Add Prometheus metrics server and pipeline metrics

pom.xml

@@ -2,7 +2,7 @@
     <modelVersion>4.0.0</modelVersion>
     <groupId>fi.hsl</groupId>
     <artifactId>transitdata-common</artifactId>
-    <version>3.0.0</version>
+    <version>3.1.0</version>
     <packaging>jar</packaging>
     <name>Common utilities for Transitdata projects</name>
 
@@ -14,6 +14,7 @@
         <spotlessMavenPlugin.version>2.43.0</spotlessMavenPlugin.version>
         <googleJavaFormat.version>1.23.0</googleJavaFormat.version>
         <jackson.version>2.20.1</jackson.version>
+        <prometheus.version>0.16.0</prometheus.version>
     </properties>
 
     <profiles>
@@ -161,7 +162,16 @@
             <version>3.26.0</version>
             <scope>test</scope>
         </dependency>
-
+        <dependency>
+            <groupId>io.prometheus</groupId>
+            <artifactId>simpleclient</artifactId>
+            <version>${prometheus.version}</version>
+        </dependency>
+        <dependency>
+            <groupId>io.prometheus</groupId>
+            <artifactId>simpleclient_httpserver</artifactId>
+            <version>${prometheus.version}</version>
+        </dependency>
     </dependencies>
 
     <build>

src/main/java/fi/hsl/common/metrics/MetricsServer.java

@@ -0,0 +1,27 @@
+package fi.hsl.common.metrics;
+
+import io.prometheus.client.exporter.HTTPServer;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.io.IOException;
+
+public class MetricsServer implements AutoCloseable {
+    private static final Logger log = LoggerFactory.getLogger(MetricsServer.class);
+
+    public final int port;
+    private final HTTPServer httpServer;
+
+    public MetricsServer(final int port) throws IOException {
+        this.port = port;
+        log.info("Starting MetricsServer on port {}, endpoint /metrics", port);
+        this.httpServer = new HTTPServer.Builder().withPort(port).build();
+    }
+
+    @Override
+    public void close() {
+        if (httpServer != null) {
+            httpServer.close();
+        }
+    }
+}

src/main/java/fi/hsl/common/metrics/PipelineMetrics.java

@@ -0,0 +1,28 @@
+package fi.hsl.common.metrics;
+
+import io.prometheus.client.Histogram;
+
+public final class PipelineMetrics {
+
+    private static final double[] LATENCY_BUCKETS = {0.05, 0.1, 0.25, 0.5, 1, 2, 5, 10, 30, 60, 120};
+
+    private static final Histogram E2E_LATENCY = Histogram.build().name("transitdata_e2e_latency_seconds")
+            .help("Internal processing latency from ingest to produced output, per pipeline.").labelNames("pipeline")
+            .buckets(LATENCY_BUCKETS).register();
+
+    private static final Histogram PUBLISH_DURATION = Histogram.build()
+            .name("transitdata_feed_publish_duration_seconds")
+            .help("Time to build and write a full feed dump, per pipeline.").labelNames("pipeline")
+            .buckets(LATENCY_BUCKETS).register();
+
+    private PipelineMetrics() {
+    }
+
+    public static void observeLatency(final String pipeline, final double seconds) {
+        E2E_LATENCY.labels(pipeline).observe(seconds);
+    }
+
+    public static void observePublishDuration(final String pipeline, final double seconds) {
+        PUBLISH_DURATION.labels(pipeline).observe(seconds);
+    }
+}

src/main/java/fi/hsl/common/pulsar/PulsarApplication.java

@@ -2,6 +2,7 @@
 
 import com.typesafe.config.Config;
 import fi.hsl.common.health.HealthServer;
+import fi.hsl.common.metrics.MetricsServer;
 import fi.hsl.common.redis.RedisClusterProperties;
 import fi.hsl.common.redis.RedisStore;
 import org.apache.pulsar.client.admin.PulsarAdmin;
@@ -46,6 +47,7 @@ public class PulsarApplication implements AutoCloseable {
     PulsarAdmin admin;
     RedisStore redisStore;
     HealthServer healthServer;
+    MetricsServer metricsServer;
 
     PulsarApplication() {
     }
@@ -120,6 +122,10 @@ public PulsarApplicationContext initialize(@NotNull Config config) throws Except
             }
         }
 
+        if (config.hasPath("metrics.enabled") && config.getBoolean("metrics.enabled")) {
+            metricsServer = new MetricsServer(config.getInt("metrics.port"));
+        }
+
         return createContext(config, client, consumer, producers, redisStore, admin, healthServer);
     }
 
@@ -322,5 +328,7 @@ public void close() {
             redisStore.close();
         if (healthServer != null)
             healthServer.close();
+        if (metricsServer != null)
+            metricsServer.close();
     }
 }

src/main/resources/common.conf

@@ -63,6 +63,13 @@ health {
   endpoint=${?HEALTH_ENDPOINT}
 }
 
+metrics {
+  enabled=false
+  enabled=${?METRICS_ENABLED}
+  port=8080
+  port=${?METRICS_PORT}
+}
+
 gtfsrt {
   serviceDayStartTime="04:30:00"
   serviceDayStartTime=${?GTFSRT_SERVICE_DAY_START_TIME}