add gauge for queue size

Author: tburch

src/main/java/com/lowtuna/jsonblob/JsonBlobApplication.java

@@ -75,7 +75,7 @@ public void run(JsonBlobConfiguration configuration, Environment environment) th
     ScheduledExecutorService scheduledExecutorService = configuration.getBlobManagerConfig().getScheduledExecutorService().instance(environment);
     ScheduledExecutorService cleanupScheduledExecutorService = configuration.getBlobManagerConfig().getCleanupScheduledExecutorService().instance(environment);
 
-    FileSystemJsonBlobManager fileSystemBlobManager = new FileSystemJsonBlobManager(configuration.getBlobManagerConfig().getFileSystemBlogDataDirectory(), scheduledExecutorService, cleanupScheduledExecutorService, environment.getObjectMapper(), configuration.getBlobManagerConfig().getBlobAccessTtl(), configuration.getBlobManagerConfig().isDeleteEnabled());
+    FileSystemJsonBlobManager fileSystemBlobManager = new FileSystemJsonBlobManager(configuration.getBlobManagerConfig().getFileSystemBlogDataDirectory(), scheduledExecutorService, cleanupScheduledExecutorService, environment.getObjectMapper(), configuration.getBlobManagerConfig().getBlobAccessTtl(), configuration.getBlobManagerConfig().isDeleteEnabled(), environment.metrics());
     environment.lifecycle().manage(fileSystemBlobManager);
 
     environment.healthChecks().register("freeSpace", new BlobDirectoryFreeSpaceHealthcheck(configuration.getBlobManagerConfig().getFileSystemBlogDataDirectory(), 5242880));

src/main/java/com/lowtuna/jsonblob/core/BlobCleanupConsumer.java

@@ -19,6 +19,8 @@
 @Slf4j
 @RequiredArgsConstructor
 public class BlobCleanupConsumer implements Runnable {
+  private static final Duration QUEUE_TIMEOUT = Duration.seconds(15);
+
   private final BlockingQueue<File> filesToProcess;
   private final Duration blobAccessTtl;
   private final FileSystemJsonBlobManager fileSystemJsonBlobManager;
@@ -27,7 +29,7 @@ public class BlobCleanupConsumer implements Runnable {
   @Override
   public void run() {
     try {
-      File file = filesToProcess.take();
+      File file = filesToProcess.poll(QUEUE_TIMEOUT.getQuantity(), QUEUE_TIMEOUT.getUnit());
       log.debug("Processing {}", file.getAbsolutePath());
       String blobId = file.getName().split("\\.", 2)[0];
       File metadataFile = fileSystemJsonBlobManager.getMetaDataFile(file.getParentFile());

src/main/java/com/lowtuna/jsonblob/core/BlobCleanupProducer.java

@@ -1,8 +1,9 @@
 package com.lowtuna.jsonblob.core;
 
+import com.codahale.metrics.Gauge;
+import com.codahale.metrics.MetricRegistry;
 import com.google.common.base.Stopwatch;
 import io.dropwizard.util.Duration;
-import lombok.AllArgsConstructor;
 import lombok.extern.slf4j.Slf4j;
 import org.apache.commons.io.DirectoryWalker;
 
@@ -17,13 +18,20 @@
 /**
  * Created by tburch on 8/18/17.
  */
-@AllArgsConstructor
 @Slf4j
 public class BlobCleanupProducer extends DirectoryWalker<Void> implements Runnable {
   private final Path dataDirectoryPath;
   private final Duration blobAccessTtl;
   private final BlockingQueue<File> filesToProcess;
 
+  public BlobCleanupProducer(Path dataDirectoryPath, Duration blobAccessTtl, BlockingQueue<File> filesToProcess, MetricRegistry metricRegistry) {
+    this.dataDirectoryPath = dataDirectoryPath;
+    this.blobAccessTtl = blobAccessTtl;
+    this.filesToProcess = filesToProcess;
+    metricRegistry.register(MetricRegistry.name(getClass(), "filesToProcessCount"), (Gauge<Integer>) () -> filesToProcess.size());
+  }
+
+
   @Override
   protected boolean handleDirectory(File directory, int depth, Collection<Void> results) throws IOException {
     if (directory.listFiles() != null && directory.listFiles().length == 0) {
@@ -44,7 +52,7 @@ protected boolean handleDirectory(File directory, int depth, Collection<Void> re
       LocalDate localDate = LocalDate.of(Integer.parseInt(dateParts[1]), Integer.parseInt(dateParts[2]), Integer.parseInt(dateParts[3]));
       process = localDate.isBefore(LocalDate.now().minusDays(blobAccessTtl.toDays()));
       if (process) {
-        log.info("Processing {} with {} blobs for un-accessed blobs", directory.getAbsolutePath(), directory.listFiles().length - 1);
+        log.info("Processing {} blobs for un-accessed blobs", directory.getAbsolutePath());
         for (File file: directory.listFiles()) {
           try {
             filesToProcess.put(file);

src/main/java/com/lowtuna/jsonblob/core/FileSystemJsonBlobManager.java

@@ -1,5 +1,6 @@
 package com.lowtuna.jsonblob.core;
 
+import com.codahale.metrics.MetricRegistry;
 import com.fasterxml.jackson.databind.ObjectMapper;
 import com.fasterxml.uuid.Generators;
 import com.google.common.base.Optional;
@@ -64,14 +65,16 @@ public class FileSystemJsonBlobManager implements JsonBlobManager, Runnable, Man
   private final Duration blobAccessTtl;
   @Getter
   private final boolean deleteEnabled;
+  private final MetricRegistry metricRegistry;
 
-  public FileSystemJsonBlobManager(File blobDataDirectory, ScheduledExecutorService scheduledExecutorService, ScheduledExecutorService cleanupScheduledExecutorService, ObjectMapper objectMapper, Duration blobAccessTtl, boolean deleteEnabled) {
+  public FileSystemJsonBlobManager(File blobDataDirectory, ScheduledExecutorService scheduledExecutorService, ScheduledExecutorService cleanupScheduledExecutorService, ObjectMapper objectMapper, Duration blobAccessTtl, boolean deleteEnabled, MetricRegistry metricRegistry) {
     this.blobDataDirectory = blobDataDirectory;
     this.scheduledExecutorService = scheduledExecutorService;
     this.cleanupScheduledExecutorService = cleanupScheduledExecutorService;
     this.objectMapper = objectMapper;
     this.blobAccessTtl = blobAccessTtl;
     this.deleteEnabled = deleteEnabled;
+    this.metricRegistry = metricRegistry;
 
     blobDataDirectory.mkdirs();
   }
@@ -316,7 +319,7 @@ public void start() throws Exception {
       cleanupScheduledExecutorService.scheduleAtFixedRate(new BlobCleanupConsumer(filesToProcess, blobAccessTtl, this, objectMapper), 0, 250, TimeUnit.MILLISECONDS);
     }
 
-    BlobCleanupProducer dataDirectoryCleaner = new BlobCleanupProducer(blobDataDirectory.toPath(), blobAccessTtl, filesToProcess);
+    BlobCleanupProducer dataDirectoryCleaner = new BlobCleanupProducer(blobDataDirectory.toPath(), blobAccessTtl, filesToProcess, metricRegistry);
     cleanupScheduledExecutorService.scheduleWithFixedDelay(dataDirectoryCleaner, 0, 1, TimeUnit.DAYS);
   }
 

src/test/java/com/lowtuna/jsonblob/core/TryBlobCleanupJob.java

@@ -1,5 +1,6 @@
 package com.lowtuna.jsonblob.core;
 
+import com.codahale.metrics.MetricRegistry;
 import com.fasterxml.jackson.databind.ObjectMapper;
 import io.dropwizard.util.Duration;
 import lombok.extern.java.Log;
@@ -36,7 +37,7 @@ public class TryBlobCleanupJob {
 
   @Before
   public void initBlobManage() {
-    this.blobManager = new FileSystemJsonBlobManager(TEMP, Executors.newSingleThreadScheduledExecutor(), Executors.newScheduledThreadPool(10), new ObjectMapper(), blobTtl, true);
+    this.blobManager = new FileSystemJsonBlobManager(TEMP, Executors.newSingleThreadScheduledExecutor(), Executors.newScheduledThreadPool(10), new ObjectMapper(), blobTtl, true, new MetricRegistry());
   }
 
   @Test