Merge 7eae0f0 into 2014630

.github/workflows/checks.yml

@@ -14,15 +14,15 @@ jobs:
     name: Lint
     uses: flyteorg/flytetools/.github/workflows/lint.yml@master
     with:
-      go-version: 1.18
+      go-version: "1.19"
 
   tests:
     name: Unit Tests
     uses: flyteorg/flytetools/.github/workflows/tests.yml@master
+    with:
+      go-version: "1.19"
     secrets:
       FLYTE_BOT_PAT: ${{ secrets.FLYTE_BOT_PAT }}
-    with:
-      go-version: 1.18
 
   docker-build:
     name: Docker Build Images
@@ -42,13 +42,13 @@ jobs:
     uses: flyteorg/flytetools/.github/workflows/integration.yml@master
     with:
       cache_key: ${{ needs.docker-build.outputs.cache_key }}
-      go-version: 1.18
+      go-version: "1.19"
 
   generate:
     name: Check Go Generate
     uses: flyteorg/flytetools/.github/workflows/go_generate.yml@master
     with:
-      go-version: 1.18
+      go-version: "1.19"
 
   bump_version:
     name: Bump Version
@@ -63,7 +63,7 @@ jobs:
     needs: [ bump_version ] # Only to ensure it can successfully build
     uses: flyteorg/flytetools/.github/workflows/goreleaser.yml@master
     with:
-      go-version: 1.18
+      go-version: "1.19"
     secrets:
       FLYTE_BOT_PAT: ${{ secrets.FLYTE_BOT_PAT }}
 

.github/workflows/codeql-analysis.yml

@@ -38,11 +38,11 @@ jobs:
 
     steps:
     - name: Checkout repository
-      uses: actions/checkout@v2
+      uses: actions/checkout@v3
 
-    - uses: actions/setup-go@v2
+    - uses: actions/setup-go@v3
       with:
-        go-version: '1.18'
+        go-version: '1.19'
 
     # Initializes the CodeQL tools for scanning.
     - name: Initialize CodeQL

.github/workflows/upgrade_automation.yml

@@ -16,6 +16,6 @@ jobs:
     uses: flyteorg/flytetools/.github/workflows/flyte_automation.yml@master
     with:
       component: ${{ github.event.inputs.component }}
-      go-version: 1.18
+      go-version: 1.19
     secrets:
       FLYTE_BOT_PAT: ${{ secrets.FLYTE_BOT_PAT }}

go.mod

@@ -1,14 +1,14 @@
 module github.com/flyteorg/flytepropeller
 
-go 1.18
+go 1.19
 
 require (
 	github.com/DiSiqueira/GoTree v1.0.1-0.20180907134536-53a8e837f295
 	github.com/benlaurie/objecthash v0.0.0-20180202135721-d1e3d6079fc1
 	github.com/fatih/color v1.13.0
-	github.com/flyteorg/flyteidl v1.3.4
-	github.com/flyteorg/flyteplugins v1.0.28
-	github.com/flyteorg/flytestdlib v1.0.11
+	github.com/flyteorg/flyteidl v1.3.6
+	github.com/flyteorg/flyteplugins v1.0.34
+	github.com/flyteorg/flytestdlib v1.0.15
 	github.com/ghodss/yaml v1.0.0
 	github.com/go-redis/redis v6.15.7+incompatible
 	github.com/go-test/deep v1.0.7

manager/doc.go

@@ -16,7 +16,7 @@ FlytePropeller Manager is configured at the root of the FlytePropeller configurt
 
 FlytePropeller Manager handles dynamic updates to both the k8s PodTemplate and shard configuration. The k8s PodTemplate resource has an associated resource version which uniquely identifies changes. Additionally, shard configuration modifications may be tracked using a simple hash. Flyte stores these values as annotations on managed FlytePropeller instances. Therefore, if either of there values change the FlytePropeller Manager instance will detect it and perform the necessary deployment updates.
 
-Shard Strategies
+# Shard Strategies
 
 Flyte defines a variety of Shard Strategies for configuring how FlyteWorkflows are sharded. These options may include the shard type (ex. hash, project, or domain) along with the number of shards or the distribution of project / domain IDs over shards.
 

pkg/compiler/builders.go

@@ -7,7 +7,7 @@ import (
 	c "github.com/flyteorg/flytepropeller/pkg/compiler/common"
 )
 
-type flyteTask = core.TaskTemplate
+type flyteTask = core.TaskTemplate //nolint:unused
 type flyteWorkflow = core.CompiledWorkflow
 type flyteNode = core.Node
 

pkg/compiler/workflow_compiler.go

@@ -5,30 +5,30 @@
 // 2) Call CompileWorkflow(...) and make sure it reports no errors.
 // 3) Use one of the transformer packages (e.g. transformer/k8s) to build the final executable workflow.
 //
-//                 +-------------------+
-//                 | start(StartNode)  |
-//                 +-------------------+
-//                   |
-//                   | wf_input
-//                   v
-//  +--------+     +-------------------+
-//  | static | --> | node_1(TaskNode)  |
-//  +--------+     +-------------------+
-//    |              |
-//    |              | x
-//    |              v
-//    |            +-------------------+
-//    +----------> | node_2(TaskNode)  |
-//                 +-------------------+
-//                   |
-//                   | n2_output
-//                   v
-//                 +-------------------+
-//                 |   end(EndNode)    |
-//                 +-------------------+
-//                 +-------------------+
-//                 | Workflow Id: repo |
-//                 +-------------------+
+//	               +-------------------+
+//	               | start(StartNode)  |
+//	               +-------------------+
+//	                 |
+//	                 | wf_input
+//	                 v
+//	+--------+     +-------------------+
+//	| static | --> | node_1(TaskNode)  |
+//	+--------+     +-------------------+
+//	  |              |
+//	  |              | x
+//	  |              v
+//	  |            +-------------------+
+//	  +----------> | node_2(TaskNode)  |
+//	               +-------------------+
+//	                 |
+//	                 | n2_output
+//	                 v
+//	               +-------------------+
+//	               |   end(EndNode)    |
+//	               +-------------------+
+//	               +-------------------+
+//	               | Workflow Id: repo |
+//	               +-------------------+
 package compiler
 
 import (

pkg/controller/composite_workqueue.go

@@ -14,10 +14,10 @@ import (
 )
 
 // A CompositeWorkQueue can be used in cases where the work is enqueued by two sources. It can be enqueued by either
-// 1. Informer for the Primary Object itself. In case of FlytePropeller, this is the workflow object
-// 2. Informer or any other process that enqueues the top-level object for re-evaluation in response to one of the
-//    sub-objects being ready. In the case of FlytePropeller this is the "Node/Task" updates, will re-enqueue the workflow
-//    to be re-evaluated
+//  1. Informer for the Primary Object itself. In case of FlytePropeller, this is the workflow object
+//  2. Informer or any other process that enqueues the top-level object for re-evaluation in response to one of the
+//     sub-objects being ready. In the case of FlytePropeller this is the "Node/Task" updates, will re-enqueue the workflow
+//     to be re-evaluated
 type CompositeWorkQueue interface {
 	workqueue.RateLimitingInterface
 	// Specialized interface that should be called to start the migration of work from SubQueue to primaryQueue

pkg/controller/config/config.go

@@ -1,33 +1,36 @@
-// Package config contains the core configuration for FlytePropeller. This configuration can be added under the ``propeller`` section.
-//  Example config:
+// Package config contains the core configuration for FlytePropeller. This configuration can be added under the “propeller“ section.
+//
+//	Example config:
+//
 // ----------------
-//  propeller:
-//     rawoutput-prefix: s3://my-container/test/
-//     metadata-prefix: metadata/propeller/sandbox
-//     workers: 4
-//     workflow-reeval-duration: 10s
-//     downstream-eval-duration: 5s
-//     limit-namespace: "all"
-//     prof-port: 11254
-//     metrics-prefix: flyte
-//     enable-admin-launcher: true
-//     max-ttl-hours: 1
-//     gc-interval: 500m
-//     queue:
-//       type: batch
-//       queue:
-//         type: bucket
-//         rate: 1000
-//         capacity: 10000
-//       sub-queue:
-//         type: bucket
-//         rate: 1000
-//         capacity: 10000
-//     # This config assumes using `make start` in flytesnacks repo to startup a DinD k3s container
-//     kube-config: "$HOME/kubeconfig/k3s/k3s.yaml"
-//     publish-k8s-events: true
-//     workflowStore:
-//       policy: "ResourceVersionCache"
+//
+//	propeller:
+//	   rawoutput-prefix: s3://my-container/test/
+//	   metadata-prefix: metadata/propeller/sandbox
+//	   workers: 4
+//	   workflow-reeval-duration: 10s
+//	   downstream-eval-duration: 5s
+//	   limit-namespace: "all"
+//	   prof-port: 11254
+//	   metrics-prefix: flyte
+//	   enable-admin-launcher: true
+//	   max-ttl-hours: 1
+//	   gc-interval: 500m
+//	   queue:
+//	     type: batch
+//	     queue:
+//	       type: bucket
+//	       rate: 1000
+//	       capacity: 10000
+//	     sub-queue:
+//	       type: bucket
+//	       rate: 1000
+//	       capacity: 10000
+//	   # This config assumes using `make start` in flytesnacks repo to startup a DinD k3s container
+//	   kube-config: "$HOME/kubeconfig/k3s/k3s.yaml"
+//	   publish-k8s-events: true
+//	   workflowStore:
+//	     policy: "ResourceVersionCache"
 package config
 
 import (

pkg/controller/controller.go

@@ -131,7 +131,7 @@ func (c *Controller) run(ctx context.Context) error {
 }
 
 // Called from leader elector -if configured- to start running as the leader.
-func (c *Controller) onStartedLeading(ctx context.Context) {
+func (c *Controller) onStartedLeading(_ context.Context) {
 	ctx, cancelNow := context.WithCancel(context.Background())
 	logger.Infof(ctx, "Acquired leader lease.")
 	go func() {
@@ -460,7 +460,7 @@ func New(ctx context.Context, cfg *config.Config, kubeclientset kubernetes.Inter
 	// Set up an event handler for when FlyteWorkflow resources change
 	flyteworkflowInformer.Informer().AddEventHandler(controller.getWorkflowUpdatesHandler())
 
-	updateHandler := flytek8s.GetPodTemplateUpdatesHandler(&flytek8s.DefaultPodTemplateStore, flyteK8sConfig.GetK8sPluginConfig().DefaultPodTemplateName)
+	updateHandler := flytek8s.GetPodTemplateUpdatesHandler(&flytek8s.DefaultPodTemplateStore)
 	podTemplateInformer.Informer().AddEventHandler(updateHandler)
 	return controller, nil
 }

pkg/controller/handler.go

@@ -163,18 +163,19 @@ func (p *Propeller) TryMutateWorkflow(ctx context.Context, originalW *v1alpha1.F
 // The return value should be an error, in the case, we wish to retry this workflow
 // <pre>
 //
-//     +--------+        +---------+        +------------+     +---------+
-//     |        |        |         |        |            |     |         |
-//     | Ready  +--------> Running +--------> Succeeding +-----> Success |
-//     |        |        |         |        |            |     |         |
-//     +--------+        +---------+        +------------+     +---------+
-//         |                  |
-//         |                  |
-//         |             +----v----+        +---------------------+        +--------+
-//         |             |         |        |     (optional)      |        |        |
-//         +-------------> Failing +--------> HandlingFailureNode +--------> Failed |
-//                       |         |        |                     |        |        |
-//                       +---------+        +---------------------+        +--------+
+//	+--------+        +---------+        +------------+     +---------+
+//	|        |        |         |        |            |     |         |
+//	| Ready  +--------> Running +--------> Succeeding +-----> Success |
+//	|        |        |         |        |            |     |         |
+//	+--------+        +---------+        +------------+     +---------+
+//	    |                  |
+//	    |                  |
+//	    |             +----v----+        +---------------------+        +--------+
+//	    |             |         |        |     (optional)      |        |        |
+//	    +-------------> Failing +--------> HandlingFailureNode +--------> Failed |
+//	                  |         |        |                     |        |        |
+//	                  +---------+        +---------------------+        +--------+
+//
 // </pre>
 func (p *Propeller) Handle(ctx context.Context, namespace, name string) error {
 	logger.Infof(ctx, "Processing Workflow.")

pkg/controller/nodes/executor.go

@@ -6,14 +6,14 @@
 // to the respective node handlers
 //
 // Available node handlers are
-// - Task: Arguably the most important handler as it handles all tasks. These include all plugins. The goal of the workflow is
-//         is to run tasks, thus every workflow will contain atleast one TaskNode (except for the case, where the workflow
-//          is purely a meta-workflow and can run other workflows
-// - SubWorkflow: This is one of the most important handlers. It can execute Workflows that are nested inside a workflow
-// - DynamicTask Handler: This is just a decorator on the Task Handler. It handles cases, in which the Task returns a futures
-//                        file. Every Task is actually executed through the DynamicTaskHandler
-// - Branch Handler: This handler is used to execute branches
-// - Start & End Node handler: these are nominal handlers for the start and end node and do no really carry a lot of logic
+//   - Task: Arguably the most important handler as it handles all tasks. These include all plugins. The goal of the workflow is
+//     is to run tasks, thus every workflow will contain atleast one TaskNode (except for the case, where the workflow
+//     is purely a meta-workflow and can run other workflows
+//   - SubWorkflow: This is one of the most important handlers. It can execute Workflows that are nested inside a workflow
+//   - DynamicTask Handler: This is just a decorator on the Task Handler. It handles cases, in which the Task returns a futures
+//     file. Every Task is actually executed through the DynamicTaskHandler
+//   - Branch Handler: This handler is used to execute branches
+//   - Start & End Node handler: these are nominal handlers for the start and end node and do no really carry a lot of logic
 package nodes
 
 import (

pkg/webhook/pod.go

@@ -1,29 +1,29 @@
 // The PodMutator is a controller-runtime webhook that intercepts Pod Creation events and mutates them. Currently, there
 // is only one registered Mutator, that's the SecretsMutator. It works as follows:
 //
-//  -  The Webhook only works on Pods. If propeller/plugins launch a resource outside of K8s (or in a separate k8s
+//   - The Webhook only works on Pods. If propeller/plugins launch a resource outside of K8s (or in a separate k8s
 //     cluster), it's the responsibility of the plugin to correctly pass secret injection information.
-//  -  When a k8s-plugin builds a resource, propeller's PluginManager will automatically inject a label `inject-flyte
+//   - When a k8s-plugin builds a resource, propeller's PluginManager will automatically inject a label `inject-flyte
 //     -secrets: true` and serialize the secret injection information into the annotations.
-//  -  If a plugin does not use the K8sPlugin interface, it's its responsibility to pass secret injection information.
-//  -  If a k8s plugin creates a CRD that launches other Pods (e.g. Spark/PyTorch... etc.), it's its responsibility to
+//   - If a plugin does not use the K8sPlugin interface, it's its responsibility to pass secret injection information.
+//   - If a k8s plugin creates a CRD that launches other Pods (e.g. Spark/PyTorch... etc.), it's its responsibility to
 //     make sure the labels/annotations set on the CRD by PluginManager are propagated to those launched Pods. This
 //     ensures secret injection happens no matter how many levels of indirections there are.
-//  -  The Webhook expects 'inject-flyte-secrets: true' as a label on the Pod. Otherwise it won't listen/observe that pod.
-//  -  Once it intercepts the admission request, it goes over all registered Mutators and invoke them in the order they
+//   - The Webhook expects 'inject-flyte-secrets: true' as a label on the Pod. Otherwise it won't listen/observe that pod.
+//   - Once it intercepts the admission request, it goes over all registered Mutators and invoke them in the order they
 //     are registered as. If a Mutator fails and it's marked as `required`, the operation will fail and the admission
 //     will be rejected.
-//  -  The SecretsMutator will attempt to lookup the requested secret from the process environment. If the secret is
+//   - The SecretsMutator will attempt to lookup the requested secret from the process environment. If the secret is
 //     already mounted, it'll inject it as plain-text into the Pod Spec (Less secure).
-//  -  If it's not found in the environment it'll, instead, fallback to the enabled Secrets Injector (K8s, Confidant,
+//   - If it's not found in the environment it'll, instead, fallback to the enabled Secrets Injector (K8s, Confidant,
 //     Vault... etc.).
-//  -  Each SecretsInjector will mutate the Pod differently depending on how its backend secrets system injects the secrets
+//   - Each SecretsInjector will mutate the Pod differently depending on how its backend secrets system injects the secrets
 //     for example:
-//      - For K8s secrets, it'll either add EnvFromSecret or VolumeMountSource (depending on the MountRequirement
-//        stated in the flyteIdl.Secret object) into the Pod. There is no validation that the secret exist and is available
-//        to the Pod at this point. If the secret is not accessible, the Pod will fail with ContainerCreationConfigError and
-//        will be retried.
-//      - For Vault secrets, it'll inject the right annotations to trigger Vault's own sidecar/webhook to mount the secret.
+//   - For K8s secrets, it'll either add EnvFromSecret or VolumeMountSource (depending on the MountRequirement
+//     stated in the flyteIdl.Secret object) into the Pod. There is no validation that the secret exist and is available
+//     to the Pod at this point. If the secret is not accessible, the Pod will fail with ContainerCreationConfigError and
+//     will be retried.
+//   - For Vault secrets, it'll inject the right annotations to trigger Vault's own sidecar/webhook to mount the secret.
 package webhook
 
 import (