Sends billing.events log lines to S3

pom.xml

@@ -203,6 +203,11 @@
       <groupId>software.amazon.awssdk</groupId>
       <artifactId>sts</artifactId>
     </dependency>
+    <!-- S3 export of billing.events log lines. -->
+    <dependency>
+      <groupId>software.amazon.awssdk</groupId>
+      <artifactId>s3</artifactId>
+    </dependency>
     <dependency>
       <groupId>com.datastax.oss</groupId>
       <artifactId>java-driver-core</artifactId>
@@ -300,6 +305,13 @@
       <artifactId>junit-jupiter</artifactId>
       <scope>test</scope>
     </dependency>
+    <!-- S3Mock for the billing S3 export integration test. -->
+    <dependency>
+      <groupId>com.adobe.testing</groupId>
+      <artifactId>s3mock-testcontainers</artifactId>
+      <version>5.1.0</version>
+      <scope>test</scope>
+    </dependency>
     <dependency>
       <groupId>com.github.docker-java</groupId>
       <artifactId>docker-java-api</artifactId>

src/main/java/io/stargate/sgv2/jsonapi/api/request/RequestContext.java

@@ -15,7 +15,7 @@
 import io.stargate.sgv2.jsonapi.config.feature.ApiFeatures;
 import io.stargate.sgv2.jsonapi.config.feature.FeaturesConfig;
 import io.stargate.sgv2.jsonapi.logging.LoggingMDCContext;
-import io.stargate.sgv2.jsonapi.service.provider.Billing;
+import io.stargate.sgv2.jsonapi.service.billing.Billing;
 import io.stargate.sgv2.jsonapi.service.schema.SchemaRegistry;
 import io.vertx.ext.web.RoutingContext;
 import jakarta.enterprise.context.RequestScoped;

src/main/java/io/stargate/sgv2/jsonapi/config/BillingConfig.java

@@ -2,7 +2,7 @@
 
 import io.smallrye.config.ConfigMapping;
 import io.smallrye.config.WithDefault;
-import io.stargate.sgv2.jsonapi.service.provider.BillingEventType;
+import io.stargate.sgv2.jsonapi.service.billing.BillingEventType;
 import java.util.List;
 import java.util.Optional;
 import java.util.Set;

src/main/java/io/stargate/sgv2/jsonapi/config/BillingS3ExportConfig.java

@@ -0,0 +1,48 @@
+package io.stargate.sgv2.jsonapi.config;
+
+import io.smallrye.config.ConfigMapping;
+import io.smallrye.config.WithDefault;
+import java.time.Duration;
+import java.util.Optional;
+
+/** Configuration for the billing S3 export (see BillingS3HandlerInstaller). */
+@ConfigMapping(prefix = "stargate.jsonapi.billing.s3")
+public interface BillingS3ExportConfig {
+
+  /** when false the export handler is never installed. */
+  @WithDefault("false")
+  boolean enabled();
+
+  /** S3 bucket region */
+  @WithDefault("us-east-2")
+  String region();
+
+  /** S3 bucket name */
+  @WithDefault("serverless-usage-dev")
+  String bucket();
+
+  /**
+   * Only for non-AWS S3 endpoints (e.g. S3Mock in tests). TODO: XXX EXPLAIN WHAT THIS SHOULD SET
+   * SET TO
+   */
+  Optional<String> endpointOverride();
+
+  /** */
+  @WithDefault("2048")
+  int maxEventsPerBatch();
+
+  /**
+   * Max bytes to include in a batch, NOTE: if a single event is bigger than this it will be sent in
+   * a batch still. 2097152 == 2 MB
+   */
+  @WithDefault("2097152")
+  long maxBytesPerBatch();
+
+  /** Age flush period: buffered events are shipped at least this often */
+  @WithDefault("PT30S")
+  Duration maxAge();
+
+  /** Bound on buffered events; beyond it new lines are dropped. */
+  @WithDefault("10000")
+  int queueCapacity();
+}

src/main/java/io/stargate/sgv2/jsonapi/metrics/BatchedLogBufferMetrics.java

@@ -0,0 +1,52 @@
+package io.stargate.sgv2.jsonapi.metrics;
+
+import io.micrometer.core.instrument.Counter;
+import io.micrometer.core.instrument.MeterRegistry;
+import io.stargate.sgv2.jsonapi.service.billing.BatchedLogBuffer;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicBoolean;
+
+/**
+ * Metrics for the billing buffer in {@link
+ * io.stargate.sgv2.jsonapi.service.billing.BatchedLogBuffer}
+ */
+public final class BatchedLogBufferMetrics extends MetricsBase {
+
+  private final AtomicBoolean bufferRegister = new AtomicBoolean(false);
+
+  private final Counter offered;
+  private final Counter dropped;
+
+  /** */
+  public BatchedLogBufferMetrics(MeterRegistry meterRegistry, String prefix) {
+    super(meterRegistry, prefix);
+
+    this.offered = newCounter("buffer.offered");
+    this.dropped = newCounter("buffer.dropped");
+
+    // Note: not recording events dropped at shutdown as a metric because when shutting down
+    // the metrics still need to be scrapped to be useful. Do it as a log message that is
+    // persistent.
+  }
+
+  public void registerBuffer(BatchedLogBuffer buffer) {
+
+    if (!bufferRegister.compareAndSet(false, true)) {
+      throw new IllegalStateException("registerBuffer() already called");
+    }
+
+    newTimeGauge(
+        "buffer.head_age_ms", () -> buffer.headEntryAge().toMillis(), TimeUnit.MILLISECONDS);
+    newGauge("buffer.size", buffer::size);
+    newGauge("buffer.remaining_capacity", buffer::remainingCapacity);
+    newGauge("buffer.bytes", buffer::queuedBytes);
+  }
+
+  public void offered() {
+    offered.increment();
+  }
+
+  public void dropped() {
+    dropped.increment();
+  }
+}

src/main/java/io/stargate/sgv2/jsonapi/metrics/BatchedLogUploaderMetrics.java

@@ -0,0 +1,48 @@
+package io.stargate.sgv2.jsonapi.metrics;
+
+import io.micrometer.core.instrument.Counter;
+import io.micrometer.core.instrument.MeterRegistry;
+import io.micrometer.core.instrument.Timer;
+import io.stargate.sgv2.jsonapi.service.billing.BatchedLogBuffer;
+import java.util.Objects;
+
+/** Metrics for billing events, mostly around what is sent to S3 */
+public final class BatchedLogUploaderMetrics extends MetricsBase {
+
+  public final Counter uploadedBatches;
+  public final Counter uploadedBytes;
+  public final Timer uploadedHeadAgeMs;
+  public final Counter uploadedEvents;
+
+  public final Counter failedBatches;
+  public final Counter failedEvents;
+
+  /** */
+  public BatchedLogUploaderMetrics(MeterRegistry meterRegistry, String prefix) {
+    super(meterRegistry, prefix);
+
+    this.uploadedBatches = newCounter("s3.uploaded.batches");
+    this.uploadedBytes = newCounter("s3.uploaded.bytes");
+    this.uploadedEvents = newCounter("s3.uploaded.events");
+    this.uploadedHeadAgeMs = newTimer("s3.uploaded.oldest_event");
+
+    this.failedBatches = newCounter("s3.failed.batches");
+    this.failedEvents = newCounter("s3.failed.events");
+  }
+
+  public void recordBatchDelivered(BatchedLogBuffer.Batch batch) {
+
+    Objects.requireNonNull(batch, "batch must not be null");
+    uploadedBatches.increment();
+    uploadedBytes.increment(batch.bytes());
+    uploadedHeadAgeMs.record(batch.oldestEventAtDuration());
+    uploadedEvents.increment(batch.size());
+  }
+
+  public void recordBatchFailed(BatchedLogBuffer.Batch batch) {
+
+    Objects.requireNonNull(batch, "batch must not be null");
+    failedBatches.increment();
+    failedEvents.increment(batch.size());
+  }
+}

src/main/java/io/stargate/sgv2/jsonapi/metrics/MetricsBase.java

@@ -0,0 +1,61 @@
+package io.stargate.sgv2.jsonapi.metrics;
+
+import io.micrometer.core.instrument.*;
+import java.util.Objects;
+import java.util.concurrent.TimeUnit;
+import java.util.function.Supplier;
+
+/** Common base for classes that create metric measures */
+public abstract class MetricsBase {
+
+  protected final MeterRegistry meterRegistry;
+  protected final String prefix;
+
+  protected MetricsBase(MeterRegistry meterRegistry, String prefix) {
+
+    this.meterRegistry = Objects.requireNonNull(meterRegistry, "meterRegistry must not be null");
+    this.prefix = Objects.requireNonNull(prefix, "prefix must not be null");
+    if (prefix.isBlank()) {
+      throw new IllegalArgumentException("prefix must not be blank");
+    }
+  }
+
+  protected String validateName(String name) {
+    Objects.requireNonNull(name, "name must not be null");
+    if (name.isBlank()) {
+      throw new IllegalArgumentException("name must not be blank");
+    }
+
+    return name.charAt(0) == '.' ? name : "." + name;
+  }
+
+  protected String fullName(String name) {
+    return prefix + validateName(name);
+  }
+
+  protected Counter newCounter(String name) {
+    return meterRegistry.counter(fullName(name));
+  }
+
+  protected Gauge newGauge(String name, Supplier<Number> func) {
+    // no null checks in the builder below
+    Objects.requireNonNull(func, "func must not be null");
+
+    return Gauge.builder(fullName(name), func)
+        .strongReference(true) // is set in builder() above just being explicit
+        .register(meterRegistry);
+  }
+
+  protected Timer newTimer(String name) {
+    return newTimer(name, 0.5, 0.95, 0.99);
+  }
+
+  protected Timer newTimer(String name, double... percentiles) {
+    return Timer.builder(fullName(name)).publishPercentiles(percentiles).register(meterRegistry);
+  }
+
+  protected TimeGauge newTimeGauge(String name, Supplier<Number> func, TimeUnit unit) {
+
+    return TimeGauge.builder(fullName(name), func, unit).register(meterRegistry);
+  }
+}

src/main/java/io/stargate/sgv2/jsonapi/service/billing/AsyncBatchedLogUploader.java

@@ -0,0 +1,28 @@
+package io.stargate.sgv2.jsonapi.service.billing;
+
+import io.smallrye.mutiny.Uni;
+
+/** A function that uploads a batch of log records, normally to S3. */
+@FunctionalInterface
+public interface AsyncBatchedLogUploader extends AutoCloseable {
+
+  /**
+   * Called to upload the batch of records.
+   *
+   * @param batch The batch of log records to upload
+   * @return A Uni of the result of the operation
+   */
+  Uni<UploadResult> upload(BatchedLogBuffer.Batch batch);
+
+  @Override
+  default void close() {}
+
+  /**
+   * Result of the upload call.
+   *
+   * @param success <code>true</code> if the operation succeeded, false otherwise.
+   * @param throwable The throwable associated with an error state.
+   * @param batch The batch that was uploaded, or attempted to be uploaded.
+   */
+  record UploadResult(boolean success, Throwable throwable, BatchedLogBuffer.Batch batch) {}
+}