Expose fault tolerant execution and filesystem exchange metrics via JMX

[arhimondr]

Description

Exposes fault tolerant execution related operational metrics via JMX to enable live monitoring

Is this change a fix, improvement, new feature, refactoring, or other?

Improvement

Is this a change to the core query engine, a connector, client library, or the SPI interfaces? (be specific)

Core, Exchange

How would you describe this change to a non-technical end user or system administrator?

N/A

Related issues, pull requests, and links

-

Documentation

(x) No documentation is needed.
( ) Sufficient documentation is included in this PR.
( ) Documentation PR is available with #prnumber.
( ) Documentation issue #issuenumber is filed, and can be handled later.

Release notes

(x) No release notes entries required.
( ) Release notes entries required with the following suggested text:

# Section
* Fix some things. ({issue}`issuenumber`)

[arhimondr]

I had to move exchange plugin from the io.trino.plugin.exchange package to the io.trino.plugin.exchange.filesystem to avoid any potential naming clashes with any future implementations when exposing JMX metrics. After changing the package name I also renamed the module to make it consistent with the package naming.

[linzebing]

When will this happen in practice?

[arhimondr]

Rather unlikely, but we have a similar safeguard in other places

[linzebing]

What is this for?

[arhimondr]

This is to export the JMX beans under a specific prefix

[linzebing]

This is not on the critical path, do we need to record here too?

[arhimondr]

It might be useful for tracking (for example if for some reason it starts to fail)

[linzebing]

For my understanding, can you briefly explain what these two annotations do?

[arhimondr]

Methods annotated with @Managed are exported via JMX. @Nested tells the framework to recurse into the object returned by a method and export nested methods annotated with @Managed.

[linzebing]

Is this an optimization? Why do we need a class member blocked?

[arhimondr]

This is to make sure only a single blocked future exists and tracked.

[linzebing]

I don't quite understand, why not simply the following:

return stats.getExchangeSourceBlocked().record(toCompletableFuture(
                nonCancellationPropagating(
                        whenAnyComplete(readers.stream()
                                .map(ExchangeStorageReader::isBlocked)
                                .collect(toImmutableList())))));

[arhimondr]

Because isBlocked is called by multiple threads concurrently and a different feature may be returned to a different thread skewing the metric while ideally we would like to keep our measurements as close as possible to the time it takes for the entire ExchangeSource to transition from "blocked" state to "non-blocked".

[losipiuk]

why not throw?

[arhimondr]

Exceptions thrown out of a listener are not logged. So it's more of a "log it or loose it".

[losipiuk]

I know currently dependent tasks are not executing until upstream task comptes. Yet it will change. In such case would we make downstream task as "FAILED" or "ABORTED".
If the latter then we should also compute stats for "ABORTED" tasks. It would be important to understand how much effort we are wasting on those.
Maybe we can just merge FAILED and ABORTED?

[arhimondr]

I'm going to add private final ExecutionStats abortedTasks = new ExecutionStats(); and store the stats for both, ABORTED and CANCELLED tasks there. Not sure if it makes sense to track metrics separately.

[losipiuk]

Should we have memory and network stats here too?
It feels not costly to add them and then we can decide which are the most important for us for tracking.

[arhimondr]

I'm not sure if there's a reliable metric for network as the network traffic can occur at many different levels (connector / exchanges / coordinator-to-worker communication). Though it certainly feels like it would make sense to record peak memory utilization. Let me add it.

[losipiuk]

nit: Use Ticker/Stopwatch pair instead.

[losipiuk]

LGTM