Write egress access stats to Data Acess V3

docs/PER_SHARE_EGRESS_MONITORING.md

@@ -9,6 +9,7 @@ When enabled, each data access emits JSON logs with:
 - Total egress bytes
 - Pricing tier (e.g., `internet_to_na_eu`, `interregion_na_to_eu`)
 - Client region code
+- Audit fields for customer audits (clientIp, rawRegionHeader, isGcpIp, tenantId)
 
 ---
 
@@ -125,6 +126,12 @@ accessLogging:
   detectGcpTraffic: true                # Enable GCP IP range lookup
   clientRegionHeader: "x-client-region" # Header with country code
   clientIpHeader: "x-forwarded-for"     # Header with client IP chain
+  # Base path for the consolidated access log Delta table.
+  # All access logs are written to a single table: access_log_br__system
+  # Omit to disable Delta writing.
+  deltaTablePath: "gs://<bucket>/datalake/data/tenant/_system"
+  deltaFlushIntervalSeconds: 60         # Max seconds between Delta flushes (default: 60)
+  deltaFlushBatchSize: 1000             # Records per flush before early trigger (default: 1000)
 ```
 
 ### GCP Load Balancer Headers
@@ -137,6 +144,73 @@ X-Client-Region-Subdivision: {client_region_subdivision}
 
 ---
 
+## Delta Lake Storage
+
+When `deltaTablePath` is configured, `ACCESS_LOG` entries are written asynchronously to
+a consolidated Delta table on GCS in addition to the JSON log stream. This enables durable storage and
+SQL-queryable access via Delta Sharing.
+
+### Consolidated Table
+
+All access logs are written to a single consolidated table `access_log_br__system`. The `tenantId`
+field is included in each record for filtering by tenant. This simplifies cross-tenant queries
+and consolidates all egress data in one location.
+
+**Note:** `access_log_br__system` has double underscore in it's name (one as a separator and one from _system tenant's name).
+
+**IMPORTANT:** The Delta table must be pre-created by the `deltalake-admin` tool during tenant
+onboarding. The Delta Sharing server does not auto-create the table schema. 
+
+| Property | Value |
+|----------|-------|
+| **Table Name** | `access_log_br__system` |
+| **GCS Path** | `{deltaTablePath}/access_log_br__system` |
+| **Partitioning** | None (unpartitioned for simplified queries) |
+| **Format** | Parquet + Delta transaction log |
+| **Pre-requisite** | Table must be created by `deltalake-admin` before first write |
+
+### Schema
+
+| Field | Type | Description |
+|-------|------|-------------|
+| `logType` | string | Always "ACCESS_LOG" |
+| `share` | string | Share name |
+| `schema` | string | Schema name |
+| `table` | string | Table name |
+| `egressBytes` | long | Bytes transferred |
+| `pricingTier` | string | GCP pricing tier (see Pricing Tiers) |
+| `timestampMs` | long | Timestamp in milliseconds |
+| `requestType` | string | "query" or "cdf_stream" |
+| `clientRegion` | string | ISO 3166-1 alpha-2 country code |
+| `tenantId` | string | Tenant identifier (derived from share name) |
+| `clientIp` | string | Client IP address for audit |
+| `rawRegionHeader` | string | Raw region header value for audit |
+| `isGcpIp` | boolean | Whether client IP is in GCP ranges |
+
+### GCS Base Paths by Environment
+
+| Environment | Table Path |
+|-------------|------------|
+| zing-dev | `gs://zing-dev-197522-dl-v1/datalake/data/tenant/_system/access_log_br__system` |
+| zing-preview | `gs://zing-preview-dl-v1/datalake/data/tenant/_system/access_log_br__system` |
+| zcloud-prod | `gs://zcloud-prod-dl-v1/datalake/data/tenant/_system/access_log_br__system` |
+| zcloud-prod2 | `gs://zcloud-prod2-dl-v1/datalake/data/tenant/_system/access_log_br__system` |
+| zcloud-prod3 | `gs://zcloud-prod3-dl-v1/datalake/data/tenant/_system/access_log_br__system` |
+
+### Delta Write Behavior
+
+- Records are buffered in a bounded in-memory queue (capacity: 100,000) and written by a
+  background daemon thread — the request path is never blocked.
+- A flush is triggered when `deltaFlushBatchSize` records accumulate or `deltaFlushIntervalSeconds`
+  elapses, whichever comes first.
+- Queue overflow silently drops records (logged as a warning); write errors are logged to stderr
+  and never propagate to clients.
+- On graceful shutdown the writer makes a best-effort attempt to drain the queue (up to ~30s) before the process exits.
+- Only `ACCESS_LOG` entries are written to Delta. `PRICING_CONTEXT` and `REQUEST_HEADERS`
+  entries are log-only.
+
+---
+
 ## Log Output
 
 **ACCESS_LOG** — Emitted for each request with non-zero egress:
@@ -150,7 +224,11 @@ X-Client-Region-Subdivision: {client_region_subdivision}
   "pricingTier": "internet_to_na_eu",
   "timestampMs": 1717502400000,
   "requestType": "query",
-  "clientRegion": "US"
+  "clientRegion": "US",
+  "tenantId": "my_tenant",
+  "clientIp": "203.0.113.45",
+  "rawRegionHeader": "US",
+  "isGcpIp": false
 }
 ```
 
@@ -191,8 +269,9 @@ X-Client-Region-Subdivision: {client_region_subdivision}
 |------|---------|
 | `GcpPricingTier.scala` | Continent mapping, egress type detection, pricing calculation |
 | `GcpIpRangeLookup.scala` | GCP IP range fetching and CIDR trie lookup |
-| `AccessLogEmitter.scala` | Log entry models and JSON emission |
-| `DeltaSharingService.scala` | Integration for query/CDF endpoints |
+| `AccessLogEmitter.scala` | Log entry models with audit fields, JSON emission, composite fan-out |
+| `DeltaAccessLogWriter.scala` | Async Delta Lake writer to consolidated `access_log_br__system` table |
+| `DeltaSharingService.scala` | Integration for query/CDF endpoints, tenant ID extraction |
 | `ServerConfig.scala` | `AccessLoggingConfig` model |
 
 ### Egress Bytes Calculation
@@ -203,7 +282,10 @@ Sum of `size` from all file actions: `AddFile`, `AddFileForCDF`, `AddCDCFile`.
 
 ## Notes
 
-- Logs emitted to `delta.sharing.access` logger
-- Zero-byte requests not logged
+- JSON logs emitted to `delta.sharing.access` logger
+- Zero-byte requests not logged (neither to JSON nor Delta)
 - GCP IP ranges refreshed every 24 hours
 - Pricing tiers match GCP documentation
+- Delta writes are additive; disabling `deltaTablePath` has no effect on JSON log output
+- All tenant access logs are consolidated in a single `access_log_br__system` table
+- Audit fields (clientIp, rawRegionHeader, isGcpIp, tenantId) support customer audit requirements

manifests/base/configmap.yaml

@@ -32,3 +32,4 @@ data:
       detectGcpTraffic: true
       clientRegionHeader: "x-client-region"
       clientIpHeader: "x-forwarded-for"
+      deltaTablePath: "gs://zing-dev-197522-dl-v1/datalake/data/tenant/_system"

manifests/zcloud-prod/configmap.yaml

@@ -27,3 +27,4 @@ data:
       detectGcpTraffic: true
       clientRegionHeader: "x-client-region"
       clientIpHeader: "x-forwarded-for"
+      deltaTablePath: "gs://zcloud-prod-dl-v1/datalake/data/tenant/_system"

manifests/zcloud-prod2/configmap.yaml

@@ -27,3 +27,4 @@ data:
       detectGcpTraffic: true
       clientRegionHeader: "x-client-region"
       clientIpHeader: "x-forwarded-for"
+      deltaTablePath: "gs://zcloud-prod2-dl-v1/datalake/data/tenant/_system"

manifests/zcloud-prod3/configmap.yaml

@@ -27,3 +27,4 @@ data:
       detectGcpTraffic: true
       clientRegionHeader: "x-client-region"
       clientIpHeader: "x-forwarded-for"
+      deltaTablePath: "gs://zcloud-prod3-dl-v1/datalake/data/tenant/_system"

manifests/zing-preview/configmap.yaml

@@ -27,3 +27,4 @@ data:
       detectGcpTraffic: true
       clientRegionHeader: "x-client-region"
       clientIpHeader: "x-forwarded-for"
+      deltaTablePath: "gs://zing-preview-dl-v1/datalake/data/tenant/_system"

scalastyle-config.xml

@@ -92,7 +92,7 @@ This file is divided into 3 sections:
  */
 
 \E)?\Q/*
- * Copyright (2021) The Delta Lake Project Authors.
+ * Copyright (\E20\d{2}\Q) The Delta Lake Project Authors.
  *
  * Licensed under the Apache License, Version 2.0 (the "License");
  * you may not use this file except in compliance with the License.

server/src/main/scala/io/delta/sharing/server/DeltaSharingService.scala

@@ -207,6 +207,18 @@ class DeltaSharingService(serverConfig: ServerConfig) {
       .toMap
   }
 
+  /**
+   * Extracts tenant_id from share name following the pattern `{tenant_id}_share`.
+   * Falls back to the share name itself if the pattern doesn't match.
+   */
+  private def extractTenantId(shareName: String): String = {
+    if (shareName.endsWith("_share")) {
+      shareName.dropRight("_share".length)
+    } else {
+      shareName
+    }
+  }
+
   private def emitQueryEgressMetric(
       req: HttpRequest,
       share: String,
@@ -228,7 +240,7 @@ class DeltaSharingService(serverConfig: ServerConfig) {
       headers,
       serverConfig.getAccessLogging)
 
-    // Emit access log with essential fields
+    // Emit access log with essential fields and audit fields
     val entry = AccessLogEntry(
       share = share,
       schema = schema,
@@ -237,7 +249,11 @@ class DeltaSharingService(serverConfig: ServerConfig) {
       timestampMs = nowMs,
       pricingTier = pricingCtx.location.pricingTier,
       clientRegion = pricingCtx.location.clientRegion,
-      requestType = AccessLogEmitter.QueryRequestType
+      requestType = AccessLogEmitter.QueryRequestType,
+      tenantId = Some(extractTenantId(share)),
+      clientIp = pricingCtx.clientIp,
+      rawRegionHeader = pricingCtx.rawRegionHeader,
+      isGcpIp = Some(pricingCtx.isGcpIp)
     )
     accessLogEmitter.record(entry)
 
@@ -292,7 +308,7 @@ class DeltaSharingService(serverConfig: ServerConfig) {
       headers,
       serverConfig.getAccessLogging)
 
-    // Emit access log with essential fields
+    // Emit access log with essential fields and audit fields
     val entry = AccessLogEntry(
       share = share,
       schema = schema,
@@ -301,7 +317,11 @@ class DeltaSharingService(serverConfig: ServerConfig) {
       timestampMs = nowMs,
       pricingTier = pricingCtx.location.pricingTier,
       clientRegion = pricingCtx.location.clientRegion,
-      requestType = AccessLogEmitter.CdfStreamRequestType
+      requestType = AccessLogEmitter.CdfStreamRequestType,
+      tenantId = Some(extractTenantId(share)),
+      clientIp = pricingCtx.clientIp,
+      rawRegionHeader = pricingCtx.rawRegionHeader,
+      isGcpIp = Some(pricingCtx.isGcpIp)
     )
     accessLogEmitter.record(entry)
 
@@ -1220,14 +1240,20 @@ object DeltaSharingService {
   }
 
   def start(serverConfig: ServerConfig): Server = {
+    val service = new DeltaSharingService(serverConfig)
+    // scalastyle:off runtimeaddshutdownhook
+    Runtime.getRuntime.addShutdownHook(new Thread(
+      () => service.accessLogEmitter.close(),
+      "delta-access-log-shutdown"))
+    // scalastyle:on runtimeaddshutdownhook
     lazy val server = {
       updateDefaultJsonPrinterForScalaPbConverterUtil()
       val builder = Server.builder()
         .defaultHostname(serverConfig.getHost)
         .disableDateHeader()
         .disableServerHeader()
         .requestTimeout(java.time.Duration.ofSeconds(serverConfig.requestTimeoutSeconds))
-        .annotatedService(serverConfig.endpoint, new DeltaSharingService(serverConfig): Any)
+        .annotatedService(serverConfig.endpoint, service: Any)
       if (serverConfig.ssl == null) {
         builder.http(serverConfig.getPort)
       } else {

server/src/main/scala/io/delta/sharing/server/config/ServerConfig.scala

@@ -152,7 +152,17 @@ case class AccessLoggingConfig(
     // Enable GCP traffic detection using GCP's published IP ranges (cloud.json).
     // When true, client IPs belonging to other GCP regions can be classified as inter-region
     // (cheaper) rather than internet egress. Set to false to disable this detection.
-    @BeanProperty var detectGcpTraffic: Boolean) extends ConfigItem {
+    @BeanProperty var detectGcpTraffic: Boolean,
+    // GCS base path for the consolidated access log Delta table. When set, ACCESS_LOG
+    // entries are written asynchronously to `{deltaTablePath}/access_log_br__system`
+    // in addition to JSON logs. The Delta table must be pre-created by deltalake-admin;
+    // the server does not auto-create schema. Leave null or empty to disable Delta writing.
+    // Example: gs://my-bucket/datalake/data/tenant/_system
+    @BeanProperty var deltaTablePath: String,
+    // How often (seconds) to flush buffered access log records to the Delta table.
+    @BeanProperty var deltaFlushIntervalSeconds: Int,
+    // Maximum number of records to buffer before triggering an early flush.
+    @BeanProperty var deltaFlushBatchSize: Int) extends ConfigItem {
 
   def this() = {
     this(
@@ -161,7 +171,10 @@ case class AccessLoggingConfig(
       clientIpHeader = "x-forwarded-for",
       pricingGroups = Collections.emptyMap(),
       sourceRegion = "",
-      detectGcpTraffic = true)
+      detectGcpTraffic = true,
+      deltaTablePath = null,
+      deltaFlushIntervalSeconds = 60,
+      deltaFlushBatchSize = 1000)
   }
 
   override def checkConfig(): Unit = {

server/src/main/scala/io/delta/sharing/server/telemetry/AccessLogEmitter.scala

@@ -1,5 +1,5 @@
 /*
- * Copyright (2021) The Delta Lake Project Authors.
+ * Copyright (2026) The Delta Lake Project Authors.
  *
  * Licensed under the Apache License, Version 2.0 (the "License");
  * you may not use this file except in compliance with the License.
@@ -62,6 +62,12 @@ import io.delta.sharing.server.config.ServerConfig
  * == Optional Context ==
  * @param clientRegion ISO 3166-1 alpha-2 country code of the client (e.g., "US", "MT")
  * @param requestType Type of request: "query" (snapshot read) or "cdf_stream" (CDF streaming)
+ *
+ * == Audit Fields (for customer audits and consolidated storage) ==
+ * @param tenantId Tenant identifier extracted from share name (for consolidated table storage)
+ * @param clientIp Client IP address from request headers (for audit)
+ * @param rawRegionHeader Raw region header value before normalization (for audit)
+ * @param isGcpIp Whether the client IP is in a known GCP public IP range (for audit)
  */
 case class AccessLogEntry(
     share: String,
@@ -71,7 +77,11 @@ case class AccessLogEntry(
     timestampMs: Long,
     pricingTier: String = "unknown",
     clientRegion: Option[String] = None,
-    requestType: String = "query")
+    requestType: String = "query",
+    tenantId: Option[String] = None,
+    clientIp: Option[String] = None,
+    rawRegionHeader: Option[String] = None,
+    isGcpIp: Option[Boolean] = None)
 
 /**
  * Captures all context information used to calculate the pricing tier.
@@ -139,6 +149,7 @@ trait AccessLogEmitter {
   def record(entry: AccessLogEntry): Unit
   def recordContext(entry: PricingContextLogEntry): Unit
   def recordHeaders(entry: RequestHeadersLogEntry): Unit
+  def close(): Unit = {}
 }
 
 object AccessLogEmitter {
@@ -148,11 +159,24 @@ object AccessLogEmitter {
   /**
    * Creates an AccessLogEmitter based on server configuration.
    * Returns a JsonAccessLogEmitter if access logging is enabled, otherwise a no-op emitter.
+   * When deltaTablePath is also set, returns a CompositeAccessLogEmitter that fans out to
+   * both the JSON log stream and the Delta table writer.
    */
   def create(serverConfig: ServerConfig): AccessLogEmitter = {
     val cfg = Option(serverConfig.getAccessLogging)
     cfg match {
-      case Some(c) if c.enabled => new JsonAccessLogEmitter()
+      case Some(c) if c.enabled =>
+        val jsonEmitter = new JsonAccessLogEmitter()
+        val deltaPath = Option(c.getDeltaTablePath).map(_.trim).filter(_.nonEmpty)
+        deltaPath match {
+          case Some(path) =>
+            val deltaWriter = new DeltaAccessLogWriter(
+              path,
+              c.getDeltaFlushIntervalSeconds,
+              c.getDeltaFlushBatchSize)
+            new CompositeAccessLogEmitter(Seq(jsonEmitter, deltaWriter))
+          case None => jsonEmitter
+        }
       case _ => NoopAccessLogEmitter
     }
   }
@@ -167,6 +191,21 @@ object NoopAccessLogEmitter extends AccessLogEmitter {
   override def recordHeaders(entry: RequestHeadersLogEntry): Unit = {}
 }
 
+/**
+ * Fans out all emitter calls to a sequence of delegate emitters.
+ * Used to write to both JSON logs and the Delta table simultaneously.
+ */
+class CompositeAccessLogEmitter(emitters: Seq[AccessLogEmitter]) extends AccessLogEmitter {
+  override def record(entry: AccessLogEntry): Unit =
+    emitters.foreach(_.record(entry))
+  override def recordContext(entry: PricingContextLogEntry): Unit =
+    emitters.foreach(_.recordContext(entry))
+  override def recordHeaders(entry: RequestHeadersLogEntry): Unit =
+    emitters.foreach(_.recordHeaders(entry))
+  override def close(): Unit =
+    emitters.foreach(_.close())
+}
+
 /**
  * Emits access log entries as JSON-structured log lines.
  * Uses a dedicated logger that can be filtered/routed separately in Cloud Logging.
@@ -192,9 +231,13 @@ class JsonAccessLogEmitter extends AccessLogEmitter {
       "requestType" -> entry.requestType
     )
 
-    // Optional context fields
+    // Optional context fields (including audit fields)
     val contextPayload = Seq(
-      entry.clientRegion.map("clientRegion" -> _)
+      entry.clientRegion.map("clientRegion" -> _),
+      entry.tenantId.map("tenantId" -> _),
+      entry.clientIp.map("clientIp" -> _),
+      entry.rawRegionHeader.map("rawRegionHeader" -> _),
+      entry.isGcpIp.map("isGcpIp" -> _)
     ).flatten.toMap
 
     val logPayload = basePayload ++ contextPayload