fix(controller): Various v2.12 fixes. Fixes #4798, #4801, #4806 (#4808)

Signed-off-by: Alex Collins <alex_collins@intuit.com>
argoproj · Jan 4, 2021 · a0024d0 · a0024d0
1 parent ee59d49
commit a0024d0
Show file tree

Hide file tree

Showing 25 changed files with 264 additions and 214 deletions.
diff --git a/Makefile b/Makefile
@@ -50,6 +50,8 @@ STATIC_BUILD          ?= true
 STATIC_FILES          ?= true
 GOTEST                ?= go test
 PROFILE               ?= minimal
+# by keeping this short we speed up the tests
+DEFAULT_REQUEUE_TIME  ?= 2s
 # whether or not to start the Argo Service in TLS mode
 SECURE                := false
 AUTH_MODE             := hybrid
@@ -450,7 +452,7 @@ endif
 	grep '127.0.0.1[[:blank:]]*postgres' /etc/hosts
 	grep '127.0.0.1[[:blank:]]*mysql' /etc/hosts
 ifeq ($(RUN_MODE),local)
-	env SECURE=$(SECURE) ALWAYS_OFFLOAD_NODE_STATUS=$(ALWAYS_OFFLOAD_NODE_STATUS) LOG_LEVEL=$(LOG_LEVEL) UPPERIO_DB_DEBUG=$(UPPERIO_DB_DEBUG) VERSION=$(VERSION) AUTH_MODE=$(AUTH_MODE) NAMESPACED=$(NAMESPACED) NAMESPACE=$(KUBE_NAMESPACE) $(GOPATH)/bin/goreman -set-ports=false -logtime=false start
+	env DEFAULT_REQUEUE_TIME=$(DEFAULT_REQUEUE_TIME) SECURE=$(SECURE) ALWAYS_OFFLOAD_NODE_STATUS=$(ALWAYS_OFFLOAD_NODE_STATUS) LOG_LEVEL=$(LOG_LEVEL) UPPERIO_DB_DEBUG=$(UPPERIO_DB_DEBUG) VERSION=$(VERSION) AUTH_MODE=$(AUTH_MODE) NAMESPACED=$(NAMESPACED) NAMESPACE=$(KUBE_NAMESPACE) $(GOPATH)/bin/goreman -set-ports=false -logtime=false start
 endif
 
 .PHONY: wait

diff --git a/Procfile b/Procfile
@@ -1,2 +1,2 @@
-controller: LEADER_ELECTION_IDENTITY=local ALWAYS_OFFLOAD_NODE_STATUS=${ALWAYS_OFFLOAD_NODE_STATUS} OFFLOAD_NODE_STATUS_TTL=30s WORKFLOW_GC_PERIOD=30s UPPERIO_DB_DEBUG=${UPPERIO_DB_DEBUG} ARCHIVED_WORKFLOW_GC_PERIOD=30s ./dist/workflow-controller --executor-image argoproj/argoexec:${VERSION} --namespaced=${NAMESPACED} --namespace ${NAMESPACE} --loglevel ${LOG_LEVEL}
+controller: DEFAULT_REQUEUE_TIME=${DEFAULT_REQUEUE_TIME} LEADER_ELECTION_IDENTITY=local ALWAYS_OFFLOAD_NODE_STATUS=${ALWAYS_OFFLOAD_NODE_STATUS} OFFLOAD_NODE_STATUS_TTL=30s WORKFLOW_GC_PERIOD=30s UPPERIO_DB_DEBUG=${UPPERIO_DB_DEBUG} ARCHIVED_WORKFLOW_GC_PERIOD=30s ./dist/workflow-controller --executor-image argoproj/argoexec:${VERSION} --namespaced=${NAMESPACED} --namespace ${NAMESPACE} --loglevel ${LOG_LEVEL}
 argo-server: UPPERIO_DB_DEBUG=${UPPERIO_DB_DEBUG} ./dist/argo --loglevel ${LOG_LEVEL} server --namespaced=${NAMESPACED} --namespace ${NAMESPACE} --auth-mode ${AUTH_MODE} --secure=$SECURE
diff --git a/manifests/base/workflow-controller/workflow-controller-deployment.yaml b/manifests/base/workflow-controller/workflow-controller-deployment.yaml
@@ -32,6 +32,20 @@ spec:
                 fieldRef:
                   apiVersion: v1
                   fieldPath: metadata.name
+          ports:
+            - name: metrics
+              containerPort: 9090
+          # Periodically check we are listening on the metrics port
+          # causing a restart if it is not OK.
+          # This takes advantage of the fact that if the metrics service has died,
+          # then the controller has died.
+          # In testing, it appears to take 60-90s from failure to restart.
+          livenessProbe:
+            httpGet:
+              port: metrics
+              path: /metrics
+            initialDelaySeconds: 30
+            periodSeconds: 30
       securityContext:
         runAsNonRoot: true
       nodeSelector:

diff --git a/manifests/install.yaml b/manifests/install.yaml
@@ -523,7 +523,16 @@ spec:
               apiVersion: v1
               fieldPath: metadata.name
         image: argoproj/workflow-controller:latest
+        livenessProbe:
+          httpGet:
+            path: /metrics
+            port: metrics
+          initialDelaySeconds: 30
+          periodSeconds: 30
         name: workflow-controller
+        ports:
+        - containerPort: 9090
+          name: metrics
         securityContext:
           capabilities:
             drop:

diff --git a/manifests/namespace-install.yaml b/manifests/namespace-install.yaml
@@ -418,7 +418,16 @@ spec:
               apiVersion: v1
               fieldPath: metadata.name
         image: argoproj/workflow-controller:latest
+        livenessProbe:
+          httpGet:
+            path: /metrics
+            port: metrics
+          initialDelaySeconds: 30
+          periodSeconds: 30
         name: workflow-controller
+        ports:
+        - containerPort: 9090
+          name: metrics
         securityContext:
           capabilities:
             drop:

diff --git a/manifests/quick-start-minimal.yaml b/manifests/quick-start-minimal.yaml
@@ -672,7 +672,16 @@ spec:
               apiVersion: v1
               fieldPath: metadata.name
         image: argoproj/workflow-controller:latest
+        livenessProbe:
+          httpGet:
+            path: /metrics
+            port: metrics
+          initialDelaySeconds: 30
+          periodSeconds: 30
         name: workflow-controller
+        ports:
+        - containerPort: 9090
+          name: metrics
         securityContext:
           capabilities:
             drop:

diff --git a/manifests/quick-start-mysql.yaml b/manifests/quick-start-mysql.yaml
@@ -761,7 +761,16 @@ spec:
               apiVersion: v1
               fieldPath: metadata.name
         image: argoproj/workflow-controller:latest
+        livenessProbe:
+          httpGet:
+            path: /metrics
+            port: metrics
+          initialDelaySeconds: 30
+          periodSeconds: 30
         name: workflow-controller
+        ports:
+        - containerPort: 9090
+          name: metrics
         securityContext:
           capabilities:
             drop:

diff --git a/manifests/quick-start-postgres.yaml b/manifests/quick-start-postgres.yaml
@@ -753,7 +753,16 @@ spec:
               apiVersion: v1
               fieldPath: metadata.name
         image: argoproj/workflow-controller:latest
+        livenessProbe:
+          httpGet:
+            path: /metrics
+            port: metrics
+          initialDelaySeconds: 30
+          periodSeconds: 30
         name: workflow-controller
+        ports:
+        - containerPort: 9090
+          name: metrics
         securityContext:
           capabilities:
             drop:

diff --git a/test/e2e/functional/stop-terminate.yaml b/test/e2e/functional/stop-terminate.yaml
@@ -1,7 +1,7 @@
 apiVersion: argoproj.io/v1alpha1
 kind: Workflow
 metadata:
-  name: stop-terminate
+  generateName: stop-terminate-
   labels:
     argo-e2e: true
 spec:
@@ -23,7 +23,7 @@ spec:
           - name: message
       container:
         image: argoproj/argosay:v2
-        args: [sleep, "10s"]
+        args: [sleep, "30s"]
 
     - name: exit
       container:

diff --git a/test/e2e/functional_test.go b/test/e2e/functional_test.go
@@ -575,19 +575,19 @@ func (s *FunctionalSuite) TestStopBehavior() {
 		When().
 		SubmitWorkflow().
 		WaitForWorkflow(fixtures.ToStart, "to start").
-		RunCli([]string{"stop", "stop-terminate"}, func(t *testing.T, output string, err error) {
+		RunCli([]string{"stop", "@latest"}, func(t *testing.T, output string, err error) {
 			assert.NoError(t, err)
-			assert.Contains(t, output, "workflow stop-terminate stopped")
+			assert.Regexp(t, "workflow stop-terminate-.* stopped", output)
 		}).
 		WaitForWorkflow(45 * time.Second).
 		Then().
-		ExpectWorkflow(func(t *testing.T, _ *metav1.ObjectMeta, status *wfv1.WorkflowStatus) {
+		ExpectWorkflow(func(t *testing.T, m *metav1.ObjectMeta, status *wfv1.WorkflowStatus) {
 			assert.Equal(t, wfv1.NodeFailed, status.Phase)
 			nodeStatus := status.Nodes.FindByDisplayName("A.onExit")
 			if assert.NotNil(t, nodeStatus) {
 				assert.Equal(t, wfv1.NodeSucceeded, nodeStatus.Phase)
 			}
-			nodeStatus = status.Nodes.FindByDisplayName("stop-terminate.onExit")
+			nodeStatus = status.Nodes.FindByDisplayName(m.Name + ".onExit")
 			if assert.NotNil(t, nodeStatus) {
 				assert.Equal(t, wfv1.NodeSucceeded, nodeStatus.Phase)
 			}
@@ -600,17 +600,17 @@ func (s *FunctionalSuite) TestTerminateBehavior() {
 		When().
 		SubmitWorkflow().
 		WaitForWorkflow(fixtures.ToStart, "to start").
-		RunCli([]string{"terminate", "stop-terminate"}, func(t *testing.T, output string, err error) {
+		RunCli([]string{"terminate", "@latest"}, func(t *testing.T, output string, err error) {
 			assert.NoError(t, err)
-			assert.Contains(t, output, "workflow stop-terminate terminated")
+			assert.Regexp(t, "workflow stop-terminate-.* terminated", output)
 		}).
 		WaitForWorkflow().
 		Then().
-		ExpectWorkflow(func(t *testing.T, _ *metav1.ObjectMeta, status *wfv1.WorkflowStatus) {
+		ExpectWorkflow(func(t *testing.T, m *metav1.ObjectMeta, status *wfv1.WorkflowStatus) {
 			assert.Equal(t, wfv1.NodeFailed, status.Phase)
 			nodeStatus := status.Nodes.FindByDisplayName("A.onExit")
 			assert.Nil(t, nodeStatus)
-			nodeStatus = status.Nodes.FindByDisplayName("stop-terminate.onExit")
+			nodeStatus = status.Nodes.FindByDisplayName(m.Name + ".onExit")
 			assert.Nil(t, nodeStatus)
 		})
 }

diff --git a/util/logs/workflow-logger.go b/util/logs/workflow-logger.go
@@ -147,7 +147,7 @@ func WorkflowLogs(ctx context.Context, wfClient versioned.Interface, kubeClient
 		ensureWeAreStreaming(&pod)
 	}
 
-	if req.GetLogOptions().Follow {
+	if logOptions.Follow {
 		wfListOptions := metav1.ListOptions{FieldSelector: "metadata.name=" + req.GetName(), ResourceVersion: "0"}
 		wfWatch, err := wfInterface.Watch(wfListOptions)
 		if err != nil {

diff --git a/workflow/controller/controller.go b/workflow/controller/controller.go
@@ -291,7 +291,7 @@ func (wfc *WorkflowController) createSynchronizationManager() error {
 	}
 
 	nextWorkflow := func(key string) {
-		wfc.wfQueue.Add(key)
+		wfc.wfQueue.AddRateLimited(key)
 	}
 
 	wfc.syncManager = sync.NewLockManager(getSyncLimit, nextWorkflow)
@@ -353,12 +353,7 @@ func (wfc *WorkflowController) notifySemaphoreConfigUpdate(cm *apiv1.ConfigMap)
 			log.Warnf("received object from indexer %s is not an unstructured", indexes.SemaphoreConfigIndexName)
 			continue
 		}
-		wf, err := util.FromUnstructured(un)
-		if err != nil {
-			log.Errorf("failed to convert to workflow from unstructured: %v", err)
-			continue
-		}
-		wfc.wfQueue.Add(fmt.Sprintf("%s/%s", wf.Namespace, wf.Name))
+		wfc.wfQueue.AddRateLimited(fmt.Sprintf("%s/%s", un.GetNamespace(), un.GetName()))
 	}
 }
 
@@ -749,7 +744,7 @@ func (wfc *WorkflowController) addWorkflowInformerHandlers() {
 					key, err := cache.DeletionHandlingMetaNamespaceKeyFunc(obj)
 					if err == nil {
 						wfc.releaseAllWorkflowLocks(obj)
-						wfc.wfQueue.AddRateLimited(key)
+						// no need to add to the queue - this workflow is done
 						wfc.throttler.Remove(key)
 					}
 				},

diff --git a/workflow/controller/controller_test.go b/workflow/controller/controller_test.go
@@ -3,6 +3,7 @@ package controller
 import (
 	"context"
 	"testing"
+	"time"
 
 	"github.com/argoproj/pkg/sync"
 	"k8s.io/apimachinery/pkg/apis/meta/v1/unstructured"
@@ -595,5 +596,6 @@ func TestNotifySemaphoreConfigUpdate(t *testing.T) {
 	assert.Equal(0, controller.wfQueue.Len())
 
 	controller.notifySemaphoreConfigUpdate(&cm)
+	time.Sleep(2 * time.Second)
 	assert.Equal(2, controller.wfQueue.Len())
 }
diff --git a/workflow/controller/estimation/estimator_factory.go b/workflow/controller/estimation/estimator_factory.go
@@ -46,7 +46,7 @@ func (f *estimatorFactory) NewEstimator(wf *wfv1.Workflow) (Estimator, error) {
 			if err != nil {
 				return defaultEstimator, fmt.Errorf("failed to list workflows by index: %v", err)
 			}
-			var newestWf *wfv1.Workflow
+			var newestUn *unstructured.Unstructured
 			for _, obj := range objs {
 				un, ok := obj.(*unstructured.Unstructured)
 				if !ok {
@@ -55,16 +55,16 @@ func (f *estimatorFactory) NewEstimator(wf *wfv1.Workflow) (Estimator, error) {
 				if un.GetLabels()[common.LabelKeyPhase] != string(wfv1.NodeSucceeded) {
 					continue
 				}
-				candidateWf, err := util.FromUnstructured(un)
+				// we use `creationTimestamp` because it's fast
+				if newestUn == nil || un.GetCreationTimestamp().After(newestUn.GetCreationTimestamp().Time) {
+					newestUn = un
+				}
+			}
+			if newestUn != nil {
+				newestWf, err := util.FromUnstructured(newestUn)
 				if err != nil {
 					return defaultEstimator, fmt.Errorf("failed convert unstructured to workflow: %w", err)
 				}
-				// we use `startedAt` because that's same as how the archive sorts
-				if newestWf == nil || candidateWf.Status.StartedAt.Time.After(newestWf.Status.StartedAt.Time) {
-					newestWf = candidateWf
-				}
-			}
-			if newestWf != nil {
 				err = f.hydrator.Hydrate(newestWf)
 				if err != nil {
 					return defaultEstimator, fmt.Errorf("failed hydrate last workflow: %w", err)
@@ -78,7 +78,7 @@ func (f *estimatorFactory) NewEstimator(wf *wfv1.Workflow) (Estimator, error) {
 			}
 			workflows, err := f.wfArchive.ListWorkflows(wf.Namespace, time.Time{}, time.Time{}, requirements, 1, 0)
 			if err != nil {
-				return nil, fmt.Errorf("failed to list archived workflows: %v", err)
+				return defaultEstimator, fmt.Errorf("failed to list archived workflows: %v", err)
 			}
 			if len(workflows) > 0 {
 				return &estimator{wf, &workflows[0]}, nil

diff --git a/workflow/controller/indexes/indexes.go b/workflow/controller/indexes/indexes.go
@@ -1,5 +1,11 @@
 package indexes
 
+// Indexers (i.e. IndexFunc) should be fast and should not return errors
+// If an indexers returns an error, the cache will panic, and crash the Go VM.
+// Indexers should be fast, if they are not, then the informer will get out of date
+// and start returning old workflows, resulting in the operator reconciling out of date
+// information, causing conflict errors and risking corrupted data.
+
 const (
 	ClusterWorkflowTemplateIndex = "clusterworkflowtemplate"
 	CronWorkflowIndex            = "cronworkflow"

diff --git a/workflow/controller/indexes/labels.go b/workflow/controller/indexes/labels.go
@@ -1,8 +1,6 @@
 package indexes
 
 import (
-	"fmt"
-
 	"k8s.io/apimachinery/pkg/api/meta"
 	"k8s.io/client-go/tools/cache"
 
@@ -18,7 +16,7 @@ func MetaWorkflowPhaseIndexFunc() cache.IndexFunc {
 	return func(obj interface{}) ([]string, error) {
 		v, err := meta.Accessor(obj)
 		if err != nil {
-			return []string{}, fmt.Errorf("object has no meta: %v", err)
+			return nil, nil
 		}
 		if value, exists := v.GetLabels()[common.LabelKeyPhase]; exists {
 			return []string{value}, nil
@@ -33,12 +31,12 @@ func MetaNamespaceLabelIndexFunc(label string) cache.IndexFunc {
 	return func(obj interface{}) ([]string, error) {
 		v, err := meta.Accessor(obj)
 		if err != nil {
-			return []string{}, fmt.Errorf("object has no meta: %v", err)
+			return nil, nil
 		}
 		if value, exists := v.GetLabels()[label]; exists {
 			return []string{MetaNamespaceLabelIndex(v.GetNamespace(), value)}, nil
 		} else {
-			return []string{}, nil
+			return nil, nil
 		}
 	}
 }
diff --git a/workflow/controller/indexes/workflow_index.go b/workflow/controller/indexes/workflow_index.go
@@ -1,9 +1,6 @@
 package indexes
 
 import (
-	"fmt"
-
-	"github.com/prometheus/common/log"
 	"k8s.io/apimachinery/pkg/api/meta"
 	"k8s.io/apimachinery/pkg/apis/meta/v1/unstructured"
 	"k8s.io/client-go/tools/cache"
@@ -15,11 +12,11 @@ import (
 func MetaWorkflowIndexFunc(obj interface{}) ([]string, error) {
 	m, err := meta.Accessor(obj)
 	if err != nil {
-		return []string{}, fmt.Errorf("object has no meta: %v", err)
+		return nil, nil
 	}
 	name, ok := m.GetLabels()[common.LabelKeyWorkflow]
 	if !ok {
-		return []string{}, fmt.Errorf("object has no workflow label")
+		return nil, nil
 	}
 	return []string{WorkflowIndexValue(m.GetNamespace(), name)}, nil
 }
@@ -32,13 +29,11 @@ func WorkflowSemaphoreKeysIndexFunc() cache.IndexFunc {
 	return func(obj interface{}) ([]string, error) {
 		un, ok := obj.(*unstructured.Unstructured)
 		if !ok {
-			log.Warnf("cannot convert obj into unstructured.Unstructured in Indexer %s", SemaphoreConfigIndexName)
-			return []string{}, nil
+			return nil, nil
 		}
 		wf, err := util.FromUnstructured(un)
 		if err != nil {
-			log.Warnf("failed to convert to workflow from unstructured: %v", err)
-			return []string{}, nil
+			return nil, nil
 		}
 		return wf.GetSemaphoreKeys(), nil
 	}