clear lp cache on workflow node finalize

Signed-off-by: Paul Dittamo <[email protected]>

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

@@ -33,7 +33,7 @@ func newMetrics(scope promutils.Scope) metrics {
 }
 
 func (w *workflowNodeHandler) FinalizeRequired() bool {
-	return false
+	return true
 }
 
 func (w *workflowNodeHandler) Setup(_ context.Context, _ interfaces.SetupContext) error {
@@ -120,8 +120,12 @@ func (w *workflowNodeHandler) Abort(ctx context.Context, nCtx interfaces.NodeExe
 	return nil
 }
 
-func (w *workflowNodeHandler) Finalize(ctx context.Context, _ interfaces.NodeExecutionContext) error {
-	logger.Warnf(ctx, "Subworkflow finalize invoked. Nothing to be done")
+func (w *workflowNodeHandler) Finalize(ctx context.Context, nCtx interfaces.NodeExecutionContext) error {
+	logger.Warnf(ctx, "Subworkflow finalize invoked. Clearing up the cache")
+	wfNode := nCtx.Node().GetWorkflowNode()
+	if wfNode.GetLaunchPlanRefID() != nil {
+		return w.lpHandler.Finalize(ctx, nCtx)
+	}
 	return nil
 }
 

flytepropeller/pkg/controller/nodes/subworkflow/launchplan.go

@@ -229,3 +229,20 @@ func (l *launchPlanHandler) HandleAbort(ctx context.Context, nCtx interfaces.Nod
 	}
 	return l.launchPlan.Kill(ctx, childID, fmt.Sprintf("cascading abort as parent execution id [%s] aborted, reason [%s]", nCtx.ExecutionContext().GetName(), reason))
 }
+
+func (l *launchPlanHandler) Finalize(ctx context.Context, nCtx interfaces.NodeExecutionContext) error {
+	parentNodeExecutionID, err := getParentNodeExecutionID(nCtx)
+	if err != nil {
+		return err
+	}
+	childID, err := GetChildWorkflowExecutionIDForExecution(
+		parentNodeExecutionID,
+		nCtx,
+	)
+	if err != nil {
+		// THIS SHOULD NEVER HAPPEN
+		return err
+	}
+
+	return l.launchPlan.Finalize(ctx, childID)
+}

flytepropeller/pkg/controller/nodes/subworkflow/launchplan/admin.go

@@ -224,6 +224,10 @@ func (a *adminLaunchPlanExecutor) Initialize(ctx context.Context) error {
 	return a.cache.Start(ctx)
 }
 
+func (a *adminLaunchPlanExecutor) Finalize(ctx context.Context, executionID *core.WorkflowExecutionIdentifier) error {
+	return a.cache.DeleteDelayed(executionID.String())
+}
+
 func (a *adminLaunchPlanExecutor) syncItem(ctx context.Context, batch cache.Batch) (
 	resp []cache.ItemSyncResponse, err error) {
 	resp = make([]cache.ItemSyncResponse, 0, len(batch))

flytepropeller/pkg/controller/nodes/subworkflow/launchplan/launchplan.go

@@ -46,6 +46,9 @@ type Executor interface {
 
 	// Initialize initializes Executor.
 	Initialize(ctx context.Context) error
+
+	// Finalize clears the cache of the LaunchPlan execution
+	Finalize(ctx context.Context, executionID *core.WorkflowExecutionIdentifier) error
 }
 
 type Reader interface {