Backport #1967 - Add datasource and taskId to metrics emitted by peons

indexing-service/src/main/java/io/druid/indexing/overlord/ForkingTaskRunner.java

@@ -52,7 +52,9 @@
 import io.druid.indexing.common.tasklogs.LogUtils;
 import io.druid.indexing.overlord.config.ForkingTaskRunnerConfig;
 import io.druid.indexing.worker.config.WorkerConfig;
+import io.druid.query.DruidMetrics;
 import io.druid.server.DruidNode;
+import io.druid.server.metrics.MonitorsConfig;
 import io.druid.tasklogs.TaskLogPusher;
 import io.druid.tasklogs.TaskLogStreamer;
 import org.apache.commons.io.FileUtils;
@@ -296,6 +298,24 @@ public TaskStatus call()
                                 }
                               }
 
+                              // Add dataSource and taskId for metrics
+                              command.add(
+                                  String.format(
+                                      "-D%s%s=%s",
+                                      MonitorsConfig.METRIC_DIMENSION_PREFIX,
+                                      DruidMetrics.DATASOURCE,
+                                      task.getDataSource()
+                                  )
+                              );
+                              command.add(
+                                  String.format(
+                                      "-D%s%s=%s",
+                                      MonitorsConfig.METRIC_DIMENSION_PREFIX,
+                                      DruidMetrics.TASK_ID,
+                                      task.getId()
+                                  )
+                              );
+
                               command.add(String.format("-Ddruid.host=%s", childHost));
                               command.add(String.format("-Ddruid.port=%d", childPort));
 

processing/src/main/java/io/druid/query/DruidMetrics.java

@@ -36,6 +36,7 @@ public class DruidMetrics
   public final static String TYPE = "type";
   public final static String INTERVAL = "interval";
   public final static String ID = "id";
+  public final static String TASK_ID = "taskId";
   public final static String STATUS = "status";
 
   // task metrics

server/src/main/java/io/druid/server/metrics/MetricsModule.java

@@ -28,15 +28,21 @@
 import com.google.inject.name.Names;
 import com.metamx.common.logger.Logger;
 import com.metamx.emitter.service.ServiceEmitter;
+import com.metamx.metrics.JvmCpuMonitor;
+import com.metamx.metrics.JvmMonitor;
 import com.metamx.metrics.Monitor;
 import com.metamx.metrics.MonitorScheduler;
+import com.metamx.metrics.SysMonitor;
 import io.druid.concurrent.Execs;
 import io.druid.guice.DruidBinders;
 import io.druid.guice.JsonConfigProvider;
 import io.druid.guice.LazySingleton;
 import io.druid.guice.ManageLifecycle;
 
+import java.util.HashMap;
 import java.util.List;
+import java.util.Map;
+import java.util.Properties;
 import java.util.Set;
 
 /**
@@ -95,4 +101,40 @@ public MonitorScheduler getMonitorScheduler(
         monitors
     );
   }
+
+  @Provides
+  @ManageLifecycle
+  public JvmMonitor getJvmMonitor(Properties props)
+  {
+    return new JvmMonitor(getDimensions(props));
+  }
+
+  @Provides
+  @ManageLifecycle
+  public JvmCpuMonitor getJvmCpuMonitor(Properties props)
+  {
+    return new JvmCpuMonitor(getDimensions(props));
+  }
+
+  @Provides
+  @ManageLifecycle
+  public SysMonitor getSysMonitor(Properties props)
+  {
+    return new SysMonitor(getDimensions(props));
+  }
+
+  private Map<String, String[]> getDimensions(Properties props)
+  {
+    Map<String, String[]> dimensions = new HashMap<>();
+    for (String property : props.stringPropertyNames()) {
+      if (property.startsWith(MonitorsConfig.METRIC_DIMENSION_PREFIX)) {
+        dimensions.put(
+            property.substring(MonitorsConfig.METRIC_DIMENSION_PREFIX.length()),
+            new String[]{props.getProperty(property)}
+        );
+      }
+    }
+    return dimensions;
+  }
+
 }

server/src/main/java/io/druid/server/metrics/MonitorsConfig.java

@@ -28,6 +28,8 @@
  */
 public class MonitorsConfig
 {
+  public final static String METRIC_DIMENSION_PREFIX = "druid.metrics.emitter.dimension.";
+
   @JsonProperty("monitors")
   @NotNull
   private List<Class<? extends Monitor>> monitors = Lists.newArrayList();