[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.