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\n  baseBackoffDuration: 10ms\n  maxBackoffDuration: 50ms\n  maxErrDuration: 40s\n "
 
 	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

@@ -3,6 +3,8 @@ package v1alpha1
 import (
 	"fmt"
 
+	"github.com/lyft/flinkk8soperator/pkg/controller/flink/client"
+
 	apiv1 "k8s.io/api/core/v1"
 
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
@@ -151,14 +153,16 @@ type FlinkJobStatus struct {
 }
 
 type FlinkApplicationStatus struct {
-	Phase            FlinkApplicationPhase `json:"phase"`
-	StartedAt        *metav1.Time          `json:"startedAt,omitempty"`
-	LastUpdatedAt    *metav1.Time          `json:"lastUpdatedAt,omitempty"`
-	Reason           string                `json:"reason,omitempty"`
-	ClusterStatus    FlinkClusterStatus    `json:"clusterStatus,omitempty"`
-	JobStatus        FlinkJobStatus        `json:"jobStatus"`
-	FailedDeployHash string                `json:"failedUpdateHash,omitEmpty"`
-	DeployHash       string                `json:"deployHash"`
+	Phase            FlinkApplicationPhase         `json:"phase"`
+	StartedAt        *metav1.Time                  `json:"startedAt,omitempty"`
+	LastUpdatedAt    *metav1.Time                  `json:"lastUpdatedAt,omitempty"`
+	Reason           string                        `json:"reason,omitempty"`
+	ClusterStatus    FlinkClusterStatus            `json:"clusterStatus,omitempty"`
+	JobStatus        FlinkJobStatus                `json:"jobStatus"`
+	FailedDeployHash string                        `json:"failedUpdateHash,omitEmpty"`
+	DeployHash       string                        `json:"deployHash"`
+	RetryCount       int32                         `json:"retryCount,omitEmpty"`
+	LastSeenError    *client.FlinkApplicationError `json:"lastSeenError,omitEmpty"`
 }
 
 func (in *FlinkApplicationStatus) GetPhase() FlinkApplicationPhase {

pkg/controller/config/config.go

@@ -11,16 +11,18 @@ 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"`
+	BaseBackoffDuration   config.Duration `json:"baseBackoffDuration" pflag:"\"100ms\",Determines the base backoff for exponential retries."`
+	MaxBackoffDuration    config.Duration `json:"maxBackoffDuration" pflag:"\"30s\",Determines the max backoff for exponential retries."`
+	MaxErrDuration        config.Duration `json:"maxErrDuration" pflag:"\"5m\",Determines the max time to wait on errors."`
 }
 
 func GetConfig() *Config {