OshiSysMonitor: Add ability to skip emitting metrics

docs/operations/metrics.md

@@ -507,6 +507,8 @@ These metrics are only available if the `OshiSysMonitor` module is included.
 |`sys/tcpv4/out/rsts`|Total "out reset" packets sent to reset the connection||Generally 0|
 |`sys/tcpv4/retrans/segs`|Total segments re-transmitted||Varies|
 
+If you want to skip some of these metrics you could specify `druid.monitoring.oshisys.skipEmitting`.
+Possible values are `mem`, `swap`, `fs`, `disk`, `net`, `cpu`, `sys`, and `tcp`.
 
 ## S3 multi-part upload
 

processing/src/main/java/org/apache/druid/java/util/metrics/OshiSysMonitor.java

@@ -19,9 +19,11 @@
 
 package org.apache.druid.java.util.metrics;
 
+import com.google.common.annotations.VisibleForTesting;
 import com.google.common.base.Preconditions;
 import com.google.common.collect.ImmutableList;
 import com.google.common.collect.ImmutableMap;
+import com.google.common.collect.ImmutableSet;
 import org.apache.druid.java.util.emitter.service.ServiceEmitter;
 import org.apache.druid.java.util.emitter.service.ServiceMetricEvent;
 import oshi.SystemInfo;
@@ -39,6 +41,8 @@
 
 import java.util.List;
 import java.util.Map;
+import java.util.Set;
+import java.util.function.Consumer;
 
 /**
  * SysMonitor implemented using {@link oshi}
@@ -72,22 +76,34 @@ public class OshiSysMonitor extends FeedDefiningMonitor
   private final TcpStats tcpStats;
 
   private final Map<String, String[]> dimensions;
+  private final Set<String> skipEmitting;
+  private final Map<String, Consumer<ServiceEmitter>> monitoringFunctions = ImmutableMap.of(
+      "mem", this::monitorMemStats,
+      "swap", this::monitorSwapStats,
+      "fs", this::monitorFsStats,
+      "disk", this::monitorDiskStats,
+      "net", this::monitorNetStats,
+      "cpu", this::monitorCpuStats,
+      "sys", this::monitorSysStats,
+      "tcp", this::monitorTcpStats
+  );
 
   public OshiSysMonitor()
   {
-    this(ImmutableMap.of());
+    this(ImmutableMap.of(), ImmutableSet.of());
   }
 
-  public OshiSysMonitor(Map<String, String[]> dimensions)
+  public OshiSysMonitor(Map<String, String[]> dimensions, Set<String> skipEmitting)
   {
-    this(dimensions, DEFAULT_METRICS_FEED);
+    this(dimensions, DEFAULT_METRICS_FEED, skipEmitting);
   }
 
-  public OshiSysMonitor(Map<String, String[]> dimensions, String feed)
+  public OshiSysMonitor(Map<String, String[]> dimensions, String feed, Set<String> skipEmitting)
   {
     super(feed);
     Preconditions.checkNotNull(dimensions);
     this.dimensions = ImmutableMap.copyOf(dimensions);
+    this.skipEmitting = ImmutableSet.copyOf(skipEmitting);
 
     this.si = new SystemInfo();
     this.hal = si.getHardware();
@@ -105,10 +121,17 @@ public OshiSysMonitor(Map<String, String[]> dimensions, String feed)
   }
 
   // Create an object with mocked systemInfo for testing purposes
+  @VisibleForTesting
   public OshiSysMonitor(SystemInfo systemInfo)
+  {
+    this(systemInfo, ImmutableSet.of());
+  }
+
+  public OshiSysMonitor(SystemInfo systemInfo, Set<String> skipEmitting)
   {
     super("metrics");
     this.dimensions = ImmutableMap.of();
+    this.skipEmitting = ImmutableSet.copyOf(skipEmitting);
 
     this.si = systemInfo;
     this.hal = si.getHardware();
@@ -127,14 +150,11 @@ public OshiSysMonitor(SystemInfo systemInfo)
   @Override
   public boolean doMonitor(ServiceEmitter emitter)
   {
-    monitorMemStats(emitter);
-    monitorSwapStats(emitter);
-    monitorFsStats(emitter);
-    monitorDiskStats(emitter);
-    monitorNetStats(emitter);
-    monitorCpuStats(emitter);
-    monitorSysStats(emitter);
-    monitorTcpStats(emitter);
+    monitoringFunctions.forEach((key, function) -> {
+      if (!skipEmitting.contains(key)) {
+        function.accept(emitter);
+      }
+    });
     return true;
   }
 

processing/src/test/java/org/apache/druid/java/util/metrics/OshiSysMonitorTest.java

@@ -21,6 +21,7 @@
 
 import com.google.common.collect.ImmutableList;
 import com.google.common.collect.ImmutableMap;
+import com.google.common.collect.ImmutableSet;
 import org.apache.druid.java.util.common.StringUtils;
 import org.apache.druid.java.util.emitter.core.Event;
 import org.apache.druid.java.util.emitter.service.ServiceEmitter;
@@ -116,6 +117,24 @@ public void testMemStats()
     emitter.verifyValue("sys/mem/free", 16L);
   }
 
+  @Test
+  public void testMemStatsSkipOthers()
+  {
+    StubServiceEmitter emitter = new StubServiceEmitter("dev/monitor-test", "localhost:0000");
+    GlobalMemory mem = Mockito.mock(GlobalMemory.class);
+    Mockito.when(mem.getTotal()).thenReturn(64L);
+    Mockito.when(mem.getAvailable()).thenReturn(16L);
+    Mockito.when(hal.getMemory()).thenReturn(mem);
+
+    OshiSysMonitor m = new OshiSysMonitor(si, ImmutableSet.of("swap", "fs", "disk", "net", "cpu", "sys", "tcp"));
+    m.start();
+    m.doMonitor(emitter);
+    m.stop();
+    Assert.assertEquals(3, emitter.getEvents().size());
+    emitter.verifyEmitted("sys/swap/pageIn", 0);
+    emitter.verifyEmitted("sys/fs/max", 0);
+  }
+
   @Test
   public void testSwapStats()
   {

server/src/main/java/org/apache/druid/server/metrics/MetricsModule.java

@@ -66,7 +66,7 @@
  */
 public class MetricsModule implements Module
 {
-  static final String MONITORING_PROPERTY_PREFIX = "druid.monitoring";
+  public static final String MONITORING_PROPERTY_PREFIX = "druid.monitoring";
   private static final Logger log = new Logger(MetricsModule.class);
   private Set<NodeRole> nodeRoles;
 
@@ -86,6 +86,7 @@ public void configure(Binder binder)
   {
     JsonConfigProvider.bind(binder, MONITORING_PROPERTY_PREFIX, DruidMonitorSchedulerConfig.class);
     JsonConfigProvider.bind(binder, MONITORING_PROPERTY_PREFIX, MonitorsConfig.class);
+    JsonConfigProvider.bind(binder, MONITORING_PROPERTY_PREFIX + "." + OshiSysConfig.PREFIX, OshiSysConfig.class);
 
     DruidBinders.metricMonitorBinder(binder); // get the binder so that it will inject the empty set at a minimum.
 
@@ -200,7 +201,11 @@ public SysMonitor getSysMonitor(DataSourceTaskIdHolder dataSourceTaskIdHolder, @
 
   @Provides
   @ManageLifecycle
-  public OshiSysMonitor getOshiSysMonitor(DataSourceTaskIdHolder dataSourceTaskIdHolder, @Self Set<NodeRole> nodeRoles)
+  public OshiSysMonitor getOshiSysMonitor(
+      DataSourceTaskIdHolder dataSourceTaskIdHolder,
+      @Self Set<NodeRole> nodeRoles,
+      OshiSysConfig oshiSysConfig
+  )
   {
     if (nodeRoles.contains(NodeRole.PEON)) {
       return new NoopOshiSysMonitor();
@@ -209,7 +214,7 @@ public OshiSysMonitor getOshiSysMonitor(DataSourceTaskIdHolder dataSourceTaskIdH
           dataSourceTaskIdHolder.getDataSource(),
           dataSourceTaskIdHolder.getTaskId()
       );
-      return new OshiSysMonitor(dimensions);
+      return new OshiSysMonitor(dimensions, oshiSysConfig.getSkipEmitting());
     }
   }
 }

server/src/main/java/org/apache/druid/server/metrics/OshiSysConfig.java

@@ -0,0 +1,47 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *   http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.druid.server.metrics;
+
+import com.fasterxml.jackson.annotation.JsonProperty;
+
+import javax.validation.constraints.NotNull;
+import java.util.ArrayList;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Set;
+
+public class OshiSysConfig
+{
+  static final String PREFIX = "oshisys";
+
+  @JsonProperty("skipEmitting")
+  @NotNull
+  private List<String> skipEmitting;
+
+  public OshiSysConfig(@JsonProperty("skipEmitting") List<String> skipEmitting)
+  {
+    this.skipEmitting = skipEmitting == null ? new ArrayList<>() : skipEmitting;
+  }
+
+  public Set<String> getSkipEmitting()
+  {
+    return new HashSet<>(skipEmitting);
+  }
+}

server/src/test/java/org/apache/druid/server/metrics/MetricsModuleTest.java

@@ -212,6 +212,44 @@ public void testGetOshiSysMonitorViaInjector()
     Assert.assertTrue(sysMonitor instanceof NoopOshiSysMonitor);
     Mockito.verify(emitter, Mockito.never()).emit(ArgumentMatchers.any(ServiceEventBuilder.class));
   }
+
+  @Test
+  public void testGetOshiSysMonitorViaInjectorBroker()
+  {
+    final Injector injector = createInjector(new Properties()
+    {
+      {
+        setProperty("druid.monitoring.oshisys.skipEmitting", "[\"mem\"]");
+      }
+    }, ImmutableSet.of(NodeRole.BROKER));
+    final OshiSysMonitor sysMonitor = injector.getInstance(OshiSysMonitor.class);
+    final ServiceEmitter emitter = Mockito.mock(ServiceEmitter.class);
+    sysMonitor.doMonitor(emitter);
+
+    Assert.assertTrue(sysMonitor instanceof OshiSysMonitor);
+    Mockito.verify(emitter, Mockito.atLeastOnce()).emit(ArgumentMatchers.any(ServiceEventBuilder.class));
+  }
+
+  @Test
+  public void testGetOshiSysMonitorViaInjectorBrokerSkipAll()
+  {
+    final Injector injector = createInjector(new Properties()
+    {
+      {
+        setProperty(
+            "druid.monitoring.oshisys.skipEmitting",
+            "[\"mem\", \"swap\", \"fs\", \"disk\", \"net\", \"cpu\", \"sys\", \"tcp\"]"
+        );
+      }
+    }, ImmutableSet.of(NodeRole.BROKER));
+    final OshiSysMonitor sysMonitor = injector.getInstance(OshiSysMonitor.class);
+    final ServiceEmitter emitter = Mockito.mock(ServiceEmitter.class);
+    sysMonitor.doMonitor(emitter);
+
+    Assert.assertTrue(sysMonitor instanceof OshiSysMonitor);
+    Mockito.verify(emitter, Mockito.never()).emit(ArgumentMatchers.any(ServiceEventBuilder.class));
+  }
+
   @Test
   public void testGetOshiSysMonitorWhenNull()
   {