Class S3AuditSink

java.lang.Object
com.mcpdbwizard.pub.S3AuditSink
All Implemented Interfaces:
McpAuditSink
public class S3AuditSink extends Object implements McpAuditSink
Writes the audit trail to S3 as rolling JSONL objects.

Select it with MCP_AUDIT_SINK=com.mcpdbwizard.pub.S3AuditSink.

Environment
MCP_AUDIT_S3_BUCKETRequired.
MCP_AUDIT_S3_PREFIXKey prefix, default "mcp-audit".
MCP_AUDIT_S3_REGIONRegion. The SDK's own resolution if unset.
MCP_AUDIT_S3_ENDPOINTOverride, for MinIO or a VPC endpoint.
MCP_AUDIT_S3_ROLL_BYTESRoll an object at this size, default 8388608L.
MCP_AUDIT_S3_ROLL_SECONDS...or after this long, default 300L.

An object store is not a stream, so this batches

A PUT per tool call would be slow and, at S3's per-request pricing, absurd. Records accumulate and become one object when it is big enough or old enough.

Keys are date-partitioned — <prefix>/<config>/yyyy/MM/dd/<epochMillis>-<uuid>.jsonl — for two concrete reasons rather than tidiness. A lifecycle rule can expire a prefix, which is how retention is done on the S3 side; and Athena or Glue can partition on the date without reading the objects.

The roll timer is not optional

Rolling only when a record arrives would mean a quiet server holds its last few records in memory indefinitely, and loses them if the process stops. A daemon thread rolls on schedule so the window in which a record exists only in memory is bounded by MCP_AUDIT_S3_ROLL_SECONDS rather than by how busy the server happens to be.

Records in that window are in memory only. SpoolingAuditSink in front is what closes it: it writes to disk before this ever sees a record, and deletes only once flush() confirms the PUT. Copyright 2003-2026 ATB Consultancy Services Ltd (formerly Orinda Software Ltd, Dublin, Ireland)

  • Field Details

  • Constructor Details

    • S3AuditSink

      public S3AuditSink()
      Built reflectively by McpAuditSinks.fromEnvironment().

    • S3AuditSink

      public S3AuditSink(software.amazon.awssdk.services.s3.S3Client theClientValue, String theBucketValue, String thePrefixValue, String theConfigValue, long theRollBytesValue, long theRollSecondsValue)
      For tests: inject a client rather than reach AWS. The roll thread is not started.

  • Method Details

    • startRolling

      public void startRolling()
      Start the scheduled roll. Separate from the constructor so a test can roll deterministically.

    • record

      public void record(McpAuditEvent theEvent)
      Record one call. Must not throw.

      Never throws.

      Specified by:
      record in interface McpAuditSink
      Parameters:
      theEvent - the call to record; never null

    • rollIfDue

      public void rollIfDue()
      Roll if the pending batch is old enough. Called by the timer, and testable on its own.

    • flush

      public boolean flush()
      Confirm that everything handed to McpAuditSink.record(com.mcpdbwizard.pub.McpAuditEvent) since the last flush is durably accepted.

      This is what makes a spool possible. A spool writes each record to disk first and may only delete it once the sink has genuinely taken it — and for an asynchronous sink like Kafka, record returning tells you nothing, because the send has not completed yet.

      The default returns true, which is correct for a sink that delivers synchronously inside record. An asynchronous sink must override it, and must return false if anything since the last flush was lost — returning true on a failed batch would let a spool delete records that never arrived, which is the one way a spool can be worse than no spool.

      Forces a roll, then reports whether anything has been lost since the previous flush. Compared against the count at the END OF THE PREVIOUS FLUSH — the rule KafkaAuditSink.flush() sets out, and it applies here for the same reason: a batch can be rejected inside record(com.mcpdbwizard.pub.McpAuditEvent) when it fills, before any flush begins.

      Specified by:
      flush in interface McpAuditSink
      Returns:
      true if everything since the last flush is safely delivered

    • getDroppedCount

      public long getDroppedCount()
      Description copied from interface: McpAuditSink
      How many records this sink is known to have LOST, or -1 when it does not report.

      The number an operator actually needs, and the one nothing surfaced before: a trail is only worth citing if it is complete, and "complete" is exactly what a drop count denies. Reported through the SPI rather than by casting to a particular sink so a status page keeps working when the sink is swapped.

      -1 means "this sink does not count", which a caller must show differently from 0. Claiming zero losses on a sink that cannot tell is the one wrong answer here.

      Specified by:
      getDroppedCount in interface McpAuditSink

    • getDeliveredCount

      public long getDeliveredCount()
      Description copied from interface: McpAuditSink
      How many records were confirmed delivered, or -1 when the sink does not report.
      Specified by:
      getDeliveredCount in interface McpAuditSink

    • getPendingCount

      public long getPendingCount()
      Description copied from interface: McpAuditSink
      How many records are written down but not yet delivered, or -1 when not reported.
      Specified by:
      getPendingCount in interface McpAuditSink

    • describe

      public String describe()
      Description copied from interface: McpAuditSink
      A short human description of what this sink is, for an operator reading a status page.

      Defaults to the class name. A wrapper overrides it to name what it wraps — otherwise a spooled Kafka sink reports only "SpoolingAuditSink", and where the records finally go is the part being asked about.

      Specified by:
      describe in interface McpAuditSink

    • close

      public void close()
      Roll what is left, then release the client.
      Specified by:
      close in interface McpAuditSink