Update resync time from 5 minute to 1 minute

workflow/controller/wrapper.go

@@ -5,16 +5,15 @@ import (
 	"fmt"
 	"github.com/argoproj/argo/pkg/apis/workflow/v1alpha1"
 	wfclientset "github.com/argoproj/argo/pkg/client/clientset/versioned"
-	"github.com/argoproj/argo/workflow/util"
 	log "github.com/sirupsen/logrus"
 	"k8s.io/apimachinery/pkg/apis/meta/v1"
 	"k8s.io/client-go/kubernetes"
-	"strconv"
 	"time"
 )
 
 var workflowRecordLimitMap = make(map[string]int)
-const aliyunRetryKey = "aliyun.retry"
+
+//const aliyunRetryKey = "aliyun.retry"
 
 // ReSyncWorkflow lookup all workflow on incorrect status
 func (wfc *WorkflowController) ReSyncWorkflow(ctx context.Context) {
@@ -42,7 +41,7 @@ func (wfc *WorkflowController) ReSyncWorkflow(ctx context.Context) {
 			checkDagWorkflowStatus(&wf, wfc.wfclientset, wfc.kubeclientset)
 		}
 
-		time.Sleep(5 * time.Minute)
+		time.Sleep(1 * time.Minute)
 	}
 
 }
@@ -83,7 +82,7 @@ func checkDagWorkflowStatus(wf *v1alpha1.Workflow, argoClient wfclientset.Interf
 
 	mapKey := fmt.Sprintf("%s-%s", wf.Namespace, wf.Name)
 	if val, ok := workflowRecordLimitMap[mapKey]; ok {
-		if val < 3 {
+		if val < 2 {
 			workflowRecordLimitMap[mapKey] += 1
 			return
 		}
@@ -92,46 +91,47 @@ func checkDagWorkflowStatus(wf *v1alpha1.Workflow, argoClient wfclientset.Interf
 		return
 	}
 
-	retryArgoInterface := argoClient.ArgoprojV1alpha1().Workflows(wf.Namespace)
+	//retryArgoInterface := argoClient.ArgoprojV1alpha1().Workflows(wf.Namespace)
 	log.Infof("ReSync Workflow %s:%s status from %s to %s", wf.Namespace, wf.Name, wf.Status.Phase, v1alpha1.NodeFailed)
 	dagNode.Phase = v1alpha1.NodeFailed
 	wf.Status.Nodes[targetNode[0].ID] = dagNode
 	wf.Status.Phase = v1alpha1.NodeFailed
 
-	wfLabelsMap := wf.Labels
-	wfRetryTimesInt := 0
-	if retryTimesStr, ok := wfLabelsMap[aliyunRetryKey]; !ok {
-		wfLabelsMap[aliyunRetryKey] = "1"
-	} else {
-		tmpRetryInt, tmpErr := strconv.Atoi(retryTimesStr)
-		if tmpErr != nil {
-			log.Error(tmpErr)
-			tmpRetryInt = 0
-		}
-		tmpRetryInt++
-		wfLabelsMap[aliyunRetryKey] = strconv.Itoa(tmpRetryInt)
-		wfRetryTimesInt = tmpRetryInt
-	}
-	wf.SetLabels(wfLabelsMap)
-
 	_, err := argoClient.ArgoprojV1alpha1().Workflows(wf.Namespace).Update(wf)
 	if err != nil {
 		log.Errorf("Update workflow %s:%s error %v", wf.Namespace, wf.Name, err)
 	}
+	delete(workflowRecordLimitMap, mapKey)
 
-	time.Sleep(10 * time.Second)
-	if wfRetryTimesInt > 3 {
-		return
-	}
-	wf, err = retryArgoInterface.Get(wf.Name, v1.GetOptions{})
-	if err != nil {
-		log.Errorf("Get workflow %s:%s error %v",  wf.Namespace, wf.Name, err)
-		return
-	}
-	log.Infof("Retry workflow %s:%s for unstable status map value %d", wf.Namespace, wf.Name, workflowRecordLimitMap[mapKey])
-	if _, tmpErr := util.RetryWorkflow(kubeClient, retryArgoInterface, wf); tmpErr != nil {
-		log.Errorf("Retry workflow %s error %v", wf.Name, tmpErr)
-	}
+	/*
+		wfLabelsMap := wf.Labels
+		wfRetryTimesInt := 0
+		if retryTimesStr, ok := wfLabelsMap[aliyunRetryKey]; !ok {
+			wfLabelsMap[aliyunRetryKey] = "1"
+		} else {
+			tmpRetryInt, tmpErr := strconv.Atoi(retryTimesStr)
+			if tmpErr != nil {
+				log.Error(tmpErr)
+				tmpRetryInt = 0
+			}
+			tmpRetryInt++
+			wfLabelsMap[aliyunRetryKey] = strconv.Itoa(tmpRetryInt)
+			wfRetryTimesInt = tmpRetryInt
+		}
+		wf.SetLabels(wfLabelsMap)
 
-	delete(workflowRecordLimitMap, mapKey)
+		time.Sleep(10 * time.Second)
+		if wfRetryTimesInt > 3 {
+			return
+		}
+		wf, err = retryArgoInterface.Get(wf.Name, v1.GetOptions{})
+		if err != nil {
+			log.Errorf("Get workflow %s:%s error %v",  wf.Namespace, wf.Name, err)
+			return
+		}
+		log.Infof("Retry workflow %s:%s for unstable status map value %d", wf.Namespace, wf.Name, workflowRecordLimitMap[mapKey])
+		if _, tmpErr := util.RetryWorkflow(kubeClient, retryArgoInterface, wf); tmpErr != nil {
+			log.Errorf("Retry workflow %s error %v", wf.Name, tmpErr)
+		}
+	*/
 }