Change the policy of the queue(DataCarrier) L1 in the metric aggregat…

…e worker to IF_POSSIBLE mode (#11554)

docs/en/changes/changes.md

@@ -39,6 +39,9 @@
 * Fix `limit` doesn't work for `findEndpoint` API in ES storage.
 * Isolate MAL CounterWindow cache by metric name.
 * Fix JDBC Log query order.
+* Change the DataCarrier IF_POSSIBLE strategy to use ArrayBlockingQueue implementation.
+* Change the policy of the queue(DataCarrier) in the L1 metric aggregate worker to IF_POSSIBLE mode. 
+* Add self-observability metric `metrics_aggregator_abandon` to count the number of abandon metrics.
 
 #### UI
 

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

@@ -25,6 +25,7 @@
 import org.apache.skywalking.oap.server.core.analysis.metrics.Metrics;
 import org.apache.skywalking.oap.server.core.worker.AbstractWorker;
 import org.apache.skywalking.oap.server.library.datacarrier.DataCarrier;
+import org.apache.skywalking.oap.server.library.datacarrier.buffer.BufferStrategy;
 import org.apache.skywalking.oap.server.library.datacarrier.consumer.BulkConsumePool;
 import org.apache.skywalking.oap.server.library.datacarrier.consumer.ConsumerPoolFactory;
 import org.apache.skywalking.oap.server.library.datacarrier.consumer.IConsumer;
@@ -46,6 +47,7 @@ public class MetricsAggregateWorker extends AbstractWorker<Metrics> {
     private AbstractWorker<Metrics> nextWorker;
     private final DataCarrier<Metrics> dataCarrier;
     private final MergableBufferedData<Metrics> mergeDataCache;
+    private CounterMetrics abandonCounter;
     private CounterMetrics aggregationCounter;
     private long lastSendTime = 0;
 
@@ -68,7 +70,7 @@ public class MetricsAggregateWorker extends AbstractWorker<Metrics> {
             queueBufferSize = 1_000;
         }
         this.dataCarrier = new DataCarrier<>(
-            "MetricsAggregateWorker." + modelName, name, queueChannelSize, queueBufferSize);
+            "MetricsAggregateWorker." + modelName, name, queueChannelSize, queueBufferSize, BufferStrategy.IF_POSSIBLE);
 
         BulkConsumePool.Creator creator = new BulkConsumePool.Creator(
             name, BulkConsumePool.Creator.recommendMaxSize() * 2, 20);
@@ -82,6 +84,11 @@ public class MetricsAggregateWorker extends AbstractWorker<Metrics> {
         MetricsCreator metricsCreator = moduleDefineHolder.find(TelemetryModule.NAME)
                                                           .provider()
                                                           .getService(MetricsCreator.class);
+        abandonCounter = metricsCreator.createCounter(
+            "metrics_aggregator_abandon", "The abandon number of rows received in aggregation",
+            new MetricsTag.Keys("metricName", "level", "dimensionality"),
+            new MetricsTag.Values(modelName, "1", "minute")
+        );
         aggregationCounter = metricsCreator.createCounter(
             "metrics_aggregation", "The number of rows in aggregation",
             new MetricsTag.Keys("metricName", "level", "dimensionality"),
@@ -95,7 +102,9 @@ public class MetricsAggregateWorker extends AbstractWorker<Metrics> {
      */
     @Override
     public final void in(Metrics metrics) {
-        dataCarrier.produce(metrics);
+        if (!dataCarrier.produce(metrics)) {
+            abandonCounter.inc();
+        }
     }
 
     /**

oap-server/server-library/library-datacarrier-queue/src/main/java/org/apache/skywalking/oap/server/library/datacarrier/buffer/ArrayBlockingQueueBuffer.java

@@ -40,8 +40,10 @@ public class ArrayBlockingQueueBuffer<T> implements QueueBuffer<T> {
 
     @Override
     public boolean save(T data) {
-        //only BufferStrategy.BLOCKING
         try {
+            if (BufferStrategy.IF_POSSIBLE.equals(strategy)) {
+                return queue.offer(data);
+            }
             queue.put(data);
         } catch (InterruptedException e) {
             // Ignore the error

oap-server/server-library/library-datacarrier-queue/src/main/java/org/apache/skywalking/oap/server/library/datacarrier/buffer/Channels.java

@@ -35,11 +35,7 @@ public Channels(int channelSize, int bufferSize, IDataPartitioner<T> partitioner
         this.strategy = strategy;
         bufferChannels = new QueueBuffer[channelSize];
         for (int i = 0; i < channelSize; i++) {
-            if (BufferStrategy.BLOCKING.equals(strategy)) {
-                bufferChannels[i] = new ArrayBlockingQueueBuffer<>(bufferSize, strategy);
-            } else {
-                bufferChannels[i] = new Buffer<>(bufferSize, strategy);
-            }
+            bufferChannels[i] = new ArrayBlockingQueueBuffer<>(bufferSize, strategy);
         }
         // noinspection PointlessArithmeticExpression
         size = 1L * channelSize * bufferSize; // it's not pointless, it prevents numeric overflow before assigning an integer to a long

oap-server/server-library/library-datacarrier-queue/src/main/java/org/apache/skywalking/oap/server/library/datacarrier/consumer/ConsumerThread.java

@@ -20,7 +20,6 @@
 
 import java.util.ArrayList;
 import java.util.List;
-import org.apache.skywalking.oap.server.library.datacarrier.buffer.Buffer;
 import org.apache.skywalking.oap.server.library.datacarrier.buffer.QueueBuffer;
 
 public class ConsumerThread<T> extends Thread {
@@ -88,9 +87,6 @@ void shutdown() {
         running = false;
     }
 
-    /**
-     * DataSource is a refer to {@link Buffer}.
-     */
     class DataSource {
         private QueueBuffer<T> sourceBuffer;