Distribute the multiple output in Logstash

In Logstash, since the configured Config becomes effective as a whole, it becomes a single output setting with a simple setting. Therefore, it is possible to set multiple outputs by conditionally branching according to items with if.

Based on the generic design introduced in this article last time, add a setting to distribute and distribute the destinations from Logstash to plural.

Setting up Logstash

• input / filter

The configuration file of input is as follows. Monitor and collect syslog logs. By tags, implement the same processing as Fluentd (td-agent).

input {
  file {
    path => "/var/log/messages"
    tags => ["sys", "logstash_messages", "%{host}" ]
    type => syslog
  }
  file {
    path => "/var/log/cron"
    tags => ["sys", "logstash_cron", "%{host}" ]
    type => syslog
  }
  file {
    path => "/var/log/secure"
    tags => ["sys", "logstash_secure", "%{host}" ]
    type => syslog
  }
}

filter {
  if [type] == "syslog" {
    grok {
      match => { "message" => "%{SYSLOGTIMESTAMP:syslog_timestamp} %{SYSLOGHOST:syslog_hostname} %{DATA:syslog_program}(?:\[%{POSINT:syslog_pid}\])?: %{GREEDYDATA:syslog_message}" }
      add_field => [ "received_at", "%{@timestamp}" ]
      add_field => [ "received_from", "%{host}" ]
    }
    date {
      match => [ "syslog_timestamp", "MMM  d HH:mm:ss", "MMM dd HH:mm:ss" ]
    }
  }
}

For this time, /var/log/secure and /var/log/cron should be made readable by chmod 644 for testing. If it is the default it can not be opened at 600 and the following error occurs.

[2017-04-10T02:14:50,228][WARN ][logstash.inputs.file     ] failed to open /var/log/cron: Permission denied - /var/log/cron
[2017-04-10T02:14:50,519][WARN ][logstash.inputs.file     ] failed to open /var/log/secure: Permission denied - /var/log/secure

• output

The settings for output are as follows. Similarly to <match tag> of Fluentd (td-agent), send and output destinations by tags. When comparing the items of the array, it becomes as follows. Elasticsearch used the same server this time, but it can specify separately. Incidentally, the filter plugin already uses a simple if branch.

The comparison basically follows the grammar of Ruby. See the official for details.

/etc/logstash/conf.d/output.conf

output {
  if [tags][1] == "logstash_messages" {
    elasticsearch {
      hosts => ["localhost:9200"]
      index => "%{tags[0]}.%{tags[1]}-%{+YYYY.MM.dd}"
    }
    file {
      path => "/var/log/logstash/%{tags[0]}/%{tags[1]}.log"
    }
  } else if [tags][1] == "logstash_cron" {
    elasticsearch {
      hosts => ["localhost:9200"]
      index => "%{tags[0]}.%{tags[1]}-%{+YYYY.MM.dd}"
    }
    file {
      path => "/var/log/logstash/others/%{tags[1]}.log"
    }
  } else {
    file {
      path => "/var/log/logstash/others/other.log"
    }
  }
}

To compare the contents of the array, specify [<type>] [<number>]. Normally it is [<type>] only.

Implementation result

Elasticsearch's index is as set, only messages and cron are created.

$ curl http://localhost:9200/sys.logstash_cron-*
{"sys.logstash_cron-2017.04.09":{"aliases":{},"mappings":{"syslog":...

$ curl http://localhost:9200/sys.logstash_messages-*
{"sys.logstash_messages-2017.04.09":{"aliases":{},"mappings":{"syslog":...

$ curl http://localhost:9200/sys.logstash_secure-*
{}

Backup to file is as follows. A file is created as set.

$ sudo tree /var/log/logstash/

/var/log/logstash/
├── logstash-plain.log
├── others
│   ├── logstash_cron.log
│   └── other.log
└── sys
    └── logstash_messages.log

Conclusion - Distribute the multiple output in Logstash

By using the if statement, Logstash was able to distribute log destinations to multiple contents for each content. Depending on the log type, Elasticsearch, file etc. Flexible storage method can be set.