Add "documentType" to "documents_processed" metric

5 files changed, 105 insertions, 28 deletions

diff --git a/docproc/src/main/java/com/yahoo/docproc/DocprocExecutor.java b/docproc/src/main/java/com/yahoo/docproc/DocprocExecutor.java
index f2d971ac5d0..0882de890a2 100644
--- a/docproc/src/main/java/com/yahoo/docproc/DocprocExecutor.java
+++ b/docproc/src/main/java/com/yahoo/docproc/DocprocExecutor.java
@@ -5,13 +5,20 @@ import com.yahoo.document.DocumentOperation;
 import com.yahoo.document.DocumentPut;
 import com.yahoo.document.json.JsonWriter;
 import com.yahoo.jdisc.Metric;
-import java.util.logging.Level;
 import com.yahoo.statistics.Counter;
 import com.yahoo.text.Utf8;
 
-import java.util.Collections;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.function.Function;
+import java.util.logging.Level;
 import java.util.logging.Logger;
 
+import static java.util.stream.Collectors.counting;
+import static java.util.stream.Collectors.groupingBy;
+
 /**
  * An executor executed incoming processings on its CallStack
  *
@@ -27,7 +34,7 @@ public class DocprocExecutor {
     private final String docCounterName;
     private final Counter docCounter;
     private final Metric metric;
-    private Metric.Context context;
+    private Function<String, Metric.Context> contexts;
     private final CallStack callStack;
 
     /**
@@ -44,7 +51,7 @@ public class DocprocExecutor {
         this.metric = callStack.getMetric();
         this.callStack = callStack;
         this.callStack.setName(name);
-        this.context = this.metric.createContext(Collections.singletonMap("chain", chainDimension));
+        this.contexts = cachedContexts(chainDimension);
     }
 
     /**
@@ -59,7 +66,7 @@ public class DocprocExecutor {
         this.docCounterName = oldExecutor.docCounterName;
         this.docCounter = oldExecutor.docCounter;
         this.metric = oldExecutor.metric;
-        this.context = oldExecutor.context;
+        this.contexts = oldExecutor.contexts;
         this.callStack = callStack;
     }
 
@@ -71,16 +78,14 @@ public class DocprocExecutor {
         return name;
     }
 
-    private void incrementNumDocsProcessed(int num) {
-        docCounter.increment(num);
-        metric.add(docCounterName, num, null);
-        metric.add(METRIC_NAME_DOCUMENTS_PROCESSED, num, this.context);
-    }
-
     private void incrementNumDocsProcessed(Processing processing) {
-        int increment = processing.getNumDocsToBeProcessed();
-        if (increment != 0) {
-            incrementNumDocsProcessed(increment);
+        List<DocumentOperation> operations = processing.getOnceOperationsToBeProcessed();
+        if ( ! operations.isEmpty()) {
+            docCounter.increment(operations.size());
+            metric.add(docCounterName, operations.size(), null);
+            operations.stream()
+                      .collect(groupingBy(operation -> operation.getId().getDocType(), counting()))
+                      .forEach((type, count) -> metric.add(METRIC_NAME_DOCUMENTS_PROCESSED, count, contexts.apply(type)));
         }
     }
 
@@ -176,4 +181,15 @@ public class DocprocExecutor {
         }
         return progress;
     }
+
+    private Function<String, Metric.Context> cachedContexts(String chainDimension) {
+        Map<String, Metric.Context> contextCache = new ConcurrentHashMap<>();
+        return documentType -> contextCache.computeIfAbsent(documentType, type -> {
+            Map<String, String> dimensions = new HashMap<>(2);
+            dimensions.put("chain", chainDimension);
+            dimensions.put("documentType", type);
+            return metric.createContext(dimensions);
+        });
+    }
+
 }
diff --git a/docproc/src/main/java/com/yahoo/docproc/Processing.java b/docproc/src/main/java/com/yahoo/docproc/Processing.java
index a029d4f4da3..63c0e143e30 100644
--- a/docproc/src/main/java/com/yahoo/docproc/Processing.java
+++ b/docproc/src/main/java/com/yahoo/docproc/Processing.java
@@ -5,6 +5,7 @@ import com.yahoo.component.provider.ComponentRegistry;
 import com.yahoo.document.DocumentOperation;
 
 import java.util.ArrayList;
+import java.util.Collections;
 import java.util.HashMap;
 import java.util.Iterator;
 import java.util.List;
@@ -44,7 +45,7 @@ public class Processing {
     /** The registry of docproc services. */
     private ComponentRegistry<DocprocService> docprocServiceRegistry = null;
 
-    private boolean getNumDocsCalled = false;
+    private boolean operationsGotten = false;
 
     /**
      * Create a Processing with no documents. Useful with DocprocService.process(Processing).
@@ -265,11 +266,11 @@ public class Processing {
         }
     }
 
-    int getNumDocsToBeProcessed() {
-        if (getNumDocsCalled) {
-            return 0;
-        }
-        getNumDocsCalled = true;
-        return getDocumentOperations().size();
+    List<DocumentOperation> getOnceOperationsToBeProcessed() {
+        if (operationsGotten)
+            return Collections.emptyList();
+
+        operationsGotten = true;
+        return getDocumentOperations();
     }
 }
diff --git a/docproc/src/main/java/com/yahoo/docproc/jdisc/metric/MockMetric.java b/docproc/src/main/java/com/yahoo/docproc/jdisc/metric/MockMetric.java
new file mode 100644
index 00000000000..a3cdd7d3394
--- /dev/null
+++ b/docproc/src/main/java/com/yahoo/docproc/jdisc/metric/MockMetric.java
@@ -0,0 +1,53 @@
+// Copyright Verizon Media. Licensed under the terms of the Apache 2.0 license. See LICENSE in the project root.
+package com.yahoo.docproc.jdisc.metric;
+
+import com.yahoo.jdisc.Metric;
+
+import java.util.Map;
+import java.util.concurrent.ConcurrentHashMap;
+
+/**
+ * Metric implementation for tests.
+ *
+ * @author jonmv
+ */
+public class MockMetric implements Metric {
+
+    private final Map<String, Map<Map<String, ?>, Double>> metrics = new ConcurrentHashMap<>();
+
+    @Override
+    public void set(String key, Number val, Context ctx) {
+        metrics.computeIfAbsent(key, k -> new ConcurrentHashMap<>())
+               .put(MapContext.emptyIfNull(ctx).dimensions, val.doubleValue());
+    }
+
+    @Override
+    public void add(String key, Number val, Context ctx) {
+        metrics.computeIfAbsent(key, k -> new ConcurrentHashMap<>())
+               .merge(MapContext.emptyIfNull(ctx).dimensions, val.doubleValue(), Double::sum);
+    }
+
+    @Override
+    public Context createContext(Map<String, ?> properties) {
+        return properties == null ? MapContext.empty : new MapContext(properties);
+    }
+
+    public Map<String, Map<Map<String, ?>, Double>> metrics() { return metrics; }
+
+    private static class MapContext implements Context {
+
+        private static final MapContext empty = new MapContext(Map.of());
+
+        private final Map<String, ?> dimensions;
+
+        private MapContext(Map<String, ?> dimensions) {
+            this.dimensions = dimensions;
+        }
+
+        private static MapContext emptyIfNull(Context context) {
+            return context == null ? empty : (MapContext) context;
+        }
+
+    }
+
+}
diff --git a/docproc/src/test/java/com/yahoo/docproc/ProcessingTestCase.java b/docproc/src/test/java/com/yahoo/docproc/ProcessingTestCase.java
index e72b9d991b6..b65327a2077 100644
--- a/docproc/src/test/java/com/yahoo/docproc/ProcessingTestCase.java
+++ b/docproc/src/test/java/com/yahoo/docproc/ProcessingTestCase.java
@@ -13,7 +13,7 @@ import static org.junit.Assert.assertNotNull;
 import static org.junit.Assert.assertTrue;
 
 /**
- * @author <a href="mailto:einarmr@yahoo-inc.com">Einar M R Rosenvinge</a>
+ * @author Einar M R Rosenvinge
  * @since 5.1.10
  */
 public class ProcessingTestCase {
diff --git a/docproc/src/test/java/com/yahoo/docproc/SimpleDocumentProcessorTestCase.java b/docproc/src/test/java/com/yahoo/docproc/SimpleDocumentProcessorTestCase.java
index 741d10bdc43..c5fb4b75cf3 100644
--- a/docproc/src/test/java/com/yahoo/docproc/SimpleDocumentProcessorTestCase.java
+++ b/docproc/src/test/java/com/yahoo/docproc/SimpleDocumentProcessorTestCase.java
@@ -2,6 +2,7 @@
 package com.yahoo.docproc;
 
 import com.yahoo.container.StatisticsConfig;
+import com.yahoo.docproc.jdisc.metric.MockMetric;
 import com.yahoo.docproc.jdisc.metric.NullMetric;
 import com.yahoo.document.DataType;
 import com.yahoo.document.DocumentId;
@@ -12,20 +13,23 @@ import com.yahoo.document.DocumentType;
 import com.yahoo.document.DocumentUpdate;
 import com.yahoo.document.datatypes.StringFieldValue;
 import com.yahoo.document.idstring.IdIdString;
+import com.yahoo.jdisc.Metric;
 import com.yahoo.statistics.StatisticsImpl;
 import org.junit.Test;
 
+import java.util.Map;
+
 import static org.junit.Assert.assertEquals;
 import static org.junit.Assert.assertNull;
 import static org.junit.Assert.assertTrue;
 
 /**
- * @author <a href="mailto:einarmr@yahoo-inc.com">Einar M R Rosenvinge</a>
+ * @author Einar M R Rosenvinge
  */
 public class SimpleDocumentProcessorTestCase {
 
-    private static DocprocService setupDocprocService(SimpleDocumentProcessor processor) {
-        CallStack stack = new CallStack("default", new StatisticsImpl(new StatisticsConfig(new StatisticsConfig.Builder())), new NullMetric());
+    private static DocprocService setupDocprocService(SimpleDocumentProcessor processor, Metric metric) {
+        CallStack stack = new CallStack("default", new StatisticsImpl(new StatisticsConfig(new StatisticsConfig.Builder())), metric);
         stack.addLast(processor);
         DocprocService service = new DocprocService("default");
         service.setCallStack(stack);
@@ -57,7 +61,8 @@ public class SimpleDocumentProcessorTestCase {
                                      new DocumentUpdate(type, "id:this:foobar::is:an:update"),
                                      new DocumentRemove(new DocumentId("id:this:foobar::is:a:remove")));
 
-        DocprocService service = setupDocprocService(new VerySimpleDocumentProcessor());
+        MockMetric metric = new MockMetric();
+        DocprocService service = setupDocprocService(new VerySimpleDocumentProcessor(), metric);
         service.getExecutor().process(p);
 
         assertEquals(3, p.getDocumentOperations().size());
@@ -66,6 +71,8 @@ public class SimpleDocumentProcessorTestCase {
         assertTrue(p.getDocumentOperations().get(1) instanceof DocumentUpdate);
         assertTrue(p.getDocumentOperations().get(2) instanceof DocumentRemove);
         assertEquals("id:foobar:foobar::12345", p.getDocumentOperations().get(2).getId().toString());
+        assertEquals(Map.of(Map.of("chain", "default", "documentType", "foobar"), 3.0),
+                     metric.metrics().get("documents_processed"));
     }
 
     @Test
@@ -73,7 +80,7 @@ public class SimpleDocumentProcessorTestCase {
         DocumentType type = createTestType();
 
         Processing p = getProcessing(new DocumentPut(type, "id:this:foobar::is:a:document"));
-        DocprocService service = setupDocprocService(new VerySimpleDocumentProcessor());
+        DocprocService service = setupDocprocService(new VerySimpleDocumentProcessor(), new MockMetric());
         service.getExecutor().process(p);
 
         assertEquals(1, p.getDocumentOperations().size());
@@ -89,7 +96,7 @@ public class SimpleDocumentProcessorTestCase {
                                      new DocumentRemove(new DocumentId("id:this:foobar::is:a:remove")),
                                      new DocumentPut(type, "id:this:foobar::is:a:document2"));
 
-        DocprocService service = setupDocprocService(new SimpleDocumentProcessorThrowingOnRemovesAndUpdates());
+        DocprocService service = setupDocprocService(new SimpleDocumentProcessorThrowingOnRemovesAndUpdates(), new MockMetric());
         try {
             service.getExecutor().process(p);
         } catch (RuntimeException re) {