Support seamless job update, job history management and manual job rollback

[elanv]

Features

• Seamless job update
  • Automate savepoint creation and Flink job update process when spec.job field is changed
• Job history management.
  • Keep job history by maintaining a certain number of k8s job resources.
• Manual rollback
  • User can rollback by specifying job revision in FlinkCluster.

CR changes

• New annotation of FlinkCluster CR: flinkclusters.flinkoperator.k8s.io/revision
  • Current job revision to display
• New field of flinkcluster.job.spec: revisionHistoryLimit
  • Number of job resources to keep
• New status.components.job.state: Updating and UpdateFailed
  • State of the job that is allowed to update is limited to Running and UpdateFailed. Except finished states - Succeded, Failed, Cancelled
  • When a job is finished, there is no big difference between creating a new job with fromSavepoint and updating.
  • Operator can't create a savepoint if job is finished.
  • The cluster may also be removed by the cleanup policy.

Job update process summary

• User modifies job spec
  • Update allowed fields: any fields except fromSavepoint
• Reconcile cycle 1
  • Build a new desired job resource
  • Request "cancel job with savepoint" for existing flink job to job manager
  • Change job state: Running --> Updating
• Reconcile cycle 2
  • Check whether "cancel job with savepoint" succeeded
  • If failed, change status and exit update process on failure: Updating --> UpdateFailed
  • If succeeded, update the incremented revision annotation of FlinkCluster.
• Reconcile cycle 3
  • Observer queries the job resource with incremented revision but it does not exist.
  • Reconciler tries to create new job because desired job exists but no observed job
• Reconcile cycle 4
  • Updater changes job status - Updating --> Running - when there is running Flink job.

Job rollback policy

• Does not rollback automatically when update fails.
  • For updates, the user will be watching the update and there may be a variety of reasons for the failure, so it seems better to provide a manual rollback measure first.
• History management and manual rollback support
  • When updating, do not delete old job but keep history like k8s Deployment-ReplicaSet relationship (Keep managed job limit like Deployment.spec.revisionHistoryLimit)
  • When a user specifies a specific job revision in the FlinkCluster CR, the operator can rollback to the existing job likewise the update process.

[elanv]

@functicons I have detailed seamless job update. May I ask for a review? If you agree, I would like to be assigned to this issue and contribute this feature.

If you think differently about this implementation details or have your own plan, please feel free to share it.

[functicons]

@elanv Thanks for the fantastic proposal! But as I understand, it might also increase the complexity of the operator. So, first I want to learn how do you compare a) adding this feature in the operator itself; b) writing a script to automate the job upgrade which first takes a savepoint, stop the existing job, then creates a new job with the savepoint? If the feature could be implemented on top of the operator as a script, we might consider adding such a script, it is easier to maintain and keeps the core operator simple. What do you think?

[elanv]

I think it's important to keep it simple. So I excluded the rollback handling when the update failed.

If you have a script-enabled environment, I think automation scripts will be enough. However, I am supporting Flink job control in java application. I think there may be a demand to deal with Flink job updates in applications in multiple languages as well as java. The burden of implementing the process is on its own, and if this feature is included, it will help a lot more people.

[functicons]

Sounds good, go head! The design looks good to me from a high level. I am just curious about why use annotation for flinkclusters.flinkoperator.k8s.io/revision instead of a field in job status?

[elanv]

Thanks for your review!

I missed the explanation. The reason for the revision as an annotation is that if you want to go back to a specific job version, you have to write that version in the FlinkCluster CR. Since status is observed information, it seems appropriate to use annotation to modify the revision. It could be added to CR spec also, but spec field is declarative and the rollback behavior is imperative, so I thought it would be better to use annotations.