From b99b4c855f5e43b940fe338c6d39ddadd11156fc Mon Sep 17 00:00:00 2001 From: Addison Higham Date: Wed, 22 Apr 2020 19:37:07 -0600 Subject: [PATCH] [functions] Provide more options for func worker prometheus metrics The broker and proxy both allow for hitting the metrics endpoint without auth. The functions worker should allow that to be configurable as well. This adds an option to allow for metrics endpoint to allow the endpoint to be hit without auth Additionally, the functions worker doesn't expose the default prometheus metrics (such as JVM info, etc). This commit implements and adds an option to support that --- .../pulsar/functions/worker/WorkerConfig.java | 11 ++++++ .../functions/worker/rest/WorkerServer.java | 11 +++++- .../rest/api/FunctionsMetricsResource.java | 39 ++++++++++++++++++- 3 files changed, 57 insertions(+), 4 deletions(-) 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 + } + } }