minor: Emit taskStatus on task/run/time from the statsd and prometheus emitters

extensions-contrib/prometheus-emitter/src/main/resources/defaultMetrics.json

@@ -142,7 +142,7 @@
   "ingest/realtime/segmentUpgrade/skipped" : { "dimensions" : ["dataSource", "reason"], "type" : "count", "help": "Number of upgrade requests a task received but did not announce." },
   "task/autoScaler/requiredCount" : { "dimensions" : ["dataSource"], "type" : "count", "help": "Count of required tasks based on the calculations of lagBased auto scaler." },
 
-  "task/run/time" : { "dimensions" : ["dataSource", "taskType"], "type" : "timer", "conversionFactor": 1000.0, "help": "Seconds taken to run a task."},
+  "task/run/time" : { "dimensions" : ["dataSource", "taskType", "taskStatus"], "type" : "timer", "conversionFactor": 1000.0, "help": "Seconds taken to run a task."},
   "task/pending/time" : { "dimensions" : ["dataSource", "taskType"], "type" : "timer", "conversionFactor": 1000.0, "help": "Seconds taken for a task to wait for running."},
   "task/action/run/time" : { "dimensions" : ["dataSource", "taskType"], "type" : "timer", "conversionFactor": 1000.0, "help": "Seconds taken to execute a task action."},
   "task/action/success/count" : { "dimensions" : ["dataSource"], "type" : "count", "help": "Number of task actions that were executed successfully during the emission period."},

extensions-contrib/prometheus-emitter/src/test/java/org/apache/druid/emitter/prometheus/MetricsTest.java

@@ -138,4 +138,17 @@ public void testMetricsConfigurationWithTimerHistogramBuckets()
     Assertions.assertArrayEquals(expectedHistogramBuckets, dimensionsAndCollector.getHistogramBuckets(), 0.0);
   }
 
+  @Test
+  public void testTaskRunTimeHasTaskStatusLabel()
+  {
+    PrometheusEmitterConfig config = new PrometheusEmitterConfig(null, "test_7", null, null, null, true, true, null, null, null, null);
+    Metrics metrics = new Metrics(config);
+    DimensionsAndCollector dimensionsAndCollector = metrics.getByName("task/run/time", "overlord");
+    Assertions.assertNotNull(dimensionsAndCollector);
+    Assertions.assertArrayEquals(
+        new String[]{"dataSource", "druid_service", "host_name", "taskStatus", "taskType"},
+        dimensionsAndCollector.getDimensions()
+    );
+  }
+
 }

extensions-contrib/prometheus-emitter/src/test/java/org/apache/druid/emitter/prometheus/PrometheusEmitterTest.java

@@ -209,15 +209,17 @@ public void testEmitterMetric()
     ServiceMetricEvent build = ServiceMetricEvent.builder()
             .setDimension("dataSource", "test")
             .setDimension("taskType", "index_parallel")
+            .setDimension("taskStatus", "SUCCESS")
             .setMetric("task/run/time", 500)
             .build(ImmutableMap.of("service", "overlord", "host", "druid.test.cn"));
     emitter.emit(build);
     double assertEpsilon = 0.0001;
+    final String[] labelNames = {"dataSource", "druid_service", "host_name", "taskStatus", "taskType", "le"};
     Assertions.assertEquals(0.0, CollectorRegistry.defaultRegistry.getSampleValue(
-            "namespace_task_run_time_bucket", new String[]{"dataSource", "druid_service", "host_name", "taskType", "le"}, new String[]{"test", "overlord", "druid.test.cn", "index_parallel", "0.1"}
+            "namespace_task_run_time_bucket", labelNames, new String[]{"test", "overlord", "druid.test.cn", "SUCCESS", "index_parallel", "0.1"}
     ), assertEpsilon);
     Assertions.assertEquals(1.0, CollectorRegistry.defaultRegistry.getSampleValue(
-            "namespace_task_run_time_bucket", new String[]{"dataSource", "druid_service", "host_name", "taskType", "le"}, new String[]{"test", "overlord", "druid.test.cn", "index_parallel", "0.5"}
+            "namespace_task_run_time_bucket", labelNames, new String[]{"test", "overlord", "druid.test.cn", "SUCCESS", "index_parallel", "0.5"}
     ), assertEpsilon);
   }
 

extensions-contrib/statsd-emitter/src/main/resources/defaultMetricDimensions.json

@@ -104,7 +104,7 @@
   "taskSlot/lazy/count" : { "dimensions" : ["category"], "type" : "gauge" },
   "taskSlot/blacklisted/count" : { "dimensions" : ["category"], "type" : "gauge" },
 
-  "task/run/time" : { "dimensions" : ["dataSource", "taskType"], "type" : "timer" },
+  "task/run/time" : { "dimensions" : ["dataSource", "taskType", "taskStatus"], "type" : "timer" },
   "segment/added/bytes" : { "dimensions" : ["dataSource", "taskType"], "type" : "count" },
   "segment/moved/bytes" : { "dimensions" : ["dataSource", "taskType"], "type" : "count" },
   "segment/nuked/bytes" : { "dimensions" : ["dataSource", "taskType"], "type" : "count" },

extensions-contrib/statsd-emitter/src/test/java/org/apache/druid/emitter/statsd/DimensionConverterTest.java

@@ -25,6 +25,8 @@
 import org.junit.jupiter.api.Assertions;
 import org.junit.jupiter.api.Test;
 
+import java.util.List;
+
 public class DimensionConverterTest
 {
   @Test
@@ -58,4 +60,42 @@ public void testConvert()
     expected.put("type", "groupBy");
     Assertions.assertEquals(expected.build(), actual.build(), "correct Dimensions");
   }
+
+  @Test
+  public void testConvertTaskRunTime()
+  {
+    DimensionConverter dimensionConverter = new DimensionConverter(new ObjectMapper(), null);
+    ServiceMetricEvent event = new ServiceMetricEvent.Builder()
+        .setDimension("dataSource", "data-source")
+        .setDimension("taskType", "index_kafka")
+        .setDimension("taskStatus", "FAILED")
+        .setDimension("taskId", "index_kafka_data-source_abc_1")
+        .setDimension("groupId", "index_kafka_data-source_abc")
+        .setDimension("description", "some very long error message")
+        .setMetric("task/run/time", 10)
+        .build("overlord", "overlordHost1");
+
+    ImmutableMap.Builder<String, String> actual = new ImmutableMap.Builder<>();
+    StatsDMetric statsDMetric = dimensionConverter.addFilteredUserDims(
+        event.getService(),
+        event.getMetric(),
+        event.getUserDims(),
+        actual
+    );
+    Assertions.assertEquals(StatsDMetric.Type.timer, statsDMetric.type, "correct StatsDMetric.Type");
+    final ImmutableMap<String, String> dims = actual.build();
+    // taskId, groupId and description stay filtered out; they are unbounded.
+    Assertions.assertEquals(
+        ImmutableMap.of("dataSource", "data-source", "taskStatus", "FAILED", "taskType", "index_kafka"),
+        dims,
+        "correct Dimensions"
+    );
+    // Dimensions are iterated in sorted order, and for non-dogstatsd output their values are
+    // appended to the dotted metric name in that order, so the emitted order is user-visible.
+    Assertions.assertEquals(
+        List.of("dataSource", "taskStatus", "taskType"),
+        List.copyOf(dims.keySet()),
+        "correct Dimension order"
+    );
+  }
 }