-
Notifications
You must be signed in to change notification settings - Fork 10
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
Keep track of HTTPServer instances (#40)
Signed-off-by: Mickael Maison <[email protected]>
- Loading branch information
Showing
8 changed files
with
321 additions
and
120 deletions.
There are no files selected for viewing
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,91 @@ | ||
| /* | ||
| * Copyright Strimzi authors. | ||
| * License: Apache License 2.0 (see the file LICENSE or http://apache.org/licenses/LICENSE-2.0.html). | ||
| */ | ||
| package io.strimzi.kafka.metrics; | ||
|
|
||
| import io.prometheus.metrics.exporter.httpserver.HTTPServer; | ||
| import io.prometheus.metrics.model.registry.PrometheusRegistry; | ||
| import org.slf4j.Logger; | ||
| import org.slf4j.LoggerFactory; | ||
|
|
||
| import java.io.IOException; | ||
| import java.util.HashMap; | ||
| import java.util.Map; | ||
| import java.util.concurrent.atomic.AtomicInteger; | ||
|
|
||
| /** | ||
| * Class to keep track of all the HTTP servers started by all the Kafka components in a JVM. | ||
| */ | ||
| public class HttpServers { | ||
|
|
||
| private final static Logger LOG = LoggerFactory.getLogger(HttpServers.class); | ||
| private static final Map<Listener, ServerCounter> SERVERS = new HashMap<>(); | ||
|
|
||
| /** | ||
| * Get or create a new HTTP server if there isn't an existing instance for the specified listener. | ||
| * @param listener The host and port | ||
| * @param registry The Prometheus registry to expose | ||
| * @return A ServerCounter instance | ||
| * @throws IOException if the HTTP server does not exist and cannot be started | ||
| */ | ||
| public synchronized static ServerCounter getOrCreate(Listener listener, PrometheusRegistry registry) throws IOException { | ||
| ServerCounter serverCounter = SERVERS.get(listener); | ||
| if (serverCounter == null) { | ||
| serverCounter = new ServerCounter(listener, registry); | ||
| SERVERS.put(listener, serverCounter); | ||
| } | ||
| serverCounter.count.incrementAndGet(); | ||
| return serverCounter; | ||
| } | ||
|
|
||
| /** | ||
| * Release an HTTP server instance. If no other components hold this instance, it is closed. | ||
| * @param serverCounter The HTTP server instance to release | ||
| */ | ||
| public synchronized static void release(ServerCounter serverCounter) { | ||
| if (serverCounter.close()) { | ||
| SERVERS.remove(serverCounter.listener); | ||
| } | ||
| } | ||
|
|
||
| /** | ||
| * Class used to keep track of the HTTP server started on a listener. | ||
| */ | ||
| public static class ServerCounter { | ||
|
|
||
| private final AtomicInteger count; | ||
| private final HTTPServer server; | ||
| private final Listener listener; | ||
|
|
||
| private ServerCounter(Listener listener, PrometheusRegistry registry) throws IOException { | ||
| this.count = new AtomicInteger(); | ||
| this.server = HTTPServer.builder() | ||
| .hostname(listener.host) | ||
| .port(listener.port) | ||
| .registry(registry) | ||
| .buildAndStart(); | ||
| LOG.debug("Started HTTP server on http://{}:{}", listener.host, server.getPort()); | ||
| this.listener = listener; | ||
| } | ||
|
|
||
| /** | ||
| * The port this HTTP server instance is listening on. If the listener port is 0, this returns the actual port | ||
| * that is used. | ||
| * @return The port number | ||
| */ | ||
| public int port() { | ||
| return server.getPort(); | ||
| } | ||
|
|
||
| private synchronized boolean close() { | ||
| int remaining = count.decrementAndGet(); | ||
| if (remaining == 0) { | ||
| server.close(); | ||
| LOG.debug("Stopped HTTP server on http://{}:{}", listener.host, server.getPort()); | ||
| return true; | ||
| } | ||
| return false; | ||
| } | ||
| } | ||
| } |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,74 @@ | ||
| /* | ||
| * Copyright Strimzi authors. | ||
| * License: Apache License 2.0 (see the file LICENSE or http://apache.org/licenses/LICENSE-2.0.html). | ||
| */ | ||
| package io.strimzi.kafka.metrics; | ||
|
|
||
| import org.apache.kafka.common.config.ConfigDef; | ||
| import org.apache.kafka.common.config.ConfigException; | ||
|
|
||
| import java.util.Objects; | ||
| import java.util.regex.Matcher; | ||
| import java.util.regex.Pattern; | ||
|
|
||
| import static io.strimzi.kafka.metrics.PrometheusMetricsReporterConfig.LISTENER_CONFIG; | ||
|
|
||
| /** | ||
| * Class parsing and handling the listener specified via {@link PrometheusMetricsReporterConfig#LISTENER_CONFIG} for | ||
| * the HTTP server used to expose the metrics. | ||
| */ | ||
| public class Listener { | ||
|
|
||
| private static final Pattern PATTERN = Pattern.compile("http://\\[?([0-9a-zA-Z\\-%._:]*)]?:([0-9]+)"); | ||
|
|
||
| final String host; | ||
| final int port; | ||
|
|
||
| /* test */ Listener(String host, int port) { | ||
| this.host = host; | ||
| this.port = port; | ||
| } | ||
|
|
||
| static Listener parseListener(String listener) { | ||
| Matcher matcher = PATTERN.matcher(listener); | ||
| if (matcher.matches()) { | ||
| String host = matcher.group(1); | ||
| int port = Integer.parseInt(matcher.group(2)); | ||
| return new Listener(host, port); | ||
| } else { | ||
| throw new ConfigException(LISTENER_CONFIG, listener, "Listener must be of format http://[host]:[port]"); | ||
| } | ||
| } | ||
|
|
||
| @Override | ||
| public String toString() { | ||
| return "http://" + host + ":" + port; | ||
| } | ||
|
|
||
| @Override | ||
| public boolean equals(Object o) { | ||
| if (this == o) return true; | ||
| if (o == null || getClass() != o.getClass()) return false; | ||
| Listener listener = (Listener) o; | ||
| return port == listener.port && Objects.equals(host, listener.host); | ||
| } | ||
|
|
||
| @Override | ||
| public int hashCode() { | ||
| return Objects.hash(host, port); | ||
| } | ||
|
|
||
| /** | ||
| * Validator to check the user provided listener configuration | ||
| */ | ||
| static class ListenerValidator implements ConfigDef.Validator { | ||
|
|
||
| @Override | ||
| public void ensureValid(String name, Object value) { | ||
| Matcher matcher = PATTERN.matcher(String.valueOf(value)); | ||
| if (!matcher.matches()) { | ||
| throw new ConfigException(name, value, "Listener must be of format http://[host]:[port]"); | ||
| } | ||
| } | ||
| } | ||
| } |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
62 changes: 62 additions & 0 deletions
62
src/test/java/io/strimzi/kafka/metrics/HttpServersTest.java
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,62 @@ | ||
| /* | ||
| * Copyright Strimzi authors. | ||
| * License: Apache License 2.0 (see the file LICENSE or http://apache.org/licenses/LICENSE-2.0.html). | ||
| */ | ||
| package io.strimzi.kafka.metrics; | ||
|
|
||
| import io.prometheus.metrics.model.registry.PrometheusRegistry; | ||
| import org.junit.jupiter.api.Test; | ||
|
|
||
| import java.io.IOException; | ||
| import java.net.HttpURLConnection; | ||
| import java.net.URL; | ||
|
|
||
| import static org.junit.jupiter.api.Assertions.assertFalse; | ||
| import static org.junit.jupiter.api.Assertions.assertSame; | ||
| import static org.junit.jupiter.api.Assertions.assertTrue; | ||
|
|
||
| public class HttpServersTest { | ||
|
|
||
| private final PrometheusRegistry registry = new PrometheusRegistry(); | ||
|
|
||
| @Test | ||
| public void testLifecycle() throws IOException { | ||
| Listener listener1 = Listener.parseListener("http://localhost:0"); | ||
| HttpServers.ServerCounter server1 = HttpServers.getOrCreate(listener1, registry); | ||
| assertTrue(listenerStarted(listener1.host, server1.port())); | ||
|
|
||
| Listener listener2 = Listener.parseListener("http://localhost:0"); | ||
| HttpServers.ServerCounter server2 = HttpServers.getOrCreate(listener2, registry); | ||
| assertTrue(listenerStarted(listener2.host, server2.port())); | ||
| assertSame(server1, server2); | ||
|
|
||
| Listener listener3 = Listener.parseListener("http://127.0.0.1:0"); | ||
| HttpServers.ServerCounter server3 = HttpServers.getOrCreate(listener3, registry); | ||
| assertTrue(listenerStarted(listener3.host, server3.port())); | ||
|
|
||
| HttpServers.release(server1); | ||
| assertTrue(listenerStarted(listener1.host, server1.port())); | ||
| assertTrue(listenerStarted(listener2.host, server2.port())); | ||
| assertTrue(listenerStarted(listener3.host, server3.port())); | ||
|
|
||
| HttpServers.release(server2); | ||
| assertFalse(listenerStarted(listener1.host, server1.port())); | ||
| assertFalse(listenerStarted(listener2.host, server2.port())); | ||
| assertTrue(listenerStarted(listener3.host, server3.port())); | ||
|
|
||
| HttpServers.release(server3); | ||
| assertFalse(listenerStarted(listener3.host, server3.port())); | ||
| } | ||
|
|
||
| private boolean listenerStarted(String host, int port) { | ||
| try { | ||
| URL url = new URL("http://" + host + ":" + port + "/metrics"); | ||
| HttpURLConnection con = (HttpURLConnection) url.openConnection(); | ||
| con.setRequestMethod("HEAD"); | ||
| con.connect(); | ||
| return con.getResponseCode() == HttpURLConnection.HTTP_OK; | ||
| } catch (IOException ioe) { | ||
| return false; | ||
| } | ||
| } | ||
| } |
Oops, something went wrong.