fix(backend): Guard against panic when MLMD execution publishing fails (#12203)

* Guard against panic when MLMD execution publishing fails

This also removes the defer that obscures the original error being
returned.

Signed-off-by: mprahl <[email protected]>

* Increase the CI timeout on deploying KFP

With seaweedfs, the image can take a long time to pull. It's better to
wait longer than to fail the workflow.

Signed-off-by: mprahl <[email protected]>

---------

Signed-off-by: mprahl <[email protected]>

Author: mprahl

.github/resources/scripts/kfp-readiness/wait_for_pods.py

@@ -77,7 +77,7 @@ def print_get_pods():
         print(f"An error occurred while running kubectl get pods: {e.stderr}")
 
 
-def check_pods(calm_time=10, timeout=600, retries_after_ready=5):
+def check_pods(calm_time=10, timeout=900, retries_after_ready=5):
     start_time = time.time()
     stable_count = 0
     previous_statuses = {}

backend/src/v2/component/launcher_v2.go

@@ -171,6 +171,11 @@ func (l *LauncherV2) Execute(ctx context.Context) (err error) {
 	var outputArtifacts []*metadata.OutputArtifact
 	status := pb.Execution_FAILED
 	defer func() {
+		if execution == nil {
+			glog.Errorf("Skipping publish since execution is nil. Original err is: %v", err)
+			return
+		}
+
 		if perr := l.publish(ctx, execution, executorOutput, outputArtifacts, status); perr != nil {
 			if err != nil {
 				err = fmt.Errorf("failed to publish execution with error %s after execution failed: %s", perr.Error(), err.Error())
@@ -312,17 +317,25 @@ func (l *LauncherV2) publish(
 	outputArtifacts []*metadata.OutputArtifact,
 	status pb.Execution_State,
 ) (err error) {
-	defer func() {
-		if err != nil {
-			err = fmt.Errorf("failed to publish results to ML Metadata: %w", err)
-		}
-	}()
-	outputParameters := executorOutput.GetParameterValues()
+	if execution == nil {
+		return fmt.Errorf("failed to publish results to ML Metadata: execution is nil")
+	}
+
+	var outputParameters map[string]*structpb.Value
+	if executorOutput != nil {
+		outputParameters = executorOutput.GetParameterValues()
+	}
+
 	// TODO(Bobgy): upload output artifacts.
 	// TODO(Bobgy): when adding artifacts, we will need execution.pipeline to be non-nil, because we need
 	// to publish output artifacts to the context too.
 	// return l.metadataClient.PublishExecution(ctx, execution, outputParameters, outputArtifacts, pb.Execution_COMPLETE)
-	return l.clientManager.MetadataClient().PublishExecution(ctx, execution, outputParameters, outputArtifacts, status)
+	err = l.clientManager.MetadataClient().PublishExecution(ctx, execution, outputParameters, outputArtifacts, status)
+	if err != nil {
+		return fmt.Errorf("failed to publish results to ML Metadata: %w", err)
+	}
+
+	return nil
 }
 
 // executeV2 handles placeholder substitution for inputs, calls execute to