Instrument ArrayNode

cmd/kubectl-flyte/cmd/testdata/array-node-cache-serialize.yaml.golden

@@ -0,0 +1,91 @@
+tasks:
+- container:
+    args:
+    - "pyflyte-fast-execute"
+    - "--additional-distribution"
+    - "s3://my-s3-bucket/flytesnacks/development/SMJBJX7BQJ6MCOABLKQT5VZXVY======/script_mode.tar.gz"
+    - "--dest-dir"
+    - "/root"
+    - "--"
+    - "pyflyte-map-execute"
+    - "--inputs"
+    - "{{.input}}"
+    - "--output-prefix"
+    - "{{.outputPrefix}}"
+    - "--raw-output-data-prefix"
+    - "{{.rawOutputDataPrefix}}"
+    - "--checkpoint-path"
+    - "{{.checkpointOutputPrefix}}"
+    - "--prev-checkpoint"
+    - "{{.prevCheckpointPrefix}}"
+    - "--resolver"
+    - "MapTaskResolver"
+    - "--"
+    - "vars"
+    - ""
+    - "resolver"
+    - "flytekit.core.python_auto_container.default_task_resolver"
+    - "task-module"
+    - "map-task"
+    - "task-name"
+    - "a_mappable_task"
+    image: "array-node:ee1ba227aa95447d04bb1761691b4d97749642dc"
+    resources:
+      limits:
+        - name: 1
+          value: "1"
+        - name: 3
+          value: "500Mi"
+      requests:
+        - name: 1
+          value: "1"
+        - name: 3
+          value: "300Mi"
+  id:
+    name: task-1
+    project: flytesnacks
+    domain: development
+  metadata:
+    discoverable: true
+    discovery_version: "1.0"
+    cache_serializable: true
+  interface:
+    inputs:
+      variables:
+        a:
+          type:
+            simple: INTEGER
+    outputs:
+      variables:
+        o0:
+          type:
+            simple: STRING
+workflow:
+  id:
+    name: workflow-with-array-node
+  interface:
+    inputs:
+      variables:
+        x:
+          type:
+            collectionType:
+              simple: INTEGER
+  nodes:
+  - id: node-1
+    inputs:
+    - binding:
+        promise:
+          node_id: start-node
+          var: x
+      var: a
+    arrayNode:
+      parallelism: 0
+      node:
+        metadata:
+          retries:
+            retries: 3
+        taskNode:
+          referenceId:
+            name: task-1
+            project: flytesnacks
+            domain: development

cmd/kubectl-flyte/cmd/testdata/array-node-cache.yaml.golden

@@ -0,0 +1,91 @@
+tasks:
+- container:
+    args:
+    - "pyflyte-fast-execute"
+    - "--additional-distribution"
+    - "s3://my-s3-bucket/flytesnacks/development/SMJBJX7BQJ6MCOABLKQT5VZXVY======/script_mode.tar.gz"
+    - "--dest-dir"
+    - "/root"
+    - "--"
+    - "pyflyte-map-execute"
+    - "--inputs"
+    - "{{.input}}"
+    - "--output-prefix"
+    - "{{.outputPrefix}}"
+    - "--raw-output-data-prefix"
+    - "{{.rawOutputDataPrefix}}"
+    - "--checkpoint-path"
+    - "{{.checkpointOutputPrefix}}"
+    - "--prev-checkpoint"
+    - "{{.prevCheckpointPrefix}}"
+    - "--resolver"
+    - "MapTaskResolver"
+    - "--"
+    - "vars"
+    - ""
+    - "resolver"
+    - "flytekit.core.python_auto_container.default_task_resolver"
+    - "task-module"
+    - "map-task"
+    - "task-name"
+    - "a_mappable_task"
+    image: "array-node:ee1ba227aa95447d04bb1761691b4d97749642dc"
+    resources:
+      limits:
+        - name: 1
+          value: "1"
+        - name: 3
+          value: "500Mi"
+      requests:
+        - name: 1
+          value: "1"
+        - name: 3
+          value: "300Mi"
+  id:
+    name: task-1
+    project: flytesnacks
+    domain: development
+  metadata:
+    discoverable: true
+    discovery_version: "1.0"
+    cache_serializable: false
+  interface:
+    inputs:
+      variables:
+        a:
+          type:
+            simple: INTEGER
+    outputs:
+      variables:
+        o0:
+          type:
+            simple: STRING
+workflow:
+  id:
+    name: workflow-with-array-node
+  interface:
+    inputs:
+      variables:
+        x:
+          type:
+            collectionType:
+              simple: INTEGER
+  nodes:
+  - id: node-1
+    inputs:
+    - binding:
+        promise:
+          node_id: start-node
+          var: x
+      var: a
+    arrayNode:
+      parallelism: 1
+      node:
+        metadata:
+          retries:
+            retries: 3
+        taskNode:
+          referenceId:
+            name: task-1
+            project: flytesnacks
+            domain: development

cmd/kubectl-flyte/cmd/testdata/array-node-inputs.yaml.golden

@@ -0,0 +1,13 @@
+literals:
+  "x":
+    collection:
+      literals:
+      - scalar:
+          primitive:
+            integer: "1"
+      - scalar:
+          primitive:
+            integer: "2"
+      - scalar:
+          primitive:
+            integer: "3"

cmd/kubectl-flyte/cmd/testdata/array-node.yaml.golden

@@ -0,0 +1,86 @@
+tasks:
+- container:
+    args:
+    - "pyflyte-fast-execute"
+    - "--additional-distribution"
+    - "s3://my-s3-bucket/flytesnacks/development/SMJBJX7BQJ6MCOABLKQT5VZXVY======/script_mode.tar.gz"
+    - "--dest-dir"
+    - "/root"
+    - "--"
+    - "pyflyte-map-execute"
+    - "--inputs"
+    - "{{.input}}"
+    - "--output-prefix"
+    - "{{.outputPrefix}}"
+    - "--raw-output-data-prefix"
+    - "{{.rawOutputDataPrefix}}"
+    - "--checkpoint-path"
+    - "{{.checkpointOutputPrefix}}"
+    - "--prev-checkpoint"
+    - "{{.prevCheckpointPrefix}}"
+    - "--resolver"
+    - "MapTaskResolver"
+    - "--"
+    - "vars"
+    - ""
+    - "resolver"
+    - "flytekit.core.python_auto_container.default_task_resolver"
+    - "task-module"
+    - "map-task"
+    - "task-name"
+    - "a_mappable_task"
+    image: "array-node:ee1ba227aa95447d04bb1761691b4d97749642dc"
+    resources:
+      limits:
+        - name: 1
+          value: "1"
+        - name: 3
+          value: "500Mi"
+      requests:
+        - name: 1
+          value: "1"
+        - name: 3
+          value: "300Mi"
+  id:
+    name: task-1
+  metadata:
+    discoverable: false
+    cache_serializable: false
+  interface:
+    inputs:
+      variables:
+        a:
+          type:
+            simple: INTEGER
+    outputs:
+      variables:
+        x:
+          type:
+            simple: STRING
+workflow:
+  id:
+    name: workflow-with-array-node
+  interface:
+    inputs:
+      variables:
+        x:
+          type:
+            collectionType:
+              simple: INTEGER
+  nodes:
+  - id: node-1
+    inputs:
+    - binding:
+        promise:
+          node_id: start-node
+          var: x
+      var: a
+    arrayNode:
+      parallelism: 1
+      node:
+        metadata:
+          retries:
+            retries: 3
+        taskNode:
+          referenceId:
+            name: task-1

events/event_recorder.go

@@ -13,7 +13,7 @@ import (
 	"github.com/golang/protobuf/proto"
 )
 
-const maxErrorMessageLength = 104857600 //100KB
+const MaxErrorMessageLength = 104857600 //100KB
 const truncationIndicator = "... <Message Truncated> ..."
 
 type recordingMetrics struct {
@@ -60,23 +60,23 @@ func (r *eventRecorder) sinkEvent(ctx context.Context, event proto.Message) erro
 
 func (r *eventRecorder) RecordNodeEvent(ctx context.Context, e *event.NodeExecutionEvent) error {
 	if err, ok := e.GetOutputResult().(*event.NodeExecutionEvent_Error); ok {
-		truncateErrorMessage(err.Error, maxErrorMessageLength)
+		truncateErrorMessage(err.Error, MaxErrorMessageLength)
 	}
 
 	return r.sinkEvent(ctx, e)
 }
 
 func (r *eventRecorder) RecordTaskEvent(ctx context.Context, e *event.TaskExecutionEvent) error {
 	if err, ok := e.GetOutputResult().(*event.TaskExecutionEvent_Error); ok {
-		truncateErrorMessage(err.Error, maxErrorMessageLength)
+		truncateErrorMessage(err.Error, MaxErrorMessageLength)
 	}
 
 	return r.sinkEvent(ctx, e)
 }
 
 func (r *eventRecorder) RecordWorkflowEvent(ctx context.Context, e *event.WorkflowExecutionEvent) error {
 	if err, ok := e.GetOutputResult().(*event.WorkflowExecutionEvent_Error); ok {
-		truncateErrorMessage(err.Error, maxErrorMessageLength)
+		truncateErrorMessage(err.Error, MaxErrorMessageLength)
 	}
 
 	return r.sinkEvent(ctx, e)

go.mod

@@ -146,3 +146,5 @@ require (
 )
 
 replace github.com/aws/amazon-sagemaker-operator-for-k8s => github.com/aws/amazon-sagemaker-operator-for-k8s v1.0.1-0.20210303003444-0fb33b1fd49d
+
+replace github.com/flyteorg/flyteidl => github.com/flyteorg/flyteidl v1.5.11-0.20230614183933-d56d4d37bf34

go.sum

@@ -242,8 +242,8 @@ github.com/fatih/color v1.7.0/go.mod h1:Zm6kSWBoL9eyXnKyktHP6abPY2pDugNf5Kwzbycv
 github.com/fatih/color v1.13.0 h1:8LOYc1KYPPmyKMuN8QV2DNRWNbLo6LZ0iLs8+mlH53w=
 github.com/fatih/color v1.13.0/go.mod h1:kLAiJbzzSOZDVNGyDpeOxJ47H46qBXwg5ILebYFFOfk=
 github.com/felixge/httpsnoop v1.0.1/go.mod h1:m8KPJKqk1gH5J9DgRY2ASl2lWCfGKXixSwevea8zH2U=
-github.com/flyteorg/flyteidl v1.5.10 h1:SHeiaWRt8EAVuFsat+BJswtc07HTZ4DqhfTEYSm621k=
-github.com/flyteorg/flyteidl v1.5.10/go.mod h1:EtE/muM2lHHgBabjYcxqe9TWeJSL0kXwbI0RgVwI4Og=
+github.com/flyteorg/flyteidl v1.5.11-0.20230614183933-d56d4d37bf34 h1:Gj5UKqJU+ozeTeYAvDWHiF4HSVufHW1W1ecymFfbbis=
+github.com/flyteorg/flyteidl v1.5.11-0.20230614183933-d56d4d37bf34/go.mod h1:EtE/muM2lHHgBabjYcxqe9TWeJSL0kXwbI0RgVwI4Og=
 github.com/flyteorg/flyteplugins v1.1.8 h1:UVYdqDdcIqz2JIso+m3MsaPSsTZJZyZQ6Eg7nhX9r/Y=
 github.com/flyteorg/flyteplugins v1.1.8/go.mod h1:sRxeatEOHq1b9bTxTRNcwoIkVTAVN9dTz8toXkfcz2E=
 github.com/flyteorg/flytestdlib v1.0.20 h1:BrCQMlpdrFAPlADFJvCyn7gm+37df9WGYqLEB1mOlCQ=

pkg/apis/flyteworkflow/v1alpha1/array.go

@@ -0,0 +1,24 @@
+package v1alpha1
+
+type ArrayNodeSpec struct {
+	SubNodeSpec     *NodeSpec
+	Parallelism     uint32
+	MinSuccesses    *uint32
+	MinSuccessRatio *float32
+}
+
+func (a *ArrayNodeSpec) GetSubNodeSpec() *NodeSpec {
+	return a.SubNodeSpec
+}
+
+func (a *ArrayNodeSpec) GetParallelism() uint32 {
+	return a.Parallelism
+}
+
+func (a *ArrayNodeSpec) GetMinSuccesses() *uint32 {
+	return a.MinSuccesses
+}
+
+func (a *ArrayNodeSpec) GetMinSuccessRatio() *float32 {
+	return a.MinSuccessRatio
+}