MONIT-28397: Fix discrepancy in buffer file data and proxy backlog charts (#856)

Author: laullon

proxy/src/main/java/com/wavefront/agent/queueing/QueueController.java

@@ -1,20 +1,24 @@
 package com.wavefront.agent.queueing;
 
 import com.google.common.annotations.VisibleForTesting;
-import com.google.common.util.concurrent.RateLimiter;
 import com.wavefront.agent.data.DataSubmissionTask;
 import com.wavefront.agent.handlers.HandlerKey;
 import com.wavefront.common.Managed;
+import com.wavefront.common.NamedThreadFactory;
 import com.wavefront.common.Pair;
 import com.wavefront.common.TaggedMetricName;
 import com.yammer.metrics.Metrics;
 import com.yammer.metrics.core.Gauge;
 import java.io.IOException;
 import java.util.Comparator;
 import java.util.List;
-import java.util.Timer;
 import java.util.TimerTask;
+import java.util.concurrent.Executors;
+import java.util.concurrent.ScheduledExecutorService;
+import java.util.concurrent.TimeUnit;
 import java.util.concurrent.atomic.AtomicBoolean;
+import java.util.concurrent.atomic.AtomicInteger;
+import java.util.concurrent.atomic.AtomicLong;
 import java.util.function.Consumer;
 import java.util.function.Supplier;
 import java.util.logging.Logger;
@@ -32,23 +36,18 @@ public class QueueController<T extends DataSubmissionTask<T>> extends TimerTask
 
   // min difference in queued timestamps for the schedule adjuster to kick in
   private static final int TIME_DIFF_THRESHOLD_SECS = 60;
-  private static final int REPORT_QUEUE_STATS_DELAY_SECS = 15;
   private static final double MIN_ADJ_FACTOR = 0.25d;
   private static final double MAX_ADJ_FACTOR = 1.5d;
+  private static ScheduledExecutorService executor;
 
   protected final HandlerKey handlerKey;
   protected final List<QueueProcessor<T>> processorTasks;
-  @Nullable private final Consumer<Integer> backlogSizeSink;
   protected final Supplier<Long> timeProvider;
-  protected final Timer timer;
-
-  @SuppressWarnings("UnstableApiUsage")
-  protected final RateLimiter reportRateLimiter = RateLimiter.create(0.1);
-
-  private long currentWeight;
-  private int queueSize;
+  private final Consumer<Integer> backlogSizeSink;
 
   private final AtomicBoolean isRunning = new AtomicBoolean(false);
+  private final AtomicLong currentWeight = new AtomicLong();
+  private final AtomicInteger queueSize = new AtomicInteger();
 
   /**
    * @param handlerKey Pipeline handler key
@@ -79,7 +78,6 @@ public QueueController(
     this.processorTasks = processorTasks;
     this.backlogSizeSink = backlogSizeSink;
     this.timeProvider = timeProvider == null ? System::currentTimeMillis : timeProvider;
-    this.timer = new Timer("timer-queuedservice-" + handlerKey.toString());
 
     Metrics.newGauge(
         new TaggedMetricName(
@@ -92,7 +90,7 @@ public QueueController(
         new Gauge<Integer>() {
           @Override
           public Integer value() {
-            return queueSize;
+            return queueSize.get();
           }
         });
     Metrics.newGauge(
@@ -101,48 +99,11 @@ public Integer value() {
         new Gauge<Long>() {
           @Override
           public Long value() {
-            return currentWeight;
+            return currentWeight.get();
           }
         });
   }
 
-  @Override
-  public void run() {
-    // 1. grab current queue sizes (tasks count) and report to EntityProperties
-    int backlog = processorTasks.stream().mapToInt(x -> x.getTaskQueue().size()).sum();
-    queueSize = backlog;
-    if (backlogSizeSink != null) {
-      backlogSizeSink.accept(backlog);
-    }
-
-    // 2. grab queue sizes (points/etc count)
-    long totalWeight = 0L;
-    for (QueueProcessor<T> task : processorTasks) {
-      TaskQueue<T> taskQueue = task.getTaskQueue();
-      if ((taskQueue != null) && (taskQueue.weight() != null)) {
-        totalWeight += taskQueue.weight();
-      }
-    }
-    long previousWeight = currentWeight;
-    currentWeight = totalWeight;
-
-    // 3. adjust timing
-    adjustTimingFactors(processorTasks);
-
-    // 4. print stats when there's backlog
-    if ((previousWeight != 0) || (currentWeight != 0)) {
-      printQueueStats();
-      if (currentWeight == 0) {
-        logger.info(
-            "["
-                + handlerKey.getHandle()
-                + "] "
-                + handlerKey.getEntityType()
-                + " backlog has been cleared!");
-      }
-    }
-  }
-
   /**
    * Compares timestamps of tasks at the head of all backing queues. If the time difference between
    * most recently queued head and the oldest queued head (across all backing queues) is less than
@@ -177,16 +138,34 @@ static <T extends DataSubmissionTask<T>> void adjustTimingFactors(
     }
   }
 
+  @Override
+  public void run() {
+    // 1. grab current queue sizes (tasks count) and report to EntityProperties
+    int backlog = processorTasks.stream().mapToInt(x -> x.getTaskQueue().size()).sum();
+    if (backlogSizeSink != null) {
+      backlogSizeSink.accept(backlog);
+    }
+    // 2. adjust timing
+    adjustTimingFactors(processorTasks);
+  }
+
   private void printQueueStats() {
-    long oldestTaskTimestamp =
-        processorTasks.stream()
-            .filter(x -> x.getTaskQueue().size() > 0)
-            .mapToLong(QueueProcessor::getHeadTaskTimestamp)
-            .min()
-            .orElse(Long.MAX_VALUE);
-    //noinspection UnstableApiUsage
-    if ((oldestTaskTimestamp < timeProvider.get() - REPORT_QUEUE_STATS_DELAY_SECS * 1000)
-        && (reportRateLimiter.tryAcquire())) {
+    // 1. grab current queue sizes (tasks count)
+    int backlog = processorTasks.stream().mapToInt(x -> x.getTaskQueue().size()).sum();
+    queueSize.set(backlog);
+
+    // 2. grab queue sizes (points/etc count)
+    long actualWeight = 0L;
+    for (QueueProcessor<T> task : processorTasks) {
+      TaskQueue<T> taskQueue = task.getTaskQueue();
+      if ((taskQueue != null) && (taskQueue.weight() != null)) {
+        actualWeight += taskQueue.weight();
+      }
+    }
+    long previousWeight = currentWeight.getAndSet(actualWeight);
+
+    // 4. print stats when there's backlog
+    if ((previousWeight != 0) || (actualWeight != 0)) {
       logger.info(
           "["
               + handlerKey.getHandle()
@@ -198,35 +177,59 @@ private void printQueueStats() {
               + currentWeight
               + " "
               + handlerKey.getEntityType());
+      if (actualWeight == 0) {
+        logger.info(
+            "["
+                + handlerKey.getHandle()
+                + "] "
+                + handlerKey.getEntityType()
+                + " backlog has been cleared!");
+      }
     }
   }
 
   @Override
   public void start() {
     if (isRunning.compareAndSet(false, true)) {
-      timer.scheduleAtFixedRate(this, 1000, 1000);
+      if ((executor == null) || (executor.isShutdown())) { // need it for unittests
+        executor = Executors.newScheduledThreadPool(2, new NamedThreadFactory("QueueController"));
+      }
+      executor.scheduleAtFixedRate(this::run, 1, 1, TimeUnit.SECONDS);
+      executor.scheduleAtFixedRate(this::printQueueStats, 10, 10, TimeUnit.SECONDS);
       processorTasks.forEach(QueueProcessor::start);
     }
   }
 
   @Override
   public void stop() {
     if (isRunning.compareAndSet(true, false)) {
-      timer.cancel();
+      executor.shutdown();
       processorTasks.forEach(QueueProcessor::stop);
     }
   }
 
   public void truncateBuffers() {
     processorTasks.forEach(
         tQueueProcessor -> {
-          System.out.print("-- size: " + tQueueProcessor.getTaskQueue().size());
+          logger.info(
+              "["
+                  + handlerKey.getHandle()
+                  + "] "
+                  + handlerKey.getEntityType()
+                  + "-- size before truncate: "
+                  + tQueueProcessor.getTaskQueue().size());
           try {
             tQueueProcessor.getTaskQueue().clear();
           } catch (IOException e) {
             e.printStackTrace();
           }
-          System.out.println("--> size: " + tQueueProcessor.getTaskQueue().size());
+          logger.info(
+              "["
+                  + handlerKey.getHandle()
+                  + "] "
+                  + handlerKey.getEntityType()
+                  + "--> size after truncate: "
+                  + tQueueProcessor.getTaskQueue().size());
         });
   }
 }