logstash清洗日志处理

用 date 插件 实现

input {
  beats {
    port => 5044
  }
}

filter {
  # 如果日志是整个 message，需要先用 grok ，也实现从日志中新增字段：logtime、loglevel、appname、pid、traceid
  grok {
    match => { "message" => "%{TIMESTAMP_ISO8601:logtime} \[%{SPACE}%{LOGLEVEL:loglevel}\]\[%{DATA:appname}\]\[%{NUMBER:pid}\]\[(?:%{NUMBER:traceid}|)\]" }
  }

  # 用 date 插件替换 @timestamp
  date {
    match => ["logtime", "ISO8601"]   # 如果是标准 ISO8601 格式
    timezone => "Asia/Shanghai"       # 明确时区
    target => "@timestamp"            # 默认就是 @timestamp，可以省略
  }

  # 可选：删除原字段
  mutate {
    remove_field => ["@version", "event", "agent", "hosts"]
  }
}

output {
  elasticsearch {
    hosts => ["http://localhost:9200"]
## 动态生成索引，这里在filebeat 中使用fields新增可选字段project、env、log_type，这里引用
    index => "%{[fields][project]}-%{[fields][env]}-%{[fields][log_type]}%{+YYYY.MM.dd}"
  }
  stdout { codec => rubydebug }
}