Журналы Postfix в ElasticSearch.

# begin input
input {
    # в бою
    udp {
       type => syslog
       port => 10555
    }

    # для отладки (в файле должна быть отдельная последняя пустая строка)
    #file {
    #    path => "/tmp/testmail.log"
    #    start_position => "beginning"
    #    sincedb_path => "/dev/null"
    #    ignore_older => 0
    #}
}
# end input

# begin filter
filter {

    # начинаем парсинг
    if [type] == "syslog" and [message] =~ "postfix" {
    #if [message] =~ "postfix" {

        # мешают парсить не нужные строки - drop

        # 1 мешает PIX - нет дополнительной информации и мешает секции if postfix/smtp
        # postfix/smtp[8026]: E854E1C0848: enabling PIX workarounds: disable_esmtp for mail.almaz.ru[1.2.3.3]:25
        if [message] =~ "PIX" { drop {} }

        # 2 не нужен вывод policy-spf
        # postfix/policy-spf[29399]: Policy action=DUNNO
        if [message] =~ "postfix/policy-spf" { drop {} }

        # 3 удаляем bounce - так как не знаем как корректно связать две очереди в строках уведомления. Само встречное письмо о доставке обрабатывается корректно
        # postfix/bounce[1736]: 640A31C0848: sender non-delivery notification: 235071C085B
        if [message] =~ "postfix/bounce" { drop {} }

        # 4 удаляем warning по поводу отсутствия записей A PTR у спам сессий. Postfix позже всё равное сделает NOQUEUE
        # postfix/smtpd[20313]: warning: hostname
        if [message] =~ "warning: hostname" { drop {} }

        # 5 удаляем статистику от Postfix session count and request rate control
        # postfix/anvil[23300]: statistics: max connection rate 2/60s for (smtp:8.183.149.173) at Jul 21 06:18:06
        # postfix/anvil[23300]: statistics: max connection count 2 for (smtp:9.255.91.105) at Jul 21 06:21:53
        # postfix/anvil[23300]: statistics: max cache size 3 at Jul 21 06:18:34
        if [message] =~ "postfix/anvil" { drop {} }

        # 6 удаляем строки не созданных SMTP сессий и вышедших по таймаутам ... всё равно номера SMTP сессии - нет
        # postfix/smtpd[27347]: timeout after END-OF-MESSAGE from ...
        # postfix/smtpd[11696]: timeout after NOOP from ...
        if [message] =~ "timeout after" { drop {} }

        # конец drop

        # список шаблонов для парсинга SMTP сессии Postfix
        grok {
            patterns_dir => ["/etc/logstash/patterns.d/"]
            match => {
                "message" => [
                        "%{SYSLOGTIMESTAMP:syslog_timestamp} %{SYSLOGHOST:syslog_hostname} %{DATA:program}(?:\[%{POSINT:syslog_pid}\])?: %{POSTFIX_MSG_ID:taskid}: client=(?<client_hostname>[^\]]+)\[%{IP:client_ip}\]",
                        "%{SYSLOGTIMESTAMP:syslog_timestamp} %{SYSLOGHOST:syslog_hostname} %{DATA:program}(?:\[%{POSINT:syslog_pid}\])?: %{POSTFIX_MSG_ID:taskid}: message-id=<(?<message_id>[^>]+)>.*",
                        "%{SYSLOGTIMESTAMP:syslog_timestamp} %{SYSLOGHOST:syslog_hostname} %{DATA:program}(?:\[%{POSINT:syslog_pid}\])?: %{POSTFIX_MSG_ID:taskid}: from=<(?<t>(?<from_mail>[^>]+)?)>, size=%{INT:size}, nrcpt=%{INT:nrcpt}.*",
                        "%{SYSLOGTIMESTAMP:syslog_timestamp} %{SYSLOGHOST:syslog_hostname} %{DATA:program}(?:\[%{POSINT:syslog_pid}\])?: %{POSTFIX_MSG_ID:taskid}: to=<(?<to_mail>[^>]+)>,.* relay=%{NOTSPACE:relay},.* status=%{NOTSPACE:status} \((?<status_msg>.+)\).*",
                        "%{SYSLOGTIMESTAMP:syslog_timestamp} %{SYSLOGHOST:syslog_hostname} %{DATA:program}(?:\[%{POSINT:syslog_pid}\])?: %{POSTFIX_MSG_ID:taskid}: removed",
                        "%{SYSLOGTIMESTAMP:syslog_timestamp} %{SYSLOGHOST:syslog_hostname} %{DATA:program}(?:\[%{POSINT:syslog_pid}\])?: %{POSTFIX_MSG_ID:taskid}: milter-reject: (?<status>END-OF-MESSAGE) from (?<client_hostname>[^\]]+)\[%{IP:client_ip}\]: (?<status_msg>.+); from=<(?<from_mail>[^>]+)> to=<(?<to_mail>[^>]+)>.*",
                        "%{SYSLOGTIMESTAMP:syslog_timestamp} %{SYSLOGHOST:syslog_hostname} %{DATA:program}(?:\[%{POSINT:syslog_pid}\])?: %{DATA:taskid}: (?<status>.+):.*from %{DATA:client_hostname}\[(?<client_ip>%{IP})\]: %{DATA:status_msg}; from=<(?<t>(?<from_mail>[^>]+)?)> to=<(?<to_mail>[^>]+)>.*",
                        "%{SYSLOGTIMESTAMP:syslog_timestamp} %{SYSLOGHOST:syslog_hostname} %{DATA:program}(?:\[%{POSINT:syslog_pid}\])?: %{NOTSPACE:status}: %{POSTFIX_MSG_ID:taskid}: (?<status_msg>.+)"
                            ]
                    }
        }
        # end grok

        # если нет имени очереди, то MTA отверг (reject) письмо, не создав SMTP сессию с taskid
        # postfix/smtpd[25614]: NOQUEUE: reject: RCPT from ....
        if [taskid] == "NOQUEUE" {
            mutate {
                remove_field => ["host", "message", "beat.version", "beat.hostname", "prospector.type", "input.type", "offset", "type", "t", "event"]
                add_tag => [ "postfix" ]
                convert => [
                    "size", "integer",
                    "nrcpt", "integer",
                    "syslog_pid", "integer"
                ]
            }
        } else if [program] == "postfix/smtpd" and [client_ip] {
            # все входящие сессии (из Интернет или наши пользователи) обладают строкой кто именно к нам подсоединился - создаём карту create_or_update
            # postfix/smtpd[12406]: connect from mail.almaz.ru[1.2.3.4]

            aggregate {
                task_id => "%{taskid}"
                code => "
                    map['debug'] ||= [];
                    map['debug'] << 1;
                    map['client_hostname'] ||= event.get('client_hostname')
                    map['client_ip'] ||= event.get('client_ip')
                    map['program'] ||= event.get('program')
                    event.cancel()
                        "
                map_action => create_or_update
            }
        } else if [program] == "postfix/cleanup" and [message_id] and ![status] {
            # когда сам сервер пишет письмо-уведомление-о-доставке, то нет никакого коннекта к нам - создаём карту create_or_update
            # postfix/cleanup[25466]: FFFFE1C0848: message-id=<ss777@mail.your-domain.ru>

            aggregate {
                task_id => "%{taskid}"
                code => "
                    map['debug'] ||= [];
                    map['debug'] << 2;
                    map['message_id'] ||= event.get('message_id')
                    event.cancel()
                        "
                map_action => create_or_update
            }
        } else if [program] == "postfix/qmgr" and [nrcpt] {
            # В письме должно быть указание КТО написал письмо или <> NULL SENDER, который ниже заменим на NULL_SENDER_RFC5321 - обновляем карту
            # postfix/qmgr[1788]: 5A05E1C053C: from=<user@domain.ru>, size=1333858, nrcpt=1 (queue active)
            # postfix/qmgr[1788]: FFFFE1C0848: from=<>, size=4353, nrcpt=1 (queue active)

            aggregate {
                task_id => "%{taskid}"
                code => "
                    map['debug'] ||= [];
                    map['debug'] << 3;
                    map['from_mail'] ||= event.get('from_mail')
                    if map['from_mail'].nil?
                        map['from_mail'] = 'NULL_SENDER_RFC5321'
                    end

                    map['size'] ||= event.get('size')
                    map['nrcpt'] ||= event.get('nrcpt')
                    event.cancel()
                        "
                map_action => update
            }
        } else if ( [program] == "postfix/smtp" or [program] == "postfix/virtual" ) {
            # доставка письма если внешнему серверу (OUT), то smtp. А если письмо нам (IN), то virtual. обновляем карту
            # postfix/smtp[24636]: 6F7971C048F: to=<external-user@gmail.com>, relay=gmail-smtp-in.l.google.com[142.250.102.27]:25, delay=3.1, delays=0.63/0/1.8/0.59, dsn=2.0.0, status=sent (250 2.0.0 OK 111132 x15-20020 - gsmtp)
            # postfix/virtual[27549]: 0FA4D1C0520: to=<your-user@your-domain.ru>, relay=virtual, delay=0.53, delays=0.44/0.01/0/0.08, dsn=2.0.0, status=sent (delivered to maildir)

            aggregate {
                task_id => "%{taskid}"
                code => "
                    map['debug'] ||= [];
                    map['debug'] << 4;
                    map['to_mail'] ||= [];
                    map['to_mail'] << event.get('to_mail');
                    map['relay'] ||= [];
                    map['relay'] << event.get('relay');
                    map['status'] ||= [];
                    map['status'] << event.get('status')
                    map['status_msg'] ||= [];
                    map['status_msg'] << event.get('status_msg')
                    event.cancel()
                        "
                map_action => update
            }
        } else if ( ( [program] == "postfix/qmgr" and ![nrcpt] ) or ( [program] == "postfix/cleanup" and [status] ) or ( [program] == "postfix/smtpd" and [status_msg] ) ) {
            # нормальная SMTP сессия заканчивается ( qmgr И нет nrcpt )
            # postfix/qmgr[1788]: FFFFE1C0848: removed
            # ИЛИ
            # сессию прекратил milter И вынес вердикт по спаму и вирусам
            # postfix/cleanup[29093]: 153E31C03DF: milter-reject:
            # закрываем карту
            # ИЛИ
            # Postfix прекратил сессию из-за превышения различных лимитов
            # postfix/smtpd[703]: warning: 097781C053C: queue file size limit exceeded

            aggregate {
                task_id => "%{taskid}"
                code => "
                    map['debug'] ||= [];
                    map['debug'] << 5;

                    map['status'] ||= [];
                    map['status'] << event.get('status')

                    map['from_mail'] ||= event.get('from_mail')

                    map['to_mail'] ||= [];
                    map['to_mail'] << event.get('to_mail');

                    map['status_msg'] ||= [];
                    map['status_msg'] << event.get('status_msg')

                    event.set('debug',map['debug']);
                    event.set('client_hostname',map['client_hostname']);
                    event.set('client_ip',map['client_ip']);
                    event.set('from_mail',map['from_mail']);
                    event.set('size',map['size']);
                    event.set('nrcpt',map['nrcpt']);
                    event.set('to_mail',map['to_mail']);
                    event.set('relay',map['relay']);
                    event.set('status',map['status']);
                    event.set('status_msg',map['status_msg']);
                    event.set('program',map['program']);
                    event.set('message_id',map['message_id']);
                        "

                map_action => update
                end_of_task => true
                timeout_task_id_field => "taskid"
                timeout_timestamp_field  => "@timestamp"
                timeout => 346000
                push_map_as_event_on_timeout => false
                push_previous_map_as_event => false
                timeout_tags => ['_aggregatetimeout']
            }
            # end aggregate
        mutate {
            convert => [
                "size", "integer",
                "nrcpt", "integer",
                "syslog_pid", "integer"
            ]
            remove_field => ["host", "message", "beat.version", "beat.hostname", "prospector.type", "input.type", "offset", "type", "t", "event"]
            add_tag => [ "postfix" ]
        }
    }
    # end if else ALL
 }
 # end if message postfix
}
# end filter

# begin output
output {

    if "postfix" in [tags] {
        # для отладки - если есть ошибки, то скидываем в отдельный текстовый файл
        if "_grokparsefailure" in [tags] or "_aggregatetimeout" in [tags] or "_aggregateexception" in [tags] {
            file {
                path => "/var/log/logstash/fail-postfix-%{+YYYY.MM.dd}.log"
            }
        } else {
            # для отладки скидываем в файл
            #file {
            #    path => "/var/log/logstash/postfix-%{+YYYY.MM.dd}.log"
            #    flush_interval => 0
            #}
            # скидываем в базу ES
            elasticsearch {
               hosts           => "localhost"
               manage_template => false
               index           => "postfix-%{+YYYY.MM.dd}"
            }

        }
        # end grok parse falure tags
    }
    # end postfix tags
}
# end output