From 71ec50fa54017c466967abd9c18d258f6900e667 Mon Sep 17 00:00:00 2001 From: Andriy Chorny Date: Wed, 12 Aug 2026 14:48:18 +0300 Subject: [PATCH] Init agents context. Add read access logs script --- AGENTS.md | 174 ++++++++ CHANGELOG-VIRTANA.md | 68 +++ Makefile | 7 +- dev/READ_AL.md | 114 ++++++ dev/read_access_logs.py | 387 ++++++++++++++++++ memory-bank/01-overview.md | 78 ++++ memory-bank/02-build-and-test.md | 46 +++ memory-bank/03-virtana-changes.md | 72 ++++ memory-bank/04-deployment.md | 48 +++ memory-bank/05-gcp-egress-pricing.md | 61 +++ .../06-egress-monitoring.md | 0 memory-bank/07-access-log-table-reference.md | 230 +++++++++++ memory-bank/README.md | 19 + 13 files changed, 1301 insertions(+), 3 deletions(-) create mode 100644 AGENTS.md create mode 100644 CHANGELOG-VIRTANA.md create mode 100644 dev/READ_AL.md create mode 100644 dev/read_access_logs.py create mode 100644 memory-bank/01-overview.md create mode 100644 memory-bank/02-build-and-test.md create mode 100644 memory-bank/03-virtana-changes.md create mode 100644 memory-bank/04-deployment.md create mode 100644 memory-bank/05-gcp-egress-pricing.md rename docs/PER_SHARE_EGRESS_MONITORING.md => memory-bank/06-egress-monitoring.md (100%) create mode 100644 memory-bank/07-access-log-table-reference.md create mode 100644 memory-bank/README.md diff --git a/AGENTS.md b/AGENTS.md new file mode 100644 index 000000000..710f950bd --- /dev/null +++ b/AGENTS.md @@ -0,0 +1,174 @@ +# Delta Sharing – Agent Instructions + +Virtana fork of [delta-io/delta-sharing](https://github.com/delta-io/delta-sharing), extending the reference server with GCS-backed telemetry and multi-environment Kubernetes deployments. + +Deeper background lives in the [memory bank](memory-bank/README.md) — overview, build/test, Virtana divergence, deployment, and GCP egress pricing. + +## Rules for Every Task + +1. **Update [CHANGELOG-VIRTANA.md](CHANGELOG-VIRTANA.md).** Any change to behaviour, configuration, build setup, or deployment gets reflected in the relevant area section (add a new area if none fits). It's an informal running overview organized by area, not a dated/versioned log — no need to track release status. That file is the record of how this fork diverges from upstream. +2. **Run the tests** for whatever you touched (see [Build & Test](#build--test)). `server/test` runs scalastyle first and fails on style violations. +3. **Verify the image builds** when you change `build.sbt`, dependencies, or anything packaging-related: + ```bash + DOCKER_DEFAULT_PLATFORM=linux/amd64 make image + ``` + The env var is required — the deployment target is amd64 and building on an arm64 Mac without it produces an unusable image. + +## Branching + +**`master` is the working branch** — do development and target PRs here. CI/CD builds and deploys from it. + +`master` was forked from upstream [`delta-io/delta-sharing` `branch-1.3`](https://github.com/delta-io/delta-sharing/tree/branch-1.3) at commit `793cc19b2a3434227ebdc7f34bb2141047a67925`. + +| Branch | Purpose | +|--------|---------| +| `master` | Working + release branch — CI/CD deploys from here | +| `virtana-1.3` | Historical fork branch | +| `main` | Upstream delta-io/delta-sharing — **not** the deployment branch | + +Never target `main` for Virtana changes. + +## Divergence from Upstream + +See [CHANGELOG-VIRTANA.md](CHANGELOG-VIRTANA.md) for the full inventory of Virtana-modified and Virtana-authored files, plus the merge risks to watch when pulling from upstream. When fixing a bug, check that file first to tell whether the code is ours or upstream's. + +## Build & Test + +Use the wrapper script — plain `sbt` may not be in PATH: + +```bash +./build/sbt server/compile # Scala 2.12 (server only) +./build/sbt client/compile # Cross-compiled 2.12 + 2.13 +./build/sbt spark/compile # Scala 2.13 only + +./build/sbt server/test # Runs scalastyle before tests +./build/sbt client/test +./build/sbt spark/test + +# Image build — always pin the platform, the deploy target is amd64 +DOCKER_DEFAULT_PLATFORM=linux/amd64 make image +DOCKER_DEFAULT_PLATFORM=linux/amd64 ./build/sbt server/docker:publishLocal +``` + +**Scala versions**: server → 2.12.18 + Spark 3.5.3 (Java 8); spark connector → 2.13.13 + Spark 4.0.0 (Java 17). + +scalastyle runs as part of `server/test` and fails the build on violations. Config: [scalastyle-config.xml](scalastyle-config.xml). Use `org.scalatest.FunSuite` (not `AnyFunSuite`) for test suites. + +Python tests: +```bash +python/dev/pytest +``` + +## Project Structure + +``` +server/src/main/scala/io/delta/sharing/server/ REST endpoints (Armeria), main entry point +server/src/main/scala/.../server/config/ ServerConfig, AccessLoggingConfig, Share/Table config +server/src/main/scala/.../server/common/ CloudFileSigner (GCS + S3), JsonPredicates +server/src/main/scala/.../server/telemetry/ Access log emission, GCP pricing tier, Delta writer +server/src/main/scala/.../kernel/ Delta Lake kernel integration +manifests/ Kustomize overlays per environment +ci/ Jenkins pipeline, deploy scripts +python/delta_sharing/ Python client library +``` + +## Server Configuration + +The server takes `--config `. Key settings (from [manifests/base/configmap.yaml](manifests/base/configmap.yaml)): + +```yaml +host: "0.0.0.0" +port: 8080 +endpoint: "/delta-sharing" +preSignedUrlTimeoutSeconds: 3600 +deltaTableCacheSize: 100 +evaluateJsonPredicateHints: true +evaluateJsonPredicateHintsV2: true +requestTimeoutSeconds: 180 +idleTimeoutSeconds: 120 # Must exceed the proxy's IdleConnTimeout or clients see EOF errors +queryTablePageSizeLimit: 10000 +perfLoggingEnabled: true + +authorization: + bearerToken: "" # Injected from $BEARER_TOKEN at deploy time + +shares: + - name: "share_name" + schemas: + - name: "schema_name" + tables: + - name: "table_name" + location: "gs://bucket/path" + cdfEnabled: false +``` + +## GCS Integration + +**Authentication**: Set `GOOGLE_APPLICATION_CREDENTIALS` to a service account JSON path. In Kubernetes, Workload Identity is used via the `dl-sharing` service account — no key file needed in production. + +**Dependencies** (in [build.sbt](build.sbt)): +- `com.google.cloud:google-cloud-storage` — GCS SDK +- `com.google.cloud.bigdataoss:gcs-connector:hadoop2-2.2.4` — Hadoop FS integration + +**GCS signing**: `server/src/main/scala/.../server/common/CloudFileSigner.scala` — generates pre-signed GCS URLs using `GoogleHadoopFileSystem` and `StorageResourceId`. + +Use `gs://` table paths (not `s3://` or `s3a://`) for GCS-backed tables. + +**GCS environments**: + +| Environment | Bucket | Region | +|------------|--------|--------| +| zing-dev | `gs://zing-dev-197522-dl-v1/` | us-central1 | +| zing-preview | `gs://zing-preview-dl-v1/` | us-central1 | +| zcloud-prod | `gs://zcloud-prod-dl-v1/` | us-central1 | +| zcloud-prod2 | `gs://zcloud-prod2-dl-v1/` | us-west4 | +| zcloud-prod3 | `gs://zcloud-prod3-dl-v1/` | australia-southeast1 | +| zcloud-emea | GCS bucket | europe-west3 | + +## Access Logging (Virtana Extension) + +Virtana-added feature that writes structured access log entries to a Delta table on GCS after each query/CDF request. See [memory-bank/06-egress-monitoring.md](memory-bank/06-egress-monitoring.md) and [memory-bank/07-access-log-table-reference.md](memory-bank/07-access-log-table-reference.md). + +**Config block**: +```yaml +accessLogging: + enabled: true + sourceRegion: "us-central1" # GCP region of this server's data bucket + detectGcpTraffic: true # Classify inter-GCP traffic for pricing tier + clientRegionHeader: "x-client-region" + clientIpHeader: "x-forwarded-for" + deltaTablePath: "gs://bucket/path/tenant/_system" + deltaFlushIntervalSeconds: 60 + deltaFlushBatchSize: 1000 +``` + +**Key classes**: +- `AccessLogEmitter` / `DeltaAccessLogWriter` — buffered async writer to GCS Delta table +- `GcpPricingTier` — classifies egress by region pair (e.g. `internet_to_na_eu`, `same_region`); refreshes GCP IP ranges from gstatic.com every 24h +- `GcpIpRangeLookup` — IP range → GCP region detection + +**Delta table** (`access_log_br__system`): a single consolidated, unpartitioned table with `tenantId` as a column for per-tenant filtering. Protocol (1,2) for Delta Standalone compatibility; pre-created by `deltalake-admin` — the writer does not create the schema. + +## Kubernetes Deployment + +Manifests use Kustomize overlays at [manifests/](manifests/). Each environment overlay extends `manifests/base/`. + +Deployment pattern: +- Init container merges base config + shares config using `envsubst` (substitutes `$BEARER_TOKEN`, `$GCP_PROJECT_ID`) +- Sidecar container `zc-api-proxy` handles JWT/Auth0 authentication in front of the sharing server on `localhost:8080` +- `dl-sharing` Kubernetes SA is bound to a GCP SA via Workload Identity + +```bash +make deploy-dev # Deploy to zing-dev +make deploy-preview # Deploy to zing-preview +make deploy-prod # Deploy to zcloud-prod +``` + +## Environment Variables + +| Variable | Where used | +|----------|-----------| +| `BEARER_TOKEN` | Kubernetes secret → injected into server config | +| `GCP_PROJECT_ID` | Kubernetes ConfigMap → injected into server config | +| `GOOGLE_APPLICATION_CREDENTIALS` | Local dev / test — path to service account JSON | +| `AZURE_TEST_ACCOUNT_KEY` | Optional — Azure blob storage integration tests | diff --git a/CHANGELOG-VIRTANA.md b/CHANGELOG-VIRTANA.md new file mode 100644 index 000000000..0e801e877 --- /dev/null +++ b/CHANGELOG-VIRTANA.md @@ -0,0 +1,68 @@ +# Virtana Changelog + +Everything Virtana has changed on top of the upstream [delta-io/delta-sharing](https://github.com/delta-io/delta-sharing) reference server. + +Fork point: upstream [`branch-1.3`](https://github.com/delta-io/delta-sharing/tree/branch-1.3) at commit `793cc19b2a3434227ebdc7f34bb2141047a67925`. + +```bash +git diff --stat 793cc19b2a3434227ebdc7f34bb2141047a67925..HEAD +``` + +**Keep this file current.** This isn't a formal historical log — just a running, informal overview of +everything Virtana has changed in this fork, organized by area. Every task that changes behaviour, +configuration, build setup, or deployment updates the relevant area below (or adds a new one if it +doesn't fit). Upstream files not listed here are unmodified — when fixing a bug, first check whether +it lives in Virtana code or upstream code. + +## Changes + +### Telemetry package (largest addition) + +`server/src/main/scala/io/delta/sharing/server/telemetry/`, entirely Virtana-authored: + +| File | Purpose | +|------|---------| +| `AccessLogEmitter.scala` | `AccessLogEntry` / `PricingContextLogEntry` models; JSON emitter | +| `DeltaAccessLogWriter.scala` | Async buffered writer to the GCS Delta table | +| `GcpPricingTier.scala` | Egress cost classification by source→destination region pair | +| `GcpIpRangeLookup.scala` | GCP IP range → region detection (refreshed from `cloud.json`) | + +Each has a matching suite under `server/src/test/scala/.../telemetry/`. + +### Upstream server files, modified + +| File | Virtana change | +|------|----------------| +| `DeltaSharingService.scala` | ~600 added lines: access log emission per query/CDF request, client region + IP header extraction, egress byte accounting, idle timeout config | +| `config/ServerConfig.scala` | New `AccessLoggingConfig` case class; new `perfLoggingEnabled` and `idleTimeoutSeconds` options | +| `DeltaSharedTableProtocol.scala` | New `CdfQueryTimings` / `TableQueryTimings` / `QueryResultTimings` observability models; `QueryResult` gained a `timings` field | +| `DeltaSharedTableLoader.scala` | `loadTableWithUpdateCost` returns `deltaLog.update()` elapsed time for perf logging | +| `standalone/internal/DeltaSharedTable.scala` | Per-phase timing instrumentation (snapshot resolve, replay, signing); near-timeout warnings | +| `standalone/internal/DeltaSharingCDCReader.scala` | CDF stream timing instrumentation | + +### Build & infrastructure + +| File | Virtana change | +|------|----------------| +| `build.sbt` | Global slf4j binding exclusions (`slf4j-log4j12`, `slf4j-reload4j`); per-dependency `ExclusionRule("org.slf4j")`; swapped `slf4j-simple` → `logback-classic`; `delta-standalone` changed from `provided` to a compile dependency (required by the Delta log writer); pinned `dockerBaseImage := "eclipse-temurin:8-jre"` | +| `server/src/main/resources/logback.xml` | New — preserves log severity in GCP Cloud Console | +| `scalastyle-config.xml` | License header regex relaxed to allow any copyright year | + +### Deployment & docs + +- `manifests/` — Kustomize base + 6 environment overlays +- `ci/` — Jenkinsfile, Makefile, `deploy.sh`, per-environment deployment YAML +- `Makefile` — image build and `deploy-*` targets +- [memory-bank/](memory-bank/README.md) — committed shared context for humans and agents: overview, + build/test, Virtana divergence, deployment, and GCP egress pricing reference. Linked from + [AGENTS.md](AGENTS.md). Replaces `docs/PER_SHARE_EGRESS_MONITORING.md` (now + `memory-bank/06-egress-monitoring.md`) and `docs/Notes.md` (now + `memory-bank/07-access-log-table-reference.md`); the old `docs/` is gone. The access log table + description was corrected there: `access_log_br__system` is unpartitioned (matches + `DeltaAccessLogWriter`), not partitioned by `year`/`month`/`day`. + +## Known divergence risks when merging upstream + +- `DeltaSharingService.scala` is heavily modified — expect conflicts on any upstream change there. +- The `delta-standalone` scope change (`provided` → compile) must be preserved or `DeltaAccessLogWriter` fails at runtime. +- slf4j exclusions must be preserved or the server emits "multiple SLF4J bindings" and loses Cloud Console severity. diff --git a/Makefile b/Makefile index 2a9c52bdf..030afbea0 100644 --- a/Makefile +++ b/Makefile @@ -2,15 +2,16 @@ REPO_ROOT := $(dir $(abspath $(lastword $(MAKEFILE_LIST)))) SERVER_VERSION := $(shell grep 'version in ThisBuild' $(REPO_ROOT)version.sbt | sed 's/.*"\(.*\)".*/\1/') SERVICE_IMAGE := $(shell grep '^SERVICE_IMAGE' $(REPO_ROOT).env | cut -d'=' -f2) -# Resolve IMAGE_TAG from ci/.env (used for push and deploy) -# IMAGE_TAG := $(shell grep '^IMAGE_TAG' $(REPO_ROOT)ci/.env | cut -d':' -f2 | tr -d ' =') +# IMAGE_TAG is normally supplied by the caller (e.g. Jenkins export). ?= leaves an +# already-set/exported value alone and only supplies "dev" for local builds - +IMAGE_TAG ?= dev .PHONY: image image: @echo "Building Docker image from fork (version $(SERVER_VERSION))" @cd $(REPO_ROOT) && build/sbt server/docker:publishLocal @echo "Image built: deltaio/delta-sharing-server:$(SERVER_VERSION)" - @docker tag deltaio/delta-sharing-server:$(SERVER_VERSION) ${SERVICE_IMAGE}:${IMAGE_TAG:-dev} + @docker tag deltaio/delta-sharing-server:$(SERVER_VERSION) ${SERVICE_IMAGE}:${IMAGE_TAG} .PHONY: push-dev push-dev: diff --git a/dev/READ_AL.md b/dev/READ_AL.md new file mode 100644 index 000000000..32e0f78e8 --- /dev/null +++ b/dev/READ_AL.md @@ -0,0 +1,114 @@ +# Read Access Logs + +Query Delta Sharing access logs via Delta Sharing protocol or directly from GCS. + +Access logs are stored in per-tenant Delta tables named `access_log_{tenant_id}`, where +`tenant_id` is extracted from the share name (pattern: `{tenant_id}_share`). + +## Installation + +```bash +# For direct GCS access (recommended) +pip3 install deltalake pandas pyarrow gcsfs + +# For Delta Sharing mode +pip3 install delta-sharing pandas +``` + +## Modes + +### Direct GCS Mode (`--direct`) + +Reads directly from GCS bucket. Requires `gcloud auth application-default login`. + +```bash +# Read _system tenant logs (default) +python3 read_access_logs.py --direct --env zing-dev + +# Read specific tenant logs +python3 read_access_logs.py --direct --env zing-dev --tenant ipa7l25ufagwjfmv + +# Filter last 7 days +python3 read_access_logs.py --direct --env zing-dev --tenant _system --days 7 + +# Filter by share name +python3 read_access_logs.py --direct --env zing-dev --tenant mytenantid --share myshare + +# Combine filters +python3 read_access_logs.py --direct --env zcloud-prod --tenant _system --days 30 --limit 100 +``` + +### Delta Sharing Mode (default) + +Requires `_system_share` to be configured on the server. + +```bash +# Using default profile +python3 read_access_logs.py --profile profile-dev.json + +# List available shares/tables +python3 read_access_logs.py --profile profile-dev.json --list-tables + +# Custom table URL +python3 read_access_logs.py --table-url "profile-prod.json#_system_share.SystemData_v0_1.access_log__system" +``` + +## Per-Tenant Table Structure + +| Share accessed | Table name | GCS path | +|----------------|------------|----------| +| `_system_share` | `access_log__system` | `gs://.../tenant/_system/access_log__system` | +| `ipa7l25ufagwjfmv_share` | `access_log_ipa7l25ufagwjfmv` | `gs://.../tenant/_system/access_log_ipa7l25ufagwjfmv` | +| `hhgp5t6oz3nvczk7_share` | `access_log_hhgp5t6oz3nvczk7` | `gs://.../tenant/_system/access_log_hhgp5t6oz3nvczk7` | + +## Environments + +| Environment | GCS Base Path | +|-----------------|----------------------------------------------------| +| `zing-dev` | gs://zing-dev-197522-dl-v1/datalake/data/tenant/_system | +| `zing-preview` | gs://zing-preview-dl-v1/datalake/data/tenant/_system | +| `zcloud-prod` | gs://zcloud-prod-dl-v1/datalake/data/tenant/_system | +| `zcloud-prod2` | gs://zcloud-prod2-dl-v1/datalake/data/tenant/_system | +| `zcloud-prod3` | gs://zcloud-prod3-dl-v1/datalake/data/tenant/_system | + +## Output Formats + +```bash +# Summary with stats (default) +python3 read_access_logs.py --direct --env zing-dev + +# Full table +python3 read_access_logs.py --direct --env zing-dev --output table + +# Export to CSV +python3 read_access_logs.py --direct --env zcloud-prod --output csv --output-file logs.csv + +# Export to JSON +python3 read_access_logs.py --direct --env zing-preview --limit 100 --output json --output-file logs.json +``` + +## Options Reference + +| Option | Description | +|------------------|--------------------------------------------------| +| `--direct` | Read directly from GCS (requires gcloud auth) | +| `--env` | Environment for direct mode | +| `--tenant` | Tenant ID for direct mode (default: `_system`) | +| `--profile` | Delta Sharing profile file | +| `--table-url` | Full Delta Sharing table URL | +| `--list-tables` | List available shares/schemas/tables | +| `--days N` | Filter to last N days | +| `--share NAME` | Filter by share name | +| `--limit N` | Limit number of records | +| `--output` | Format: `summary`, `table`, `csv`, `json` | +| `--output-file` | Write output to file (for csv/json) | + +## Output Columns + +- `timestampMs` - Request timestamp (milliseconds) +- `share` - Share name accessed +- `schema` - Schema name +- `table` - Table name +- `egressBytes` - Bytes transferred +- `pricingTier` - GCS pricing tier +- `clientRegion` - Client's region \ No newline at end of file diff --git a/dev/read_access_logs.py b/dev/read_access_logs.py new file mode 100644 index 000000000..36e2931d8 --- /dev/null +++ b/dev/read_access_logs.py @@ -0,0 +1,387 @@ +#!/usr/bin/env python3 +""" +Read access logs via Delta Sharing or directly from GCS. + +Two modes: +1. Via Delta Sharing (requires _system_share to be configured in server) +2. Direct from GCS (requires gcloud auth and deltalake library) + +Install: + pip3 install delta-sharing pandas + # For direct GCS access: + pip3 install deltalake gcsfs + +Run (via Delta Sharing): + python3 read_access_logs.py --profile profile-dev.json + python3 read_access_logs.py --days 7 --limit 100 + +Run (direct GCS - when Delta Sharing share not configured): + python3 read_access_logs.py --direct --env zing-dev --tenant _system + python3 read_access_logs.py --direct --env zing-dev --tenant ipa7l25ufagwjfmv +""" + +from __future__ import annotations + +import argparse +from datetime import datetime, timezone, timedelta + +import pandas as pd + +# GCS base paths by environment (per-tenant tables live under these) +ENV_BASE_PATHS = { + "zing-dev": "gs://zing-dev-197522-dl-v1/datalake/data/tenant/_system", + "zing-preview": "gs://zing-preview-dl-v1/datalake/data/tenant/_system", + "zcloud-prod": "gs://zcloud-prod-dl-v1/datalake/data/tenant/_system", + "zcloud-prod2": "gs://zcloud-prod2-dl-v1/datalake/data/tenant/_system", + "zcloud-prod3": "gs://zcloud-prod3-dl-v1/datalake/data/tenant/_system", +} + +# Default table URL format: #.. +DEFAULT_TABLE_URL = "profile-dev.json#_system_share.SystemData_v0_1.access_log__system" + + +def list_available_tables(profile: str) -> None: + """List all shares, schemas, and tables available through the profile.""" + import delta_sharing + + print(f"Listing tables from profile: {profile}") + client = delta_sharing.SharingClient(profile) + + shares = client.list_shares() + print(f"\nFound {len(shares)} shares:") + + for share in shares: + print(f"\n Share: {share.name}") + schemas = client.list_schemas(share) + for schema in schemas: + print(f" Schema: {schema.name}") + tables = client.list_tables(schema) + for table in tables: + table_url = f"{profile}#{share.name}.{schema.name}.{table.name}" + print(f" Table: {table.name}") + print(f" URL: {table_url}") + + +def read_access_logs_direct( + env: str, + tenant: str = "_system", + days: int | None = None, + share_filter: str | None = None, + limit: int | None = None, +) -> pd.DataFrame: + """Read access logs directly from GCS using deltalake library.""" + try: + import deltalake + except ImportError: + print("Missing deltalake library. Install with: pip3 install deltalake gcsfs") + raise SystemExit(1) + + base_path = ENV_BASE_PATHS[env] + table_path = f"{base_path}/access_log_{tenant}" + print(f"Reading directly from GCS: {table_path}") + + try: + dt = deltalake.DeltaTable(table_path) + print(f"Table version: {dt.version()}") + print(f"Files: {len(dt.file_uris())}") + + if len(dt.file_uris()) == 0: + print("\nWARNING: Delta table has 0 files tracked.") + print("The parquet files may exist but aren't in the Delta log.") + print("This happens when writes fail due to protocol version mismatch.") + print("\nTo fix: delete the table and let the server recreate it:") + print(f" gsutil -m rm -r {table_path}/") + return pd.DataFrame() + + df = dt.to_pandas() + except Exception as e: + print(f"Error reading Delta table: {e}") + print("\nTrying to read orphaned parquet files directly...") + return read_orphaned_parquet(table_path) + + return apply_filters(df, days, share_filter, limit) + + +def read_orphaned_parquet(table_path: str) -> pd.DataFrame: + """Read parquet files directly when Delta log is broken.""" + try: + import gcsfs + import pyarrow.parquet as pq + except ImportError: + print("Missing gcsfs/pyarrow. Install with: pip3 install gcsfs pyarrow") + raise SystemExit(1) + + # Strip gs:// prefix for gcsfs + gcs_path = table_path.replace("gs://", "") + fs = gcsfs.GCSFileSystem() + + # Find all parquet files + parquet_files = [] + for root, dirs, files in fs.walk(gcs_path): + for f in files: + if f.endswith(".parquet"): + parquet_files.append(f"gs://{root}/{f}") + + print(f"Found {len(parquet_files)} orphaned parquet files") + + if not parquet_files: + return pd.DataFrame() + + # Read all parquet files + dfs = [] + for pf in parquet_files[:100]: # Limit to first 100 files + try: + df = pq.read_table(pf, filesystem=fs).to_pandas() + dfs.append(df) + except Exception as e: + print(f" Warning: could not read {pf}: {e}") + + if not dfs: + return pd.DataFrame() + + return pd.concat(dfs, ignore_index=True) + + +def apply_filters( + df: pd.DataFrame, + days: int | None = None, + share_filter: str | None = None, + limit: int | None = None, +) -> pd.DataFrame: + """Apply common filters to the DataFrame.""" + if df.empty: + return df + + # Filter by date range + if days is not None and "timestampMs" in df.columns: + now = datetime.now(timezone.utc) + cutoff_ms = int((now - timedelta(days=days)).timestamp() * 1000) + df = df[df["timestampMs"] >= cutoff_ms] + print(f"After date filter ({days} days): {len(df):,} rows") + + # Filter by share name + if share_filter and "share" in df.columns: + df = df[df["share"] == share_filter] + print(f"After share filter ({share_filter}): {len(df):,} rows") + + # Sort by timestamp descending + if "timestampMs" in df.columns: + df = df.sort_values("timestampMs", ascending=False) + + if limit: + df = df.head(limit) + + return df + + +def read_access_logs( + table_url: str, + days: int | None = None, + share_filter: str | None = None, + limit: int | None = None, +) -> pd.DataFrame: + """Read access logs via Delta Sharing.""" + import delta_sharing + + print(f"Reading from: {table_url}") + + # Get table version first + try: + version = delta_sharing.get_table_version(table_url) + print(f"Table version: {version}") + except Exception as e: + print(f"Warning: Could not get table version: {e}") + + # Load table as pandas DataFrame + df = delta_sharing.load_as_pandas(table_url) + print(f"Loaded {len(df):,} rows") + + return apply_filters(df, days, share_filter, limit) + + +def summarize(df: pd.DataFrame) -> None: + """Print summary statistics.""" + print(f"\n{'=' * 60}") + print(f"Total records: {len(df):,}") + + if df.empty: + print("No data found.") + print(f"{'=' * 60}\n") + return + + # Show columns + print(f"Columns: {list(df.columns)}") + + if "egressBytes" in df.columns: + total_bytes = df["egressBytes"].sum() + print(f"Total egress: {total_bytes:,.0f} bytes ({total_bytes / 1e9:.2f} GB)") + + if "share" in df.columns: + print(f"\nBy share:") + share_stats = ( + df.groupby("share") + .agg( + { + "egressBytes": "sum", + } + ) + .rename(columns={"egressBytes": "bytes"}) + ) + share_stats["requests"] = df.groupby("share").size() + share_stats["egress_gb"] = share_stats["bytes"] / 1e9 + print(share_stats.sort_values("bytes", ascending=False).head(10).to_string()) + + if "pricingTier" in df.columns: + print(f"\nBy pricing tier:") + tier_stats = df.groupby("pricingTier")["egressBytes"].sum() + print(tier_stats.sort_values(ascending=False).to_string()) + + if "timestampMs" in df.columns: + ts = pd.to_datetime(df["timestampMs"], unit="ms", utc=True) + print(f"\nTime range: {ts.min()} to {ts.max()}") + + print(f"{'=' * 60}\n") + + +def main() -> int: + parser = argparse.ArgumentParser( + description=__doc__, + formatter_class=argparse.RawDescriptionHelpFormatter, + ) + parser.add_argument( + "--direct", + action="store_true", + help="Read directly from GCS instead of via Delta Sharing", + ) + parser.add_argument( + "--env", + choices=list(ENV_BASE_PATHS.keys()), + default="zing-dev", + help="Environment for --direct mode", + ) + parser.add_argument( + "--tenant", + default="_system", + help="Tenant ID for --direct mode (e.g., _system, ipa7l25ufagwjfmv)", + ) + parser.add_argument( + "--table-url", + default=DEFAULT_TABLE_URL, + help="Delta Sharing table URL: #..
", + ) + parser.add_argument( + "--profile", + help="Override profile file (e.g., profile-dev.json, profile-prod.json)", + ) + parser.add_argument( + "--list-tables", + action="store_true", + help="List all available shares/schemas/tables and exit", + ) + parser.add_argument( + "--days", + type=int, + help="Filter to last N days", + ) + parser.add_argument( + "--share", + help="Filter by share name (the 'share' column in access logs)", + ) + parser.add_argument( + "--limit", + type=int, + help="Limit number of records returned", + ) + parser.add_argument( + "--output", + choices=["summary", "json", "csv", "table"], + default="summary", + help="Output format", + ) + parser.add_argument( + "--output-file", + help="Write output to file (for json/csv)", + ) + args = parser.parse_args() + + # List tables mode (via Delta Sharing only) + if args.list_tables: + profile = args.profile or args.table_url.split("#")[0] + list_available_tables(profile) + return 0 + + # Read data + try: + if args.direct: + df = read_access_logs_direct( + env=args.env, + tenant=args.tenant, + days=args.days, + share_filter=args.share, + limit=args.limit, + ) + else: + # Build table URL + table_url = args.table_url + if args.profile: + parts = table_url.split("#") + if len(parts) == 2: + table_url = f"{args.profile}#{parts[1]}" + else: + table_url = args.profile + + df = read_access_logs( + table_url=table_url, + days=args.days, + share_filter=args.share, + limit=args.limit, + ) + except Exception as e: + print(f"Error reading table: {e}") + import traceback + + traceback.print_exc() + return 1 + + # Output + if args.output == "summary": + summarize(df) + if not df.empty: + print("Recent entries:") + # Show a subset of columns for readability + display_cols = [ + c + for c in [ + "timestampMs", + "share", + "schema", + "table", + "egressBytes", + "pricingTier", + "clientRegion", + ] + if c in df.columns + ] + print(df[display_cols].head(10).to_string()) + elif args.output == "json": + output = df.to_json(orient="records", date_format="iso", indent=2) + if args.output_file: + with open(args.output_file, "w") as f: + f.write(output) + print(f"Wrote {len(df)} records to {args.output_file}") + else: + print(output) + elif args.output == "csv": + if args.output_file: + df.to_csv(args.output_file, index=False) + print(f"Wrote {len(df)} records to {args.output_file}") + else: + print(df.to_csv(index=False)) + elif args.output == "table": + print(df.to_string()) + + return 0 + + +if __name__ == "__main__": + raise SystemExit(main()) diff --git a/memory-bank/01-overview.md b/memory-bank/01-overview.md new file mode 100644 index 000000000..310d472dc --- /dev/null +++ b/memory-bank/01-overview.md @@ -0,0 +1,78 @@ +# Overview + +Virtana fork of [delta-io/delta-sharing](https://github.com/delta-io/delta-sharing). It extends the +reference sharing server with GCS-backed access-log telemetry and multi-environment Kubernetes +deployment. + +- Fork point: upstream [`branch-1.3`](https://github.com/delta-io/delta-sharing/tree/branch-1.3) at + commit `793cc19b2a3434227ebdc7f34bb2141047a67925`. +- The org is Virtana; the git remote path is still `github.com/zenoss`. + +## Branching + +| Branch | Purpose | +|--------|---------| +| `master` | Working + release branch — CI/CD builds and deploys from here. Target PRs here. | +| `virtana-1.3` | Historical fork branch | +| `main` | Upstream mirror — never target it for Virtana changes | + +## Layout + +``` +server/src/main/scala/io/delta/sharing/server/ Armeria REST endpoints, main entry point + .../server/config/ ServerConfig, AccessLoggingConfig, Share/Table config + .../server/common/ CloudFileSigner (GCS + S3), JsonPredicates + .../server/telemetry/ Virtana access logging + egress pricing tier + .../kernel/ Delta Lake kernel integration +client/, spark/, python/delta_sharing/ Client libraries +manifests/ Kustomize base + per-environment overlays +ci/ Jenkinsfile, Makefile, deploy.sh +memory-bank/ This documentation set +dev/ Local scratch scripts — out of scope +``` + +Server entry points worth knowing: `DeltaSharingService.scala`, `DeltaSharedTableLoader.scala`, +`DeltaSharedTableProtocol.scala`, `SharedTableManager.scala`. + +## Server configuration + +The server takes `--config `. Base config lives in +[manifests/base/configmap.yaml](../manifests/base/configmap.yaml). + +```yaml +host: "0.0.0.0" +port: 8080 +endpoint: "/delta-sharing" +preSignedUrlTimeoutSeconds: 3600 +deltaTableCacheSize: 100 +evaluateJsonPredicateHints: true +evaluateJsonPredicateHintsV2: true +requestTimeoutSeconds: 180 +idleTimeoutSeconds: 120 # must exceed the proxy's IdleConnTimeout or clients see EOF errors +queryTablePageSizeLimit: 10000 +perfLoggingEnabled: true + +authorization: + bearerToken: "" # injected from $BEARER_TOKEN at deploy time + +shares: + - name: "share_name" + schemas: + - name: "schema_name" + tables: + - name: "table_name" + location: "gs://bucket/path" + cdfEnabled: false +``` + +`accessLogging` is a Virtana addition — see [03-virtana-changes.md](03-virtana-changes.md). + +## GCS integration + +- **Auth**: `GOOGLE_APPLICATION_CREDENTIALS` pointing at a service account JSON for local work; in + Kubernetes, Workload Identity via the `dl-sharing` service account (no key file). +- **Dependencies** (in [build.sbt](../build.sbt)): `com.google.cloud:google-cloud-storage`, + `com.google.cloud.bigdataoss:gcs-connector:hadoop2-2.2.4`. +- **Signing**: `server/src/main/scala/.../server/common/CloudFileSigner.scala` generates pre-signed + GCS URLs via `GoogleHadoopFileSystem` and `StorageResourceId`. +- Use `gs://` table paths, not `s3://` or `s3a://`. diff --git a/memory-bank/02-build-and-test.md b/memory-bank/02-build-and-test.md new file mode 100644 index 000000000..be719bb84 --- /dev/null +++ b/memory-bank/02-build-and-test.md @@ -0,0 +1,46 @@ +# Build & Test + +Use the wrapper script — plain `sbt` may not be on PATH. + +```bash +./build/sbt server/compile # Scala 2.12 (server only) +./build/sbt client/compile # cross-compiled 2.12 + 2.13 +./build/sbt spark/compile # Scala 2.13 only + +./build/sbt server/test # runs scalastyle first, fails on violations +./build/sbt client/test +./build/sbt spark/test + +python/dev/pytest # Python client tests + +DOCKER_DEFAULT_PLATFORM=linux/amd64 make image +DOCKER_DEFAULT_PLATFORM=linux/amd64 ./build/sbt server/docker:publishLocal +``` + +## Things that bite + +- `DOCKER_DEFAULT_PLATFORM=linux/amd64` is mandatory. The deploy target is amd64; building on an + arm64 Mac without it produces an unusable image. +- Test suites must use `org.scalatest.FunSuite`, **not** `AnyFunSuite`. +- scalastyle runs as part of `server/test` and fails the build. Config: + [scalastyle-config.xml](../scalastyle-config.xml) (license header regex relaxed to allow any year). +- In [build.sbt](../build.sbt): `delta-standalone` must stay a compile dependency (not `provided`) + or `DeltaAccessLogWriter` fails at runtime; the slf4j exclusions plus `logback-classic` must + survive or the server emits multiple-binding warnings and loses severity in GCP Cloud Console. + +## Versions + +| Module | Scala | Spark | Java | +|--------|-------|-------|------| +| server | 2.12.18 | 3.5.3 | 8 | +| spark connector | 2.13.13 | 4.0.0 | 17 | + +## Checklist for every change + +1. Update the relevant area section under `## Changes` in + [CHANGELOG-VIRTANA.md](../CHANGELOG-VIRTANA.md) (add a new area if none fits) for any change to + behaviour, configuration, build setup, or deployment. It's an informal running overview organized + by area, not a dated/versioned log. +2. Run the tests for whatever you touched. +3. Rebuild the image if `build.sbt`, dependencies, or packaging changed. +4. Write "Virtana", not "Zenoss", in docs. diff --git a/memory-bank/03-virtana-changes.md b/memory-bank/03-virtana-changes.md new file mode 100644 index 000000000..600465dd7 --- /dev/null +++ b/memory-bank/03-virtana-changes.md @@ -0,0 +1,72 @@ +# What Virtana Changed + +[CHANGELOG-VIRTANA.md](../CHANGELOG-VIRTANA.md) is the authoritative log; this file is the shape of +the divergence. When fixing a bug, check here first to tell whether the code is ours or upstream's. + +## New — telemetry package (largest addition) + +`server/src/main/scala/io/delta/sharing/server/telemetry/`, entirely Virtana-authored: + +| File | Purpose | +|------|---------| +| `AccessLogEmitter.scala` | `AccessLogEntry` / `PricingContextLogEntry` models; JSON emitter | +| `DeltaAccessLogWriter.scala` | Async buffered writer to the GCS Delta table | +| `GcpPricingTier.scala` | Egress cost classification by source→destination region pair (e.g. `same_region`, `internet_to_na_eu`); refreshes GCP IP ranges from gstatic.com every 24h | +| `GcpIpRangeLookup.scala` | GCP IP range → region detection | + +Each has a matching suite under `server/src/test/scala/.../telemetry/`. + +### Config block + +```yaml +accessLogging: + enabled: true + sourceRegion: "us-central1" # GCP region of this server's data bucket + detectGcpTraffic: true # classify inter-GCP traffic for pricing tier + clientRegionHeader: "x-client-region" + clientIpHeader: "x-forwarded-for" + deltaTablePath: "gs://bucket/path/tenant/_system" + deltaFlushIntervalSeconds: 60 + deltaFlushBatchSize: 1000 +``` + +### Output table + +`access_log_br__system` — a single consolidated, **unpartitioned** table; `tenantId` is a column so +queries can filter by tenant. Protocol (1,2) for Delta Standalone compatibility. The table must be +pre-created by `deltalake-admin`; the writer does not create the schema. + +Background: [06-egress-monitoring.md](06-egress-monitoring.md), +[07-access-log-table-reference.md](07-access-log-table-reference.md). + +## Modified — upstream server files + +| File | Virtana change | +|------|----------------| +| `DeltaSharingService.scala` | ~600 added lines: access log emission per query/CDF request, client region + IP header extraction, egress byte accounting, idle timeout config | +| `config/ServerConfig.scala` | New `AccessLoggingConfig` case class; new `perfLoggingEnabled` and `idleTimeoutSeconds` options | +| `DeltaSharedTableProtocol.scala` | New `CdfQueryTimings` / `TableQueryTimings` / `QueryResultTimings` observability models; `QueryResult` gained a `timings` field | +| `DeltaSharedTableLoader.scala` | `loadTableWithUpdateCost` returns `deltaLog.update()` elapsed time for perf logging | +| `standalone/internal/DeltaSharedTable.scala` | Per-phase timing instrumentation (snapshot resolve, replay, signing); near-timeout warnings | +| `standalone/internal/DeltaSharingCDCReader.scala` | CDF stream timing instrumentation | + +## Modified — build & infrastructure + +| File | Virtana change | +|------|----------------| +| `build.sbt` | Global slf4j binding exclusions (`slf4j-log4j12`, `slf4j-reload4j`); per-dependency `ExclusionRule("org.slf4j")`; `slf4j-simple` → `logback-classic`; `delta-standalone` moved from `provided` to a compile dependency; pinned `dockerBaseImage := "eclipse-temurin:8-jre"` | +| `server/src/main/resources/logback.xml` | New — preserves log severity in GCP Cloud Console | +| `scalastyle-config.xml` | License header regex relaxed to allow any copyright year | + +## New — deployment & docs + +`manifests/` (Kustomize base + 6 overlays), `ci/` (Jenkinsfile, Makefile, `deploy.sh`, +per-environment deployment YAML), root `Makefile`, `memory-bank/`. + +## Merge risks when pulling from upstream + +1. `DeltaSharingService.scala` is heavily modified — expect conflicts on any upstream change there. +2. The `delta-standalone` scope change (`provided` → compile) must be preserved or + `DeltaAccessLogWriter` fails at runtime. +3. The slf4j exclusions must be preserved or the server emits "multiple SLF4J bindings" and loses + Cloud Console severity. diff --git a/memory-bank/04-deployment.md b/memory-bank/04-deployment.md new file mode 100644 index 000000000..bcd9e155e --- /dev/null +++ b/memory-bank/04-deployment.md @@ -0,0 +1,48 @@ +# Deployment + +Kustomize overlays live in [manifests/](../manifests/). Each environment overlay extends +`manifests/base/`: `zing-dev`, `zing-preview`, `zcloud-prod`, `zcloud-prod2`, `zcloud-prod3`, +`zcloud-emea`. + +## Pod shape + +- An init container merges the base config with the shares config using `envsubst`, substituting + `$BEARER_TOKEN` and `$GCP_PROJECT_ID`. +- A sidecar container `zc-api-proxy` handles JWT/Auth0 authentication in front of the sharing + server on `localhost:8080`. +- The `dl-sharing` Kubernetes service account is bound to a GCP service account via Workload + Identity. + +## Make targets + +```bash +DOCKER_DEFAULT_PLATFORM=linux/amd64 make image # build/sbt server/docker:publishLocal, then tag +make push-dev # push to gcr.io/zing-dev-197522/ +make deploy-dev # ci/deploy.sh dev +make deploy-preview # ci/deploy.sh preview +make deploy-prod # ci/deploy.sh prod +``` + +Version comes from [version.sbt](../version.sbt); `SERVICE_IMAGE` comes from `.env` and `IMAGE_TAG` +from `ci/.env`. CI lives in [ci/Jenkinsfile](../ci/Jenkinsfile) with per-environment +`ci/deployment-{dev,preview,prod}.yaml`. + +## GCS environments + +| Environment | Bucket | Region | +|-------------|--------|--------| +| zing-dev | `gs://zing-dev-197522-dl-v1/` | us-central1 | +| zing-preview | `gs://zing-preview-dl-v1/` | us-central1 | +| zcloud-prod | `gs://zcloud-prod-dl-v1/` | us-central1 | +| zcloud-prod2 | `gs://zcloud-prod2-dl-v1/` | us-west4 | +| zcloud-prod3 | `gs://zcloud-prod3-dl-v1/` | australia-southeast1 | +| zcloud-emea | GCS bucket | europe-west3 | + +## Environment variables + +| Variable | Where used | +|----------|-----------| +| `BEARER_TOKEN` | Kubernetes secret → injected into server config | +| `GCP_PROJECT_ID` | Kubernetes ConfigMap → injected into server config | +| `GOOGLE_APPLICATION_CREDENTIALS` | Local dev/test — path to service account JSON | +| `AZURE_TEST_ACCOUNT_KEY` | Optional — Azure blob storage integration tests | diff --git a/memory-bank/05-gcp-egress-pricing.md b/memory-bank/05-gcp-egress-pricing.md new file mode 100644 index 000000000..5470f1da4 --- /dev/null +++ b/memory-bank/05-gcp-egress-pricing.md @@ -0,0 +1,61 @@ +# GCP Egress Pricing + +Reference data behind `GcpPricingTier.scala` and the egress classification in the access log. + +## Premium tier internet egress (VM → internet) + +Tiered by monthly volume: + +| Destination | 0–1 TiB | 1–10 TiB | 10 TiB+ | +|-------------|---------|----------|---------| +| North America ↔ Europe | $0.12 | $0.11 | $0.085 | +| Any GCP region → Asia (excl. Indonesia/Korea) | $0.12 | $0.11 | $0.085 | +| Any GCP region → Indonesia/Korea | $0.19 | $0.18 | $0.15 | +| Any GCP region → South America | $0.19 | $0.18 | $0.15 | + +## Inter-region egress (VM → VM, VM → Google service), per GiB + +| Region pair | Price | +|-------------|-------| +| North America ↔ North America | $0.02 | +| North America ↔ Europe | $0.05 | +| Asia ↔ Asia | $0.08 | +| Any ↔ Australia/Indonesia | $0.10 | +| Any ↔ South America | $0.14 | + +## Geographic grouping + +- **North America**: us-central1, us-east1, us-west1, us-west2, … +- **Europe**: europe-west1 (Belgium), europe-west2 (London), europe-north1 (Finland), … +- **Asia**: asia-east1, asia-northeast1, asia-southeast1, asia-south1, … +- **South America**: southamerica-east1 +- **Australia**: australia-southeast1 + +## How the server classifies a request + +Inputs: configured source region (the data bucket's region), destination (client region or IP +geolocation), egress bytes, and request type (`query` or `cdf_stream`). + +Implementation in +[server/src/main/scala/io/delta/sharing/server/DeltaSharingService.scala](../server/src/main/scala/io/delta/sharing/server/DeltaSharingService.scala): + +- `ClientLocationContext` — region, subdivision, IP, and pricing group. +- `buildClientLocationContext()` — resolves client location from request headers. +- `DefaultPricingGroupsByRegion` — country code → pricing group: + - `na_eu`: US, CA, MX, GB, IE, DE, FR, NL, BE, CH, AT, ES, PT, IT, SE, NO, DK, FI, PL, CZ, HU, RO + - `apac`: JP, KR, IN, SG, HK, TW, ID, MY, PH, TH, VN, AU, NZ + - `latam`: BR, AR, CL, CO, PE + +Header detection order: + +1. Region: `x-client-region`, `x-appengine-country`, `cf-ipcountry`, `cloudfront-viewer-country` +2. Subdivision: `x-client-region-subdivision`, `x-appengine-region` +3. IP: `x-forwarded-for`, `x-envoy-external-address`, `x-real-ip`, `true-client-ip` + +Resolution: try subdivision code, then region code, against the configured pricing map, then the +default map, then a `"*"` wildcard entry; fall back to the raw region, then `"unknown"`. + +## Known gaps + +- Client region/country often absent — needs IP geolocation or a custom header from the proxy. +- Pricing group shows `unknown` when no mapping resolves. diff --git a/docs/PER_SHARE_EGRESS_MONITORING.md b/memory-bank/06-egress-monitoring.md similarity index 100% rename from docs/PER_SHARE_EGRESS_MONITORING.md rename to memory-bank/06-egress-monitoring.md diff --git a/memory-bank/07-access-log-table-reference.md b/memory-bank/07-access-log-table-reference.md new file mode 100644 index 000000000..42eba1921 --- /dev/null +++ b/memory-bank/07-access-log-table-reference.md @@ -0,0 +1,230 @@ +# System Tenant: Delta Sharing Access Logs + +Technical reference for writing Delta Sharing access logs to the Data Access feature. + +> The schema in this file is the original table design. The shipped writer +> (`DeltaAccessLogWriter.scala`) writes an **unpartitioned** table with additional audit columns — +> see [06-egress-monitoring.md](06-egress-monitoring.md) for the schema actually in use. Treat that +> file as authoritative for the writer; this one for table locations, retention, and query patterns. + +## Overview + +The Delta Sharing server (deployed in Kubernetes) writes access logs to a Delta table managed by the `_system` tenant. This enables billing analytics, usage monitoring, and audit trails for all Delta Sharing activity. + +## Table Details + +| Property | Value | +|----------|-------| +| **Table Name** | `access_log_br__system` | +| **Template** | `access_log_br` | +| **Partitioning** | None — the shipped writer writes unpartitioned files | +| **Change Data Feed** | Enabled | + +### GCS Locations by Environment + +| Environment | Table Location | +|-------------|----------------| +| 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` | + +## Schema + +All columns must be provided by the writer. The `year`, `month`, and `day` partition columns are computed by the writer from `timestampMs`. + +### Required Columns + +| Column | Type | Description | +|--------|------|-------------| +| `logType` | STRING | Log entry type, e.g., `"ACCESS_LOG"` | +| `share` | STRING | Name of the Delta Share accessed | +| `schema` | STRING | Schema name within the share | +| `table` | STRING | Table name accessed | +| `egressBytes` | LONG | Number of bytes transferred | +| `timestampMs` | LONG | Unix timestamp in **milliseconds** | +| `year` | INT | Partition: `YEAR(timestampMs)` - computed by writer | +| `month` | INT | Partition: `MONTH(timestampMs)` - computed by writer | +| `day` | INT | Partition: `DAY(timestampMs)` - computed by writer | + +### Optional Columns + +| Column | Type | Description | +|--------|------|-------------| +| `pricingTier` | STRING | Egress pricing tier (e.g., `"internet_to_na_eu"`, `"same_region"`) | +| `requestType` | STRING | Type of request (e.g., `"query"`, `"metadata"`, `"getFiles"`) | +| `clientRegion` | STRING | Client's geographic region (e.g., `"US"`, `"EU"`, `"APAC"`) | + +**Note**: The table uses protocol (1,2) for compatibility with Delta Standalone writers. No generated columns - the writer must compute partition values. + +## Writing Access Logs + +### JSON Record Format + +```json +{ + "logType": "ACCESS_LOG", + "share": "customer123_share", + "schema": "DataAccess_v0_1", + "table": "metric_ag_customer123", + "egressBytes": 1048576, + "pricingTier": "internet_to_na_eu", + "timestampMs": 1717502400000, + "requestType": "query", + "clientRegion": "US" +} +``` + +### Using Spark/PySpark + +```python +from pyspark.sql import SparkSession +from pyspark.sql.types import StructType, StructField, StringType, LongType + +# Schema for writing (excludes generated columns) +write_schema = StructType( + [ + StructField("logType", StringType(), False), + StructField("share", StringType(), False), + StructField("schema", StringType(), False), + StructField("table", StringType(), False), + StructField("egressBytes", LongType(), False), + StructField("pricingTier", StringType(), True), + StructField("timestampMs", LongType(), False), + StructField("requestType", StringType(), True), + StructField("clientRegion", StringType(), True), + ] +) + +# Write to Delta table +df.write.format("delta").mode("append").save( + "gs:///datalake/data/tenant/_system/access_log_br__system" +) +``` + +### Using Delta Lake Rust/Python SDK + +```python +import deltalake + +# Append records +deltalake.write_deltalake( + "gs:///datalake/data/tenant/_system/access_log_br__system", + data, # pandas DataFrame or PyArrow Table + mode="append", +) +``` + +## Integration Notes for Delta Sharing Server + +### Kubernetes Deployment Considerations + +1. **GCS Credentials**: The server pod needs a GCP service account with write access to the `_system` tenant path. Mount the service account key as a secret or use Workload Identity. + +2. **Batching**: Consider batching log writes to reduce I/O overhead: + - Buffer logs in memory (e.g., 1000 records or 60 seconds) + - Write in batch to Delta table + - Flush on graceful shutdown + +3. **Async Writing**: Log writes should be async to avoid impacting request latency. Use a background thread/coroutine with a bounded queue. + +4. **Error Handling**: Log write failures should not fail client requests. Log errors to stderr and emit metrics for monitoring. + +### Environment Detection + +Determine the correct GCS bucket based on the Kubernetes namespace or environment variable: + +```python +ENV_BUCKETS = { + "zing-dev": "zing-dev-197522-dl-v1", + "zing-preview": "zing-preview-dl-v1", + "zcloud-prod": "zcloud-prod-dl-v1", + "zcloud-prod2": "zcloud-prod2-dl-v1", + "zcloud-prod3": "zcloud-prod3-dl-v1", +} + + +def get_access_log_path(env: str) -> str: + bucket = ENV_BUCKETS[env] + return f"gs://{bucket}/datalake/data/tenant/_system/access_log_br__system" +``` + +### Timestamp Handling + +- `timestampMs` must be Unix epoch in **milliseconds** (not seconds) +- Use `System.currentTimeMillis()` (Java), `time.time_ns() // 1_000_000` (Python), or `Date.now()` (JS) +- The `timestamp`, `year`, `month`, `day` columns are generated automatically on write + +### Pricing Tier Values + +Suggested values for `pricingTier` based on GCP egress pricing: + +| Value | Description | +|-------|-------------| +| `same_region` | Client and server in same GCP region | +| `same_continent` | Cross-region, same continent | +| `internet_to_na_eu` | Internet egress to North America/Europe | +| `internet_to_apac` | Internet egress to Asia-Pacific | +| `internet_to_other` | Internet egress to other regions | + +### Request Type Values + +Suggested values for `requestType`: + +| Value | Description | +|-------|-------------| +| `listShares` | List available shares | +| `listSchemas` | List schemas in a share | +| `listTables` | List tables in a schema | +| `getTableMetadata` | Get table metadata | +| `getTableVersion` | Get table version | +| `query` | Query/read table data | +| `getFiles` | Get files for a table version | + +## Querying Access Logs + +Once data is written, it's accessible via Delta Sharing under: +- **Share**: `_system_share` +- **Schema**: `SystemData_v0_1` +- **Table**: `access_log_br__system` + +### Example Queries + +```sql +-- Total egress by share (last 30 days) +SELECT share, SUM(egressBytes) as total_bytes +FROM access_log_br__system +WHERE year = 2026 AND month = 6 +GROUP BY share +ORDER BY total_bytes DESC; + +-- Request counts by type +SELECT requestType, COUNT(*) as requests +FROM access_log_br__system +WHERE timestamp >= current_timestamp - INTERVAL 7 DAYS +GROUP BY requestType; + +-- Daily egress trend +SELECT year, month, day, SUM(egressBytes) as daily_bytes +FROM access_log_br__system +GROUP BY year, month, day +ORDER BY year, month, day; +``` + +## Retention + +| Environment | Retention | +|-------------|-----------| +| zing-dev | 30 days | +| zing-preview | 90 days | +| Production (prod/prod2/prod3) | 450 days (15 months) | + +## See Also + +- [06-egress-monitoring.md](06-egress-monitoring.md) — how the server classifies egress and writes + these records +- [../AGENTS.md](../AGENTS.md) — project overview and CLI commands + +The schema template (`templates/access_log_br.json`) and the per-environment `_system` tenant +configs live in the `deltalake-admin` repo, which pre-creates this table during tenant onboarding. diff --git a/memory-bank/README.md b/memory-bank/README.md new file mode 100644 index 000000000..75a01a52c --- /dev/null +++ b/memory-bank/README.md @@ -0,0 +1,19 @@ +# Memory Bank + +Shared, committed context for anyone (human or agent) working on this fork. +Read the file that matches your task. + +| File | Contents | +|------|----------| +| [01-overview.md](01-overview.md) | Fork point, branching, project layout, server config keys, GCS integration | +| [02-build-and-test.md](02-build-and-test.md) | Build/test commands, Scala versions, style rules, per-task checklist | +| [03-virtana-changes.md](03-virtana-changes.md) | Divergence from upstream: telemetry package, modified files, merge risks | +| [04-deployment.md](04-deployment.md) | Kustomize overlays, make targets, GCS environments, env vars, CI | +| [05-gcp-egress-pricing.md](05-gcp-egress-pricing.md) | GCP egress pricing tiers and how classification is implemented | +| [06-egress-monitoring.md](06-egress-monitoring.md) | Per-share egress monitoring: tier resolution, config, Delta writer behaviour, log formats | +| [07-access-log-table-reference.md](07-access-log-table-reference.md) | `access_log_br__system` table reference: schema, locations, example queries, retention | + +Related: [../AGENTS.md](../AGENTS.md) (agent instructions), [../CHANGELOG-VIRTANA.md](../CHANGELOG-VIRTANA.md) +(authoritative divergence log). + +`dev/` is local scratch tooling and is intentionally not documented here.