Improve staleness logic by improving error handling and retrying a subset of errors

config/test/flinkk8soperator_config.yaml

@@ -4,7 +4,6 @@
 operator:
   ingressUrlFormat: "{{$jobCluster}}.lyft.xyz"
   containerNameFormat: "%s-unknown"
-  statemachineStalenessDuration: 10m
 logger:
   show-source: true
   level: 4

integ/main_test.go

@@ -76,12 +76,11 @@ func (s *IntegSuite) SetUpSuite(c *C) {
 
 	if runDirect {
 		config := controllerConfig.Config{
-			LimitNamespace:                namespace,
-			UseProxy:                      true,
-			ResyncPeriod:                  flyteConfig.Duration{Duration: 3 * time.Second},
-			StatemachineStalenessDuration: flyteConfig.Duration{Duration: 30 * time.Second},
-			MetricsPrefix:                 "flinkk8soperator",
-			ProxyPort:                     flyteConfig.Port{Port: 8001},
+			LimitNamespace: namespace,
+			UseProxy:       true,
+			ResyncPeriod:   flyteConfig.Duration{Duration: 3 * time.Second},
+			MetricsPrefix:  "flinkk8soperator",
+			ProxyPort:      flyteConfig.Port{Port: 8001},
 		}
 
 		log.Info("Running operator directly")

integ/utils/utils.go

@@ -139,7 +139,7 @@ func (f *TestUtil) CreateCRD() error {
 
 func (f *TestUtil) CreateOperator() error {
 	configValue := make(map[string]string)
-	configValue["development"] = "operator:\n  containerNameFormat: \"%s-unknown\"\n  statemachineStalenessDuration: 40s\n  resyncPeriod: 5s"
+	configValue["development"] = "operator:\n  containerNameFormat: \"%s-unknown\"\n  resyncPeriod: 5s"
 
 	configMap := v1.ConfigMap{
 		ObjectMeta: metav1.ObjectMeta{

local_config.yaml

@@ -5,7 +5,6 @@
 operator:
   useKubectlProxy: true
   containerNameFormat: "%s-unknown"
-  statemachineStalenessDuration: 1m
   metricsPrefix: "flinkk8soperator"
   resyncPeriod: 10s
   proxyPort: 8001

pkg/apis/app/v1alpha1/types.go

@@ -159,6 +159,8 @@ type FlinkApplicationStatus struct {
 	JobStatus        FlinkJobStatus        `json:"jobStatus"`
 	FailedDeployHash string                `json:"failedUpdateHash,omitEmpty"`
 	DeployHash       string                `json:"deployHash"`
+	RetryCount       int32                 `json:"retryCount,omitEmpty"`
+	LastSeenError    string                `json:"lastSeenError,omitEmpty"`
 }
 
 func (in *FlinkApplicationStatus) GetPhase() FlinkApplicationPhase {

pkg/controller/config/config.go

@@ -11,16 +11,15 @@ const configSectionKey = "operator"
 var ConfigSection = config.MustRegisterSection(configSectionKey, &Config{})
 
 type Config struct {
-	ResyncPeriod                  config.Duration `json:"resyncPeriod" pflag:"\"30s\",Determines the resync period for all watchers."`
-	LimitNamespace                string          `json:"limitNamespace" pflag:"\"\",Namespaces to watch for by flink operator"`
-	MetricsPrefix                 string          `json:"metricsPrefix" pflag:"\"flinkk8soperator\",Prefix for metrics propagated to prometheus"`
-	ProfilerPort                  config.Port     `json:"prof-port" pflag:"\"10254\",Profiler port"`
-	FlinkIngressURLFormat         string          `json:"ingressUrlFormat"`
-	UseProxy                      bool            `json:"useKubectlProxy"`
-	ProxyPort                     config.Port     `json:"ProxyPort" pflag:"\"8001\",The port at which flink cluster runs locally"`
-	ContainerNameFormat           string          `json:"containerNameFormat"`
-	Workers                       int             `json:"workers" pflag:"4,Number of routines to process custom resource"`
-	StatemachineStalenessDuration config.Duration `json:"statemachineStalenessDuration" pflag:"\"5m\",Duration for statemachine staleness."`
+	ResyncPeriod          config.Duration `json:"resyncPeriod" pflag:"\"30s\",Determines the resync period for all watchers."`
+	LimitNamespace        string          `json:"limitNamespace" pflag:"\"\",Namespaces to watch for by flink operator"`
+	MetricsPrefix         string          `json:"metricsPrefix" pflag:"\"flinkk8soperator\",Prefix for metrics propagated to prometheus"`
+	ProfilerPort          config.Port     `json:"prof-port" pflag:"\"10254\",Profiler port"`
+	FlinkIngressURLFormat string          `json:"ingressUrlFormat"`
+	UseProxy              bool            `json:"useKubectlProxy"`
+	ProxyPort             config.Port     `json:"ProxyPort" pflag:"\"8001\",The port at which flink cluster runs locally"`
+	ContainerNameFormat   string          `json:"containerNameFormat"`
+	Workers               int             `json:"workers" pflag:"4,Number of routines to process custom resource"`
 }
 
 func GetConfig() *Config {

pkg/controller/flink/client/api.go

@@ -101,18 +101,18 @@ func (c *FlinkJobManagerClient) GetJobConfig(ctx context.Context, url, jobID str
 	response, err := c.executeRequest(ctx, httpGet, url, nil)
 	if err != nil {
 		c.metrics.getJobConfigFailureCounter.Inc(ctx)
-		return nil, errors.Wrap(err, "GetJobConfig API request failed")
+		return nil, GetError(err, "GetJobConfig", globalFailure)
 	}
 
 	if response != nil && !response.IsSuccess() {
 		c.metrics.getJobConfigFailureCounter.Inc(ctx)
 		logger.Errorf(ctx, fmt.Sprintf("Get Jobconfig failed with response %v", response))
-		return nil, errors.New(fmt.Sprintf("Get Jobconfig failed with status %v", response.Status()))
+		return nil, GetError(err, "GetJobConfig", response.Status())
 	}
 	var jobConfigResponse JobConfigResponse
 	if err := json.Unmarshal(response.Body(), &jobConfigResponse); err != nil {
 		logger.Errorf(ctx, "Unable to Unmarshal jobPlanResponse %v, err: %v", response, err)
-		return nil, err
+		return nil, GetError(err, "GetJobConfig", jsonUnmarshalError)
 	}
 	c.metrics.getJobConfigSuccessCounter.Inc(ctx)
 	return &jobConfigResponse, nil
@@ -123,19 +123,19 @@ func (c *FlinkJobManagerClient) GetClusterOverview(ctx context.Context, url stri
 	response, err := c.executeRequest(ctx, httpGet, url, nil)
 	if err != nil {
 		c.metrics.getClusterFailureCounter.Inc(ctx)
-		return nil, errors.Wrap(err, "GetClusterOverview API request failed")
+		return nil, GetError(err, "GetClusterOverview", globalFailure)
 	}
 	if response != nil && !response.IsSuccess() {
 		c.metrics.getClusterFailureCounter.Inc(ctx)
 		if response.StatusCode() != int(http.StatusNotFound) || response.StatusCode() != int(http.StatusServiceUnavailable) {
 			logger.Errorf(ctx, fmt.Sprintf("Get cluster overview failed with response %v", response))
 		}
-		return nil, errors.New(fmt.Sprintf("Get cluster overview failed with status %v", response.Status()))
+		return nil, GetError(err, "GetClusterOverview", response.Status())
 	}
 	var clusterOverviewResponse ClusterOverviewResponse
 	if err = json.Unmarshal(response.Body(), &clusterOverviewResponse); err != nil {
 		logger.Errorf(ctx, "Unable to Unmarshal clusterOverviewResponse %v, err: %v", response, err)
-		return nil, err
+		return nil, GetError(err, "GetClusterOverview", jsonUnmarshalError)
 	}
 	c.metrics.getClusterSuccessCounter.Inc(ctx)
 	return &clusterOverviewResponse, nil
@@ -174,17 +174,17 @@ func (c *FlinkJobManagerClient) CancelJobWithSavepoint(ctx context.Context, url
 	response, err := c.executeRequest(ctx, httpPost, url, cancelJobRequest)
 	if err != nil {
 		c.metrics.cancelJobFailureCounter.Inc(ctx)
-		return "", errors.Wrap(err, "Cancel job API request failed")
+		return "", GetError(err, "CancelJobWithSavepoint", globalFailure)
 	}
 	if response != nil && !response.IsSuccess() {
 		c.metrics.cancelJobFailureCounter.Inc(ctx)
 		logger.Errorf(ctx, fmt.Sprintf("Cancel job failed with response %v", response))
-		return "", errors.New(fmt.Sprintf("Cancel job failed with status %v", response.Status()))
+		return "", GetError(err, "CancelJobWithSavepoint", response.Status())
 	}
 	var cancelJobResponse CancelJobResponse
 	if err = json.Unmarshal(response.Body(), &cancelJobResponse); err != nil {
 		logger.Errorf(ctx, "Unable to Unmarshal cancelJobResponse %v, err: %v", response, err)
-		return "", err
+		return "", GetError(err, "CancelJobWithSavepoint", jsonUnmarshalError)
 	}
 	c.metrics.cancelJobSuccessCounter.Inc(ctx)
 	return cancelJobResponse.TriggerID, nil
@@ -198,12 +198,12 @@ func (c *FlinkJobManagerClient) ForceCancelJob(ctx context.Context, url string,
 	response, err := c.executeRequest(ctx, httpPatch, url, nil)
 	if err != nil {
 		c.metrics.forceCancelJobFailureCounter.Inc(ctx)
-		return errors.Wrap(err, "Force cancel job API request failed")
+		return GetError(err, "ForceCancelJob", globalFailure)
 	}
 	if response != nil && !response.IsSuccess() {
 		c.metrics.forceCancelJobFailureCounter.Inc(ctx)
 		logger.Errorf(ctx, fmt.Sprintf("Force cancel job failed with response %v", response))
-		return errors.New(fmt.Sprintf("Force cancel job failed with status %v", response.Status()))
+		return GetError(err, "ForceCancelJob", response.Status())
 	}
 
 	c.metrics.forceCancelJobFailureCounter.Inc(ctx)
@@ -217,18 +217,17 @@ func (c *FlinkJobManagerClient) SubmitJob(ctx context.Context, url string, jarID
 	response, err := c.executeRequest(ctx, httpPost, url, submitJobRequest)
 	if err != nil {
 		c.metrics.submitJobFailureCounter.Inc(ctx)
-		return nil, errors.Wrap(err, "Submit job API request failed")
+		return nil, GetError(err, "SubmitJob", globalFailure)
 	}
 	if response != nil && !response.IsSuccess() {
 		c.metrics.submitJobFailureCounter.Inc(ctx)
 		logger.Warnf(ctx, fmt.Sprintf("Job submission failed with response %v", response))
-		return nil, errors.New(fmt.Sprintf("Job submission failed with status %v\n%s",
-			response.Status(), string(response.Body())))
+		return nil, GetError(err, "SubmitJob", response.Status())
 	}
 	var submitJobResponse SubmitJobResponse
 	if err = json.Unmarshal(response.Body(), &submitJobResponse); err != nil {
 		logger.Errorf(ctx, "Unable to Unmarshal submitJobResponse %v, err: %v", response, err)
-		return nil, err
+		return nil, GetError(err, "SubmitJob", jsonUnmarshalError)
 	}
 
 	c.metrics.submitJobSuccessCounter.Inc(ctx)
@@ -242,17 +241,17 @@ func (c *FlinkJobManagerClient) CheckSavepointStatus(ctx context.Context, url st
 	response, err := c.executeRequest(ctx, httpGet, url, nil)
 	if err != nil {
 		c.metrics.checkSavepointFailureCounter.Inc(ctx)
-		return nil, errors.Wrap(err, "Check savepoint status API request failed")
+		return nil, GetError(err, "CheckSavepointStatus", globalFailure)
 	}
 	if response != nil && !response.IsSuccess() {
 		c.metrics.checkSavepointFailureCounter.Inc(ctx)
 		logger.Errorf(ctx, fmt.Sprintf("Check savepoint status failed with response %v", response))
-		return nil, errors.New(fmt.Sprintf("Check savepoint status failed with status %v", response.Status()))
+		return nil, GetError(err, "CheckSavepointStatus", response.Status())
 	}
 	var savepointResponse SavepointResponse
 	if err = json.Unmarshal(response.Body(), &savepointResponse); err != nil {
 		logger.Errorf(ctx, "Unable to Unmarshal savepointResponse %v, err: %v", response, err)
-		return nil, err
+		return nil, GetError(err, "CheckSavepointStatus", jsonUnmarshalError)
 	}
 	c.metrics.cancelJobSuccessCounter.Inc(ctx)
 	return &savepointResponse, nil
@@ -263,18 +262,18 @@ func (c *FlinkJobManagerClient) GetJobs(ctx context.Context, url string) (*GetJo
 	response, err := c.executeRequest(ctx, httpGet, url, nil)
 	if err != nil {
 		c.metrics.getJobsFailureCounter.Inc(ctx)
-		return nil, errors.Wrap(err, "Get jobs API request failed")
+		return nil, GetError(err, "GetJobs", globalFailure)
 	}
 	if response != nil && !response.IsSuccess() {
 		c.metrics.getJobsFailureCounter.Inc(ctx)
 		logger.Errorf(ctx, fmt.Sprintf("GetJobs failed with response %v", response))
-		return nil, errors.New(fmt.Sprintf("GetJobs request failed with status %v", response.Status()))
+		return nil, GetError(err, "GetJobs", response.Status())
 	}
 	var getJobsResponse GetJobsResponse
 	if err = json.Unmarshal(response.Body(), &getJobsResponse); err != nil {
 		logger.Errorf(ctx, "%v", getJobsResponse)
 		logger.Errorf(ctx, "Unable to Unmarshal getJobsResponse %v, err: %v", response, err)
-		return nil, err
+		return nil, GetError(err, "GetJobs", response.Status())
 	}
 	c.metrics.getJobsSuccessCounter.Inc(ctx)
 	return &getJobsResponse, nil
@@ -285,11 +284,11 @@ func (c *FlinkJobManagerClient) GetLatestCheckpoint(ctx context.Context, url str
 	response, err := c.executeRequest(ctx, httpGet, endpoint, nil)
 	if err != nil {
 		c.metrics.getCheckpointsFailureCounter.Inc(ctx)
-		return nil, errors.Wrap(err, "get checkpoints failed")
+		return nil, GetError(err, "GetLatestCheckpoint", globalFailure)
 	}
 	if response != nil && !response.IsSuccess() {
 		c.metrics.getCheckpointsFailureCounter.Inc(ctx)
-		return nil, errors.New(fmt.Sprintf("get checkpoints failed with response %v", response))
+		return nil, GetError(err, "GetLatestCheckpoint", response.Status())
 	}
 
 	var checkpointResponse CheckpointResponse
@@ -305,11 +304,11 @@ func (c *FlinkJobManagerClient) GetTaskManagers(ctx context.Context, url string)
 	endpoint := url + taskmanagersURL
 	response, err := c.executeRequest(ctx, httpGet, endpoint, nil)
 	if err != nil {
-		return nil, errors.Wrap(err, "get taskmanagers failed")
+		return nil, GetError(err, "GetTaskManagers", globalFailure)
 	}
 
 	if response != nil && !response.IsSuccess() {
-		return nil, errors.New(fmt.Sprintf("get taskmanagers failed with response %v", response))
+		return nil, GetError(err, "GetTaskManagers", response.Status())
 	}
 
 	var taskmanagerResponse TaskManagersResponse
@@ -326,11 +325,11 @@ func (c *FlinkJobManagerClient) GetCheckpointCounts(ctx context.Context, url str
 	response, err := c.executeRequest(ctx, httpGet, endpoint, nil)
 	if err != nil {
 		c.metrics.getCheckpointsFailureCounter.Inc(ctx)
-		return nil, errors.Wrap(err, "get checkpoints failed")
+		return nil, GetError(err, "GetCheckpointCounts", globalFailure)
 	}
 	if response != nil && !response.IsSuccess() {
 		c.metrics.getCheckpointsFailureCounter.Inc(ctx)
-		return nil, errors.New(fmt.Sprintf("get checkpoints failed with response %v", response))
+		return nil, GetError(err, "GetCheckpointCounts", response.Status())
 	}
 
 	var checkpointResponse CheckpointResponse
@@ -346,11 +345,11 @@ func (c *FlinkJobManagerClient) GetJobOverview(ctx context.Context, url string,
 	endpoint := fmt.Sprintf(url+getJobsOverviewURL, jobID)
 	response, err := c.executeRequest(ctx, httpGet, endpoint, nil)
 	if err != nil {
-		return nil, errors.Wrap(err, "get job overview failed")
+		return nil, GetError(err, "GetJobOverview", globalFailure)
 	}
 	if response != nil && !response.IsSuccess() {
 		c.metrics.getCheckpointsFailureCounter.Inc(ctx)
-		return nil, errors.New(fmt.Sprintf("get job overview failed with response %v", response))
+		return nil, GetError(err, "GetJobOverview", response.Status())
 	}
 
 	var jobOverviewResponse FlinkJobOverview