How to Implement Log Aggregation with Fluent Bit 3.0 and Elasticsearch 8.12 for Kubernetes 1.37

#!/bin/bash

set -euo pipefail
trap 'echo "Prerequisite check failed at line $LINENO. Exiting."; exit 1' ERR

# Check kubectl version
echo "Checking kubectl version..."
KUBECTL_VERSION=$(kubectl version --client -o json | jq -r '.clientVersion.gitVersion')
if [[ ! "$KUBECTL_VERSION" =~ v1\.37\. ]]; then
  echo "Error: kubectl version must be 1.37.0+, got $KUBECTL_VERSION"
  exit 1
fi

# Check Helm version
echo "Checking Helm version..."
HELM_VERSION=$(helm version --short | cut -d'v' -f2)
if [[ ! "$HELM_VERSION" =~ 3\.14\. ]]; then
  echo "Error: Helm version must be 3.14.0+, got $HELM_VERSION"
  exit 1
fi

# Check cluster node count
echo "Checking cluster nodes..."
NODE_COUNT=$(kubectl get nodes --no-headers | wc -l)
if [ "$NODE_COUNT" -lt 3 ]; then
  echo "Error: Cluster must have at least 3 nodes, got $NODE_COUNT"
  exit 1
fi

# Check available storage class
echo "Checking storage class..."
kubectl get storageclass gp3 > /dev/null 2>&1 || { echo "Error: gp3 storage class not found"; exit 1; }

echo "All prerequisites validated successfully."

# Elasticsearch 8.12 Helm values
replicaCount: 3
image: "docker.elastic.co/elasticsearch/elasticsearch:8.12.0"
resources:
  requests:
    cpu: 1000m
    memory: 4Gi
  limits:
    cpu: 2000m
    memory: 8Gi
volumeClaimTemplate:
  accessModes: ["ReadWriteOnce"]
  resources:
    requests:
      storage: 100Gi
  storageClassName: gp3
esConfig:
  elasticsearch.yml: |
    xpack.security.enabled: true
    xpack.security.enrollment.enabled: true
    indices.breaker.total.use_real_memory: false
    cluster.name: k8s-1.37-log-cluster
    node.name: ${HOSTNAME}

[INPUT]
    name kubernetes
    tag kube.*
    match *
    kube_url https://kubernetes.default.svc:443
    kube_ca_file /var/run/secrets/kubernetes.io/serviceaccount/ca.crt
    kube_token_file /var/run/secrets/kubernetes.io/serviceaccount/token
    kube_tag_prefix kube.var.log.containers.
    merge_log true
    merge_log_key log_processed
    buffer.size 50MB
    buffer.max_records 100000

[FILTER]
    name kubernetes
    match kube.*
    merge_log true
    keep_log false
    labels on
    annotations on

[OUTPUT]
    name es
    match kube.*
    host elasticsearch-master.logging.svc.cluster.local
    port 9200
    index fluent-bit-k8s-1.37-%Y.%m.%d
    type _doc
    user elastic
    password ${ES_PASSWORD}
    tls on
    tls.verify off
    es_version 8

package main

import (
    "context"
    "flag"
    "fmt"
    "os"
    "io/ioutil"
    "gopkg.in/yaml.v3"
    "k8s.io/client-go/kubernetes"
    "k8s.io/client-go/tools/clientcmd"
    "k8s.io/apimachinery/pkg/api/errors"
    metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
)

// FluentBitConfig represents the structure of a Fluent Bit ConfigMap
type FluentBitConfig struct {
    ApiVersion string `yaml:"apiVersion"`
    Kind       string `yaml:"kind"`
    Metadata   struct {
        Name      string `yaml:"name"`
        Namespace string `yaml:"namespace"`
    } `yaml:"metadata"`
    Data struct {
        FluentBitConf string `yaml:"fluent-bit.conf"`
        ParsersConf   string `yaml:"parsers.conf"`
    } `yaml:"data"`
}

func main() {
    // Parse command line flags
    configPath := flag.String("config", "fluent-bit-configmap.yaml", "Path to Fluent Bit ConfigMap YAML")
    kubeconfig := flag.String("kubeconfig", os.Getenv("KUBECONFIG"), "Path to kubeconfig file")
    flag.Parse()

    // Read Fluent Bit config file
    yamlFile, err := ioutil.ReadFile(*configPath)
    if err != nil {
        fmt.Fprintf(os.Stderr, "Error reading config file: %v\n", err)
        os.Exit(1)
    }

    // Parse YAML into struct
    var fbConfig FluentBitConfig
    err = yaml.Unmarshal(yamlFile, &fbConfig)
    if err != nil {
        fmt.Fprintf(os.Stderr, "Error parsing YAML: %v\n", err)
        os.Exit(1)
    }

    // Validate ConfigMap metadata
    if fbConfig.ApiVersion != "v1" {
        fmt.Fprintf(os.Stderr, "Invalid apiVersion: expected v1, got %s\n", fbConfig.ApiVersion)
        os.Exit(1)
    }
    if fbConfig.Kind != "ConfigMap" {
        fmt.Fprintf(os.Stderr, "Invalid kind: expected ConfigMap, got %s\n", fbConfig.Kind)
        os.Exit(1)
    }
    if fbConfig.Metadata.Namespace != "logging" {
        fmt.Fprintf(os.Stderr, "Invalid namespace: expected logging, got %s\n", fbConfig.Metadata.Namespace)
        os.Exit(1)
    }

    // Connect to Kubernetes cluster
    config, err := clientcmd.BuildConfigFromFlags("", *kubeconfig)
    if err != nil {
        fmt.Fprintf(os.Stderr, "Error building kubeconfig: %v\n", err)
        os.Exit(1)
    }
    clientset, err := kubernetes.NewForConfig(config)
    if err != nil {
        fmt.Fprintf(os.Stderr, "Error creating Kubernetes client: %v\n", err)
        os.Exit(1)
    }

    // Check if Fluent Bit DaemonSet exists in the namespace
    _, err = clientset.AppsV1().DaemonSets(fbConfig.Metadata.Namespace).Get(
        context.Background(),
        "fluent-bit",
        metav1.GetOptions{},
    )
    if err != nil {
        if errors.IsNotFound(err) {
            fmt.Fprintf(os.Stderr, "Fluent Bit DaemonSet not found in namespace %s\n", fbConfig.Metadata.Namespace)
            os.Exit(1)
        }
        fmt.Fprintf(os.Stderr, "Error checking DaemonSet: %v\n", err)
        os.Exit(1)
    }

    fmt.Println("Fluent Bit configuration is valid and DaemonSet is running.")
}

[STORAGE]
    storage.path /var/log/fluent-bit
    storage.sync normal
    storage.backlog.mem_limit 1GB

[INPUT]
    name kubernetes
    buffer.size 50MB
    buffer.max_records 100000

PUT _ilm/policy/fluent-bit-logs
{
  "policy": {
    "phases": {
      "hot": {
        "actions": {
          "rollover": {
            "max_size": "50GB",
            "max_age": "1d"
          },
          "set_priority": {
            "priority": 100
          }
        }
      },
      "warm": {
        "min_age": "3d",
        "actions": {
          "shrink": {
            "number_of_shards": 1
          },
          "set_priority": {
            "priority": 50
          }
        }
      },
      "delete": {
        "min_age": "30d",
        "actions": {
          "delete": {}
        }
      }
    }
  }
}

kubectl run fluent-bit-dry-run --image=fluent/fluent-bit:3.0.2 --rm -it \
  --namespace logging \
  --overrides='{
    "spec": {
      "containers": [{
        "name": "fluent-bit",
        "image": "fluent/fluent-bit:3.0.2",
        "args": ["--dry-run", "-c", "/fluent-bit/etc/fluent-bit.conf"],
        "volumeMounts": [{
          "name": "config",
          "mountPath": "/fluent-bit/etc"
        }]
      }],
      "volumes": [{
        "name": "config",
        "configMap": {
          "name": "fluent-bit-config"
        }
      }]
    }
  }'

import os
import sys
import time
import json
import random
import string
import argparse
from elasticsearch import Elasticsearch, helpers
from datetime import datetime
import logging

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

def generate_log_line():
    """Generate a sample Kubernetes log line matching Fluent Bit's output format"""
    timestamp = datetime.utcnow().isoformat() + "Z"
    pod_name = f"app-{random.randint(1, 100)}-{''.join(random.choices(string.ascii_lowercase, k=5))}"
    namespace = random.choice(["default", "logging", "app-prod", "app-staging"])
    container = random.choice(["app", "sidecar", "init"])
    log_level = random.choice(["INFO", "WARN", "ERROR", "DEBUG"])
    message = f"Sample log message {''.join(random.choices(string.ascii_letters + string.digits, k=20))}"

    return {
        "@timestamp": timestamp,
        "kubernetes": {
            "pod_name": pod_name,
            "namespace": namespace,
            "container": container
        },
        "log_level": log_level,
        "message": message,
        "cluster": "eks-1-37-prod"
    }

def benchmark_elasticsearch(host, port, index_name, num_logs, batch_size):
    """Benchmark Elasticsearch 8.12 index throughput"""
    es = Elasticsearch(f"https://{host}:{port}",
        ca_certs="/etc/elasticsearch/certs/ca.crt",
        basic_auth=("elastic", os.getenv("ES_PASSWORD")),
        verify_certs=True
    )

    # Check if Elasticsearch is reachable
    try:
        info = es.info()
        logger.info(f"Connected to Elasticsearch {info['version']['number']}")
    except Exception as e:
        logger.error(f"Failed to connect to Elasticsearch: {e}")
        sys.exit(1)

    # Create index with 8.12-optimized settings
    index_settings = {
        "settings": {
            "number_of_shards": 3,
            "number_of_replicas": 1,
            "index.refresh_interval": "5s",
            "index.codec": "best_compression"  # New in ES 8.12
        },
        "mappings": {
            "properties": {
                "@timestamp": {"type": "date"},
                "kubernetes.pod_name": {"type": "keyword"},
                "kubernetes.namespace": {"type": "keyword"},
                "log_level": {"type": "keyword"}
            }
        }
    }

    try:
        if not es.indices.exists(index=index_name):
            es.indices.create(index=index_name, body=index_settings)
            logger.info(f"Created index {index_name}")
    except Exception as e:
        logger.error(f"Failed to create index: {e}")
        sys.exit(1)

    # Generate log batch generator
    def log_generator():
        for i in range(num_logs):
            yield {
                "_index": index_name,
                "_source": generate_log_line()
            }
            if i % 10000 == 0:
                logger.info(f"Generated {i} logs")

    # Run benchmark
    logger.info(f"Indexing {num_logs} logs with batch size {batch_size}")
    start_time = time.time()
    count = 0
    for ok, item in helpers.streaming_bulk(es, log_generator(), chunk_size=batch_size):
        if not ok:
            logger.error(f"Failed to index document: {item}")
        else:
            count += 1
    end_time = time.time()

    # Calculate metrics
    duration = end_time - start_time
    throughput = count / duration
    logger.info(f"Indexed {count} logs in {duration:.2f}s")
    logger.info(f"Throughput: {throughput:.2f} logs/sec")
    logger.info(f"Average latency per log: {(duration / count) * 1000:.2f}ms")

    # Cleanup
    es.indices.delete(index=index_name)
    logger.info(f"Deleted test index {index_name}")

if __name__ == "__main__":
    parser = argparse.ArgumentParser(description="Benchmark Elasticsearch 8.12 log indexing throughput")
    parser.add_argument("--host", default="elasticsearch.logging.svc.cluster.local", help="Elasticsearch host")
    parser.add_argument("--port", type=int, default=9200, help="Elasticsearch port")
    parser.add_argument("--index", default="fluent-bit-benchmark", help="Test index name")
    parser.add_argument("--num-logs", type=int, default=1000000, help="Number of logs to index")
    parser.add_argument("--batch-size", type=int, default=5000, help="Bulk batch size")
    args = parser.parse_args()

    if not os.getenv("ES_PASSWORD"):
        logger.error("ES_PASSWORD environment variable is not set")
        sys.exit(1)

    benchmark_elasticsearch(args.host, args.port, args.index, args.num_logs, args.batch_size)

#!/bin/bash

set -euo pipefail
trap 'echo "Error occurred at line $LINENO. Exiting."; exit 1' ERR

# Configuration variables
KUBE_NAMESPACE="logging"
FLUENT_BIT_VERSION="3.0.2"
ELASTICSEARCH_VERSION="8.12.0"
HELM_REPO_ELASTIC="https://artifacthub.io/packages/helm/elastic/elasticsearch"
HELM_REPO_FLUENT="https://fluent.github.io/helm-charts"

# Step 1: Create namespace
echo "Creating namespace $KUBE_NAMESPACE..."
kubectl create namespace "$KUBE_NAMESPACE" --dry-run=client -o yaml | kubectl apply -f -

# Step 2: Add Helm repos
echo "Adding Helm repositories..."
helm repo add elastic "$HELM_REPO_ELASTIC" || { echo "Failed to add elastic repo"; exit 1; }
helm repo add fluent "$HELM_REPO_FLUENT" || { echo "Failed to add fluent repo"; exit 1; }
helm repo update || { echo "Failed to update Helm repos"; exit 1; }

# Step 3: Deploy Elasticsearch 8.12
echo "Deploying Elasticsearch $ELASTICSEARCH_VERSION..."
helm upgrade --install elasticsearch elastic/elasticsearch \
    --namespace "$KUBE_NAMESPACE" \
    --version "8.12.0" \
    --values - <<EOF
replicaCount: 3
image: "docker.elastic.co/elasticsearch/elasticsearch:8.12.0"
resources:
  requests:
    cpu: 1000m
    memory: 4Gi
  limits:
    cpu: 2000m
    memory: 8Gi
volumeClaimTemplate:
  accessModes: ["ReadWriteOnce"]
  resources:
    requests:
      storage: 100Gi
  storageClassName: gp3
esConfig:
  elasticsearch.yml: |
    xpack.security.enabled: true
    xpack.security.enrollment.enabled: true
    indices.breaker.total.use_real_memory: false
EOF

# Wait for Elasticsearch to be ready
echo "Waiting for Elasticsearch to be ready..."
kubectl rollout status statefulset/elasticsearch-master --namespace "$KUBE_NAMESPACE" --timeout=600s || { echo "Elasticsearch rollout failed"; exit 1; }

# Step 4: Get Elasticsearch password
echo "Retrieving Elasticsearch password..."
ES_PASSWORD=$(kubectl get secrets --namespace "$KUBE_NAMESPACE" elasticsearch-master-credentials -o jsonpath="{.data.password}" | base64 --decode)
if [ -z "$ES_PASSWORD" ]; then
    echo "Failed to retrieve Elasticsearch password"
    exit 1
fi
export ES_PASSWORD

# Step 5: Deploy Fluent Bit 3.0
echo "Deploying Fluent Bit $FLUENT_BIT_VERSION..."
helm upgrade --install fluent-bit fluent/fluent-bit \
    --namespace "$KUBE_NAMESPACE" \
    --version "0.39.0" \
    --values - <<EOF
image:
  repository: fluent/fluent-bit
  tag: 3.0.2
resources:
  requests:
    cpu: 100m
    memory: 128Mi
  limits:
    cpu: 500m
    memory: 512Mi
input:
  - name: kubernetes
    tag: kube.*
    match: *
    kube_url: https://kubernetes.default.svc:443
    kube_ca_file: /var/run/secrets/kubernetes.io/serviceaccount/ca.crt
    kube_token_file: /var/run/secrets/kubernetes.io/serviceaccount/token
    kube_tag_prefix: kube.var.log.containers.
    merge_log: true
    merge_log_key: log_processed
output:
  - name: es
    match: kube.*
    host: elasticsearch-master.logging.svc.cluster.local
    port: 9200
    index: fluent-bit-k8s-1.37
    type: _doc
    user: elastic
    password: "$ES_PASSWORD"
    tls: on
    tls.verify: off
EOF

# Wait for Fluent Bit to be ready
echo "Waiting for Fluent Bit to be ready..."
kubectl rollout status daemonset/fluent-bit --namespace "$KUBE_NAMESPACE" --timeout=300s || { echo "Fluent Bit rollout failed"; exit 1; }

# Step 6: Validate pipeline
echo "Validating log pipeline..."
sleep 60  # Wait for logs to be indexed
LOG_COUNT=$(curl -s -u "elastic:$ES_PASSWORD" "https://elasticsearch-master.logging.svc.cluster.local:9200/fluent-bit-k8s-1.37/_count" | jq -r '.count')
if [ "$LOG_COUNT" -gt 0 ]; then
    echo "Pipeline validation successful: $LOG_COUNT logs indexed"
else
    echo "Pipeline validation failed: no logs found"
    exit 1
fi

echo "Deployment complete! Log into Kibana to view logs."

[FILTER]
    name multiline
    match kube.*
    multiline.key_content log
    multiline.parser java

├── deploy/
│   ├── elasticsearch/
│   │   └── values.yaml
│   └── fluent-bit/
│       ├── configmap.yaml
│       └── values.yaml
├── scripts/
│   ├── validate-config.go
│   ├── benchmark-es.py
│   └── deploy-stack.sh
├── terraform/
│   ├── main.tf
│   └── variables.tf
└── README.md