[functions] Don't auth functions worker metrics endpoint

pulsar-functions/runtime/src/main/java/org/apache/pulsar/functions/worker/WorkerConfig.java

@@ -100,6 +100,17 @@ public class WorkerConfig implements Serializable, PulsarConfiguration {
         doc = "The port for serving worker https requests"
     )
     private Integer workerPortTls;
+    @FieldContext(
+            category = CATEGORY_WORKER,
+            doc = "Whether the '/metrics' endpoint requires authentication. Defaults to true."
+                    + "'authenticationEnabled' must also be set for this to take effect."
+    )
+    private boolean authenticateMetricsEndpoint = true;
+    @FieldContext(
+            category = CATEGORY_WORKER,
+            doc = "Whether the '/metrics' endpoint should return default prometheus metrics. Defaults to false."
+    )
+    private boolean includeStandardPrometheusMetrics = false;
     @FieldContext(
         category = CATEGORY_WORKER,
         doc = "Classname of Pluggable JVM GC metrics logger that can log GC specific metrics")

pulsar-functions/worker/src/main/java/org/apache/pulsar/functions/worker/rest/WorkerServer.java

@@ -23,13 +23,15 @@
 import java.net.BindException;
 import java.net.URI;
 import java.util.ArrayList;
+import java.util.Collections;
 import java.util.EnumSet;
 import java.util.List;
 import java.util.Optional;
 import java.util.TimeZone;
 
 import javax.servlet.DispatcherType;
 
+import io.prometheus.client.exporter.MetricsServlet;
 import lombok.extern.slf4j.Slf4j;
 
 import org.apache.pulsar.broker.web.AuthenticationFilter;
@@ -103,7 +105,8 @@ private void init() {
                 newServletContextHandler("/admin/v2", new ResourceConfig(Resources.getApiV2Resources()), workerService));
         handlers.add(
                 newServletContextHandler("/admin/v3", new ResourceConfig(Resources.getApiV3Resources()), workerService));
-        handlers.add(newServletContextHandler("/", new ResourceConfig(Resources.getRootResources()), workerService));
+        // don't require auth for metrics or config routes
+        handlers.add(newServletContextHandler("/", new ResourceConfig(Resources.getRootResources()), workerService, workerConfig.isAuthenticateMetricsEndpoint()));
 
         RequestLogHandler requestLogHandler = new RequestLogHandler();
         Slf4jRequestLog requestLog = new Slf4jRequestLog();
@@ -142,6 +145,10 @@ private void init() {
     }
 
     public static ServletContextHandler newServletContextHandler(String contextPath, ResourceConfig config, WorkerService workerService) {
+        return newServletContextHandler(contextPath, config, workerService, true);
+    }
+
+    public static ServletContextHandler newServletContextHandler(String contextPath, ResourceConfig config, WorkerService workerService, boolean requireAuthentication) {
         final ServletContextHandler contextHandler =
                 new ServletContextHandler(ServletContextHandler.NO_SESSIONS);
 
@@ -153,7 +160,7 @@ public static ServletContextHandler newServletContextHandler(String contextPath,
         final ServletHolder apiServlet =
                 new ServletHolder(new ServletContainer(config));
         contextHandler.addServlet(apiServlet, "/*");
-        if (workerService.getWorkerConfig().isAuthenticationEnabled()) {
+        if (workerService.getWorkerConfig().isAuthenticationEnabled() && requireAuthentication) {
             FilterHolder filter = new FilterHolder(new AuthenticationFilter(workerService.getAuthenticationService()));
             contextHandler.addFilter(filter, MATCH_ALL, EnumSet.allOf(DispatcherType.class));
         }

pulsar-functions/worker/src/main/java/org/apache/pulsar/functions/worker/rest/api/FunctionsMetricsResource.java

@@ -20,6 +20,8 @@
 
 import io.netty.buffer.ByteBuf;
 import io.netty.buffer.ByteBufAllocator;
+import io.prometheus.client.CollectorRegistry;
+import io.prometheus.client.exporter.common.TextFormat;
 import org.apache.pulsar.common.util.SimpleTextOutputStream;
 import org.apache.pulsar.functions.worker.FunctionsStatsGenerator;
 import org.apache.pulsar.functions.worker.WorkerService;
@@ -31,17 +33,26 @@
 import javax.ws.rs.core.MediaType;
 import javax.ws.rs.core.Response;
 import javax.ws.rs.core.StreamingOutput;
+import java.io.IOException;
+import java.io.Writer;
+import java.nio.CharBuffer;
+import java.nio.charset.StandardCharsets;
 
 @Path("/")
 public class FunctionsMetricsResource extends FunctionApiResource {
     @Path("metrics")
     @GET
     @Produces(MediaType.TEXT_PLAIN)
-    public Response getMetrics() {
+    public Response getMetrics() throws IOException {
 
         WorkerService workerService = get();
-
         ByteBuf buf = ByteBufAllocator.DEFAULT.heapBuffer();
+        // if request, also attach the prometheus metrics
+        if (workerService.getWorkerConfig().isIncludeStandardPrometheusMetrics()) {
+            Writer writer = new BufWriter(buf);
+            TextFormat.write004(writer, CollectorRegistry.defaultRegistry.metricFamilySamples());
+        }
+
         try {
             SimpleTextOutputStream stream = new SimpleTextOutputStream(buf);
             FunctionsStatsGenerator.generate(workerService,"default", stream);
@@ -60,4 +71,28 @@ public Response getMetrics() {
             buf.release();
         }
     }
+
+    private static class BufWriter extends Writer {
+        private final ByteBuf buf;
+
+        public BufWriter(ByteBuf buf) {
+            this.buf = buf;
+        }
+
+        @Override
+        public void write(char[] cbuf, int off, int len) throws IOException {
+            buf.writeCharSequence(CharBuffer.wrap(cbuf, off, len), StandardCharsets.UTF_8);
+        }
+
+        @Override
+        public void flush() throws IOException {
+            // noop
+
+        }
+
+        @Override
+        public void close() throws IOException {
+            // noop
+        }
+    }
 }