Opinion: Vector 0.38 Is Better Than Fluentd 1.16 for Cloud-Native Log Forwarding

[sources.kubernetes_logs]
type = "kubernetes"
namespaces = ["production", "staging"]
exclude_paths = ["/var/log/pods/*/fluentd*"]

[sinks.elasticsearch]
type = "elasticsearch"
inputs = ["kubernetes_logs"]
endpoints = ["https://es.example.com:9200"]
index = "logs-%Y-%m-%d"

  @type kubernetes
  @id kubernetes_logs
  @label @k8s
  namespaces production,staging
  exclude_path /var/log/pods/*/fluentd*

    @type json
    time_key time
    time_format %ISO8601


    @type file
    path /var/log/fluentd-buffer/kubernetes
    flush_mode interval
    flush_interval 5s
    chunk_limit_size 2M
    total_limit_size 100M




  @type elasticsearch
  @id elasticsearch
  host es.example.com
  port 9200
  index_name logs-${tag_parts[3]}-%Y-%m-%d

    @type file
    path /var/log/fluentd-buffer/elasticsearch
    flush_mode interval
    flush_interval 5s
    chunk_limit_size 2M
    total_limit_size 100M

#!/usr/bin/env python3
"""Vector 0.38 vs Fluentd 1.16 Throughput Benchmark Tool
Requires: pip install requests pyyaml kubernetes
Benchmark configuration: 1KB log events, 60-second test duration, 8 vCPU node
"""

import time
import json
import random
import string
import subprocess
import argparse
import logging
from typing import Dict, List, Optional

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

def generate_log_event(size_kb: int = 1) -> str:
    """Generate a random log event of specified size in KB"""
    # Base log structure with common fields
    base_log = {
        "timestamp": time.time_ns(),
        "level": random.choice(["INFO", "WARN", "ERROR"]),
        "service": random.choice(["api", "worker", "scheduler"]),
        "pod": f"pod-{random.randint(1000, 9999)}",
        "namespace": "production",
        "message": ""
    }
    # Fill message to reach target size (1KB = 1024 bytes)
    target_bytes = size_kb * 1024
    current_size = len(json.dumps(base_log).encode("utf-8"))
    padding_needed = target_bytes - current_size
    if padding_needed > 0:
        base_log["message"] = ''.join(random.choices(string.ascii_letters, k=padding_needed))
    return json.dumps(base_log)

def run_vector_benchmark(vector_endpoint: str, duration_sec: int = 60) -> Dict:
    """Send logs to Vector 0.38 HTTP sink and measure throughput"""
    import requests
    logger.info(f"Starting Vector benchmark to {vector_endpoint} for {duration_sec}s")
    start_time = time.time()
    event_count = 0
    errors = 0

    while (time.time() - start_time) < duration_sec:
        try:
            event = generate_log_event()
            response = requests.post(
                vector_endpoint,
                data=event,
                headers={"Content-Type": "application/json"},
                timeout=1
            )
            if response.status_code == 200:
                event_count += 1
            else:
                errors += 1
                logger.warning(f"Vector returned {response.status_code}")
        except Exception as e:
            errors += 1
            logger.error(f"Vector benchmark error: {str(e)}")

    elapsed = time.time() - start_time
    return {
        "tool": "Vector 0.38",
        "throughput_eps": event_count / elapsed,
        "total_events": event_count,
        "errors": errors,
        "elapsed_sec": elapsed
    }

def run_fluentd_benchmark(fluentd_endpoint: str, duration_sec: int = 60) -> Dict:
    """Send logs to Fluentd 1.16 forward protocol and measure throughput"""
    import socket
    logger.info(f"Starting Fluentd benchmark to {fluentd_endpoint} for {duration_sec}s")
    start_time = time.time()
    event_count = 0
    errors = 0
    host, port = fluentd_endpoint.split(":")
    port = int(port)

    while (time.time() - start_time) < duration_sec:
        try:
            event = generate_log_event()
            # Fluentd forward protocol: [tag, time, record]
            tag = "benchmark.logs"
            timestamp = int(time.time())
            payload = json.dumps([tag, timestamp, json.loads(event)])
            sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
            sock.settimeout(1)
            sock.connect((host, port))
            sock.sendall(payload.encode("utf-8"))
            sock.close()
            event_count += 1
        except Exception as e:
            errors += 1
            logger.error(f"Fluentd benchmark error: {str(e)}")

    elapsed = time.time() - start_time
    return {
        "tool": "Fluentd 1.16",
        "throughput_eps": event_count / elapsed,
        "total_events": event_count,
        "errors": errors,
        "elapsed_sec": elapsed
    }

if __name__ == "__main__":
    parser = argparse.ArgumentParser(description="Vector vs Fluentd Benchmark")
    parser.add_argument("--vector-endpoint", default="http://localhost:8686", help="Vector HTTP sink URL")
    parser.add_argument("--fluentd-endpoint", default="localhost:24224", help="Fluentd forward endpoint")
    parser.add_argument("--duration", type=int, default=60, help="Test duration in seconds")
    args = parser.parse_args()

    logger.info("Starting benchmark suite")
    vector_results = run_vector_benchmark(args.vector_endpoint, args.duration)
    fluentd_results = run_fluentd_benchmark(args.fluentd_endpoint, args.duration)

    logger.info("Benchmark Results:")
    logger.info(f"Vector 0.38: {vector_results['throughput_eps']:.2f} events/sec, {vector_results['errors']} errors")
    logger.info(f"Fluentd 1.16: {fluentd_results['throughput_eps']:.2f} events/sec, {fluentd_results['errors']} errors")
    logger.info(f"Vector is {vector_results['throughput_eps'] / fluentd_results['throughput_eps']:.2f}x faster")

package main

import (
    "fmt"
    "os"
    "io/ioutil"
    "log"
    "errors"
    "github.com/BurntSushi/toml" // TOML parser for Vector configs
)

// VectorConfig represents the top-level structure of a Vector 0.38 configuration file
type VectorConfig struct {
    Sources  map[string]Source  `toml:"sources"`
    Sinks    map[string]Sink    `toml:"sinks"`
    Transforms map[string]Transform `toml:"transforms"`
}

// Source defines a Vector log source (e.g., kubernetes, file, stdin)
type Source struct {
    Type        string                 `toml:"type"`
    Options     map[string]interface{} `toml:"options"`
}

// Sink defines a Vector log destination (e.g., elasticsearch, s3, stdout)
type Sink struct {
    Type        string                 `toml:"type"`
    Inputs      []string               `toml:"inputs"`
    Options     map[string]interface{} `toml:"options"`
}

// Transform defines a Vector processing step (e.g., filter, remap)
type Transform struct {
    Type        string                 `toml:"type"`
    Inputs      []string               `toml:"inputs"`
    Options     map[string]interface{} `toml:"options"`
}

// validateVectorConfig checks a Vector 0.38 config for common errors
func validateVectorConfig(configPath string) error {
    // Read config file
    data, err := ioutil.ReadFile(configPath)
    if err != nil {
        return fmt.Errorf("failed to read config file: %w", err)
    }

    // Parse TOML
    var config VectorConfig
    if _, err := toml.Decode(string(data), &config); err != nil {
        return fmt.Errorf("failed to parse TOML: %w", err)
    }

    // Validate at least one source exists
    if len(config.Sources) == 0 {
        return errors.New("no sources defined in config")
    }

    // Validate all sinks reference valid inputs
    for sinkName, sink := range config.Sinks {
        for _, input := range sink.Inputs {
            // Check if input is a source or transform
            _, isSource := config.Sources[input]
            _, isTransform := config.Transforms[input]
            if !isSource && !isTransform {
                return fmt.Errorf("sink %s references invalid input %s", sinkName, input)
            }
        }
    }

    // Validate all transforms reference valid inputs
    for transformName, transform := range config.Transforms {
        for _, input := range transform.Inputs {
            _, isSource := config.Sources[input]
            _, isTransform := config.Transforms[input]
            if !isSource && !isTransform {
                return fmt.Errorf("transform %s references invalid input %s", transformName, input)
            }
        }
    }

    // Check for required Kubernetes source fields if type is kubernetes
    for sourceName, source := range config.Sources {
        if source.Type == "kubernetes" {
            if source.Options["namespaces"] == nil && source.Options["pods"] == nil {
                log.Printf("warning: kubernetes source %s has no namespace/pod filters, may collect all logs", sourceName)
            }
        }
    }

    return nil
}

func main() {
    if len(os.Args) < 2 {
        log.Fatal("usage: vector-config-validator ")
    }
    configPath := os.Args[1]

    log.Printf("Validating Vector config: %s", configPath)
    err := validateVectorConfig(configPath)
    if err != nil {
        log.Fatalf("Validation failed: %v", err)
    }

    log.Println("Config validation successful")
    // Print config summary
    data, _ := ioutil.ReadFile(configPath)
    var config VectorConfig
    toml.Decode(string(data), &config)
    fmt.Printf("Sources: %d\n", len(config.Sources))
    fmt.Printf("Sinks: %d\n", len(config.Sinks))
    fmt.Printf("Transforms: %d\n", len(config.Transforms))
}

use std::net::TcpStream;
use std::io::Write;
use std::time::{SystemTime, UNIX_EPOCH};
use serde_json::{json, Value};
use std::error::Error;
use std::thread;
use std::time::Duration;

// Fluentd forward protocol message structure: [tag, timestamp, record]
type FluentdMessage = (String, i64, Value);

// Generate a random log event for testing
fn generate_log_event() -> Value {
    let timestamp = SystemTime::now()
        .duration_since(UNIX_EPOCH)
        .unwrap()
        .as_nanos() as i64;

    json!({
        "timestamp": timestamp,
        "level": if rand::random() { "INFO" } else { "ERROR" },
        "service": "benchmark-client",
        "pod": format!("pod-{}", rand::random::()),
        "namespace": "production",
        "message": "test log message for Fluentd 1.16 benchmarking"
    })
}

// Send a single message to Fluentd via forward protocol
fn send_fluentd_message(stream: &mut TcpStream, tag: &str, record: Value) -> Result<(), Box> {
    let timestamp = SystemTime::now()
        .duration_since(UNIX_EPOCH)
        .unwrap()
        .as_secs() as i64;

    // Construct forward protocol payload: [tag, timestamp, record]
    let payload = serde_json::to_vec(&json!([tag, timestamp, record]))?;

    // Send payload with length prefix (Fluentd expects 4-byte big-endian length)
    let len = payload.len() as u32;
    stream.write_all(&len.to_be_bytes())?;
    stream.write_all(&payload)?;
    stream.flush()?;

    Ok(())
}

// Run Fluentd benchmark client
fn run_fluentd_benchmark(host: &str, port: u16, duration_sec: u64) -> Result<(), Box> {
    let address = format!("{}:{}", host, port);
    let start_time = SystemTime::now();
    let mut event_count = 0;
    let mut error_count = 0;

    println!("Starting Fluentd 1.16 benchmark to {}", address);

    while SystemTime::now()
        .duration_since(start_time)
        .unwrap()
        .as_secs() < duration_sec
    {
        // Reconnect for each message to simulate real-world client behavior
        match TcpStream::connect(&address) {
            Ok(mut stream) => {
                let event = generate_log_event();
                match send_fluentd_message(&mut stream, "benchmark.logs", event) {
                    Ok(_) => event_count += 1,
                    Err(e) => {
                        error_count += 1;
                        eprintln!("Failed to send message: {}", e);
                    }
                }
            }
            Err(e) => {
                error_count += 1;
                eprintln!("Failed to connect to Fluentd: {}", e);
                thread::sleep(Duration::from_millis(100));
            }
        }
    }

    let elapsed = SystemTime::now()
        .duration_since(start_time)
        .unwrap()
        .as_secs_f64();

    println!("Fluentd 1.16 Benchmark Results:");
    println!("Duration: {:.2}s", elapsed);
    println!("Total events: {}", event_count);
    println!("Throughput: {:.2} events/sec", event_count as f64 / elapsed);
    println!("Errors: {}", error_count);

    Ok(())
}

fn main() -> Result<(), Box> {
    // Parse command line arguments
    let args: Vec = std::env::args().collect();
    if args.len() < 3 {
        eprintln!("Usage: {}   [duration_sec]", args[0]);
        return Ok(());
    }

    let host = &args[1];
    let port: u16 = args[2].parse()?;
    let duration = if args.len() > 3 { args[3].parse().unwrap_or(60) } else { 60 };

    run_fluentd_benchmark(host, port, duration)?;

    Ok(())
}

# Vector VRL script to process payment logs
if .message =~ r'credit_card: (\d{16})' {
  .credit_card = r'credit_card: (\d{16})'
  .credit_card_masked = "****-****-****-" + .credit_card[12:16]
  del(.credit_card)
}
if .level == "ERROR" {
  .alert = true
}

[sources.k8s_logs]
type = "kubernetes"
# Automatically collect logs from all pods in production namespace
namespaces = ["production"]
# Exclude Vector's own logs to prevent feedback loops
exclude_paths = ["/var/log/pods/*/vector*"]

# Fluentd config to forward logs to Vector

  @type forward

    host vector-sidecar
    port 24224



# Vector config to ingest Fluentd logs
[sources.fluentd_forward]
type = "fluentd"
listen_port = 24224

[sinks.elasticsearch]
type = "elasticsearch"
inputs = ["fluentd_forward"]
endpoints = ["https://es.example.com:9200"]