Logql Generator

# Common LogQL Query Examples
# These queries demonstrate typical use cases for log analysis with Grafana Loki

## === BASIC LOG QUERIES ===

# Find all error logs
{job="app"} |= "error"

# Find logs containing "error" or "fatal"
{job="app"} |~ "error|fatal"

# Find errors NOT containing "timeout"
{job="app"} |= "error" != "timeout"

# Regex pattern matching
{job="nginx"} |~ "HTTP/[0-9.]+ (4|5)[0-9]{2}"

## === JSON LOG PARSING ===

# Parse JSON logs and filter by level
{app="api"} | json | level="error"

# Parse JSON and filter by multiple fields
{app="api"} | json | method="POST" | status_code >= 400

# Parse nested JSON fields
{app="api"} | json request.method="POST", response.status_code >= 500

# Extract and display specific JSON fields
{app="api"} | json | line_format "{{.level}}: {{.message}}"

## === LOGFMT PARSING ===

# Parse logfmt format
{app="app"} | logfmt

# Parse and filter logfmt logs
{app="app"} | logfmt | level="error" | caller="database.go"

## === LOGFMT PARSER FLAGS (Loki 3.x) ===

# Basic logfmt parsing
{app="api"} | logfmt

# Strict mode - fail on malformed key=value pairs (stops on error)
{app="api"} | logfmt --strict

# Keep standalone keys as labels with empty string value
{app="api"} | logfmt --keep-empty

# Combine flags for strict parsing with empty key retention
{app="api"} | logfmt --strict --keep-empty

# Strict mode with label extraction parameters
{app="api"} | logfmt --strict host, fwd_ip="fwd"

# Detect malformed log entries using strict mode
{app="api"} | logfmt --strict | __error__ != ""

# Production query - strict mode with error filtering
{app="api"} | logfmt --strict | __error__="" | level="error"

## === JSON PARSER WITH PARAMETER EXTRACTION ===

# Extract all fields (default behavior)
{app="api"} | json

# Extract specific fields by name (more efficient)
{app="api"} | json status, method, duration

# Extract with custom label names using expressions
{app="api"} | json first_server="servers[0]", ua="request.headers[\"User-Agent\"]"

# Extract arrays/objects as JSON strings
{app="api"} | json server_list="servers", headers="request.headers"

# Shorthand - label name equals field name
{app="api"} | json servers

# Extract nested request fields using dot notation
{app="api"} | json method="request.method", status="response.status_code"

# Extract nested fields with array access
{app="api"} | json first_item="items[0].name", last_item="items[-1].name"

# Combined with bracket notation for special characters
{app="api"} | json content_type="headers[\"Content-Type\"]"

# Combined with filtering on extracted fields
{app="api"} | json status="response.status" | status >= 400

# Extract only what you need for performance
{app="api"} | json level, message, trace_id | level="error"

## === PATTERN EXTRACTION ===

# Extract fields using pattern syntax
{job="nginx"} | pattern "<ip> - - [<timestamp>] \"<method> <path> <protocol>\" <status> <size>"

# Pattern extraction with filtering
{job="nginx"} | pattern "<ip> - - [<_>] \"<method> <_>\" <status> <_>" | status >= 400

# Pattern with wildcards
{service_name="distributor"} |> "<_> level=debug <_> msg=\"POST /push <_>\""

## === REGEX PARSING ===

# Extract fields with named capture groups
{app="app"} | regexp "(?P<level>\\w+): (?P<message>.+)"

# Extract IP addresses and HTTP codes
{job="nginx"} | regexp "(?P<ip>\\d+\\.\\d+\\.\\d+\\.\\d+) .* (?P<status>\\d{3})"

## === ERROR ANALYSIS ===

# Count errors over time
count_over_time({app="api"} | json | level="error" [5m])

# Error rate per second
sum(rate({app="api"} | json | level="error" [5m]))

# Error percentage
(
  sum(rate({app="api"} | json | level="error" [5m]))
  /
  sum(rate({app="api"}[5m]))
) * 100

# Errors by type
sum by (error_type) (
  count_over_time({app="api"} | json | level="error" [5m])
)

# Top 10 error messages
topk(10,
  sum by (error_message) (
    count_over_time({app="api"} | json | level="error" [1h])
  )
)

## === PERFORMANCE MONITORING ===

# Find slow requests (>1 second)
{app="api"} | json | duration > 1

# Average response time
avg_over_time({app="api"} | json | unwrap duration [5m])

# 95th percentile latency
quantile_over_time(0.95, {app="api"} | json | unwrap duration [5m])

# 99th percentile latency
quantile_over_time(0.99, {app="api"} | json | unwrap duration [5m])

# Max response time in last 5 minutes
max_over_time({app="api"} | json | unwrap duration [5m])

# Requests by latency bucket
sum by (le) (
  count_over_time({app="api"} | json | duration > 0 [5m])
)

## === TRAFFIC ANALYSIS ===

# Total log volume (logs per second)
sum(rate({namespace="production"}[5m]))

# Logs per second by application
sum by (app) (rate({namespace="production"}[5m]))

# Bytes per second
sum(bytes_rate({job="app"}[5m]))

# Request rate by endpoint
sum by (endpoint) (rate({app="api"} | json [5m]))

# HTTP status code distribution
sum by (status_code) (
  count_over_time({app="api"} | json [5m])
)

# Requests per second by method
sum by (method) (rate({app="api"} | json [5m]))

## === SECURITY MONITORING ===

# Failed login attempts
{app="auth"} | json | event="login_failed"

# Failed logins by user
sum by (username) (
  count_over_time({app="auth"} | json | event="login_failed" [1h])
)

# Top 10 users with failed logins
topk(10,
  sum by (username) (
    count_over_time({app="auth"} | json | event="login_failed" [1h])
  )
)

# Suspicious activity (unauthorized access attempts)
{app="api"} | json | (status_code == 401 or status_code == 403)

# Access from external IPs (requires a parsed label that holds the IP, e.g. source_ip)
{app="api"} | json | source_ip != ip("10.0.0.0/8") | source_ip != ip("172.16.0.0/12") | source_ip != ip("192.168.0.0/16")

# SQL injection attempts
{app="api"} |~ "union.*select|select.*from.*where"

## === USER BEHAVIOR TRACKING ===

# Top 10 users by activity
topk(10,
  sum by (user_id) (rate({app="api"}[5m]))
)

# User sessions by endpoint
sum by (user_id, endpoint) (
  count_over_time({app="api"} | json [1h])
)

# User actions per minute
sum by (user_id, action) (
  rate({app="api"} | json [1m])
)

## === APPLICATION DEBUGGING ===

# Logs from specific trace ID
{app="api"} | json | trace_id="abc123xyz"

# Logs for specific user session
{app="api"} | json | session_id="sess_12345"

# Debug logs with context
{app="api", level="debug"} | json | line_format "{{.timestamp}} [{{.trace_id}}] {{.message}}"

# Stack traces (multi-line logs)
{app="api"} |= "Exception" | json

# Database query errors
{app="api"} | json | component="database" | level="error"

## === LOG VOLUME AND RATES ===

# Total logs in last hour
sum(count_over_time({namespace="production"}[1h]))

# Peak log rate in last 24 hours
max_over_time(
  sum(rate({namespace="production"}[5m]))[24h:5m]
)

# Average log rate by hour
avg_over_time(
  sum(rate({namespace="production"}[5m]))[1h:5m]
)

# Logs by namespace
sum by (namespace) (rate({job="kubernetes-pods"}[5m]))

## === FILTERING AND TRANSFORMATION ===

# Remove ANSI color codes
{app="app"} | decolorize

# Keep only specific labels
{app="api"} | json | keep namespace, pod, level, message

# Drop noisy labels
{app="api"} | json | drop instance, pod

# Rename labels
{app="api"} | json | label_format env=`{{.environment}}`, svc=`{{.service}}`

# NOTE: LogQL does NOT have a native `dedup` operator
# Deduplication is a UI-level feature in Grafana's Explore panel
# For programmatic deduplication, use metric aggregations:
# sum by (message) (count_over_time({app="api"} | json [5m])) > 0

## === IP ADDRESS FILTERING ===

# The ip() function filters based on a label value containing an IP address
# It supports: single IPs, CIDR notation, and IP ranges

# Logs from specific IP (exact match)
{app="api"} | json | client_ip="192.168.1.100"

# Logs from IP range using CIDR notation (apply ip() to a specific label)
{job="nginx"} | logfmt | remote_addr = ip("192.168.1.0/24")

# Logs NOT from internal network (negate with !=)
{job="nginx"} | logfmt | remote_addr != ip("10.0.0.0/8")

# Logs from IP range (start-end format)
{job="nginx"} | logfmt | remote_addr = ip("192.168.4.5-192.168.4.20")

# Logs from multiple IP ranges (combine with or)
{job="nginx"} | logfmt | remote_addr = ip("192.168.0.0/16") or remote_addr = ip("10.0.0.0/8")

# Exclude specific IP within a range
{job="nginx"} | logfmt | remote_addr = ip("192.168.4.0/24") | remote_addr != ip("192.168.4.2")

# IPv6 address range filtering
{job="nginx"} | logfmt | remote_addr = ip("2001:db8::1-2001:db8::8")

# Line filter with IP (less precise, may have false positives)
{job="nginx"} |= ip("192.168.4.5/16")

## === TIME-BASED COMPARISONS ===

# Current error rate vs 1 hour ago
sum(rate({app="api"} | json | level="error" [5m]))
-
sum(rate({app="api"} | json | level="error" [5m] offset 1h))

# Current vs yesterday same time
sum(rate({app="api"}[5m]))
/
sum(rate({app="api"}[5m] offset 1d))

# Week-over-week comparison
sum(rate({app="api"} | json | level="error" [5m]))
/
sum(rate({app="api"} | json | level="error" [5m] offset 1w))

## === AGGREGATION EXAMPLES ===

# Sum of request durations
sum(sum_over_time({app="api"} | json | unwrap duration [5m]))

# Average by pod
avg by (pod) (
  avg_over_time({app="api"} | json | unwrap duration [5m])
)

# Maximum without specific labels
max without (instance, pod) (
  max_over_time({app="api"} | json | unwrap response_size [5m])
)

# Count distinct values (approximate)
count(
  count by (user_id) (
    {app="api"} | json
  )
)

## === ALERTING EXAMPLES ===

# High error rate (>5%)
(
  sum(rate({app="api"} | json | level="error" [5m]))
  /
  sum(rate({app="api"}[5m]))
) > 0.05

# Low log volume (potential issue)
sum(rate({app="api"}[5m])) < 0.1

# High latency (95th percentile >2s)
quantile_over_time(0.95, {app="api"} | json | unwrap duration [5m]) > 2

# No logs received (dead service)
absent_over_time({app="api"}[5m])

# Too many failed logins
sum(rate({app="auth"} | json | event="login_failed" [5m])) > 10

## === ADVANCED QUERIES ===

# Complex filtering with multiple conditions
{app="api"}
| json
| (status_code >= 400 and status_code < 500)
  or
  (status_code >= 500 and duration > 1)
| method != "HEAD"
| path !~ "/health|/metrics"

# Multi-stage parsing and formatting
{job="nginx"}
| pattern "<ip> - - [<timestamp>] \"<method> <path> <_>\" <status> <size>"
| json
| label_format status_class="{{if ge .status 500}}5xx{{else if ge .status 400}}4xx{{else}}ok{{end}}"
| line_format "{{.status_class}}: {{.method}} {{.path}}"

# Probabilistic top K (for large datasets)
approx_topk(10,
  sum by (endpoint) (rate({app="api"}[5m]))
)

# Percentage per status code of total traffic (many-to-one matching against total)
# LogQL supports vector matching modifiers for arithmetic/comparison:
# on(...), ignoring(...), group_left, and group_right.
sum by (status_code) (rate({app="api"} | json [5m]))
/ on() group_left
sum(rate({app="api"}[5m]))

# Percentage per status code within each app (many-to-one matching)
sum by (app, status_code) (rate({job="http-server"} | json [5m]))
/ on(app) group_left
sum by (app) (rate({job="http-server"} | json [5m]))

## === STRUCTURED METADATA (Loki 3.x) ===

# Structured metadata is NOT in stream selector - it comes AFTER the selector
# This enables high-cardinality data without index bloat

# Query logs with structured metadata (correct syntax)
{app="api"} | trace_id="abc123"

# Combine structured metadata with other filters
{app="api"} | trace_id="abc123" | json | level="error"

# Multiple structured metadata filters
{app="api"} | user_id="12345" | request_id="req-abc"

# WRONG: Don't put structured metadata in stream selector!
# {app="api", trace_id="abc123"}  # This won't work!

## === QUERY ACCELERATION (Loki 3.x) ===

# ACCELERATED: Structured metadata filter BEFORE parser (bloom filters used)
{cluster="prod"} | detected_level="error" | logfmt | json

# NOT ACCELERATED: Filter AFTER parser (bloom filters skipped)
{cluster="prod"} | logfmt | json | detected_level="error"

# Accelerated with multiple conditions
{app="api"} | trace_id="abc123" | service="payment" | json | level="error"

# Accelerated with OR conditions
{app="api"} | detected_level="error" or detected_level="warn" | json

## === APPROX_TOPK (Probabilistic Top-K) — Loki 3.3+ only ===

# Faster alternative to topk for high-cardinality data
# Returns approximate results, great for large datasets
# REQUIRES: Loki 3.3 or newer — do NOT use when version is unknown or < 3.3

# Approximate top 10 endpoints by request rate
approx_topk(10,
  sum by (endpoint) (rate({app="api"}[5m]))
)

# Approximate top 20 users by error count
approx_topk(20,
  sum by (user_id) (count_over_time({app="api"} | json | level="error" [1h]))
)

# Use when topk times out or hits series limits
approx_topk(50,
  sum by (trace_id) (rate({app="api"}[5m]))
)

## === VECTOR() FOR RELIABLE ALERTING ===

# Ensures a value is always returned (prevents "no data" alert states)

# Always returns a value (0 when no matches)
sum(count_over_time({app="api"} | json | level="error" [5m])) or vector(0)

# Use in alerting rules
sum(rate({app="api"} | json | level="error" [5m])) or vector(0) > 10

# Percentage calculation with fallback
(
  sum(rate({app="api"} | json | level="error" [5m])) or vector(0)
)
/
(
  sum(rate({app="api"}[5m])) or vector(1)
) * 100

## === __ERROR__ LABEL DEBUGGING ===

# Show only lines that failed to parse
{app="api"} | json | __error__ != ""

# Show only successfully parsed lines (production use)
{app="api"} | json | __error__="" | level="error"

# Debug parse errors with details
{app="api"} | json | __error__ != "" | line_format "ERROR: {{.__error__}} LINE: {{.__line__}}"

# Count errors by parser error type
sum by (__error__) (count_over_time({app="api"} | json | __error__ != "" [5m]))

## === TEMPLATE FUNCTIONS ===

# String manipulation
{app="api"} | json | line_format "{{.path | replace \" \" \"_\" | trunc 50 | upper}}"

# Trim whitespace
{app="api"} | json | line_format "IP: {{.client_ip | trim}}"

# Indent vs nindent
# indent: indents every line in a string
{app="api"} | json | line_format "Message:{{indent 4 .message}}"

# nindent: prepends a newline BEFORE indenting (useful for YAML-like output)
{app="api"} | json | line_format "Details:{{nindent 4 .stack_trace}}"

# nindent is particularly useful for multi-line content
{app="api"} | json | line_format "Error Report:{{nindent 2 .error}}{{nindent 2 .context}}"

# Date formatting
{app="api"} | json | line_format "{{__timestamp__ | date \"2006-01-02T15:04:05\"}}: {{.message}}"

# Math operations
{app="api"} | json | line_format "Duration: {{div .duration_ms 1000}}s"

# Conditional formatting
{app="api"} | json | label_format severity="{{if ge .status_code 500}}critical{{else if ge .status_code 400}}warning{{else}}info{{end}}"

# Access original line
{app="api"} | json | line_format "ORIGINAL: {{__line__ | lower}}"

# Printf formatting
{app="api"} | json | line_format "{{printf \"%-40.40s\" .request_uri}} {{printf \"%5.5s\" .method}}"

# Iterate over JSON array in log
{job="api"} | json | line_format "{{ range $item := fromJson .items }}{{ $item.name }} {{ end }}"

## === LINE MATCH PATTERN OPERATOR (|>) ===

# Faster than regex for pattern matching with wildcards
# <_> is a wildcard for any arbitrary text

# Match pattern with wildcards
{service_name="distributor"} |> "<_> level=debug <_> msg=\"POST /push <_>\""

# Filter out debug logs with pattern
{service_name="api"} !> "<_> level=debug <_>"

# Match HTTP request patterns
{job="nginx"} |> "<_> \"GET /api/<_>\" 200 <_>"

## === DECOLORIZE ===

# Remove ANSI color codes from terminal output
{app="app"} | decolorize

# Decolorize before parsing
{app="app"} | decolorize | json | level="error"

## === UNPACK PARSER ===

# Unpack data that was packed by Promtail's pack stage
{cluster="us-central1", job="myjob"} | unpack

# Unpack and filter by embedded label
{cluster="us-central1", job="myjob"} | unpack | container="myapp"

# Unpack, filter, then parse the original log
{cluster="us-central1", job="myjob"} | unpack | container="myapp" | json

## === ADDITIONAL UNWRAPPED RANGE FUNCTIONS ===

# First value in interval (useful for starting values)
first_over_time({app="api"} | json | unwrap request_count [5m])

# Last value in interval (useful for ending values)
last_over_time({app="api"} | json | unwrap request_count [5m])

# Standard deviation of durations (useful for detecting variability)
stddev_over_time({app="api"} | json | unwrap duration [5m])

# Standard variance of durations
stdvar_over_time({app="api"} | json | unwrap duration [5m])

# Rate counter - treats values as a counter metric (for monotonically increasing values)
rate_counter({app="api"} | json | unwrap total_requests [5m])

# Bytes over time - total bytes in a time range
bytes_over_time({app="api"}[5m])

## === SORTING AND ORDERING ===

# Sort results ascending by value
sort(sum by (app) (rate({job="api"}[5m])))

# Sort results descending by value
sort_desc(sum by (app) (rate({job="api"}[5m])))

# Combine sort with topk for ordered top results
sort_desc(topk(10, sum by (endpoint) (rate({app="api"}[5m]))))

## === LABEL_REPLACE FUNCTION ===

# Extract service name from label using regex capture group
label_replace(
  rate({job="api-server", service="payment:v2"} |= "err" [1m]),
  "service_name", "$1", "service", "(.*):.*"
)

# Add environment label based on namespace pattern
label_replace(
  sum by (namespace) (rate({job="app"}[5m])),
  "env", "$1", "namespace", "(prod|staging|dev).*"
)

# Create simplified label from complex one
label_replace(
  sum by (pod) (rate({job="kubernetes-pods"}[5m])),
  "app", "$1", "pod", "([a-z-]+)-[a-z0-9]+-[a-z0-9]+"
)

## === LABEL_REPLACE CHAINING (Advanced) ===

# Chain multiple label_replace calls for complex transformations
# Example: Extract app, version, and region from pod name "myapp-v2-us-east-1-abc123-xyz789"
label_replace(
  label_replace(
    label_replace(
      sum by (pod) (rate({job="kubernetes-pods"}[5m])),
      "app", "$1", "pod", "([a-z-]+)-v[0-9]+-.*"
    ),
    "version", "$1", "pod", "[a-z-]+-v([0-9]+)-.*"
  ),
  "region", "$1", "pod", "[a-z-]+-v[0-9]+-([a-z]+-[a-z]+-[0-9]+)-.*"
)

# Extract team and service from job label like "team-platform/service-api"
label_replace(
  label_replace(
    sum by (job) (rate({namespace="production"}[5m])),
    "team", "$1", "job", "team-([^/]+)/.*"
  ),
  "service", "$1", "job", "[^/]+/service-(.*)"
)

# Create environment label from namespace and add severity from level
label_replace(
  label_replace(
    sum by (namespace, level) (rate({job="app"} | json [5m])),
    "env", "production", "namespace", "prod.*"
  ),
  "severity", "critical", "level", "error|fatal"
)

# Extract domain from URL-based endpoint labels
label_replace(
  sum by (endpoint) (rate({app="api"} | json [5m])),
  "domain", "$1", "endpoint", "https?://([^/]+)/.*"
)

# Transform pod names to deployment names (strip replicaset hash)
label_replace(
  sum by (pod) (rate({namespace="prod"}[5m])),
  "deployment", "$1", "pod", "(.*)-[a-f0-9]+-[a-z0-9]+"
)

## === CONVERSION FUNCTIONS ===

# Convert duration strings like "1.5s" or "150ms" to seconds
avg_over_time({app="api"} | json | unwrap duration_seconds(response_time) [5m])

# Convert byte strings like "1.5KB" or "10MB" to bytes
sum_over_time({app="api"} | json | unwrap bytes(payload_size) [5m])

## === FLOAT TEMPLATE FUNCTIONS ===

# Division with float result
{app="api"} | json | line_format "Duration: {{divf .duration_ns 1000000}}ms"

# Subtraction with floats
{app="api"} | json | line_format "Change: {{subf .current .previous}}"

# Floor and ceiling
{app="api"} | json | line_format "Floor: {{floor .value}} Ceil: {{ceil .value}}"

# Round to 2 decimal places
{app="api"} | json | line_format "Rounded: {{round .percentage 2}}%"

## === REGEX TEMPLATE FUNCTIONS ===

# Replace with regex capture groups
{app="api"} | json | line_format "{{regexReplaceAll \"user_(\\\\d+)\" .message \"User ID: $1\"}}"

# Literal replacement (no capture group expansion)
{app="api"} | json | line_format "{{regexReplaceAllLiteral \"error\" .message \"ERROR\"}}"

## === AUTOMATIC LABELS (service_name and detected_level) ===

# Loki automatically populates certain labels to improve log exploration

## service_name Label (Auto-populated)
# - Auto-populated by Loki when no service name is provided
# - Uses container name as fallback if not specified
# - Used in Grafana's Logs Drilldown feature
# - Can be customized via discover_service_name in limits_config

# Query by auto-generated service_name
{service_name="my-api"}

# Combine service_name with other filters
{service_name="my-api"} | json | level="error"

# Error rate by service_name
sum by (service_name) (rate({namespace="production"} | json | level="error" [5m]))

# Top services by error count
topk(10, sum by (service_name) (count_over_time({namespace="prod"} | json | level="error" [1h])))

## detected_level Label (Structured Metadata - Auto-detected)
# - Auto-detected when discover_log_levels: true is configured
# - Stored as structured metadata (NOT indexed)
# - Values: debug, info, warn, error, critical, fatal
# - Requires allow_structured_metadata: true in limits_config
# - Place BEFORE parsers for query acceleration with bloom filters!

# Query by auto-detected log level (accelerated with bloom filters)
{cluster="prod"} | detected_level="error" | json

# Combine detected_level with other structured metadata
{app="api"} | detected_level="warn" | trace_id="abc123" | json

# Multiple detected_level values with OR (accelerated)
{app="api"} | detected_level="error" or detected_level="critical" | json

# Error count using detected_level
sum(count_over_time({namespace="prod"} | detected_level="error" | json [5m]))

# Error rate by detected_level
sum by (detected_level) (rate({app="api"} | detected_level!="" | json [5m]))

# WRONG: detected_level AFTER parser loses acceleration!
# {cluster="prod"} | json | detected_level="error"  # NOT accelerated

# CORRECT: detected_level BEFORE parser (accelerated)
{cluster="prod"} | detected_level="error" | json

## === GRAFANA ALLOY / PROMTAIL MIGRATION ===

# Note: Promtail is deprecated in favor of Grafana Alloy
# Commercial support for Promtail ends February 28, 2026
# For new deployments, use Grafana Alloy instead of Promtail

# Query logs collected by Alloy (same LogQL syntax)
{collector="alloy"}

# Query logs with Alloy-specific labels
{job="alloy"} | json | level="error"

#!/usr/bin/env bash
#
# Run logql-generator regression checks.
# - Static contract checks always run.
# - Runtime Loki checks are controlled by RUN_LOKI_RUNTIME_TESTS:
#   auto (default): run only when Docker is available
#   1/true: require runtime checks (fail if Docker is unavailable)
#   0/false: skip runtime checks
#

set -euo pipefail

SCRIPT_DIR="$(cd "$(dirname "${BASH_SOURCE[0]}")" && pwd)"
readonly SCRIPT_DIR
readonly SKILL_DIR="$(cd "$SCRIPT_DIR/.." && pwd)"
readonly TEST_DIR="$SKILL_DIR/tests"

if [[ ! -d "$TEST_DIR" ]]; then
    echo "FAIL: tests directory not found: $TEST_DIR" >&2
    exit 1
fi

RUN_MODE="${RUN_LOKI_RUNTIME_TESTS:-auto}"

echo "Running logql-generator regression checks..."
echo "RUN_LOKI_RUNTIME_TESTS=${RUN_MODE}"
echo ""

PYTHONDONTWRITEBYTECODE=1 \
RUN_LOKI_RUNTIME_TESTS="$RUN_MODE" \
python3 -m unittest discover -s "$TEST_DIR" -p "test_*.py" -v

echo ""
echo "All logql-generator regression checks finished."

#!/usr/bin/env python3
"""Runtime Loki integration tests for logql-generator query regressions."""

from __future__ import annotations

import json
import os
import shutil
import socket
import subprocess
import time
import unittest
from urllib import error, parse, request


RUN_MODE = os.getenv("RUN_LOKI_RUNTIME_TESTS", "auto").strip().lower()
LOKI_IMAGE = os.getenv("LOKI_IMAGE", "grafana/loki:3.6.2").strip()
STARTUP_TIMEOUT_SECONDS = int(os.getenv("LOKI_STARTUP_TIMEOUT_SECONDS", "60"))
QUERY_TIMEOUT_SECONDS = int(os.getenv("LOKI_QUERY_TIMEOUT_SECONDS", "25"))

_REQUIRE_VALUES = {"1", "true", "yes", "required"}
_SKIP_VALUES = {"0", "false", "no", "off", "skip"}


class LokiRuntimeIntegrationTests(unittest.TestCase):
    """Validate key query examples against a real ephemeral Loki runtime."""

    container_name: str | None = None
    base_url: str | None = None

    @staticmethod
    def _docker_available() -> bool:
        if shutil.which("docker") is None:
            return False
        try:
            subprocess.run(
                ["docker", "info"],
                check=True,
                stdout=subprocess.DEVNULL,
                stderr=subprocess.DEVNULL,
            )
        except (OSError, subprocess.CalledProcessError):
            return False
        return True

    @staticmethod
    def _reserve_host_port() -> int:
        with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as sock:
            sock.bind(("127.0.0.1", 0))
            return int(sock.getsockname()[1])

    @classmethod
    def _stop_container(cls) -> None:
        if not cls.container_name:
            return
        subprocess.run(
            ["docker", "rm", "-f", cls.container_name],
            check=False,
            stdout=subprocess.DEVNULL,
            stderr=subprocess.DEVNULL,
        )
        cls.container_name = None
        cls.base_url = None

    @classmethod
    def setUpClass(cls) -> None:
        if RUN_MODE in _SKIP_VALUES:
            raise unittest.SkipTest(
                "Runtime Loki tests skipped by RUN_LOKI_RUNTIME_TESTS."
            )

        docker_ready = cls._docker_available()
        if RUN_MODE in _REQUIRE_VALUES and not docker_ready:
            raise RuntimeError(
                "RUN_LOKI_RUNTIME_TESTS requires Docker, but Docker is unavailable."
            )
        if RUN_MODE not in _REQUIRE_VALUES and not docker_ready:
            raise unittest.SkipTest(
                "Docker unavailable; skipping runtime Loki tests (auto mode)."
            )

        host_port = cls._reserve_host_port()
        cls.container_name = f"logql-generator-loki-{os.getpid()}-{int(time.time())}"
        cls.base_url = f"http://127.0.0.1:{host_port}"

        try:
            subprocess.run(
                [
                    "docker",
                    "run",
                    "-d",
                    "--rm",
                    "--name",
                    cls.container_name,
                    "-p",
                    f"{host_port}:3100",
                    LOKI_IMAGE,
                    "-config.file=/etc/loki/local-config.yaml",
                ],
                check=True,
                stdout=subprocess.DEVNULL,
                stderr=subprocess.PIPE,
                text=True,
            )
            cls._wait_until_ready()
            cls._push_sample_logs()
        except Exception:
            cls._stop_container()
            raise

    @classmethod
    def tearDownClass(cls) -> None:
        cls._stop_container()

    @classmethod
    def _wait_until_ready(cls) -> None:
        assert cls.base_url is not None
        ready_url = f"{cls.base_url}/ready"
        deadline = time.time() + STARTUP_TIMEOUT_SECONDS
        last_error: str | None = None

        while time.time() < deadline:
            try:
                with request.urlopen(ready_url, timeout=4) as response:
                    body = response.read().decode("utf-8", errors="ignore").lower()
                    if response.status == 200 and "ready" in body:
                        return
            except error.HTTPError as exc:  # pragma: no cover - retry loop
                try:
                    exc.read()
                finally:
                    exc.close()
                last_error = f"HTTP {exc.code}"
            except Exception as exc:  # pragma: no cover - retry loop
                last_error = str(exc)
            time.sleep(1)

        raise RuntimeError(f"Loki did not become ready in time: {last_error}")

    @classmethod
    def _push_sample_logs(cls) -> None:
        assert cls.base_url is not None
        now_ns = time.time_ns()
        # Keep log lines within a recent 5m range window for rate()/bytes_over_time().
        samples = {
            "streams": [
                {
                    "stream": {"app": "api", "job": "http-server"},
                    "values": [
                        [str(now_ns - 90_000_000_000), '{"status_code":200,"duration":0.12,"level":"info"}'],
                        [str(now_ns - 60_000_000_000), '{"status_code":500,"duration":1.42,"level":"error"}'],
                        [str(now_ns - 30_000_000_000), '{"status_code":500,"duration":0.98,"level":"error"}'],
                    ],
                },
                {
                    "stream": {"app": "billing", "job": "http-server"},
                    "values": [
                        [str(now_ns - 85_000_000_000), '{"status_code":200,"duration":0.10,"level":"info"}'],
                        [str(now_ns - 55_000_000_000), '{"status_code":404,"duration":0.23,"level":"warn"}'],
                        [str(now_ns - 25_000_000_000), '{"status_code":500,"duration":1.33,"level":"error"}'],
                    ],
                },
            ]
        }
        payload = json.dumps(samples, separators=(",", ":")).encode("utf-8")
        req = request.Request(
            f"{cls.base_url}/loki/api/v1/push",
            data=payload,
            method="POST",
            headers={"Content-Type": "application/json"},
        )
        with request.urlopen(req, timeout=10) as response:
            if response.status != 204:
                raise RuntimeError(f"Unexpected Loki push status code: {response.status}")

    def _query(self, query: str) -> list[dict]:
        assert self.base_url is not None
        url = f"{self.base_url}/loki/api/v1/query?{parse.urlencode({'query': query})}"
        req = request.Request(url, method="GET")
        try:
            with request.urlopen(req, timeout=10) as response:
                payload = json.loads(response.read().decode("utf-8"))
        except error.HTTPError as exc:
            body = exc.read().decode("utf-8", errors="ignore")
            raise AssertionError(f"Loki query failed ({exc.code}): {body}") from exc

        self.assertEqual(payload.get("status"), "success", msg=str(payload))
        return payload.get("data", {}).get("result", [])

    def _query_until_non_empty(self, query: str) -> list[dict]:
        deadline = time.time() + QUERY_TIMEOUT_SECONDS
        last_error: Exception | None = None
        last_result: list[dict] = []

        while time.time() < deadline:
            try:
                last_result = self._query(query)
                if last_result:
                    return last_result
            except Exception as exc:  # pragma: no cover - retry loop
                last_error = exc
            time.sleep(1)

        if last_error is not None:
            raise AssertionError(f"Query did not stabilize: {last_error}") from last_error
        raise AssertionError(f"Query returned no results within timeout: {query}")

    def test_ratio_against_total_with_on_group_left_executes(self) -> None:
        query = (
            'sum by (status_code) (rate({app="api"} | json [5m]))'
            " / on() group_left "
            'sum(rate({app="api"}[5m]))'
        )
        result = self._query_until_non_empty(query)
        self.assertGreaterEqual(len(result), 2)
        for series in result:
            self.assertIn("status_code", series.get("metric", {}))

    def test_many_to_one_ratio_with_group_left_executes(self) -> None:
        query = (
            'sum by (app, status_code) (rate({job="http-server"} | json [5m]))'
            " / on(app) group_left "
            'sum by (app) (rate({job="http-server"} | json [5m]))'
        )
        result = self._query_until_non_empty(query)
        apps = set()
        for series in result:
            metric = series.get("metric", {})
            self.assertIn("app", metric)
            self.assertIn("status_code", metric)
            apps.add(metric["app"])
        self.assertIn("api", apps)
        self.assertIn("billing", apps)

    def test_bytes_over_time_executes_without_unwrap(self) -> None:
        query = 'bytes_over_time({app="api"}[5m])'
        result = self._query_until_non_empty(query)
        self.assertTrue(any(series.get("metric", {}).get("app") == "api" for series in result))


if __name__ == "__main__":
    unittest.main()

#!/usr/bin/env python3
"""Regression tests for logql-generator documentation contracts."""

from pathlib import Path
import re
import unittest


SKILL_DIR = Path(__file__).resolve().parent.parent
COMMON_QUERIES = SKILL_DIR / "examples" / "common_queries.logql"


def _resolve_skill_md() -> Path:
    candidate = SKILL_DIR / "SKILL.md"
    if candidate.exists():
        return candidate
    raise FileNotFoundError(f"Could not find skill markdown file: {candidate}")


SKILL_MD = _resolve_skill_md()


def _between(text: str, start_marker: str, end_marker: str) -> str:
    start = text.find(start_marker)
    end = text.find(end_marker, start + len(start_marker))
    if start == -1 or end == -1:
        raise AssertionError(
            f"Could not find section boundaries: {start_marker!r} -> {end_marker!r}"
        )
    return text[start:end]


class TestVectorMatchingGuidance(unittest.TestCase):
    """Ensure ratio examples and guidance reflect actual LogQL vector matching behavior."""

    @classmethod
    def setUpClass(cls) -> None:
        cls.examples_text = COMMON_QUERIES.read_text(encoding="utf-8")

    def test_does_not_claim_vector_matching_is_unsupported(self) -> None:
        bad_claim = re.compile(
            r"does\s+not\s+support.*on\(\).*group_left",
            re.IGNORECASE | re.DOTALL,
        )
        self.assertIsNone(
            bad_claim.search(self.examples_text),
            "Examples must not claim that on()/group_left() are unsupported in LogQL.",
        )

    def test_has_ratio_example_with_on_group_left_against_total(self) -> None:
        self.assertRegex(
            self.examples_text,
            re.compile(
                r"sum by \(status_code\)\s*\(rate\(\{app=\"api\"\} \| json \[5m\]\)\)\s*"
                r"/ on\(\) group_left\s*sum\(rate\(\{app=\"api\"\}\[5m\]\)\)",
                re.DOTALL,
            ),
        )

    def test_has_many_to_one_ratio_example_with_group_left(self) -> None:
        self.assertRegex(
            self.examples_text,
            re.compile(
                r"sum by \(app, status_code\)\s*\(rate\(\{job=\"http-server\"\} \| json \[5m\]\)\)\s*"
                r"/ on\(app\) group_left\s*"
                r"sum by \(app\)\s*\(rate\(\{job=\"http-server\"\} \| json \[5m\]\)\)",
                re.DOTALL,
            ),
        )


class TestBytesOverTimePlacement(unittest.TestCase):
    """Ensure bytes_over_time remains classified as a log-range aggregation."""

    @classmethod
    def setUpClass(cls) -> None:
        cls.skill_text = SKILL_MD.read_text(encoding="utf-8")
        cls.log_range_section = _between(
            cls.skill_text,
            "### Log Range Aggregations",
            "### Unwrapped Range Aggregations",
        )
        cls.unwrapped_section = _between(
            cls.skill_text,
            "### Unwrapped Range Aggregations",
            "### Aggregation Operators",
        )

    def test_bytes_over_time_is_listed_in_log_range_section(self) -> None:
        self.assertIn("`bytes_over_time(log-range)`", self.log_range_section)

    def test_bytes_over_time_is_not_listed_in_unwrapped_section(self) -> None:
        self.assertNotIn("bytes_over_time", self.unwrapped_section)

    def test_skill_has_bytes_over_time_usage_rule(self) -> None:
        self.assertIn(
            "Use `bytes_over_time(<log-range>)` for raw log-byte volume.",
            self.skill_text,
        )
        self.assertIn(
            "Use `| unwrap bytes(field)` with unwrapped range aggregations",
            self.skill_text,
        )


if __name__ == "__main__":
    unittest.main()