We Ditched New Relic for Grafana 10 and Loki 2.9: 60% Monitoring Cost Savings

package main

import (
    "bytes"
    "context"
    "encoding/json"
    "fmt"
    "log"
    "net/http"
    "os"
    "os/signal"
    "syscall"
    "time"

    "github.com/prometheus/common/model"
    "github.com/grafana/loki-client-go/pkg/loki"
    "github.com/grafana/loki-client-go/pkg/logproto"
)

// LokiConfig holds configuration for the Loki client
type LokiConfig struct {
    URL      string        // Loki push endpoint (e.g., http://loki:3100/loki/api/v1/push)
    BatchSize int          // Number of logs to batch before pushing
    BatchWait time.Duration // Max time to wait before flushing batch
    Timeout   time.Duration // HTTP timeout for push requests
}

// LokiClient wraps the official Loki client with retry logic and error handling
type LokiClient struct {
    client *loki.Client
    config LokiConfig
}

// NewLokiClient initializes a new Loki client with validation and retries
func NewLokiClient(cfg LokiConfig) (*LokiClient, error) {
    if cfg.URL == "" {
        return nil, fmt.Errorf("loki URL must not be empty")
    }
    if cfg.BatchSize <= 0 {
        cfg.BatchSize = 100 // default batch size
    }
    if cfg.BatchWait <= 0 {
        cfg.BatchWait = 5 * time.Second // default batch wait
    }
    if cfg.Timeout <= 0 {
        cfg.Timeout = 10 * time.Second // default HTTP timeout
    }

    // Initialize official Loki client from grafana/loki-client-go
    client, err := loki.New(cfg.URL, loki.Config{
        BatchSize: cfg.BatchSize,
        BatchWait: cfg.BatchWait,
        Timeout:   cfg.Timeout,
    })
    if err != nil {
        return nil, fmt.Errorf("failed to initialize loki client: %\w", err)
    }

    return &LokiClient{client: client, config: cfg}, nil
}

// PushLog sends a structured log entry to Loki with labels and retries on failure
func (lc *LokiClient) PushLog(ctx context.Context, labels model.LabelSet, message string) error {
    // Validate inputs
    if len(labels) == 0 {
        return fmt.Errorf("labels must not be empty")
    }
    if message == "" {
        return fmt.Errorf("log message must not be empty")
    }

    // Retry up to 3 times on failure
    var err error
    for i := 0; i < 3; i++ {
        select {
        case <-ctx.Done():
            return ctx.Err()
        default:
            err = lc.client.Handle(ctx, labels, message)
            if err == nil {
                return nil
            }
            log.Printf("retry %d/3 failed to push log: %v", i+1, err)
            time.Sleep(time.Duration(i+1) * 100 * time.Millisecond) // exponential backoff
        }
    }
    return fmt.Errorf("failed to push log after 3 retries: %\w", err)
}

// Shutdown flushes any pending logs and closes the client
func (lc *LokiClient) Shutdown() error {
    return lc.client.Stop()
}

func main() {
    // Load config from environment variables
    cfg := LokiConfig{
        URL:       os.Getenv("LOKI_URL"),
        BatchSize: 100,
        BatchWait: 5 * time.Second,
        Timeout:   10 * time.Second,
    }
    if cfg.URL == "" {
        cfg.URL = "http://localhost:3100/loki/api/v1/push"
    }

    // Initialize client
    client, err := NewLokiClient(cfg)
    if err != nil {
        log.Fatalf("failed to create loki client: %v", err)
    }
    defer func() {
        if err := client.Shutdown(); err != nil {
            log.Printf("failed to shutdown loki client: %v", err)
        }
    }()

    // Handle OS signals for graceful shutdown
    ctx, cancel := context.WithCancel(context.Background())
    sigChan := make(chan os.Signal, 1)
    signal.Notify(sigChan, syscall.SIGINT, syscall.SIGTERM)
    go func() {
        <-sigChan
        log.Println("received shutdown signal, flushing logs...")
        cancel()
    }()

    // Send sample logs in a loop
    ticker := time.NewTicker(1 * time.Second)
    defer ticker.Stop()
    for {
        select {
        case <-ctx.Done():
            log.Println("context cancelled, exiting...")
            return
        case <-ticker.C:
            labels := model.LabelSet{
                "app":       "sample-service",
                "env":       "production",
                "level":     "info",
                "trace_id":  "a1b2c3d4e5f6a7b8c9d0e1f2a3b4c5d6",
            }
            message := fmt.Sprintf("sample log entry at %s", time.Now().Format(time.RFC3339))
            if err := client.PushLog(ctx, labels, message); err != nil {
                log.Printf("failed to push log: %v", err)
            }
        }
    }
}

import os
import time
import json
import logging
from typing import Dict, List, Optional
import requests
from requests.adapters import HTTPAdapter
from urllib3.util.retry import Retry

# Configure logging
logging.basicConfig(
    level=logging.INFO,
    format="%(asctime)s - %(levelname)s - %(message)s"
)
logger = logging.getLogger(__name__)

class LokiQueryClient:
    """Client to query Loki 2.9 logs via the Query API with retry logic and pagination."""

    def __init__(self, loki_url: str = "http://localhost:3100"):
        """Initialize Loki client with base URL and retry configuration.

        Args:
            loki_url: Base URL of Loki instance (e.g., http://loki:3100)
        """
        self.loki_url = loki_url.rstrip("/")
        self.session = self._configure_session()

    def _configure_session(self) -> requests.Session:
        """Configure requests session with retry logic for transient errors."""
        session = requests.Session()
        retry_strategy = Retry(
            total=3,
            backoff_factor=1,
            status_forcelist=[429, 500, 502, 503, 504],
            allowed_methods=["GET", "POST"]
        )
        adapter = HTTPAdapter(max_retries=retry_strategy)
        session.mount("http://", adapter)
        session.mount("https://", adapter)
        return session

    def query_range(
        self,
        query: str,
        start: Optional[int] = None,
        end: Optional[int] = None,
        limit: int = 1000,
        direction: str = "backward"
    ) -> List[Dict]:
        """Query Loki for logs within a time range with pagination.

        Args:
            query: LogQL query string (e.g., {app="my-service"} |= "error")
            start: Start timestamp in Unix nanoseconds (default: 1 hour ago)
            end: End timestamp in Unix nanoseconds (default: now)
            limit: Max number of logs to return per page
            direction: Pagination direction (forward/backward)

        Returns:
            List of log entries with labels and timestamp
        """
        if start is None:
            start = int(time.time() * 1e9) - (3600 * 1e9)  # 1 hour ago
        if end is None:
            end = int(time.time() * 1e9)

        logs = []
        next_token = None

        while True:
            params = {
                "query": query,
                "start": start,
                "end": end,
                "limit": limit,
                "direction": direction,
            }
            if next_token:
                params["nextToken"] = next_token

            try:
                response = self.session.get(
                    f"{self.loki_url}/loki/api/v1/query_range",
                    params=params,
                    timeout=30
                )
                response.raise_for_status()
            except requests.exceptions.RequestException as e:
                logger.error(f"Failed to query Loki: {e}")
                raise

            data = response.json()
            if data.get("status") != "success":
                raise RuntimeError(f"Loki query failed: {data.get('error', 'unknown error')}")

            # Extract logs from response
            result_type = data["data"]["resultType"]
            if result_type == "streams":
                for stream in data["data"]["result"]:
                    labels = stream["stream"]
                    for value in stream["values"]:
                        timestamp_ns, log_line = value
                        logs.append({
                            "timestamp": int(timestamp_ns),
                            "labels": labels,
                            "message": log_line
                        })
            elif result_type == "matrix":
                logger.warning("Matrix result type not supported for log queries")
                break

            # Check for next page
            next_token = data["data"].get("nextToken")
            if not next_token:
                break

        return logs

    def aggregate_errors_by_service(self, hours_ago: int = 1) -> Dict[str, int]:
        """Aggregate error logs by service name over a time window.

        Args:
            hours_ago: Number of hours to look back for errors

        Returns:
            Dictionary mapping service name to error count
        """
        start = int(time.time() * 1e9) - (hours_ago * 3600 * 1e9)
        query = '{level="error"}'  # Query all error-level logs
        logs = self.query_range(query, start=start)

        error_counts = {}
        for log in logs:
            service = log["labels"].get("app", "unknown")
            error_counts[service] = error_counts.get(service, 0) + 1

        return error_counts

def main():
    # Load Loki URL from environment variable
    loki_url = os.getenv("LOKI_URL", "http://localhost:3100")
    client = LokiQueryClient(loki_url=loki_url)

    try:
        logger.info("Querying error logs from the last hour...")
        error_counts = client.aggregate_errors_by_service(hours_ago=1)

        if not error_counts:
            logger.info("No error logs found in the last hour.")
            return

        logger.info("Error counts by service:")
        for service, count in sorted(error_counts.items(), key=lambda x: -x[1]):
            logger.info(f"  {service}: {count} errors")

        # Print as JSON for easy parsing
        print(json.dumps(error_counts, indent=2))

    except Exception as e:
        logger.error(f"Failed to run query: {e}")
        raise

if __name__ == "__main__":
    main()

#!/bin/bash
set -euo pipefail  # Exit on error, undefined variable, pipe failure

# Configuration
LOKI_VERSION="2.9.2"
GRAFANA_VERSION="10.2.3"
HELM_REPO_URL="https://grafana.github.io/helm-charts"
NAMESPACE="observability"
S3_BUCKET="my-company-loki-logs-2024"
AWS_REGION="us-east-1"

# Logging function
log() {
    echo "[$(date +'%Y-%m-%dT%H:%M:%S%z')] $1"
}

# Error handling function
error() {
    log "ERROR: $1" >&2
    exit 1
}

# Check prerequisites
check_prerequisites() {
    log "Checking prerequisites..."
    for cmd in helm kubectl aws; do
        if ! command -v "$cmd" &> /dev/null; then
            error "$cmd is not installed or not in PATH"
        fi
    done

    # Check kubectl connection
    if ! kubectl cluster-info &> /dev/null; then
        error "kubectl is not connected to a Kubernetes cluster"
    fi

    # Check AWS credentials
    if ! aws sts get-caller-identity &> /dev/null; then
        error "AWS credentials are not configured or invalid"
    fi
}

# Create S3 bucket for Loki storage
create_s3_bucket() {
    log "Creating S3 bucket $S3_BUCKET in $AWS_REGION..."
    if aws s3api head-bucket --bucket "$S3_BUCKET" 2>/dev/null; then
        log "S3 bucket $S3_BUCKET already exists"
        return
    fi

    aws s3api create-bucket \
        --bucket "$S3_BUCKET" \
        --region "$AWS_REGION" \
        --create-bucket-configuration LocationConstraint="$AWS_REGION" || error "Failed to create S3 bucket"

    # Enable versioning and lifecycle policies
    aws s3api put-bucket-versioning \
        --bucket "$S3_BUCKET" \
        --versioning-configuration Status=Enabled || error "Failed to enable S3 versioning"

    log "S3 bucket $S3_BUCKET created successfully"
}

# Add Grafana Helm repo
add_helm_repo() {
    log "Adding Grafana Helm repo..."
    helm repo add grafana "$HELM_REPO_URL" || error "Failed to add Grafana Helm repo"
    helm repo update || error "Failed to update Helm repos"
}

# Create namespace
create_namespace() {
    log "Creating namespace $NAMESPACE..."
    kubectl create namespace "$NAMESPACE" --dry-run=client -o yaml | kubectl apply -f - || error "Failed to create namespace"
}

# Deploy Loki 2.9 via Helm
deploy_loki() {
    log "Deploying Loki $LOKI_VERSION to $NAMESPACE..."
    helm upgrade --install loki grafana/loki \
        --namespace "$NAMESPACE" \
        --version "$LOKI_VERSION" \
        --values - <