[metrics] Metrics http interface unified authentication configuration (…

…apache#14756)

conf/broker.conf

@@ -1244,6 +1244,9 @@ webSocketMaxTextFrameSize=1048576
 
 ### --- Metrics --- ###
 
+# Whether the '/metrics' endpoint requires authentication. Defaults to false
+authenticateMetricsEndpoint=false
+
 # Enable topic level metrics
 exposeTopicLevelMetricsInPrometheus=true
 

conf/proxy.conf

@@ -312,3 +312,8 @@ zookeeperSessionTimeoutMs=-1
 # ZooKeeper cache expiry time in seconds
 # Deprecated: use metadataStoreCacheExpirySeconds
 zooKeeperCacheExpirySeconds=-1
+
+### --- Metrics --- ###
+
+# Whether the '/metrics' endpoint requires authentication. Defaults to true
+authenticateMetricsEndpoint=true

pulsar-broker-common/src/main/java/org/apache/pulsar/broker/ServiceConfiguration.java

@@ -2324,6 +2324,12 @@ public class ServiceConfiguration implements PulsarConfiguration {
     private int webSocketMaxTextFrameSize = 1048576;
 
     /**** --- Metrics. --- ****/
+    @FieldContext(
+            category = CATEGORY_METRICS,
+            doc = "Whether the '/metrics' endpoint requires authentication. Defaults to false."
+                    + "'authenticationEnabled' must also be set for this to take effect."
+    )
+    private boolean authenticateMetricsEndpoint = false;
     @FieldContext(
         category = CATEGORY_METRICS,
         doc = "If true, export topic level metrics otherwise namespace level"

pulsar-broker/src/main/java/org/apache/pulsar/broker/PulsarService.java

@@ -880,7 +880,7 @@ private void addWebServerHandlers(WebService webService,
         // Add metrics servlet
         webService.addServlet("/metrics",
                 new ServletHolder(metricsServlet),
-                false, attributeMap);
+                config.isAuthenticateMetricsEndpoint(), attributeMap);
 
         // Add websocket service
         addWebSocketServiceHandler(webService, attributeMap, config);

pulsar-broker/src/test/java/org/apache/pulsar/broker/stats/MetricsAuthenticationTest.java

@@ -0,0 +1,68 @@
+/**
+ * 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.pulsar.broker.stats;
+
+import com.google.common.collect.Sets;
+import javax.ws.rs.client.Client;
+import javax.ws.rs.core.Response;
+import lombok.extern.slf4j.Slf4j;
+import org.apache.pulsar.broker.auth.MockedPulsarServiceBaseTest;
+import org.glassfish.jersey.client.ClientConfig;
+import org.glassfish.jersey.logging.LoggingFeature;
+import org.testng.Assert;
+import org.testng.annotations.AfterMethod;
+import org.testng.annotations.BeforeMethod;
+import org.testng.annotations.Test;
+
+@Slf4j
+public class MetricsAuthenticationTest extends MockedPulsarServiceBaseTest {
+
+    @BeforeMethod(alwaysRun = true)
+    @Override
+    protected void setup() throws Exception {
+        conf.setAuthenticationEnabled(true);
+        conf.setAuthenticationProviders(
+                Sets.newHashSet("org.apache.pulsar.broker.auth.MockAuthenticationProvider"));
+        conf.setAuthorizationEnabled(true);
+    }
+
+    @AfterMethod(alwaysRun = true)
+    @Override
+    protected void cleanup() throws Exception {
+        super.internalCleanup();
+        resetConfig();
+    }
+
+    @Test
+    void testGetMetricsByAuthenticate() throws Exception {
+        conf.setAuthenticateMetricsEndpoint(true);
+        super.internalSetup();
+        Client client = javax.ws.rs.client.ClientBuilder.newClient(new ClientConfig().register(LoggingFeature.class));
+        Response r = client.target(this.pulsar.getWebServiceAddress()).path("/metrics").request().get();
+        Assert.assertEquals(r.getStatus(), Response.Status.UNAUTHORIZED.getStatusCode());
+    }
+
+    @Test
+    void testGetMetricsByDefault() throws Exception {
+        super.internalSetup();
+        Client client = javax.ws.rs.client.ClientBuilder.newClient(new ClientConfig().register(LoggingFeature.class));
+        Response r = client.target(this.pulsar.getWebServiceAddress()).path("/metrics").request().get();
+        Assert.assertEquals(r.getStatus(), Response.Status.OK.getStatusCode());
+    }
+}

pulsar-proxy/src/test/java/org/apache/pulsar/proxy/server/ProxyWithJwtAuthorizationTest.java

@@ -22,14 +22,20 @@
 
 import com.google.common.collect.Sets;
 
-import java.util.*;
-import java.util.concurrent.TimeUnit;
-
 import io.jsonwebtoken.Jwts;
 import io.jsonwebtoken.SignatureAlgorithm;
+import java.util.Base64;
+import java.util.HashSet;
+import java.util.Optional;
+import java.util.Set;
+import java.util.concurrent.TimeUnit;
+import javax.crypto.SecretKey;
+import javax.ws.rs.client.Client;
+import javax.ws.rs.core.Response;
 import org.apache.pulsar.broker.authentication.AuthenticationProviderToken;
 import org.apache.pulsar.broker.authentication.AuthenticationService;
 import org.apache.pulsar.broker.authentication.utils.AuthTokenUtils;
+import org.apache.pulsar.broker.resources.PulsarResources;
 import org.apache.pulsar.client.admin.PulsarAdmin;
 import org.apache.pulsar.client.api.*;
 import org.apache.pulsar.client.impl.auth.AuthenticationToken;
@@ -38,6 +44,9 @@
 import org.apache.pulsar.common.policies.data.ClusterData;
 import org.apache.pulsar.common.policies.data.SubscriptionAuthMode;
 import org.apache.pulsar.common.policies.data.TenantInfoImpl;
+import org.apache.pulsar.metadata.impl.ZKMetadataStore;
+import org.glassfish.jersey.client.ClientConfig;
+import org.glassfish.jersey.logging.LoggingFeature;
 import org.mockito.Mockito;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
@@ -46,8 +55,6 @@
 import org.testng.annotations.BeforeMethod;
 import org.testng.annotations.Test;
 
-import javax.crypto.SecretKey;
-
 public class ProxyWithJwtAuthorizationTest extends ProducerConsumerBase {
     private static final Logger log = LoggerFactory.getLogger(ProxyWithAuthorizationTest.class);
 
@@ -364,6 +371,40 @@ public void testProxyAuthorizationWithPrefixSubscriptionAuthMode() throws Except
         log.info("-- Exiting {} test --", methodName);
     }
 
+    @Test
+    void testGetMetrics() throws Exception {
+        log.info("-- Starting {} test --", methodName);
+        startProxy();
+        PulsarResources resource = new PulsarResources(new ZKMetadataStore(mockZooKeeper),
+                new ZKMetadataStore(mockZooKeeperGlobal));
+        AuthenticationService authService = new AuthenticationService(
+                PulsarConfigurationLoader.convertFrom(proxyConfig));
+        proxyConfig.setAuthenticateMetricsEndpoint(false);
+        WebServer webServer = new WebServer(proxyConfig, authService);
+        ProxyServiceStarter.addWebServerHandlers(webServer, proxyConfig, proxyService,
+                new BrokerDiscoveryProvider(proxyConfig, resource));
+        webServer.start();
+        Client client = javax.ws.rs.client.ClientBuilder.newClient(new ClientConfig().register(LoggingFeature.class));
+        try {
+            Response r = client.target(webServer.getServiceUri()).path("/metrics").request().get();
+            Assert.assertEquals(r.getStatus(), Response.Status.OK.getStatusCode());
+        } finally {
+            webServer.stop();
+        }
+        proxyConfig.setAuthenticateMetricsEndpoint(true);
+        webServer = new WebServer(proxyConfig, authService);
+        ProxyServiceStarter.addWebServerHandlers(webServer, proxyConfig, proxyService,
+                new BrokerDiscoveryProvider(proxyConfig, resource));
+        webServer.start();
+        try {
+            Response r = client.target(webServer.getServiceUri()).path("/metrics").request().get();
+            Assert.assertEquals(r.getStatus(), Response.Status.UNAUTHORIZED.getStatusCode());
+        } finally {
+            webServer.stop();
+        }
+        log.info("-- Exiting {} test --", methodName);
+    }
+
     private void createAdminClient() throws Exception {
         admin = spy(PulsarAdmin.builder().serviceHttpUrl(brokerUrl.toString())
                 .authentication(AuthenticationFactory.token(ADMIN_TOKEN)).build());