ELFK采集Oracle日志：使用Logstash从Kafka消费Oracle日志，并吐到ES

input {
  kafka {
    bootstrap_servers => "kafka-server1:9092,kafka-server2:9092"
    topics => ["oracle-logs"]
    group_id => "logstash-oracle-group"
    consumer_threads => 3
    codec => "json"
  }
}
 
filter {
  date {
    match => ["timestamp", "yyyy-MM-dd HH:mm:ss.SSS"]
    target => "@timestamp"
  }
  mutate {
    remove_field => ["timestamp", "host", "path", "logsource"]
  }
}
 
output {
  elasticsearch {
    hosts => ["http://es-server1:9200", "http://es-server2:9200"]
    index => "oracle-logs-%{+YYYY.MM.dd}"
    document_type => "oracle_log_entry"
  }
}

这个配置文件定义了Logstash的输入、过滤和输出。它从Kafka的"oracle-logs"主题读取JSON格式的Oracle日志事件，将时间戳转换为Logstash可以理解的格式，并且调整事件结构，最后将处理过的数据送入Elasticsearch，索引名以日期格式命名。这个配置假设你已经有一个运行中的Kafka集群和Elasticsearch集群。