Add metrics for prometheus observability

configurations/default/env.yml.tmp

@@ -17,3 +17,7 @@ GTFS_DATABASE_URL: jdbc:postgresql://localhost/catalogue # If running via docker
 # GTFS_DATABASE_PASSWORD:
 #MONGO_HOST: mongo-host:27017 # If running via docker, this is mongo:27017
 MONGO_DB_NAME: catalogue
+
+# If set, this API key must be passed to access the /metrics endpoint.
+# Pass via header "X-API-Key" or query parameter "api_key".
+# METRICS_API_KEY: your-secret-api-key

configurations/default/server.yml.tmp

@@ -16,6 +16,8 @@ application:
     s3_region: us-east-1
     gtfs_s3_bucket: bucket-name
 modules:
+  metrics:
+    enabled: true
   enterprise:
     enabled: false
     # Setting this to true will upload all feeds to S3 instead of linking to their URL

configurations/test/env.yml.tmp

@@ -24,3 +24,7 @@ MONGO_DB_NAME: catalogue
 #MONGO_PASSWORD: password
 #MONGO_PROTOCOL: mongodb+srv
 #MONGO_USER: user
+
+# If set, this API key must be passed to access the /metrics endpoint.
+# Pass via header "X-API-Key" or query parameter "api_key".
+# METRICS_API_KEY: your-secret-api-key

configurations/test/server.yml.tmp

@@ -15,6 +15,8 @@ application:
     s3_region: us-east-1
     gtfs_s3_bucket: bucket-name
 modules:
+  metrics:
+    enabled: true
   enterprise:
     enabled: false
   editor:

pom.xml

@@ -210,6 +210,18 @@
         </repository>
     </repositories>
 
+
+    <dependencyManagement>
+        <dependencies>
+            <dependency>
+                <groupId>io.micrometer</groupId>
+                <artifactId>micrometer-bom</artifactId>
+                <version>1.17.0</version>
+                <type>pom</type>
+                <scope>import</scope>
+            </dependency>
+        </dependencies>
+    </dependencyManagement>
     <dependencies>
         <!-- Handles HTTP server -->
         <dependency>
@@ -231,6 +243,12 @@
             <version>1.2.13</version>
         </dependency>
 
+        <!-- Metrics -->
+        <dependency>
+            <groupId>io.micrometer</groupId>
+            <artifactId>micrometer-registry-prometheus</artifactId>
+        </dependency>
+
         <!-- Used to connect to and import legacy editor MapDBs -->
         <dependency>
             <groupId>org.mapdb</groupId>

src/main/java/com/conveyal/datatools/common/status/MonitorableJob.java

@@ -1,6 +1,7 @@
 package com.conveyal.datatools.common.status;
 
 import com.conveyal.datatools.manager.auth.Auth0UserProfile;
+import com.conveyal.datatools.manager.metrics.MetricsService;
 import com.conveyal.datatools.manager.utils.JobUtils;
 import com.fasterxml.jackson.annotation.JsonIgnore;
 import com.fasterxml.jackson.annotation.JsonProperty;
@@ -215,6 +216,7 @@ public void run () {
         } finally {
             LOG.info("{} (jobId={}) {} in {} ms", type, jobId, status.error ? "errored" : "completed", status.duration);
             active = false;
+            MetricsService.recordJobOutcome(type, status.error, status.duration);
         }
     }
 

src/main/java/com/conveyal/datatools/common/utils/Scheduler.java

@@ -1,8 +1,10 @@
 package com.conveyal.datatools.common.utils;
 
+import com.conveyal.datatools.manager.DataManager;
 import com.conveyal.datatools.manager.auth.Auth0UserProfile;
 import com.conveyal.datatools.manager.jobs.FeedExpirationNotificationJob;
 import com.conveyal.datatools.manager.jobs.FetchSingleFeedJob;
+import com.conveyal.datatools.manager.metrics.MetricsService;
 import com.conveyal.datatools.manager.models.FeedSource;
 import com.conveyal.datatools.manager.models.FeedVersion;
 import com.conveyal.datatools.manager.models.Project;
@@ -48,10 +50,22 @@ public class Scheduler {
     public final static ListMultimap<String, ScheduledJob> scheduledJobsForFeedSources =
         synchronizedListMultimap(ArrayListMultimap.create());
 
+    // Use a separate thread to handle updating metrics inventory.
+    private static final ScheduledExecutorService metricsInventoryScheduler =
+        Executors.newSingleThreadScheduledExecutor(r -> {
+            Thread t = new Thread(r, "metrics-refresh");
+            t.setDaemon(true);
+            return t;
+        });
+
+
     /**
      * A method to initialize all scheduled tasks upon server startup.
      */
     public static void initialize() {
+        if (DataManager.isModuleEnabled("metrics")) {
+            startInventoryRefresh(5, TimeUnit.MINUTES);
+        }
         LOG.info("Scheduling recurring feed auto fetches for all projects.");
         for (Project project : Persistence.projects.getAll()) {
             handleAutoFeedFetch(project);
@@ -64,6 +78,17 @@ public static void initialize() {
         }
     }
 
+    /**
+     * Schedules a task to refresh the metrics inventory on a regular interval
+     * @param period refresh interval
+     * @param unit unit for the refresh interval
+     */
+    public static void startInventoryRefresh(long period, TimeUnit unit) {
+        LOG.info("Creating a separate thread for refreshing metrics inventory.");
+        metricsInventoryScheduler.scheduleAtFixedRate(MetricsService::refreshInventory,
+            0, period, unit);
+    }
+
     /**
      * Convenience method for scheduling one-off jobs for a feed source.
      */

src/main/java/com/conveyal/datatools/manager/DataManager.java

@@ -15,6 +15,7 @@
 import com.conveyal.datatools.manager.controllers.api.FeedVersionController;
 import com.conveyal.datatools.manager.controllers.api.GtfsPlusController;
 import com.conveyal.datatools.manager.controllers.api.LabelController;
+import com.conveyal.datatools.manager.controllers.api.MetricsController;
 import com.conveyal.datatools.manager.controllers.api.NoteController;
 import com.conveyal.datatools.manager.controllers.api.OrganizationController;
 import com.conveyal.datatools.manager.controllers.api.ProjectController;
@@ -70,6 +71,8 @@ public class DataManager {
     public static final String GTFS_PLUS_SUBDIR = "gtfsplus";
     private static final Logger LOG = LoggerFactory.getLogger(DataManager.class);
 
+    public static long serverStartTime;
+
     // These fields hold YAML files that represent the server configuration.
     private static JsonNode envConfig;
     private static JsonNode serverConfig;
@@ -102,7 +105,7 @@ public class DataManager {
     public static final Map<String, RequestSummary> lastRequestForUser = new HashMap<>();
 
     public static void main(String[] args) throws IOException {
-        long serverStartTime = System.currentTimeMillis();
+        serverStartTime = System.currentTimeMillis();
         initializeApplication(args);
 
         registerRoutes();
@@ -188,7 +191,11 @@ private static void loadProperties() {
      * modules and sets other core routes (e.g., 404 response) and response headers (e.g., API content type is JSON).
      */
     static void registerRoutes() throws IOException {
+
         CorsFilter.apply();
+        if (isModuleEnabled("metrics")) {
+            MetricsController.register();
+        }
         // Initialize GTFS GraphQL API service
         // FIXME: Add user permissions check to ensure user has access to feeds.
         GraphQLController.initialize(GTFS_DATA_SOURCE, GTFS_API_PREFIX);

src/main/java/com/conveyal/datatools/manager/controllers/api/MetricsController.java

@@ -0,0 +1,34 @@
+package com.conveyal.datatools.manager.controllers.api;
+
+import com.conveyal.datatools.manager.metrics.MetricsService;
+
+import spark.Request;
+import spark.Response;
+
+import static com.conveyal.datatools.common.utils.SparkUtils.logMessageAndHalt;
+import static com.conveyal.datatools.manager.DataManager.getConfigPropertyAsText;
+import static spark.Spark.get;
+
+public class MetricsController {
+    private static final String METRICS_API_KEY_CONFIG = "METRICS_API_KEY";
+
+    /** GET /metrics — Prometheus scrape endpoint. */
+    private static String getMetrics(Request req, Response res) {
+        String apiKey = getConfigPropertyAsText(METRICS_API_KEY_CONFIG);
+        if (apiKey != null && !apiKey.isEmpty()) {
+            String providedKey = req.headers("X-API-Key");
+            if (providedKey == null) {
+                providedKey = req.queryParams("api_key");
+            }
+            if (!apiKey.equals(providedKey)) {
+                logMessageAndHalt(req, 401, "Invalid or missing API key");
+            }
+        }
+        res.type("text/plain; version=0.0.4; charset=utf-8");
+        return MetricsService.registry().scrape();
+    }
+
+    public static void register() {
+        get("/metrics", MetricsController::getMetrics);
+    }
+}

src/main/java/com/conveyal/datatools/manager/metrics/MetricsService.java

@@ -0,0 +1,140 @@
+package com.conveyal.datatools.manager.metrics;
+
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.ConcurrentMap;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicLong;
+
+import org.bson.Document;
+
+import com.conveyal.datatools.common.status.MonitorableJob.JobType;
+import com.conveyal.datatools.manager.DataManager;
+import com.conveyal.datatools.manager.persistence.Persistence;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import io.micrometer.core.instrument.Counter;
+import io.micrometer.core.instrument.Gauge;
+import io.micrometer.core.instrument.Timer;
+import io.micrometer.core.instrument.binder.jvm.ClassLoaderMetrics;
+import io.micrometer.core.instrument.binder.jvm.JvmGcMetrics;
+import io.micrometer.core.instrument.binder.jvm.JvmMemoryMetrics;
+import io.micrometer.core.instrument.binder.jvm.JvmThreadDeadlockMetrics;
+import io.micrometer.core.instrument.binder.jvm.JvmThreadMetrics;
+import io.micrometer.core.instrument.binder.system.ProcessorMetrics;
+import io.micrometer.prometheusmetrics.PrometheusConfig;
+import io.micrometer.prometheusmetrics.PrometheusMeterRegistry;
+
+public class MetricsService {
+    private static final Logger LOG = LoggerFactory.getLogger(MetricsService.class);
+
+    private static final PrometheusMeterRegistry registry;
+
+    private static final AtomicLong feedSourceCount = new AtomicLong(0);
+    private static final AtomicLong feedVersionCount = new AtomicLong(0);
+    private static final AtomicLong projectCount = new AtomicLong(0);
+    private static final AtomicLong organizationCount = new AtomicLong(0);
+    private static final AtomicLong lastRefreshEpoch = new AtomicLong(0);
+    private static final Counter refreshFailures;
+
+    private static final ConcurrentMap<JobType, Counter> failedJobsByType = new ConcurrentHashMap<>();
+    private static final ConcurrentMap<JobType, Counter> completedJobsByType = new ConcurrentHashMap<>();
+    private static final ConcurrentMap<JobType, Timer> jobDurationByType = new ConcurrentHashMap<>();
+    private static Timer jobDuration;
+
+    static {
+        registry = new PrometheusMeterRegistry(PrometheusConfig.DEFAULT);
+        populateAndRegisterJobMetricsByType();
+
+        new ClassLoaderMetrics().bindTo(registry);
+        new JvmMemoryMetrics().bindTo(registry);
+        new JvmGcMetrics().bindTo(registry);
+        new ProcessorMetrics().bindTo(registry);
+        new JvmThreadMetrics().bindTo(registry);
+        new JvmThreadDeadlockMetrics().bindTo(registry);
+
+        Gauge.builder("datatools.feed.sources", feedSourceCount, AtomicLong::get)
+            .description("Total number of Feed Sources")
+            .strongReference(true)
+            .register(registry);
+        Gauge.builder("datatools.feed.versions", feedVersionCount, AtomicLong::get)
+            .description("Total number of Feed Versions")
+            .strongReference(true)
+            .register(registry);
+        Gauge.builder("datatools.projects", projectCount, AtomicLong::get)
+            .description("Total number of Projects")
+            .strongReference(true)
+            .register(registry);
+        Gauge.builder("datatools.organizations", organizationCount, AtomicLong::get)
+            .description("Total number of Organizations")
+            .strongReference(true)
+            .register(registry);
+        Gauge.builder("datatools.inventory.refresh.age.seconds",
+                lastRefreshEpoch, e -> e.get() == 0
+                    ? Double.NaN
+                    : (System.currentTimeMillis() - e.get()) / 1000.0)
+            .description("Seconds since last inventory refresh")
+            .strongReference(true)
+            .register(registry);
+        refreshFailures = Counter.builder("datatools.inventory.refresh.failures")
+                .description("Total number of failed Refreshes")
+                .register(registry);
+        Gauge.builder("datatools.uptime.seconds",
+                DataManager.serverStartTime,
+                start -> (System.currentTimeMillis() - start) / 1000.0)
+            .description("Server uptime in seconds")
+            .register(registry);
+    }
+
+    private static void populateAndRegisterJobMetricsByType() {
+        for(JobType type : JobType.values()) {
+            failedJobsByType.put(type, Counter.builder(String.format("datatools.jobs.%s.failed", type.name()))
+                .description(String.format("Number of failed jobs of type %s", type.name()))
+                .register(registry));
+            completedJobsByType.put(type, Counter.builder(String.format("datatools.jobs.%s.completed", type.name()))
+                .description(String.format("Number of completed jobs of type %s", type.name()))
+                .register(registry));
+            jobDurationByType.put(type, Timer.builder(String.format("datatools.jobs.%s.duration", type.name()))
+                .description(String.format("Execution duration of jobs of type %s", type.name()))
+                .register(registry));
+        }
+
+        Gauge.builder("datatools.jobs.failed", failedJobsByType,
+                map -> map.values().stream().mapToDouble(Counter::count).sum())
+            .description("Number of failed jobs")
+            .register(registry);
+        Gauge.builder("datatools.jobs.completed", completedJobsByType,
+                map -> map.values().stream().mapToDouble(Counter::count).sum())
+            .description("Number of completed jobs")
+            .register(registry);
+        jobDuration = Timer.builder("datatools.jobs.duration")
+            .description("Job execution duration")
+            .register(registry);
+    }
+
+    public static PrometheusMeterRegistry registry() {
+        return registry;
+    }
+
+    public static void refreshInventory() {
+        try {
+            Document allDocuments = new Document();
+            feedSourceCount.set(Persistence.feedSources.count(allDocuments));
+            feedVersionCount.set(Persistence.feedVersions.count(allDocuments));
+            projectCount.set(Persistence.projects.count(allDocuments));
+            organizationCount.set(Persistence.organizations.count(allDocuments));
+            lastRefreshEpoch.set(System.currentTimeMillis());
+        } catch (Exception e) {
+            refreshFailures.increment();
+			LOG.warn("Inventory refresh failed", e);
+        }
+    }
+
+    public static void recordJobOutcome(JobType type, boolean error, long duration) {
+        ConcurrentMap<JobType, Counter> outcomeCountersByType = error ? failedJobsByType : completedJobsByType;
+
+        outcomeCountersByType.get(type).increment();
+        jobDuration.record(duration, TimeUnit.MILLISECONDS);
+        jobDurationByType.get(type).record(duration, TimeUnit.MILLISECONDS);
+    }
+}