как я могу извлечь исходные имена хостов из журналов JSON на logstash

я собираю журналы с помощью OSSEC и пересылаю журналы JSON в logstash с помощью logstash-forwarder. это моя конфигурация logstash.

input {   
  lumberjack {
    port => 10516
    type => "lumberjack"
    ssl_certificate => "/etc/pki/tls/certs/logstash-forwarder.crt"
    ssl_key => "/etc/pki/tls/private/logstash-forwarder.key"
    codec => json
    }
}

filter {
  json {
    source => "message"
  }
}
output {
  elasticsearch {
    host => localhost
  }
}

я хотел бы извлечь хост, указанный в поле «местоположение» внутри скобок, и создать специальный тег, потому что logstash видит OSSEC только как исходный хост, потому что он пересылает журналы. ниже приведен пример вывода logstash.

{
  "_index": "logstash-2015.09.23",
  "_type": "ossec-alerts",
  "_id": "AU_4Q1Hc5OjGfEBnRiWa",
  "_score": null,
  "_source": {
    "rule": {
      "level": 3,
      "comment": "Nginx error message.",
      "sidid": 31301
    },
    "srcip": "192.168.192.10",
    "location": "(logstash) 192.168.212.104->/var/log/nginx/error.log",
    "full_log": "2015/09/23 11:33:24 [error] 1057#0: *562 connect() failed (111: Connection refused) while connecting to upstream, client: 192.168.192.10, server: _, request: \"POST /elasticsearch/.kibana/__kibanaQueryValidator/_validate/query?explain=true&ignore_unavailable=true HTTP/1.1\", upstream: \"http://[::1]:5601/elasticsearch/.kibana/__kibanaQueryValidator/_validate/query?explain=true&ignore_unavailable=true\", host: \"192.168.212.104\", referrer: \"http://192.168.212.104/\"",
    "@version": "1",
    "@timestamp": "2015-09-23T03:33:25.588Z",
    "type": "ossec-alerts",
    "file": "/var/ossec/logs/alerts/alerts.json",
    "host": "ossec",
    "offset": "51048"
  },
  "fields": {
    "@timestamp": [
      1442979205588
    ]
  },
  "sort": [
    1442979205588
  ]
}

json filter logstash logstash-forwarder
person Jace    schedule 23.09.2015    source источник

Ответы (2)


arrow_upward
1
arrow_downward

Вам нужен фильтр Grok. Вы можете использовать отладчик grok, чтобы найти наиболее подходящие шаблоны. Следующий шаблон должен работать для вашего поля location:

\(%{HOST:host}\) %{IP:srcip}->%{PATH:path}

В разделе фильтра logstash:

grok {
    match => { "location" => "\(%{HOST:host}\) %{IP:srcip}->%{PATH:path}" }
    overwrite => [ "host", "srcip" ]
}

overwrite необходимо, потому что у вас уже есть поля host и srcip.

person hurb    schedule 23.09.2015

arrow_upward
1
arrow_downward

После того как вы примените фильтр json{}, у вас останется куча полей. Теперь к этим полям можно применять дополнительные фильтры, включая grok{} для создания дополнительных полей!

person Alain Collins    schedule 23.09.2015
comment
О, очень плохо. Извините, это было довольно одновременно. Опять ты больше рыбак ;) - person hurb; 23.09.2015
comment
Виновен по обвинению! :) - person Alain Collins; 23.09.2015