Fix the priority setting doesn't work of the ALS analyzers (#6121)

CHANGES.md

@@ -43,6 +43,7 @@ Release Notes.
 * Fix the uuid field in GRPCConfigWatcherRegister is not updated.
 * Support Envoy {AccessLog,Metrics}Service API V3.
 * Adopt the [MAL](docs/en/concepts-and-designs/mal.md) in Envoy metrics service analyzer.
+* Fix the priority setting doesn't work of the ALS analyzers.
 
 #### UI
 * Fix un-removed tags in trace query.

oap-server/server-receiver-plugin/envoy-metrics-receiver-plugin/src/main/java/org/apache/skywalking/oap/server/receiver/envoy/AccessLogServiceGRPCHandler.java

@@ -18,6 +18,7 @@
 
 package org.apache.skywalking.oap.server.receiver.envoy;
 
+import io.envoyproxy.envoy.data.accesslog.v3.HTTPAccessLogEntry;
 import io.envoyproxy.envoy.service.accesslog.v2.AccessLogServiceGrpc;
 import io.envoyproxy.envoy.service.accesslog.v3.StreamAccessLogsMessage;
 import io.envoyproxy.envoy.service.accesslog.v3.StreamAccessLogsResponse;
@@ -29,6 +30,7 @@ import org.apache.skywalking.aop.server.receiver.mesh.TelemetryDataDispatcher;
 import org.apache.skywalking.apm.network.servicemesh.v3.ServiceMeshMetric;
 import org.apache.skywalking.oap.server.library.module.ModuleManager;
 import org.apache.skywalking.oap.server.library.module.ModuleStartException;
+import org.apache.skywalking.oap.server.library.util.CollectionUtils;
 import org.apache.skywalking.oap.server.receiver.envoy.als.ALSHTTPAnalysis;
 import org.apache.skywalking.oap.server.receiver.envoy.als.Role;
 import org.apache.skywalking.oap.server.telemetry.TelemetryModule;
@@ -47,7 +49,8 @@ public class AccessLogServiceGRPCHandler extends AccessLogServiceGrpc.AccessLogS
     private final HistogramMetrics histogram;
     private final CounterMetrics sourceDispatcherCounter;
 
-    public AccessLogServiceGRPCHandler(ModuleManager manager, EnvoyMetricReceiverConfig config) throws ModuleStartException {
+    public AccessLogServiceGRPCHandler(ModuleManager manager,
+                                       EnvoyMetricReceiverConfig config) throws ModuleStartException {
         ServiceLoader<ALSHTTPAnalysis> alshttpAnalyses = ServiceLoader.load(ALSHTTPAnalysis.class);
         envoyHTTPAnalysisList = new ArrayList<>();
         for (String httpAnalysisName : config.getAlsHTTPAnalysis()) {
@@ -62,9 +65,18 @@ public class AccessLogServiceGRPCHandler extends AccessLogServiceGrpc.AccessLogS
         LOGGER.debug("envoy HTTP analysis: " + envoyHTTPAnalysisList);
 
         MetricsCreator metricCreator = manager.find(TelemetryModule.NAME).provider().getService(MetricsCreator.class);
-        counter = metricCreator.createCounter("envoy_als_in_count", "The count of envoy ALS metric received", MetricsTag.EMPTY_KEY, MetricsTag.EMPTY_VALUE);
-        histogram = metricCreator.createHistogramMetric("envoy_als_in_latency", "The process latency of service ALS metric receiver", MetricsTag.EMPTY_KEY, MetricsTag.EMPTY_VALUE);
-        sourceDispatcherCounter = metricCreator.createCounter("envoy_als_source_dispatch_count", "The count of envoy ALS metric received", MetricsTag.EMPTY_KEY, MetricsTag.EMPTY_VALUE);
+        counter = metricCreator.createCounter(
+            "envoy_als_in_count", "The count of envoy ALS metric received", MetricsTag.EMPTY_KEY,
+            MetricsTag.EMPTY_VALUE
+        );
+        histogram = metricCreator.createHistogramMetric(
+            "envoy_als_in_latency", "The process latency of service ALS metric receiver", MetricsTag.EMPTY_KEY,
+            MetricsTag.EMPTY_VALUE
+        );
+        sourceDispatcherCounter = metricCreator.createCounter(
+            "envoy_als_source_dispatch_count", "The count of envoy ALS metric received", MetricsTag.EMPTY_KEY,
+            MetricsTag.EMPTY_VALUE
+        );
     }
 
     @Override
@@ -93,9 +105,10 @@ public class AccessLogServiceGRPCHandler extends AccessLogServiceGrpc.AccessLogS
                     StreamAccessLogsMessage.LogEntriesCase logCase = message.getLogEntriesCase();
 
                     if (LOGGER.isDebugEnabled()) {
-                        LOGGER.debug("Messaged is identified from Envoy[{}], role[{}] in [{}]. Received msg {}", identifier
-                            .getNode()
-                            .getId(), role, logCase, message);
+                        LOGGER.debug(
+                            "Messaged is identified from Envoy[{}], role[{}] in [{}]. Received msg {}", identifier
+                                .getNode()
+                                .getId(), role, logCase, message);
                     }
 
                     switch (logCase) {
@@ -103,10 +116,16 @@ public class AccessLogServiceGRPCHandler extends AccessLogServiceGrpc.AccessLogS
                             StreamAccessLogsMessage.HTTPAccessLogEntries logs = message.getHttpLogs();
 
                             List<ServiceMeshMetric.Builder> sourceResult = new ArrayList<>();
-                            for (ALSHTTPAnalysis analysis : envoyHTTPAnalysisList) {
-                                logs.getLogEntryList().forEach(log -> {
-                                    sourceResult.addAll(analysis.analysis(identifier, log, role));
-                                });
+                            for (final HTTPAccessLogEntry log : logs.getLogEntryList()) {
+                                for (ALSHTTPAnalysis analysis : envoyHTTPAnalysisList) {
+                                    final List<ServiceMeshMetric.Builder> result =
+                                        analysis.analysis(identifier, log, role);
+                                    if (CollectionUtils.isNotEmpty(result)) {
+                                        // Once the analysis has results, don't need to continue analysis in lower priority analyzers.
+                                        sourceResult.addAll(result);
+                                        break;
+                                    }
+                                }
                             }
 
                             sourceDispatcherCounter.inc(sourceResult.size());