Kaustubh Butte created FLINK-40407:
--------------------------------------
Summary: [FLINK-autoscaler] Add reason tags to
`autoscaler.scalings` counter to explain why scaling was triggered
Key: FLINK-40407
URL: https://issues.apache.org/jira/browse/FLINK-40407
Project: Flink
Issue Type: Improvement
Components: Autoscaler
Affects Versions: 1.20.1, 2.0.0
Reporter: Kaustubh Butte
Currently, the `autoscaler.scalings` counter is incremented whenever a scaling
decision is applied, but there is no metric indicating WHY the autoscaler
decided to scale. Understanding the scaling trigger is critical for:
- Incident triage ("did it scale because of a traffic spike or a Kafka lag
buildup?")
- Validating autoscaler behavior ("is this job scaling due to genuine load or
a false-positive signal?")
- Building targeted alerts ("alert on backlog-driven scale-ups but not routine
load adjustments")
The information needed to classify the reason is already computed in the
autoscaler's evaluation pipeline but is discarded after the scaling decision is
made.
*Proposed reasons:*
*Scale-up:*
|Reason|Condition|Meaning|
|`backlog`|`CATCH_UP_DATA_RATE.current > 0` and contributes >10% of
`targetCapacity`|Consumer lag accumulated; scaling up to catch up|
|`high_load`|`CATCH_UP_DATA_RATE ~ 0` and `TRUE_PROCESSING_RATE <
SCALE_UP_RATE_THRESHOLD`|Sustained input rate exceeds processing capacity|
|`input_spike`|`TARGET_DATA_RATE.current > 1.5 *
TARGET_DATA_RATE.average`|Sudden traffic increase|
*Scale-down:*
|Reason|Condition|Meaning|
|`low_utilization`|`TRUE_PROCESSING_RATE >
SCALE_DOWN_RATE_THRESHOLD`|Processing capacity far exceeds demand|
*Classification logic:*
The target capacity is already decomposed in
`AutoScalerUtils.getTargetProcessingCapacity()`:
```
targetCapacity = lagCatchupRate + restartCatchupRate + inputTargetAtUtilization
```
A single vertex can have multiple reasons simultaneously (e.g., backlog AND
high load both contributing to targetCapacity). And a single scaling cycle
evaluates all vertices, each potentially with different reasons. Therefore,
reasons are collected as a set per vertex, then aggregated across all vertices
into a sorted pipe-delimited string.
*Per-vertex classification* (inside
`JobVertexScaler.computeScaleTargetParallelism()`, inline with the decision):
- If `lagCatchupRate / targetCapacity > 0.1` then add `backlog`
- If `TARGET_DATA_RATE.current / TARGET_DATA_RATE.average > 1.5` then add
`input_spike`
- If scale-up and no specific reason matched, add `high_load` (fallback)
- If scale-down, add `low_util`
*Job-level aggregation* (in `ScalingExecutor`):
- Collect all reason sets from all vertices in the scaling summary
- Merge into a single `TreeSet<String>` (alphabetically sorted)
- Join with `|` delimiter: e.g., `"backlog|high_load"`
This approach prevents algorithm-reason drift because the classification is
done inline in the same function that computes the scale factor, reading the
same computed variables. If the algorithm changes, the developer sees the
reason-tagging code right there.
*Maximum cardinality:* 15 distinct tag values (2^4 - 1 combinations of 4 atomic
reasons). In practice fewer, since `high_load` is only added when no other
scale-up reason applies.
*Implementation:*
1. Add `Set<String> scaleReasons` to `ParallelismChange` in `JobVertexScaler`,
tag reasons inline alongside scale factor computation
2. Add `Set<String> reasons` to `ScalingSummary`, propagated from
`ParallelismChange`
3. In `ScalingExecutor`, aggregate reasons from all vertices into a sorted
pipe-delimited string
4. In `AutoscalerFlinkMetrics`, register tagged sub-counters dynamically using
`MetricGroup.addGroup("reason", reasonTag).counter("scalings")`
5. Keep the existing `numScalings` counter incrementing for backward
compatibility
*Result in metric reporters:*
- `autoscaler.scalings` (total, backward compatible)
- `autoscaler.scalings \{reason=backlog}` – pure backlog catch-up
- `autoscaler.scalings \{reason=backlog|high_load}` – backlog + steady-state
overload
- `autoscaler.scalings \{reason=high_load}` – sustained input rate exceeds
capacity
- `autoscaler.scalings \{reason=input_spike}` – sudden traffic increase
- `autoscaler.scalings \{reason=low_util}` – scale-down, overcapacity
- (and other sorted combinations up to 15 total)
*Files affected:*
- `JobVertexScaler.java` – add `Set<String> scaleReasons` to
`ParallelismChange`, tag reasons inline
- `ScalingSummary.java` – add `Set<String> reasons` field
- `ScalingExecutor.java` – aggregate reasons across vertices into
pipe-delimited string
- `AutoscalerFlinkMetrics.java` – add dynamically registered tagged counters
- `JobAutoScalerImpl.java` – pass aggregated reason string to metrics
All changes are in the `flink-autoscaler` module (standalone,
platform-agnostic). Works with both K8s and YARN deployments.
*Relationship to other tickets:*
This is a companion to the `balanced` counter disambiguation ticket. Both can
be implemented together or independently. Together they provide complete
observability over every autoscaler decision cycle: why it scaled, why it
didn't scale, or why it errored.
--
This message was sent by Atlassian Jira
(v8.20.10#820010)
