Wire in MetricReporter into the backend servlet to publish metrics

-------------
Created by MOE: https://github.com/google/moe
MOE_MIGRATED_REVID=130468842

java/google/registry/monitoring/metrics/MetricExporter.java

@@ -18,6 +18,10 @@ import com.google.common.base.Optional;
 import com.google.common.collect.ImmutableList;
 import com.google.common.util.concurrent.AbstractExecutionThreadService;
 import java.util.concurrent.BlockingQueue;
+import java.util.concurrent.Executor;
+import java.util.concurrent.Executors;
+import java.util.concurrent.ThreadFactory;
+import java.util.logging.Logger;
 
 /**
  * Background service to asynchronously push bundles of {@link MetricPoint} instances to a {@link
@@ -25,17 +29,24 @@ import java.util.concurrent.BlockingQueue;
  */
 class MetricExporter extends AbstractExecutionThreadService {
 
+  private static final Logger logger = Logger.getLogger(MetricReporter.class.getName());
+
   private final BlockingQueue<Optional<ImmutableList<MetricPoint<?>>>> writeQueue;
   private final MetricWriter writer;
+  private final ThreadFactory threadFactory;
 
   MetricExporter(
-      BlockingQueue<Optional<ImmutableList<MetricPoint<?>>>> writeQueue, MetricWriter writer) {
+      BlockingQueue<Optional<ImmutableList<MetricPoint<?>>>> writeQueue,
+      MetricWriter writer,
+      ThreadFactory threadFactory) {
     this.writeQueue = writeQueue;
     this.writer = writer;
+    this.threadFactory = threadFactory;
   }
 
   @Override
   protected void run() throws Exception {
+    logger.info("Started up MetricExporter");
     while (isRunning()) {
       Optional<ImmutableList<MetricPoint<?>>> batch = writeQueue.take();
       if (batch.isPresent()) {
@@ -44,9 +55,15 @@ class MetricExporter extends AbstractExecutionThreadService {
         }
         writer.flush();
       } else {
+        logger.info("Received a poison pill, stopping now");
         // An absent optional indicates that the Reporter wants this service to shut down.
         return;
       }
     }
   }
+
+  @Override
+  protected Executor executor() {
+    return Executors.newSingleThreadExecutor(threadFactory);
+  }
 }

java/google/registry/monitoring/metrics/MetricRegistryImpl.java

@@ -81,7 +81,6 @@ public final class MetricRegistryImpl implements MetricRegistry {
       String description,
       String valueDisplayName,
       ImmutableSet<LabelDescriptor> labels) {
-
     Counter metric = new Counter(name, description, valueDisplayName, labels);
     registerMetric(name, metric);
     logger.info("Registered new counter: " + name);

java/google/registry/monitoring/metrics/MetricReporter.java

@@ -15,6 +15,7 @@
 package google.registry.monitoring.metrics;
 
 import static com.google.common.base.Preconditions.checkArgument;
+import static com.google.common.util.concurrent.MoreExecutors.directExecutor;
 
 import com.google.common.annotations.VisibleForTesting;
 import com.google.common.base.Optional;
@@ -22,8 +23,15 @@ import com.google.common.collect.ImmutableList;
 import com.google.common.util.concurrent.AbstractScheduledService;
 import java.util.concurrent.ArrayBlockingQueue;
 import java.util.concurrent.BlockingQueue;
+import java.util.concurrent.Executors;
+import java.util.concurrent.ScheduledExecutorService;
+import java.util.concurrent.ThreadFactory;
 import java.util.concurrent.TimeUnit;
+import java.util.concurrent.TimeoutException;
+import java.util.logging.Level;
 import java.util.logging.Logger;
+import javax.inject.Inject;
+import javax.inject.Named;
 
 /**
  * Engine to write metrics to a {@link MetricWriter} on a regular periodic basis.
@@ -38,18 +46,26 @@ public class MetricReporter extends AbstractScheduledService {
   private final long writeInterval;
   private final MetricRegistry metricRegistry;
   private final BlockingQueue<Optional<ImmutableList<MetricPoint<?>>>> writeQueue;
-  private final MetricExporter metricExporter;
+  private MetricExporter metricExporter;
+  private final MetricWriter metricWriter;
+  private final ThreadFactory threadFactory;
 
   /**
    * Returns a new MetricReporter.
    *
    * @param metricWriter {@link MetricWriter} implementation to write metrics to.
    * @param writeInterval time period between metric writes, in seconds.
+   * @param threadFactory factory to use when creating background threads.
    */
-  public MetricReporter(MetricWriter metricWriter, long writeInterval) {
+  @Inject
+  public MetricReporter(
+      MetricWriter metricWriter,
+      @Named("metricsWriteInterval") long writeInterval,
+      @Named("metricsBackgroundThreadFactory") ThreadFactory threadFactory) {
     this(
         metricWriter,
         writeInterval,
+        threadFactory,
         MetricRegistryImpl.getDefault(),
         new ArrayBlockingQueue<Optional<ImmutableList<MetricPoint<?>>>>(1000));
   }
@@ -58,19 +74,27 @@ public class MetricReporter extends AbstractScheduledService {
   MetricReporter(
       MetricWriter metricWriter,
       long writeInterval,
+      ThreadFactory threadFactory,
       MetricRegistry metricRegistry,
       BlockingQueue<Optional<ImmutableList<MetricPoint<?>>>> writeQueue) {
     checkArgument(writeInterval > 0, "writeInterval must be greater than zero");
 
+    this.metricWriter = metricWriter;
     this.writeInterval = writeInterval;
+    this.threadFactory = threadFactory;
     this.metricRegistry = metricRegistry;
     this.writeQueue = writeQueue;
-    this.metricExporter = new MetricExporter(writeQueue, metricWriter);
+    this.metricExporter = new MetricExporter(writeQueue, metricWriter, threadFactory);
   }
 
   @Override
   protected void runOneIteration() {
     logger.info("Running background metric push");
+
+    if (metricExporter.state() == State.FAILED) {
+      startMetricExporter();
+    }
+
     ImmutableList.Builder<MetricPoint<?>> points = new ImmutableList.Builder<>();
 
     /*
@@ -80,12 +104,12 @@ public class MetricReporter extends AbstractScheduledService {
     for (Metric<?> metric : metricRegistry.getRegisteredMetrics()) {
       points.addAll(metric.getTimestampedValues());
       logger.fine(String.format("Enqueued metric %s", metric));
-      MetricMetrics.pushedPoints.incrementBy(
-          1, metric.getMetricSchema().kind().name(), metric.getValueClass().toString());
+      MetricMetrics.pushedPoints.incrementBy(1,
+          metric.getMetricSchema().kind().name(), metric.getValueClass().toString());
     }
 
     if (!writeQueue.offer(Optional.of(points.build()))) {
-      logger.warning("writeQueue full, dropped a reporting interval of points");
+      logger.severe("writeQueue full, dropped a reporting interval of points");
     }
 
     MetricMetrics.pushIntervals.incrementBy(1);
@@ -98,12 +122,17 @@ public class MetricReporter extends AbstractScheduledService {
     runOneIteration();
 
     writeQueue.offer(Optional.<ImmutableList<MetricPoint<?>>>absent());
-    metricExporter.stopAsync().awaitTerminated();
+    try {
+      metricExporter.stopAsync().awaitTerminated(10, TimeUnit.SECONDS);
+      logger.info("Shut down MetricExporter");
+    } catch (TimeoutException timeoutException) {
+      logger.severe("Failed to shut down MetricExporter: " + timeoutException);
+    }
   }
 
   @Override
   protected void startUp() {
-    metricExporter.startAsync().awaitRunning();
+    startMetricExporter();
   }
 
   @Override
@@ -111,4 +140,38 @@ public class MetricReporter extends AbstractScheduledService {
     // Start writing after waiting for one writeInterval.
     return Scheduler.newFixedDelaySchedule(writeInterval, writeInterval, TimeUnit.SECONDS);
   }
+
+  @Override
+  protected ScheduledExecutorService executor() {
+    final ScheduledExecutorService executor =
+        Executors.newSingleThreadScheduledExecutor(threadFactory);
+    addListener(
+        new Listener() {
+          @Override
+          public void terminated(State from) {
+            executor.shutdown();
+          }
+
+          @Override
+          public void failed(State from, Throwable failure) {
+            executor.shutdown();
+          }
+        },
+        directExecutor());
+    return executor;
+  }
+
+  @VisibleForTesting
+  void startMetricExporter() {
+    // Services in the FAILED state must be reconstructed, they can't be started
+    if (metricExporter.state() == State.FAILED) {
+      logger.log(
+          Level.SEVERE,
+          "MetricExporter died unexpectedly, restarting",
+          metricExporter.failureCause());
+      this.metricExporter = new MetricExporter(writeQueue, metricWriter, threadFactory);
+    }
+
+    this.metricExporter.startAsync();
+  }
 }