We Cut Log Costs by 35% Using Vector 0.30 and Loki 3.0: Lessons from a 3-Month Tuning

# Vector 0.30 Configuration for Kubernetes Log Ingestion
# Author: Platform Team, Version: 1.2.0, Last Updated: 2024-09-15
# Components:
# 1. kubernetes_logs: Collect logs from all pods in prod namespace
# 2. filter: Drop low-value debug logs and health check endpoints
# 3. remap: Parse JSON logs, add pod metadata, mask PII
# 4. adaptive_buffer: Handle traffic spikes without OOM
# 5. loki_sink: Send processed logs to Loki 3.0 cluster

[sources.k8s_logs]
type = "kubernetes_logs"
include_namespaces = ["prod", "staging"]
exclude_namespaces = ["kube-system"]
max_line_bytes = 1048576  # 1MB max line size
auto_partial_merge = true
partial_merge_timeout_secs = 30
# Handle pod churn: refresh pod list every 5 seconds
pod_refresh_interval_secs = 5
# Error handling: retry failed pod metadata fetches 3 times
metadata_fetch_retry_limit = 3
metadata_fetch_retry_backoff_secs = 2

[transforms.filter_low_value]
type = "filter"
inputs = ["k8s_logs"]
# Drop debug logs from non-critical services
condition = '!(severity == "DEBUG" && service != "payment-gateway")'
# Drop k8s health check logs
condition = '!(message contains "healthcheck" && pod_labels.app contains "k8s")'

[transforms.parse_and_enrich]
type = "remap"
inputs = ["filter_low_value"]
source = '''
  # Parse JSON-encoded log lines, default to raw text if parse fails
  parsed = parse_json(.message) ?? { "raw_message": .message }
  # Merge parsed fields into root event
  for_each(parsed) { |key, value| .[key] = value }
  # Add pod metadata from Vector's k8s enrichment
  .pod_name = .kubernetes.pod_name
  .namespace = .kubernetes.namespace
  .service = .kubernetes.labels.app ?? "unknown"
  .node_name = .kubernetes.node_name
  # Mask email addresses and credit card numbers (PCI compliance)
  .message = mask(.message, pattern: r"\b[A-Za-z0-9._%+-]+@[A-Za-z0-9.-]+\.[A-Z|a-z]{2,}\b", replace: "[EMAIL_REDACTED]")
  .message = mask(.message, pattern: r"\b\d{4}-\d{4}-\d{4}-\d{4}\b", replace: "[CARD_REDACTED]")
  # Add timestamp in Loki-compatible RFC3339 format
  .timestamp = now() | format_timestamp("%Y-%m-%dT%H:%M:%S%.fZ")
'''

[transforms.adaptive_buffer]
type = "adaptive_buffer"
inputs = ["parse_and_enrich"]
# Buffer up to 10GB on disk during traffic spikes
max_size_bytes = 10737418240
# Flush to sink when buffer reaches 50% capacity or every 5 seconds
flush_when_full = false
flush_interval_secs = 5
# Error handling: retry failed flushes 5 times with exponential backoff
retry_max_attempts = 5
retry_backoff_secs = 2
retry_max_backoff_secs = 30
# Dead letter queue for logs that fail processing after retries
dead_letter_queue.enabled = true
dead_letter_queue.max_size_bytes = 1073741824  # 1GB DLQ
dead_letter_queue.dir = "/var/lib/vector/dlq"

[sinks.loki_cluster]
type = "loki"
inputs = ["adaptive_buffer"]
endpoint = "https://loki.prod.internal:3100"
# Loki 3.0 supports tenant header for multi-tenancy
tenant_id = "prod"
# Batch up to 1MB or 1000 events before sending
batch.max_bytes = 1048576
batch.max_events = 1000
batch.timeout_secs = 3
# Auth: use service account token for Loki auth
auth.strategy = "bearer"
auth.token = "${LOKI_BEARER_TOKEN}"
# Error handling: retry HTTP 429 (rate limit) and 5xx errors
retry_on_status_codes = [429, 500, 502, 503, 504]
retry_max_attempts = 10
retry_backoff_secs = 1
# Label config: map Vector fields to Loki labels (critical for query performance)
labels = { 
  "service" = "service",
  "namespace" = "namespace",
  "pod_name" = "pod_name",
  "severity" = "severity"
}
# Remove fields that are already in labels to reduce payload size
remove_fields = ["service", "namespace", "pod_name", "severity", "kubernetes"]

# Loki 3.0 Configuration for Production Workload
# Author: Platform Team, Version: 3.0.1, Last Updated: 2024-09-10
# Key Changes from Loki 2.9: TSDB index backend, reduced memory overhead

# Auth: Require tenant ID for all requests
auth_enabled: true

# Server config: listen on all interfaces, enable TLS
server:
  http_listen_port: 3100
  grpc_listen_port: 9095
  http_server_read_timeout: 30s
  http_server_write_timeout: 30s
  # TLS config for encryption in transit
  tls_cert_file: /etc/loki/tls/cert.pem
  tls_key_file: /etc/loki/tls/key.pem
  # Error handling: enable request logging for debugging
  log_request_at_info_level: false
  log_format: json

# Distributor config: handle incoming log writes
distributor:
  # Rate limit: 10k events/sec per tenant
  rate_limit:
    enabled: true
    limit_per_second: 10000
    burst_size: 20000
  # Hashing: use consistent hashing to distribute to ingesters
  ring:
    kvstore:
      store: memberlist
    heartbeat_timeout: 1m
    replication_factor: 3
  # Error handling: reject requests with invalid labels
  reject_old_samples: true
  reject_old_samples_max_age: 10m

# Ingester config: receive and batch logs before writing to storage
ingester:
  # Lifecycle: flush to storage when chunk reaches 1.5MB or 15 minutes
  chunk_idle_period: 15m
  chunk_block_size: 262144  # 256KB blocks
  chunk_target_size: 1572864  # 1.5MB target chunk size
  max_chunk_age: 2h
  # TSDB index config (new in Loki 3.0)
  index:
    tsdb:
      dir: /var/lib/loki/tsdb
      # Create new TSDB shard every 24 hours
      shard_period: 24h
      # Retain TSDB shards for 30 days
      retention_period: 720h
  # WAL config for crash recovery
  wal:
    dir: /var/lib/loki/wal
    enabled: true
    flush_on_shutdown: true
  # Error handling: limit per-tenant ingester memory to 4GB
  per_tenant_max_ingestion_rate: 10000
  max_series_per_query: 100000
  max_line_size: 1048576  # 1MB max line

# Storage config: TSDB index + GCS for chunk storage
storage_config:
  # TSDB index backend (Loki 3.0 default)
  tsdb:
    dir: /var/lib/loki/tsdb
  # GCS chunk storage for long-term retention
  gcs:
    bucket_name: "prod-loki-chunks"
    service_account_file: /etc/loki/gcs-sa.json
  # BoltDB is deprecated in Loki 3.0, disable for compatibility
  boltdb_shipper:
    active_index_type: tsdb

# Compactor config: merge small chunks and delete expired data
compactor:
  working_directory: /var/lib/loki/compactor
  shared_store: gcs
  # Compact chunks every 10 minutes
  compaction_interval: 10m
  # Retention: delete chunks older than 30 days
  retention_enabled: true
  retention_period: 720h
  # Error handling: retry failed compactions 3 times
  compaction_retry_max_attempts: 3

# Querier config: handle log query requests
querier:
  # Max concurrent queries per tenant: 10
  max_concurrent: 10
  # Timeout for queries: 30 seconds
  timeout: 30s
  # TSDB query optimization: enable block caching
  tsdb:
    block_cache:
      enabled: true
      max_size_bytes: 1073741824  # 1GB block cache
    index_cache:
      enabled: true
      max_size_bytes: 536870912  # 512MB index cache

# Table manager: manage index table lifecycle (deprecated in 3.0, disable)
table_manager:
  retention_deletes_enabled: false
  retention_period: 0s

// log-benchmark.go: Benchmark log pipeline egress and query latency
// Author: Platform Team, Version: 1.0.0, Last Updated: 2024-09-20
// Dependencies: github.com/prometheus/client_golang v1.19.0
// Build: go build -o log-benchmark log-benchmark.go

package main

import (
    "bytes"
    "context"
    "encoding/json"
    "fmt"
    "log"
    "math/rand"
    "net/http"
    "os"
    "time"

    "github.com/prometheus/client_golang/prometheus"
    "github.com/prometheus/client_golang/prometheus/promauto"
    "github.com/prometheus/client_golang/prometheus/promhttp"
)

// Prometheus metrics for benchmarking
var (
    egressBytes = promauto.NewCounterVec(
        prometheus.CounterOpts{
            Name: "benchmark_egress_bytes_total",
            Help: "Total egress bytes sent to log sinks",
        },
        []string{"sink"},
    )
    queryLatency = promauto.NewHistogramVec(
        prometheus.HistogramOpts{
            Name: "benchmark_query_latency_ms",
            Help: "Query latency in milliseconds",
            Buckets: []float64{50, 100, 200, 500, 1000, 2000, 5000},
        },
        []string{"backend"},
    )
    errorCount = promauto.NewCounterVec(
        prometheus.CounterOpts{
            Name: "benchmark_errors_total",
            Help: "Total benchmark errors",
        },
        []string{"stage"},
    )
)

// Synthetic log generator: generate 1k random log lines per second
func generateLogs(ctx context.Context, logChan chan<- []byte) {
    services := []string{"payment-gateway", "user-service", "inventory-api", "checkout-worker"}
    severities := []string{"INFO", "WARN", "ERROR", "DEBUG"}
    rand.Seed(time.Now().UnixNano())

    for {
        select {
        case <-ctx.Done():
            return
        default:
            // Generate 100 log lines per batch
            batch := make([]byte, 0)
            for i := 0; i < 100; i++ {
                logLine := map[string]interface{}{
                    "timestamp": time.Now().Format(time.RFC3339Nano),
                    "service":   services[rand.Intn(len(services))],
                    "severity":  severities[rand.Intn(len(severities))],
                    "message":   fmt.Sprintf("synthetic-log-%d", rand.Intn(100000)),
                    "trace_id":  fmt.Sprintf("%x", rand.Int63()),
                    "node":      fmt.Sprintf("node-%d", rand.Intn(10)),
                }
                line, err := json.Marshal(logLine)
                if err != nil {
                    errorCount.WithLabelValues("log-generation").Inc()
                    log.Printf("failed to marshal log line: %v", err)
                    continue
                }
                batch = append(batch, line...)
                batch = append(batch, '\n')
            }
            logChan <- batch
            // Throttle to ~1k events/sec
            time.Sleep(100 * time.Millisecond)
        }
    }
}

// Send logs to Vector sink and measure egress
func sendToVector(ctx context.Context, logChan <-chan []byte) {
    vectorEndpoint := os.Getenv("VECTOR_ENDPOINT")
    if vectorEndpoint == "" {
        vectorEndpoint = "http://vector.prod.internal:8686"
    }
    client := &http.Client{Timeout: 5 * time.Second}

    for {
        select {
        case <-ctx.Done():
            return
        case batch := <-logChan:
            resp, err := client.Post(vectorEndpoint+"/logs", "application/json", bytes.NewBuffer(batch))
            if err != nil {
                errorCount.WithLabelValues("vector-send").Inc()
                log.Printf("failed to send to Vector: %v", err)
                continue
            }
            resp.Body.Close()
            egressBytes.WithLabelValues("vector").Add(float64(len(batch)))
            if resp.StatusCode != http.StatusOK {
                errorCount.WithLabelValues("vector-send").Inc()
                log.Printf("Vector returned status %d", resp.StatusCode)
            }
        }
    }
}

// Run query against Loki and measure latency
func queryLoki() {
    lokiEndpoint := os.Getenv("LOKI_ENDPOINT")
    if lokiEndpoint == "" {
        lokiEndpoint = "http://loki.prod.internal:3100"
    }
    client := &http.Client{Timeout: 30 * time.Second}
    query := `{service="payment-gateway"} | json | line_format "{{.message}}"`

    start := time.Now()
    resp, err := client.Get(fmt.Sprintf("%s/loki/api/v1/query?query=%s&time=%d", lokiEndpoint, query, time.Now().Unix()))
    if err != nil {
        errorCount.WithLabelValues("loki-query").Inc()
        log.Printf("failed to query Loki: %v", err)
        return
    }
    defer resp.Body.Close()
    latency := time.Since(start).Milliseconds()
    queryLatency.WithLabelValues("loki").Observe(float64(latency))
}

// Run query against Elasticsearch and measure latency
func queryElasticsearch() {
    esEndpoint := os.Getenv("ES_ENDPOINT")
    if esEndpoint == "" {
        esEndpoint = "http://elasticsearch.prod.internal:9200"
    }
    client := &http.Client{Timeout: 30 * time.Second}
    query := `{"query": {"match": {"service": "payment-gateway"}}}`

    start := time.Now()
    resp, err := client.Post(fmt.Sprintf("%s/_search", esEndpoint), "application/json", bytes.NewBufferString(query))
    if err != nil {
        errorCount.WithLabelValues("es-query").Inc()
        log.Printf("failed to query Elasticsearch: %v", err)
        return
    }
    defer resp.Body.Close()
    latency := time.Since(start).Milliseconds()
    queryLatency.WithLabelValues("elasticsearch").Observe(float64(latency))
}

func main() {
    ctx, cancel := context.WithCancel(context.Background())
    defer cancel()

    // Start Prometheus metrics endpoint
    go func() {
        http.Handle("/metrics", promhttp.Handler())
        log.Fatal(http.ListenAndServe(":9090", nil))
    }()

    // Start log generator
    logChan := make(chan []byte, 100)
    go generateLogs(ctx, logChan)

    // Start Vector sender
    go sendToVector(ctx, logChan)

    // Run periodic queries (every 30 seconds)
    ticker := time.NewTicker(30 * time.Second)
    defer ticker.Stop()
    for {
        select {
        case <-ticker.C:
            go queryLoki()
            go queryElasticsearch()
        case <-ctx.Done():
            return
        }
    }
}

[transforms.label_optimize]
type = "remap"
inputs = ["parse_and_enrich"]
source = '''
  # Drop high-cardinality labels that aren't used in queries
  del(.trace_id)
  del(.user_id)
  del(.pod_ip)
  # Only keep labels that are used in >5% of queries
  .severity = upcase(.severity)  # Normalize severity to reduce cardinality
'''

[transforms.adaptive_buffer]
type = "adaptive_buffer"
inputs = ["label_optimize"]
max_size_bytes = 10737418240  # 10GB max buffer
flush_interval_secs = 5
dead_letter_queue.enabled = true
dead_letter_queue.max_size_bytes = 1073741824  # 1GB DLQ

compactor:
  compaction_interval: 10m
  retention_enabled: true
  retention_period: 720h  # 30 days
querier:
  tsdb:
    block_cache:
      enabled: true
      max_size_bytes: 1073741824  # 1GB