feat: allow global API key on OpenAI connector (#110)

Because

- We need connectors like OpenAI to use globally configured secrets to
run executions without credentials.

This commit

- Injects a global secret for the OpenAI API key on intialization.
- Transforms a keyword value in that configuration parameter into the
secret.

pkg/base/execution.go

@@ -6,6 +6,7 @@ import (
 	"strconv"
 	"strings"
 
+	"github.com/instill-ai/x/errmsg"
 	"github.com/santhosh-tekuri/jsonschema/v5"
 	"go.uber.org/zap"
 	"google.golang.org/protobuf/encoding/protojson"
@@ -138,3 +139,18 @@ func (e *ExecutionWrapper) Execute(inputs []*structpb.Struct) ([]*structpb.Struc
 
 	return outputs, err
 }
+
+// CredentialGlobalSecret is a keyword to reference a global secret in a
+// component configuration. When a component detects this value in a
+// configuration parameter, it will used the pre-configured value, injected at
+// initialization.
+const CredentialGlobalSecret = "__INSTILL_CREDENTIAL"
+
+// NewUnresolvedGlobalSecret returns an end-user error signaling that  the
+// connection configuration references a global secret that
+func NewUnresolvedGlobalSecret(key string) error {
+	return errmsg.AddMessage(
+		fmt.Errorf("unresolved global secret"),
+		fmt.Sprintf("The connection field %s can't reference a global secret.", key),
+	)
+}

pkg/connector/main.go

@@ -42,7 +42,13 @@ type connector struct {
 	con base.IConnector
 }
 
-func Init(logger *zap.Logger, usageHandler base.UsageHandler) *ConnectorStore {
+// ConnectionSecrets contains the global connection secrets of each
+// implemented connector (referenced by ID). Connectors may use these secrets
+// to skip the connector configuration step and have a ready-to-run
+// connection.
+type ConnectionSecrets map[string]map[string]any
+
+func Init(logger *zap.Logger, usageHandler base.UsageHandler, secrets ConnectionSecrets) *ConnectorStore {
 	once.Do(func() {
 
 		conStore = &ConnectorStore{
@@ -53,7 +59,14 @@ func Init(logger *zap.Logger, usageHandler base.UsageHandler) *ConnectorStore {
 		conStore.Import(stabilityai.Init(logger, usageHandler))
 		conStore.Import(instill.Init(logger, usageHandler))
 		conStore.Import(huggingface.Init(logger, usageHandler))
-		conStore.Import(openai.Init(logger, usageHandler))
+
+		{
+			// OpenAI
+			conn := openai.Init(logger, usageHandler)
+			conn = conn.WithGlobalCredentials(secrets[conn.GetID()])
+			conStore.Import(conn)
+		}
+
 		conStore.Import(archetypeai.Init(logger, usageHandler))
 		conStore.Import(numbers.Init(logger, usageHandler))
 		conStore.Import(airbyte.Init(logger, usageHandler))

pkg/connector/openai/v0/main.go

@@ -6,6 +6,7 @@ import (
 	"encoding/base64"
 	"encoding/json"
 	"fmt"
+	"strings"
 	"sync"
 
 	"github.com/gabriel-vasile/mimetype"
@@ -24,6 +25,9 @@ const (
 	speechRecognitionTask = "TASK_SPEECH_RECOGNITION"
 	textToSpeechTask      = "TASK_TEXT_TO_SPEECH"
 	textToImageTask       = "TASK_TEXT_TO_IMAGE"
+
+	cfgAPIKey       = "api_key"
+	cfgOrganization = "organization"
 )
 
 var (
@@ -35,39 +39,93 @@ var (
 	openAIJSON []byte
 
 	once sync.Once
-	con  *connector
+	con  *Connector
 )
 
-type connector struct {
+// Connector executes queries against OpenAI.
+type Connector struct {
 	base.BaseConnector
+
+	// Global secrets.
+	globalAPIKey string
 }
 
 type execution struct {
 	base.BaseConnectorExecution
 }
 
-func Init(l *zap.Logger, u base.UsageHandler) *connector {
+// Init returns an initialized OpenAI connector.
+func Init(l *zap.Logger, u base.UsageHandler) *Connector {
 	once.Do(func() {
-		con = &connector{
+		con = &Connector{
 			BaseConnector: base.BaseConnector{
 				Logger:       l,
 				UsageHandler: u,
 			},
 		}
+
 		err := con.LoadConnectorDefinition(definitionJSON, tasksJSON, map[string][]byte{"openai.json": openAIJSON})
 		if err != nil {
 			panic(err)
 		}
 	})
+
 	return con
 }
 
-func (c *connector) CreateExecution(sysVars map[string]any, connection *structpb.Struct, task string) (*base.ExecutionWrapper, error) {
+// The connection parameter is defined with snake_case, but the
+// environment variable configuration loader replaces underscores by dots,
+// so we can't use the parameter key directly.
+func readFromSecrets(key string, s map[string]any) string {
+	sanitized := strings.ReplaceAll(key, "_", "")
+	if v, ok := s[sanitized].(string); ok {
+		return v
+	}
+
+	return ""
+}
+
+// WithGlobalCredentials reads the global connection configuration, which can
+// be used to execute the connector with globally defined secrets.
+func (c *Connector) WithGlobalCredentials(s map[string]any) *Connector {
+	c.globalAPIKey = readFromSecrets(cfgAPIKey, s)
+
+	return c
+}
+
+// CreateExecution initializes a connector executor that can be used in a
+// pipeline trigger.
+func (c *Connector) CreateExecution(sysVars map[string]any, connection *structpb.Struct, task string) (*base.ExecutionWrapper, error) {
+	resolvedConnection, err := c.resolveSecrets(connection)
+	if err != nil {
+		return nil, err
+	}
+
 	return &base.ExecutionWrapper{Execution: &execution{
-		BaseConnectorExecution: base.BaseConnectorExecution{Connector: c, SystemVariables: sysVars, Connection: connection, Task: task},
+		BaseConnectorExecution: base.BaseConnectorExecution{
+			Connector:       c,
+			SystemVariables: sysVars,
+			Connection:      resolvedConnection,
+			Task:            task,
+		},
 	}}, nil
 }
 
+// resolveSecrets looks for references to a global secret in the connection
+// and replaces them by the global secret injected during initialization.
+func (c *Connector) resolveSecrets(conn *structpb.Struct) (*structpb.Struct, error) {
+	apiKey := conn.GetFields()[cfgAPIKey].GetStringValue()
+	if apiKey == base.CredentialGlobalSecret {
+		if c.globalAPIKey == "" {
+			return nil, base.NewUnresolvedGlobalSecret(cfgAPIKey)
+		}
+
+		conn.GetFields()[cfgAPIKey] = structpb.NewStringValue(c.globalAPIKey)
+	}
+
+	return conn, nil
+}
+
 // getBasePath returns OpenAI's API URL. This configuration param allows us to
 // override the API the connector will point to. It isn't meant to be exposed
 // to users. Rather, it can serve to test the logic against a fake server.
@@ -82,11 +140,11 @@ func getBasePath(config *structpb.Struct) string {
 }
 
 func getAPIKey(config *structpb.Struct) string {
-	return config.GetFields()["api_key"].GetStringValue()
+	return config.GetFields()[cfgAPIKey].GetStringValue()
 }
 
 func getOrg(config *structpb.Struct) string {
-	val, ok := config.GetFields()["organization"]
+	val, ok := config.GetFields()[cfgOrganization]
 	if !ok {
 		return ""
 	}
@@ -328,7 +386,7 @@ func (e *execution) Execute(inputs []*structpb.Struct) ([]*structpb.Struct, erro
 }
 
 // Test checks the connector state.
-func (c *connector) Test(sysVars map[string]any, connection *structpb.Struct) error {
+func (c *Connector) Test(_ map[string]any, connection *structpb.Struct) error {
 	models := ListModelsResponse{}
 	req := newClient(connection, c.Logger).R().SetResult(&models)