[jira] [Created] (FLINK-40407) [FLINK-autoscaler] Add reason tags to `autoscaler.scalings` counter to explain why scaling was triggered

["Kaustubh Butte (Jira)" <ji...@apache.org>]

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)