Improve logging when operator failed before generating artifact (#1252)

src/golang/lib/engine/aq_engine.go

@@ -1005,6 +1005,10 @@ func (eng *aqEngine) execute(
 			// and check operators with warning severity.
 			if execState.HasBlockingFailure() {
 				log.Infof("Stopping execution of operator %v", op.ID())
+				if execState.Error != nil {
+					log.Infof("Reason: %s", execState.Error.Message())
+				}
+
 				for id, dagOp := range workflowDag.Operators() {
 					log.Infof("Checking status of operator %v", id)
 					// Skip if this operator has already been completed or is in progress.

src/golang/lib/workflow/artifact/artifact.go

@@ -207,12 +207,14 @@ func (a *ArtifactImpl) updateArtifactResultAfterComputation(
 	execState *shared.ExecutionState,
 ) {
 	changes := map[string]interface{}{
+		models.ArtifactResultMetadata:  nil,
 		models.ArtifactResultStatus:    execState.Status,
 		models.ArtifactResultExecState: execState,
-		models.ArtifactResultMetadata:  nil,
 	}
 
-	if a.Computed(ctx) {
+	metadataExists := utils.ObjectExistsInStorage(ctx, a.storageConfig, a.execPaths.ArtifactMetadataPath)
+
+	if a.Computed(ctx) && metadataExists {
 		var artifactResultMetadata shared.ArtifactResultMetadata
 		err := utils.ReadFromStorage(
 			ctx,
@@ -322,6 +324,11 @@ func (a *ArtifactImpl) GetMetadata(ctx context.Context) (*shared.ArtifactResultM
 			return nil, nil
 		}
 
+		// If the path is not available, we assume the data is not available.
+		if !utils.ObjectExistsInStorage(ctx, a.storageConfig, a.execPaths.ArtifactMetadataPath) {
+			return nil, nil
+		}
+
 		var metadata shared.ArtifactResultMetadata
 		err := utils.ReadFromStorage(ctx, a.storageConfig, a.execPaths.ArtifactMetadataPath, &metadata)
 		if err != nil {

src/golang/lib/workflow/utils/storage.go

@@ -24,6 +24,10 @@ func CleanupStorageFiles(ctx context.Context, storageConfig *shared.StorageConfi
 }
 
 func ObjectExistsInStorage(ctx context.Context, storageConfig *shared.StorageConfig, path string) bool {
+	if path == "" {
+		return false
+	}
+
 	return storage.NewStorage(storageConfig).Exists(ctx, path)
 }