Add labels to Pipelined tasks metrics (#1057)

Add the taskName labels to the Pipelined Task metrics.
Signed-off-by: Adrian Sutton <adrian.sutton@consensys.net>

ethereum/eth/src/main/java/tech/pegasys/pantheon/ethereum/eth/manager/task/AbstractPipelinedTask.java

@@ -53,15 +53,21 @@ public abstract class AbstractPipelinedTask<I, O> extends AbstractEthTask<List<O
     outboundQueue = new LinkedBlockingQueue<>(outboundBacklogSize);
     results = new ArrayList<>();
     this.inboundQueueCounter =
-        metricsSystem.createCounter(
+        metricsSystem
-            MetricCategory.SYNCHRONIZER,
+            .createLabelledCounter(
-            "inboundQueueCounter",
+                MetricCategory.SYNCHRONIZER,
-            "count of queue items that started processing");
+                "inboundQueueCounter",
+                "count of queue items that started processing",
+                "taskName")
+            .labels(getClass().getSimpleName());
     this.outboundQueueCounter =
-        metricsSystem.createCounter(
+        metricsSystem
-            MetricCategory.SYNCHRONIZER,
+            .createLabelledCounter(
-            "outboundQueueCounter",
+                MetricCategory.SYNCHRONIZER,
-            "count of queue items that finished processing");
+                "outboundQueueCounter",
+                "count of queue items that finished processing",
+                "taskName")
+            .labels(getClass().getSimpleName());
   }
 
   @Override

ethereum/eth/src/test/java/tech/pegasys/pantheon/ethereum/eth/sync/fullsync/IncrementerTest.java

@@ -34,10 +34,11 @@ import tech.pegasys.pantheon.metrics.noop.NoOpMetricsSystem;
 import tech.pegasys.pantheon.metrics.prometheus.MetricsConfiguration;
 import tech.pegasys.pantheon.metrics.prometheus.PrometheusMetricsSystem;
 
+import java.util.Arrays;
+import java.util.Collections;
 import java.util.List;
 import java.util.stream.Collectors;
 
-import org.assertj.core.util.Lists;
 import org.junit.Test;
 
 public class IncrementerTest {
@@ -95,12 +96,43 @@ public class IncrementerTest {
 
     final List<Observation> metrics =
         metricsSystem.getMetrics(MetricCategory.SYNCHRONIZER).collect(Collectors.toList());
-    final Observation inboundObservation =
+
-        new Observation(MetricCategory.SYNCHRONIZER, "inboundQueueCounter", 6.0, Lists.emptyList());
+    // the first iteration gets the genesis block, which results in no data
-    final Observation outboundObservation =
+    // being passed downstream.  So observed value is 2.
+    final Observation headerInboundObservation =
+        new Observation(
+            MetricCategory.SYNCHRONIZER,
+            "inboundQueueCounter",
+            2.0,
+            Collections.singletonList("ParallelDownloadHeadersTask"));
+    final Observation headerOutboundObservation =
         new Observation(
-            MetricCategory.SYNCHRONIZER, "outboundQueueCounter", 5.0, Lists.emptyList());
+            MetricCategory.SYNCHRONIZER,
-    assertThat(metrics).contains(inboundObservation, outboundObservation);
+            "outboundQueueCounter",
+            1.0,
+            Collections.singletonList("ParallelDownloadHeadersTask"));
+    assertThat(metrics).contains(headerInboundObservation, headerOutboundObservation);
+
+    for (final String label :
+        Arrays.asList(
+            "ParallelValidateHeadersTask",
+            "ParallelDownloadBodiesTask",
+            "ParallelExtractTxSignaturesTask",
+            "ParallelValidateAndImportBodiesTask")) {
+      final Observation inboundObservation =
+          new Observation(
+              MetricCategory.SYNCHRONIZER,
+              "inboundQueueCounter",
+              1.0,
+              Collections.singletonList(label));
+      final Observation outboundObservation =
+          new Observation(
+              MetricCategory.SYNCHRONIZER,
+              "outboundQueueCounter",
+              1.0,
+              Collections.singletonList(label));
+      assertThat(metrics).contains(inboundObservation, outboundObservation);
+    }
   }
 
   private FullSyncDownloader<?> downloader(final SynchronizerConfiguration syncConfig) {