Add datasource and taskId to metrics emitted by peons

This PR adds the datasource and taskId to the jvm and sys metrics
emitted by the peons.

fix spelling

review comment

review comment

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

@@ -52,7 +52,9 @@ import io.druid.indexing.common.task.Task;
 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;
@@ -297,6 +299,24 @@ public class ForkingTaskRunner implements TaskRunner, TaskLogStreamer
                                 }
                               }
 
+                              // 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.Provides;
 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 class MetricsModule implements Module
         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 @@ import java.util.List;
  */
 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();