Wf new plugin iface

Gopkg.toml

@@ -58,8 +58,9 @@ required = [
 
 [[constraint]]
   name = "github.com/lyft/flyteplugins"
+  revision = "798101a85483ce46f727f6f1129890eefd358e72"
   source = "https://github.com/lyft/flyteplugins"
-  version = "^0.1.3"
+  #version = "^0.1.3"
 
 [[override]]
   name = "github.com/lyft/flytestdlib"

cmd/controller/cmd/root.go

@@ -16,9 +16,6 @@ import (
 
 	config2 "github.com/lyft/flytepropeller/pkg/controller/config"
 
-	"github.com/lyft/flyteplugins/go/tasks"
-
-	"github.com/lyft/flyteplugins/go/tasks/v1/flytek8s"
 	v1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 
 	"github.com/lyft/flytestdlib/config/viper"
@@ -193,15 +190,6 @@ func executeRootCmd(cfg *config2.Config) {
 		limitNamespace = cfg.LimitNamespace
 	}
 
-	err = flytek8s.Initialize(ctx, limitNamespace, cfg.DownstreamEval.Duration)
-	if err != nil {
-		logger.Panicf(ctx, "Failed to initialize k8s plugins. Error: %v", err)
-	}
-
-	if err := tasks.Load(ctx); err != nil {
-		logger.Fatalf(ctx, "Failed to load task plugins. [%v]", err)
-	}
-
 	mgr, err := manager.New(kubecfg, manager.Options{
 		Namespace:  limitNamespace,
 		SyncPeriod: &cfg.DownstreamEval.Duration,

cmd/controller/main.go

@@ -1,6 +1,11 @@
 package main
 
 import (
+	// _ "github.com/lyft/flyteplugins/go/tasks/plugins/array/awsbatch"
+	_ "github.com/lyft/flyteplugins/go/tasks/plugins/array/k8s"
+	_ "github.com/lyft/flyteplugins/go/tasks/plugins/hive"
+	_ "github.com/lyft/flyteplugins/go/tasks/plugins/k8s/container"
+
 	"github.com/lyft/flytepropeller/cmd/controller/cmd"
 )
 

cmd/kubectl-flyte/cmd/printers/node.go

@@ -68,7 +68,7 @@ func (p NodeStatusPrinter) NodeInfo(wName string, node v1alpha1.BaseNode, nodeSt
 
 func (p NodePrinter) BranchNodeInfo(node v1alpha1.ExecutableNode, nodeStatus v1alpha1.ExecutableNodeStatus) []string {
 	info := p.BaseNodeInfo(node, nodeStatus)
-	branchStatus := nodeStatus.GetOrCreateBranchStatus()
+	branchStatus := nodeStatus.GetBranchStatus()
 	info = append(info, branchStatus.GetPhase().String())
 	if branchStatus.GetFinalizedNode() != nil {
 		info = append(info, *branchStatus.GetFinalizedNode())

config.yaml

@@ -22,13 +22,14 @@ propeller:
       capacity: 1000
   kube-config: "$HOME/.kube/config"
   publish-k8s-events: true
+tasks:
+  task-plugins:
+    enabled-plugins:
+      - container
+      - k8s-array
+      - qubole-hive-executor
 # Sample plugins config
 plugins:
-  # Set of enabled plugins at root level
-  enabled-plugins:
-    - container
-    - waitable
-    - K8S-ARRAY
   # All k8s plugins default configuration
   k8s:
     inject-finalizer: true
@@ -46,24 +47,6 @@ plugins:
       - FLYTE_AWS_ENDPOINT: "http://minio.flyte:9000"
       - FLYTE_AWS_ACCESS_KEY_ID: minio
       - FLYTE_AWS_SECRET_ACCESS_KEY: miniostorage
-  # Spark Plugin configuration
-  spark:
-    spark-config-default:
-      - spark.hadoop.mapreduce.fileoutputcommitter.algorithm.version: "2"
-      - spark.kubernetes.allocation.batch.size: "50"
-      - spark.hadoop.fs.s3a.acl.default: "BucketOwnerFullControl"
-      - spark.hadoop.fs.s3n.impl: "org.apache.hadoop.fs.s3a.S3AFileSystem"
-      - spark.hadoop.fs.AbstractFileSystem.s3n.impl: "org.apache.hadoop.fs.s3a.S3A"
-      - spark.hadoop.fs.s3.impl: "org.apache.hadoop.fs.s3a.S3AFileSystem"
-      - spark.hadoop.fs.AbstractFileSystem.s3.impl: "org.apache.hadoop.fs.s3a.S3A"
-      - spark.hadoop.fs.s3a.impl: "org.apache.hadoop.fs.s3a.S3AFileSystem"
-      - spark.hadoop.fs.AbstractFileSystem.s3a.impl: "org.apache.hadoop.fs.s3a.S3A"
-      - spark.hadoop.fs.s3a.multipart.threshold: "536870912"
-      - spark.blacklist.enabled: "true"
-      - spark.blacklist.timeout: "5m"
-  # Waitable plugin configuration
-  waitable:
-    console-uri: http://localhost:30081/console
   # Logging configuration
   logs:
     kubernetes-enabled: true
@@ -89,11 +72,9 @@ admin:
   endpoint: localhost:8089
   insecure: true
 catalog-cache:
-  type: catalog
+  type: noop
   endpoint: datacatalog:8089
   insecure: true
-errors:
-  show-source: true
 logger:
   level: 4
   show-source: true

pkg/apis/flyteworkflow/v1alpha1/iface.go

@@ -8,7 +8,6 @@ import (
 	"k8s.io/apimachinery/pkg/types"
 
 	"github.com/lyft/flyteidl/gen/pb-go/flyteidl/core"
-	types2 "github.com/lyft/flyteplugins/go/tasks/v1/types"
 	"github.com/lyft/flytestdlib/storage"
 )
 
@@ -173,7 +172,7 @@ type MutableDynamicNodeStatus interface {
 }
 
 // Interface for Branch node. All the methods are purely read only except for the GetExecutionStatus.
-// Phase returns ExecutableBranchNodeStatus, which permits some mutations
+// p returns ExecutableBranchNodeStatus, which permits some mutations
 type ExecutableBranchNode interface {
 	GetIf() ExecutableIfBlock
 	GetElse() *NodeID
@@ -205,18 +204,19 @@ type MutableNodeStatus interface {
 	SetCached()
 	ResetDirty()
 
+	GetBranchStatus() MutableBranchNodeStatus
 	GetOrCreateBranchStatus() MutableBranchNodeStatus
-	GetOrCreateWorkflowStatus() MutableWorkflowNodeStatus
+	GetWorkflowStatus() MutableWorkflowNodeStatus
 	ClearWorkflowStatus()
 	GetOrCreateTaskStatus() MutableTaskNodeStatus
+	GetTaskStatus() MutableTaskNodeStatus
 	ClearTaskStatus()
-	GetOrCreateSubWorkflowStatus() MutableSubWorkflowNodeStatus
-	ClearSubWorkflowStatus()
 	GetOrCreateDynamicNodeStatus() MutableDynamicNodeStatus
+	GetDynamicNodeStatus() MutableDynamicNodeStatus
 	ClearDynamicNodeStatus()
 }
 
-// Interface for a Node Phase. This provides a mutable API.
+// Interface for a Node p. This provides a mutable API.
 type ExecutableNodeStatus interface {
 	NodeStatusGetter
 	MutableNodeStatus
@@ -233,7 +233,6 @@ type ExecutableNodeStatus interface {
 	GetAttempts() uint32
 	GetWorkflowNodeStatus() ExecutableWorkflowNodeStatus
 	GetTaskNodeStatus() ExecutableTaskNodeStatus
-	GetSubWorkflowNodeStatus() ExecutableSubWorkflowNodeStatus
 
 	IsCached() bool
 	IsDirty() bool
@@ -249,16 +248,18 @@ type MutableSubWorkflowNodeStatus interface {
 }
 
 type ExecutableTaskNodeStatus interface {
-	GetPhase() types2.TaskPhase
+	GetPhase() int
 	GetPhaseVersion() uint32
-	GetCustomState() types2.CustomState
+	GetPluginState() []byte
+	GetPluginStateVersion() uint32
 }
 
 type MutableTaskNodeStatus interface {
 	ExecutableTaskNodeStatus
-	SetPhase(phase types2.TaskPhase)
+	SetPhase(phase int)
 	SetPhaseVersion(version uint32)
-	SetCustomState(state types2.CustomState)
+	SetPluginState([]byte)
+	SetPluginStateVersion(uint32)
 }
 
 // Interface for a Child Workflow Node
@@ -287,7 +288,7 @@ type ExecutableNode interface {
 	GetRetryStrategy() *RetryStrategy
 }
 
-// Interface for the Workflow Phase. This is the mutable portion for a Workflow
+// Interface for the Workflow p. This is the mutable portion for a Workflow
 type ExecutableWorkflowStatus interface {
 	NodeStatusGetter
 	UpdatePhase(p WorkflowPhase, msg string)
@@ -336,7 +337,7 @@ type ExecutableSubWorkflow interface {
 type WorkflowMeta interface {
 	GetExecutionID() ExecutionID
 	GetK8sWorkflowID() types.NamespacedName
-	NewControllerRef() metav1.OwnerReference
+	GetOwnerReference() metav1.OwnerReference
 	GetNamespace() string
 	GetCreationTimestamp() metav1.Time
 	GetAnnotations() map[string]string
@@ -345,9 +346,13 @@ type WorkflowMeta interface {
 	GetServiceAccountName() string
 }
 
+type TaskDetailsGetter interface {
+	GetTask(id TaskID) (ExecutableTask, error)
+}
+
 type WorkflowMetaExtended interface {
 	WorkflowMeta
-	GetTask(id TaskID) (ExecutableTask, error)
+	TaskDetailsGetter
 	FindSubWorkflow(subID WorkflowID) ExecutableSubWorkflow
 	GetExecutionStatus() ExecutableWorkflowStatus
 }