[flyte-core] Flyte Connection

flyteadmin/pkg/manager/impl/execution_manager.go

@@ -427,6 +427,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) {
@@ -957,23 +988,29 @@ func (m *ExecutionManager) launchExecutionAndPrepareModel(
 		return nil, 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/manager/impl/execution_manager_test.go

@@ -5495,6 +5495,93 @@ func TestGetClusterAssignment(t *testing.T) {
 	})
 }
 
+func TestGetExternalResourceAttribute(t *testing.T) {
+	externalResourceAttribute := &admin.ExternalResourceAttributes{
+		Connections: map[string]*core.Connection{
+			"conn1": {
+				Secrets: map[string]string{"key1": "value1"},
+				Configs: map[string]string{"key2": "value2"},
+			},
+		},
+	}
+	resourceManager := managerMocks.MockResourceManager{}
+	resourceManager.GetResourceFunc = func(ctx context.Context,
+		request managerInterfaces.ResourceRequest) (*managerInterfaces.ResourceResponse, error) {
+		assert.EqualValues(t, request, managerInterfaces.ResourceRequest{
+			Project:      workflowIdentifier.Project,
+			Domain:       workflowIdentifier.Domain,
+			ResourceType: admin.MatchableResource_EXTERNAL_RESOURCE,
+		})
+		return &managerInterfaces.ResourceResponse{
+			Attributes: &admin.MatchingAttributes{
+				Target: &admin.MatchingAttributes_ExternalResourceAttributes{
+					ExternalResourceAttributes: externalResourceAttribute,
+				},
+			},
+		}, nil
+	}
+
+	executionManager := ExecutionManager{resourceManager: &resourceManager}
+
+	t.Run("value from db", func(t *testing.T) {
+		era, err := executionManager.getExternalResourceAttributes(context.TODO(), &admin.ExecutionCreateRequest{
+			Project: workflowIdentifier.Project,
+			Domain:  workflowIdentifier.Domain,
+			Spec:    &admin.ExecutionSpec{},
+		})
+		assert.NoError(t, err)
+		assert.True(t, proto.Equal(era, externalResourceAttribute))
+	})
+	t.Run("value from request", func(t *testing.T) {
+		reqExternalResourceAttribute := &admin.ExternalResourceAttributes{
+			Connections: map[string]*core.Connection{
+				"conn2": {
+					Secrets: map[string]string{"key1": "value1"},
+					Configs: map[string]string{"key2": "value2"},
+				},
+			},
+		}
+		era, err := executionManager.getExternalResourceAttributes(context.TODO(), &admin.ExecutionCreateRequest{
+			Project: workflowIdentifier.Project,
+			Domain:  workflowIdentifier.Domain,
+			Spec: &admin.ExecutionSpec{
+				ExternalResourceAttributes: reqExternalResourceAttribute,
+			},
+		})
+		assert.NoError(t, err)
+		assert.True(t, proto.Equal(era, reqExternalResourceAttribute))
+	})
+	t.Run("value from config", func(t *testing.T) {
+		name := "conn3"
+		connections := map[string]runtimeInterfaces.Connection{
+			name: {
+				Secrets: map[string]string{"key1": "value1"},
+				Configs: map[string]string{"key2": "value2"},
+			},
+		}
+		externalResourceProvider := &runtimeIFaceMocks.ExternalResourceConfiguration{}
+		externalResourceProvider.OnGetExternalResource().Return(runtimeInterfaces.ExternalResource{
+			Connections: connections,
+		})
+		mockConfig := getMockExecutionsConfigProvider()
+		mockConfig.(*runtimeMocks.MockConfigurationProvider).AddExternalResourceConfiguration(externalResourceProvider)
+
+		executionManager := ExecutionManager{
+			resourceManager: &managerMocks.MockResourceManager{},
+			config:          mockConfig,
+		}
+
+		era, err := executionManager.getExternalResourceAttributes(context.TODO(), &admin.ExecutionCreateRequest{
+			Project: workflowIdentifier.Project,
+			Domain:  workflowIdentifier.Domain,
+			Spec:    &admin.ExecutionSpec{},
+		})
+		assert.NoError(t, err)
+		assert.Equal(t, connections[name].Secrets, era.GetConnections()[name].Secrets)
+		assert.Equal(t, connections[name].Configs, era.GetConnections()[name].Configs)
+	})
+}
+
 func TestResolvePermissions(t *testing.T) {
 	assumableIamRole := "role"
 	k8sServiceAccount := "sa"

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/runtime/mocks/mock_configuration_provider.go

@@ -17,6 +17,7 @@ type MockConfigurationProvider struct {
 	namespaceMappingConfiguration       interfaces.NamespaceMappingConfiguration
 	qualityOfServiceConfiguration       interfaces.QualityOfServiceConfiguration
 	clusterPoolAssignmentConfiguration  interfaces.ClusterPoolAssignmentConfiguration
+	externalResourceConfiguration       interfaces.ExternalResourceConfiguration
 }
 
 func (p *MockConfigurationProvider) ApplicationConfiguration() interfaces.ApplicationConfiguration {
@@ -79,6 +80,14 @@ func (p *MockConfigurationProvider) AddClusterPoolAssignmentConfiguration(cfg in
 	p.clusterPoolAssignmentConfiguration = cfg
 }
 
+func (p *MockConfigurationProvider) ExternalResourceConfiguration() interfaces.ExternalResourceConfiguration {
+	return p.externalResourceConfiguration
+}
+
+func (p *MockConfigurationProvider) AddExternalResourceConfiguration(cfg interfaces.ExternalResourceConfiguration) {
+	p.externalResourceConfiguration = cfg
+}
+
 func NewMockConfigurationProvider(
 	applicationConfiguration interfaces.ApplicationConfiguration,
 	queueConfiguration interfaces.QueueConfiguration,
@@ -94,6 +103,9 @@ func NewMockConfigurationProvider(
 	mockClusterPoolAssignmentConfiguration := &ifaceMocks.ClusterPoolAssignmentConfiguration{}
 	mockClusterPoolAssignmentConfiguration.OnGetClusterPoolAssignments().Return(make(map[string]interfaces.ClusterPoolAssignment))
 
+	mockExternalResourceConfiguration := &ifaceMocks.ExternalResourceConfiguration{}
+	mockExternalResourceConfiguration.OnGetExternalResource().Return(interfaces.ExternalResource{})
+
 	return &MockConfigurationProvider{
 		applicationConfiguration:           applicationConfiguration,
 		queueConfiguration:                 queueConfiguration,
@@ -103,5 +115,6 @@ func NewMockConfigurationProvider(
 		namespaceMappingConfiguration:      namespaceMappingConfiguration,
 		qualityOfServiceConfiguration:      mockQualityOfServiceConfiguration,
 		clusterPoolAssignmentConfiguration: mockClusterPoolAssignmentConfiguration,
+		externalResourceConfiguration:      mockExternalResourceConfiguration,
 	}
 }

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{