[Improve][Zeta] Add RequestSlotOperation observability metrics

[dybyte]

Purpose of this pull request

This PR adds lightweight observability metrics for RequestSlotOperation in Zeta.

RequestSlotOperation is used by the active master during resource allocation to request slots from worker nodes. When job startup or resource allocation becomes slow, operators currently have limited visibility into whether slot requests are succeeding, returning no available slot, or failing at the master-to-worker invocation layer.

This PR adds master-side Prometheus exports for:

• request_slot_operation_total{address, result}
• request_slot_operation_last_invocation_latency_ms{address}
• request_slot_operation_max_invocation_latency_ms{address}

The result label has the following meanings:

• success: the worker returned an assigned slot.
• no_slot: the request reached a worker and completed, but the worker did not return a suitable slot.
• failure: the master-to-worker invocation failed or the operation completed exceptionally.

These metrics are intentionally aggregate-only. They do not add job, worker, slot, or resource-profile labels, to avoid high-cardinality telemetry.

Does this PR introduce any user-facing change?

Yes.

This PR adds new Prometheus telemetry metrics to the Zeta metrics endpoint and updates the telemetry documentation in both English and Chinese.

It does not change slot allocation behavior, retry behavior, operation serialization, or runtime scheduling semantics.

How was this patch tested?

• Added collector-level tests for RequestSlotOperationExports
  • returns empty on non-master nodes
  • returns empty when the coordinator is not ready
  • exports metric families when ResourceManager stats are available
  • verifies result labels and exported sample values

Check list

• If any new Jar binary package adding in your PR, please add License Notice according
  New License Guide
• If necessary, please update the documentation to describe the new feature. https://github.com/apache/seatunnel/tree/dev/docs
• If necessary, please update incompatible-changes.md to describe the incompatibility caused by this PR.
• If you are contributing the connector code, please check that the following files are updated:
  1. Update plugin-mapping.properties and add new connector information in it
  2. Update the pom file of seatunnel-dist
  3. Add ci label in label-scope-conf
  4. Add e2e testcase in seatunnel-e2e
  5. Update connector plugin_config

[DanielLeens]

Thanks for the contribution. I reviewed the new Zeta resource-manager telemetry path from the slot request call site through the Prometheus exporter.

What this PR adds

• User pain: operators currently cannot tell whether slot allocation is slow, failing, returning no slot, or simply waiting on general resource pressure.
• Fix approach: record master-side RequestSlotOperation success/no_slot/failure counts plus last/max invocation latency, export them through telemetry, and document the new metrics in EN/ZH docs.
• One-line summary: the implementation is well scoped and does not change allocation behavior, but the core recording path needs a small behavior test before merge.

Runtime path checked

Job needs slots
  -> AbstractResourceManager.applyResources(jobId, resourceProfiles, tagFilter)
      -> new ResourceRequestHandler(...).request(tagFilter)
          -> requestSlots()
              -> singleResourceRequestToMember(index, resource, worker)
                  -> sendToMember(new RequestSlotOperation(jobId, resource), workerAddress)
                  -> whenComplete(...)
                      -> error != null: recordRequestSlotOperationFailure(elapsedMs)
                      -> slotProfile == null: recordRequestSlotOperationNoSlot(elapsedMs)
                      -> slotProfile != null: recordRequestSlotOperationSuccess(elapsedMs)

Prometheus scrape
  -> RequestSlotOperationExports.collect()
      -> isMaster && isCoordinatorReady
      -> getCoordinatorService().getResourceManager()
      -> getRequestSlotOperationStats()
      -> request_slot_operation_total{result=...}
      -> request_slot_operation_last/max_invocation_latency_ms

Findings

Issue 1: the exporter is tested, but the real RequestSlotOperation recording path is not

• Location: seatunnel-engine/seatunnel-engine-server/src/main/java/org/apache/seatunnel/engine/server/resourcemanager/ResourceRequestHandler.java:232
• Why it matters: the added test verifies that RequestSlotOperationExports can export a mocked RequestSlotOperationStats object, but it does not verify that ResourceRequestHandler actually updates AbstractResourceManager stats on the three real completion paths: success, no_slot, and failure.
• Risk: a future change could accidentally stop recording one of these paths while the exporter test still passes, because the exporter test bypasses the real slot-request lifecycle entirely.
• Suggested fix: add a focused unit test around the existing fake resource manager / request-slot retry test utilities that drives one success, one null-slot result, and one failed request, then asserts getRequestSlotOperationStats() updates successCount, noSlotCount, failureCount, and the latency fields. This should not require a real cluster or E2E test.
• Severity: Medium
• Already raised by others: no.

Testing/stability: the new exporter tests have no sleeps, ports, external services, or order-sensitive assertions. Stability rating: has risk, because the key recording behavior is not covered yet.

I did not run local tests; this review is based on local diff inspection, call-chain tracing, docs comparison, and the current GitHub check state.

Merge conclusion

Conclusion: can merge after fixes

1. Blocking items

• Issue 1: add a targeted behavior test for the real RequestSlotOperation stats recording paths.

2. Non-blocking suggestions

• RequestSlotOperationExports.collect() can use new ArrayList<>() instead of the raw new ArrayList().

Current gate: Build is still queued: https://github.com/apache/seatunnel/runs/86580989038. Once the small test gap is fixed and CI is green, I am happy to re-review.

[DanielLeens]

Thanks for the update. I re-reviewed the latest head and the source-level blocker from my previous round is closed for me now.

What changed

• User pain: operators could not tell whether slot allocation was slow, failed, returned no slot, or was simply waiting on general resource pressure.
• Fix approach: the master now records RequestSlotOperation success / no_slot / failure counts plus last/max invocation latency, and exports them through telemetry with EN/ZH docs.
• One-line summary: this is a low-risk observability improvement for the Zeta slot request path, without changing allocation behavior.

Runtime path checked

Job needs execution resources
  -> AbstractResourceManager.applyResources(jobId, resourceProfiles, tagFilter)
      -> new ResourceRequestHandler(...).request(tagFilter)
          -> requestSlots()
              -> singleResourceRequestToMember(index, resource, worker)
                  -> sendToMember(new RequestSlotOperation(jobId, resource), workerAddress)
                  -> whenComplete(...)
                      -> error != null
                          -> recordRequestSlotOperationFailure(elapsedMs)
                          -> preserve the existing failure propagation
                      -> slotProfile == null
                          -> recordRequestSlotOperationNoSlot(elapsedMs)
                          -> keep the existing no-slot retry/failure behavior
                      -> slotProfile != null
                          -> recordRequestSlotOperationSuccess(elapsedMs)
                          -> heartbeat + addSlotToCacheMap

Prometheus scrape
  -> RequestSlotOperationExports.collect()
      -> return empty when this node is not master or coordinator is not ready
      -> getCoordinatorService().getResourceManager()
      -> getRequestSlotOperationStats()
      -> export request_slot_operation_total{result=...}
      -> export last/max invocation latency gauges

Review result

• The previous blocker is fixed: ResourceManagerTest now drives the real applyResources() path for success/no_slot and the exceptional path for failure, instead of only testing the exporter with a mocked stats object.
• The earlier non-blocking raw collection suggestion is also fixed: RequestSlotOperationExports.collect() now uses new ArrayList<>().
• I do not see a new source-side blocker in the latest head. The implementation keeps the existing slot allocation, retry, exception, and release behavior intact.
• Test stability looks good: the new tests do not use sleeps, fixed ports, containers, wall-clock assertions, or order-sensitive checks. The asynchronous path asserts counts only, while deterministic latency values are covered through direct aggregation calls.
• EN/ZH telemetry docs are consistent for the new metric names, labels, and result meanings.

I did not run local tests; this review is based on the full current diff, call-chain tracing, the added tests, the EN/ZH docs, and the current GitHub check state.

Merge conclusion

Conclusion: can merge after the remaining gate clears

1. Blocking items

• No remaining source-side blocker from my side. The previous test coverage gap is resolved.
• The remaining gate is CI / merge status: Build is still in progress for the latest head, so I am not approving ahead of the required checks.

2. Non-blocking suggestions

• None from this round.

Overall, this looks like a focused and maintainable Zeta observability improvement. One extra queue fact I recorded: the branch is currently diverged from dev (ahead_by=3, behind_by=8). Since the current Build is still running rather than failing, I would not ask for a sync-only follow-up yet; let's wait for the latest Build result first.

Note: my earlier changes requested entry may still remain visible in GitHub's review summary until a write-capable maintainer or the final review gate updates it. From Daniel's technical review side, the source blocker is cleared and the remaining item is the current CI / merge gate.

[nzw921rx]

Thanks for the update. I re-reviewed the latest head. The source-level blocker from the previous round is closed, but I found one new blocker in the Prometheus collection path that is not covered by the existing review.

What changed

• User pain: operators could not tell whether slot allocation was slow, failed, returned no slot, or was simply waiting on general resource pressure.
• Fix approach: the master now records RequestSlotOperation success / no_slot / failure counts plus last/max invocation latency, and exports them through telemetry with EN/ZH docs.
• One-line summary: the recording path is focused and preserves the existing slot-request outcomes, but the exporter currently makes a read-only Prometheus scrape initialize the resource manager.

Runtime path checked

Job needs execution resources
  -> AbstractResourceManager.applyResources(jobId, resourceProfiles, tagFilter)
      -> new ResourceRequestHandler(...).request(tagFilter)
          -> requestSlots()
              -> singleResourceRequestToMember(index, resource, worker)
                  -> sendToMember(new RequestSlotOperation(jobId, resource), workerAddress)
                  -> whenComplete(...)
                      -> error != null
                          -> recordRequestSlotOperationFailure(elapsedMs)
                          -> preserve the existing failure propagation
                      -> slotProfile == null
                          -> recordRequestSlotOperationNoSlot(elapsedMs)
                          -> keep the existing no-slot retry/failure behavior
                      -> slotProfile != null
                          -> recordRequestSlotOperationSuccess(elapsedMs)
                          -> heartbeat + addSlotToCacheMap

Prometheus scrape
  -> RequestSlotOperationExports.collect()
      -> return empty when this node is not master or coordinator is not ready
      -> getCoordinatorService().getResourceManager()
          -> resourceManager == null
              -> create ResourceManager
              -> manager.init()
                  -> send SyncWorkerProfileOperation to every live member
                  -> join all returned futures
      -> getRequestSlotOperationStats()
      -> export request_slot_operation_total{result=...}
      -> export last/max invocation latency gauges

Review result

• The previous blocker is fixed: ResourceManagerTest now drives the real applyResources() path for success/no_slot and the exceptional path for failure, instead of only testing the exporter with mocked stats.
• The earlier non-blocking raw collection suggestion is also fixed: RequestSlotOperationExports.collect() now uses new ArrayList<>().
• The slot-request recording path preserves the existing allocation, retry, exception, heartbeat, and release behavior.
• The counter names, labels, and values are consistent between the exporter tests and the EN/ZH telemetry docs.
• The new tests do not use sleeps, fixed ports, containers, wall-clock assertions, or order-sensitive checks.
• The remaining source-side blocker is the exporter calling the lazy-initializing resource-manager accessor from the Prometheus scrape path.

Findings

Issue 1: a Prometheus scrape can initialize and block on the ResourceManager

• Location: seatunnel-engine/seatunnel-engine-server/src/main/java/org/apache/seatunnel/engine/server/telemetry/metrics/exports/RequestSlotOperationExports.java:44
• Why it matters: RequestSlotOperationExports.collect() calls CoordinatorService#getResourceManager(). That method is explicitly a lazy-initializing accessor: when the field is null, it creates the resource manager and calls manager.init() while holding the coordinator monitor. AbstractResourceManager#init() then sends SyncWorkerProfileOperation to every live member and joins all returned futures.
• Risk: on a newly active master, a read-only /metrics scrape can create runtime state, issue cluster RPCs, block while waiting for members, and make concurrent scheduling/resource requests wait for the same initialization lock. This is a scheduling-side effect introduced by telemetry and contradicts the PR's stated guarantee that allocation/runtime behavior is unchanged. The subsequent resourceManager == null check cannot prevent this because the lazy getter initializes the instance instead of returning null.
• Suggested fix: expose a non-initializing accessor such as getInitializedResourceManager() or peekResourceManager() and use it from the collector. If no resource manager exists yet, return no samples or zero-valued samples without creating it. Add a focused test proving that collection on an active coordinator with an uninitialized resource manager does not invoke the factory or init().
• Severity: High / blocking.
• Already raised by others: no. The existing review only covered the recording-path tests and raw collection type.

Relevant code:

• RequestSlotOperationExports.collect()
• CoordinatorService#getResourceManager()
• AbstractResourceManager#init()

I did not run local Maven commands. This review is based on the full current diff, call-chain and lifecycle tracing, the added tests, the EN/ZH docs, existing review comments, and the current GitHub check state.

Merge conclusion

Conclusion: can merge after fixes

1. Blocking items

• Issue 1: make the telemetry collection path observe an already initialized resource manager without initializing it.
• The current GitHub Build check has completed with failure. The only failing job is the unrelated Flink HTTP E2E HttpIT.testSourceToAssertSink (expected: <0> but was: <1>), so it appears unrelated to this Zeta-only change, but the required check still needs a successful rerun before merge.

2. Non-blocking suggestions

• None from this round.

Overall, the slot-operation metrics and their direct tests are focused and maintainable. Once the scrape-side initialization is removed and the required CI gate is green, this should be ready for another review.

[DanielLeens]

Thanks @nzw921rx for flagging that exporter path. I rechecked the current head against the same call chain, and I agree this is still a real blocker on the latest revision.

The core problem is that RequestSlotOperationExports.collect() still calls CoordinatorService#getResourceManager(), and that accessor lazily creates and initializes the resource manager. So a read-only metrics scrape can still trigger manager.init(), send sync operations to members, and block on cluster work before returning metrics. That is not just a telemetry read; it changes runtime behavior from the scrape path.

So from Daniel's side I would keep this PR in changes requested territory until the collector switches to a non-initializing accessor (for example a peek / getInitialized... style path) and proves with a focused test that collecting metrics on an active coordinator does not bootstrap the resource manager.

I do not have an extra source-side blocker to add on top of the one you described, but I do agree this one needs a code update before merge.

[SEZ9]

Thanks for working on this. I re-reviewed the latest head 8ef702a541d7 from scratch.

Review conclusion

Conclusion: LGTM, can merge

Nice work — thanks again for the contribution. Happy to discuss any of the above.

[DanielLeens]

Thanks @SEZ9 for the rereview. I would like to respectfully push back on the latest LGTM here.

I rechecked the current head against the same exporter path, and my earlier blocker still stands: RequestSlotOperationExports.collect() still reaches CoordinatorService#getResourceManager(), and that accessor can lazily initialize the resource manager during a read-only metrics scrape. In other words, collecting telemetry can still change runtime behavior and send cluster sync work from the scrape path.

Until the collector switches to a non-initializing accessor and we have a focused regression test proving a metrics scrape does not bootstrap the resource manager, I still would not merge this revision from Daniel's side.