Fluentbit Generator

# Fluent Bit Configuration: Application Logs with Multi-line Parsing
# This configuration tails application logs with multi-line support for stack traces
# and forwards to Elasticsearch.

[SERVICE]
    Flush         1
    Log_Level     info
    Daemon        Off
    HTTP_Server   On
    HTTP_Port     2020
    Parsers_File  parsers.conf

[INPUT]
    Name              tail
    Tag               app.myapp
    Path              /var/log/app/*.log
    Parser            json
    Multiline.Parser  multiline-java
    DB                /var/log/flb_app.db
    Mem_Buf_Limit     100MB
    Skip_Long_Lines   On
    Read_from_Head    Off

[FILTER]
    Name          modify
    Match         *
    Add           app_name myapp
    Add           environment production
    Add           version 1.0.0

[FILTER]
    Name          parser
    Match         *
    Key_Name      log
    Parser        json
    Reserve_Data  On
    Preserve_Key  Off

[FILTER]
    Name          grep
    Match         *
    # Only forward ERROR and above
    Regex         level (ERROR|FATAL|CRITICAL)

[OUTPUT]
    Name              es
    Match             *
    Host              elasticsearch
    Port              9200
    Index             app-logs
    Retry_Limit       3
    storage.total_limit_size 10M
    tls               On
    tls.verify        On

# Fluent Bit Configuration: Kubernetes to AWS CloudWatch Logs
# This configuration collects Kubernetes logs and forwards them to CloudWatch.

[SERVICE]
    Flush         1
    Log_Level     info
    Daemon        Off
    HTTP_Server   On
    HTTP_Port     2020
    storage.metrics on
    Parsers_File  parsers.conf

[INPUT]
    Name              tail
    Tag               kube.*
    Path              /var/log/containers/*.log
    Exclude_Path      /var/log/containers/*fluent-bit*.log
    Parser            docker
    DB                /var/log/flb_kube.db
    Mem_Buf_Limit     50MB
    Skip_Long_Lines   On

[FILTER]
    Name                kubernetes
    Match               kube.*
    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           On
    Keep_Log            Off
    Labels              On

[FILTER]
    Name          modify
    Match         *
    Add           cluster my-eks-cluster
    Add           region us-east-1

[OUTPUT]
    Name              cloudwatch_logs
    Match             *
    region            us-east-1
    log_group_name    /aws/kubernetes/my-cluster/logs
    log_stream_prefix from-fluent-bit-
    auto_create_group On
    Retry_Limit       3
    # IAM role authentication (no hardcoded credentials)

# Fluent Bit Configuration: File Tailing to S3
# This configuration tails log files and archives them to Amazon S3.

[SERVICE]
    Flush         5
    Log_Level     info
    Daemon        Off
    HTTP_Server   On
    HTTP_Port     2020
    Parsers_File  parsers.conf

[INPUT]
    Name              tail
    Tag               files.application
    Path              /var/log/app/*.log
    DB                /var/log/flb_files.db
    Mem_Buf_Limit     100MB
    Skip_Long_Lines   On
    Refresh_Interval  10

[FILTER]
    Name          modify
    Match         *
    Add           source file-collector
    Add           host ${HOSTNAME}

[FILTER]
    Name          parser
    Match         *
    Key_Name      log
    Parser        json
    Reserve_Data  On

[OUTPUT]
    Name              s3
    Match             *
    bucket            my-logs-archive-bucket
    region            us-east-1
    total_file_size   100M
    upload_timeout    10m
    use_put_object    Off
    compression       gzip
    # S3 key format with time-based partitioning
    s3_key_format     /logs/app/%Y/%m/%d/%H-%M-%S-$UUID.gz
    Retry_Limit       3
    # IAM role authentication (no hardcoded credentials)
    # Credentials loaded from environment or IAM role

# Fluent Bit Configuration: Full Production Setup
# This is a complete, production-ready configuration with all best practices:
# - Proper buffer limits and retry logic
# - Security (TLS, no hardcoded credentials)
# - Performance optimization (compression, batching)
# - Reliability (filesystem buffering, health checks)
# - Observability (metrics, health endpoints)

[SERVICE]
    # Flush logs every 1 second
    Flush         1

    # Log level for production
    Log_Level     info

    # Run in foreground (for containers)
    Daemon        Off

    # HTTP server for health checks and metrics
    HTTP_Server   On
    HTTP_Listen   0.0.0.0
    HTTP_Port     2020

    # Enable storage metrics for monitoring
    storage.metrics on

    # Parser configuration
    Parsers_File  parsers.conf

    # Grace period for shutdown (seconds)
    Grace         30

    # Enable hot reload
    Hot_Reload    On

# ===== INPUTS =====

[INPUT]
    Name              tail
    Tag               kube.*
    Path              /var/log/containers/*.log
    # Prevent log loops by excluding Fluent Bit's own logs
    Exclude_Path      /var/log/containers/*fluent-bit*.log,/var/log/containers/*fluentbit*.log
    Parser            docker
    # Position database for crash recovery
    DB                /var/log/flb_kube.db
    # Memory buffer limit per input (prevents OOM)
    Mem_Buf_Limit     50MB
    # Skip lines longer than 32KB (prevents hang)
    Skip_Long_Lines   On
    # Refresh file list every 10 seconds
    Refresh_Interval  10
    # Don't read from beginning (only new logs)
    Read_from_Head    Off
    # Rotate wait time
    Rotate_Wait       30
    # Skip empty lines
    Skip_Empty_Lines  On

# ===== FILTERS =====

# Kubernetes metadata enrichment
[FILTER]
    Name                kubernetes
    Match               kube.*
    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 parsed JSON logs
    Merge_Log           On
    # Keep original log field for downstream grep filters
    Keep_Log            On
    # Honor pod annotations for parser
    K8S-Logging.Parser  On
    # Honor pod annotations for exclude
    K8S-Logging.Exclude On
    # Include pod labels
    Labels              On
    # Exclude annotations to reduce size
    Annotations         Off
    # Use TLS
    tls.verify          On
    # Cache settings
    Use_Kubelet         Off
    Buffer_Size         0

# Add cluster and environment identifiers
[FILTER]
    Name          modify
    Match         *
    Add           cluster_name production-eks-cluster
    Add           environment production
    Add           region us-east-1
    Add           version 1.0.0

# Lift nested Kubernetes fields to top level with prefix
[FILTER]
    Name          nest
    Match         *
    Operation     lift
    Nested_under  kubernetes
    Add_prefix    k8s_

# Filter out noisy logs (health checks, metrics endpoints)
[FILTER]
    Name          grep
    Match         *
    Exclude       log /health
    Exclude       log /metrics
    Exclude       log /readyz
    Exclude       log /livez

# Rate limiting: max 1000 logs per 5 second window
[FILTER]
    Name          throttle
    Match         *
    Rate          1000
    Window        5
    Interval      1m
    Print_Status  true

# ===== OUTPUTS =====

# Primary: Elasticsearch for real-time search and analysis
[OUTPUT]
    Name              es
    Match             *
    Host              elasticsearch.logging.svc.cluster.local
    Port              9200
    # Use Logstash format for time-based indices
    Logstash_Format   On
    Logstash_Prefix   k8s-prod
    # Daily indices
    Logstash_DateFormat %Y.%m.%d
    # Include tag in index
    Include_Tag_Key   On
    Tag_Key           @tag
    # Time key for @timestamp field
    Time_Key          @timestamp
    # Generate unique document IDs
    Generate_ID       On
    # Replace dots in field names with underscores
    Replace_Dots      On
    # Retry configuration
    Retry_Limit       3
    # Filesystem buffering for reliability
    storage.type      filesystem
    storage.path      /var/log/fluent-bit-buffer/
    storage.total_limit_size 5G
    # TLS configuration
    tls               On
    tls.verify        On
    # Authentication via environment variables
    HTTP_User         ${ES_USER}
    HTTP_Passwd       ${ES_PASSWORD}
    # Buffer configuration
    Buffer_Size       False
    Type              _doc
    # Trace errors
    Trace_Error       On
    Trace_Output      On

# Secondary: S3 for long-term archival and compliance
[OUTPUT]
    Name              s3
    Match             *
    bucket            prod-k8s-logs-archive
    region            us-east-1
    # Upload when file reaches 100MB
    total_file_size   100M
    # Or after 10 minutes
    upload_timeout    10m
    # Use multipart upload (recommended)
    use_put_object    Off
    # Compression (reduces costs)
    compression       gzip
    # S3 key format with time-based partitioning
    s3_key_format     /logs/k8s/%Y/%m/%d/$TAG[0]/%H-%M-%S-$UUID.gz
    # Retry configuration
    Retry_Limit       3
    # Filesystem buffering
    storage.type      filesystem
    storage.path      /var/log/fluent-bit-buffer-s3/
    storage.total_limit_size 2G
    # IAM role authentication (no hardcoded credentials)
    # AWS credentials loaded from environment or IAM role
    # Store class for cost optimization
    store_dir         /tmp/fluent-bit/s3
    # Content type
    content_type      application/gzip

# Tertiary: CloudWatch for AWS-native monitoring
[OUTPUT]
    Name              cloudwatch_logs
    Match             kube.*
    region            us-east-1
    log_group_name    /aws/kubernetes/production-cluster
    log_stream_prefix fluent-bit-
    auto_create_group On
    # Retry configuration
    Retry_Limit       3
    # Use STS for authentication (IAM role)
    # Log key for message field
    log_key           log

# Fluent Bit Configuration: HTTP Webhook to Kafka
# This configuration receives logs via HTTP webhook and forwards them to Kafka.

[SERVICE]
    Flush         1
    Log_Level     info
    Daemon        Off
    HTTP_Server   On
    HTTP_Port     2020
    Parsers_File  parsers.conf

[INPUT]
    Name          http
    Tag           webhook.events
    Listen        0.0.0.0
    Port          9880
    Buffer_Size   32KB

[FILTER]
    Name          modify
    Match         *
    Add           source webhook
    Add           collector fluent-bit
    Add           timestamp ${TIME}

[FILTER]
    Name          parser
    Match         *
    Key_Name      log
    Parser        json
    Reserve_Data  On

[OUTPUT]
    Name              kafka
    Match             *
    Brokers           kafka-broker-1:9092,kafka-broker-2:9092,kafka-broker-3:9092
    Topics            application-logs
    Format            json
    Timestamp_Key     @timestamp
    Retry_Limit       3
    # Kafka producer configuration
    rdkafka.queue.buffering.max.messages     100000
    rdkafka.request.required.acks            1
    rdkafka.message.send.max.retries         3
    rdkafka.compression.type                 gzip

# Fluent Bit Configuration: Kubernetes to Elasticsearch
# This configuration collects logs from Kubernetes pods, enriches them with metadata,
# and forwards them to Elasticsearch with Logstash format indexing.

[SERVICE]
    # Flush interval in seconds
    Flush        1

    # Log level: off, error, warn, info, debug, trace
    Log_Level    info

    # Daemon mode (Off for containers)
    Daemon       Off

    # HTTP server for health checks and metrics
    HTTP_Server  On
    HTTP_Listen  0.0.0.0
    HTTP_Port    2020

    # Enable storage metrics
    storage.metrics on

    # Parser configuration file
    Parsers_File parsers.conf

[INPUT]
    Name              tail
    Tag               kube.*
    Path              /var/log/containers/*.log
    Exclude_Path      /var/log/containers/*fluent-bit*.log
    Parser            docker
    DB                /var/log/flb_kube.db
    Mem_Buf_Limit     50MB
    Skip_Long_Lines   On
    Refresh_Interval  10
    Read_from_Head    Off

[FILTER]
    Name                kubernetes
    Match               kube.*
    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           On
    Keep_Log            Off
    K8S-Logging.Parser  On
    K8S-Logging.Exclude On
    Labels              On
    Annotations         Off
    Buffer_Size         0

[FILTER]
    Name          modify
    Match         *
    Add           cluster_name production-cluster
    Add           environment production

[FILTER]
    Name          nest
    Match         *
    Operation     lift
    Nested_under  kubernetes
    Add_prefix    k8s_

[OUTPUT]
    Name              es
    Match             *
    Host              elasticsearch.logging.svc
    Port              9200
    Logstash_Format   On
    Logstash_Prefix   k8s
    Retry_Limit       3
    storage.total_limit_size 5M
    tls               On
    tls.verify        On
    Buffer_Size       False
    Type              _doc

# Fluent Bit Configuration: Kubernetes to Grafana Loki
# This configuration collects logs from Kubernetes pods, enriches them with metadata,
# and forwards them to Loki with proper labels.

[SERVICE]
    Flush        1
    Log_Level    info
    Daemon       Off
    HTTP_Server  On
    HTTP_Listen  0.0.0.0
    HTTP_Port    2020
    storage.metrics on
    Parsers_File parsers.conf

[INPUT]
    Name              tail
    Tag               kube.*
    Path              /var/log/containers/*.log
    Exclude_Path      /var/log/containers/*fluent-bit*.log
    Parser            docker
    DB                /var/log/flb_kube.db
    Mem_Buf_Limit     50MB
    Skip_Long_Lines   On
    Refresh_Interval  10

[FILTER]
    Name                kubernetes
    Match               kube.*
    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           On
    Keep_Log            Off
    K8S-Logging.Parser  On
    K8S-Logging.Exclude On
    Labels              On

[FILTER]
    Name          modify
    Match         *
    Add           cluster production-cluster

[OUTPUT]
    Name              loki
    Match             *
    Host              loki.logging.svc
    Port              3100
    labels            job=fluent-bit, cluster=production-cluster
    auto_kubernetes_labels on
    remove_keys       kubernetes,stream
    line_format       json
    Retry_Limit       3

# Fluent Bit Configuration: Kubernetes to OpenTelemetry
# This configuration collects logs from Kubernetes pods, enriches them with metadata,
# and forwards them to an OpenTelemetry Collector using the OTLP protocol over HTTP.

[SERVICE]
    # Flush interval in seconds
    Flush        1

    # Log level: off, error, warn, info, debug, trace
    Log_Level    info

    # Daemon mode (Off for containers)
    Daemon       Off

    # HTTP server for health checks and metrics
    HTTP_Server  On
    HTTP_Listen  0.0.0.0
    HTTP_Port    2020

    # Enable storage metrics
    storage.metrics on
    Parsers_File  parsers.conf

[INPUT]
    Name              tail
    Tag               kube.*
    Path              /var/log/containers/*.log
    Exclude_Path      /var/log/containers/*fluent-bit*.log
    Parser            docker
    DB                /var/log/flb_kube.db
    Mem_Buf_Limit     50MB
    Skip_Long_Lines   On
    Refresh_Interval  10

[FILTER]
    Name                kubernetes
    Match               kube.*
    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           On
    Keep_Log            Off
    K8S-Logging.Parser  On
    K8S-Logging.Exclude On
    Labels              On
    Annotations         On

[FILTER]
    Name          modify
    Match         *
    Add           cluster_name my-cluster
    Add           environment production

[OUTPUT]
    Name                 opentelemetry
    Match                *
    Host                 opentelemetry-collector.observability.svc
    Port                 4318
    # Use HTTP protocol for OTLP
    logs_uri             /v1/logs
    # Add resource attributes
    add_label            cluster my-cluster
    add_label            environment production
    # TLS configuration
    tls                  On
    tls.verify           On
    # Retry configuration
    Retry_Limit          3

# Fluent Bit Configuration: Lua Scripting Filter
# This configuration demonstrates using Lua scripts for advanced log filtering,
# transformation, and enrichment. The Lua filter allows custom processing logic
# that goes beyond built-in filters.

[SERVICE]
    # Flush interval in seconds
    Flush        1

    # Log level: off, error, warn, info, debug, trace
    Log_Level    info

    # Daemon mode (Off for containers)
    Daemon       Off

    # HTTP server for health checks and metrics
    HTTP_Server  On
    HTTP_Listen  0.0.0.0
    HTTP_Port    2020

    # Enable storage metrics
    storage.metrics on

    # Parser configuration file
    Parsers_File parsers.conf

[INPUT]
    Name              tail
    Tag               app.*
    Path              /var/log/app/*.log
    Parser            json
    DB                /var/log/flb_app.db
    Mem_Buf_Limit     100MB
    Skip_Long_Lines   On

[FILTER]
    Name          parser
    Match         *
    Key_Name      log
    Parser        json
    Reserve_Data  On

[FILTER]
    Name    lua
    Match   *
    script  /fluent-bit/scripts/filter.lua
    call    process_record

[FILTER]
    Name          modify
    Match         *
    Add           processed_by lua_filter

[OUTPUT]
    Name              es
    Match             *
    Host              elasticsearch
    Port              9200
    Index             app-logs
    Retry_Limit       3
    storage.total_limit_size 10M

# Example Lua script content (save to /fluent-bit/scripts/filter.lua):
# function process_record(tag, timestamp, record)
#     -- Add custom field
#     record["custom_field"] = "custom_value"
#
#     -- Transform existing field
#     if record["level"] then
#         record["severity"] = string.upper(record["level"])
#     end
#
#     -- Filter out specific records (return -1 to drop)
#     if record["message"] and string.match(record["message"], "DEBUG") then
#         return -1, timestamp, record
#     end
#
#     -- Return modified record
#     return 1, timestamp, record
# end

# Fluent Bit Configuration: Multi-Destination
# This configuration sends logs to multiple destinations: Elasticsearch for search,
# S3 for archival, and stdout for debugging.

[SERVICE]
    Flush         1
    Log_Level     info
    Daemon        Off
    HTTP_Server   On
    HTTP_Port     2020
    Parsers_File  parsers.conf

[INPUT]
    Name              tail
    Tag               app.logs
    Path              /var/log/app/*.log
    Parser            json
    DB                /var/log/flb_app.db
    Mem_Buf_Limit     100MB
    Skip_Long_Lines   On

[FILTER]
    Name          modify
    Match         *
    Add           environment production
    Add           region us-east-1

[FILTER]
    Name          parser
    Match         *
    Key_Name      message
    Parser        json
    Reserve_Data  On

# Primary destination: Elasticsearch for real-time search
[OUTPUT]
    Name              es
    Match             *
    Host              elasticsearch.logging.svc
    Port              9200
    Index             app-logs
    Retry_Limit       3
    storage.total_limit_size 5M

# Secondary destination: S3 for long-term archival
[OUTPUT]
    Name              s3
    Match             *
    bucket            logs-archive-prod
    region            us-east-1
    total_file_size   100M
    compression       gzip
    s3_key_format     /logs/app/%Y/%m/%d/%H-%M-%S-$UUID.gz
    Retry_Limit       3

# Debug destination: stdout
[OUTPUT]
    Name          stdout
    Match         *
    Format        json_lines

# Fluent Bit Parser Definitions
# This file contains common parsers for various log formats.

[PARSER]
    Name        docker
    Format      json
    Time_Key    time
    Time_Format %Y-%m-%dT%H:%M:%S.%LZ
    Time_Keep   Off
    # Command      |  Decoder | Field | Optional Action
    # =============|==================|=================
    Decode_Field_As escaped_utf8 log

[PARSER]
    Name        json
    Format      json
    Time_Key    timestamp
    Time_Format %Y-%m-%dT%H:%M:%S.%LZ
    Time_Keep   Off

[PARSER]
    Name        syslog-rfc3164
    Format      regex
    Regex       /^\<(?<pri>[0-9]+)\>(?<time>[^ ]* {1,2}[^ ]* [^ ]*) (?<host>[^ ]*) (?<ident>[a-zA-Z0-9_\/\.\-]*)(?:\[(?<pid>[0-9]+)\])?(?:[^\:]*\:)? *(?<message>.*)$/
    Time_Key    time
    Time_Format %b %d %H:%M:%S
    Time_Keep   Off

[PARSER]
    Name        syslog-rfc5424
    Format      regex
    Regex       /^\<(?<pri>[0-9]{1,5})\>1 (?<time>[^ ]+) (?<host>[^ ]+) (?<ident>[^ ]+) (?<pid>[-0-9]+) (?<msgid>[^ ]+) (?<extradata>(\[(.*)\]|-)) (?<message>.+)$/
    Time_Key    time
    Time_Format %Y-%m-%dT%H:%M:%S.%L%z
    Time_Keep   Off

[PARSER]
    Name        nginx
    Format      regex
    Regex       ^(?<remote>[^ ]*) (?<host>[^ ]*) (?<user>[^ ]*) \[(?<time>[^\]]*)\] "(?<method>\S+)(?: +(?<path>[^\"]*?)(?: +\S*)?)?" (?<code>[^ ]*) (?<size>[^ ]*)(?: "(?<referer>[^\"]*)" "(?<agent>[^\"]*)")?$
    Time_Key    time
    Time_Format %d/%b/%Y:%H:%M:%S %z

[PARSER]
    Name        apache
    Format      regex
    Regex       ^(?<host>[^ ]*) [^ ]* (?<user>[^ ]*) \[(?<time>[^\]]*)\] "(?<method>\S+)(?: +(?<path>[^ ]*) +\S*)?" (?<code>[^ ]*) (?<size>[^ ]*)(?: "(?<referer>[^\"]*)" "(?<agent>[^\"]*)")?$
    Time_Key    time
    Time_Format %d/%b/%Y:%H:%M:%S %z

[PARSER]
    Name        apache_error
    Format      regex
    Regex       ^\[[^ ]* (?<time>[^\]]*)\] \[(?<level>[^\]]*)\](?: \[pid (?<pid>[^\]]*)\])?( \[client (?<client>[^\]]*)\])? (?<message>.*)$

[PARSER]
    Name        mongodb
    Format      regex
    Regex       ^(?<time>[^ ]*)\s+(?<severity>\w)\s+(?<component>[^ ]+)\s+\[(?<context>[^\]]+)]\s+(?<message>.*?) *(?<ms>(\d+))?(:?ms)?$
    Time_Format %Y-%m-%dT%H:%M:%S.%L
    Time_Keep   Off
    Time_Key    time

[PARSER]
    Name        cri
    Format      regex
    Regex       ^(?<time>[^ ]+) (?<stream>stdout|stderr) (?<logtag>[^ ]*) (?<message>.*)$
    Time_Key    time
    Time_Format %Y-%m-%dT%H:%M:%S.%L%z

[MULTILINE_PARSER]
    Name          multiline-java
    Type          regex
    Flush_timeout 1000
    # Java stack traces start with timestamp, Exception, or indented "at"
    rule          "start_state"   "/(^\d{4}-\d{2}-\d{2}|^[A-Za-z].*Exception|^    at )/"  "cont"
    rule          "cont"          "/^[\s]+/"                                                "cont"

[MULTILINE_PARSER]
    Name          multiline-python
    Type          regex
    Flush_timeout 1000
    # Python tracebacks
    rule          "start_state"   "/^Traceback \(most recent/"                              "cont"
    rule          "cont"          "/^[\s]+/"                                                 "cont"

[MULTILINE_PARSER]
    Name          multiline-go
    Type          regex
    Flush_timeout 1000
    # Go panic traces
    rule          "start_state"   "/^(panic:|goroutine )/"                                  "cont"
    rule          "cont"          "/^[\s]+/"                                                 "cont"

[MULTILINE_PARSER]
    Name          multiline-ruby
    Type          regex
    Flush_timeout 1000
    # Ruby exceptions
    rule          "start_state"   "/^[A-Z][a-z]*Error:/"                                    "cont"
    rule          "cont"          "/^\s+from /"                                              "cont"

# Fluent Bit Configuration: Prometheus Metrics Collection
# This configuration collects metrics from node_exporter and Fluent Bit's own HTTP server,
# and forwards them to Prometheus using the remote_write API.

[SERVICE]
    # Flush interval in seconds
    Flush        15

    # Log level: off, error, warn, info, debug, trace
    Log_Level    info

    # Daemon mode (Off for containers)
    Daemon       Off

    # HTTP server for health checks and metrics
    HTTP_Server  On
    HTTP_Listen  0.0.0.0
    HTTP_Port    2020

    # Enable storage metrics
    storage.metrics on

[INPUT]
    Name              node_exporter_metrics
    Tag               node_metrics
    Scrape_interval   15

[INPUT]
    Name              prometheus_scrape
    Tag               k8s_metrics
    Host              127.0.0.1
    Port              2020
    Scrape_interval   15

[FILTER]
    Name          modify
    Match         *
    Add           cluster my-cluster

[OUTPUT]
    Name              prometheus_remote_write
    Match             *
    Host              prometheus.monitoring.svc
    Port              9090
    Uri               /api/v1/write
    # Add labels to all metrics
    add_label         cluster my-cluster
    # TLS configuration
    tls               On
    tls.verify        On
    # Retry configuration
    Retry_Limit       3
    # Compression
    compression       snappy

# Fluent Bit Configuration: Stream Processor
# This configuration demonstrates using the Stream Processor for SQL-like
# transformations, aggregations, and real-time analytics on log data.
# Stream tasks can create derived streams for alerts, metrics, and analytics.

[SERVICE]
    # Flush interval in seconds
    Flush        1

    # Log level: off, error, warn, info, debug, trace
    Log_Level    info

    # Daemon mode (Off for containers)
    Daemon       Off

    # HTTP server for health checks and metrics
    HTTP_Server  On
    HTTP_Listen  0.0.0.0
    HTTP_Port    2020

    # Enable storage metrics
    storage.metrics on

    # Parser configuration file
    Parsers_File parsers.conf

[INPUT]
    Name              tail
    Tag               app.*
    Path              /var/log/app/*.log
    Parser            json
    DB                /var/log/flb_app.db
    Mem_Buf_Limit     100MB
    Skip_Long_Lines   On

# Stream Processor for advanced SQL-like transformations
[STREAM_TASK]
    Name    error_aggregation
    Exec    CREATE STREAM errors AS \
            SELECT \
                level, \
                COUNT(*) as error_count, \
                TUMBLE_START() as window_start \
            FROM TAG:'app.*' \
            WHERE level = 'error' \
            GROUP BY level, TUMBLE(time, INTERVAL '1' MINUTE);

[STREAM_TASK]
    Name    high_latency_detection
    Exec    CREATE STREAM high_latency AS \
            SELECT \
                service, \
                endpoint, \
                response_time_ms \
            FROM TAG:'app.*' \
            WHERE response_time_ms > 1000;

[FILTER]
    Name          modify
    Match         *
    Add           processed_by stream_processor

[OUTPUT]
    Name              es
    Match             app.*
    Host              elasticsearch
    Port              9200
    Index             app-logs
    Retry_Limit       3

[OUTPUT]
    Name              es
    Match             errors
    Host              elasticsearch
    Port              9200
    Index             error-metrics
    Retry_Limit       3

[OUTPUT]
    Name              es
    Match             high_latency
    Host              elasticsearch
    Port              9200
    Index             performance-alerts
    Retry_Limit       3

# Fluent Bit Configuration: Syslog Collection and Forwarding
# This configuration receives syslog messages and forwards them to a remote syslog server.

[SERVICE]
    Flush         5
    Log_Level     info
    Daemon        Off
    HTTP_Server   On
    HTTP_Port     2020
    Parsers_File  parsers.conf

[INPUT]
    Name          syslog
    Tag           syslog.messages
    Parser        syslog-rfc3164
    Listen        0.0.0.0
    Port          5140
    Mode          tcp
    Buffer_Size   32KB

[FILTER]
    Name          modify
    Match         *
    Add           source fluent-bit-collector
    Add           collector_host ${HOSTNAME}

[OUTPUT]
    Name          syslog
    Match         *
    Host          syslog-server.example.com
    Port          514
    Mode          tcp
    Syslog_Format rfc5424
    Retry_Limit   5

# Optional: Also output to stdout for debugging
[OUTPUT]
    Name          stdout
    Match         *
    Format        json_lines

# Fluent Bit Configuration: Kubernetes Error Logs to Elasticsearch
# This configuration collects logs from Kubernetes pods, filters for error-level
# logs only (ERROR, FATAL, CRITICAL), applies throttling, and forwards to Elasticsearch.
#
# Generated for:
# - Cluster: prod-cluster
# - Environment: production
# - Output: Elasticsearch at elasticsearch.logging.svc:9200
# - Index prefix: k8s-errors

# ===== SERVICE SECTION =====
[SERVICE]
    # Flush interval in seconds
    Flush        1

    # Log level: off, error, warn, info, debug, trace
    Log_Level    info

    # Daemon mode (Off for containers)
    Daemon       Off

    # HTTP server for health checks and metrics
    HTTP_Server  On
    HTTP_Listen  0.0.0.0
    HTTP_Port    2020

    # Enable storage metrics
    storage.metrics on

    # Parser configuration file
    Parsers_File parsers.conf

# ===== INPUT SECTION =====
[INPUT]
    Name              tail
    Tag               kube.*
    Path              /var/log/containers/*.log
    # Prevent log loops by excluding Fluent Bit's own logs
    Exclude_Path      /var/log/containers/*fluent-bit*.log
    Parser            docker
    # Position database for crash recovery
    DB                /var/log/flb_kube.db
    # Memory buffer limit per input (prevents OOM)
    Mem_Buf_Limit     50MB
    # Skip lines longer than 32KB (prevents hang)
    Skip_Long_Lines   On
    # Refresh file list every 10 seconds
    Refresh_Interval  10
    # Don't read from beginning (only new logs)
    Read_from_Head    Off

# ===== FILTER SECTION =====

# Kubernetes metadata enrichment
[FILTER]
    Name                kubernetes
    Match               kube.*
    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 parsed JSON logs
    Merge_Log           On
    # Keep original log field for downstream parser/grep filters
    Keep_Log            On
    # Honor pod annotations for parser
    K8S-Logging.Parser  On
    # Honor pod annotations for exclude
    K8S-Logging.Exclude On
    # Include pod labels
    Labels              On
    # Exclude annotations to reduce size
    Annotations         Off
    Buffer_Size         0

# Add cluster and environment identifiers
[FILTER]
    Name          modify
    Match         *
    Add           cluster_name prod-cluster
    Add           environment production

# Parse JSON logs to extract structured fields (including level)
[FILTER]
    Name          parser
    Match         *
    Key_Name      log
    Parser        json
    Reserve_Data  On
    Preserve_Key  Off

# Filter to include only error-level logs (ERROR, FATAL, CRITICAL)
# This significantly reduces log volume and storage costs
[FILTER]
    Name          grep
    Match         *
    Regex         level (ERROR|FATAL|CRITICAL|error|fatal|critical)

# Rate limiting: max 500 logs per 5 second window
# Prevents overwhelming downstream systems during log storms
[FILTER]
    Name          throttle
    Match         *
    Rate          500
    Window        5
    Interval      1m
    Print_Status  true

# Lift nested Kubernetes fields to top level with prefix
[FILTER]
    Name          nest
    Match         *
    Operation     lift
    Nested_under  kubernetes
    Add_prefix    k8s_

# ===== OUTPUT SECTION =====

# Elasticsearch output for error logs
[OUTPUT]
    Name              es
    Match             *
    Host              elasticsearch.logging.svc
    Port              9200
    # Use Logstash format for time-based indices
    Logstash_Format   On
    Logstash_Prefix   k8s-errors
    # Retry configuration
    Retry_Limit       3
    # Buffer configuration
    storage.total_limit_size 5M
    # TLS configuration
    tls               On
    tls.verify        Off  # Internal cluster with self-signed certs
    # Buffer settings
    Buffer_Size       False
    Type              _doc

# Fluent Bit Parser Definitions
# This file contains common parsers for various log formats.

[PARSER]
    Name        docker
    Format      json
    Time_Key    time
    Time_Format %Y-%m-%dT%H:%M:%S.%LZ
    Time_Keep   Off
    # Command      |  Decoder | Field | Optional Action
    # =============|==================|=================
    Decode_Field_As escaped_utf8 log

[PARSER]
    Name        json
    Format      json
    Time_Key    timestamp
    Time_Format %Y-%m-%dT%H:%M:%S.%LZ
    Time_Keep   Off

[PARSER]
    Name        syslog-rfc3164
    Format      regex
    Regex       /^\<(?<pri>[0-9]+)\>(?<time>[^ ]* {1,2}[^ ]* [^ ]*) (?<host>[^ ]*) (?<ident>[a-zA-Z0-9_\/\.\-]*)(?:\[(?<pid>[0-9]+)\])?(?:[^\:]*\:)? *(?<message>.*)$/
    Time_Key    time
    Time_Format %b %d %H:%M:%S
    Time_Keep   Off

[PARSER]
    Name        syslog-rfc5424
    Format      regex
    Regex       /^\<(?<pri>[0-9]{1,5})\>1 (?<time>[^ ]+) (?<host>[^ ]+) (?<ident>[^ ]+) (?<pid>[-0-9]+) (?<msgid>[^ ]+) (?<extradata>(\[(.*)\]|-)) (?<message>.+)$/
    Time_Key    time
    Time_Format %Y-%m-%dT%H:%M:%S.%L%z
    Time_Keep   Off

[PARSER]
    Name        nginx
    Format      regex
    Regex       ^(?<remote>[^ ]*) (?<host>[^ ]*) (?<user>[^ ]*) \[(?<time>[^\]]*)\] "(?<method>\S+)(?: +(?<path>[^\"]*?)(?: +\S*)?)?" (?<code>[^ ]*) (?<size>[^ ]*)(?: "(?<referer>[^\"]*)" "(?<agent>[^\"]*)")?$
    Time_Key    time
    Time_Format %d/%b/%Y:%H:%M:%S %z

[PARSER]
    Name        apache
    Format      regex
    Regex       ^(?<host>[^ ]*) [^ ]* (?<user>[^ ]*) \[(?<time>[^\]]*)\] "(?<method>\S+)(?: +(?<path>[^ ]*) +\S*)?" (?<code>[^ ]*) (?<size>[^ ]*)(?: "(?<referer>[^\"]*)" "(?<agent>[^\"]*)")?$
    Time_Key    time
    Time_Format %d/%b/%Y:%H:%M:%S %z

[PARSER]
    Name        apache_error
    Format      regex
    Regex       ^\[[^ ]* (?<time>[^\]]*)\] \[(?<level>[^\]]*)\](?: \[pid (?<pid>[^\]]*)\])?( \[client (?<client>[^\]]*)\])? (?<message>.*)$

[PARSER]
    Name        mongodb
    Format      regex
    Regex       ^(?<time>[^ ]*)\s+(?<severity>\w)\s+(?<component>[^ ]+)\s+\[(?<context>[^\]]+)]\s+(?<message>.*?) *(?<ms>(\d+))?(:?ms)?$
    Time_Format %Y-%m-%dT%H:%M:%S.%L
    Time_Keep   Off
    Time_Key    time

[PARSER]
    Name        cri
    Format      regex
    Regex       ^(?<time>[^ ]+) (?<stream>stdout|stderr) (?<logtag>[^ ]*) (?<message>.*)$
    Time_Key    time
    Time_Format %Y-%m-%dT%H:%M:%S.%L%z

[MULTILINE_PARSER]
    Name          multiline-java
    Type          regex
    Flush_timeout 1000
    # Java stack traces start with timestamp, Exception, or indented "at"
    rule          "start_state"   "/(^\d{4}-\d{2}-\d{2}|^[A-Za-z].*Exception|^    at )/"  "cont"
    rule          "cont"          "/^[\s]+/"                                                "cont"

[MULTILINE_PARSER]
    Name          multiline-python
    Type          regex
    Flush_timeout 1000
    # Python tracebacks
    rule          "start_state"   "/^Traceback \(most recent/"                              "cont"
    rule          "cont"          "/^[\s]+/"                                                 "cont"

[MULTILINE_PARSER]
    Name          multiline-go
    Type          regex
    Flush_timeout 1000
    # Go panic traces
    rule          "start_state"   "/^(panic:|goroutine )/"                                  "cont"
    rule          "cont"          "/^[\s]+/"                                                 "cont"

[MULTILINE_PARSER]
    Name          multiline-ruby
    Type          regex
    Flush_timeout 1000
    # Ruby exceptions
    rule          "start_state"   "/^[A-Z][a-z]*Error:/"                                    "cont"
    rule          "cont"          "/^\s+from /"                                              "cont"

#!/usr/bin/env python3
"""
Fluent Bit Configuration Generator

Generates production-ready Fluent Bit configurations based on common use cases.
Supports multiple input sources, filters, and output destinations.
"""

import argparse
import inspect
import sys
import warnings
from typing import Dict, Optional, Callable, Any
from urllib.parse import urlparse


class FluentBitConfigGenerator:
    """Generates Fluent Bit configuration files with best practices built-in."""

    _DEPRECATED_KWARG_ALIASES: Dict[str, Dict[str, str]] = {
        "syslog-forward": {
            "forward_host": "syslog_host",
            "forward_port": "syslog_port",
        },
        "file-tail-s3": {
            "file_path": "log_path",
        },
    }
    _DEPRECATION_REMOVAL_DATE = "2026-09-01"

    def __init__(self) -> None:
        """Initialize the generator with available use cases."""
        self.use_cases: Dict[str, Callable[..., str]] = {
            "kubernetes-elasticsearch": self._generate_k8s_elasticsearch,
            "kubernetes-loki": self._generate_k8s_loki,
            "kubernetes-cloudwatch": self._generate_k8s_cloudwatch,
            "kubernetes-opentelemetry": self._generate_k8s_opentelemetry,
            "application-multiline": self._generate_app_multiline,
            "syslog-forward": self._generate_syslog_forward,
            "file-tail-s3": self._generate_file_s3,
            "http-kafka": self._generate_http_kafka,
            "multi-destination": self._generate_multi_destination,
            "prometheus-metrics": self._generate_prometheus_metrics,
            "lua-filtering": self._generate_lua_filtering,
            "stream-processor": self._generate_stream_processor,
            "custom": self._generate_custom,
        }

    def generate(self, use_case: str, **kwargs) -> str:
        """
        Generate configuration for specified use case.

        Args:
            use_case: The name of the use case to generate
            **kwargs: Additional parameters specific to the use case

        Returns:
            Generated Fluent Bit configuration as a string

        Raises:
            ValueError: If use case is not recognized or kwargs are invalid
        """
        if use_case not in self.use_cases:
            available = ", ".join(self.use_cases.keys())
            raise ValueError(
                f"Unknown use case: {use_case}\n"
                f"Available use cases: {available}"
            )
        normalized_kwargs = self._normalize_kwargs(use_case, kwargs)
        return self.use_cases[use_case](**normalized_kwargs)

    def _allowed_kwargs_for(self, use_case: str) -> set[str]:
        """Return accepted keyword names for a use case method signature."""
        signature = inspect.signature(self.use_cases[use_case])
        return {
            name
            for name, parameter in signature.parameters.items()
            if parameter.kind
            in (inspect.Parameter.POSITIONAL_OR_KEYWORD, inspect.Parameter.KEYWORD_ONLY)
        }

    def _normalize_kwargs(self, use_case: str, kwargs: Dict[str, Any]) -> Dict[str, Any]:
        """Map deprecated kwargs and reject unknown kwargs."""
        normalized = dict(kwargs)

        for old_name, new_name in self._DEPRECATED_KWARG_ALIASES.get(use_case, {}).items():
            if old_name not in normalized:
                continue

            old_value = normalized.pop(old_name)
            if new_name not in normalized or normalized[new_name] is None:
                normalized[new_name] = old_value

            warnings.warn(
                (
                    f"'{old_name}' is deprecated for use case '{use_case}'; "
                    f"use '{new_name}' instead. Support will be removed after "
                    f"{self._DEPRECATION_REMOVAL_DATE}."
                ),
                DeprecationWarning,
                stacklevel=3,
            )

        allowed_kwargs = self._allowed_kwargs_for(use_case)
        unknown_kwargs = sorted(set(normalized) - allowed_kwargs)
        if unknown_kwargs:
            unknown_text = ", ".join(unknown_kwargs)
            raise ValueError(
                f"Unknown parameter(s) for use case '{use_case}': {unknown_text}"
            )

        return normalized

    @staticmethod
    def _parse_otlp_endpoint(endpoint: str) -> tuple[str, int, str]:
        """
        Parse OTLP endpoint into host/port/base path.

        Supports host:port, URL formats, and IPv6 bracket notation.
        """
        normalized = endpoint.strip()
        parsed = urlparse(normalized if "://" in normalized else f"//{normalized}")

        host = parsed.hostname
        if not host:
            raise ValueError(
                f"Invalid OTLP endpoint: {endpoint}. Expected host:port or URL."
            )

        port = parsed.port if parsed.port is not None else 4318
        base_path = parsed.path.rstrip("/")
        logs_uri = f"{base_path}/v1/logs" if base_path else "/v1/logs"
        return host, port, logs_uri

    def _generate_service_section(
        self,
        flush: int = 1,
        log_level: str = "info",
        http_server: bool = True,
        parsers_file: Optional[str] = None,
    ) -> str:
        """
        Generate SERVICE section with global Fluent Bit configuration.

        Args:
            flush: Flush interval in seconds (default: 1)
            log_level: Logging level (default: "info")
            http_server: Enable HTTP server for metrics (default: True)
            parsers_file: Path to parsers configuration file (optional)

        Returns:
            SERVICE section configuration string
        """
        config = f"""[SERVICE]
    # Flush interval in seconds
    Flush        {flush}

    # Log level: off, error, warn, info, debug, trace
    Log_Level    {log_level}

    # Daemon mode (Off for containers)
    Daemon       Off
"""

        if http_server:
            config += """
    # HTTP server for health checks and metrics
    HTTP_Server  On
    HTTP_Listen  0.0.0.0
    HTTP_Port    2020

    # Enable storage metrics
    storage.metrics on
"""

        if parsers_file:
            config += f"""
    # Parser configuration file
    Parsers_File {parsers_file}
"""

        return config

    def _tls_block(
        self,
        tls_verify: bool = True,
        tls_ca_file: Optional[str] = None,
        indent: int = 4,
    ) -> str:
        """Generate TLS configuration lines for an OUTPUT block.

        Args:
            tls_verify: Verify server TLS certificates (default: True).
                        Set to False only for self-signed certs when a
                        custom CA file cannot be provided.
            tls_ca_file: Path to a custom CA certificate file.  When
                         provided, verification is always enabled using
                         this path as the trust anchor — the recommended
                         approach for self-signed certificates.
            indent: Leading spaces for each generated line.

        Returns:
            TLS configuration lines as a string (no trailing newline).
        """
        pad = " " * indent
        lines = [f"{pad}tls               On"]
        if tls_ca_file:
            lines.append(f"{pad}tls.verify        On")
            lines.append(f"{pad}tls.ca_file       {tls_ca_file}")
        elif tls_verify:
            lines.append(f"{pad}tls.verify        On")
        else:
            lines.append(f"{pad}tls.verify        Off")
        return "\n".join(lines)

    def _generate_k8s_elasticsearch(
        self,
        es_host: str = "elasticsearch.logging.svc",
        es_port: int = 9200,
        es_index_prefix: str = "k8s",
        cluster_name: str = "my-cluster",
        environment: str = "production",
        container_runtime: str = "cri",
        tls_verify: bool = True,
        tls_ca_file: Optional[str] = None,
        **kwargs
    ) -> str:
        """
        Generate Kubernetes to Elasticsearch configuration.

        Args:
            es_host: Elasticsearch hostname
            es_port: Elasticsearch port
            es_index_prefix: Index prefix for Logstash format
            cluster_name: Kubernetes cluster name
            environment: Environment identifier
            container_runtime: Container runtime parser to use ("cri" for containerd/CRI-O,
                               "docker" for Docker). Default: "cri" (Kubernetes 1.24+).
            tls_verify: Verify TLS certificates (default: True).
                        Set to False only when self-signed certs are in use
                        and no CA file can be provided.
            tls_ca_file: Path to custom CA file for self-signed certificates.
                         When set, tls_verify is implicitly enabled.

        Returns:
            Complete Fluent Bit configuration string
        """
        if es_host is None:
            es_host = "elasticsearch.logging.svc"
        if es_port is None:
            es_port = 9200
        if es_index_prefix is None:
            es_index_prefix = "k8s"
        if cluster_name is None:
            cluster_name = "my-cluster"
        if environment is None:
            environment = "production"
        if container_runtime is None:
            container_runtime = "cri"
        config = self._generate_service_section(parsers_file="parsers.conf")

        config += f"""
[INPUT]
    Name              tail
    Tag               kube.*
    Path              /var/log/containers/*.log
    Exclude_Path      /var/log/containers/*fluent-bit*.log
    Parser            {container_runtime}
    DB                /var/log/flb_kube.db
    Mem_Buf_Limit     50MB
    Skip_Long_Lines   On
    Refresh_Interval  10
    Read_from_Head    Off

[FILTER]
    Name                kubernetes
    Match               kube.*
    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           On
    Keep_Log            Off
    K8S-Logging.Parser  On
    K8S-Logging.Exclude On
    Labels              On
    Annotations         Off
    Buffer_Size         0

[FILTER]
    Name          modify
    Match         *
    Add           cluster_name {cluster_name}
    Add           environment {environment}

[FILTER]
    Name          nest
    Match         *
    Operation     lift
    Nested_under  kubernetes
    Add_prefix    k8s_

[OUTPUT]
    Name              es
    Match             *
    Host              {es_host}
    Port              {es_port}
    Logstash_Format   On
    Logstash_Prefix   {es_index_prefix}
    Retry_Limit       3
    storage.total_limit_size 5M
    Buffer_Size       False
    Type              _doc
{self._tls_block(tls_verify, tls_ca_file)}
"""

        return config

    def _generate_k8s_loki(
        self,
        loki_host: str = "loki.logging.svc",
        loki_port: int = 3100,
        cluster_name: str = "my-cluster",
        container_runtime: str = "cri",
        **kwargs
    ) -> str:
        """
        Generate Kubernetes to Loki configuration.

        Args:
            loki_host: Loki hostname
            loki_port: Loki port
            cluster_name: Kubernetes cluster name
            container_runtime: Container runtime parser to use ("cri" for containerd/CRI-O,
                               "docker" for Docker). Default: "cri" (Kubernetes 1.24+).

        Returns:
            Complete Fluent Bit configuration string
        """
        if loki_host is None:
            loki_host = "loki.logging.svc"
        if loki_port is None:
            loki_port = 3100
        if cluster_name is None:
            cluster_name = "my-cluster"
        if container_runtime is None:
            container_runtime = "cri"
        config = self._generate_service_section(parsers_file="parsers.conf")

        config += f"""
[INPUT]
    Name              tail
    Tag               kube.*
    Path              /var/log/containers/*.log
    Exclude_Path      /var/log/containers/*fluent-bit*.log
    Parser            {container_runtime}
    DB                /var/log/flb_kube.db
    Mem_Buf_Limit     50MB
    Skip_Long_Lines   On
    Refresh_Interval  10

[FILTER]
    Name                kubernetes
    Match               kube.*
    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           On
    Keep_Log            Off
    K8S-Logging.Parser  On
    K8S-Logging.Exclude On
    Labels              On

[FILTER]
    Name          modify
    Match         *
    Add           cluster {cluster_name}

[OUTPUT]
    Name              loki
    Match             *
    Host              {loki_host}
    Port              {loki_port}
    labels            job=fluent-bit, cluster={cluster_name}
    auto_kubernetes_labels on
    remove_keys       kubernetes,stream
    line_format       json
    Retry_Limit       3
"""

        return config

    def _generate_k8s_cloudwatch(
        self,
        aws_region: str = "us-east-1",
        log_group_name: str = "/aws/kubernetes/logs",
        cluster_name: str = "my-cluster",
        container_runtime: str = "cri",
        **kwargs
    ) -> str:
        """
        Generate Kubernetes to CloudWatch configuration.

        Args:
            aws_region: AWS region for CloudWatch
            log_group_name: CloudWatch log group name
            cluster_name: Kubernetes cluster name
            container_runtime: Container runtime parser to use ("cri" for containerd/CRI-O,
                               "docker" for Docker). Default: "cri" (Kubernetes 1.24+).

        Returns:
            Complete Fluent Bit configuration string
        """
        if aws_region is None:
            aws_region = "us-east-1"
        if log_group_name is None:
            log_group_name = "/aws/kubernetes/logs"
        if cluster_name is None:
            cluster_name = "my-cluster"
        if container_runtime is None:
            container_runtime = "cri"
        config = self._generate_service_section(parsers_file="parsers.conf")

        config += f"""
[INPUT]
    Name              tail
    Tag               kube.*
    Path              /var/log/containers/*.log
    Exclude_Path      /var/log/containers/*fluent-bit*.log
    Parser            {container_runtime}
    DB                /var/log/flb_kube.db
    Mem_Buf_Limit     50MB
    Skip_Long_Lines   On

[FILTER]
    Name                kubernetes
    Match               kube.*
    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           On
    Keep_Log            Off
    Labels              On

[FILTER]
    Name          modify
    Match         *
    Add           cluster {cluster_name}

[OUTPUT]
    Name              cloudwatch_logs
    Match             *
    region            {aws_region}
    log_group_name    {log_group_name}
    log_stream_prefix from-fluent-bit-
    auto_create_group On
    Retry_Limit       3
"""

        return config

    def _generate_k8s_opentelemetry(
        self,
        otlp_endpoint: str = "opentelemetry-collector.observability.svc:4318",
        cluster_name: str = "my-cluster",
        environment: str = "production",
        container_runtime: str = "cri",
        tls_verify: bool = True,
        tls_ca_file: Optional[str] = None,
        **kwargs
    ) -> str:
        """
        Generate Kubernetes to OpenTelemetry configuration.

        Args:
            otlp_endpoint: OpenTelemetry Collector endpoint (HTTP)
            cluster_name: Kubernetes cluster name
            environment: Environment identifier
            container_runtime: Container runtime parser to use ("cri" for containerd/CRI-O,
                               "docker" for Docker). Default: "cri" (Kubernetes 1.24+).
            tls_verify: Verify TLS certificates (default: True).
                        Set to False only when self-signed certs are in use
                        and no CA file can be provided.
            tls_ca_file: Path to custom CA file for self-signed certificates.
                         When set, tls_verify is implicitly enabled.

        Returns:
            Complete Fluent Bit configuration string
        """
        if otlp_endpoint is None:
            otlp_endpoint = "opentelemetry-collector.observability.svc:4318"
        if cluster_name is None:
            cluster_name = "my-cluster"
        if environment is None:
            environment = "production"
        if container_runtime is None:
            container_runtime = "cri"
        host, port, logs_uri = self._parse_otlp_endpoint(otlp_endpoint)
        config = self._generate_service_section(parsers_file="parsers.conf")

        config += f"""
[INPUT]
    Name              tail
    Tag               kube.*
    Path              /var/log/containers/*.log
    Exclude_Path      /var/log/containers/*fluent-bit*.log
    Parser            {container_runtime}
    DB                /var/log/flb_kube.db
    Mem_Buf_Limit     50MB
    Skip_Long_Lines   On
    Refresh_Interval  10

[FILTER]
    Name                kubernetes
    Match               kube.*
    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           On
    Keep_Log            Off
    K8S-Logging.Parser  On
    K8S-Logging.Exclude On
    Labels              On
    Annotations         On

[FILTER]
    Name          modify
    Match         *
    Add           cluster_name {cluster_name}
    Add           environment {environment}

[OUTPUT]
    Name                 opentelemetry
    Match                *
    Host                 {host}
    Port                 {port}
    # Use HTTP protocol for OTLP
    logs_uri             {logs_uri}
    # Add resource attributes
    add_label            cluster {cluster_name}
    add_label            environment {environment}
    # TLS configuration
{self._tls_block(tls_verify, tls_ca_file, indent=4)}
    # Retry configuration
    Retry_Limit          3
"""

        return config

    def _generate_app_multiline(
        self,
        log_path: str = "/var/log/app/*.log",
        language: str = "java",
        app_name: str = "myapp",
        environment: str = "production",
        es_host: str = "elasticsearch",
        **kwargs
    ) -> str:
        """
        Generate application logs with multiline parsing configuration.

        Args:
            log_path: Path to application log files
            language: Programming language for multiline parser (java, python, go, ruby)
            app_name: Application name
            environment: Environment identifier
            es_host: Elasticsearch hostname

        Returns:
            Complete Fluent Bit configuration string
        """
        if log_path is None:
            log_path = "/var/log/app/*.log"
        if language is None:
            language = "java"
        if app_name is None:
            app_name = "myapp"
        if environment is None:
            environment = "production"
        if es_host is None:
            es_host = "elasticsearch"
        config = self._generate_service_section(parsers_file="parsers.conf")

        config += f"""
[INPUT]
    Name              tail
    Tag               app.*
    Path              {log_path}
    Multiline.Parser  multiline-{language}
    DB                /var/log/flb_app.db
    Mem_Buf_Limit     100MB
    Skip_Long_Lines   On

[FILTER]
    Name          modify
    Match         *
    Add           app_name {app_name}
    Add           environment {environment}

[FILTER]
    Name          parser
    Match         *
    Key_Name      log
    Parser        json
    Reserve_Data  On
    Preserve_Key  Off

[OUTPUT]
    Name              es
    Match             *
    Host              {es_host}
    Port              9200
    Index             app-logs
    Retry_Limit       3
    storage.total_limit_size 10M
"""

        return config

    def _generate_syslog_forward(
        self,
        listen_port: int = 5140,
        syslog_host: str = "syslog-server.example.com",
        syslog_port: int = 514,
        **kwargs
    ) -> str:
        """
        Generate syslog collection and forwarding configuration.

        Args:
            listen_port: Port to listen for syslog messages
            syslog_host: Destination syslog server hostname
            syslog_port: Destination syslog server port

        Returns:
            Complete Fluent Bit configuration string
        """
        if listen_port is None:
            listen_port = 5140
        if syslog_host is None:
            syslog_host = "syslog-server.example.com"
        if syslog_port is None:
            syslog_port = 514
        config = self._generate_service_section(flush=5, parsers_file="parsers.conf")

        config += f"""
[INPUT]
    Name          syslog
    Tag           syslog.*
    Parser        syslog-rfc3164
    Listen        0.0.0.0
    Port          {listen_port}
    Mode          tcp
    Buffer_Size   32KB

[FILTER]
    Name          modify
    Match         *
    Add           source fluent-bit

[OUTPUT]
    Name          syslog
    Match         *
    Host          {syslog_host}
    Port          {syslog_port}
    Mode          tcp
    Syslog_Format rfc5424
    Retry_Limit   5

[OUTPUT]
    Name          stdout
    Match         *
    Format        json_lines
"""

        return config

    def _generate_file_s3(
        self,
        log_path: str = "/var/log/app/*.log",
        s3_bucket: str = "my-logs-bucket",
        s3_region: str = "us-east-1",
        **kwargs
    ) -> str:
        """
        Generate file tailing to S3 configuration.

        Args:
            log_path: Path to log files to tail
            s3_bucket: S3 bucket name
            s3_region: AWS region for S3

        Returns:
            Complete Fluent Bit configuration string
        """
        if log_path is None:
            log_path = "/var/log/app/*.log"
        if s3_bucket is None:
            s3_bucket = "my-logs-bucket"
        if s3_region is None:
            s3_region = "us-east-1"
        config = self._generate_service_section(flush=5)

        config += f"""
[INPUT]
    Name              tail
    Tag               files.*
    Path              {log_path}
    DB                /var/log/flb_files.db
    Mem_Buf_Limit     100MB
    Skip_Long_Lines   On

[FILTER]
    Name          modify
    Match         *
    Add           source file-collector

[OUTPUT]
    Name              s3
    Match             *
    bucket            {s3_bucket}
    region            {s3_region}
    total_file_size   100M
    upload_timeout    10m
    use_put_object    Off
    compression       gzip
    s3_key_format     /fluent-bit-logs/%Y/%m/%d/$TAG[0]/%H-%M-%S-$UUID.gz
    Retry_Limit       3
"""

        return config

    def _generate_http_kafka(
        self,
        http_port: int = 9880,
        kafka_brokers: str = "kafka:9092",
        kafka_topic: str = "logs",
        **kwargs
    ) -> str:
        """
        Generate HTTP webhook to Kafka configuration.

        Args:
            http_port: Port to listen for HTTP requests
            kafka_brokers: Comma-separated Kafka broker addresses
            kafka_topic: Kafka topic name

        Returns:
            Complete Fluent Bit configuration string
        """
        if http_port is None:
            http_port = 9880
        if kafka_brokers is None:
            kafka_brokers = "kafka:9092"
        if kafka_topic is None:
            kafka_topic = "logs"
        config = self._generate_service_section()

        config += f"""
[INPUT]
    Name          http
    Tag           webhook.*
    Listen        0.0.0.0
    Port          {http_port}
    Buffer_Size   32KB

[FILTER]
    Name          modify
    Match         *
    Add           source webhook

[OUTPUT]
    Name              kafka
    Match             *
    Brokers           {kafka_brokers}
    Topics            {kafka_topic}
    Format            json
    Timestamp_Key     @timestamp
    Retry_Limit       3
    rdkafka.queue.buffering.max.messages 100000
    rdkafka.request.required.acks        1
"""

        return config

    def _generate_multi_destination(
        self,
        es_host: str = "elasticsearch",
        s3_bucket: str = "logs-archive",
        **kwargs
    ) -> str:
        """
        Generate multi-destination configuration.

        Args:
            es_host: Elasticsearch hostname
            s3_bucket: S3 bucket name for archival

        Returns:
            Complete Fluent Bit configuration string
        """
        if es_host is None:
            es_host = "elasticsearch"
        if s3_bucket is None:
            s3_bucket = "logs-archive"
        config = self._generate_service_section(parsers_file="parsers.conf")

        config += f"""
[INPUT]
    Name              tail
    Tag               app.*
    Path              /var/log/app/*.log
    Parser            json
    DB                /var/log/flb_app.db
    Mem_Buf_Limit     100MB

[FILTER]
    Name          modify
    Match         *
    Add           environment production

[OUTPUT]
    Name              es
    Match             *
    Host              {es_host}
    Port              9200
    Index             app-logs
    Retry_Limit       3

[OUTPUT]
    Name              s3
    Match             *
    bucket            {s3_bucket}
    region            us-east-1
    total_file_size   100M
    compression       gzip
    s3_key_format     /logs/%Y/%m/%d/%H-%M-%S-$UUID.gz
    Retry_Limit       3

[OUTPUT]
    Name          stdout
    Match         *
    Format        json_lines
"""

        return config

    def _generate_prometheus_metrics(
        self,
        prometheus_host: str = "prometheus.monitoring.svc",
        prometheus_port: int = 9090,
        scrape_interval: int = 15,
        cluster_name: str = "my-cluster",
        tls_verify: bool = True,
        tls_ca_file: Optional[str] = None,
        **kwargs
    ) -> str:
        """
        Generate Prometheus metrics collection and forwarding configuration.

        Args:
            prometheus_host: Prometheus remote write endpoint hostname
            prometheus_port: Prometheus remote write endpoint port
            scrape_interval: Metrics scrape interval in seconds
            cluster_name: Kubernetes cluster name
            tls_verify: Verify TLS certificates (default: True).
                        Set to False only when self-signed certs are in use
                        and no CA file can be provided.
            tls_ca_file: Path to custom CA file for self-signed certificates.
                         When set, tls_verify is implicitly enabled.

        Returns:
            Complete Fluent Bit configuration string
        """
        if prometheus_host is None:
            prometheus_host = "prometheus.monitoring.svc"
        if prometheus_port is None:
            prometheus_port = 9090
        if scrape_interval is None:
            scrape_interval = 15
        if cluster_name is None:
            cluster_name = "my-cluster"
        config = self._generate_service_section(flush=scrape_interval)

        config += f"""
[INPUT]
    Name              node_exporter_metrics
    Tag               node_metrics
    Scrape_interval   {scrape_interval}

[INPUT]
    Name              prometheus_scrape
    Tag               k8s_metrics
    Host              127.0.0.1
    Port              2020
    Scrape_interval   {scrape_interval}

[OUTPUT]
    Name              prometheus_remote_write
    Match             *
    Host              {prometheus_host}
    Port              {prometheus_port}
    Uri               /api/v1/write
    # Add labels to all metrics
    add_label         cluster {cluster_name}
    # TLS configuration
{self._tls_block(tls_verify, tls_ca_file)}
    # Retry configuration
    Retry_Limit       3
    # Compression
    compression       snappy
"""

        return config

    def _generate_lua_filtering(
        self,
        log_path: str = "/var/log/app/*.log",
        lua_script_path: str = "/fluent-bit/scripts/filter.lua",
        es_host: str = "elasticsearch",
        **kwargs
    ) -> str:
        """
        Generate configuration with Lua scripting filter.

        Args:
            log_path: Path to log files
            lua_script_path: Path to Lua filter script
            es_host: Elasticsearch hostname

        Returns:
            Complete Fluent Bit configuration string
        """
        if log_path is None:
            log_path = "/var/log/app/*.log"
        if lua_script_path is None:
            lua_script_path = "/fluent-bit/scripts/filter.lua"
        if es_host is None:
            es_host = "elasticsearch"
        config = self._generate_service_section(parsers_file="parsers.conf")

        config += f"""
[INPUT]
    Name              tail
    Tag               app.*
    Path              {log_path}
    Parser            json
    DB                /var/log/flb_app.db
    Mem_Buf_Limit     100MB
    Skip_Long_Lines   On

[FILTER]
    Name          parser
    Match         *
    Key_Name      log
    Parser        json
    Reserve_Data  On

[FILTER]
    Name    lua
    Match   *
    script  {lua_script_path}
    call    process_record

[FILTER]
    Name          modify
    Match         *
    Add           processed_by lua_filter

[OUTPUT]
    Name              es
    Match             *
    Host              {es_host}
    Port              9200
    Index             app-logs
    Retry_Limit       3
    storage.total_limit_size 10M

# Example Lua script content (save to {lua_script_path}):
# function process_record(tag, timestamp, record)
#     -- Add custom field
#     record["custom_field"] = "custom_value"
#
#     -- Transform existing field
#     if record["level"] then
#         record["severity"] = string.upper(record["level"])
#     end
#
#     -- Filter out specific records (return -1 to drop)
#     if record["message"] and string.match(record["message"], "DEBUG") then
#         return -1, timestamp, record
#     end
#
#     -- Return modified record
#     return 1, timestamp, record
# end
"""

        return config

    def _generate_stream_processor(
        self,
        log_path: str = "/var/log/app/*.log",
        es_host: str = "elasticsearch",
        **kwargs
    ) -> str:
        """
        Generate configuration with Stream Processor for advanced log processing.

        Args:
            log_path: Path to log files
            es_host: Elasticsearch hostname

        Returns:
            Complete Fluent Bit configuration string
        """
        if log_path is None:
            log_path = "/var/log/app/*.log"
        if es_host is None:
            es_host = "elasticsearch"
        config = self._generate_service_section(parsers_file="parsers.conf")

        config += f"""
[INPUT]
    Name              tail
    Tag               app.*
    Path              {log_path}
    Parser            json
    DB                /var/log/flb_app.db
    Mem_Buf_Limit     100MB
    Skip_Long_Lines   On

# Stream Processor for advanced SQL-like transformations
[STREAM_TASK]
    Name    error_aggregation
    Exec    CREATE STREAM errors AS \\
            SELECT \\
                level, \\
                COUNT(*) as error_count, \\
                TUMBLE_START() as window_start \\
            FROM TAG:'app.*' \\
            WHERE level = 'error' \\
            GROUP BY level, TUMBLE(time, INTERVAL '1' MINUTE);

[STREAM_TASK]
    Name    high_latency_detection
    Exec    CREATE STREAM high_latency AS \\
            SELECT \\
                service, \\
                endpoint, \\
                response_time_ms \\
            FROM TAG:'app.*' \\
            WHERE response_time_ms > 1000;

[FILTER]
    Name          modify
    Match         *
    Add           processed_by stream_processor

[OUTPUT]
    Name              es
    Match             app.*
    Host              {es_host}
    Port              9200
    Index             app-logs
    Retry_Limit       3

[OUTPUT]
    Name              es
    Match             errors
    Host              {es_host}
    Port              9200
    Index             error-metrics
    Retry_Limit       3

[OUTPUT]
    Name              es
    Match             high_latency
    Host              {es_host}
    Port              9200
    Index             performance-alerts
    Retry_Limit       3
"""

        return config

    def _generate_custom(self, **kwargs) -> str:
        """
        Generate minimal custom configuration template.

        Returns:
            Basic Fluent Bit configuration string
        """
        config = self._generate_service_section()

        config += """
[INPUT]
    Name              tail
    Tag               custom.*
    Path              /var/log/*.log
    DB                /var/log/flb_custom.db
    Mem_Buf_Limit     50MB
    Skip_Long_Lines   On

[FILTER]
    Name          modify
    Match         *
    Add           custom_field custom_value

[OUTPUT]
    Name          stdout
    Match         *
    Format        json_lines
"""

        return config


def main() -> None:
    """Main entry point for the configuration generator."""
    use_case_arg_map: Dict[str, list[str]] = {
        "kubernetes-elasticsearch": [
            "cluster_name",
            "environment",
            "es_host",
            "es_port",
            "es_index_prefix",
            "container_runtime",
            "tls_verify",
            "tls_ca_file",
        ],
        "kubernetes-loki": ["cluster_name", "loki_host", "loki_port", "container_runtime"],
        "kubernetes-cloudwatch": ["cluster_name", "aws_region", "log_group_name", "container_runtime"],
        "kubernetes-opentelemetry": [
            "cluster_name",
            "environment",
            "otlp_endpoint",
            "container_runtime",
            "tls_verify",
            "tls_ca_file",
        ],
        "application-multiline": [
            "environment",
            "es_host",
            "log_path",
            "app_name",
            "language",
        ],
        "syslog-forward": [
            "listen_port",
            "syslog_host",
            "syslog_port",
            "forward_host",
            "forward_port",
        ],
        "file-tail-s3": ["log_path", "file_path", "s3_bucket", "s3_region"],
        "http-kafka": ["http_port", "kafka_brokers", "kafka_topic"],
        "multi-destination": ["es_host", "s3_bucket"],
        "prometheus-metrics": [
            "cluster_name",
            "prometheus_host",
            "prometheus_port",
            "scrape_interval",
            "tls_verify",
            "tls_ca_file",
        ],
        "lua-filtering": ["es_host", "log_path", "lua_script_path"],
        "stream-processor": ["es_host", "log_path"],
        "custom": [],
    }

    parser = argparse.ArgumentParser(
        description="Generate Fluent Bit configurations",
        formatter_class=argparse.RawDescriptionHelpFormatter,
    )

    parser.add_argument(
        "--use-case",
        required=True,
        choices=[
            "kubernetes-elasticsearch",
            "kubernetes-loki",
            "kubernetes-cloudwatch",
            "kubernetes-opentelemetry",
            "application-multiline",
            "syslog-forward",
            "file-tail-s3",
            "http-kafka",
            "multi-destination",
            "prometheus-metrics",
            "lua-filtering",
            "stream-processor",
            "custom",
        ],
        help="Configuration use case to generate",
    )

    parser.add_argument(
        "--output",
        default="fluent-bit.conf",
        help="Output file path (default: fluent-bit.conf)",
    )

    # Common options
    parser.add_argument("--cluster-name", default=None, help="Kubernetes cluster name")
    parser.add_argument("--environment", default=None, help="Environment name")

    # Elasticsearch options
    parser.add_argument("--es-host", default=None, help="Elasticsearch host")
    parser.add_argument("--es-port", type=int, default=None, help="Elasticsearch port")
    parser.add_argument("--es-index-prefix", default=None, help="Elasticsearch index prefix")

    # Loki options
    parser.add_argument("--loki-host", default=None, help="Loki host")
    parser.add_argument("--loki-port", type=int, default=None, help="Loki port")

    # CloudWatch options
    parser.add_argument("--aws-region", default=None, help="AWS region")
    parser.add_argument("--log-group-name", default=None, help="CloudWatch log group")

    # OpenTelemetry options
    parser.add_argument(
        "--otlp-endpoint",
        default=None,
        help="OpenTelemetry Collector OTLP endpoint (HTTP)",
    )

    # Application options
    parser.add_argument("--log-path", default=None, help="Log file path")
    parser.add_argument("--file-path", default=None, help=argparse.SUPPRESS)
    parser.add_argument("--app-name", default=None, help="Application name")
    parser.add_argument(
        "--language",
        default=None,
        choices=["java", "python", "go", "ruby"],
        help="Language for multiline parsing",
    )

    # Kubernetes container runtime
    parser.add_argument(
        "--container-runtime",
        default=None,
        choices=["docker", "cri"],
        help=(
            "Container runtime parser for Kubernetes log tailing. "
            "Use 'cri' for containerd/CRI-O (Kubernetes 1.24+, default) "
            "or 'docker' for legacy Docker shim clusters."
        ),
    )

    # S3 options
    parser.add_argument("--s3-bucket", default=None, help="S3 bucket name")
    parser.add_argument("--s3-region", default=None, help="S3 region")

    # Syslog options
    parser.add_argument("--listen-port", type=int, default=None, help="Syslog listen port")
    parser.add_argument("--syslog-host", default=None, help="Destination syslog host")
    parser.add_argument("--syslog-port", type=int, default=None, help="Destination syslog port")
    parser.add_argument("--forward-host", default=None, help=argparse.SUPPRESS)
    parser.add_argument("--forward-port", type=int, default=None, help=argparse.SUPPRESS)

    # Kafka options
    parser.add_argument("--http-port", type=int, default=None, help="HTTP listen port")
    parser.add_argument("--kafka-brokers", default=None, help="Kafka brokers")
    parser.add_argument("--kafka-topic", default=None, help="Kafka topic")

    # Prometheus options
    parser.add_argument("--prometheus-host", default=None, help="Prometheus host")
    parser.add_argument("--prometheus-port", type=int, default=None, help="Prometheus port")
    parser.add_argument(
        "--scrape-interval",
        type=int,
        default=None,
        help="Metrics scrape interval in seconds",
    )

    # Lua options
    parser.add_argument("--lua-script-path", default=None, help="Path to Lua script")

    # TLS options (applies to elasticsearch, opentelemetry, prometheus-metrics)
    tls_group = parser.add_mutually_exclusive_group()
    tls_group.add_argument(
        "--tls-verify",
        dest="tls_verify",
        action="store_true",
        default=None,
        help="Verify TLS certificates (default: enabled)",
    )
    tls_group.add_argument(
        "--no-tls-verify",
        dest="tls_verify",
        action="store_false",
        help=(
            "Disable TLS certificate verification. "
            "Prefer --tls-ca-file for self-signed certificates instead."
        ),
    )
    parser.add_argument(
        "--tls-ca-file",
        default=None,
        help=(
            "Path to custom CA certificate file for self-signed certificates. "
            "When provided, TLS verification is always enabled."
        ),
    )

    args = parser.parse_args()

    # Generate configuration
    generator = FluentBitConfigGenerator()
    try:
        effective_args: Dict[str, Any] = vars(args).copy()
        known_use_case_keys = {name for values in use_case_arg_map.values() for name in values}
        unsupported = sorted(
            key
            for key in known_use_case_keys
            if key not in use_case_arg_map[args.use_case] and effective_args.get(key) is not None
        )
        if unsupported:
            unsupported_text = ", ".join(unsupported)
            raise ValueError(
                f"Unsupported parameter(s) for use case '{args.use_case}': "
                f"{unsupported_text}"
            )

        selected_kwargs = {
            key: effective_args[key]
            for key in use_case_arg_map[args.use_case]
            if effective_args.get(key) is not None
        }
        config = generator.generate(args.use_case, **selected_kwargs)

        # Write to file with error handling
        try:
            with open(args.output, "w", encoding="utf-8") as f:
                f.write(config)
        except PermissionError as e:
            print(f"Error: Permission denied writing to {args.output}: {e}", file=sys.stderr)
            sys.exit(1)
        except IOError as e:
            print(f"Error: Failed to write configuration file: {e}", file=sys.stderr)
            sys.exit(1)

        print(f"✓ Configuration generated successfully: {args.output}")
        print(f"\nUse case: {args.use_case}")
        print(f"\nNext steps:")
        print(f"1. Review the configuration: cat {args.output}")
        print(f"2. Customize parameters as needed")
        print(f"3. Validate the configuration: fluent-bit -c {args.output} --dry-run")
        print(f"4. Test the configuration: fluent-bit -c {args.output}")

    except ValueError as e:
        print(f"Error: {e}", file=sys.stderr)
        sys.exit(1)
    except Exception as e:
        print(f"Error: Unexpected error generating configuration: {e}", file=sys.stderr)
        sys.exit(1)


if __name__ == "__main__":
    main()

#!/usr/bin/env python3
"""
Tests for generate_config.py

Run with:
    python3 -m unittest scripts/test_generate_config.py -v
or:
    python3 -m pytest scripts/test_generate_config.py -v
"""

import re
import subprocess
import sys
import tempfile
import unittest
import warnings
from pathlib import Path

# Make sure the script directory is importable
SCRIPT_DIR = Path(__file__).parent
sys.path.insert(0, str(SCRIPT_DIR))

from generate_config import FluentBitConfigGenerator  # noqa: E402


ALL_USE_CASES = [
    "kubernetes-elasticsearch",
    "kubernetes-loki",
    "kubernetes-cloudwatch",
    "kubernetes-opentelemetry",
    "application-multiline",
    "syslog-forward",
    "file-tail-s3",
    "http-kafka",
    "multi-destination",
    "prometheus-metrics",
    "lua-filtering",
    "stream-processor",
    "custom",
]

GENERATOR = SCRIPT_DIR / "generate_config.py"


class TestAllUseCasesGenerate(unittest.TestCase):
    """Every use case must generate without raising and return a non-empty string."""

    def setUp(self):
        self.gen = FluentBitConfigGenerator()

    def _check(self, use_case, **kwargs):
        result = self.gen.generate(use_case, **kwargs)
        self.assertIsInstance(result, str)
        self.assertGreater(len(result.strip()), 0, f"{use_case} returned empty config")

    def test_kubernetes_elasticsearch(self):
        self._check("kubernetes-elasticsearch")

    def test_kubernetes_loki(self):
        self._check("kubernetes-loki")

    def test_kubernetes_cloudwatch(self):
        self._check("kubernetes-cloudwatch")

    def test_kubernetes_opentelemetry(self):
        self._check("kubernetes-opentelemetry")

    def test_application_multiline(self):
        self._check("application-multiline")

    def test_syslog_forward(self):
        self._check("syslog-forward")

    def test_file_tail_s3(self):
        self._check("file-tail-s3")

    def test_http_kafka(self):
        self._check("http-kafka")

    def test_multi_destination(self):
        self._check("multi-destination")

    def test_prometheus_metrics(self):
        self._check("prometheus-metrics")

    def test_lua_filtering(self):
        self._check("lua-filtering")

    def test_stream_processor(self):
        self._check("stream-processor")

    def test_custom(self):
        self._check("custom")

    def test_unknown_use_case_raises(self):
        with self.assertRaises(ValueError):
            self.gen.generate("nonexistent-use-case")


class TestKwargCompatibilityAndValidation(unittest.TestCase):
    """Deprecated aliases should map correctly and unknown kwargs must fail fast."""

    def setUp(self):
        self.gen = FluentBitConfigGenerator()

    def test_syslog_legacy_aliases_are_mapped_with_warning(self):
        with warnings.catch_warnings(record=True) as caught:
            warnings.simplefilter("always", DeprecationWarning)
            config = self.gen.generate(
                "syslog-forward",
                forward_host="legacy-syslog.internal",
                forward_port=1514,
            )

        self.assertIn("Host          legacy-syslog.internal", config)
        self.assertIn("Port          1514", config)

        messages = [str(item.message) for item in caught]
        self.assertTrue(any("forward_host" in msg and "syslog_host" in msg for msg in messages))
        self.assertTrue(any("forward_port" in msg and "syslog_port" in msg for msg in messages))

    def test_file_path_alias_is_mapped_with_warning(self):
        with warnings.catch_warnings(record=True) as caught:
            warnings.simplefilter("always", DeprecationWarning)
            config = self.gen.generate("file-tail-s3", file_path="/tmp/legacy.log")

        self.assertIn("Path              /tmp/legacy.log", config)
        self.assertTrue(any("file_path" in str(item.message) for item in caught))

    def test_new_kwarg_wins_when_old_and_new_are_both_passed(self):
        with warnings.catch_warnings(record=True):
            warnings.simplefilter("always", DeprecationWarning)
            config = self.gen.generate(
                "syslog-forward",
                syslog_host="preferred.example.com",
                forward_host="legacy.example.com",
            )

        self.assertIn("Host          preferred.example.com", config)
        self.assertNotIn("Host          legacy.example.com", config)

    def test_unknown_kwarg_is_rejected_for_syslog_forward(self):
        with self.assertRaisesRegex(ValueError, "Unknown parameter\\(s\\).*unexpected_flag"):
            self.gen.generate("syslog-forward", unexpected_flag=True)

    def test_unknown_kwarg_is_rejected_for_file_tail_s3(self):
        with self.assertRaisesRegex(ValueError, "Unknown parameter\\(s\\).*extra_option"):
            self.gen.generate("file-tail-s3", extra_option="x")


class TestOtlpEndpointParsing(unittest.TestCase):
    """_parse_otlp_endpoint should handle valid inputs and reject invalid ones."""

    def _parse(self, endpoint):
        return FluentBitConfigGenerator._parse_otlp_endpoint(endpoint)

    def test_host_port(self):
        host, port, uri = self._parse("collector.svc:4318")
        self.assertEqual(host, "collector.svc")
        self.assertEqual(port, 4318)
        self.assertEqual(uri, "/v1/logs")

    def test_http_url(self):
        host, port, uri = self._parse("http://otel.example.com:4318")
        self.assertEqual(host, "otel.example.com")
        self.assertEqual(port, 4318)
        self.assertEqual(uri, "/v1/logs")

    def test_url_with_base_path(self):
        host, port, uri = self._parse("http://otel.example.com:4318/prefix")
        self.assertEqual(uri, "/prefix/v1/logs")

    def test_default_port_when_omitted(self):
        host, port, uri = self._parse("collector.example.com")
        self.assertEqual(port, 4318)

    def test_invalid_endpoint_raises(self):
        with self.assertRaises(ValueError):
            self._parse("://no-host")

    def test_whitespace_stripped(self):
        host, port, uri = self._parse("  collector.svc:4318  ")
        self.assertEqual(host, "collector.svc")


class TestTlsBlock(unittest.TestCase):
    """_tls_block should produce correct lines for all four TLS combinations."""

    def setUp(self):
        self.gen = FluentBitConfigGenerator()

    def test_verify_on_no_ca(self):
        block = self.gen._tls_block(tls_verify=True, tls_ca_file=None)
        self.assertIn("tls               On", block)
        self.assertIn("tls.verify        On", block)
        self.assertNotIn("tls.ca_file", block)

    def test_verify_off_no_ca(self):
        block = self.gen._tls_block(tls_verify=False, tls_ca_file=None)
        self.assertIn("tls               On", block)
        self.assertIn("tls.verify        Off", block)
        self.assertNotIn("tls.ca_file", block)

    def test_ca_file_forces_verify_on(self):
        block = self.gen._tls_block(tls_verify=False, tls_ca_file="/etc/certs/ca.crt")
        self.assertIn("tls.verify        On", block)
        self.assertIn("tls.ca_file       /etc/certs/ca.crt", block)
        self.assertNotIn("tls.verify        Off", block)

    def test_verify_on_with_ca_file(self):
        block = self.gen._tls_block(tls_verify=True, tls_ca_file="/etc/certs/ca.crt")
        self.assertIn("tls.verify        On", block)
        self.assertIn("tls.ca_file       /etc/certs/ca.crt", block)

    def test_custom_indent(self):
        block = self.gen._tls_block(indent=8)
        self.assertTrue(block.startswith(" " * 8))


class TestContainerRuntime(unittest.TestCase):
    """K8s generators must use the correct Parser based on container_runtime."""

    def setUp(self):
        self.gen = FluentBitConfigGenerator()

    K8S_USE_CASES = [
        "kubernetes-elasticsearch",
        "kubernetes-loki",
        "kubernetes-cloudwatch",
        "kubernetes-opentelemetry",
    ]

    def _parser_lines(self, config: str) -> list[str]:
        """Return lines that contain 'Parser' outside of comments."""
        return [
            line
            for line in config.splitlines()
            if re.search(r"^\s+Parser\s", line) and not line.strip().startswith("#")
        ]

    def test_default_is_cri(self):
        for uc in self.K8S_USE_CASES:
            with self.subTest(use_case=uc):
                config = self.gen.generate(uc)
                parser_lines = self._parser_lines(config)
                self.assertTrue(
                    any("cri" in l for l in parser_lines),
                    f"{uc}: expected 'cri' parser line, got: {parser_lines}",
                )
                self.assertFalse(
                    any("docker" in l for l in parser_lines),
                    f"{uc}: unexpected 'docker' parser line, got: {parser_lines}",
                )

    def test_docker_override(self):
        for uc in self.K8S_USE_CASES:
            with self.subTest(use_case=uc):
                config = self.gen.generate(uc, container_runtime="docker")
                parser_lines = self._parser_lines(config)
                self.assertTrue(
                    any("docker" in l for l in parser_lines),
                    f"{uc}: expected 'docker' parser line, got: {parser_lines}",
                )
                self.assertFalse(
                    any(re.search(r"\bcri\b", l) for l in parser_lines),
                    f"{uc}: unexpected 'cri' in parser line, got: {parser_lines}",
                )

    def test_explicit_none_falls_back_to_cri(self):
        config = self.gen.generate("kubernetes-loki", container_runtime=None)
        parser_lines = self._parser_lines(config)
        self.assertTrue(any("cri" in l for l in parser_lines))


class TestApplicationMultiline(unittest.TestCase):
    """application-multiline INPUT must not mix Parser and Multiline.Parser."""

    def setUp(self):
        self.gen = FluentBitConfigGenerator()

    def _input_block(self, config: str) -> str:
        """Extract the first [INPUT] block from config."""
        match = re.search(r"\[INPUT\](.*?)(?=\[|\Z)", config, re.DOTALL)
        return match.group(0) if match else ""

    def test_no_plain_parser_in_input(self):
        config = self.gen.generate("application-multiline")
        input_block = self._input_block(config)
        # Must NOT have a bare `Parser` line (only Multiline.Parser is allowed)
        plain_parser = re.findall(r"^\s+Parser\s", input_block, re.MULTILINE)
        self.assertEqual(
            plain_parser,
            [],
            f"INPUT block must not contain plain 'Parser' alongside Multiline.Parser: {input_block}",
        )

    def test_multiline_parser_present(self):
        config = self.gen.generate("application-multiline", language="java")
        self.assertIn("Multiline.Parser  multiline-java", config)

    def test_ruby_language(self):
        config = self.gen.generate("application-multiline", language="ruby")
        self.assertIn("Multiline.Parser  multiline-ruby", config)
        input_block = self._input_block(config)
        plain_parser = re.findall(r"^\s+Parser\s", input_block, re.MULTILINE)
        self.assertEqual(plain_parser, [])

    def test_python_language(self):
        config = self.gen.generate("application-multiline", language="python")
        self.assertIn("Multiline.Parser  multiline-python", config)

    def test_go_language(self):
        config = self.gen.generate("application-multiline", language="go")
        self.assertIn("Multiline.Parser  multiline-go", config)


class TestPrometheusMetrics(unittest.TestCase):
    """prometheus-metrics must not contain a [FILTER] block."""

    def setUp(self):
        self.gen = FluentBitConfigGenerator()

    def test_no_filter_block(self):
        config = self.gen.generate("prometheus-metrics")
        # [FILTER] blocks silently no-op on metrics records — must be absent
        self.assertNotIn("[FILTER]", config, "prometheus-metrics must not contain a [FILTER] block")

    def test_output_contains_add_label(self):
        config = self.gen.generate("prometheus-metrics", cluster_name="test-cluster")
        self.assertIn("add_label", config)
        self.assertIn("test-cluster", config)


class TestFalsyDefaults(unittest.TestCase):
    """Fix 5: falsy values like port=0 or host='' must not be replaced by defaults."""

    def setUp(self):
        self.gen = FluentBitConfigGenerator()

    def test_zero_port_not_replaced(self):
        # Port 0 is falsy; the old `or` pattern would replace it with the default.
        config = self.gen.generate("kubernetes-loki", loki_port=0)
        self.assertIn("Port              0", config)

    def test_empty_string_host_not_replaced(self):
        # Empty string is falsy; old `or` pattern would silently revert to default.
        config = self.gen.generate("file-tail-s3", s3_bucket="")
        self.assertIn("bucket            ", config)
        # Default "my-logs-bucket" must NOT appear when caller explicitly passes ""
        self.assertNotIn("my-logs-bucket", config)

    def test_zero_scrape_interval_not_replaced(self):
        config = self.gen.generate("prometheus-metrics", scrape_interval=0)
        self.assertIn("Scrape_interval   0", config)

    def test_none_falls_back_to_default(self):
        # None should still fall back to the declared default
        config = self.gen.generate("kubernetes-loki", loki_port=None)
        self.assertIn("Port              3100", config)


class TestCLI(unittest.TestCase):
    """Smoke tests for the CLI entry point via subprocess."""

    def _run(self, *args, expect_success=True):
        result = subprocess.run(
            [sys.executable, str(GENERATOR), *args],
            capture_output=True,
            text=True,
        )
        if expect_success:
            self.assertEqual(
                result.returncode,
                0,
                f"CLI failed.\nstdout: {result.stdout}\nstderr: {result.stderr}",
            )
        return result

    def test_help(self):
        result = self._run("--help", expect_success=True)
        self.assertIn("--use-case", result.stdout)

    def test_kubernetes_loki_default(self):
        with tempfile.NamedTemporaryFile(suffix=".conf", delete=False) as tmp:
            self._run("--use-case", "kubernetes-loki", "--output", tmp.name)
            content = Path(tmp.name).read_text()
        self.assertIn("[INPUT]", content)
        self.assertIn("[OUTPUT]", content)
        self.assertIn("cri", content)

    def test_container_runtime_docker_cli(self):
        with tempfile.NamedTemporaryFile(suffix=".conf", delete=False) as tmp:
            self._run(
                "--use-case", "kubernetes-loki",
                "--container-runtime", "docker",
                "--output", tmp.name,
            )
            content = Path(tmp.name).read_text()
        self.assertIn("docker", content)
        self.assertNotIn("Parser            cri", content)

    def test_language_ruby_cli(self):
        with tempfile.NamedTemporaryFile(suffix=".conf", delete=False) as tmp:
            self._run(
                "--use-case", "application-multiline",
                "--language", "ruby",
                "--output", tmp.name,
            )
            content = Path(tmp.name).read_text()
        self.assertIn("multiline-ruby", content)

    def test_prometheus_metrics_cli(self):
        with tempfile.NamedTemporaryFile(suffix=".conf", delete=False) as tmp:
            self._run("--use-case", "prometheus-metrics", "--output", tmp.name)
            content = Path(tmp.name).read_text()
        self.assertNotIn("[FILTER]", content)

    def test_syslog_forward_legacy_aliases_cli(self):
        with tempfile.NamedTemporaryFile(suffix=".conf", delete=False) as tmp:
            self._run(
                "--use-case", "syslog-forward",
                "--forward-host", "legacy-syslog.internal",
                "--forward-port", "1514",
                "--output", tmp.name,
            )
            content = Path(tmp.name).read_text()
        self.assertIn("Host          legacy-syslog.internal", content)
        self.assertIn("Port          1514", content)

    def test_file_tail_s3_legacy_file_path_cli(self):
        with tempfile.NamedTemporaryFile(suffix=".conf", delete=False) as tmp:
            self._run(
                "--use-case", "file-tail-s3",
                "--file-path", "/tmp/legacy.log",
                "--output", tmp.name,
            )
            content = Path(tmp.name).read_text()
        self.assertIn("Path              /tmp/legacy.log", content)

    def test_unsupported_parameter_for_use_case_exits_nonzero(self):
        result = self._run(
            "--use-case", "syslog-forward",
            "--es-host", "elasticsearch.local",
            expect_success=False,
        )
        self.assertNotEqual(result.returncode, 0)
        self.assertIn("Unsupported parameter(s) for use case 'syslog-forward'", result.stderr)

    def test_unknown_use_case_exits_nonzero(self):
        result = self._run("--use-case", "nonexistent", expect_success=False)
        self.assertNotEqual(result.returncode, 0)

    def test_invalid_language_exits_nonzero(self):
        result = self._run(
            "--use-case", "application-multiline",
            "--language", "cobol",
            expect_success=False,
        )
        self.assertNotEqual(result.returncode, 0)

    def test_invalid_container_runtime_exits_nonzero(self):
        result = self._run(
            "--use-case", "kubernetes-loki",
            "--container-runtime", "podman",
            expect_success=False,
        )
        self.assertNotEqual(result.returncode, 0)


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