From 51f6917444cec5497675604dc4ce89a92d04deda Mon Sep 17 00:00:00 2001 From: Evan Jarrett Date: Thu, 8 Jan 2026 22:52:32 -0600 Subject: [PATCH] add log shipper begin envvar cleanup --- .env.appview.example | 38 ++- .env.example | 293 ++++++++++++++++++++ .env.hold.example | 26 ++ .gitignore | 3 + CLAUDE.md | 1 - cmd/appview/serve.go | 70 ++--- cmd/hold/main.go | 16 +- deploy/.env.prod.template | 13 - deploy/docker-compose.prod.yml | 1 - docker-compose.yml | 40 ++- docs/DEVELOPMENT.md | 4 - docs/appview.md | 12 - pkg/appview/config.go | 95 +++---- pkg/appview/db/annotations_test.go | 2 +- pkg/appview/db/device_store_test.go | 2 +- pkg/appview/db/hold_store_test.go | 2 +- pkg/appview/db/oauth_store_test.go | 6 +- pkg/appview/db/queries_test.go | 18 +- pkg/appview/db/readonly.go | 8 +- pkg/appview/db/readonly_test.go | 2 +- pkg/appview/db/schema.go | 10 +- pkg/appview/db/session_store_test.go | 2 +- pkg/appview/db/tag_delete_test.go | 2 +- pkg/appview/handlers/api.go | 44 +-- pkg/appview/handlers/device_test.go | 2 +- pkg/appview/middleware/auth_test.go | 4 +- pkg/auth/hold_remote_test.go | 2 +- pkg/auth/token/handler_test.go | 2 +- pkg/hold/config.go | 63 +++++ pkg/logging/logger.go | 44 ++- pkg/logging/shipper.go | 308 +++++++++++++++++++++ pkg/logging/shipper_test.go | 386 +++++++++++++++++++++++++++ pkg/logging/victoria.go | 112 ++++++++ 33 files changed, 1430 insertions(+), 203 deletions(-) create mode 100644 .env.example create mode 100644 pkg/logging/shipper.go create mode 100644 pkg/logging/shipper_test.go create mode 100644 pkg/logging/victoria.go diff --git a/.env.appview.example b/.env.appview.example index b86cab7..4445db0 100644 --- a/.env.appview.example +++ b/.env.appview.example @@ -1,3 +1,9 @@ +# ============================================================================== +# DEPRECATED: This file is deprecated. Use .env.example instead. +# This file will be removed in a future version. +# See .env.example for the unified configuration file. +# ============================================================================== + # ATCR AppView Configuration # Copy this file to .env.appview and fill in your values # Load with: source .env.appview && ./bin/atcr-appview serve @@ -9,9 +15,6 @@ # HTTP listen address (default: :5000) ATCR_HTTP_ADDR=:5000 -# Debug listen address (default: :5001) -# ATCR_DEBUG_ADDR=:5001 - # Base URL for the AppView service (REQUIRED for production) # Used to generate OAuth redirect URIs and JWT realms # Development: Auto-detected from ATCR_HTTP_ADDR (e.g., http://127.0.0.1:5000) @@ -63,19 +66,10 @@ ATCR_DEFAULT_HOLD_DID=did:web:127.0.0.1:8080 # UI Configuration # ============================================================================== -# Enable web UI (default: true) -# Set to "false" to disable web interface and run registry-only -ATCR_UI_ENABLED=true - # SQLite database path for UI data (sessions, stars, pull counts, etc.) # Default: /var/lib/atcr/ui.db # ATCR_UI_DATABASE_PATH=/var/lib/atcr/ui.db -# Skip database migrations on startup (default: false) -# Set to "true" to skip running migrations (useful for tests or fresh databases) -# Production: Keep as "false" to ensure migrations are applied -SKIP_DB_MIGRATIONS=false - # ============================================================================== # Logging Configuration # ============================================================================== @@ -86,6 +80,26 @@ ATCR_LOG_LEVEL=debug # Log formatter: text, json (default: text) # ATCR_LOG_FORMATTER=text +# ============================================================================== +# Remote Log Shipping (optional) +# ============================================================================== + +# Backend: victoria, opensearch, loki (empty = disabled) +# ATCR_LOG_SHIPPER_BACKEND=victoria + +# Remote log service URL +# ATCR_LOG_SHIPPER_URL=http://victorialogs:9428 + +# Number of logs to batch before flushing (default: 100) +# ATCR_LOG_SHIPPER_BATCH_SIZE=100 + +# Max time between flushes (default: 5s) +# ATCR_LOG_SHIPPER_FLUSH_INTERVAL=5s + +# Basic auth credentials (optional) +# ATCR_LOG_SHIPPER_USERNAME= +# ATCR_LOG_SHIPPER_PASSWORD= + # ============================================================================== # Hold Health Check Configuration # ============================================================================== diff --git a/.env.example b/.env.example new file mode 100644 index 0000000..fc653eb --- /dev/null +++ b/.env.example @@ -0,0 +1,293 @@ +# ============================================================================== +# ATCR Configuration +# ============================================================================== +# This file contains ALL configuration options for both AppView and Hold services. +# Copy to .env and uncomment/modify the values you need. +# +# QUICKSTART (minimum for local development): +# HOLD_PUBLIC_URL=http://127.0.0.1:8080 +# ATCR_DEFAULT_HOLD_DID=did:web:127.0.0.1:8080 +# +# QUICKSTART (minimum for production): +# APPVIEW_DOMAIN=atcr.io +# HOLD_DOMAIN=hold01.atcr.io +# HOLD_OWNER=did:plc:your-did +# AWS_ACCESS_KEY_ID=xxx +# AWS_SECRET_ACCESS_KEY=xxx +# S3_BUCKET=xxx +# S3_ENDPOINT=https://xxx +# +# ============================================================================== + +# ============================================================================== +# DOMAIN CONFIGURATION (Production) +# ============================================================================== +# These are used by docker-compose.prod.yml to derive other values automatically. +# For local dev, skip these and set the explicit URLs below instead. + +# Main AppView domain (registry API + web UI) +# APPVIEW_DOMAIN=atcr.io + +# Hold service domain +# Used to derive: HOLD_PUBLIC_URL, ATCR_DEFAULT_HOLD_DID +# HOLD_DOMAIN=hold01.atcr.io + +# ============================================================================== +# APPVIEW - SERVER CONFIGURATION +# ============================================================================== + +# HTTP listen address +# Default: :5000 +# ATCR_HTTP_ADDR=:5000 + +# Public URL for OAuth redirect URIs and JWT realms +# Development: Auto-detected from ATCR_HTTP_ADDR (e.g., http://127.0.0.1:5000) +# Production: Set to your public URL (e.g., https://atcr.io) +# ATCR_BASE_URL=https://atcr.io + +# Service name for JWT issuer/service fields +# Default: Derived from ATCR_BASE_URL hostname, or "atcr.io" +# ATCR_SERVICE_NAME=atcr.io + +# ============================================================================== +# APPVIEW - STORAGE CONFIGURATION (REQUIRED) +# ============================================================================== + +# Default hold service DID for users without their own storage (REQUIRED) +# Format: did:web:hostname[:port] +# Docker dev: did:web:172.28.0.3:8080 +# Local dev: did:web:127.0.0.1:8080 +# Production: did:web:hold01.atcr.io +ATCR_DEFAULT_HOLD_DID=did:web:127.0.0.1:8080 + +# ============================================================================== +# APPVIEW - AUTHENTICATION +# ============================================================================== + +# Path to JWT signing private key (auto-generated if missing) +# Default: /var/lib/atcr/auth/private-key.pem +# ATCR_AUTH_KEY_PATH=/var/lib/atcr/auth/private-key.pem + +# Path to JWT signing certificate (auto-generated if missing) +# Default: /var/lib/atcr/auth/private-key.crt +# ATCR_AUTH_CERT_PATH=/var/lib/atcr/auth/private-key.crt + +# JWT token expiration in seconds +# Default: 300 (5 minutes) +# ATCR_TOKEN_EXPIRATION=300 + +# Path to OAuth client P-256 signing key (auto-generated for production) +# Used for confidential OAuth client authentication +# Localhost deployments always use public OAuth clients (no key needed) +# Default: /var/lib/atcr/oauth/client.key +# ATCR_OAUTH_KEY_PATH=/var/lib/atcr/oauth/client.key + +# OAuth client display name (shown in authorization screens) +# Default: AT Container Registry +# ATCR_CLIENT_NAME=AT Container Registry + +# ============================================================================== +# APPVIEW - WEB UI +# ============================================================================== + +# SQLite database path for UI data (sessions, stars, pull counts, etc.) +# Default: /var/lib/atcr/ui.db +# ATCR_UI_DATABASE_PATH=/var/lib/atcr/ui.db + +# ============================================================================== +# APPVIEW - JETSTREAM (ATProto Event Streaming) +# ============================================================================== + +# Jetstream WebSocket URL for real-time ATProto events +# Default: wss://jetstream2.us-west.bsky.network/subscribe +# JETSTREAM_URL=wss://jetstream2.us-west.bsky.network/subscribe + +# Enable backfill worker to sync historical records +# Default: true +# ATCR_BACKFILL_ENABLED=true + +# ATProto relay endpoint for backfill sync API +# Default: https://relay1.us-east.bsky.network +# ATCR_RELAY_ENDPOINT=https://relay1.us-east.bsky.network + +# Backfill sync interval +# Default: 1h +# Examples: 30m, 1h, 2h, 24h +# ATCR_BACKFILL_INTERVAL=1h + +# ============================================================================== +# APPVIEW - HEALTH CHECKS +# ============================================================================== + +# How often to check health of hold endpoints in the background +# Default: 15m +# ATCR_HEALTH_CHECK_INTERVAL=15m + +# How long to cache health check results +# Default: 15m +# ATCR_HEALTH_CACHE_TTL=15m + +# ============================================================================== +# HOLD SERVICE - SERVER CONFIGURATION (REQUIRED) +# ============================================================================== + +# Public URL of hold service (REQUIRED) +# The hostname becomes the hold name/record key +# Local dev: http://127.0.0.1:8080 +# Production: https://hold01.atcr.io +HOLD_PUBLIC_URL=http://127.0.0.1:8080 + +# HTTP listen address +# Default: :8080 +# HOLD_SERVER_ADDR=:8080 + +# Allow public blob reads (pulls) without authentication +# Writes (pushes) always require crew membership via PDS +# Default: false +# HOLD_PUBLIC=false + +# ATProto relay endpoint for requesting crawl on startup +# Makes the hold's embedded PDS discoverable by the relay network +# Default: (empty - disabled) +# Set to https://bsky.network to enable +# HOLD_RELAY_ENDPOINT=https://bsky.network + +# ============================================================================== +# HOLD SERVICE - EMBEDDED PDS +# ============================================================================== + +# Directory path for embedded PDS carstore (SQLite database) +# Default: /var/lib/atcr-hold +# If empty, embedded PDS is disabled +# Note: This is a directory path, NOT a file path +# Carstore creates db.sqlite3 inside this directory +HOLD_DATABASE_DIR=/var/lib/atcr-hold + +# Path to signing key (auto-generated on first run if missing) +# Default: {HOLD_DATABASE_DIR}/signing.key +# HOLD_KEY_PATH=/var/lib/atcr-hold/signing.key + +# ============================================================================== +# HOLD SERVICE - REGISTRATION & ACCESS CONTROL +# ============================================================================== + +# Your ATProto DID (REQUIRED for registration) +# Get your DID: https://bsky.social/xrpc/com.atproto.identity.resolveHandle?handle=yourhandle.bsky.social +# On first run with HOLD_OWNER set: +# 1. Hold service prints OAuth URL to logs +# 2. Visit URL to authorize +# 3. Hold creates captain + crew records +# 4. Registration complete! +# HOLD_OWNER=did:plc:your-did-here + +# Allow any authenticated user to register as crew +# Default: false (only explicit crew members can write) +# Set to true for open/community holds +# HOLD_ALLOW_ALL_CREW=false + +# ============================================================================== +# HOLD SERVICE - BLUESKY INTEGRATION +# ============================================================================== + +# Enable Bluesky posts when users push container images +# When enabled, creates posts announcing image pushes +# Default: false +# HOLD_BLUESKY_POSTS_ENABLED=false + +# Avatar image URL to download during bootstrap +# HOLD_PROFILE_AVATAR=https://imgs.blue/evan.jarrett.net/1TpTOdtS60GdJWBYEqtK22y688jajbQ9a5kbYRFtwuqrkBAE + +# ============================================================================== +# HOLD SERVICE - ADMIN +# ============================================================================== + +# Enable admin panel +# Default: false +# HOLD_ADMIN_ENABLED=false + +# ============================================================================== +# STORAGE - S3 CONFIGURATION +# ============================================================================== + +# Storage driver type +# Options: s3, filesystem +# Default: s3 +STORAGE_DRIVER=s3 + +# S3 Access Credentials +AWS_ACCESS_KEY_ID=your_access_key +AWS_SECRET_ACCESS_KEY=your_secret_key + +# S3 Region +# For third-party S3 providers, this is ignored when S3_ENDPOINT is set, +# but must be a valid AWS region to pass validation. +# Default: us-east-1 +AWS_REGION=us-east-1 + +# S3 Bucket Name +S3_BUCKET=atcr-blobs + +# S3 Endpoint (for S3-compatible services) +# Examples: +# - Storj: https://gateway.storjshare.io +# - UpCloud: https://[bucket-id].upcloudobjects.com +# - Minio: http://minio:9000 +# Leave empty for AWS S3 +# S3_ENDPOINT=https://gateway.storjshare.io + +# ============================================================================== +# STORAGE - FILESYSTEM CONFIGURATION +# ============================================================================== + +# Root directory for filesystem storage (when STORAGE_DRIVER=filesystem) +# Default: /var/lib/atcr/hold +# STORAGE_ROOT_DIR=/var/lib/atcr/hold + +# ============================================================================== +# LOGGING (Shared by AppView and Hold) +# ============================================================================== + +# Log level: debug, info, warn, error +# Default: info +ATCR_LOG_LEVEL=info + +# Log formatter: text, json +# Default: text +# ATCR_LOG_FORMATTER=text + +# ============================================================================== +# REMOTE LOG SHIPPING (Optional) +# ============================================================================== + +# Backend: victoria, opensearch, loki (empty = disabled) +# ATCR_LOG_SHIPPER_BACKEND=victoria + +# Remote log service URL +# ATCR_LOG_SHIPPER_URL=http://victorialogs:9428 + +# Number of logs to batch before flushing +# Default: 100 +# ATCR_LOG_SHIPPER_BATCH_SIZE=100 + +# Max time between flushes +# Default: 5s +# ATCR_LOG_SHIPPER_FLUSH_INTERVAL=5s + +# Basic auth credentials (optional) +# ATCR_LOG_SHIPPER_USERNAME= +# ATCR_LOG_SHIPPER_PASSWORD= + +# ============================================================================== +# DEVELOPMENT / TESTING +# ============================================================================== + +# Enable test mode +# - Uses HTTP for local DID resolution +# - Adds transition:generic scope for OAuth +# - Uses localhost for OAuth redirects while storing real URL in hold record +# Default: false +# TEST_MODE=false + +# Disable presigned URLs (force proxy mode for testing) +# Default: false +# DISABLE_PRESIGNED_URLS=false diff --git a/.env.hold.example b/.env.hold.example index a490c98..8f7fbcb 100644 --- a/.env.hold.example +++ b/.env.hold.example @@ -1,3 +1,9 @@ +# ============================================================================== +# DEPRECATED: This file is deprecated. Use .env.example instead. +# This file will be removed in a future version. +# See .env.example for the unified configuration file. +# ============================================================================== + # ATCR Hold Service Configuration # Copy this file to .env and fill in your values @@ -125,3 +131,23 @@ ATCR_LOG_LEVEL=debug # Log formatter: text, json (default: text) # ATCR_LOG_FORMATTER=text + +# ============================================================================== +# Remote Log Shipping (optional) +# ============================================================================== + +# Backend: victoria, opensearch, loki (empty = disabled) +# ATCR_LOG_SHIPPER_BACKEND=victoria + +# Remote log service URL +# ATCR_LOG_SHIPPER_URL=http://victorialogs:9428 + +# Number of logs to batch before flushing (default: 100) +# ATCR_LOG_SHIPPER_BATCH_SIZE=100 + +# Max time between flushes (default: 5s) +# ATCR_LOG_SHIPPER_FLUSH_INTERVAL=5s + +# Basic auth credentials (optional) +# ATCR_LOG_SHIPPER_USERNAME= +# ATCR_LOG_SHIPPER_PASSWORD= diff --git a/.gitignore b/.gitignore index 23c8aff..01093c3 100644 --- a/.gitignore +++ b/.gitignore @@ -12,6 +12,9 @@ tmp/ # Environment configuration .env +# Docker-created quota config (actual config is in deploy/quotas.yaml) +quotas.yaml + # Generated assets (run go generate to rebuild) pkg/appview/licenses/spdx-licenses.json pkg/appview/static/js/htmx.min.js diff --git a/CLAUDE.md b/CLAUDE.md index 016ef28..54719bc 100644 --- a/CLAUDE.md +++ b/CLAUDE.md @@ -669,7 +669,6 @@ See `.env.appview.example` for all available options. Key environment variables: - `ATCR_TOKEN_EXPIRATION` - JWT expiration in seconds (default: 300) **UI:** -- `ATCR_UI_ENABLED` - Enable web interface (default: true) - `ATCR_UI_DATABASE_PATH` - SQLite database path (default: `/var/lib/atcr/ui.db`) **Jetstream:** diff --git a/cmd/appview/serve.go b/cmd/appview/serve.go index d5c71d2..a06e762 100644 --- a/cmd/appview/serve.go +++ b/cmd/appview/serve.go @@ -5,7 +5,6 @@ import ( "database/sql" "encoding/json" "fmt" - "html/template" "log/slog" "net/http" "os" @@ -66,14 +65,22 @@ func serveRegistry(cmd *cobra.Command, args []string) error { return fmt.Errorf("failed to load config from environment: %w", err) } - // Initialize structured logging - logging.InitLogger(cfg.LogLevel) + // Initialize structured logging with optional remote shipping + logging.InitLoggerWithShipper(cfg.LogLevel, logging.ShipperConfig{ + Backend: cfg.LogShipper.Backend, + URL: cfg.LogShipper.URL, + BatchSize: cfg.LogShipper.BatchSize, + FlushInterval: cfg.LogShipper.FlushInterval, + Service: "appview", + Username: cfg.LogShipper.Username, + Password: cfg.LogShipper.Password, + }) slog.Info("Configuration loaded successfully from environment") // Initialize UI database first (required for all stores) slog.Info("Initializing UI database", "path", cfg.UI.DatabasePath) - uiDatabase, uiReadOnlyDB, uiSessionStore := db.InitializeDatabase(cfg.UI.Enabled, cfg.UI.DatabasePath, cfg.UI.SkipDBMigrations) + uiDatabase, uiReadOnlyDB, uiSessionStore := db.InitializeDatabase(cfg.UI.DatabasePath) if uiDatabase == nil { return fmt.Errorf("failed to initialize UI database - required for session storage") } @@ -183,32 +190,28 @@ func serveRegistry(cmd *cobra.Command, args []string) error { mainRouter.Use(chimiddleware.GetHead) // Automatically handle HEAD requests for GET routes mainRouter.Use(routes.CORSMiddleware()) - // Load templates if UI is enabled - var uiTemplates *template.Template - if cfg.UI.Enabled { - var err error - uiTemplates, err = appview.Templates() - if err != nil { - slog.Warn("Failed to load UI templates", "error", err) - } else { - // Register UI routes with dependencies - routes.RegisterUIRoutes(mainRouter, routes.UIDependencies{ - Database: uiDatabase, - ReadOnlyDB: uiReadOnlyDB, - SessionStore: uiSessionStore, - OAuthClientApp: oauthClientApp, - OAuthStore: oauthStore, - Refresher: refresher, - BaseURL: baseURL, - DeviceStore: deviceStore, - HealthChecker: healthChecker, - ReadmeFetcher: readmeFetcher, - Templates: uiTemplates, - DefaultHoldDID: defaultHoldDID, - }) - } + // Load templates (UI is always enabled) + uiTemplates, err := appview.Templates() + if err != nil { + return fmt.Errorf("failed to load UI templates: %w", err) } + // Register UI routes with dependencies + routes.RegisterUIRoutes(mainRouter, routes.UIDependencies{ + Database: uiDatabase, + ReadOnlyDB: uiReadOnlyDB, + SessionStore: uiSessionStore, + OAuthClientApp: oauthClientApp, + OAuthStore: oauthStore, + Refresher: refresher, + BaseURL: baseURL, + DeviceStore: deviceStore, + HealthChecker: healthChecker, + ReadmeFetcher: readmeFetcher, + Templates: uiTemplates, + DefaultHoldDID: defaultHoldDID, + }) + // Create OAuth server oauthServer := oauth.NewServer(oauthClientApp) // Connect server to refresher for cache invalidation @@ -458,15 +461,8 @@ func serveRegistry(cmd *cobra.Command, args []string) error { // Register credential helper version API (public endpoint) mainRouter.Handle("/api/credential-helper/version", &uihandlers.CredentialHelperVersionHandler{ - Version: cfg.CredentialHelper.Version, TangledRepo: cfg.CredentialHelper.TangledRepo, - Checksums: cfg.CredentialHelper.Checksums, }) - if cfg.CredentialHelper.Version != "" { - slog.Info("Credential helper version API enabled", - "endpoint", "/api/credential-helper/version", - "version", cfg.CredentialHelper.Version) - } // Create HTTP server server := &http.Server{ @@ -500,14 +496,18 @@ func serveRegistry(cmd *cobra.Command, args []string) error { defer cancel() if err := server.Shutdown(shutdownCtx); err != nil { + logging.Shutdown() // Flush remaining logs return fmt.Errorf("server shutdown error: %w", err) } case err := <-errChan: // Stop health worker on error (workerCancel called by defer) healthWorker.Stop() + logging.Shutdown() // Flush remaining logs return fmt.Errorf("server error: %w", err) } + // Flush any remaining logs before exit + logging.Shutdown() return nil } diff --git a/cmd/hold/main.go b/cmd/hold/main.go index f321d0a..2ccf14e 100644 --- a/cmd/hold/main.go +++ b/cmd/hold/main.go @@ -35,8 +35,16 @@ func main() { os.Exit(1) } - // Initialize structured logging - logging.InitLogger(cfg.LogLevel) + // Initialize structured logging with optional remote shipping + logging.InitLoggerWithShipper(cfg.LogLevel, logging.ShipperConfig{ + Backend: cfg.LogShipper.Backend, + URL: cfg.LogShipper.URL, + BatchSize: cfg.LogShipper.BatchSize, + FlushInterval: cfg.LogShipper.FlushInterval, + Service: "hold", + Username: cfg.LogShipper.Username, + Password: cfg.LogShipper.Password, + }) // Initialize embedded PDS if database path is configured // This must happen before creating HoldService since service needs PDS for authorization @@ -234,6 +242,7 @@ func main() { select { case err := <-serverErr: slog.Error("Server failed", "error", err) + logging.Shutdown() // Flush remaining logs os.Exit(1) case sig := <-sigChan: slog.Info("Received signal, shutting down gracefully", "signal", sig) @@ -275,5 +284,8 @@ func main() { } else { slog.Info("Server shutdown complete") } + + // Flush any remaining logs before exit + logging.Shutdown() } } diff --git a/deploy/.env.prod.template b/deploy/.env.prod.template index b079216..af8d1de 100644 --- a/deploy/.env.prod.template +++ b/deploy/.env.prod.template @@ -150,16 +150,6 @@ ATCR_TOKEN_EXPIRATION=300 # Default: AT Container Registry # ATCR_CLIENT_NAME=AT Container Registry -# Enable web UI -# Default: true -ATCR_UI_ENABLED=true - -# Skip database migrations on startup -# Default: false (migrations are applied on startup) -# Set to "true" only for testing or when migrations are managed externally -# Production: Keep as "false" to ensure migrations are applied -SKIP_DB_MIGRATIONS=false - # ============================================================================== # Logging Configuration # ============================================================================== @@ -212,9 +202,6 @@ ATCR_BACKFILL_INTERVAL=1h # Override service name (defaults to APPVIEW_DOMAIN) # ATCR_SERVICE_NAME=atcr.io -# Debug listen address (optional - for pprof debugging) -# ATCR_DEBUG_ADDR=:5001 - # ============================================================================== # CHECKLIST # ============================================================================== diff --git a/deploy/docker-compose.prod.yml b/deploy/docker-compose.prod.yml index 43797fc..7fc6da6 100644 --- a/deploy/docker-compose.prod.yml +++ b/deploy/docker-compose.prod.yml @@ -59,7 +59,6 @@ services: ATCR_TOKEN_EXPIRATION: ${ATCR_TOKEN_EXPIRATION:-300} # UI configuration - ATCR_UI_ENABLED: ${ATCR_UI_ENABLED:-true} ATCR_UI_DATABASE_PATH: /var/lib/atcr/ui.db # Logging diff --git a/docker-compose.yml b/docker-compose.yml index 505f42d..e1ecbaa 100644 --- a/docker-compose.yml +++ b/docker-compose.yml @@ -14,13 +14,21 @@ services: # Server configuration ATCR_HTTP_ADDR: :5000 ATCR_DEFAULT_HOLD_DID: did:web:172.28.0.3:8080 - # UI configuration - ATCR_UI_ENABLED: "true" ATCR_BACKFILL_ENABLED: "true" # Test mode - fallback to default hold when user's hold is unreachable TEST_MODE: "true" # Logging ATCR_LOG_LEVEL: debug + # Log shipping (uncomment to enable) + ATCR_LOG_SHIPPER_BACKEND: victoria + ATCR_LOG_SHIPPER_URL: http://172.28.0.10:9428 + # Limit local Docker logs - real logs go to Victoria Logs + # Local logs just for live tailing (docker logs -f) + logging: + driver: json-file + options: + max-size: "10m" + max-file: "1" volumes: # Mount source code for Air hot reload - .:/app @@ -56,7 +64,17 @@ services: # DISABLE_PRESIGNED_URLS: true # Logging ATCR_LOG_LEVEL: debug + # Log shipping (uncomment to enable) + ATCR_LOG_SHIPPER_BACKEND: victoria + ATCR_LOG_SHIPPER_URL: http://172.28.0.10:9428 # Storage config comes from env_file (STORAGE_DRIVER, AWS_*, S3_*) + # Limit local Docker logs - real logs go to Victoria Logs + # Local logs just for live tailing (docker logs -f) + logging: + driver: json-file + options: + max-size: "10m" + max-file: "1" build: context: . dockerfile: Dockerfile.dev @@ -82,6 +100,23 @@ services: atcr-network: ipv4_address: 172.28.0.3 + # Victoria Logs for centralized log storage + # Uncomment to enable, then set ATCR_LOG_SHIPPER_* env vars above + victorialogs: + image: victoriametrics/victoria-logs:latest + container_name: victorialogs + ports: + - "9428:9428" + volumes: + - victorialogs-data:/victoria-logs-data + command: + - "-storageDataPath=/victoria-logs-data" + - "-retentionPeriod=7d" + restart: unless-stopped + networks: + atcr-network: + ipv4_address: 172.28.0.10 + networks: atcr-network: driver: bridge @@ -94,3 +129,4 @@ volumes: atcr-auth: atcr-ui: go-mod-cache: + victorialogs-data: diff --git a/docs/DEVELOPMENT.md b/docs/DEVELOPMENT.md index 0ef3869..f4a7ea7 100644 --- a/docs/DEVELOPMENT.md +++ b/docs/DEVELOPMENT.md @@ -165,9 +165,6 @@ services: # Auth ATCR_AUTH_KEY_PATH: "/var/lib/atcr/auth/private-key.pem" - # UI - ATCR_UI_ENABLED: "true" - # Jetstream (optional) # JETSTREAM_URL: "wss://jetstream2.us-east.bsky.network/subscribe" # ATCR_BACKFILL_ENABLED: "false" @@ -524,7 +521,6 @@ export ATCR_BASE_URL=http://localhost:5000 export ATCR_DEFAULT_HOLD_DID=did:web:hold01.atcr.io export ATCR_UI_DATABASE_PATH=/tmp/atcr-ui.db export ATCR_AUTH_KEY_PATH=/tmp/atcr-auth-key.pem -export ATCR_UI_ENABLED=true # Or use .env file source .env.appview diff --git a/docs/appview.md b/docs/appview.md index e4830a0..d2ea1de 100644 --- a/docs/appview.md +++ b/docs/appview.md @@ -115,11 +115,6 @@ Or via Docker Compose (recommended). - **Description:** Service name used for JWT `service` and `issuer` fields. Controls token scope. - **Example:** `atcr.io`, `registry.example.com` -#### `ATCR_DEBUG_ADDR` -- **Default:** `:5001` -- **Description:** Debug listen address for pprof debugging endpoints -- **Example:** `:5001`, `:6060` - ### Storage Configuration #### `ATCR_DEFAULT_HOLD_DID` ⚠️ REQUIRED @@ -150,11 +145,6 @@ Or via Docker Compose (recommended). ### Web UI Configuration -#### `ATCR_UI_ENABLED` -- **Default:** `true` -- **Description:** Enable the web interface. Set to `false` to run registry API only (no web UI, no database). -- **Use case:** API-only deployments where you don't need the browsing interface - #### `ATCR_UI_DATABASE_PATH` - **Default:** `/var/lib/atcr/ui.db` - **Description:** SQLite database path for UI data (OAuth sessions, stars, pull counts, repository metadata) @@ -245,7 +235,6 @@ Open to all ATProto users: # AppView config ATCR_BASE_URL=https://registry.example.com ATCR_DEFAULT_HOLD_DID=did:web:hold01.example.com -ATCR_UI_ENABLED=true ATCR_BACKFILL_ENABLED=true # Hold config (linked hold service) @@ -261,7 +250,6 @@ Restricted to crew members only: # AppView config ATCR_BASE_URL=https://registry.internal.example.com ATCR_DEFAULT_HOLD_DID=did:web:hold.internal.example.com -ATCR_UI_ENABLED=true # Hold config (linked hold service) HOLD_PUBLIC=false # Require auth for pulls diff --git a/pkg/appview/config.go b/pkg/appview/config.go index f26d628..3f8bb85 100644 --- a/pkg/appview/config.go +++ b/pkg/appview/config.go @@ -13,7 +13,6 @@ import ( "net/url" "os" "strconv" - "strings" "time" "github.com/distribution/distribution/v3/configuration" @@ -23,6 +22,7 @@ import ( type Config struct { Version string `yaml:"version"` LogLevel string `yaml:"log_level"` + LogShipper LogShipperConfig `yaml:"log_shipper"` Server ServerConfig `yaml:"server"` UI UIConfig `yaml:"ui"` Health HealthConfig `yaml:"health"` @@ -32,6 +32,28 @@ type Config struct { Distribution *configuration.Configuration `yaml:"-"` // Wrapped distribution config for compatibility } +// LogShipperConfig defines remote log shipping settings +type LogShipperConfig struct { + // Backend selects the log shipping backend (from env: ATCR_LOG_SHIPPER_BACKEND) + // Valid values: "victoria", "opensearch", "loki", or empty to disable + Backend string `yaml:"backend"` + + // URL is the remote log service endpoint (from env: ATCR_LOG_SHIPPER_URL) + URL string `yaml:"url"` + + // BatchSize is the number of logs to batch before flushing (from env: ATCR_LOG_SHIPPER_BATCH_SIZE, default: 100) + BatchSize int `yaml:"batch_size"` + + // FlushInterval is the max time between flushes (from env: ATCR_LOG_SHIPPER_FLUSH_INTERVAL, default: 5s) + FlushInterval time.Duration `yaml:"flush_interval"` + + // Username for basic auth (from env: ATCR_LOG_SHIPPER_USERNAME, optional) + Username string `yaml:"username"` + + // Password for basic auth (from env: ATCR_LOG_SHIPPER_PASSWORD, optional) + Password string `yaml:"password"` +} + // ServerConfig defines server settings type ServerConfig struct { // Addr is the HTTP listen address (from env: ATCR_HTTP_ADDR, default: ":5000") @@ -48,9 +70,6 @@ type ServerConfig struct { // TestMode enables HTTP for local DID resolution and transition:generic scope (from env: TEST_MODE) TestMode bool `yaml:"test_mode"` - // DebugAddr is the debug/pprof HTTP listen address (from env: ATCR_DEBUG_ADDR, default: ":5001") - DebugAddr string `yaml:"debug_addr"` - // OAuthKeyPath is the path to the OAuth client P-256 signing key (from env: ATCR_OAUTH_KEY_PATH, default: "/var/lib/atcr/oauth/client.key") // Auto-generated on first run for production (non-localhost) deployments OAuthKeyPath string `yaml:"oauth_key_path"` @@ -62,14 +81,8 @@ type ServerConfig struct { // UIConfig defines web UI settings type UIConfig struct { - // Enabled controls whether the web UI is enabled (from env: ATCR_UI_ENABLED, default: true) - Enabled bool `yaml:"enabled"` - // DatabasePath is the path to the UI SQLite database (from env: ATCR_UI_DATABASE_PATH, default: "/var/lib/atcr/ui.db") DatabasePath string `yaml:"database_path"` - - // SkipDBMigrations controls whether to skip running database migrations (from env: SKIP_DB_MIGRATIONS, default: false) - SkipDBMigrations bool `yaml:"skip_db_migrations"` } // HealthConfig defines health check and cache settings @@ -112,19 +125,11 @@ type AuthConfig struct { ServiceName string `yaml:"service_name"` } -// CredentialHelperConfig defines credential helper version and download settings +// CredentialHelperConfig defines credential helper download settings type CredentialHelperConfig struct { - // Version is the latest credential helper version (from env: ATCR_CREDENTIAL_HELPER_VERSION) - // e.g., "v0.0.2" - Version string `yaml:"version"` - - // TangledRepo is the Tangled repository URL for downloads (from env: ATCR_CREDENTIAL_HELPER_TANGLED_REPO) - // Default: "https://tangled.org/@evan.jarrett.net/at-container-registry" + // TangledRepo is the Tangled repository URL for downloads + // Hardcoded default: "https://tangled.org/@evan.jarrett.net/at-container-registry" TangledRepo string `yaml:"tangled_repo"` - - // Checksums is a comma-separated list of platform:sha256 pairs (from env: ATCR_CREDENTIAL_HELPER_CHECKSUMS) - // e.g., "linux_amd64:abc123,darwin_arm64:def456" - Checksums map[string]string `yaml:"-"` } // LoadConfigFromEnv builds a complete configuration from environment variables @@ -137,9 +142,16 @@ func LoadConfigFromEnv() (*Config, error) { // Logging configuration cfg.LogLevel = getEnvOrDefault("ATCR_LOG_LEVEL", "info") + // Log shipper configuration + cfg.LogShipper.Backend = os.Getenv("ATCR_LOG_SHIPPER_BACKEND") + cfg.LogShipper.URL = os.Getenv("ATCR_LOG_SHIPPER_URL") + cfg.LogShipper.BatchSize = getIntOrDefault("ATCR_LOG_SHIPPER_BATCH_SIZE", 100) + cfg.LogShipper.FlushInterval = getDurationOrDefault("ATCR_LOG_SHIPPER_FLUSH_INTERVAL", 5*time.Second) + cfg.LogShipper.Username = os.Getenv("ATCR_LOG_SHIPPER_USERNAME") + cfg.LogShipper.Password = os.Getenv("ATCR_LOG_SHIPPER_PASSWORD") + // Server configuration cfg.Server.Addr = getEnvOrDefault("ATCR_HTTP_ADDR", ":5000") - cfg.Server.DebugAddr = getEnvOrDefault("ATCR_DEBUG_ADDR", ":5001") cfg.Server.DefaultHoldDID = os.Getenv("ATCR_DEFAULT_HOLD_DID") if cfg.Server.DefaultHoldDID == "" { return nil, fmt.Errorf("ATCR_DEFAULT_HOLD_DID is required") @@ -155,9 +167,7 @@ func LoadConfigFromEnv() (*Config, error) { } // UI configuration - cfg.UI.Enabled = os.Getenv("ATCR_UI_ENABLED") != "false" cfg.UI.DatabasePath = getEnvOrDefault("ATCR_UI_DATABASE_PATH", "/var/lib/atcr/ui.db") - cfg.UI.SkipDBMigrations = os.Getenv("SKIP_DB_MIGRATIONS") == "true" // Health and cache configuration cfg.Health.CacheTTL = getDurationOrDefault("ATCR_HEALTH_CACHE_TTL", 15*time.Minute) @@ -184,10 +194,8 @@ func LoadConfigFromEnv() (*Config, error) { // Derive service name from base URL or env var (used for JWT issuer and service) cfg.Auth.ServiceName = getServiceName(cfg.Server.BaseURL) - // Credential helper configuration - cfg.CredentialHelper.Version = os.Getenv("ATCR_CREDENTIAL_HELPER_VERSION") - cfg.CredentialHelper.TangledRepo = getEnvOrDefault("ATCR_CREDENTIAL_HELPER_TANGLED_REPO", "https://tangled.org/@evan.jarrett.net/at-container-registry") - cfg.CredentialHelper.Checksums = parseChecksums(os.Getenv("ATCR_CREDENTIAL_HELPER_CHECKSUMS")) + // Credential helper configuration (hardcoded - no env vars needed) + cfg.CredentialHelper.TangledRepo = "https://tangled.org/@evan.jarrett.net/at-container-registry" // Build distribution configuration for compatibility with distribution library distConfig, err := buildDistributionConfig(cfg) @@ -233,9 +241,6 @@ func buildDistributionConfig(cfg *Config) (*configuration.Configuration, error) Headers: map[string][]string{ "X-Content-Type-Options": {"nosniff"}, }, - Debug: configuration.Debug{ - Addr: cfg.Server.DebugAddr, - }, } // Storage (fake in-memory placeholder - all real storage is proxied) @@ -380,23 +385,19 @@ func getDurationOrDefault(envKey string, defaultValue time.Duration) time.Durati return parsed } -// parseChecksums parses a comma-separated list of platform:sha256 pairs -// e.g., "linux_amd64:abc123,darwin_arm64:def456" -func parseChecksums(checksumsStr string) map[string]string { - checksums := make(map[string]string) - if checksumsStr == "" { - return checksums +// getIntOrDefault parses an int from environment variable or returns default +// Logs a warning if parsing fails +func getIntOrDefault(envKey string, defaultValue int) int { + envVal := os.Getenv(envKey) + if envVal == "" { + return defaultValue } - for pair := range strings.SplitSeq(checksumsStr, ",") { - parts := strings.SplitN(strings.TrimSpace(pair), ":", 2) - if len(parts) == 2 { - platform := strings.TrimSpace(parts[0]) - hash := strings.TrimSpace(parts[1]) - if platform != "" && hash != "" { - checksums[platform] = hash - } - } + parsed, err := strconv.Atoi(envVal) + if err != nil { + slog.Warn("Invalid int, using default", "env_key", envKey, "env_value", envVal, "default", defaultValue) + return defaultValue } - return checksums + + return parsed } diff --git a/pkg/appview/db/annotations_test.go b/pkg/appview/db/annotations_test.go index 2a73897..00e97d6 100644 --- a/pkg/appview/db/annotations_test.go +++ b/pkg/appview/db/annotations_test.go @@ -21,7 +21,7 @@ func TestAnnotations_Placeholder(t *testing.T) { func setupAnnotationsTestDB(t *testing.T) *sql.DB { t.Helper() // Use file::memory: with cache=shared to ensure all connections share the same in-memory DB - db, err := InitDB("file::memory:?cache=shared", true) + db, err := InitDB("file::memory:?cache=shared") if err != nil { t.Fatalf("Failed to initialize test database: %v", err) } diff --git a/pkg/appview/db/device_store_test.go b/pkg/appview/db/device_store_test.go index cdd5242..005b325 100644 --- a/pkg/appview/db/device_store_test.go +++ b/pkg/appview/db/device_store_test.go @@ -14,7 +14,7 @@ func setupTestDB(t *testing.T) *DeviceStore { t.Helper() // Use file::memory: with cache=shared to ensure all connections share the same in-memory DB // This prevents race conditions where different connections see different databases - db, err := InitDB("file::memory:?cache=shared", true) + db, err := InitDB("file::memory:?cache=shared") if err != nil { t.Fatalf("Failed to initialize test database: %v", err) } diff --git a/pkg/appview/db/hold_store_test.go b/pkg/appview/db/hold_store_test.go index e1955cc..6216208 100644 --- a/pkg/appview/db/hold_store_test.go +++ b/pkg/appview/db/hold_store_test.go @@ -81,7 +81,7 @@ func TestNullString(t *testing.T) { func setupHoldTestDB(t *testing.T) *sql.DB { t.Helper() // Use file::memory: with cache=shared to ensure all connections share the same in-memory DB - db, err := InitDB("file::memory:?cache=shared", true) + db, err := InitDB("file::memory:?cache=shared") if err != nil { t.Fatalf("Failed to initialize test database: %v", err) } diff --git a/pkg/appview/db/oauth_store_test.go b/pkg/appview/db/oauth_store_test.go index 783cb28..6a1ae56 100644 --- a/pkg/appview/db/oauth_store_test.go +++ b/pkg/appview/db/oauth_store_test.go @@ -12,7 +12,7 @@ import ( func TestInvalidateSessionsWithMismatchedScopes(t *testing.T) { // Create in-memory test database - db, err := InitDB(":memory:", true) + db, err := InitDB(":memory:") if err != nil { t.Fatalf("Failed to init database: %v", err) } @@ -232,7 +232,7 @@ func TestScopesMatch(t *testing.T) { func TestOAuthStoreSessionLifecycle(t *testing.T) { // Basic test to ensure SaveSession, GetSession, DeleteSession work correctly - db, err := InitDB(":memory:", true) + db, err := InitDB(":memory:") if err != nil { t.Fatalf("Failed to init database: %v", err) } @@ -304,7 +304,7 @@ func TestOAuthStoreSessionLifecycle(t *testing.T) { } func TestCleanupOldSessions(t *testing.T) { - db, err := InitDB(":memory:", true) + db, err := InitDB(":memory:") if err != nil { t.Fatalf("Failed to init database: %v", err) } diff --git a/pkg/appview/db/queries_test.go b/pkg/appview/db/queries_test.go index d329972..03aa879 100644 --- a/pkg/appview/db/queries_test.go +++ b/pkg/appview/db/queries_test.go @@ -7,7 +7,7 @@ import ( func TestGetRepositoryMetadata(t *testing.T) { // Create in-memory test database - db, err := InitDB(":memory:", true) + db, err := InitDB(":memory:") if err != nil { t.Fatalf("Failed to init database: %v", err) } @@ -143,7 +143,7 @@ func TestGetRepositoryMetadata(t *testing.T) { func TestInsertManifest(t *testing.T) { // Create in-memory test database - db, err := InitDB(":memory:", true) + db, err := InitDB(":memory:") if err != nil { t.Fatalf("Failed to init database: %v", err) } @@ -320,7 +320,7 @@ func TestInsertManifest(t *testing.T) { func TestUserManagement(t *testing.T) { // Create in-memory test database - db, err := InitDB(":memory:", true) + db, err := InitDB(":memory:") if err != nil { t.Fatalf("Failed to init database: %v", err) } @@ -432,7 +432,7 @@ func TestUserManagement(t *testing.T) { func TestManifestOperations(t *testing.T) { // Create in-memory test database - db, err := InitDB(":memory:", true) + db, err := InitDB(":memory:") if err != nil { t.Fatalf("Failed to init database: %v", err) } @@ -609,7 +609,7 @@ func TestManifestOperations(t *testing.T) { func TestIsManifestTagged(t *testing.T) { // Create in-memory test database - db, err := InitDB(":memory:", true) + db, err := InitDB(":memory:") if err != nil { t.Fatalf("Failed to init database: %v", err) } @@ -675,7 +675,7 @@ func TestIsManifestTagged(t *testing.T) { func TestTagOperations(t *testing.T) { // Create in-memory test database - db, err := InitDB(":memory:", true) + db, err := InitDB(":memory:") if err != nil { t.Fatalf("Failed to init database: %v", err) } @@ -838,7 +838,7 @@ func TestTagOperations(t *testing.T) { func TestGetTagsWithPlatforms(t *testing.T) { // Create in-memory test database - db, err := InitDB(":memory:", true) + db, err := InitDB(":memory:") if err != nil { t.Fatalf("Failed to init database: %v", err) } @@ -980,7 +980,7 @@ func TestGetTagsWithPlatforms(t *testing.T) { func TestUpdateUserHandle(t *testing.T) { // Create in-memory test database - db, err := InitDB(":memory:", true) + db, err := InitDB(":memory:") if err != nil { t.Fatalf("Failed to init database: %v", err) } @@ -1201,7 +1201,7 @@ func TestParseTimestamp(t *testing.T) { } func TestDeleteUserData(t *testing.T) { - db, err := InitDB(":memory:", true) + db, err := InitDB(":memory:") if err != nil { t.Fatalf("Failed to init database: %v", err) } diff --git a/pkg/appview/db/readonly.go b/pkg/appview/db/readonly.go index 7e208dc..a86c0f0 100644 --- a/pkg/appview/db/readonly.go +++ b/pkg/appview/db/readonly.go @@ -57,11 +57,7 @@ func init() { // InitializeDatabase initializes the SQLite database and session store // Returns: (read-write DB, read-only DB, session store) -func InitializeDatabase(uiEnabled bool, dbPath string, skipMigrations bool) (*sql.DB, *sql.DB, *SessionStore) { - if !uiEnabled { - return nil, nil, nil - } - +func InitializeDatabase(dbPath string) (*sql.DB, *sql.DB, *SessionStore) { // Ensure directory exists dbDir := filepath.Dir(dbPath) if err := os.MkdirAll(dbDir, 0700); err != nil { @@ -70,7 +66,7 @@ func InitializeDatabase(uiEnabled bool, dbPath string, skipMigrations bool) (*sq } // Initialize read-write database (for writes and auth operations) - database, err := InitDB(dbPath, skipMigrations) + database, err := InitDB(dbPath) if err != nil { slog.Warn("Failed to initialize UI database", "error", err) return nil, nil, nil diff --git a/pkg/appview/db/readonly_test.go b/pkg/appview/db/readonly_test.go index 2282e97..cbb43e1 100644 --- a/pkg/appview/db/readonly_test.go +++ b/pkg/appview/db/readonly_test.go @@ -19,7 +19,7 @@ func TestAuthorizerBlocksSensitiveTables(t *testing.T) { defer os.Unsetenv("ATCR_UI_DATABASE_PATH") // Initialize database (creates schema) - database, err := InitDB(dbPath, true) + database, err := InitDB(dbPath) if err != nil { t.Fatalf("Failed to initialize database: %v", err) } diff --git a/pkg/appview/db/schema.go b/pkg/appview/db/schema.go index bc3f92a..614cbf4 100644 --- a/pkg/appview/db/schema.go +++ b/pkg/appview/db/schema.go @@ -26,7 +26,7 @@ var migrationsFS embed.FS var schemaSQL string // InitDB initializes the SQLite database with the schema -func InitDB(path string, skipMigrations bool) (*sql.DB, error) { +func InitDB(path string) (*sql.DB, error) { db, err := sql.Open("sqlite3", path) if err != nil { return nil, err @@ -54,12 +54,10 @@ func InitDB(path string, skipMigrations bool) (*sql.DB, error) { } } - // Run migrations unless skipped + // Run migrations // For fresh databases, migrations are recorded but not executed (schema.sql is already complete) - if !skipMigrations { - if err := runMigrations(db, !isExisting); err != nil { - return nil, err - } + if err := runMigrations(db, !isExisting); err != nil { + return nil, err } return db, nil diff --git a/pkg/appview/db/session_store_test.go b/pkg/appview/db/session_store_test.go index 41c6d40..3d3448a 100644 --- a/pkg/appview/db/session_store_test.go +++ b/pkg/appview/db/session_store_test.go @@ -13,7 +13,7 @@ import ( func setupSessionTestDB(t *testing.T) *SessionStore { t.Helper() // Use file::memory: with cache=shared to ensure all connections share the same in-memory DB - db, err := InitDB("file::memory:?cache=shared", true) + db, err := InitDB("file::memory:?cache=shared") if err != nil { t.Fatalf("Failed to initialize test database: %v", err) } diff --git a/pkg/appview/db/tag_delete_test.go b/pkg/appview/db/tag_delete_test.go index 88a6317..28892c0 100644 --- a/pkg/appview/db/tag_delete_test.go +++ b/pkg/appview/db/tag_delete_test.go @@ -11,7 +11,7 @@ import ( // This simulates what Jetstream does: encode repo/tag to rkey, then decode and delete func TestTagDeleteRoundTrip(t *testing.T) { // Create in-memory test database - db, err := InitDB(":memory:", true) + db, err := InitDB(":memory:") if err != nil { t.Fatalf("Failed to init database: %v", err) } diff --git a/pkg/appview/handlers/api.go b/pkg/appview/handlers/api.go index 70c6e8d..9ac6197 100644 --- a/pkg/appview/handlers/api.go +++ b/pkg/appview/handlers/api.go @@ -6,7 +6,6 @@ import ( "fmt" "log/slog" "net/http" - "strings" "atcr.io/pkg/appview/db" "atcr.io/pkg/appview/middleware" @@ -246,48 +245,19 @@ type CredentialHelperVersionResponse struct { } // CredentialHelperVersionHandler returns the latest credential helper version info +// Note: Version info is fetched dynamically from TangledRepo's releases type CredentialHelperVersionHandler struct { - Version string TangledRepo string - Checksums map[string]string -} - -// Supported platforms for download URLs -var credentialHelperPlatforms = []struct { - key string // API key (e.g., "linux_amd64") - os string // OS name in archive (e.g., "Linux") - arch string // Arch name in archive (e.g., "x86_64") - ext string // Archive extension (e.g., "tar.gz" or "zip") -}{ - {"linux_amd64", "Linux", "x86_64", "tar.gz"}, - {"linux_arm64", "Linux", "arm64", "tar.gz"}, - {"darwin_amd64", "Darwin", "x86_64", "tar.gz"}, - {"darwin_arm64", "Darwin", "arm64", "tar.gz"}, - {"windows_amd64", "Windows", "x86_64", "zip"}, - {"windows_arm64", "Windows", "arm64", "zip"}, } func (h *CredentialHelperVersionHandler) ServeHTTP(w http.ResponseWriter, r *http.Request) { - // Check if version is configured - if h.Version == "" { - http.Error(w, "Credential helper version not configured", http.StatusServiceUnavailable) - return - } - - // Build download URLs for all platforms - // URL format: {TangledRepo}/tags/{version}/download/docker-credential-atcr_{version_without_v}_{OS}_{Arch}.{ext} - downloadURLs := make(map[string]string) - versionWithoutV := strings.TrimPrefix(h.Version, "v") - - for _, p := range credentialHelperPlatforms { - filename := fmt.Sprintf("docker-credential-atcr_%s_%s_%s.%s", versionWithoutV, p.os, p.arch, p.ext) - downloadURLs[p.key] = fmt.Sprintf("%s/tags/%s/download/%s", h.TangledRepo, h.Version, filename) - } - + // This endpoint directs users to the Tangled repository for downloads + // Version info should be fetched from the repository's releases page response := CredentialHelperVersionResponse{ - Latest: h.Version, - DownloadURLs: downloadURLs, - Checksums: h.Checksums, + Latest: "", + DownloadURLs: map[string]string{"tangled_repo": h.TangledRepo}, + Checksums: nil, + ReleaseNotes: "Visit the Tangled repository for the latest releases: " + h.TangledRepo, } render.SetContentType(render.ContentTypeJSON) diff --git a/pkg/appview/handlers/device_test.go b/pkg/appview/handlers/device_test.go index 87a9f70..9ac02d0 100644 --- a/pkg/appview/handlers/device_test.go +++ b/pkg/appview/handlers/device_test.go @@ -18,7 +18,7 @@ import ( // setupTestDB creates an in-memory SQLite database with full schema for testing func setupTestDB(t *testing.T) *sql.DB { - database, err := db.InitDB(":memory:", true) + database, err := db.InitDB(":memory:") if err != nil { t.Fatalf("Failed to initialize test database: %v", err) } diff --git a/pkg/appview/middleware/auth_test.go b/pkg/appview/middleware/auth_test.go index 3704bc6..94623ec 100644 --- a/pkg/appview/middleware/auth_test.go +++ b/pkg/appview/middleware/auth_test.go @@ -26,7 +26,7 @@ func TestGetUser_NoContext(t *testing.T) { // setupTestDB creates an in-memory SQLite database for testing func setupTestDB(t *testing.T) *sql.DB { - database, err := db.InitDB(":memory:", true) + database, err := db.InitDB(":memory:") require.NoError(t, err) t.Cleanup(func() { @@ -307,7 +307,7 @@ func TestOptionalAuth_InvalidSession(t *testing.T) { func TestMiddleware_ConcurrentAccess(t *testing.T) { // Use a shared in-memory database for concurrent access // (SQLite's default :memory: creates separate DBs per connection) - database, err := db.InitDB("file::memory:?cache=shared", true) + database, err := db.InitDB("file::memory:?cache=shared") require.NoError(t, err) t.Cleanup(func() { database.Close() diff --git a/pkg/auth/hold_remote_test.go b/pkg/auth/hold_remote_test.go index 99aa141..3d9820b 100644 --- a/pkg/auth/hold_remote_test.go +++ b/pkg/auth/hold_remote_test.go @@ -34,7 +34,7 @@ func TestNewRemoteHoldAuthorizer_TestMode(t *testing.T) { // setupTestDB creates an in-memory database for testing func setupTestDB(t *testing.T) *sql.DB { - testDB, err := db.InitDB(":memory:", true) + testDB, err := db.InitDB(":memory:") if err != nil { t.Fatalf("Failed to initialize test database: %v", err) } diff --git a/pkg/auth/token/handler_test.go b/pkg/auth/token/handler_test.go index da5ea1e..9927e43 100644 --- a/pkg/auth/token/handler_test.go +++ b/pkg/auth/token/handler_test.go @@ -52,7 +52,7 @@ func getSharedTestKey(t *testing.T) string { // setupTestDeviceStore creates an in-memory SQLite database for testing func setupTestDeviceStore(t *testing.T) (*db.DeviceStore, *sql.DB) { - testDB, err := db.InitDB(":memory:", true) + testDB, err := db.InitDB(":memory:") if err != nil { t.Fatalf("Failed to initialize test database: %v", err) } diff --git a/pkg/hold/config.go b/pkg/hold/config.go index ae5d98f..7143a0c 100644 --- a/pkg/hold/config.go +++ b/pkg/hold/config.go @@ -24,6 +24,7 @@ import ( type Config struct { Version string `yaml:"version"` LogLevel string `yaml:"log_level"` + LogShipper LogShipperConfig `yaml:"log_shipper"` Storage StorageConfig `yaml:"storage"` Server ServerConfig `yaml:"server"` Registration RegistrationConfig `yaml:"registration"` @@ -31,6 +32,28 @@ type Config struct { Admin AdminConfig `yaml:"admin"` } +// LogShipperConfig defines remote log shipping settings +type LogShipperConfig struct { + // Backend selects the log shipping backend (from env: ATCR_LOG_SHIPPER_BACKEND) + // Valid values: "victoria", "opensearch", "loki", or empty to disable + Backend string `yaml:"backend"` + + // URL is the remote log service endpoint (from env: ATCR_LOG_SHIPPER_URL) + URL string `yaml:"url"` + + // BatchSize is the number of logs to batch before flushing (from env: ATCR_LOG_SHIPPER_BATCH_SIZE, default: 100) + BatchSize int `yaml:"batch_size"` + + // FlushInterval is the max time between flushes (from env: ATCR_LOG_SHIPPER_FLUSH_INTERVAL, default: 5s) + FlushInterval time.Duration `yaml:"flush_interval"` + + // Username for basic auth (from env: ATCR_LOG_SHIPPER_USERNAME, optional) + Username string `yaml:"username"` + + // Password for basic auth (from env: ATCR_LOG_SHIPPER_PASSWORD, optional) + Password string `yaml:"password"` +} + // AdminConfig defines admin panel settings type AdminConfig struct { // Enabled controls whether the admin panel is accessible (from env: HOLD_ADMIN_ENABLED) @@ -114,6 +137,14 @@ func LoadConfigFromEnv() (*Config, error) { // Logging configuration cfg.LogLevel = getEnvOrDefault("ATCR_LOG_LEVEL", "info") + // Log shipper configuration + cfg.LogShipper.Backend = os.Getenv("ATCR_LOG_SHIPPER_BACKEND") + cfg.LogShipper.URL = os.Getenv("ATCR_LOG_SHIPPER_URL") + cfg.LogShipper.BatchSize = getIntOrDefault("ATCR_LOG_SHIPPER_BATCH_SIZE", 100) + cfg.LogShipper.FlushInterval = getDurationOrDefault("ATCR_LOG_SHIPPER_FLUSH_INTERVAL", 5*time.Second) + cfg.LogShipper.Username = os.Getenv("ATCR_LOG_SHIPPER_USERNAME") + cfg.LogShipper.Password = os.Getenv("ATCR_LOG_SHIPPER_PASSWORD") + // Server configuration cfg.Server.Addr = getEnvOrDefault("HOLD_SERVER_ADDR", ":8080") cfg.Server.PublicURL = os.Getenv("HOLD_PUBLIC_URL") @@ -217,6 +248,38 @@ func getEnvOrDefault(key, defaultValue string) string { return defaultValue } +// getIntOrDefault parses an int from environment variable or returns default +func getIntOrDefault(envKey string, defaultValue int) int { + envVal := os.Getenv(envKey) + if envVal == "" { + return defaultValue + } + + var parsed int + if _, err := fmt.Sscanf(envVal, "%d", &parsed); err != nil { + slog.Warn("Invalid int, using default", "env_key", envKey, "env_value", envVal, "default", defaultValue) + return defaultValue + } + + return parsed +} + +// getDurationOrDefault parses a duration from environment variable or returns default +func getDurationOrDefault(envKey string, defaultValue time.Duration) time.Duration { + envVal := os.Getenv(envKey) + if envVal == "" { + return defaultValue + } + + parsed, err := time.ParseDuration(envVal) + if err != nil { + slog.Warn("Invalid duration, using default", "env_key", envKey, "env_value", envVal, "default", defaultValue) + return defaultValue + } + + return parsed +} + // RequestCrawl sends a crawl request to the ATProto relay for the given hostname. // This makes the hold's PDS discoverable by the relay network. func RequestCrawl(relayEndpoint, publicURL string) error { diff --git a/pkg/logging/logger.go b/pkg/logging/logger.go index 4a2ba64..79701dc 100644 --- a/pkg/logging/logger.go +++ b/pkg/logging/logger.go @@ -27,6 +27,9 @@ var ( debugEnabled atomic.Bool revertTimer *time.Timer revertMu sync.Mutex + + // asyncHandler holds the global async handler for shutdown + asyncHandler *AsyncHandler ) // InitLogger initializes the global slog default logger with the specified log level. @@ -36,6 +39,20 @@ var ( // // Also starts a signal handler for SIGUSR1 to toggle debug mode at runtime. func InitLogger(level string) { + InitLoggerWithShipper(level, ShipperConfig{}) +} + +// InitLoggerWithShipper initializes the global slog default logger with the specified +// log level and optional remote log shipping. +// Valid levels: debug, info, warn, error (case-insensitive) +// If level is empty or invalid, defaults to INFO. +// Call this from main() at startup. +// +// If shipperCfg.Backend is non-empty, logs will be shipped to the configured +// remote service in addition to stdout. +// +// Also starts a signal handler for SIGUSR1 to toggle debug mode at runtime. +func InitLoggerWithShipper(level string, shipperCfg ShipperConfig) { var logLevel slog.Level switch strings.ToLower(strings.TrimSpace(level)) { @@ -69,13 +86,36 @@ func InitLogger(level string) { }, } - handler := slog.NewTextHandler(os.Stdout, opts) - slog.SetDefault(slog.New(handler)) + // Create stdout handler + stdoutHandler := slog.NewTextHandler(os.Stdout, opts) + + // Create shipper if configured + var shipper Shipper + if shipperCfg.Backend != "" { + var err error + shipper, err = NewShipper(shipperCfg) + if err != nil { + // Log error but continue without shipping + fmt.Fprintf(os.Stderr, "log shipper initialization failed: %v (continuing with stdout only)\n", err) + } + } + + // Create async handler (wraps stdout + optional shipper) + asyncHandler = NewAsyncHandler(stdoutHandler, shipper, shipperCfg, opts) + slog.SetDefault(slog.New(asyncHandler)) // Start signal handler for dynamic debug toggle go handleDebugSignal() } +// Shutdown flushes any remaining logs and closes the log shipper. +// Call this during graceful shutdown to ensure all logs are delivered. +func Shutdown() { + if asyncHandler != nil { + asyncHandler.Shutdown() + } +} + func handleDebugSignal() { sigChan := make(chan os.Signal, 1) signal.Notify(sigChan, syscall.SIGUSR1) diff --git a/pkg/logging/shipper.go b/pkg/logging/shipper.go new file mode 100644 index 0000000..8699488 --- /dev/null +++ b/pkg/logging/shipper.go @@ -0,0 +1,308 @@ +// Package logging provides centralized structured logging with optional remote log shipping. +package logging + +import ( + "context" + "fmt" + "log/slog" + "sync" + "time" +) + +// Default configuration values +const ( + DefaultBatchSize = 100 + DefaultFlushInterval = 5 * time.Second +) + +// Shipper defines the interface for log shipping backends. +// Implementations should be safe for concurrent use. +type Shipper interface { + // Ship sends a batch of log entries to the remote service. + // Returns an error if the batch could not be shipped. + Ship(ctx context.Context, entries []LogEntry) error + + // Close cleanly shuts down the shipper, releasing any resources. + Close() error +} + +// LogEntry represents a single log entry to be shipped. +type LogEntry struct { + Time time.Time + Level slog.Level + Message string + Source string + Attrs map[string]any +} + +// ShipperConfig configures the log shipper. +type ShipperConfig struct { + // Backend selects the shipping backend: "victoria", "opensearch", "loki", etc. + // Empty string disables remote shipping (stdout only). + Backend string + + // URL is the remote service endpoint URL. + URL string + + // BatchSize is the number of logs to batch before flushing. + // Default: 100 + BatchSize int + + // FlushInterval is the maximum time between flushes. + // Default: 5s + FlushInterval time.Duration + + // Service identifies the source service ("appview" or "hold"). + // Added to all log entries. + Service string + + // Username for basic auth (optional). + Username string + + // Password for basic auth (optional). + Password string +} + +// NewShipper creates a shipper for the configured backend. +// Returns nil if no backend is configured (remote shipping disabled). +func NewShipper(cfg ShipperConfig) (Shipper, error) { + switch cfg.Backend { + case "victoria": + return NewVictoriaShipper(cfg) + case "opensearch": + return nil, fmt.Errorf("opensearch backend not yet implemented") + case "loki": + return nil, fmt.Errorf("loki backend not yet implemented") + case "": + return nil, nil // No remote shipping + default: + return nil, fmt.Errorf("unknown log shipper backend: %s", cfg.Backend) + } +} + +// asyncState holds the shared state for async log shipping. +// This is separate from AsyncHandler to allow WithAttrs/WithGroup +// to create new handlers that share the same batch and flush state. +type asyncState struct { + shipper Shipper + + // Batching + batch []LogEntry + batchMu sync.Mutex + batchSize int + + // Async flush + flushInterval time.Duration + flushCh chan struct{} + doneCh chan struct{} + wg sync.WaitGroup +} + +// AsyncHandler is an slog.Handler that writes to stdout and optionally +// ships logs to a remote service asynchronously. +type AsyncHandler struct { + stdout slog.Handler + opts *slog.HandlerOptions + state *asyncState // Shared state for batching and flushing +} + +// NewAsyncHandler creates a new AsyncHandler that wraps stdout logging +// and optionally ships logs to a remote service. +func NewAsyncHandler(stdout slog.Handler, shipper Shipper, cfg ShipperConfig, opts *slog.HandlerOptions) *AsyncHandler { + batchSize := cfg.BatchSize + if batchSize <= 0 { + batchSize = DefaultBatchSize + } + + flushInterval := cfg.FlushInterval + if flushInterval <= 0 { + flushInterval = DefaultFlushInterval + } + + state := &asyncState{ + shipper: shipper, + batch: make([]LogEntry, 0, batchSize), + batchSize: batchSize, + flushInterval: flushInterval, + flushCh: make(chan struct{}, 1), + doneCh: make(chan struct{}), + } + + h := &AsyncHandler{ + stdout: stdout, + opts: opts, + state: state, + } + + // Start background flusher if shipping is enabled + if shipper != nil { + state.wg.Add(1) + go h.runFlusher() + } + + return h +} + +// Enabled reports whether the handler handles records at the given level. +func (h *AsyncHandler) Enabled(ctx context.Context, level slog.Level) bool { + return h.stdout.Enabled(ctx, level) +} + +// Handle handles the Record by writing to stdout and queuing for remote shipping. +func (h *AsyncHandler) Handle(ctx context.Context, r slog.Record) error { + // Always write to stdout + if err := h.stdout.Handle(ctx, r); err != nil { + return err + } + + // Skip remote shipping if no shipper configured + if h.state.shipper == nil { + return nil + } + + // Build log entry + entry := LogEntry{ + Time: r.Time, + Level: r.Level, + Message: r.Message, + Attrs: make(map[string]any), + } + + r.Attrs(func(a slog.Attr) bool { + if a.Key == slog.SourceKey { + if src, ok := a.Value.Any().(*slog.Source); ok { + entry.Source = shortenSource(src.File, src.Line) + } + } else { + entry.Attrs[a.Key] = resolveAttrValue(a.Value) + } + return true + }) + + // Add to batch + h.state.batchMu.Lock() + h.state.batch = append(h.state.batch, entry) + shouldFlush := len(h.state.batch) >= h.state.batchSize + h.state.batchMu.Unlock() + + if shouldFlush { + h.triggerFlush() + } + + return nil +} + +// WithAttrs returns a new Handler with the given attributes added. +func (h *AsyncHandler) WithAttrs(attrs []slog.Attr) slog.Handler { + return &AsyncHandler{ + stdout: h.stdout.WithAttrs(attrs), + opts: h.opts, + state: h.state, // Share the same state + } +} + +// WithGroup returns a new Handler with the given group name. +func (h *AsyncHandler) WithGroup(name string) slog.Handler { + return &AsyncHandler{ + stdout: h.stdout.WithGroup(name), + opts: h.opts, + state: h.state, // Share the same state + } +} + +// triggerFlush signals the flusher goroutine to flush immediately. +func (h *AsyncHandler) triggerFlush() { + select { + case h.state.flushCh <- struct{}{}: + default: // Flush already pending + } +} + +// runFlusher runs in a goroutine and periodically flushes the batch. +func (h *AsyncHandler) runFlusher() { + defer h.state.wg.Done() + + ticker := time.NewTicker(h.state.flushInterval) + defer ticker.Stop() + + for { + select { + case <-ticker.C: + h.flush() + case <-h.state.flushCh: + h.flush() + case <-h.state.doneCh: + h.flush() // Final flush + return + } + } +} + +// flush sends the current batch to the remote service. +func (h *AsyncHandler) flush() { + h.state.batchMu.Lock() + if len(h.state.batch) == 0 { + h.state.batchMu.Unlock() + return + } + + // Take ownership of the batch + batch := h.state.batch + h.state.batch = make([]LogEntry, 0, h.state.batchSize) + h.state.batchMu.Unlock() + + // Ship with a timeout context + ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second) + defer cancel() + + if err := h.state.shipper.Ship(ctx, batch); err != nil { + // Log to stderr (not through slog to avoid recursion) + fmt.Printf("log shipper error: %v (dropped %d entries)\n", err, len(batch)) + } +} + +// Shutdown flushes any remaining logs and closes the shipper. +// Call this during graceful shutdown. +func (h *AsyncHandler) Shutdown() { + if h.state.shipper == nil { + return + } + + close(h.state.doneCh) + h.state.wg.Wait() + + if err := h.state.shipper.Close(); err != nil { + fmt.Printf("log shipper close error: %v\n", err) + } +} + +// resolveAttrValue converts slog.Value to a plain Go value for JSON encoding. +func resolveAttrValue(v slog.Value) any { + switch v.Kind() { + case slog.KindString: + return v.String() + case slog.KindInt64: + return v.Int64() + case slog.KindUint64: + return v.Uint64() + case slog.KindFloat64: + return v.Float64() + case slog.KindBool: + return v.Bool() + case slog.KindDuration: + return v.Duration().String() + case slog.KindTime: + return v.Time().Format(time.RFC3339Nano) + case slog.KindGroup: + attrs := v.Group() + m := make(map[string]any, len(attrs)) + for _, a := range attrs { + m[a.Key] = resolveAttrValue(a.Value) + } + return m + case slog.KindAny: + return v.Any() + default: + return v.String() + } +} diff --git a/pkg/logging/shipper_test.go b/pkg/logging/shipper_test.go new file mode 100644 index 0000000..6caa171 --- /dev/null +++ b/pkg/logging/shipper_test.go @@ -0,0 +1,386 @@ +package logging + +import ( + "context" + "encoding/json" + "io" + "log/slog" + "net/http" + "net/http/httptest" + "strings" + "sync" + "sync/atomic" + "testing" + "time" +) + +func TestNewShipper(t *testing.T) { + tests := []struct { + name string + cfg ShipperConfig + wantErr bool + errMsg string + }{ + { + name: "empty backend returns nil", + cfg: ShipperConfig{Backend: ""}, + wantErr: false, + }, + { + name: "victoria backend requires URL", + cfg: ShipperConfig{Backend: "victoria", URL: ""}, + wantErr: true, + errMsg: "URL is required", + }, + { + name: "victoria backend with URL succeeds", + cfg: ShipperConfig{Backend: "victoria", URL: "http://localhost:9428"}, + wantErr: false, + }, + { + name: "unknown backend returns error", + cfg: ShipperConfig{Backend: "unknown"}, + wantErr: true, + errMsg: "unknown log shipper backend", + }, + { + name: "opensearch not implemented", + cfg: ShipperConfig{Backend: "opensearch"}, + wantErr: true, + errMsg: "not yet implemented", + }, + { + name: "loki not implemented", + cfg: ShipperConfig{Backend: "loki"}, + wantErr: true, + errMsg: "not yet implemented", + }, + } + + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + shipper, err := NewShipper(tt.cfg) + if tt.wantErr { + if err == nil { + t.Errorf("NewShipper() expected error, got nil") + } else if tt.errMsg != "" && !strings.Contains(err.Error(), tt.errMsg) { + t.Errorf("NewShipper() error = %v, want error containing %q", err, tt.errMsg) + } + return + } + if err != nil { + t.Errorf("NewShipper() unexpected error: %v", err) + } + if tt.cfg.Backend == "" && shipper != nil { + t.Error("NewShipper() with empty backend should return nil shipper") + } + if shipper != nil { + shipper.Close() + } + }) + } +} + +func TestVictoriaShipper_Ship(t *testing.T) { + var receivedLogs []map[string]any + var mu sync.Mutex + + server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + if r.Method != http.MethodPost { + t.Errorf("expected POST, got %s", r.Method) + } + if !strings.Contains(r.URL.Path, "/insert/jsonline") { + t.Errorf("expected /insert/jsonline path, got %s", r.URL.Path) + } + if r.Header.Get("Content-Type") != "application/stream+json" { + t.Errorf("expected application/stream+json content type, got %s", r.Header.Get("Content-Type")) + } + + body, _ := io.ReadAll(r.Body) + lines := strings.Split(strings.TrimSpace(string(body)), "\n") + + mu.Lock() + for _, line := range lines { + var doc map[string]any + if err := json.Unmarshal([]byte(line), &doc); err != nil { + t.Errorf("failed to unmarshal log line: %v", err) + } + receivedLogs = append(receivedLogs, doc) + } + mu.Unlock() + + w.WriteHeader(http.StatusOK) + })) + defer server.Close() + + shipper, err := NewVictoriaShipper(ShipperConfig{ + URL: server.URL, + Service: "test-service", + }) + if err != nil { + t.Fatalf("NewVictoriaShipper() error: %v", err) + } + defer shipper.Close() + + entries := []LogEntry{ + { + Time: time.Date(2024, 1, 8, 12, 0, 0, 0, time.UTC), + Level: slog.LevelInfo, + Message: "test message 1", + Source: "test.go:42", + Attrs: map[string]any{"key1": "value1"}, + }, + { + Time: time.Date(2024, 1, 8, 12, 0, 1, 0, time.UTC), + Level: slog.LevelError, + Message: "test message 2", + Source: "test.go:43", + Attrs: map[string]any{"key2": 123}, + }, + } + + ctx := context.Background() + if err := shipper.Ship(ctx, entries); err != nil { + t.Fatalf("Ship() error: %v", err) + } + + mu.Lock() + defer mu.Unlock() + + if len(receivedLogs) != 2 { + t.Errorf("expected 2 logs, got %d", len(receivedLogs)) + } + + // Check first log + if receivedLogs[0]["_msg"] != "test message 1" { + t.Errorf("expected _msg 'test message 1', got %v", receivedLogs[0]["_msg"]) + } + if receivedLogs[0]["level"] != "INFO" { + t.Errorf("expected level 'INFO', got %v", receivedLogs[0]["level"]) + } + if receivedLogs[0]["service"] != "test-service" { + t.Errorf("expected service 'test-service', got %v", receivedLogs[0]["service"]) + } + if receivedLogs[0]["key1"] != "value1" { + t.Errorf("expected key1 'value1', got %v", receivedLogs[0]["key1"]) + } + + // Check second log + if receivedLogs[1]["level"] != "ERROR" { + t.Errorf("expected level 'ERROR', got %v", receivedLogs[1]["level"]) + } +} + +func TestVictoriaShipper_BasicAuth(t *testing.T) { + var authHeader string + + server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + authHeader = r.Header.Get("Authorization") + w.WriteHeader(http.StatusOK) + })) + defer server.Close() + + shipper, err := NewVictoriaShipper(ShipperConfig{ + URL: server.URL, + Username: "testuser", + Password: "testpass", + }) + if err != nil { + t.Fatalf("NewVictoriaShipper() error: %v", err) + } + defer shipper.Close() + + entries := []LogEntry{{Time: time.Now(), Level: slog.LevelInfo, Message: "test"}} + if err := shipper.Ship(context.Background(), entries); err != nil { + t.Fatalf("Ship() error: %v", err) + } + + if authHeader == "" { + t.Error("expected Authorization header to be set") + } + if !strings.HasPrefix(authHeader, "Basic ") { + t.Errorf("expected Basic auth, got: %s", authHeader) + } +} + +func TestVictoriaShipper_EmptyBatch(t *testing.T) { + var called bool + + server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + called = true + w.WriteHeader(http.StatusOK) + })) + defer server.Close() + + shipper, _ := NewVictoriaShipper(ShipperConfig{URL: server.URL}) + defer shipper.Close() + + if err := shipper.Ship(context.Background(), nil); err != nil { + t.Errorf("Ship() with nil entries should not error: %v", err) + } + if err := shipper.Ship(context.Background(), []LogEntry{}); err != nil { + t.Errorf("Ship() with empty entries should not error: %v", err) + } + if called { + t.Error("Ship() should not make HTTP request for empty batch") + } +} + +func TestVictoriaShipper_ServerError(t *testing.T) { + server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + w.WriteHeader(http.StatusInternalServerError) + w.Write([]byte("internal error")) + })) + defer server.Close() + + shipper, _ := NewVictoriaShipper(ShipperConfig{URL: server.URL}) + defer shipper.Close() + + entries := []LogEntry{{Time: time.Now(), Level: slog.LevelInfo, Message: "test"}} + err := shipper.Ship(context.Background(), entries) + if err == nil { + t.Error("Ship() should return error on server error") + } + if !strings.Contains(err.Error(), "500") { + t.Errorf("error should contain status code, got: %v", err) + } +} + +func TestAsyncHandler_Batching(t *testing.T) { + var shipCount atomic.Int32 + var totalEntries atomic.Int32 + + mockShipper := &mockShipper{ + shipFunc: func(ctx context.Context, entries []LogEntry) error { + shipCount.Add(1) + totalEntries.Add(int32(len(entries))) + return nil + }, + } + + cfg := ShipperConfig{ + BatchSize: 5, + FlushInterval: 100 * time.Millisecond, + } + + stdoutHandler := slog.NewTextHandler(io.Discard, nil) + handler := NewAsyncHandler(stdoutHandler, mockShipper, cfg, nil) + + // Log 12 entries - should trigger 2 batch flushes (at 5 and 10) plus 2 remaining + for i := 0; i < 12; i++ { + record := slog.NewRecord(time.Now(), slog.LevelInfo, "test message", 0) + handler.Handle(context.Background(), record) + } + + // Wait for flush interval to trigger + time.Sleep(200 * time.Millisecond) + + handler.Shutdown() + + if totalEntries.Load() != 12 { + t.Errorf("expected 12 total entries shipped, got %d", totalEntries.Load()) + } +} + +func TestAsyncHandler_Shutdown(t *testing.T) { + var shipped []LogEntry + var mu sync.Mutex + + mockShipper := &mockShipper{ + shipFunc: func(ctx context.Context, entries []LogEntry) error { + mu.Lock() + shipped = append(shipped, entries...) + mu.Unlock() + return nil + }, + } + + cfg := ShipperConfig{ + BatchSize: 100, // Large batch size so nothing flushes immediately + FlushInterval: 10 * time.Second, + } + + stdoutHandler := slog.NewTextHandler(io.Discard, nil) + handler := NewAsyncHandler(stdoutHandler, mockShipper, cfg, nil) + + // Log a few entries + for i := 0; i < 3; i++ { + record := slog.NewRecord(time.Now(), slog.LevelInfo, "test", 0) + handler.Handle(context.Background(), record) + } + + // Entries should be pending (not shipped yet due to large batch size) + mu.Lock() + if len(shipped) != 0 { + t.Errorf("expected 0 shipped before shutdown, got %d", len(shipped)) + } + mu.Unlock() + + // Shutdown should flush pending entries + handler.Shutdown() + + mu.Lock() + if len(shipped) != 3 { + t.Errorf("expected 3 shipped after shutdown, got %d", len(shipped)) + } + mu.Unlock() +} + +func TestAsyncHandler_NoShipper(t *testing.T) { + cfg := ShipperConfig{} + stdoutHandler := slog.NewTextHandler(io.Discard, nil) + handler := NewAsyncHandler(stdoutHandler, nil, cfg, nil) + + // Should not panic with nil shipper + record := slog.NewRecord(time.Now(), slog.LevelInfo, "test", 0) + if err := handler.Handle(context.Background(), record); err != nil { + t.Errorf("Handle() with nil shipper should not error: %v", err) + } + + // Shutdown should not panic + handler.Shutdown() +} + +func TestResolveAttrValue(t *testing.T) { + tests := []struct { + name string + value slog.Value + expected any + }{ + {"string", slog.StringValue("test"), "test"}, + {"int64", slog.Int64Value(42), int64(42)}, + {"uint64", slog.Uint64Value(42), uint64(42)}, + {"float64", slog.Float64Value(3.14), 3.14}, + {"bool", slog.BoolValue(true), true}, + {"duration", slog.DurationValue(5 * time.Second), "5s"}, + } + + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + got := resolveAttrValue(tt.value) + if got != tt.expected { + t.Errorf("resolveAttrValue() = %v, want %v", got, tt.expected) + } + }) + } +} + +// mockShipper implements Shipper for testing +type mockShipper struct { + shipFunc func(ctx context.Context, entries []LogEntry) error + closeFunc func() error +} + +func (m *mockShipper) Ship(ctx context.Context, entries []LogEntry) error { + if m.shipFunc != nil { + return m.shipFunc(ctx, entries) + } + return nil +} + +func (m *mockShipper) Close() error { + if m.closeFunc != nil { + return m.closeFunc() + } + return nil +} diff --git a/pkg/logging/victoria.go b/pkg/logging/victoria.go new file mode 100644 index 0000000..33c10cd --- /dev/null +++ b/pkg/logging/victoria.go @@ -0,0 +1,112 @@ +package logging + +import ( + "bytes" + "context" + "encoding/json" + "fmt" + "io" + "net/http" + "time" +) + +// VictoriaShipper ships logs to Victoria Logs using the native JSON lines endpoint. +type VictoriaShipper struct { + url string + client *http.Client + service string + username string + password string +} + +// NewVictoriaShipper creates a new Victoria Logs shipper. +func NewVictoriaShipper(cfg ShipperConfig) (*VictoriaShipper, error) { + if cfg.URL == "" { + return nil, fmt.Errorf("victoria logs URL is required") + } + + return &VictoriaShipper{ + url: cfg.URL, + service: cfg.Service, + client: &http.Client{ + Timeout: 30 * time.Second, + }, + username: cfg.Username, + password: cfg.Password, + }, nil +} + +// Ship sends a batch of log entries to Victoria Logs. +func (v *VictoriaShipper) Ship(ctx context.Context, entries []LogEntry) error { + if len(entries) == 0 { + return nil + } + + var buf bytes.Buffer + + for _, entry := range entries { + doc := map[string]any{ + // Victoria Logs special fields + "_time": entry.Time.UTC().Format(time.RFC3339Nano), + "_msg": entry.Message, + + // Standard fields + "level": entry.Level.String(), + "source": entry.Source, + } + + // Add service if configured + if v.service != "" { + doc["service"] = v.service + } + + // Add all custom attributes + for k, val := range entry.Attrs { + // Don't overwrite special fields + if k != "_time" && k != "_msg" && k != "level" && k != "source" && k != "service" { + doc[k] = val + } + } + + if err := json.NewEncoder(&buf).Encode(doc); err != nil { + return fmt.Errorf("failed to encode log entry: %w", err) + } + } + + // Use the JSON lines endpoint with stream fields for efficient querying + url := v.url + "/insert/jsonline?_stream_fields=service,level" + + req, err := http.NewRequestWithContext(ctx, http.MethodPost, url, &buf) + if err != nil { + return fmt.Errorf("failed to create request: %w", err) + } + + req.Header.Set("Content-Type", "application/stream+json") + + // Add basic auth if configured + if v.username != "" && v.password != "" { + req.SetBasicAuth(v.username, v.password) + } + + resp, err := v.client.Do(req) + if err != nil { + return fmt.Errorf("failed to send logs: %w", err) + } + defer resp.Body.Close() + + if resp.StatusCode >= 400 { + body, _ := io.ReadAll(io.LimitReader(resp.Body, 1024)) + return fmt.Errorf("victoria logs error %d: %s", resp.StatusCode, string(body)) + } + + // Drain and close body to allow connection reuse + _, _ = io.Copy(io.Discard, resp.Body) + + return nil +} + +// Close releases any resources held by the shipper. +func (v *VictoriaShipper) Close() error { + v.client.CloseIdleConnections() + return nil +}