schongloo opened a new pull request, #1231:
URL: https://github.com/apache/flink-kubernetes-operator/pull/1231
## What is the purpose of the change
This PR adds the `ADVANCED` transition mode to Blue/Green Deployments
(FLIP-504), for review, without the automatic gate injection of #1043: the job
places the gate in its own code. It is #1043 without the JVM agent and the
StreamGraph rewriting, so it no longer relies on `@Internal` Flink APIs, and it
adds about 3,200 fewer lines (5,359 instead of 8,571). The coordination, the
teardown, the gate timeout and the aborts are the same code. Automatic
injection could follow as its own PR on top of this one: it is the revert of
this branch's last commit.
The `BASIC` mode from FLIP-503 switches whole deployments: once the new
deployment is stable the old one is retired, and while both run they write the
same records. `ADVANCED` coordinates the two jobs record by record. A gate in
both jobs decides which job writes each record from the record's event time and
the job's watermark. Every record that arrives within the job's watermark is
written exactly once across the transition. The old deployment is torn down
only once all its gate subtasks have handed over, and with a savepoint, so
sinks that commit on checkpoints commit what it wrote. The limits of that
guarantee (late records right around the cutover, an abort after the cutover
point) are listed under [Known limitations](#known-limitations).
The mode is opt-in: `BASIC` stays the default, and existing Blue/Green
deployments are unaffected.
Besides the unit tests, an end-to-end test runs a transition of a job built
for the purpose and checks that every record is written exactly once, against
Flink 1.20, 2.0, 2.1 and 2.2, natively and standalone (see [Verifying this
change](#verifying-this-change)). An abort after the cutover point is not
covered end to end yet.
Suggested reading order:
1. The Advanced Transition Mode section of
`docs/content/docs/managing/bluegreen-deployments.md`, for the behavior and its
guarantees.
2. The operator side: the Blue/Green state machine in
`BlueGreenDeploymentService`, and `BlueGreenTransitionUtils`.
3. The gate itself, in `flink-kubernetes-operator-bluegreen-client`:
`GateProcessFunction` and `WatermarkGateProcessFunction`.
## Brief change log
**API**
- `FlinkBlueGreenDeployment` gains `spec.transitionMode` (`BASIC` /
`ADVANCED`). `BASIC` is the default on both the CRD schema and the spec type,
so deployments that omit it keep working. The field is immutable after the
initial deployment: a change is rejected, the spec reverted and a Warning event
emitted.
**Operator**
- In `ADVANCED` mode the operator keeps a ConfigMap per Blue/Green pair,
through which it and the gates of both jobs coordinate. Writes are a
compare-and-set on the resourceVersion: the first write wins, and writes for
another transition are refused. Each side writes only its own turn: the active
gate signals that it waits for the cutover point (the watermark toggle), and
only the standby sets it, from its own watermark.
- It sets `bluegreen.active-deployment-type` and `bluegreen.configmap.name`
in the `flinkConfiguration` of both deployments, the keys the gate reads from
the job's configuration.
- The standby is torn down once every gate subtask has handed over, as
reported by a `bluegreenGateHandOverDone` gauge read over the REST API. It is
stopped with a savepoint, then deleted.
- `kubernetes.operator.bluegreen.gate.timeout` (defaults to the abort grace
period, and must be longer than the deployment deletion delay) bounds the wait
for the gate: past it, the transition aborts. A job without a gate therefore
aborts at the timeout. Every abort hands the gate back to the deployment that
keeps running, and an abort after the cutover point redeploys it from the
transition savepoint.
- `ADVANCED` rejects the `STATELESS` upgrade mode. In either mode, a
transition that changes the state backend is rejected while the savepoints are
native, and a canonical savepoint that fails on Flink 2.x names the native
format in the error.
**Gate (`flink-kubernetes-operator-bluegreen-client`)**
- `GateProcessFunction` defines the gate contract, and
`WatermarkGateProcessFunction` implements it. A job creates it with
`WatermarkGateProcessFunction.create(config, extractor)`, passing its
configuration and a `WatermarkExtractor` that returns each record's event time.
The job's watermark drives only the gate's state, and nothing falls back to the
wall clock.
- The active gate passes every record once its own watermark passes the
toggle, so late records are not dropped after the hand-over.
- The gate logs what each subtask does with its records once each time it
changes, at INFO.
- One client artifact serves Flink 1.x and 2.x jobs (see
[Artifacts](#artifacts)). It brings fabric8's JDK HTTP client, and leaves its
Flink dependencies to the Flink distribution.
**Tests and docs**
- End-to-end: `test_bluegreen_advanced.sh` with the new
`examples/flink-bluegreen-advanced-example` job, and the stateless, last-state
and ingress rotation Blue/Green tests, all in the CI matrix.
- The client tests run against Flink 1.20, 2.0, 2.1 and 2.2 in CI.
- User documentation, see [Documentation](#documentation).
## Using it
```yaml
spec:
transitionMode: ADVANCED
```
```java
StreamExecutionEnvironment env =
StreamExecutionEnvironment.getExecutionEnvironment();
events.process(
WatermarkGateProcessFunction.create(
env.getConfiguration().toMap(), (MyEvent event) ->
event.getEventTime()))
.name("BlueGreen-Gate")
.uid("bluegreen-gate")
.sinkTo(sink);
```
The gate goes in front of the operator that writes the records, for a sink
with a committer its writer. With the same parallelism, Flink chains it to that
operator: in the example job the chain is `BlueGreen-Gate -> Sink: Writer ->
Sink: Committer`, on Flink 1.20 and 2.2 alike.
<a name="hand-over"></a>
### Hand-over guarantees
- Every record that arrives within the job's watermark is written exactly
once: the standby writes records older than the toggle until its watermark
passes it, the active the rest.
- Late records are written by the standby before the crossing and by the
active after it. Only late records arriving right around the crossing can be
written twice or dropped, since the two jobs' watermarks do not advance in
lockstep.
- A record without a timestamp counts as older than the toggle.
- An abort before the toggle is set needs nothing, since the standby still
passed every record. After it, the survivor is redeployed from the transition
savepoint, so records written since may be written twice, never lost.
## Requirements
- **Java 17.** The gate client is Java 17 bytecode, like the rest of the
operator, so a job on Java 11 fails with `UnsupportedClassVersionError` when it
loads the gate. `BASIC` and non-Blue/Green deployments are not affected.
- **A `DataStream` to place the gate on**, in the job's code.
<a name="artifacts"></a>
## Artifacts
The client is one artifact,
`flink-kubernetes-operator-bluegreen-client:<version>`, for jobs on Flink 1.x
and 2.x. Built against the operator's Flink version, every Flink class, method
and field it references resolves with the same descriptor and kind in Flink
1.19, 1.20, 2.0, 2.1 and 2.2, and the end-to-end jobs on Flink 2.x bundle it as
built. Its Flink dependencies are `provided`, so it does not bring the
operator's Flink version into the job.
<a name="known-limitations"></a>
## Known limitations
- `ADVANCED` needs a change to the job: the gate is placed in its code. SQL
jobs that write through `INSERT INTO` have no `DataStream` to place it on.
- `ADVANCED` needs event-time watermarks: without them there is no toggle,
and the transition aborts after the gate timeout.
- Late records arriving right around the crossing can be written twice or
dropped, and an abort after the cutover point may write records twice.
- A gate subtask that receives no records counts as done after 30 s without
one.
- On Flink 2.x, operators using the async state API cannot take a canonical
savepoint, so a job using them needs
`kubernetes.operator.savepoint.format.type: NATIVE` for every transition except
a stateless one, in either mode, and cannot change its state backend through a
transition.
## Verifying this change
- `WatermarkGateProcessFunctionTest`: the hand-over between a standby and an
active gate, late records, records without a timestamp, record time only, the
turns each side writes, the logging and the hand-over gauge.
- `GateKubernetesServiceTest`: the compare-and-set against a Kubernetes mock
server.
- `FlinkBlueGreenDeploymentControllerTest`: the `ADVANCED` state machine,
from the ConfigMap and validation to the teardown, the gate timeout and the
aborts, plus the savepoint format checks.
- `BlueGreenTransitionUtilsTest`, `BlueGreenUtilsTest` and
`FlinkBlueGreenDeploymentSpecDiffTest`: the validations and `transitionMode`
immutability.
`mvn verify` passes across all modules, except the three tests that need a
local kube-apiserver. The client tests also pass against Flink 2.0.1, 2.1.1 and
2.2.1. Built against Flink 1.20 and against 2.2, the example job's `main()`
yields the chain `BlueGreen-Gate -> Sink: Writer -> Sink: Committer`, and its
JobGraph builds, which serializes the gate with its extractor.
TODO before leaving draft: link the run of the Blue/Green end-to-end jobs
for this branch.
## Does this pull request potentially affect one of the following parts:
- Dependencies (does it add or upgrade a dependency): **yes**. One new
module, the client. It brings fabric8's `kubernetes-httpclient-jdk`, takes its
Flink dependencies as `provided`, and excludes the Flink dependencies of
`flink-kubernetes-operator-api`. No existing module's dependencies change.
- The public API, i.e., is any changes to the `CustomResourceDescriptors`:
**yes**. `FlinkBlueGreenDeployment` gains `spec.transitionMode` (`BASIC` /
`ADVANCED`, default `BASIC`), immutable after the initial deployment.
- Core observer or reconciler logic that is regularly executed: **yes**, for
`FlinkBlueGreenDeployment` only. `FlinkService` gains `getMinSubtaskMetrics`,
used only by the Blue/Green teardown. `FlinkDeployment` and `FlinkSessionJob`
reconciliation is untouched, and `BASIC` behaviour is preserved.
## Documentation
A new Advanced Transition Mode section under [Blue/Green
Deployments](docs/content/docs/managing/bluegreen-deployments.md) covers how
the hand-over works and what it guarantees, the teardown and aborts, the
requirements, enabling it with the gate in the job's code, the extractor and
gate placement. The Blue/Green requirements also name the savepoint format of
transitions. Zero-Downtime Upgrades, Controllers, the custom resource overview,
the glossary, the generated CRD reference and the Chinese pages are updated to
match.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
