Cleanup flyte propeller

Signed-off-by: Jason Parraga <[email protected]>

Author: Sovietaced

flytepropeller/events/local_eventsink.go

@@ -65,8 +65,5 @@ func NewFileSink(path string) (EventSink, error) {
 
 	w := bufio.NewWriter(f)
 
-	if err != nil {
-		return nil, err
-	}
 	return &localSink{writer: &FileWriter{ioWriter: w}}, nil
 }

flytepropeller/pkg/apis/flyteworkflow/v1alpha1/workflow_status.go

@@ -145,7 +145,7 @@ func (in *WorkflowStatus) GetNodeExecutionStatus(ctx context.Context, id NodeID)
 
 	dataDir, err := in.ConstructNodeDataDir(ctx, id)
 	if err != nil {
-		logger.Errorf(ctx, "Failed to construct data dir for node [%v], exec id [%v]", id)
+		logger.Errorf(ctx, "Failed to construct data dir for node [%v]", id)
 		return n
 	}
 

flytepropeller/pkg/controller/controller.go

@@ -446,9 +446,7 @@ func New(ctx context.Context, cfg *config.Config, kubeClientset kubernetes.Inter
 		return nil, errors.Wrapf(err, "failed to create node handler factory")
 	}
 
-	nodeExecutor, err := nodes.NewExecutor(ctx, cfg.NodeConfig, store, controller.enqueueWorkflowForNodeUpdates, eventSink,
-		launchPlanActor, launchPlanActor, storage.DataReference(cfg.DefaultRawOutputPrefix), kubeClient,
-		catalogClient, recoveryClient, cfg.LiteralOffloadingConfig, &cfg.EventConfig, cfg.ClusterID, signalClient, nodeHandlerFactory, scope)
+	nodeExecutor, err := nodes.NewExecutor(ctx, cfg.NodeConfig, store, controller.enqueueWorkflowForNodeUpdates, eventSink, storage.DataReference(cfg.DefaultRawOutputPrefix), catalogClient, recoveryClient, cfg.LiteralOffloadingConfig, &cfg.EventConfig, cfg.ClusterID, nodeHandlerFactory, scope)
 	if err != nil {
 		return nil, errors.Wrapf(err, "Failed to create Controller.")
 	}

flytepropeller/pkg/controller/handler.go

@@ -206,7 +206,7 @@ func (p *Propeller) Handle(ctx context.Context, namespace, name string) error {
 			logger.Warningf(ctx, "Workflow namespace[%v]/name[%v] Stale.", namespace, name)
 			return nil
 		}
-		logger.Warningf(ctx, "Failed to GetWorkflow, retrying with back-off", fetchErr)
+		logger.Warningf(ctx, "Failed to GetWorkflow, retrying with back-off: %v", fetchErr)
 		return fetchErr
 	}
 

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

@@ -23,12 +23,10 @@ import (
 	"github.com/flyteorg/flyte/flytepropeller/pkg/controller/nodes"
 	"github.com/flyteorg/flyte/flytepropeller/pkg/controller/nodes/catalog"
 	"github.com/flyteorg/flyte/flytepropeller/pkg/controller/nodes/errors"
-	gatemocks "github.com/flyteorg/flyte/flytepropeller/pkg/controller/nodes/gate/mocks"
 	"github.com/flyteorg/flyte/flytepropeller/pkg/controller/nodes/handler"
 	"github.com/flyteorg/flyte/flytepropeller/pkg/controller/nodes/interfaces"
 	"github.com/flyteorg/flyte/flytepropeller/pkg/controller/nodes/interfaces/mocks"
 	recoverymocks "github.com/flyteorg/flyte/flytepropeller/pkg/controller/nodes/recovery/mocks"
-	"github.com/flyteorg/flyte/flytepropeller/pkg/controller/nodes/subworkflow/launchplan"
 	"github.com/flyteorg/flyte/flytestdlib/bitarray"
 	"github.com/flyteorg/flyte/flytestdlib/contextutils"
 	"github.com/flyteorg/flyte/flytestdlib/promutils"
@@ -52,22 +50,18 @@ var (
 
 func createArrayNodeHandler(ctx context.Context, t *testing.T, nodeHandler interfaces.NodeHandler, dataStore *storage.DataStore, scope promutils.Scope) (interfaces.NodeHandler, error) {
 	// mock components
-	adminClient := launchplan.NewFailFastLaunchPlanExecutor()
 	enqueueWorkflowFunc := func(workflowID v1alpha1.WorkflowID) {}
 	eventConfig := &config.EventConfig{ErrorOnAlreadyExists: true}
 	offloadingConfig := config.LiteralOffloadingConfig{Enabled: false}
 	literalOffloadingConfig := config.LiteralOffloadingConfig{Enabled: true, MinSizeInMBForOffloading: 1024, MaxSizeInMBForOffloading: 1024 * 1024}
 	mockEventSink := eventmocks.NewMockEventSink()
 	mockHandlerFactory := &mocks.HandlerFactory{}
 	mockHandlerFactory.EXPECT().GetHandler(mock.Anything).Return(nodeHandler, nil)
-	mockKubeClient := execmocks.NewFakeKubeClient()
 	mockRecoveryClient := &recoverymocks.Client{}
-	mockSignalClient := &gatemocks.SignalServiceClient{}
 	noopCatalogClient := catalog.NOOPCatalog{}
 
 	// create node executor
-	nodeExecutor, err := nodes.NewExecutor(ctx, config.GetConfig().NodeConfig, dataStore, enqueueWorkflowFunc, mockEventSink, adminClient,
-		adminClient, "s3://bucket/", mockKubeClient, noopCatalogClient, mockRecoveryClient, offloadingConfig, eventConfig, "clusterID", mockSignalClient, mockHandlerFactory, scope)
+	nodeExecutor, err := nodes.NewExecutor(ctx, config.GetConfig().NodeConfig, dataStore, enqueueWorkflowFunc, mockEventSink, "s3://bucket/", noopCatalogClient, mockRecoveryClient, offloadingConfig, eventConfig, "clusterID", mockHandlerFactory, scope)
 	assert.NoError(t, err)
 
 	// return ArrayNodeHandler

flytepropeller/pkg/controller/nodes/attr_path_resolver.go

@@ -17,7 +17,7 @@ import (
 // resolveAttrPathInPromise resolves the literal with attribute path
 // If the promise is chained with attributes (e.g. promise.a["b"][0]), then we need to resolve the promise
 func resolveAttrPathInPromise(ctx context.Context, datastore *storage.DataStore, nodeID string, literal *core.Literal, bindAttrPath []*core.PromiseAttribute) (*core.Literal, error) {
-	var currVal *core.Literal = literal
+	var currVal = literal
 	var tmpVal *core.Literal
 	var err error
 	var exist bool

flytepropeller/pkg/controller/nodes/branch/evaluator.go

@@ -15,7 +15,6 @@ import (
 const ErrorCodeUserProvidedError = "UserProvidedError"
 const ErrorCodeMalformedBranch = "MalformedBranchUserError"
 const ErrorCodeCompilerError = "CompilerError"
-const ErrorCodeFailedFetchOutputs = "FailedFetchOutputs"
 
 func EvaluateComparison(expr *core.ComparisonExpression, nodeInputs *core.LiteralMap) (bool, error) {
 	var lValue *core.Literal
@@ -94,7 +93,7 @@ func EvaluateIfBlock(block v1alpha1.ExecutableIfBlock, nodeInputs *core.LiteralM
 // Decides the branch to be taken, returns the nodeId of the selected node or an error
 // The branchNode is marked as success. This is used by downstream node to determine if it can be executed
 // All downstream nodes are marked as skipped
-func DecideBranch(ctx context.Context, nl executors.NodeLookup, nodeID v1alpha1.NodeID, node v1alpha1.ExecutableBranchNode, nodeInputs *core.LiteralMap) (*v1alpha1.NodeID, error) {
+func DecideBranch(ctx context.Context, nl executors.NodeLookup, node v1alpha1.ExecutableBranchNode, nodeInputs *core.LiteralMap) (*v1alpha1.NodeID, error) {
 	var selectedNodeID *v1alpha1.NodeID
 	var skippedNodeIds []*v1alpha1.NodeID
 	var err error

flytepropeller/pkg/controller/nodes/branch/evaluator_test.go

@@ -355,7 +355,7 @@ func TestDecideBranch(t *testing.T) {
 			DataReferenceConstructor: dataStore,
 		}
 		branchNode := &v1alpha1.BranchNodeSpec{}
-		b, err := DecideBranch(ctx, w, "n1", branchNode, nil)
+		b, err := DecideBranch(ctx, w, branchNode, nil)
 		assert.Error(t, err)
 		assert.Nil(t, b)
 	})
@@ -381,7 +381,7 @@ func TestDecideBranch(t *testing.T) {
 				ThenNode: nil,
 			},
 		}
-		b, err := DecideBranch(ctx, w, "n1", branchNode, inputs)
+		b, err := DecideBranch(ctx, w, branchNode, inputs)
 		assert.Error(t, err)
 		assert.Nil(t, b)
 		e, ok := errors.GetErrorCode(err)
@@ -416,7 +416,7 @@ func TestDecideBranch(t *testing.T) {
 				ThenNode: &n1,
 			},
 		}
-		b, err := DecideBranch(ctx, w, "n1", branchNode, inputs)
+		b, err := DecideBranch(ctx, w, branchNode, inputs)
 		assert.NoError(t, err)
 		assert.NotNil(t, b)
 		assert.Equal(t, n1, *b)
@@ -466,7 +466,7 @@ func TestDecideBranch(t *testing.T) {
 				},
 			},
 		}
-		b, err := DecideBranch(ctx, w, "n", branchNode, inputs)
+		b, err := DecideBranch(ctx, w, branchNode, inputs)
 		assert.NoError(t, err)
 		assert.NotNil(t, b)
 		assert.Equal(t, n1, *b)
@@ -517,7 +517,7 @@ func TestDecideBranch(t *testing.T) {
 				},
 			},
 		}
-		b, err := DecideBranch(ctx, w, "n", branchNode, inputs)
+		b, err := DecideBranch(ctx, w, branchNode, inputs)
 		assert.NoError(t, err)
 		assert.NotNil(t, b)
 		assert.Equal(t, n2, *b)
@@ -570,7 +570,7 @@ func TestDecideBranch(t *testing.T) {
 			},
 			Else: &n3,
 		}
-		b, err := DecideBranch(ctx, w, "n", branchNode, inputs)
+		b, err := DecideBranch(ctx, w, branchNode, inputs)
 		assert.NoError(t, err)
 		assert.NotNil(t, b)
 		assert.Equal(t, n3, *b)
@@ -620,7 +620,7 @@ func TestDecideBranch(t *testing.T) {
 			},
 			Else: &n3,
 		}
-		b, err := DecideBranch(ctx, w, "n", branchNode, inputs)
+		b, err := DecideBranch(ctx, w, branchNode, inputs)
 		assert.Error(t, err)
 		assert.Nil(t, b)
 		ec, ok := errors.GetErrorCode(err)
@@ -677,7 +677,7 @@ func TestDecideBranch(t *testing.T) {
 			},
 		}
 
-		b, err := DecideBranch(ctx, w, "n", branchNode, inputs)
+		b, err := DecideBranch(ctx, w, branchNode, inputs)
 		assert.Error(t, err)
 		assert.Nil(t, b)
 		ec, ok := errors.GetErrorCode(err)

flytepropeller/pkg/controller/nodes/branch/handler.go

@@ -45,7 +45,7 @@ func (b *branchHandler) HandleBranchNode(ctx context.Context, branchNode v1alpha
 			errMsg := fmt.Sprintf("Failed to read input. Error [%s]", err)
 			return handler.DoTransition(handler.TransitionTypeEphemeral, handler.PhaseInfoFailure(core.ExecutionError_SYSTEM, errors.RuntimeExecutionError, errMsg, nil)), nil
 		}
-		finalNodeID, err := DecideBranch(ctx, nl, nCtx.NodeID(), branchNode, nodeInputs)
+		finalNodeID, err := DecideBranch(ctx, nl, branchNode, nodeInputs)
 		if err != nil {
 			ec, ok := stdErrors.GetErrorCode(err)
 			if ok {

flytepropeller/pkg/controller/nodes/catalog/datacatalog/datacatalog.go

@@ -151,7 +151,7 @@ func (m *CatalogClient) Get(ctx context.Context, key catalog.Key) (catalog.Entry
 func (m *CatalogClient) createDataset(ctx context.Context, key catalog.Key, metadata *datacatalog.Metadata) (*datacatalog.DatasetID, error) {
 	datasetID, err := GenerateDatasetIDForTask(ctx, key)
 	if err != nil {
-		logger.Errorf(ctx, "DataCatalog failed to generate dataset for ID: %s, err: %s", key.Identifier, err)
+		logger.Errorf(ctx, "DataCatalog failed to generate dataset for ID: %v, err: %s", key.Identifier, err)
 		return nil, err
 	}
 
@@ -164,7 +164,7 @@ func (m *CatalogClient) createDataset(ctx context.Context, key catalog.Key, meta
 	if err != nil {
 		logger.Debugf(ctx, "Create dataset %v return err %v", datasetID, err)
 		if status.Code(err) == codes.AlreadyExists {
-			logger.Debugf(ctx, "Create Dataset for ID %s already exists", key.Identifier)
+			logger.Debugf(ctx, "Create Dataset for ID %v already exists", key.Identifier)
 		} else {
 			logger.Errorf(ctx, "Unable to create dataset %s, err: %s", datasetID, err)
 			return nil, err