[flytepropeller][flyteadmin] Streaming Decks

Signed-off-by: Future-Outlier <[email protected]>
flyteorg · Oct 2, 2024 · d3cf4b8 · d3cf4b8
1 parent 66ff152
commit d3cf4b8
Showing 1 changed file with 21 additions and 0 deletions.
diff --git a/flytepropeller/pkg/controller/nodes/task/handler.go b/flytepropeller/pkg/controller/nodes/task/handler.go
@@ -523,6 +523,27 @@ func (t Handler) invokePlugin(ctx context.Context, p pluginCore.Plugin, tCtx *ta
 			&event.TaskNodeMetadata{
 				CheckpointUri: tCtx.ow.GetCheckpointPrefix().String(),
 			})
+	// s3://my-s3-bucket/metadata/propeller/flytesnacks-development-a5sj4d6vs9s8r2ltn6hr/n0/data/0/deck.html
+	case pluginCore.PhaseRunning:
+		var deckURI *storage.DataReference
+		if tCtx.ow.GetReader() != nil {
+			exists, err := tCtx.ow.GetReader().DeckExists(ctx)
+			if err != nil {
+				logger.Errorf(ctx, "Failed to check deck file existence. Error: %v", err)
+				return pluginTrns, regErrors.Wrapf(err, "failed to check existence of deck file")
+			} else if exists {
+				deckURIValue := tCtx.ow.GetDeckPath()
+				deckURI = &deckURIValue
+			}
+		}
+
+		deckURIValue := storage.DataReference("s3://my-s3-bucket/metadata/propeller/flytesnacks-development-a5sj4d6vs9s8r2ltn6hr/n0/data/0/deck.html")
+		deckURI = &deckURIValue
+
+		pluginTrns.ObserveSuccess(tCtx.ow.GetOutputPath(), deckURI,
+			&event.TaskNodeMetadata{
+				CheckpointUri: tCtx.ow.GetCheckpointPrefix().String(),
+			})
 	}
 
 	return pluginTrns, nil