Elasticsearchの仕切り直し導入 その7（CentOS 7.5、2019/01/27）

Elasticsearchの使い方がなんとなく使い方が分かってきたので、CISCOのsyslogを取り込んでみます。Logstashは様々なInputに対応しており、下記にpluginの一覧とその説明があります。
https://www.elastic.co/guide/en/logstash/current/input-plugins.html

今回はsyslogプラグインを使うことにします。syslogプラグインを使うと、hostやfacilityなどを自動的にデコードしてくれます。データ取り込み時に@timestampが設定されますが、CISCOのsyslogメッセージ内にもタイムスタンプが埋め込まれているので、分かりやすいように後者も抽出してみることにします。
なお、CISCO側の設定は下記のようになっています。Logstash側での処理を考えると、既存のsyslogサービスとポート番号がバッティングしないよう、このようにしています。

Router-C841MJ#show running-config
service timestamps log datetime msec localtime show-timezone year
clock timezone JST 9 0
logging facility local0
logging host 192.168.100.10 transport udp port 1514

CISCOのsyslogメッセージ内のタイムスタンプ（後述のmessageがsyslogの生データになります）を切り出すにあたり、下記パターンを独自に定義しています。実際のところ、c_monthなどは後で使わないので名前を付ける必要はないのですが、デバッグも兼ねてこのようにしています。

[root@centos ]# cat /etc/logstash/patterns/pattern-cisco
CISCOTIME     %{MONTH:c_month} %{MONTHDAY:c_mnum} %{YEAR:c_year} %{HOUR:c_hour}:%{MINUTE:c_min}:%{SECOND:c_second}

なお、Logstashに標準で定義されている正規表現パターンは下記を参照して下さい。
https://github.com/elastic/logstash/blob/v1.4.2/patterns/grok-patterns

今回作ったコンフィグは下記の通りです。

[root@centos ]# cat cisco-syslog.conf

input {
  # cisco syslog
  syslog {
    port => 1514
    type => "syslog-cisco-841mj"
  }
  # debug
  stdin {}
}

filter {

  grok {
    patterns_dir => "/etc/logstash/patterns/"

    match => [ "message",
      "%{CISCOTIME:tmp_timestamp} %{WORD}: %{GREEDYDATA:cisco_syslog_message}"
    ]
  }

  date {
    match => [ "tmp_timestamp", "MMM dd yyyy HH:mm:ss.SSS" ]
    locale => "en"
    timezone => "Asia/Tokyo"
    target => "@c_timestamp"
  }

}

output {
  if "_grokparsefailure" in [tags] {
    file {
      path => "/var/log/logstash/fail.log"
    }
  }
  stdout { codec => rubydebug }

}

これで実際にsyslogを送出すると、下記のような出力が得られます。

{
                   "c_min" => "05",
                 "c_month" => "Jan",
                    "tags" => [
        [0] "_grokparsefailure_sysloginput"
    ],
          "severity_label" => "Emergency",
                  "c_mnum" => "27",
    "cisco_syslog_message" => "%SYS-5-CONFIG_I: Configured from console by test on vty0 (192.168.100.11)",
                "facility" => 0,
                "priority" => 0,
                    "type" => "syslog-cisco-841mj",
           "tmp_timestamp" => "Jan 27 2019 20:05:45.316",
            "@c_timestamp" => 2019-01-27T11:05:45.316Z,
                "severity" => 0,
              "@timestamp" => 2019-01-27T11:05:46.379Z,
          "facility_label" => "kernel",
                  "c_year" => "2019",
                 "message" => "<133>12919: Jan 27 2019 20:05:45.316 JST: %SYS-5-CONFIG_I: Configured from console by test on vty0 (192.168.100.11)",
                    "host" => "192.168.100.11",
                  "c_hour" => "20",
                "c_second" => "45.316",
                "@version" => "1"
}

出力を簡単に説明します。
c_min：syslog生データ（message）から、独自パターンでマッチさせた「分」データ
c_month：syslog生データ（message）から、独自パターンでマッチさせた「月」データ
_grokparsefailure_sysloginput：syslogメッセージのデコードに失敗している場合に出力されるものと思いますが、どうしようもないので放置
c_mnum：syslog生データ（message）から、独自パターンでマッチさせた「日」データ
cisco_syslog_message：syslog生データ（message）から、ゴミを除いたsyslogメッセージの実体
type：処理ルーチンの中で私が加えた値
tmp_timestamp：syslog生データ（message）から、独自パターンでマッチさせたタイムスタンプ部分を格納
@c_timestamp：syslog生データ（message）から、独自パターンでマッチさせたタイムスタンプ部分を時刻型に変換したもの
@timestamp：データ受信時に刻印したタイムスタンプ（自動生成）
c_year：syslog生データ（message）から、独自パターンでマッチさせた「年」データ
message：syslog生データそのもの（自動生成）
host：syslog送出元（自動生成）
c_hour：syslog生データ（message）から、独自パターンでマッチさせた「時」データ
c_second：syslog生データ（message）から、独自パターンでマッチさせた「秒」データ

タグ：elasticsearch