Trim error messages read from errors.pb to avoid spaming Etcd. and Ad…

…min (flyteorg#141)


Co-authored-by: Ketan Umare <kumare@lyft.com>

flytepropeller/pkg/controller/nodes/task/config/config.go

@@ -25,6 +25,7 @@ var (
 			BaseSecond:  2,
 			MaxDuration: config.Duration{Duration: time.Minute * 10},
 		},
+		MaxErrorMessageLength: 2048,
 	}
 
 	section = config.MustRegisterSection(SectionKey, defaultConfig)
@@ -35,6 +36,7 @@ type Config struct {
 	MaxPluginPhaseVersions int32            `json:"max-plugin-phase-versions" pflag:",Maximum number of plugin phase versions allowed for one phase."`
 	BarrierConfig          BarrierConfig    `json:"barrier" pflag:",Config for Barrier implementation"`
 	BackOffConfig          BackOffConfig    `json:"backoff" pflag:",Config for Exponential BackOff implementation"`
+	MaxErrorMessageLength  int              `json:"maxLogMessageLength" pflag:",Max length of error message."`
 }
 
 type BarrierConfig struct {

flytepropeller/pkg/controller/nodes/task/handler_test.go

@@ -574,6 +574,7 @@ func Test_task_Handle_NoCatalog(t *testing.T) {
 			nCtx := createNodeContext(tt.args.startingPluginPhase, uint32(tt.args.startingPluginPhaseVersion), tt.args.expectedState, ev, "test", state)
 			c := &pluginCatalogMocks.Client{}
 			tk := Handler{
+				cfg: &config.Config{MaxErrorMessageLength: 100},
 				plugins: map[pluginCore.TaskType]pluginCore.Plugin{
 					"test": fakeplugins.NewPhaseBasedPlugin(),
 				},

flytepropeller/pkg/controller/nodes/task/pre_post_execution.go

@@ -7,12 +7,13 @@ import (
 	"github.com/lyft/flyteplugins/go/tasks/pluginmachinery/catalog"
 	pluginCore "github.com/lyft/flyteplugins/go/tasks/pluginmachinery/core"
 	"github.com/lyft/flyteplugins/go/tasks/pluginmachinery/io"
-	"github.com/lyft/flytepropeller/pkg/apis/flyteworkflow/v1alpha1"
-	errors2 "github.com/lyft/flytepropeller/pkg/controller/nodes/errors"
 	"github.com/lyft/flytestdlib/logger"
 	"github.com/pkg/errors"
 	"google.golang.org/grpc/codes"
 	"google.golang.org/grpc/status"
+
+	"github.com/lyft/flytepropeller/pkg/apis/flyteworkflow/v1alpha1"
+	errors2 "github.com/lyft/flytepropeller/pkg/controller/nodes/errors"
 )
 
 func (t *Handler) CheckCatalogCache(ctx context.Context, tr pluginCore.TaskReader, inputReader io.InputReader, outputWriter io.OutputWriter) (bool, error) {
@@ -94,6 +95,14 @@ func (t *Handler) ValidateOutputAndCacheAdd(ctx context.Context, nodeID v1alpha1
 		if err != nil {
 			return nil, err
 		}
+
+		if taskErr.ExecutionError == nil {
+			taskErr.ExecutionError = &core.ExecutionError{Kind: core.ExecutionError_UNKNOWN, Code: "Unknown", Message: "Unknown"}
+		}
+		// Errors can be arbitrary long since they are written by containers/potentially 3rd party plugins. This ensures
+		// the error message length will never be big enough to cause write failures to Etcd. or spam Admin DB with huge
+		// objects.
+		taskErr.Message = trimErrorMessage(taskErr.Message, t.cfg.MaxErrorMessageLength)
 		return &taskErr, nil
 	}
 

flytepropeller/pkg/controller/nodes/task/transformer.go

@@ -42,6 +42,14 @@ func ToTaskEventPhase(p pluginCore.Phase) core.TaskExecution_Phase {
 	}
 }
 
+func trimErrorMessage(original string, maxLength int) string {
+	if len(original) <= maxLength {
+		return original
+	}
+
+	return original[0:maxLength/2] + original[len(original)-maxLength/2:]
+}
+
 func ToTaskExecutionEvent(taskExecID *core.TaskExecutionIdentifier, in io.InputFilePaths, out io.OutputFilePaths, info pluginCore.PhaseInfo) (*event.TaskExecutionEvent, error) {
 	// Transitions to a new phase
 

flytepropeller/pkg/controller/nodes/task/transformer_test.go

@@ -27,6 +27,29 @@ func TestToTaskEventPhase(t *testing.T) {
 	assert.Equal(t, core.TaskExecution_QUEUED, ToTaskEventPhase(pluginCore.PhaseQueued))
 }
 
+func Test_trimErrorMessage(t *testing.T) {
+	const inputStr = "0123456789"
+	t.Run("Length less or equal than max", func(t *testing.T) {
+		input := inputStr
+		assert.Equal(t, input, trimErrorMessage(input, 10))
+	})
+
+	t.Run("Length > max", func(t *testing.T) {
+		input := inputStr
+		assert.Equal(t, "01236789", trimErrorMessage(input, 8))
+	})
+
+	t.Run("Odd Max", func(t *testing.T) {
+		input := inputStr
+		assert.Equal(t, "01236789", trimErrorMessage(input, 9))
+	})
+
+	t.Run("Odd input", func(t *testing.T) {
+		input := "012345678"
+		assert.Equal(t, "012345678", trimErrorMessage(input, 9))
+	})
+}
+
 func TestToTaskExecutionEvent(t *testing.T) {
 	tkID := &core.Identifier{}
 	nodeID := &core.NodeExecutionIdentifier{}