Add statsd module

metrics-statsd/pom.xml

@@ -0,0 +1,34 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<project xmlns="http://maven.apache.org/POM/4.0.0"
+         xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
+         xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
+  <modelVersion>4.0.0</modelVersion>
+  <parent>
+    <groupId>io.avaje</groupId>
+    <artifactId>avaje-metrics-parent</artifactId>
+    <version>9.5</version>
+  </parent>
+
+  <artifactId>avaje-metrics-statsd</artifactId>
+
+  <dependencies>
+    <dependency>
+      <groupId>io.avaje</groupId>
+      <artifactId>avaje-metrics</artifactId>
+      <version>9.5</version>
+    </dependency>
+    <dependency>
+      <groupId>com.datadoghq</groupId>
+      <artifactId>java-dogstatsd-client</artifactId>
+      <version>4.4.4</version>
+    </dependency>
+    <dependency>
+      <groupId>io.ebean</groupId>
+      <artifactId>ebean-api</artifactId>
+      <version>17.0.0-RC1</version>
+      <scope>provided</scope>
+      <optional>true</optional>
+    </dependency>
+  </dependencies>
+
+</project>

metrics-statsd/src/main/java/io/avaje/metrics/statsd/DatabaseReporter.java

@@ -0,0 +1,102 @@
+package io.avaje.metrics.statsd;
+
+import com.timgroup.statsd.StatsDClient;
+import io.ebean.Database;
+import io.ebean.datasource.DataSourcePool;
+import io.ebean.meta.MetaCountMetric;
+import io.ebean.meta.MetaQueryMetric;
+import io.ebean.meta.MetaTimedMetric;
+import io.ebean.meta.ServerMetrics;
+
+import javax.sql.DataSource;
+
+final class DatabaseReporter implements StatsdReporter.Reporter {
+
+  private final Database database;
+
+  private DatabaseReporter(Database database) {
+    this.database = database;
+  }
+
+  static StatsdReporter.Reporter reporter(Database database) {
+    return new DatabaseReporter(database);
+  }
+
+  @Override
+  public void report(StatsDClient sender) {
+    new DReport(database, sender).report();
+  }
+
+  private static final class DReport {
+
+    private final StatsDClient reporter;
+    private final ServerMetrics dbMetrics;
+    private final long epochSecs;
+    private final String dbName;
+
+    private DReport(Database database, StatsDClient reporter) {
+      this.reporter = reporter;
+      this.dbName = database.name();
+      this.dbMetrics = database.metaInfo().collectMetrics();
+      this.epochSecs = System.currentTimeMillis() / 1000;
+      DataSource dataSource = database.dataSource();
+      if (dataSource instanceof DataSourcePool) {
+        DataSourcePool pool = (DataSourcePool) database;
+        reporter.gaugeWithTimestamp("pool.size", pool.size(), epochSecs, "db", dbName, "type", "main");
+      }
+      DataSource readDatasource = database.readOnlyDataSource();
+      if (readDatasource instanceof DataSourcePool && readDatasource != dataSource) {
+        DataSourcePool readOnlyPool = (DataSourcePool) readDatasource;
+        reporter.gaugeWithTimestamp("pool.size", readOnlyPool.size(), epochSecs,"db", dbName, "type", "readonly");
+      }
+    }
+
+    private String nm(String name, String suffix) {
+      return name + suffix;
+    }
+
+    private void report() {
+      for (MetaTimedMetric timedMetric : dbMetrics.timedMetrics()) {
+        reportTimedMetric(timedMetric);
+      }
+      for (MetaQueryMetric queryMetric : dbMetrics.queryMetrics()) {
+        reportQueryMetric(queryMetric);
+      }
+      for (MetaCountMetric countMetric : dbMetrics.countMetrics()) {
+        reportCountMetric(countMetric);
+      }
+    }
+
+    private void reportCountMetric(MetaCountMetric countMetric) {
+      reporter.count(nm(countMetric.name(), ".count"), countMetric.count(), "db", dbName);
+    }
+
+    private void reportTimedMetric(MetaTimedMetric metric) {
+      final String name = metric.name();
+      if (name.startsWith("txn")) {
+        reportMetric(metric, "txn", "name", name, "db", dbName);
+      } else {
+        reportMetric(metric, name, "db", dbName);
+      }
+    }
+
+    private void reportQueryMetric(MetaQueryMetric metric) {
+      final String name = metric.name();
+      if (name.startsWith("orm")) {
+        reportMetric(metric, "db.query", "query", name, "db", dbName, "type", "orm");
+      } else if (name.startsWith("sql")) {
+        reportMetric(metric, "db.query", "query", name, "db", dbName, "type", "orm");
+      } else {
+        reportMetric(metric, "db.query", "query", name, "db", dbName, "type", "other");
+      }
+    }
+
+    private void reportMetric(MetaTimedMetric metric, String name, String... tags) {
+      reporter.countWithTimestamp(nm(name, ".count"), metric.count(), epochSecs, tags);
+      reporter.gaugeWithTimestamp(nm(name, ".max"), metric.max(), epochSecs, tags);
+      reporter.gaugeWithTimestamp(nm(name, ".mean"), metric.mean(), epochSecs, tags);
+      reporter.gaugeWithTimestamp(nm(name, ".total"), metric.total(), epochSecs, tags);
+    }
+  }
+
+}

metrics-statsd/src/main/java/io/avaje/metrics/statsd/Reporter.java

@@ -0,0 +1,103 @@
+package io.avaje.metrics.statsd;
+
+import com.timgroup.statsd.StatsDClient;
+import io.avaje.metrics.*;
+import org.jspecify.annotations.NullMarked;
+
+import java.util.List;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicBoolean;
+
+@NullMarked
+final class Reporter implements Runnable, AutoCloseable, StatsdReporter {
+
+  private final MetricRegistry registry;
+  private final StatsDClient client;
+  private final long timedThreshold;
+  private final List<StatsdReporter.Reporter> reporters;
+  private final ScheduledTask scheduledTask;
+  private final AtomicBoolean started = new AtomicBoolean(false);
+
+  Reporter(MetricRegistry registry, StatsDClient client, long timedThreshold, int schedule,
+           TimeUnit scheduleTimeUnit, List<StatsdReporter.Reporter> reporters) {
+    this.registry = registry;
+    this.client = client;
+    this.timedThreshold = timedThreshold;
+    this.reporters = reporters;
+    this.scheduledTask = ScheduledTask.builder()
+      .schedule(schedule, schedule, scheduleTimeUnit)
+      .task(this)
+      .build();
+  }
+
+  @Override
+  public void start() {
+    scheduledTask.start();
+    started.set(true);
+  }
+
+  @Override
+  public void close() {
+    if (started.get()) {
+      scheduledTask.cancel(true);
+    }
+  }
+
+  @Override
+  public void run() {
+    var visitor = new AvajeMetricsVisitor();
+    for (Metric.Statistics metric : registry.collectMetrics()) {
+      metric.visit(visitor);
+    }
+    for (StatsdReporter.Reporter reporter : reporters) {
+      reporter.report(client);
+    }
+  }
+
+  private final class AvajeMetricsVisitor implements Metric.Visitor {
+
+    private final long epochSecs = System.currentTimeMillis() / 1000;
+
+    private void sendValues(Meter.Stats stats, String name, String... tags) {
+      if (stats.count() > 0) {
+        client.countWithTimestamp(name + ".count", stats.count(), epochSecs, tags);
+        client.gaugeWithTimestamp(name + ".total", stats.total(), epochSecs, tags);
+        client.gaugeWithTimestamp(name + ".mean", stats.mean(), epochSecs, tags);
+        client.gaugeWithTimestamp(name + ".max", stats.max(), epochSecs, tags);
+      }
+    }
+
+    @Override
+    public void visit(Timer.Stats timed) {
+      if (timedThreshold == 0 || timedThreshold < timed.total()) {
+        if (timed.name().startsWith("web.api")) {
+          sendValues(timed, "web.api", "name", timed.name());
+        } else if (timed.name().startsWith("app.")) {
+          sendValues(timed, "app.method", "name", timed.name());
+        } else {
+          sendValues(timed, timed.name());
+        }
+      }
+    }
+
+    @Override
+    public void visit(Meter.Stats stats) {
+      sendValues(stats, stats.name());
+    }
+
+    @Override
+    public void visit(Counter.Stats counter) {
+      client.countWithTimestamp(counter.name(), counter.count(), epochSecs);
+    }
+
+    @Override
+    public void visit(GaugeDouble.Stats gauge) {
+      client.gaugeWithTimestamp(gauge.name(), gauge.value(), epochSecs);
+    }
+
+    @Override
+    public void visit(GaugeLong.Stats gauge) {
+      client.gaugeWithTimestamp(gauge.name(), gauge.value(), epochSecs);
+    }
+  }
+}

metrics-statsd/src/main/java/io/avaje/metrics/statsd/StatsdBuilder.java

@@ -0,0 +1,98 @@
+package io.avaje.metrics.statsd;
+
+import com.timgroup.statsd.NonBlockingStatsDClient;
+import com.timgroup.statsd.NonBlockingStatsDClientBuilder;
+import com.timgroup.statsd.StatsDClient;
+import io.avaje.metrics.MetricRegistry;
+import io.avaje.metrics.Metrics;
+import io.ebean.Database;
+
+import java.util.ArrayList;
+import java.util.List;
+import java.util.concurrent.TimeUnit;
+
+import static java.util.Objects.requireNonNull;
+
+final class StatsdBuilder implements StatsdReporter.Builder {
+
+  private final List<StatsdReporter.Reporter> reporters = new ArrayList<>();
+  private MetricRegistry registry;
+  private String hostname = "localhost";
+  private int port = NonBlockingStatsDClient.DEFAULT_DOGSTATSD_PORT;
+  private StatsDClient client;
+  private long timedThresholdMicros;
+  private String[] tags;
+  private int schedule = 60;
+  private TimeUnit scheduleTimeUnit = TimeUnit.SECONDS;
+
+  @Override
+  public StatsdReporter.Builder hostname(String hostname) {
+    this.hostname = requireNonNull(hostname);
+    return this;
+  }
+
+  @Override
+  public StatsdReporter.Builder port(int port) {
+    this.port = port;
+    return this;
+  }
+
+  @Override
+  public StatsdReporter.Builder client(StatsDClient client) {
+    this.client = client;
+    return this;
+  }
+
+  @Override
+  public StatsdReporter.Builder timedThresholdMicros(long timedThresholdMicros) {
+    this.timedThresholdMicros = timedThresholdMicros;
+    return this;
+  }
+
+  @Override
+  public StatsdReporter.Builder tags(String[] tags) {
+    this.tags = tags;
+    return this;
+  }
+
+  @Override
+  public StatsdReporter.Builder schedule(int schedule, TimeUnit timeUnit) {
+    this.schedule = schedule;
+    this.scheduleTimeUnit = requireNonNull(timeUnit);
+    return this;
+  }
+
+  @Override
+  public StatsdReporter.Builder registry(MetricRegistry registry) {
+    this.registry = requireNonNull(registry);
+    return this;
+  }
+
+  @Override
+  public StatsdReporter.Builder database(Database database) {
+    reporters.add(DatabaseReporter.reporter(database));
+    return this;
+  }
+
+  @Override
+  public StatsdReporter.Builder reporter(StatsdReporter.Reporter reporter) {
+    reporters.add(requireNonNull(reporter));
+    return this;
+  }
+
+  @Override
+  public Reporter build() {
+    if (registry == null) {
+      registry = Metrics.registry();
+    }
+    if (client == null) {
+      client = new NonBlockingStatsDClientBuilder()
+        .hostname(requireNonNull(hostname))
+        .port(port)
+        .constantTags(tags)
+        .build();
+    }
+
+    return new Reporter(registry, client, timedThresholdMicros, schedule, scheduleTimeUnit, reporters);
+  }
+}