This is an automated email from the ASF dual-hosted git repository. exceptionfactory pushed a commit to branch support/nifi-1.x in repository https://gitbox.apache.org/repos/asf/nifi.git
commit 62ff15a8389f8509370f8c5206c053efebf2eac2 Author: Timea Barna <[email protected]> AuthorDate: Tue Jan 16 13:51:55 2024 +0100 NIFI-12506 Added Threading for Status Analytics Retrieval This closes #8251 Signed-off-by: David Handermann <[email protected]> --- .../apache/nifi/web/StandardNiFiServiceFacade.java | 34 ++---- .../PredictionBasedParallelProcessingService.java | 35 ++++++ .../util/ThreadPoolParallelProcessingService.java | 123 +++++++++++++++++++++ .../src/main/resources/nifi-web-api-context.xml | 5 + 4 files changed, 174 insertions(+), 23 deletions(-) diff --git a/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-web/nifi-web-api/src/main/java/org/apache/nifi/web/StandardNiFiServiceFacade.java b/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-web/nifi-web-api/src/main/java/org/apache/nifi/web/StandardNiFiServiceFacade.java index 049710d081..c2d5a7fd16 100644 --- a/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-web/nifi-web-api/src/main/java/org/apache/nifi/web/StandardNiFiServiceFacade.java +++ b/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-web/nifi-web-api/src/main/java/org/apache/nifi/web/StandardNiFiServiceFacade.java @@ -96,7 +96,6 @@ import org.apache.nifi.controller.service.ControllerServiceReference; import org.apache.nifi.controller.service.ControllerServiceState; import org.apache.nifi.controller.status.ProcessGroupStatus; import org.apache.nifi.controller.status.ProcessorStatus; -import org.apache.nifi.controller.status.analytics.StatusAnalytics; import org.apache.nifi.controller.status.history.ProcessGroupStatusDescriptor; import org.apache.nifi.diagnostics.SystemDiagnostics; import org.apache.nifi.events.BulletinFactory; @@ -354,6 +353,7 @@ import org.apache.nifi.web.revision.RevisionUpdate; import org.apache.nifi.web.revision.StandardRevisionClaim; import org.apache.nifi.web.revision.StandardRevisionUpdate; import org.apache.nifi.web.revision.UpdateRevisionTask; +import org.apache.nifi.web.util.PredictionBasedParallelProcessingService; import org.apache.nifi.web.util.SnippetUtils; import org.slf4j.Logger; import org.slf4j.LoggerFactory; @@ -443,6 +443,8 @@ public class StandardNiFiServiceFacade implements NiFiServiceFacade { private final JvmMetricsRegistry jvmMetricsRegistry = new JvmMetricsRegistry(); private final ConnectionAnalyticsMetricsRegistry connectionAnalyticsMetricsRegistry = new ConnectionAnalyticsMetricsRegistry(); private final ClusterMetricsRegistry clusterMetricsRegistry = new ClusterMetricsRegistry(); + private PredictionBasedParallelProcessingService parallelProcessingService; + // ----------------------------------------- // Synchronization methods @@ -6116,28 +6118,10 @@ public class StandardNiFiServiceFacade implements NiFiServiceFacade { PrometheusMetricsUtil.createAggregatedNifiMetrics(nifiMetricsRegistry, aggregatedMetrics, instanceId,ROOT_PROCESS_GROUP, rootPGName, rootPGId); // Get Connection Status Analytics (predictions, e.g.) - Set<Connection> connections = controllerFacade.getFlowManager().findAllConnections(); - for (Connection c : connections) { - // If a ResourceNotFoundException is thrown, analytics hasn't been enabled - try { - final StatusAnalytics statusAnalytics = controllerFacade.getConnectionStatusAnalytics(c.getIdentifier()); - PrometheusMetricsUtil.createConnectionStatusAnalyticsMetrics(connectionAnalyticsMetricsRegistry, - statusAnalytics, - instanceId, - "Connection", - c.getName(), - c.getIdentifier(), - c.getProcessGroup().getIdentifier(), - c.getSource().getName(), - c.getSource().getIdentifier(), - c.getDestination().getName(), - c.getDestination().getIdentifier() - ); - PrometheusMetricsUtil.aggregateConnectionPredictionMetrics(aggregatedMetrics, statusAnalytics.getPredictions()); - } catch (ResourceNotFoundException rnfe) { - break; - } - } + Collection<Map<String, Long>> predictions = parallelProcessingService.createConnectionStatusAnalyticsMetricsAndCollectPredictions( + controllerFacade, connectionAnalyticsMetricsRegistry, instanceId); + + predictions.forEach((prediction) -> PrometheusMetricsUtil.aggregateConnectionPredictionMetrics(aggregatedMetrics, prediction)); PrometheusMetricsUtil.createAggregatedConnectionStatusAnalyticsMetrics(connectionAnalyticsMetricsRegistry, aggregatedMetrics, instanceId, ROOT_PROCESS_GROUP, rootPGName, rootPGId); // Create a query to get all bulletins @@ -6497,4 +6481,8 @@ public class StandardNiFiServiceFacade implements NiFiServiceFacade { public void setFlowRegistryDAO(FlowRegistryDAO flowRegistryDao) { this.flowRegistryDAO = flowRegistryDao; } + + public void setParallelProcessingService(PredictionBasedParallelProcessingService parallelProcessingService) { + this.parallelProcessingService = parallelProcessingService; + } } diff --git a/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-web/nifi-web-api/src/main/java/org/apache/nifi/web/util/PredictionBasedParallelProcessingService.java b/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-web/nifi-web-api/src/main/java/org/apache/nifi/web/util/PredictionBasedParallelProcessingService.java new file mode 100644 index 0000000000..397ae8d881 --- /dev/null +++ b/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-web/nifi-web-api/src/main/java/org/apache/nifi/web/util/PredictionBasedParallelProcessingService.java @@ -0,0 +1,35 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package org.apache.nifi.web.util; + +import org.apache.nifi.prometheus.util.ConnectionAnalyticsMetricsRegistry; +import org.apache.nifi.web.controller.ControllerFacade; + +import java.util.Collection; +import java.util.Map; + +public interface PredictionBasedParallelProcessingService { + /** + * @return a connection prediction collection + * + * @param controllerFacade controller facade + * @param connectionAnalyticsMetricsRegistry connection analytics metrics registry + * @param instanceId instance id of the flow controller + */ + Collection<Map<String, Long>> createConnectionStatusAnalyticsMetricsAndCollectPredictions( + ControllerFacade controllerFacade, ConnectionAnalyticsMetricsRegistry connectionAnalyticsMetricsRegistry, String instanceId); +} diff --git a/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-web/nifi-web-api/src/main/java/org/apache/nifi/web/util/ThreadPoolParallelProcessingService.java b/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-web/nifi-web-api/src/main/java/org/apache/nifi/web/util/ThreadPoolParallelProcessingService.java new file mode 100644 index 0000000000..60c759b96f --- /dev/null +++ b/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-web/nifi-web-api/src/main/java/org/apache/nifi/web/util/ThreadPoolParallelProcessingService.java @@ -0,0 +1,123 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package org.apache.nifi.web.util; + +import org.apache.nifi.connectable.Connection; +import org.apache.nifi.controller.status.analytics.StatusAnalytics; +import org.apache.nifi.prometheus.util.ConnectionAnalyticsMetricsRegistry; +import org.apache.nifi.prometheus.util.PrometheusMetricsUtil; +import org.apache.nifi.util.FormatUtils; +import org.apache.nifi.util.NiFiProperties; +import org.apache.nifi.web.controller.ControllerFacade; + +import javax.ws.rs.WebApplicationException; +import java.io.Closeable; +import java.io.IOException; +import java.util.ArrayList; +import java.util.Collection; +import java.util.Collections; +import java.util.Map; +import java.util.Set; +import java.util.UUID; +import java.util.concurrent.CountDownLatch; +import java.util.concurrent.ForkJoinPool; +import java.util.concurrent.ForkJoinWorkerThread; +import java.util.concurrent.TimeUnit; + +public class ThreadPoolParallelProcessingService implements PredictionBasedParallelProcessingService, Closeable { + private static final int PARALLEL_PROCESSING_THREADS = 6; + private boolean analyticsEnabled; + private ForkJoinPool parallelProcessingThreadPool; + private long parallelProcessingTimeout; + + + public ThreadPoolParallelProcessingService(final NiFiProperties properties) { + // We need to make processing timeout shorter than the web request timeout as if they overlap Jetty may throw IllegalStateException + parallelProcessingTimeout = Math.round(FormatUtils.getPreciseTimeDuration( + properties.getProperty(NiFiProperties.WEB_REQUEST_TIMEOUT, "1 min"), TimeUnit.MILLISECONDS)) - 5000; + + analyticsEnabled = Boolean.parseBoolean( + properties.getProperty(NiFiProperties.ANALYTICS_PREDICTION_ENABLED, Boolean.FALSE.toString())); + + if (analyticsEnabled) { + this.parallelProcessingThreadPool = createParallelProcessingThreadPool(); + } + } + + @Override + public Collection<Map<String, Long>> createConnectionStatusAnalyticsMetricsAndCollectPredictions( + final ControllerFacade controllerFacade, final ConnectionAnalyticsMetricsRegistry connectionAnalyticsMetricsRegistry, final String instanceId) { + + Collection<Map<String, Long>> predictions = Collections.synchronizedList(new ArrayList<>()); + + if (!analyticsEnabled) { + return predictions; + } + + final Set<Connection> connections = controllerFacade.getFlowManager().findAllConnections(); + final CountDownLatch countDownLatch = new CountDownLatch(connections.size()); + try { + parallelProcessingThreadPool.submit( + () -> connections.parallelStream().forEach((c) -> { + try { + final StatusAnalytics statusAnalytics = controllerFacade.getConnectionStatusAnalytics(c.getIdentifier()); + PrometheusMetricsUtil.createConnectionStatusAnalyticsMetrics(connectionAnalyticsMetricsRegistry, + statusAnalytics, + instanceId, + "Connection", + c.getName(), + c.getIdentifier(), + c.getProcessGroup().getIdentifier(), + c.getSource().getName(), + c.getSource().getIdentifier(), + c.getDestination().getName(), + c.getDestination().getIdentifier() + ); + predictions.add(statusAnalytics.getPredictions()); + } finally { + countDownLatch.countDown(); + } + })); + } finally { + try { + boolean finished = countDownLatch.await(parallelProcessingTimeout, TimeUnit.MILLISECONDS); + if (!finished) { + throw new WebApplicationException("Populating flow metrics timed out"); + } + } catch (InterruptedException e) { + throw new WebApplicationException("Populating flow metrics cancelled"); + } + } + return predictions; + } + + private ForkJoinPool createParallelProcessingThreadPool() { + final ForkJoinPool.ForkJoinWorkerThreadFactory factory = pool -> { + final ForkJoinWorkerThread worker = ForkJoinPool.defaultForkJoinWorkerThreadFactory.newThread(pool); + worker.setName("analytics-prediction-parallel-processing-thread-" + UUID.randomUUID()); + return worker; + }; + return new ForkJoinPool(PARALLEL_PROCESSING_THREADS, factory, null, false); + } + + @Override + public void close() throws IOException { + if (parallelProcessingThreadPool != null) { + parallelProcessingThreadPool.shutdown(); + } + } +} diff --git a/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-web/nifi-web-api/src/main/resources/nifi-web-api-context.xml b/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-web/nifi-web-api/src/main/resources/nifi-web-api-context.xml index 33c83b90ef..2c4127126d 100644 --- a/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-web/nifi-web-api/src/main/resources/nifi-web-api-context.xml +++ b/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-web/nifi-web-api/src/main/resources/nifi-web-api-context.xml @@ -366,6 +366,7 @@ <property name="leaderElectionManager" ref="leaderElectionManager" /> <property name="flowRegistryDAO" ref="flowRegistryDAO" /> <property name="parameterContextDAO" ref="parameterContextDAO" /> + <property name="parallelProcessingService" ref="parallelProcessingService" /> </bean> <!-- component ui extension configuration context --> @@ -786,4 +787,8 @@ <!-- NiFi locking --> <bean id="serviceFacadeLock" class="org.apache.nifi.web.NiFiServiceFacadeLock"/> + <bean id="parallelProcessingService" class="org.apache.nifi.web.util.ThreadPoolParallelProcessingService"> + <constructor-arg ref="nifiProperties" /> + </bean> + </beans>
