Metrics combination API supports abandoning results (#6318)

CHANGES.md

@@ -9,11 +9,12 @@ Release Notes.
 
 #### Java Agent
 * Remove invalid mysql configuration in agent.config.
-* Add net.bytebuddy.agent.builder.AgentBuilder.RedefinitionStrategy.Listener to show detail message when redifine errors occur
+* Add net.bytebuddy.agent.builder.AgentBuilder.RedefinitionStrategy.Listener to show detail message when redefine errors occur
 
 
 #### OAP-Backend
 * Allow user-defined `JAVA_OPTS` in the startup script.
+* Metrics combination API supports abandoning results.
 
 #### UI
 

oap-server/exporter/src/test/java/org/apache/skywalking/oap/server/exporter/provider/grpc/MockMetrics.java

@@ -29,8 +29,8 @@ public class MockMetrics extends Metrics {
     }
 
     @Override
-    public void combine(Metrics metrics) {
-
+    public boolean combine(Metrics metrics) {
+        return true;
     }
 
     @Override

oap-server/server-alarm-plugin/src/test/java/org/apache/skywalking/oap/server/core/alarm/provider/RunningRuleTest.java

@@ -422,8 +422,8 @@ public class RunningRuleTest {
         }
 
         @Override
-        public void combine(Metrics metrics) {
-
+        public boolean combine(Metrics metrics) {
+            return true;
         }
 
         @Override
@@ -479,8 +479,8 @@ public class RunningRuleTest {
         }
 
         @Override
-        public void combine(Metrics metrics) {
-
+        public boolean combine(Metrics metrics) {
+            return true;
         }
 
         @Override
@@ -531,8 +531,8 @@ public class RunningRuleTest {
         }
 
         @Override
-        public void combine(Metrics metrics) {
-
+        public boolean combine(Metrics metrics) {
+            return true;
         }
 
         @Override

oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/analysis/data/MergableBufferedData.java

@@ -51,7 +51,10 @@ public class MergableBufferedData<METRICS extends Metrics> implements BufferedDa
         if (existed == null) {
             buffer.put(id, data);
         } else {
-            existed.combine(data);
+            final boolean isAbandoned = !existed.combine(data);
+            if (isAbandoned) {
+                buffer.remove(id);
+            }
         }
     }
 

oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/analysis/manual/endpoint/EndpointTraffic.java

@@ -108,8 +108,8 @@ public class EndpointTraffic extends Metrics {
     }
 
     @Override
-    public void combine(final Metrics metrics) {
-
+    public boolean combine(final Metrics metrics) {
+        return true;
     }
 
     @Override

oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/analysis/manual/instance/InstanceTraffic.java

@@ -73,7 +73,7 @@ public class InstanceTraffic extends Metrics {
     private JsonObject properties;
 
     @Override
-    public void combine(final Metrics metrics) {
+    public boolean combine(final Metrics metrics) {
         final InstanceTraffic instanceTraffic = (InstanceTraffic) metrics;
         this.lastPingTimestamp = instanceTraffic.getLastPingTimestamp();
         if (instanceTraffic.getProperties() != null && instanceTraffic.getProperties().size() > 0) {
@@ -85,6 +85,7 @@ public class InstanceTraffic extends Metrics {
         if (this.getTimeBucket() > metrics.getTimeBucket()) {
             this.setTimeBucket(metrics.getTimeBucket());
         }
+        return true;
     }
 
     @Override

oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/analysis/manual/networkalias/NetworkAddressAlias.java

@@ -67,7 +67,7 @@ public class NetworkAddressAlias extends Metrics {
     private long lastUpdateTimeBucket;
 
     @Override
-    public void combine(final Metrics metrics) {
+    public boolean combine(final Metrics metrics) {
         NetworkAddressAlias alias = (NetworkAddressAlias) metrics;
         this.representServiceId = alias.getRepresentServiceId();
         this.representServiceInstanceId = alias.getRepresentServiceInstanceId();
@@ -78,6 +78,7 @@ public class NetworkAddressAlias extends Metrics {
         if (this.getTimeBucket() > metrics.getTimeBucket()) {
             this.setTimeBucket(metrics.getTimeBucket());
         }
+        return true;
     }
 
     @Override

oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/analysis/manual/relation/endpoint/EndpointRelationServerSideMetrics.java

@@ -69,8 +69,8 @@ public class EndpointRelationServerSideMetrics extends Metrics {
     }
 
     @Override
-    public void combine(Metrics metrics) {
-
+    public boolean combine(Metrics metrics) {
+        return true;
     }
 
     @Override

oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/analysis/manual/relation/instance/ServiceInstanceRelationClientSideMetrics.java

@@ -77,8 +77,8 @@ public class ServiceInstanceRelationClientSideMetrics extends Metrics {
     }
 
     @Override
-    public void combine(Metrics metrics) {
-
+    public boolean combine(Metrics metrics) {
+        return true;
     }
 
     @Override

oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/analysis/manual/relation/instance/ServiceInstanceRelationServerSideMetrics.java

@@ -77,8 +77,8 @@ public class ServiceInstanceRelationServerSideMetrics extends Metrics {
     }
 
     @Override
-    public void combine(Metrics metrics) {
-
+    public boolean combine(Metrics metrics) {
+        return true;
     }
 
     @Override

oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/analysis/manual/relation/service/ServiceRelationClientSideMetrics.java

@@ -67,8 +67,8 @@ public class ServiceRelationClientSideMetrics extends Metrics {
     }
 
     @Override
-    public void combine(Metrics metrics) {
-
+    public boolean combine(Metrics metrics) {
+        return true;
     }
 
     @Override

oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/analysis/manual/relation/service/ServiceRelationServerSideMetrics.java

@@ -67,8 +67,8 @@ public class ServiceRelationServerSideMetrics extends Metrics {
     }
 
     @Override
-    public void combine(Metrics metrics) {
-
+    public boolean combine(Metrics metrics) {
+        return true;
     }
 
     @Override

oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/analysis/manual/service/ServiceTraffic.java

@@ -123,8 +123,8 @@ public class ServiceTraffic extends Metrics {
     }
 
     @Override
-    public void combine(final Metrics metrics) {
-
+    public boolean combine(final Metrics metrics) {
+        return true;
     }
 
     @Override

oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/analysis/meter/function/HistogramFunction.java

@@ -75,16 +75,17 @@ public abstract class HistogramFunction extends Metrics implements AcceptableVal
     }
 
     @Override
-    public void combine(final Metrics metrics) {
+    public boolean combine(final Metrics metrics) {
         HistogramFunction histogram = (HistogramFunction) metrics;
 
         if (!dataset.keysEqual(histogram.getDataset())) {
             log.warn("Incompatible input [{}}] for current HistogramFunction[{}], entity {}",
                      histogram, this, entityId
             );
-            return;
+            return true;
         }
         this.dataset.append(histogram.dataset);
+        return true;
     }
 
     @Override

oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/analysis/meter/function/PercentileFunction.java

@@ -122,14 +122,14 @@ public abstract class PercentileFunction extends Metrics implements AcceptableVa
     }
 
     @Override
-    public void combine(final Metrics metrics) {
+    public boolean combine(final Metrics metrics) {
         PercentileFunction percentile = (PercentileFunction) metrics;
 
         if (!dataset.keysEqual(percentile.getDataset())) {
             log.warn("Incompatible input [{}}] for current PercentileFunction[{}], entity {}",
                      percentile, this, entityId
             );
-            return;
+            return true;
         }
         if (ranks.size() > 0) {
             IntList ranksOfThat = percentile.getRanks();
@@ -137,11 +137,11 @@ public abstract class PercentileFunction extends Metrics implements AcceptableVa
                 log.warn("Incompatible ranks size = [{}}] for current PercentileFunction[{}]",
                          ranks.size(), this.ranks.size()
                 );
-                return;
+                return true;
             } else {
                 if (!this.ranks.equals(percentile.getRanks())) {
                     log.warn("Rank {} doesn't exist in the previous ranks {}", percentile.getRanks(), ranks);
-                    return;
+                    return true;
                 }
             }
         }
@@ -149,6 +149,7 @@ public abstract class PercentileFunction extends Metrics implements AcceptableVa
         this.dataset.append(percentile.dataset);
 
         this.isCalculated = false;
+        return true;
     }
 
     @Override

oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/analysis/meter/function/avg/AvgFunction.java

@@ -80,9 +80,10 @@ public abstract class AvgFunction extends Metrics implements AcceptableValue<Lon
     }
 
     @Override
-    public final void combine(Metrics metrics) {
+    public final boolean combine(Metrics metrics) {
         AvgFunction longAvgMetrics = (AvgFunction) metrics;
         combine(longAvgMetrics.summation, longAvgMetrics.count);
+        return true;
     }
 
     @Override

oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/analysis/meter/function/avg/AvgHistogramFunction.java

@@ -101,10 +101,11 @@ public abstract class AvgHistogramFunction extends Metrics implements Acceptable
     }
 
     @Override
-    public void combine(final Metrics metrics) {
+    public boolean combine(final Metrics metrics) {
         AvgHistogramFunction histogram = (AvgHistogramFunction) metrics;
         this.summation.append(histogram.summation);
         this.count.append(histogram.count);
+        return true;
     }
 
     @Override

oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/analysis/meter/function/avg/AvgHistogramPercentileFunction.java

@@ -154,7 +154,7 @@ public abstract class AvgHistogramPercentileFunction extends Metrics implements
     }
 
     @Override
-    public void combine(final Metrics metrics) {
+    public boolean combine(final Metrics metrics) {
         AvgHistogramPercentileFunction percentile = (AvgHistogramPercentileFunction) metrics;
 
         if (ranks.size() > 0) {
@@ -162,11 +162,11 @@ public abstract class AvgHistogramPercentileFunction extends Metrics implements
                 log.warn("Incompatible ranks size = [{}}] for current PercentileFunction[{}]",
                          ranks.size(), this.ranks.size()
                 );
-                return;
+                return true;
             } else {
                 if (!this.ranks.equals(percentile.getRanks())) {
                     log.warn("Rank {} doesn't exist in the previous ranks {}", percentile.getRanks(), ranks);
-                    return;
+                    return true;
                 }
             }
         }
@@ -175,6 +175,7 @@ public abstract class AvgHistogramPercentileFunction extends Metrics implements
         this.count.append(percentile.count);
 
         this.isCalculated = false;
+        return true;
     }
 
     @Override

oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/analysis/meter/function/avg/AvgLabeledFunction.java

@@ -72,10 +72,11 @@ public abstract class AvgLabeledFunction extends Metrics implements AcceptableVa
     private DataTable value = new DataTable(30);
 
     @Override
-    public final void combine(Metrics metrics) {
+    public final boolean combine(Metrics metrics) {
         AvgLabeledFunction longAvgMetrics = (AvgLabeledFunction) metrics;
         summation.append(longAvgMetrics.summation);
         count.append(longAvgMetrics.count);
+        return true;
     }
 
     @Override

oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/analysis/meter/function/latest/LatestFunction.java

@@ -72,9 +72,10 @@ public abstract class LatestFunction extends Metrics implements AcceptableValue<
         this.value = value;
     }
 
-    @Override public final void combine(Metrics metrics) {
+    @Override public final boolean combine(Metrics metrics) {
         LatestFunction latestFunction = (LatestFunction) metrics;
         combine(latestFunction.value);
+        return true;
     }
 
     @Override public void calculate() {

oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/analysis/metrics/ApdexMetrics.java

@@ -79,10 +79,11 @@ public abstract class ApdexMetrics extends Metrics implements IntValueHolder {
     }
 
     @Override
-    public final void combine(Metrics metrics) {
+    public final boolean combine(Metrics metrics) {
         tNum += ((ApdexMetrics) metrics).tNum;
         sNum += ((ApdexMetrics) metrics).sNum;
         totalNum += ((ApdexMetrics) metrics).totalNum;
+        return true;
     }
 
     @Override

oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/analysis/metrics/CPMMetrics.java

@@ -47,9 +47,10 @@ public abstract class CPMMetrics extends Metrics implements LongValueHolder {
     }
 
     @Override
-    public final void combine(Metrics metrics) {
+    public final boolean combine(Metrics metrics) {
         CPMMetrics cpmMetrics = (CPMMetrics) metrics;
         combine(cpmMetrics.total);
+        return true;
     }
 
     @Override

oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/analysis/metrics/CountMetrics.java

@@ -42,9 +42,10 @@ public abstract class CountMetrics extends Metrics implements LongValueHolder {
     }
 
     @Override
-    public final void combine(Metrics metrics) {
+    public final boolean combine(Metrics metrics) {
         CountMetrics countMetrics = (CountMetrics) metrics;
         combine(countMetrics.value);
+        return true;
     }
 
     @Override

oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/analysis/metrics/DoubleAvgMetrics.java

@@ -54,9 +54,10 @@ public abstract class DoubleAvgMetrics extends Metrics implements DoubleValueHol
     }
 
     @Override
-    public final void combine(Metrics metrics) {
+    public final boolean combine(Metrics metrics) {
         DoubleAvgMetrics doubleAvgMetrics = (DoubleAvgMetrics) metrics;
         combine(doubleAvgMetrics.summation, doubleAvgMetrics.count);
+        return true;
     }
 
     @Override

oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/analysis/metrics/HistogramMetrics.java

@@ -75,9 +75,10 @@ public abstract class HistogramMetrics extends Metrics {
     }
 
     @Override
-    public void combine(Metrics metrics) {
+    public boolean combine(Metrics metrics) {
         HistogramMetrics histogramMetrics = (HistogramMetrics) metrics;
         this.dataset.append(histogramMetrics.dataset);
+        return true;
     }
 
     /**

oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/analysis/metrics/LongAvgMetrics.java

@@ -54,9 +54,10 @@ public abstract class LongAvgMetrics extends Metrics implements LongValueHolder
     }
 
     @Override
-    public final void combine(Metrics metrics) {
+    public final boolean combine(Metrics metrics) {
         LongAvgMetrics longAvgMetrics = (LongAvgMetrics) metrics;
         combine(longAvgMetrics.summation, longAvgMetrics.count);
+        return true;
     }
 
     @Override

oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/analysis/metrics/MaxDoubleMetrics.java

@@ -43,9 +43,10 @@ public abstract class MaxDoubleMetrics extends Metrics implements DoubleValueHol
     }
 
     @Override
-    public final void combine(Metrics metrics) {
+    public final boolean combine(Metrics metrics) {
         MaxDoubleMetrics maxDoubleMetrics = (MaxDoubleMetrics) metrics;
         combine(maxDoubleMetrics.value);
+        return true;
     }
 
     @Override

oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/analysis/metrics/MaxLongMetrics.java

@@ -46,9 +46,10 @@ public abstract class MaxLongMetrics extends Metrics implements LongValueHolder
     }
 
     @Override
-    public final void combine(Metrics metrics) {
+    public final boolean combine(Metrics metrics) {
         MaxLongMetrics maxLongMetrics = (MaxLongMetrics) metrics;
         combine(maxLongMetrics.value);
+        return true;
     }
 
     @Override

oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/analysis/metrics/Metrics.java

@@ -56,8 +56,9 @@ public abstract class Metrics extends StreamData implements StorageData {
      * Merge the given metrics instance, these two must be the same metrics type.
      *
      * @param metrics to be merged
+     * @return {@code true} if the combined metrics should be continuously processed. {@code false} means it should be abandoned, and the implementation needs to keep the data unaltered in this case.
      */
-    public abstract void combine(Metrics metrics);
+    public abstract boolean combine(Metrics metrics);
 
     /**
      * Calculate the metrics final value when required.

oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/analysis/metrics/MinDoubleMetrics.java

@@ -43,9 +43,10 @@ public abstract class MinDoubleMetrics extends Metrics implements DoubleValueHol
     }
 
     @Override
-    public final void combine(Metrics metrics) {
+    public final boolean combine(Metrics metrics) {
         MinDoubleMetrics minDoubleMetrics = (MinDoubleMetrics) metrics;
         combine(minDoubleMetrics.value);
+        return true;
     }
 
     @Override

oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/analysis/metrics/MinLongMetrics.java

@@ -43,9 +43,10 @@ public abstract class MinLongMetrics extends Metrics implements LongValueHolder
     }
 
     @Override
-    public final void combine(Metrics metrics) {
+    public final boolean combine(Metrics metrics) {
         MinLongMetrics minLongMetrics = (MinLongMetrics) metrics;
         combine(minLongMetrics.value);
+        return true;
     }
 
     @Override

oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/analysis/metrics/PercentMetrics.java

@@ -54,9 +54,10 @@ public abstract class PercentMetrics extends Metrics implements IntValueHolder {
     }
 
     @Override
-    public final void combine(Metrics metrics) {
+    public final boolean combine(Metrics metrics) {
         total += ((PercentMetrics) metrics).total;
         match += ((PercentMetrics) metrics).match;
+        return true;
     }
 
     @Override

oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/analysis/metrics/PercentileMetrics.java

@@ -77,11 +77,12 @@ public abstract class PercentileMetrics extends Metrics implements MultiIntValue
     }
 
     @Override
-    public void combine(Metrics metrics) {
+    public boolean combine(Metrics metrics) {
         this.isCalculated = false;
 
         PercentileMetrics percentileMetrics = (PercentileMetrics) metrics;
         this.dataset.append(percentileMetrics.dataset);
+        return true;
     }
 
     @Override

oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/analysis/metrics/PxxMetrics.java

@@ -78,11 +78,12 @@ public abstract class PxxMetrics extends Metrics implements IntValueHolder {
     }
 
     @Override
-    public void combine(Metrics metrics) {
+    public boolean combine(Metrics metrics) {
         this.isCalculated = false;
 
         PxxMetrics pxxMetrics = (PxxMetrics) metrics;
         this.detailGroup.append(pxxMetrics.detailGroup);
+        return true;
     }
 
     @Override

oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/analysis/metrics/RateMetrics.java

@@ -55,9 +55,10 @@ public abstract class RateMetrics extends Metrics implements IntValueHolder {
     }
 
     @Override
-    public final void combine(Metrics metrics) {
+    public final boolean combine(Metrics metrics) {
         denominator += ((RateMetrics) metrics).denominator;
         numerator += ((RateMetrics) metrics).numerator;
+        return true;
     }
 
     @Override

oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/analysis/metrics/SumMetrics.java

@@ -42,9 +42,10 @@ public abstract class SumMetrics extends Metrics implements LongValueHolder {
     }
 
     @Override
-    public final void combine(Metrics metrics) {
+    public final boolean combine(Metrics metrics) {
         SumMetrics sumMetrics = (SumMetrics) metrics;
         combine(sumMetrics.value);
+        return true;
     }
 
     @Override

oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/analysis/worker/MetricsPersistentWorker.java

@@ -174,7 +174,10 @@ public class MetricsPersistentWorker extends PersistenceWorker<Metrics> {
                     /*
                      * Merge metrics into cachedMetrics, change only happens inside cachedMetrics.
                      */
-                    cachedMetrics.combine(metrics);
+                    final boolean isAbandoned = !cachedMetrics.combine(metrics);
+                    if (isAbandoned) {
+                        continue;
+                    }
                     cachedMetrics.calculate();
                     prepareRequests.add(metricsDAO.prepareBatchUpdate(model, cachedMetrics));
                     nextWorker(cachedMetrics);

oap-server/server-core/src/test/java/org/apache/skywalking/oap/server/core/analysis/metrics/MetricsTest.java

@@ -78,8 +78,8 @@ public class MetricsTest {
         }
 
         @Override
-        public void combine(Metrics metrics) {
-
+        public boolean combine(Metrics metrics) {
+            return true;
         }
 
         @Override