[FLINK-29134] fetch metrics may cause oom(ThreadPool task pile up)

• Type: Improvement
• Status: Closed
• Priority: Major
• Resolution: Fixed
• Affects Version/s: 1.11.0, 1.15.2
• Fix Version/s: 1.17.0, 1.15.3, 1.16.1
• Component/s: Runtime / Metrics
• Labels: pull-request-available
• Assignee: Yuxin Tan
• Reporter: Sitan Pang
• Created: 30/Aug/22 06:43
• Updated: 26/Oct/22 02:49
• Resolved: 26/Oct/22 02:49

Description

When we queryMetrics we use thread pool to process the data which are returned by TMs. 

private void queryMetrics(final MetricQueryServiceGateway queryServiceGateway) {
    LOG.debug("Query metrics for {}.", queryServiceGateway.getAddress());

    queryServiceGateway
            .queryMetrics(timeout)
            .whenCompleteAsync(
                    (MetricDumpSerialization.MetricSerializationResult result, Throwable t) -> {
                        if (t != null) {
                            LOG.debug("Fetching metrics failed.", t);
                        } else {
                            metrics.addAll(deserializer.deserialize(result));
                        }
                    },
                    executor);
} 

The only condition we will fetch metrics is update time is larger than updateInterval

public void update() {
    synchronized (this) {
        long currentTime = System.currentTimeMillis();
        if (currentTime - lastUpdateTime > updateInterval) {
            lastUpdateTime = currentTime;
            fetchMetrics();
        }
    }
} 

Therefore, if we could not process the data in update-interval-time, metrics data will accumulate.

Besides, webMonitorEndpoint, restHandlers and metrics share the thread pool.

When we open ui, it maybe even worse.

final ScheduledExecutorService executor =
        WebMonitorEndpoint.createExecutorService(
                configuration.getInteger(RestOptions.SERVER_NUM_THREADS),
                configuration.getInteger(RestOptions.SERVER_THREAD_PRIORITY),
                "DispatcherRestEndpoint");

final long updateInterval =
        configuration.getLong(MetricOptions.METRIC_FETCHER_UPDATE_INTERVAL);
final MetricFetcher metricFetcher =
        updateInterval == 0
                ? VoidMetricFetcher.INSTANCE
                : MetricFetcherImpl.fromConfiguration(
                        configuration,
                        metricQueryServiceRetriever,
                        dispatcherGatewayRetriever,
                        executor);

webMonitorEndpoint =
        restEndpointFactory.createRestEndpoint(
                configuration,
                dispatcherGatewayRetriever,
                resourceManagerGatewayRetriever,
                blobServer,
                executor,
                metricFetcher,
                highAvailabilityServices.getClusterRestEndpointLeaderElectionService(),
                fatalErrorHandler); 

 

 

 

Issue Links

• links to: GitHub Pull Request #21132