Assets

input {
    kafka {
        topics => ["sensor01"]
        bootstrap_servers => "kafka:9092"
        auto_offset_reset => "latest"
        group_id => "digitbrain-logstash-0"
        decorate_events => "extended"
    }
}
filter {
    mutate {
        add_field => {
            "lc_topic" => "%{[@metadata][kafka][topic]}"
            "kafka.timestamp" => "%{[@metadata][kafka][timestamp]}"
            "kafka.partition" => "%{[@metadata][kafka][partition]}"
            "kafka.offset" => "%{[@metadata][kafka][offset]}"
        }
    }
    mutate {
        lowercase => [ "lc_topic" ]
    }
    json {
        source => "message"
    }
}
output {
    elasticsearch {
        hosts => ["https://es01:9200"]
        ilm_rollover_alias => "kafka-%{[lc_topic]}"
        ilm_enabled => "true"
        ilm_pattern => "000001"
        ssl_verification_mode => "none"
        user => "elastic"
        # This should match the password set in .env for ELASTIC_PASSWORD
        password => "elastic"
    }
}

---