diff --git a/pulsar-functions/runtime/src/main/java/org/apache/pulsar/functions/worker/WorkerConfig.java b/pulsar-functions/runtime/src/main/java/org/apache/pulsar/functions/worker/WorkerConfig.java index b1cf7cbd27130..7fc4d1b9439a7 100644 --- a/pulsar-functions/runtime/src/main/java/org/apache/pulsar/functions/worker/WorkerConfig.java +++ b/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") diff --git a/pulsar-functions/worker/src/main/java/org/apache/pulsar/functions/worker/rest/WorkerServer.java b/pulsar-functions/worker/src/main/java/org/apache/pulsar/functions/worker/rest/WorkerServer.java index 6f1fb9bb11169..56ee46cf86f1d 100644 --- a/pulsar-functions/worker/src/main/java/org/apache/pulsar/functions/worker/rest/WorkerServer.java +++ b/pulsar-functions/worker/src/main/java/org/apache/pulsar/functions/worker/rest/WorkerServer.java @@ -23,6 +23,7 @@ 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; @@ -30,6 +31,7 @@ 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)); } diff --git a/pulsar-functions/worker/src/main/java/org/apache/pulsar/functions/worker/rest/api/FunctionsMetricsResource.java b/pulsar-functions/worker/src/main/java/org/apache/pulsar/functions/worker/rest/api/FunctionsMetricsResource.java index 9d7b316553195..e9cfbfa77aa0c 100644 --- a/pulsar-functions/worker/src/main/java/org/apache/pulsar/functions/worker/rest/api/FunctionsMetricsResource.java +++ b/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 + } + } }