Add "documentType" to "documents_processed" metric

3 files changed, 91 insertions, 21 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;
+        }
+
+    }
+
+}