Cấu Hình Logstash Sửa Đổi Bắn Chéo

ưới đây là một phiên bản sửa đổi của cấu hình Logstash của bạn, được cải thiện về cấu trúc và cú pháp để dễ dàng sử dụng.

Cấu Hình Logstash Đã Được Sửa Đổi

input {
  kafka {
    bootstrap_servers => "10.100.30.32:9092"
    topics => ["ESMART-CATEGORY-LOGS", "ESMART-GATEWAY-LOGS"]
    group_id => "log_consumer_group"
    auto_offset_reset => "earliest"
  }
}

filter {
  grok {
    match => {
      "message" => """
        {"instant":{"epochSecond":%{NUMBER:epoch_sec}," +
        "\"nanoOfSecond\":%{NUMBER:nano_sec}," +
        "\"thread\":\"%{DATA:thread}\"," +
        "\"level\":\"%{WORD:level}\"," +
        "\"loggerName\":\"%{DATA:logger_name}\"," +
        "\"message\":\"%{GREEDYDATA:log_message}\"," +
        "\"endOfBatch\":%{GREEDYDATA:end_of_batch}," +
        "\"loggerFqcn\":\"%{DATA:logger_fqcn}\"," +
        "\"contextMap\":{\"traceId\":\"%{DATA:trace_id}\"," +
        "\"spanId\":\"%{DATA:span_id}\"," +
        "\"className\":\"%{DATA:class_name}\"," +
        "\"clientIp\":\"%{IP:client_ip}\"," +
        "\"clientMessageId\":\"%{DATA:client_message_id}\"," +
        "\"clientTime\":\"%{TIMESTAMP_ISO8601:client_time}\"," +
        "\"duration\":%{NUMBER:duration}," +
        "\"methodName\":\"%{DATA:method_name}\"," +
        "\"path\":\"%{DATA:path}\"," +
        "\"stackTrace\":\"%{GREEDYDATA:stack_trace}\"}," +
        "\"threadId\":%{NUMBER:thread_id}," +
        "\"threadPriority\":%{NUMBER:thread_priority}," +
        "\"logType\":\"%{WORD:log_type}\"," +
        "\"application\":\"%{DATA:application}\"," +
        "\"localIp\":\"%{IP:local_ip}\"}
      """
    }
  }
  
  mutate {
    add_field => {
      "ts" => "%{epoch_sec}"
      "ip" => "%{client_ip}"
      "msgId" => "%{client_message_id}"
      "duration" => "%{duration}"
      "method" => "%{method_name}"
      "path" => "%{path}"
      "app" => "%{application}"
      "localIp" => "%{local_ip}"
      "logType" => "%{log_type}"
    }
    remove_field => [
      "instant", "thread", "level", "logger_name", "message", 
      "endOfBatch", "logger_fqcn", "contextMap", 
      "threadId", "threadPriority"
    ]
  }

  date {
    match => [ "client_time", "ISO8601" ]  # Changed to match the correct field
    timezone => "+07:00"
    target => "@timestamp"
  }
  
  ruby {
    code => "event.set('indexDay', event.get('@timestamp').time.localtime('+07:00').strftime('%Y%m%d'))"
  }
}

output {
  elasticsearch {
    hosts => ["http://10.152.183.57:9200"]
    template => "/usr/share/logstash/templates/logstash_template.json"
    template_name => "logstash"
    template_overwrite => true
    index => "logstash-%{indexDay}"
    document_type => "_doc"
  }
  stdout {
    codec => rubydebug
  }
}

Các Điểm Cần Lưu Ý