Postfix 日志 + Logstash + 聚合答案

【问题标题】：Postfix Logs + Logstash + AggregatePostfix 日志 + Logstash + 聚合
【发布时间】：2019-10-29 05:44:26
【问题描述】：

我无法正确配置 Logstash。我关心的后缀日志中有两行：

Jun 14 09:06:22 devmailforwarder postfix/smtp[1994]: A03CA9F532: to=<person@gmail.com>, relay=server[0.0.0.0]:25, delay=0.02, delays=0.01/0.01/0/0.01, dsn=2.0.0, status=sent (250 2.0.0 Ok: queued as A0B4D5C49)

Jun 14 09:15:04 devmailforwarder postfix/cleanup[2023]: 0E1969F533: warning: header Subject: subjectline from server[0.0.0.0]; from=<from@gmail.com> to=<to@gmail.com> proto=SMTP helo=<server>

我的 grok 过滤器模式是：

POSTFIX_QUEUEID ([0-9A-F]{6,}|[0-9a-zA-Z]{15,})
POSTFIX_STATUS (?<=status=)(.*)(?= \()
POSTFIX_PROCESS (?=postfix\/)(.*?\[)(.*?)(?=: )
POSTFIX_TO (?<=to=<)(.*?)(?=>,)
POSTFIX_RELAY (?<=relay=)(.*?)(?=,)
POSTFIX_SUBJECT (?<=Subject: )(.*)(?= from )

SMTP  ^%{SYSLOGTIMESTAMP:timestamp}%{SPACE}%{DATA:hostname}%{SPACE}%{POSTFIX_PROCESS:process}%{GREEDYDATA}%{POSTFIX_QUEUEID:queueid}%{GREEDYDATA}%{POSTFIX_TO:to}%{GREEDYDATA}%{POSTFIX_RELAY:relay}%{GREEDYDATA}%{POSTFIX_STATUS:status}%{SPACE}%{GREEDYDATA:response}
CLEANUP ^%{SYSLOGTIMESTAMP:timestamp}%{SPACE}%{DATA:hostname}%{SPACE}%{POSTFIX_PROCESS:process}:%{SPACE}%{POSTFIX_QUEUEID:queueid}%{GREEDYDATA}%{POSTFIX_SUBJECT:subject}%{GREEDYDATA:something2}

（非工作）Logstash 配置为：

input {
  file {
    path => "/var/log/mail.log*"
    exclude => "*.gz"
    start_position => "beginning"
    type => "postfix"
  }
}
    filter {
  grok {
    patterns_dir => ["/etc/logstash/conf.d/patterns"]
    match => { "message" => ["%{SMTP}", "%{SUBJECT}"] }
  }

  if "_grokparsefailure" in [tags] {
    drop {}
  }

  mutate {
    add_field => { "logstashSource" => "source-server" }
  }

  aggregate {
    task_id => "%{POSTFIX_QUEUEID}"
      code => "
        map['to'] ||= event.get('to')
        map['from'] ||= event.get('from')
        map['relay'] ||= event.get('relay')
        map['status'] ||= event.get('status')
        map['response'] ||= event.get('response')
        map['from'] ||= event.get('timestamp')
        map['relay'] ||= event.get('hostname')
        map['status'] ||= event.get('process')
        map['response'] ||= event.get('queueid')
        map['subject'] ||= event.get('subject')
        "
    map_action => "create_or_update"
    push_previous_map_as_event => true
    timeout => 2
    timeout_tags => ['aggregated']
  }
}

output {
  if [type] == "postfix" {
    file {
      path => "/var/log/logstash/postfix.log"
    }
  }
}

我的目标是创建一个包含每个字段的 elasticsearch 文档。清理消息总是首先出现在日志中。日志由唯一的队列 ID 匹配。我正在努力使聚合件正常工作。

【问题讨论】：

标签： logstash postfix-mta

【解决方案1】：

解决了。配置如下。还需要更新logstash.yml来添加

pipeline.workers: 1

filter {
  grok {
    patterns_dir => ["/etc/logstash/conf.d/patterns"]
    match => { "message" => ["%{SMTP}", "%{SUBJECT}", "%{CONNECTION}"] }
  }

  if "_grokparsefailure" in [tags] {
    drop {}
  }

  mutate {
    add_field => { "logstashSource" => "logstash-server-name" }
  }

  if ("" in [queueid]) {
    aggregate {
      task_id => "%{queueid}"
        code => "
          map['to'] ||= event.get('to')
          map['from'] ||= event.get('from')
          map['relay'] ||= event.get('relay')
          map['status'] ||= event.get('status')
          map['response'] ||= event.get('response')
          map['from'] ||= event.get('timestamp')
          map['relay'] ||= event.get('hostname')
          map['status'] ||= event.get('status')
          map['subject'] ||= event.get('subject')
          map['queueid'] ||= event.get('queueid')
          "
      timeout => 2
      timeout_tags => ['aggregated']
      map_action => 'create_or_update'
      push_map_as_event_on_timeout => true
    }
  }
}

output {
  if ("aggregated" in [tags] or "" in [connection])
  {
    elasticsearch {
      index => "postfix-%{+YYYY.MM.dd}"
      hosts => "your-es-host-here"
    }
    file {
      path => "/var/log/logstash/postfix.log"
    }
  }
}

【讨论】：