KIP-606 MetricsContext

core/src/main/java/io/confluent/rest/Application.java

@@ -19,7 +19,12 @@
 import com.fasterxml.jackson.databind.ObjectMapper;
 import com.fasterxml.jackson.jaxrs.base.JsonParseExceptionMapper;
 
+import io.confluent.rest.metrics.RestMetricsContext;
 import org.apache.kafka.common.config.ConfigException;
+import org.apache.kafka.common.metrics.JmxReporter;
+import org.apache.kafka.common.metrics.MetricConfig;
+import org.apache.kafka.common.metrics.Metrics;
+import org.apache.kafka.common.metrics.MetricsReporter;
 import org.eclipse.jetty.jaas.JAASLoginService;
 import org.eclipse.jetty.security.ConstraintMapping;
 import org.eclipse.jetty.security.ConstraintSecurityHandler;
@@ -69,10 +74,6 @@
 import javax.servlet.ServletException;
 import javax.ws.rs.core.Configurable;
 
-import org.apache.kafka.common.metrics.JmxReporter;
-import org.apache.kafka.common.metrics.MetricConfig;
-import org.apache.kafka.common.metrics.Metrics;
-import org.apache.kafka.common.metrics.MetricsReporter;
 import io.confluent.rest.auth.AuthUtil;
 import io.confluent.rest.exceptions.ConstraintViolationExceptionMapper;
 import io.confluent.rest.exceptions.GenericExceptionMapper;
@@ -93,6 +94,7 @@
 public abstract class Application<T extends RestConfig> {
   // CHECKSTYLE_RULES.ON: ClassDataAbstractionCoupling
   protected T config;
+  protected final RestMetricsContext metricsContext;
   private final String path;
 
   protected ApplicationServer server;
@@ -119,9 +121,10 @@ public Application(T config, String path) {
     List<MetricsReporter> reporters =
         config.getConfiguredInstances(RestConfig.METRICS_REPORTER_CLASSES_CONFIG,
                                       MetricsReporter.class);
-    reporters.add(new JmxReporter(config.getString(RestConfig.METRICS_JMX_PREFIX_CONFIG)));
-    this.metrics = new Metrics(metricConfig, reporters, config.getTime());
+    reporters.add(new JmxReporter());
 
+    metricsContext = config.getMetricsContext();
+    this.metrics = new Metrics(metricConfig, reporters, config.getTime(), metricsContext);
     this.getMetricsTags().putAll(
             parseListToMap(config.getList(RestConfig.METRICS_TAGS_CONFIG)));
 
@@ -134,6 +137,7 @@ public final String getPath() {
     return path;
   }
 
+
   /**
    * Returns {@link Metrics} object
    */

core/src/main/java/io/confluent/rest/RestConfig.java

@@ -16,22 +16,26 @@
 
 package io.confluent.rest;
 
+import io.confluent.rest.extension.ResourceExtension;
+import io.confluent.rest.metrics.RestMetricsContext;
+import java.util.Arrays;
+import java.util.Collections;
+import java.util.List;
+import java.util.Map;
+import java.util.TreeMap;
 import org.apache.kafka.common.config.AbstractConfig;
 import org.apache.kafka.common.config.ConfigException;
 import org.apache.kafka.common.config.ConfigDef;
 import org.apache.kafka.common.config.ConfigDef.Type;
 import org.apache.kafka.common.config.ConfigDef.Importance;
 import org.apache.kafka.common.utils.Time;
 
-import io.confluent.rest.extension.ResourceExtension;
-
-import java.util.Arrays;
-import java.util.Collections;
-import java.util.List;
-import java.util.Map;
-import java.util.TreeMap;
+import static org.apache.kafka.clients.CommonClientConfigs.METRICS_CONTEXT_PREFIX;
 
 public class RestConfig extends AbstractConfig {
+
+  private final RestMetricsContext metricsContext;
+
   public static final String DEBUG_CONFIG = "debug";
   protected static final String DEBUG_CONFIG_DOC =
       "Boolean indicating whether extra debugging information is generated in some "
@@ -667,16 +671,23 @@ private static ConfigDef incompleteBaseConfigDef() {
 
   public RestConfig(ConfigDef definition, Map<?, ?> originals) {
     super(definition, originals);
+    metricsContext = new RestMetricsContext(
+            this.getString(METRICS_JMX_PREFIX_CONFIG),
+            originalsWithPrefix(METRICS_CONTEXT_PREFIX));
   }
 
   public RestConfig(ConfigDef definition) {
-    super(definition, new TreeMap<>());
+    this(definition, new TreeMap<>());
   }
 
   public Time getTime() {
     return defaultTime;
   }
 
+  public RestMetricsContext getMetricsContext() {
+    return metricsContext;
+  }
+
   public static void validateHttpResponseHeaderConfig(String config) {
     try {
       // validate format

core/src/main/java/io/confluent/rest/metrics/RestMetricsContext.java

@@ -0,0 +1,63 @@
+/*
+ * Copyright 2014 Confluent Inc.
+ *
+ * Licensed 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 io.confluent.rest.metrics;
+
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.Map;
+import org.apache.kafka.common.metrics.MetricsContext;
+
+public final class RestMetricsContext implements MetricsContext {
+  /**
+   * Client or Service's metadata map.
+   */
+  private final Map<String, String> contextLabels;
+
+  public RestMetricsContext(String namespace,
+                            Map<String, Object> config) {
+    contextLabels = new HashMap<>();
+    contextLabels.put(MetricsContext.NAMESPACE, namespace);
+    config.forEach((key, value) -> contextLabels.put(key, value.toString()));
+  }
+
+  /**
+   * Sets a {@link MetricsContext} key, value pair.
+   */
+  public void setLabel(String labelKey, String labelValue) {
+    this.contextLabels.put(labelKey, labelValue);
+  }
+
+  public void setLabels(Map<String, String> labels) {
+    labels.forEach(this::setLabel);
+  }
+
+  /**
+   * Returns the value associated with  the specified label else
+   * {@code null}.
+   */
+  public String getLabel(String label) {
+    return contextLabels.get(label);
+  }
+
+  /**
+   * Returns {@link MetricsContext} as an unmodifiable Map.
+   */
+  @Override
+  public Map<String, String> contextLabels() {
+    return Collections.unmodifiableMap(contextLabels);
+  }
+}

core/src/test/java/io/confluent/rest/ApplicationTest.java

@@ -16,8 +16,39 @@
 
 package io.confluent.rest;
 
-import com.google.common.collect.ImmutableMap;
+import static io.confluent.rest.metrics.TestRestMetricsContext.RESOURCE_LABEL_TYPE;
+import static org.apache.kafka.common.metrics.MetricsContext.NAMESPACE;
+import static org.hamcrest.CoreMatchers.containsString;
+import static org.hamcrest.CoreMatchers.is;
+import static org.hamcrest.MatcherAssert.assertThat;
+import static org.junit.Assert.assertArrayEquals;
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertFalse;
+import static org.junit.Assert.assertNotNull;
+import static org.junit.Assert.assertTrue;
 
+import com.google.common.collect.ImmutableMap;
+import io.confluent.rest.extension.ResourceExtension;
+import java.io.IOException;
+import java.lang.management.ManagementFactory;
+import java.net.URI;
+import java.net.URL;
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Map;
+import java.util.concurrent.atomic.AtomicBoolean;
+import java.util.stream.Collectors;
+import javax.management.MBeanServer;
+import javax.management.ObjectName;
+import javax.ws.rs.GET;
+import javax.ws.rs.Path;
+import javax.ws.rs.Produces;
+import javax.ws.rs.core.Configurable;
+import javax.ws.rs.core.MediaType;
 import org.apache.http.client.methods.CloseableHttpResponse;
 import org.apache.http.client.methods.HttpGet;
 import org.apache.http.impl.client.CloseableHttpClient;
@@ -39,36 +70,6 @@
 import org.junit.Test;
 import org.junit.rules.ExpectedException;
 
-import java.io.IOException;
-import java.net.URI;
-import java.net.URL;
-import java.util.ArrayList;
-import java.util.Arrays;
-import java.util.Collections;
-import java.util.HashMap;
-import java.util.HashSet;
-import java.util.List;
-import java.util.Map;
-import java.util.concurrent.atomic.AtomicBoolean;
-import java.util.stream.Collectors;
-
-import javax.ws.rs.GET;
-import javax.ws.rs.Path;
-import javax.ws.rs.Produces;
-import javax.ws.rs.core.Configurable;
-import javax.ws.rs.core.MediaType;
-
-import io.confluent.rest.extension.ResourceExtension;
-
-import static org.hamcrest.CoreMatchers.containsString;
-import static org.hamcrest.CoreMatchers.is;
-import static org.hamcrest.MatcherAssert.assertThat;
-import static org.junit.Assert.assertArrayEquals;
-import static org.junit.Assert.assertEquals;
-import static org.junit.Assert.assertFalse;
-import static org.junit.Assert.assertNotNull;
-import static org.junit.Assert.assertTrue;
-
 public class ApplicationTest {
 
   private static final String REALM = "realm";
@@ -324,6 +325,56 @@ public void shouldShutdownProperlyEvenIfResourceExtensionThrowsOnShutdown() thro
     assertThat("shutdown called", TestApp.SHUTDOWN_CALLED.get(), is(true));
   }
 
+  @Test
+  public void testDefaultMetricsContext() throws Exception {
+    TestApp testApp = new TestApp();
+
+    assertEquals(testApp.metricsContext.getLabel(RESOURCE_LABEL_TYPE),
+            RestConfig.METRICS_JMX_PREFIX_DEFAULT);
+    assertEquals(testApp.metricsContext.getLabel(NAMESPACE),
+            RestConfig.METRICS_JMX_PREFIX_DEFAULT);
+  }
+
+  @Test
+  public void testMetricsContextResourceOverride() throws Exception  {
+    Map<String, Object> props = new HashMap<>();
+    props.put("metrics.context.resource.type", "FooApp");
+
+    TestApp testApp = new TestApp(props);
+
+    assertEquals(testApp.metricsContext.getLabel(RESOURCE_LABEL_TYPE), "FooApp");
+    assertEquals(testApp.metricsContext.getLabel(NAMESPACE), RestConfig.METRICS_JMX_PREFIX_DEFAULT);
+
+    /* Only NameSpace should be propagated to JMX */
+    String jmx_domain =  RestConfig.METRICS_JMX_PREFIX_DEFAULT;
+    String mbean_name = String.format("%s:type=kafka-metrics-count", jmx_domain);
+
+    MBeanServer server = ManagementFactory.getPlatformMBeanServer();
+    assertNotNull(server.getObjectInstance(new ObjectName(mbean_name)));
+  }
+
+  @Test
+  public void testMetricsContextJMXPrefixPropagation() throws Exception  {
+    Map<String, Object> props = new HashMap<>();
+    props.put(RestConfig.METRICS_JMX_PREFIX_CONFIG, "FooApp");
+
+    TestApp testApp = new TestApp(props);
+
+    assertEquals(testApp.metricsContext.getLabel(RESOURCE_LABEL_TYPE), "FooApp");
+    assertEquals(testApp.metricsContext.getLabel(NAMESPACE), "FooApp");
+  }
+
+  @Test
+  public void testMetricsContextJMXBeanRegistration() throws Exception  {
+    Map<String, Object> props = new HashMap<>();
+    props.put(RestConfig.METRICS_JMX_PREFIX_CONFIG, "FooApp");
+
+    new TestApp(props);
+
+    MBeanServer server = ManagementFactory.getPlatformMBeanServer();
+    assertNotNull(server.getObjectInstance(new ObjectName("FooApp:type=kafka-metrics-count")));
+  }
+
   private void assertExpectedUri(URI uri, String scheme, String host, int port) {
     assertEquals("Scheme should be " + scheme, scheme, uri.getScheme());
     assertEquals("Host should be " + host, host, uri.getHost());

core/src/test/java/io/confluent/rest/TestRestConfig.java

@@ -16,6 +16,10 @@
 
 package io.confluent.rest;
 
+import static org.apache.kafka.clients.CommonClientConfigs.METRICS_CONTEXT_PREFIX;
+
+import io.confluent.rest.metrics.RestMetricsContext;
+import io.confluent.rest.metrics.TestRestMetricsContext;
 import org.apache.kafka.common.config.ConfigDef;
 
 import java.util.Map;
@@ -36,4 +40,11 @@ public TestRestConfig() {
   public TestRestConfig(Map<?,?> originals) {
     super(config, originals);
   }
+
+  @Override
+  public RestMetricsContext getMetricsContext() {
+    return new TestRestMetricsContext(
+            getString(METRICS_JMX_PREFIX_CONFIG),
+            originalsWithPrefix(METRICS_CONTEXT_PREFIX)).metricsContext();
+  }
 }