Setup for blue/green deployment mode (#180)

* Working version 1

* Create setup for blue green deploys

* [WIP] Setup status sub-resource for blue green deploys

* Updates

* fix bug

* Fixes

* Make running jobs calculation idempotent

* Fix bugs

* Reset running jobs in recovering phase

* Make status index calculation simpler

* Add container env and annotations

* Update CRD to v1beta2

* Update CRD to v1beta2

* Fix CRD update issues

* Fix lint

* Merge master and restore v1beta1 to original version

* Upgrade integ test to v1beta2

* Backward compatibility changes

* Work around status subresource bug

* Rename status array to VersionStatuses and add comment on k8s bug

* Remove DesiredApplicationCount

* Minor updates

* Minor updates

* Initialize counter

* Handle edge case for jobId

* Debug

* Debug

* fixes

* Fix edge case

* Fix unit tests

* Debug logs

* Fix overwriting of versionstatuses

* Remove debug logs

* Merge master

* Revert CRD upgrade

* Keep Status.ClusterStatus and Status.JobStatus unchanged for Dual mode

* Remove unwarranted changes

* Make import name more descriptive

* Revert file name to add_v1beta1

* Remove an unnecessary diff

* Remove an unnecessary diff

pkg/apis/app/v1beta1/types.go

@@ -168,19 +168,37 @@ 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:"failedDeployHash,omitempty"`
-	RollbackHash       string                 `json:"rollbackHash,omitempty"`
-	DeployHash         string                 `json:"deployHash"`
-	SavepointTriggerID string                 `json:"savepointTriggerId,omitempty"`
-	SavepointPath      string                 `json:"savepointPath,omitempty"`
-	RetryCount         int32                  `json:"retryCount,omitempty"`
-	LastSeenError      *FlinkApplicationError `json:"lastSeenError,omitempty"`
+	Phase           FlinkApplicationPhase `json:"phase"`
+	StartedAt       *metav1.Time          `json:"startedAt,omitempty"`
+	LastUpdatedAt   *metav1.Time          `json:"lastUpdatedAt,omitempty"`
+	Reason          string                `json:"reason,omitempty"`
+	DeployVersion   string                `json:"deployVersion,omitempty"`
+	UpdatingVersion string                `json:"updatingVersion,omitempty"`
+	// To ensure backward compatibility, repeat ClusterStatus and JobStatus
+	ClusterStatus      FlinkClusterStatus              `json:"clusterStatus,omitempty"`
+	JobStatus          FlinkJobStatus                  `json:"jobStatus,omitempty"`
+	VersionStatuses    []FlinkApplicationVersionStatus `json:"versionStatuses,omitempty"`
+	FailedDeployHash   string                          `json:"failedDeployHash,omitempty"`
+	RollbackHash       string                          `json:"rollbackHash,omitempty"`
+	DeployHash         string                          `json:"deployHash"`
+	SavepointTriggerID string                          `json:"savepointTriggerId,omitempty"`
+	SavepointPath      string                          `json:"savepointPath,omitempty"`
+	RetryCount         int32                           `json:"retryCount,omitempty"`
+	LastSeenError      *FlinkApplicationError          `json:"lastSeenError,omitempty"`
+}
+
+type FlinkApplicationVersion string
+
+const (
+	BlueFlinkApplication  FlinkApplicationVersion = "Blue"
+	GreenFlinkApplication FlinkApplicationVersion = "Green"
+)
+
+type FlinkApplicationVersionStatus struct {
+	Version       FlinkApplicationVersion `json:"appVersion,omitempty"`
+	VersionHash   string                  `json:"versionHash,omitempty"`
+	ClusterStatus FlinkClusterStatus      `json:"clusterStatus,omitempty"`
+	JobStatus     FlinkJobStatus          `json:"jobStatus,omitempty"`
 }
 
 func (in *FlinkApplicationStatus) GetPhase() FlinkApplicationPhase {
@@ -226,6 +244,8 @@ const (
 	FlinkApplicationRecovering      FlinkApplicationPhase = "Recovering"
 	FlinkApplicationRollingBackJob  FlinkApplicationPhase = "RollingBackJob"
 	FlinkApplicationDeployFailed    FlinkApplicationPhase = "DeployFailed"
+	FlinkApplicationDualRunning     FlinkApplicationPhase = "DualRunning"
+	FlinkApplicationTeardown        FlinkApplicationPhase = "Teardown"
 )
 
 var FlinkApplicationPhases = []FlinkApplicationPhase{
@@ -240,17 +260,35 @@ var FlinkApplicationPhases = []FlinkApplicationPhase{
 	FlinkApplicationRecovering,
 	FlinkApplicationDeployFailed,
 	FlinkApplicationRollingBackJob,
+	FlinkApplicationDualRunning,
+	FlinkApplicationTeardown,
 }
 
 func IsRunningPhase(phase FlinkApplicationPhase) bool {
 	return phase == FlinkApplicationRunning || phase == FlinkApplicationDeployFailed
 }
 
+func IsBlueGreenDeploymentMode(mode DeploymentMode) bool {
+	// Backaward compatibility between v1beta1 and v1beta1
+	if mode == DeploymentModeDual {
+		return false
+	}
+	return mode == DeploymentModeBlueGreen
+}
+
+func GetMaxRunningJobs(mode DeploymentMode) int32 {
+	if IsBlueGreenDeploymentMode(mode) {
+		return int32(2)
+	}
+	return int32(1)
+}
+
 type DeploymentMode string
 
 const (
-	DeploymentModeSingle DeploymentMode = "Single"
-	DeploymentModeDual   DeploymentMode = "Dual"
+	DeploymentModeSingle    DeploymentMode = "Single"
+	DeploymentModeDual      DeploymentMode = "Dual"
+	DeploymentModeBlueGreen DeploymentMode = "BlueGreen"
 )
 
 type DeleteMode string