[flyte-core] Flyte Connection

flyteadmin/pkg/manager/impl/execution_manager.go

@@ -421,6 +421,37 @@ func (m *ExecutionManager) getClusterAssignment(ctx context.Context, request *ad
 	}, nil
 }
 
+func (m *ExecutionManager) getExternalResourceAttributes(ctx context.Context, request *admin.ExecutionCreateRequest) (
+	*admin.ExternalResourceAttributes, error) {
+	if request.Spec.ExternalResourceAttributes != nil {
+		return request.Spec.ExternalResourceAttributes, nil
+	}
+
+	resource, err := m.resourceManager.GetResource(ctx, interfaces.ResourceRequest{
+		Project:      request.Project,
+		Domain:       request.Domain,
+		ResourceType: admin.MatchableResource_EXTERNAL_RESOURCE,
+	})
+	if err != nil && !errors.IsDoesNotExistError(err) {
+		logger.Errorf(ctx, "Failed to get external resource with error: %v", err)
+		return nil, err
+	}
+	if resource != nil && resource.Attributes.GetExternalResourceAttributes() != nil {
+		return resource.Attributes.GetExternalResourceAttributes(), nil
+	}
+
+	externalResource := m.config.ExternalResourceConfiguration().GetExternalResource()
+	connections := make(map[string]*core.Connection)
+	for key, connection := range externalResource.Connections {
+		connections[key] = &core.Connection{
+			Secrets: connection.Secrets,
+			Configs: connection.Configs,
+		}
+	}
+
+	return &admin.ExternalResourceAttributes{Connections: connections}, nil
+}
+
 func (m *ExecutionManager) launchSingleTaskExecution(
 	ctx context.Context, request admin.ExecutionCreateRequest, requestedAt time.Time) (
 	context.Context, *models.Execution, error) {
@@ -951,23 +982,29 @@ func (m *ExecutionManager) launchExecutionAndPrepareModel(
 		return nil, nil, err
 	}
 
+	externalResourceAttribute, err := m.getExternalResourceAttributes(ctx, &request)
+	if err != nil {
+		return nil, nil, err
+	}
+
 	var executionClusterLabel *admin.ExecutionClusterLabel
 	if requestSpec.ExecutionClusterLabel != nil {
 		executionClusterLabel = requestSpec.ExecutionClusterLabel
 	}
 
 	executionParameters := workflowengineInterfaces.ExecutionParameters{
-		Inputs:                executionInputs,
-		AcceptedAt:            requestedAt,
-		Labels:                labels,
-		Annotations:           annotations,
-		ExecutionConfig:       executionConfig,
-		TaskResources:         &platformTaskResources,
-		EventVersion:          m.config.ApplicationConfiguration().GetTopLevelConfig().EventVersion,
-		RoleNameKey:           m.config.ApplicationConfiguration().GetTopLevelConfig().RoleNameKey,
-		RawOutputDataConfig:   rawOutputDataConfig,
-		ClusterAssignment:     clusterAssignment,
-		ExecutionClusterLabel: executionClusterLabel,
+		Inputs:                     executionInputs,
+		AcceptedAt:                 requestedAt,
+		Labels:                     labels,
+		Annotations:                annotations,
+		ExecutionConfig:            executionConfig,
+		TaskResources:              &platformTaskResources,
+		EventVersion:               m.config.ApplicationConfiguration().GetTopLevelConfig().EventVersion,
+		RoleNameKey:                m.config.ApplicationConfiguration().GetTopLevelConfig().RoleNameKey,
+		RawOutputDataConfig:        rawOutputDataConfig,
+		ClusterAssignment:          clusterAssignment,
+		ExecutionClusterLabel:      executionClusterLabel,
+		ExternalResourceAttributes: externalResourceAttribute,
 	}
 
 	overrides, err := m.addPluginOverrides(ctx, &workflowExecutionID, launchPlan.GetSpec().WorkflowId.Name, launchPlan.Id.Name)

flyteadmin/pkg/runtime/configuration_provider.go

@@ -16,6 +16,7 @@ type ConfigurationProvider struct {
 	namespaceMappingConfiguration       interfaces.NamespaceMappingConfiguration
 	qualityOfServiceConfiguration       interfaces.QualityOfServiceConfiguration
 	clusterPoolAssignmentConfiguration  interfaces.ClusterPoolAssignmentConfiguration
+	externalResourceConfiguration       interfaces.ExternalResourceConfiguration
 }
 
 func (p *ConfigurationProvider) ApplicationConfiguration() interfaces.ApplicationConfiguration {
@@ -58,6 +59,10 @@ func (p *ConfigurationProvider) ClusterPoolAssignmentConfiguration() interfaces.
 	return p.clusterPoolAssignmentConfiguration
 }
 
+func (p *ConfigurationProvider) ExternalResourceConfiguration() interfaces.ExternalResourceConfiguration {
+	return p.externalResourceConfiguration
+}
+
 func NewConfigurationProvider() interfaces.Configuration {
 	return &ConfigurationProvider{
 		applicationConfiguration:            NewApplicationConfigurationProvider(),
@@ -70,5 +75,6 @@ func NewConfigurationProvider() interfaces.Configuration {
 		namespaceMappingConfiguration:       NewNamespaceMappingConfigurationProvider(),
 		qualityOfServiceConfiguration:       NewQualityOfServiceConfigProvider(),
 		clusterPoolAssignmentConfiguration:  NewClusterPoolAssignmentConfigurationProvider(),
+		externalResourceConfiguration:       NewExternalResourceConfigurationProvider(),
 	}
 }

flyteadmin/pkg/runtime/external_resource_provider.go

@@ -0,0 +1,21 @@
+package runtime
+
+import (
+	"github.com/flyteorg/flyte/flyteadmin/pkg/runtime/interfaces"
+	"github.com/flyteorg/flyte/flytestdlib/config"
+)
+
+const externalResourcesKey = "externalResources"
+
+var externalResourcesConfig = config.MustRegisterSection(externalResourcesKey, &interfaces.ExternalResourceConfig{})
+
+// ExternalResourcesConfigurationProvider Implementation of an interfaces.ExternalResourceConfiguration
+type ExternalResourcesConfigurationProvider struct{}
+
+func (e ExternalResourcesConfigurationProvider) GetExternalResource() interfaces.ExternalResource {
+	return externalResourcesConfig.GetConfig().(*interfaces.ExternalResourceConfig).ExternalResource
+}
+
+func NewExternalResourceConfigurationProvider() interfaces.ExternalResourceConfiguration {
+	return &ExternalResourcesConfigurationProvider{}
+}

flyteadmin/pkg/runtime/interfaces/configuration.go

@@ -12,4 +12,5 @@ type Configuration interface {
 	NamespaceMappingConfiguration() NamespaceMappingConfiguration
 	QualityOfServiceConfiguration() QualityOfServiceConfiguration
 	ClusterPoolAssignmentConfiguration() ClusterPoolAssignmentConfiguration
+	ExternalResourceConfiguration() ExternalResourceConfiguration
 }

flyteadmin/pkg/runtime/interfaces/external_resource_configuration.go

@@ -0,0 +1,20 @@
+package interfaces
+
+//go:generate mockery -name ExternalResourceConfiguration -output=mocks -case=underscore
+
+type Connection struct {
+	Secrets map[string]string `json:"secrets"`
+	Configs map[string]string `json:"configs"`
+}
+
+type ExternalResource struct {
+	Connections map[string]Connection `json:"connections"`
+}
+
+type ExternalResourceConfig struct {
+	ExternalResource ExternalResource `json:"externalResource"`
+}
+
+type ExternalResourceConfiguration interface {
+	GetExternalResource() ExternalResource
+}

flyteadmin/pkg/workflowengine/impl/prepare_execution.go

@@ -42,7 +42,7 @@ func addPermissions(securityCtx *core.SecurityContext, roleNameKey string, flyte
 
 func addExecutionOverrides(taskPluginOverrides []*admin.PluginOverride,
 	workflowExecutionConfig *admin.WorkflowExecutionConfig, recoveryExecution *core.WorkflowExecutionIdentifier,
-	taskResources *interfaces.TaskResources, flyteWf *v1alpha1.FlyteWorkflow) {
+	taskResources *interfaces.TaskResources, externalResource *admin.ExternalResourceAttributes, flyteWf *v1alpha1.FlyteWorkflow) {
 	executionConfig := v1alpha1.ExecutionConfig{
 		TaskPluginImpls: make(map[string]v1alpha1.TaskPluginOverride),
 		RecoveryExecution: v1alpha1.WorkflowExecutionIdentifier{
@@ -107,6 +107,21 @@ func addExecutionOverrides(taskPluginOverrides []*admin.PluginOverride,
 			Limits:   limits,
 		}
 	}
+
+	if externalResource != nil && externalResource.GetConnections() != nil {
+		connections := make(map[string]v1alpha1.Connection)
+		for name, connection := range externalResource.GetConnections() {
+			connections[name] = v1alpha1.Connection{
+				Secrets: connection.GetSecrets(),
+				Configs: connection.GetConfigs(),
+			}
+		}
+
+		executionConfig.ExternalResourceAttribute = v1alpha1.ExternalResourceAttributes{
+			Connections: connections,
+		}
+	}
+
 	flyteWf.ExecutionConfig = executionConfig
 }
 
@@ -140,7 +155,7 @@ func PrepareFlyteWorkflow(data interfaces.ExecutionData, flyteWorkflow *v1alpha1
 	}
 	flyteWorkflow.WorkflowMeta.EventVersion = v1alpha1.EventVersion(data.ExecutionParameters.EventVersion)
 	addExecutionOverrides(data.ExecutionParameters.TaskPluginOverrides, data.ExecutionParameters.ExecutionConfig,
-		data.ExecutionParameters.RecoveryExecution, data.ExecutionParameters.TaskResources, flyteWorkflow)
+		data.ExecutionParameters.RecoveryExecution, data.ExecutionParameters.TaskResources, data.ExecutionParameters.ExternalResourceAttributes, flyteWorkflow)
 
 	if data.ExecutionParameters.RawOutputDataConfig != nil {
 		flyteWorkflow.RawOutputDataConfig = v1alpha1.RawOutputDataConfig{

flyteadmin/pkg/workflowengine/interfaces/executor.go

@@ -18,19 +18,20 @@ type TaskResources struct {
 }
 
 type ExecutionParameters struct {
-	Inputs                *core.LiteralMap
-	AcceptedAt            time.Time
-	Labels                map[string]string
-	Annotations           map[string]string
-	TaskPluginOverrides   []*admin.PluginOverride
-	ExecutionConfig       *admin.WorkflowExecutionConfig
-	RecoveryExecution     *core.WorkflowExecutionIdentifier
-	TaskResources         *TaskResources
-	EventVersion          int
-	RoleNameKey           string
-	RawOutputDataConfig   *admin.RawOutputDataConfig
-	ClusterAssignment     *admin.ClusterAssignment
-	ExecutionClusterLabel *admin.ExecutionClusterLabel
+	Inputs                     *core.LiteralMap
+	AcceptedAt                 time.Time
+	Labels                     map[string]string
+	Annotations                map[string]string
+	TaskPluginOverrides        []*admin.PluginOverride
+	ExecutionConfig            *admin.WorkflowExecutionConfig
+	RecoveryExecution          *core.WorkflowExecutionIdentifier
+	TaskResources              *TaskResources
+	EventVersion               int
+	RoleNameKey                string
+	RawOutputDataConfig        *admin.RawOutputDataConfig
+	ClusterAssignment          *admin.ClusterAssignment
+	ExecutionClusterLabel      *admin.ExecutionClusterLabel
+	ExternalResourceAttributes *admin.ExternalResourceAttributes
 }
 
 // ExecutionData includes all parameters required to create an execution CRD object.