Address comments

Author: Merteg

README.md

@@ -447,7 +447,9 @@ Fields emitted in each log line:
 - `table`
 - `requestType` (`query` or `cdf_stream`)
 - `egressBytes`
+- `pricingTier`
 - `timestampMs`
+- `clientRegion` (if available)
 
 To enable this feature, configure the `accessLogging` block in the server yaml.
 

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

@@ -213,6 +213,10 @@ class DeltaSharingService(serverConfig: ServerConfig) {
       schema: String,
       table: String,
       actions: Seq[Object]): Unit = {
+    if (!Option(serverConfig.getAccessLogging).exists(_.enabled)) {
+      return
+    }
+
     val bytes = DeltaSharingService.extractEgressBytes(actions)
     if (bytes <= 0) {
       return
@@ -273,6 +277,10 @@ class DeltaSharingService(serverConfig: ServerConfig) {
       schema: String,
       table: String,
       actions: Seq[Object]): Unit = {
+    if (!Option(serverConfig.getAccessLogging).exists(_.enabled)) {
+      return
+    }
+
     val bytes = DeltaSharingService.extractEgressBytes(actions)
     if (bytes <= 0) {
       return
@@ -1120,11 +1128,26 @@ object DeltaSharingService {
   }
 
   private def isValidIp(ip: String): Boolean = {
-    try {
-      InetAddress.getByName(ip)
-      true
-    } catch {
-      case _: Throwable => false
+    // Validate IP format without DNS resolution to avoid latency and security issues
+    val ipv4Pattern = """^(\d{1,3})\.(\d{1,3})\.(\d{1,3})\.(\d{1,3})$""".r
+    val ipv6Pattern = """^([0-9a-fA-F:]+)$""".r
+
+    ip match {
+      case ipv4Pattern(a, b, c, d) =>
+        Seq(a, b, c, d).forall { octet =>
+          val n = octet.toInt
+          n >= 0 && n <= 255
+        }
+      case ipv6Pattern(_) =>
+        try {
+          // For IPv6, use InetAddress but only for format validation
+          // This won't trigger DNS since it matches the IPv6 pattern
+          InetAddress.getByName(ip)
+          true
+        } catch {
+          case _: Throwable => false
+        }
+      case _ => false
     }
   }
 

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

@@ -144,13 +144,14 @@ case class AccessLoggingConfig(
     // Optional mapping from region codes to pricing group labels.
     // Keys should be uppercase location codes (for example: US, DE).
     // A wildcard key "*" can be used as a catch-all default.
+    // NOTE: pricingGroups is reserved for future use and not currently applied.
     @BeanProperty var pricingGroups: java.util.Map[String, String],
     // The GCP region where this server runs (for example: us-central1).
     // Used for pricing tier calculation based on source→destination pairs.
     @BeanProperty var sourceRegion: String,
-    // Enable GCP inter-region traffic detection via X-Envoy-Peer-Metadata header.
-    // When true, traffic from other GCP regions is classified as inter-region (cheaper)
-    // rather than internet egress.
+    // 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 {
 
   def this() = {

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

@@ -16,6 +16,8 @@
 
 package io.delta.sharing.server.telemetry
 
+import java.util.Locale
+
 import org.slf4j.LoggerFactory
 
 import io.delta.sharing.server.common.JsonUtils
@@ -230,13 +232,35 @@ class JsonAccessLogEmitter extends AccessLogEmitter {
     logger.info(JsonUtils.toJson(logPayload))
   }
 
+  // Headers that should be redacted to prevent leaking secrets/PII
+  private val sensitiveHeaders = Set(
+    "authorization",
+    "cookie",
+    "set-cookie",
+    "x-api-key",
+    "x-auth-token",
+    "proxy-authorization",
+    "www-authenticate",
+    "x-csrf-token",
+    "x-xsrf-token"
+  )
+
   override def recordHeaders(entry: RequestHeadersLogEntry): Unit = {
+    // Redact sensitive headers to prevent leaking secrets/PII
+    val redactedHeaders = entry.headers.map { case (key, value) =>
+      if (sensitiveHeaders.contains(key.toLowerCase(Locale.ROOT))) {
+        key -> "[REDACTED]"
+      } else {
+        key -> value
+      }
+    }
+
     val logPayload = Map(
       "logType" -> "REQUEST_HEADERS",
       "share" -> entry.share,
       "table" -> entry.table,
       "timestampMs" -> entry.timestampMs,
-      "headers" -> entry.headers
+      "headers" -> redactedHeaders
     )
 
     logger.info(JsonUtils.toJson(logPayload))

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

@@ -24,6 +24,8 @@ import scala.collection.mutable
 import scala.util.{Failure, Success, Try}
 import scala.util.control.NonFatal
 
+import org.slf4j.LoggerFactory
+
 import io.delta.sharing.server.common.JsonUtils
 
 /**
@@ -37,6 +39,8 @@ import io.delta.sharing.server.common.JsonUtils
  */
 object GcpIpRangeLookup {
 
+  private val logger = LoggerFactory.getLogger("delta.sharing.gcp.ip.lookup")
+
   private val GCP_IP_RANGES_URL = "https://www.gstatic.com/ipranges/cloud.json"
   private val REFRESH_INTERVAL_HOURS = 24
   private val CONNECTION_TIMEOUT_MS = 10000
@@ -230,14 +234,9 @@ object GcpIpRangeLookup {
           cachedTrie.set(newTrie)
           lastRefreshTime.set(System.currentTimeMillis())
           isInitialized.set(true)
-          // scalastyle:off println
-          System.out.println(s"[GcpIpRangeLookup] Loaded ${ranges.prefixes.size} IP ranges " +
-            s"(syncToken: ${ranges.syncToken})")
-          // scalastyle:on println
+          logger.info(s"Loaded ${ranges.prefixes.size} IP ranges (syncToken: ${ranges.syncToken})")
         case Failure(e) =>
-          // scalastyle:off println
-          System.err.println(s"[GcpIpRangeLookup] Failed to refresh IP ranges: ${e.getMessage}")
-          // scalastyle:on println
+          logger.warn(s"Failed to refresh IP ranges: ${e.getMessage}")
           // Keep using the old trie if we have one
           if (!isInitialized.get()) {
             // First-time failure - create empty trie
@@ -247,9 +246,7 @@ object GcpIpRangeLookup {
       }
     } catch {
       case NonFatal(e) =>
-        // scalastyle:off println
-        System.err.println(s"[GcpIpRangeLookup] Unexpected error during refresh: ${e.getMessage}")
-        // scalastyle:on println
+        logger.error(s"Unexpected error during refresh: ${e.getMessage}")
     }
   }
 

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

@@ -355,8 +355,10 @@ object GcpPricingTier {
       return (SAME_REGION, None)
     }
 
-    // If GCP traffic detection is enabled, use IP range lookup for accurate region detection
-    if (detectGcpTraffic) {
+    // If GCP traffic detection is enabled and the IP looks like GCP, use IP range lookup for
+    // accurate region detection. This avoids unnecessary initialization/refresh overhead for
+    // non-GCP internet clients.
+    if (detectGcpTraffic && clientIp.exists(isGcpPublicIp)) {
       clientIp.flatMap(GcpIpRangeLookup.lookupRegion) match {
         case Some(clientGcpRegion) =>
           // We have exact GCP region from IP lookup

server/src/universal/conf/delta-sharing-server.yaml.template

@@ -79,13 +79,12 @@ accessLogging:
   clientIpHeader: "x-forwarded-for"
   # Optional mapping from region codes to billing price groups.
   # Use uppercase keys. A wildcard key "*" can be used as a catch-all.
-  # If empty, server applies a built-in default map (na_eu/apac/latam buckets).
+  # NOTE: pricingGroups is reserved for future use and not currently applied.
   pricingGroups: {}
   # GCP region where this server runs (for example: us-central1).
-  # Required for GCP pricing tier calculation.
   # Used to determine source→destination pairs for egress cost attribution.
   sourceRegion: ""
-  # Enable GCP inter-region traffic detection via X-Envoy-Peer-Metadata header.
-  # When true, traffic from other GCP regions is classified as inter-region (cheaper)
-  # rather than internet egress. Set to false to disable detection.
+  # 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.
   detectGcpTraffic: true