Make execution transformer configurable

pradithya · pradithya · commit 3adb19bcbaef · 2023-05-06T16:09:04.000+08:00
Signed-off-by: Pradithya Aria Pura &lt;pradithya.aria@gmail.com&gt;
diff --git a/pkg/manager/impl/execution_manager.go b/pkg/manager/impl/execution_manager.go
@@ -1442,7 +1442,12 @@ func (m *ExecutionManager) ListExecutions(
 		logger.Debugf(ctx, "Failed to list executions using input [%+v] with err %v", listExecutionsInput, err)
 		return nil, err
 	}
-	executionList, err := transformers.FromExecutionModels(output.Executions, transformers.ListExecutionTransformerOptions)
+
+	listExecutionTransformer := &transformers.ExecutionTransformerOptions{
+		TrimErrorMessage:      m.config.ApplicationConfiguration().GetTopLevelConfig().ListExecutionTransformersConfig.TrimErrorMessages,
+		MaxErrorMessageLength: m.config.ApplicationConfiguration().GetTopLevelConfig().ListExecutionTransformersConfig.MaxErrorMessageLength,
+	}
+	executionList, err := transformers.FromExecutionModels(output.Executions, listExecutionTransformer)
 	if err != nil {
 		logger.Errorf(ctx,
 			"Failed to transform execution models [%+v] with err: %v", output.Executions, err)
diff --git a/pkg/manager/impl/node_execution_manager.go b/pkg/manager/impl/node_execution_manager.go
@@ -333,6 +333,11 @@ func (m *NodeExecutionManager) transformNodeExecutionModel(ctx context.Context,
 
 func (m *NodeExecutionManager) transformNodeExecutionModelList(ctx context.Context, nodeExecutionModels []models.NodeExecution) ([]*admin.NodeExecution, error) {
 	nodeExecutions := make([]*admin.NodeExecution, len(nodeExecutionModels))
+	listExecutionTransformer := &transformers.ExecutionTransformerOptions{
+		TrimErrorMessage:      m.config.ApplicationConfiguration().GetTopLevelConfig().ListExecutionTransformersConfig.TrimErrorMessages,
+		MaxErrorMessageLength: m.config.ApplicationConfiguration().GetTopLevelConfig().ListExecutionTransformersConfig.MaxErrorMessageLength,
+	}
+
 	for idx, nodeExecutionModel := range nodeExecutionModels {
 		nodeExecution, err := m.transformNodeExecutionModel(ctx, nodeExecutionModel, &core.NodeExecutionIdentifier{
 			ExecutionId: &core.WorkflowExecutionIdentifier{
@@ -341,7 +346,7 @@ func (m *NodeExecutionManager) transformNodeExecutionModelList(ctx context.Conte
 				Name:    nodeExecutionModel.Name,
 			},
 			NodeId: nodeExecutionModel.NodeID,
-		}, transformers.ListExecutionTransformerOptions)
+		}, listExecutionTransformer)
 		if err != nil {
 			return nil, err
 		}
diff --git a/pkg/manager/impl/node_execution_manager_test.go b/pkg/manager/impl/node_execution_manager_test.go
@@ -537,7 +537,8 @@ func TestTransformNodeExecutionModelList(t *testing.T) {
 		})
 
 	manager := NodeExecutionManager{
-		db: repository,
+		db:     repository,
+		config: getMockExecutionsConfigProvider(),
 	}
 	nodeExecutions, err := manager.transformNodeExecutionModelList(ctx, []models.NodeExecution{
 		{
diff --git a/pkg/manager/impl/testutils/config.go b/pkg/manager/impl/testutils/config.go
@@ -30,5 +30,8 @@ func GetApplicationConfigWithDefaultDomains() runtimeInterfaces.ApplicationConfi
 		Scheme: common.Local, SignedURL: runtimeInterfaces.SignedURL{
 			Enabled: true,
 		}})
+
+	config.GetTopLevelConfig().ListExecutionTransformersConfig.TrimErrorMessages = true
+	config.GetTopLevelConfig().ListExecutionTransformersConfig.MaxErrorMessageLength = 10240
 	return &config
 }
diff --git a/pkg/repositories/transformers/execution.go b/pkg/repositories/transformers/execution.go
@@ -22,8 +22,6 @@ import (
 	"k8s.io/apimachinery/pkg/util/sets"
 )
 
-const trimmedErrMessageLen = 10240
-
 var clusterReassignablePhases = sets.NewString(core.WorkflowExecution_UNDEFINED.String(), core.WorkflowExecution_QUEUED.String())
 
 // CreateExecutionModelInput encapsulates request parameters for calls to CreateExecutionModel.
@@ -47,13 +45,11 @@ type CreateExecutionModelInput struct {
 }
 
 type ExecutionTransformerOptions struct {
-	TrimErrorMessage bool
+	TrimErrorMessage      bool
+	MaxErrorMessageLength int
 }
 
 var DefaultExecutionTransformerOptions = &ExecutionTransformerOptions{}
-var ListExecutionTransformerOptions = &ExecutionTransformerOptions{
-	TrimErrorMessage: true,
-}
 
 // CreateExecutionModel transforms a ExecutionCreateRequest to a Execution model
 func CreateExecutionModel(input CreateExecutionModelInput) (*models.Execution, error) {
@@ -328,8 +324,8 @@ func FromExecutionModel(executionModel models.Execution, opts *ExecutionTransfor
 	}
 	if closure.GetError() != nil && opts != nil && opts.TrimErrorMessage && len(closure.GetError().Message) > 0 {
 		trimmedErrOutputResult := closure.GetError()
-		if len(trimmedErrOutputResult.Message) > trimmedErrMessageLen {
-			trimmedErrOutputResult.Message = trimmedErrOutputResult.Message[0:trimmedErrMessageLen]
+		if len(trimmedErrOutputResult.Message) > opts.MaxErrorMessageLength {
+			trimmedErrOutputResult.Message = trimmedErrOutputResult.Message[0:opts.MaxErrorMessageLength]
 		}
 		closure.OutputResult = &admin.ExecutionClosure_Error{
 			Error: trimmedErrOutputResult,
diff --git a/pkg/repositories/transformers/execution_test.go b/pkg/repositories/transformers/execution_test.go
@@ -570,6 +570,7 @@ func TestFromExecutionModel_Aborted(t *testing.T) {
 }
 
 func TestFromExecutionModel_Error(t *testing.T) {
+	trimmedErrMessageLen := 10240
 	extraLongErrMsg := string(make([]byte, 2*trimmedErrMessageLen))
 	execErr := &core.ExecutionError{
 		Code:    "CODE",
@@ -590,7 +591,8 @@ func TestFromExecutionModel_Error(t *testing.T) {
 		Closure: executionClosureBytes,
 	}
 	execution, err := FromExecutionModel(executionModel, &ExecutionTransformerOptions{
-		TrimErrorMessage: true,
+		TrimErrorMessage:      true,
+		MaxErrorMessageLength: trimmedErrMessageLen,
 	})
 	expectedExecErr := execErr
 	expectedExecErr.Message = string(make([]byte, trimmedErrMessageLen))
diff --git a/pkg/repositories/transformers/node_execution.go b/pkg/repositories/transformers/node_execution.go
@@ -323,8 +323,8 @@ func FromNodeExecutionModel(nodeExecutionModel models.NodeExecution, opts *Execu
 	}
 	if closure.GetError() != nil && opts != nil && opts.TrimErrorMessage && len(closure.GetError().Message) > 0 {
 		trimmedErrOutputResult := closure.GetError()
-		if len(trimmedErrOutputResult.Message) > trimmedErrMessageLen {
-			trimmedErrOutputResult.Message = trimmedErrOutputResult.Message[0:trimmedErrMessageLen]
+		if len(trimmedErrOutputResult.Message) > opts.MaxErrorMessageLength {
+			trimmedErrOutputResult.Message = trimmedErrOutputResult.Message[0:opts.MaxErrorMessageLength]
 		}
 		closure.OutputResult = &admin.NodeExecutionClosure_Error{
 			Error: trimmedErrOutputResult,
diff --git a/pkg/repositories/transformers/node_execution_test.go b/pkg/repositories/transformers/node_execution_test.go
@@ -528,6 +528,7 @@ func TestFromNodeExecutionModel(t *testing.T) {
 }
 
 func TestFromNodeExecutionModel_Error(t *testing.T) {
+	trimmedErrMessageLen := 10240
 	extraLongErrMsg := string(make([]byte, 2*trimmedErrMessageLen))
 	execErr := &core.ExecutionError{
 		Code:    "CODE",
@@ -551,7 +552,10 @@ func TestFromNodeExecutionModel_Error(t *testing.T) {
 		NodeExecutionMetadata: nodeExecutionMetadataBytes,
 		InputURI:              "input uri",
 		Duration:              duration,
-	}, &ExecutionTransformerOptions{TrimErrorMessage: true})
+	}, &ExecutionTransformerOptions{
+		TrimErrorMessage:      true,
+		MaxErrorMessageLength: trimmedErrMessageLen,
+	})
 	assert.Nil(t, err)
 
 	expectedExecErr := execErr
diff --git a/pkg/repositories/transformers/task_execution.go b/pkg/repositories/transformers/task_execution.go
@@ -441,8 +441,8 @@ func FromTaskExecutionModel(taskExecutionModel models.TaskExecution, opts *Execu
 	}
 	if closure.GetError() != nil && opts != nil && opts.TrimErrorMessage && len(closure.GetError().Message) > 0 {
 		trimmedErrOutputResult := closure.GetError()
-		if len(trimmedErrOutputResult.Message) > trimmedErrMessageLen {
-			trimmedErrOutputResult.Message = trimmedErrOutputResult.Message[0:trimmedErrMessageLen]
+		if len(trimmedErrOutputResult.Message) > opts.MaxErrorMessageLength {
+			trimmedErrOutputResult.Message = trimmedErrOutputResult.Message[0:opts.MaxErrorMessageLength]
 		}
 		closure.OutputResult = &admin.TaskExecutionClosure_Error{
 			Error: trimmedErrOutputResult,
diff --git a/pkg/repositories/transformers/task_execution_test.go b/pkg/repositories/transformers/task_execution_test.go
@@ -600,6 +600,7 @@ func TestFromTaskExecutionModel(t *testing.T) {
 }
 
 func TestFromTaskExecutionModel_Error(t *testing.T) {
+	trimmedErrMessageLen := 10240
 	extraLongErrMsg := string(make([]byte, 2*trimmedErrMessageLen))
 	execErr := &core.ExecutionError{
 		Code:    "CODE",
@@ -633,7 +634,8 @@ func TestFromTaskExecutionModel_Error(t *testing.T) {
 		Closure:  closureBytes,
 	}
 	taskExecution, err := FromTaskExecutionModel(taskExecutionModel, &ExecutionTransformerOptions{
-		TrimErrorMessage: true,
+		TrimErrorMessage:      true,
+		MaxErrorMessageLength: trimmedErrMessageLen,
 	})
 
 	expectedExecErr := execErr
@@ -653,7 +655,8 @@ func TestFromTaskExecutionModel_Error(t *testing.T) {
 	})
 	taskExecutionModel.Closure = closureBytes
 	taskExecution, err = FromTaskExecutionModel(taskExecutionModel, &ExecutionTransformerOptions{
-		TrimErrorMessage: true,
+		TrimErrorMessage:      true,
+		MaxErrorMessageLength: trimmedErrMessageLen,
 	})
 	expectedExecErr = execErr
 	expectedExecErr.Message = string(make([]byte, 10))
diff --git a/pkg/runtime/application_config_provider.go b/pkg/runtime/application_config_provider.go
@@ -33,6 +33,11 @@ var flyteAdminConfig = config.MustRegisterSection(flyteAdmin, &interfaces.Applic
 	MaxParallelism:              25,
 	K8SServiceAccount:           "",
 	UseOffloadedWorkflowClosure: false,
+
+	ListExecutionTransformersConfig: interfaces.ExecutionTransformersConfig{
+		TrimErrorMessages:     true,
+		MaxErrorMessageLength: 10240,
+	},
 })
 
 var schedulerConfig = config.MustRegisterSection(scheduler, &interfaces.SchedulerConfig{
diff --git a/pkg/runtime/interfaces/application_configuration.go b/pkg/runtime/interfaces/application_configuration.go
@@ -48,6 +48,13 @@ type PostgresConfig struct {
 	Debug        bool   `json:"debug" pflag:" Whether or not to start the database connection with debug mode enabled."`
 }
 
+type ExecutionTransformersConfig struct {
+	// TrimErrorMessages indicates whether error messages returned by the list workflow execution API should be trimmed.
+	TrimErrorMessages bool `json:"trimErrorMessages"`
+	// The maximum length of an error message returned by the list workflow execution API.
+	MaxErrorMessageLength int `json:"maxErrorMessageLength"`
+}
+
 // ApplicationConfig is the base configuration to start admin
 type ApplicationConfig struct {
 	// The RoleName key inserted as an annotation (https://kubernetes.io/docs/concepts/overview/working-with-objects/annotations/)
@@ -94,6 +101,9 @@ type ApplicationConfig struct {
 
 	// Enabling will use Storage (s3/gcs/etc) to offload static parts of CRDs.
 	UseOffloadedWorkflowClosure bool `json:"useOffloadedWorkflowClosure"`
+
+	// Configures the execution transformers
+	ListExecutionTransformersConfig ExecutionTransformersConfig `json:"listExecutionTransformersConfig"`
 }
 
 func (a *ApplicationConfig) GetRoleNameKey() string {

Original file line number	Diff line number	Diff line change
`@@ -537,7 +537,8 @@ func TestTransformNodeExecutionModelList(t *testing.T) {`
`537`	`537`	`})`
`538`	`538`
`539`	`539`	`manager := NodeExecutionManager{`
`540`		`- db: repository,`
	`540`	`+ db: repository,`
	`541`	`+ config: getMockExecutionsConfigProvider(),`
`541`	`542`	`}`
`542`	`543`	`nodeExecutions, err := manager.transformNodeExecutionModelList(ctx, []models.NodeExecution{`
`543`	`544`	`{`
Original file line number	Diff line number	Diff line change
`@@ -30,5 +30,8 @@ func GetApplicationConfigWithDefaultDomains() runtimeInterfaces.ApplicationConfi`
`30`	`30`	`Scheme: common.Local, SignedURL: runtimeInterfaces.SignedURL{`
`31`	`31`	`Enabled: true,`
`32`	`32`	`}})`
	`33`	`+`
	`34`	`+ config.GetTopLevelConfig().ListExecutionTransformersConfig.TrimErrorMessages = true`
	`35`	`+ config.GetTopLevelConfig().ListExecutionTransformersConfig.MaxErrorMessageLength = 10240`
`33`	`36`	`return &config`
`34`	`37`	`}`
Original file line number	Diff line number	Diff line change
`@@ -22,8 +22,6 @@ import (`
`22`	`22`	`"k8s.io/apimachinery/pkg/util/sets"`
`23`	`23`	`)`
`24`	`24`
`25`		`-const trimmedErrMessageLen = 10240`
`26`		`-`
`27`	`25`	`var clusterReassignablePhases = sets.NewString(core.WorkflowExecution_UNDEFINED.String(), core.WorkflowExecution_QUEUED.String())`
`28`	`26`
`29`	`27`	`// CreateExecutionModelInput encapsulates request parameters for calls to CreateExecutionModel.`
`@@ -47,13 +45,11 @@ type CreateExecutionModelInput struct {`
`47`	`45`	`}`
`48`	`46`
`49`	`47`	`type ExecutionTransformerOptions struct {`
`50`		`- TrimErrorMessage bool`
	`48`	`+ TrimErrorMessage bool`
	`49`	`+ MaxErrorMessageLength int`
`51`	`50`	`}`
`52`	`51`
`53`	`52`	`var DefaultExecutionTransformerOptions = &ExecutionTransformerOptions{}`
`54`		`-var ListExecutionTransformerOptions = &ExecutionTransformerOptions{`
`55`		`- TrimErrorMessage: true,`
`56`		`-}`
`57`	`53`
`58`	`54`	`// CreateExecutionModel transforms a ExecutionCreateRequest to a Execution model`
`59`	`55`	`func CreateExecutionModel(input CreateExecutionModelInput) (*models.Execution, error) {`
`@@ -328,8 +324,8 @@ func FromExecutionModel(executionModel models.Execution, opts *ExecutionTransfor`
`328`	`324`	`}`
`329`	`325`	`if closure.GetError() != nil && opts != nil && opts.TrimErrorMessage && len(closure.GetError().Message) > 0 {`
`330`	`326`	`trimmedErrOutputResult := closure.GetError()`
`331`		`- if len(trimmedErrOutputResult.Message) > trimmedErrMessageLen {`
`332`		`- trimmedErrOutputResult.Message = trimmedErrOutputResult.Message[0:trimmedErrMessageLen]`
	`327`	`+ if len(trimmedErrOutputResult.Message) > opts.MaxErrorMessageLength {`
	`328`	`+ trimmedErrOutputResult.Message = trimmedErrOutputResult.Message[0:opts.MaxErrorMessageLength]`
`333`	`329`	`}`
`334`	`330`	`closure.OutputResult = &admin.ExecutionClosure_Error{`
`335`	`331`	`Error: trimmedErrOutputResult,`
Original file line number	Diff line number	Diff line change
`@@ -323,8 +323,8 @@ func FromNodeExecutionModel(nodeExecutionModel models.NodeExecution, opts *Execu`
`323`	`323`	`}`
`324`	`324`	`if closure.GetError() != nil && opts != nil && opts.TrimErrorMessage && len(closure.GetError().Message) > 0 {`
`325`	`325`	`trimmedErrOutputResult := closure.GetError()`
`326`		`- if len(trimmedErrOutputResult.Message) > trimmedErrMessageLen {`
`327`		`- trimmedErrOutputResult.Message = trimmedErrOutputResult.Message[0:trimmedErrMessageLen]`
	`326`	`+ if len(trimmedErrOutputResult.Message) > opts.MaxErrorMessageLength {`
	`327`	`+ trimmedErrOutputResult.Message = trimmedErrOutputResult.Message[0:opts.MaxErrorMessageLength]`
`328`	`328`	`}`
`329`	`329`	`closure.OutputResult = &admin.NodeExecutionClosure_Error{`
`330`	`330`	`Error: trimmedErrOutputResult,`
Original file line number	Diff line number	Diff line change
`@@ -441,8 +441,8 @@ func FromTaskExecutionModel(taskExecutionModel models.TaskExecution, opts *Execu`
`441`	`441`	`}`
`442`	`442`	`if closure.GetError() != nil && opts != nil && opts.TrimErrorMessage && len(closure.GetError().Message) > 0 {`
`443`	`443`	`trimmedErrOutputResult := closure.GetError()`
`444`		`- if len(trimmedErrOutputResult.Message) > trimmedErrMessageLen {`
`445`		`- trimmedErrOutputResult.Message = trimmedErrOutputResult.Message[0:trimmedErrMessageLen]`
	`444`	`+ if len(trimmedErrOutputResult.Message) > opts.MaxErrorMessageLength {`
	`445`	`+ trimmedErrOutputResult.Message = trimmedErrOutputResult.Message[0:opts.MaxErrorMessageLength]`
`446`	`446`	`}`
`447`	`447`	`closure.OutputResult = &admin.TaskExecutionClosure_Error{`
`448`	`448`	`Error: trimmedErrOutputResult,`