Fix logging emitter resource packaging and rename to avoid classpath collision

docs/configuration/index.md

@@ -2019,6 +2019,8 @@ log4j config to route these logs to different sources based on the feed of the e
 |--------|-----------|--------|
 |`druid.emitter.logging.loggerClass`|The class used for logging.|`org.apache.druid.java.util.emitter.core.LoggingEmitter`|
 |`druid.emitter.logging.logLevel`|Choices: debug, info, warn, error. The log level at which message are logged.|info|
+|`druid.emitter.logging.shouldFilterMetrics`|When true, only metrics listed in the allow list are emitted; non-metric events (e.g. alerts) are always emitted. When false, all events are logged (backward-compatible).|false|
+|`druid.emitter.logging.allowedMetricsPath`|Path to a JSON file whose keys are the allowed metric names. Only used when `shouldFilterMetrics` is true. If null or empty, the bundled classpath resource `loggingEmitterAllowedMetrics.json` is used. If a path is set but the file is missing, a warning is logged and the emitter falls back to the default classpath resource.|null|
 
 #### HTTP emitter module
 

processing/pom.xml

@@ -531,6 +531,9 @@
     </plugins>
 
     <resources>
+      <resource>
+        <directory>src/main/resources</directory>
+      </resource>
       <resource>
         <directory>
           ${project.build.directory}/hyperic-sigar-${sigar.base.version}/sigar-bin/lib

processing/src/main/java/org/apache/druid/java/util/emitter/core/Emitters.java

@@ -140,6 +140,16 @@ static Map<String, Object> makeLoggingMap(Properties props)
     loggingMap.put(
         "logLevel", props.getProperty("org.apache.druid.java.util.emitter.logging.level", "debug")
     );
+    if (props.containsKey("org.apache.druid.java.util.emitter.logging.shouldFilterMetrics")) {
+      loggingMap.put(
+          "shouldFilterMetrics", Boolean.parseBoolean(props.getProperty("org.apache.druid.java.util.emitter.logging.shouldFilterMetrics"))
+      );
+    }
+    if (props.containsKey("org.apache.druid.java.util.emitter.logging.allowedMetricsPath")) {
+      loggingMap.put(
+          "allowedMetricsPath", props.getProperty("org.apache.druid.java.util.emitter.logging.allowedMetricsPath")
+      );
+    }
     return loggingMap;
   }
 

processing/src/main/java/org/apache/druid/java/util/emitter/core/LoggingEmitter.java

@@ -22,36 +22,128 @@
 /**
  */
 
+import com.fasterxml.jackson.core.type.TypeReference;
 import com.fasterxml.jackson.databind.ObjectMapper;
+import com.google.common.base.Strings;
+import org.apache.druid.error.DruidException;
 import org.apache.druid.java.util.common.StringUtils;
 import org.apache.druid.java.util.common.lifecycle.LifecycleStart;
 import org.apache.druid.java.util.common.lifecycle.LifecycleStop;
 import org.apache.druid.java.util.common.logger.Logger;
+import org.apache.druid.java.util.emitter.service.ServiceMetricEvent;
 import org.slf4j.MarkerFactory;
 
+import javax.annotation.Nullable;
+import java.io.File;
+import java.io.FileInputStream;
+import java.io.FileNotFoundException;
+import java.io.IOException;
+import java.io.InputStream;
+import java.util.Collections;
+import java.util.Map;
+import java.util.Set;
 import java.util.concurrent.RejectedExecutionException;
 import java.util.concurrent.atomic.AtomicBoolean;
 
 /**
  */
 public class LoggingEmitter implements Emitter
 {
+  private static final Logger LOGGER = new Logger(LoggingEmitter.class);
+  private static final String DEFAULT_ALLOWED_METRICS_RESOURCE = "loggingEmitterAllowedMetrics.json";
+
   private final Logger log;
   private final Level level;
   private final ObjectMapper jsonMapper;
+  @Nullable
+  private final Set<String> allowedMetrics;
 
   private final AtomicBoolean started = new AtomicBoolean(false);
 
   public LoggingEmitter(LoggingEmitterConfig config, ObjectMapper jsonMapper)
   {
-    this(new Logger(config.getLoggerClass()), Level.toLevel(config.getLogLevel()), jsonMapper);
+    this(
+        new Logger(config.getLoggerClass()),
+        Level.toLevel(config.getLogLevel()),
+        jsonMapper,
+        config.shouldFilterMetrics(),
+        config.getAllowedMetricsPath()
+    );
   }
 
   public LoggingEmitter(Logger log, Level level, ObjectMapper jsonMapper)
+  {
+    this(log, level, jsonMapper, false, null);
+  }
+
+  public LoggingEmitter(
+      Logger log,
+      Level level,
+      ObjectMapper jsonMapper,
+      boolean shouldFilterMetrics,
+      @Nullable String allowedMetricsPath
+  )
   {
     this.log = log;
     this.level = level;
     this.jsonMapper = jsonMapper;
+    this.allowedMetrics = shouldFilterMetrics ? loadAllowedMetrics(allowedMetricsPath, jsonMapper) : null;
+  }
+
+  /**
+   * Loads the allowed metric names from a JSON file. If the path is null or empty,
+   * loads from the bundled classpath resource (loggingEmitterAllowedMetrics.json).
+   * If a custom path is provided but the file is missing, logs a warning and falls
+   * back to the default classpath resource.
+   */
+  private static Set<String> loadAllowedMetrics(@Nullable String path, ObjectMapper jsonMapper)
+  {
+    final InputStream is = openAllowedMetricsStream(path);
+    try {
+      final Map<String, Object> metricsMap = jsonMapper.readValue(is, new TypeReference<Map<String, Object>>() {});
+      return Collections.unmodifiableSet(metricsMap.keySet());
+    }
+    catch (IOException e) {
+      final String source = path == null || Strings.isNullOrEmpty(path) ? DEFAULT_ALLOWED_METRICS_RESOURCE : path;
+      throw DruidException.forPersona(DruidException.Persona.OPERATOR)
+                          .ofCategory(DruidException.Category.RUNTIME_FAILURE)
+                          .build(e, "Allowed metrics file must be a JSON object with metric names as keys; failed to parse [%s]", source);
+    }
+  }
+
+  /**
+   * Opens the allowed metrics configuration stream. Uses the bundled
+   * loggingEmitterAllowedMetrics.json classpath resource when path is null/empty.
+   * When a custom path is specified but the file is missing, logs a warning and
+   * falls back to the default classpath resource.
+   */
+  private static InputStream openAllowedMetricsStream(@Nullable String path)
+  {
+    if (Strings.isNullOrEmpty(path)) {
+      LOGGER.info("Using default allowed metrics configuration from classpath resource [%s]", DEFAULT_ALLOWED_METRICS_RESOURCE);
+      return openDefaultAllowedMetricsResource();
+    }
+    try {
+      final InputStream is = new FileInputStream(new File(path));
+      LOGGER.info("Using allowed metrics configuration at [%s]", path);
+      return is;
+    }
+    catch (FileNotFoundException e) {
+      LOGGER.warn(e, "Allowed metrics file [%s] not found, falling back to default classpath resource [%s]",
+                  path, DEFAULT_ALLOWED_METRICS_RESOURCE);
+      return openDefaultAllowedMetricsResource();
+    }
+  }
+
+  private static InputStream openDefaultAllowedMetricsResource()
+  {
+    final InputStream is = LoggingEmitter.class.getClassLoader().getResourceAsStream(DEFAULT_ALLOWED_METRICS_RESOURCE);
+    if (is == null) {
+      throw DruidException.forPersona(DruidException.Persona.OPERATOR)
+                          .ofCategory(DruidException.Category.NOT_FOUND)
+                          .build("Could not find default allowed metrics resource [%s] on classpath", DEFAULT_ALLOWED_METRICS_RESOURCE);
+    }
+    return is;
   }
 
   @Override
@@ -95,6 +187,16 @@ public void emit(Event event)
         throw new RejectedExecutionException("Service not started.");
       }
     }
+
+    // Allowlist filtering: only applies to ServiceMetricEvents.
+    // Non-metric events (alerts, etc.) always pass through.
+    if (allowedMetrics != null && event instanceof ServiceMetricEvent) {
+      final String metricName = ((ServiceMetricEvent) event).getMetric();
+      if (!allowedMetrics.contains(metricName)) {
+        return;
+      }
+    }
+
     try {
       switch (level) {
         case TRACE:

processing/src/main/java/org/apache/druid/java/util/emitter/core/LoggingEmitterConfig.java

@@ -21,6 +21,7 @@
 
 import com.fasterxml.jackson.annotation.JsonProperty;
 
+import javax.annotation.Nullable;
 import javax.validation.constraints.NotNull;
 
 /**
@@ -35,6 +36,26 @@ public class LoggingEmitterConfig
   @JsonProperty
   private String logLevel = "info";
 
+  /**
+   * When true, only metrics listed in the allowed metrics configuration are emitted.
+   * If {@link #allowedMetricsPath} is null/empty, the bundled default allowlist
+   * (loggingEmitterAllowedMetrics.json on the classpath) is used. If a path is
+   * provided, it is loaded from that file instead.
+   * Defaults to false (emit all metrics, backward-compatible behavior).
+   */
+  @JsonProperty("shouldFilterMetrics")
+  private boolean shouldFilterMetrics = false;
+
+  /**
+   * Optional path to a JSON file containing an array of allowed metric names.
+   * Only used when {@link #shouldFilterMetrics} is true.
+   * If null or empty, the bundled default resource (loggingEmitterAllowedMetrics.json)
+   * is loaded from the classpath.
+   */
+  @JsonProperty
+  @Nullable
+  private String allowedMetricsPath = null;
+
   public String getLoggerClass()
   {
     return loggerClass;
@@ -45,12 +66,25 @@ public String getLogLevel()
     return logLevel;
   }
 
+  public boolean shouldFilterMetrics()
+  {
+    return shouldFilterMetrics;
+  }
+
+  @Nullable
+  public String getAllowedMetricsPath()
+  {
+    return allowedMetricsPath;
+  }
+
   @Override
   public String toString()
   {
     return "LoggingEmitterConfig{" +
            "loggerClass='" + loggerClass + '\'' +
            ", logLevel='" + logLevel + '\'' +
+           ", shouldFilterMetrics=" + shouldFilterMetrics +
+           ", allowedMetricsPath='" + allowedMetricsPath + '\'' +
            '}';
   }
 }