input {
    beats {
        port => 5044
        host => "0.0.0.0"
        ecs_compatibility => "v1"
        ssl => true
        ssl_certificate_authorities => "/usr/share/logstash/config/certificates/ca/ca.crt"
        ssl_key => "/usr/share/logstash/config/certificates/{{ logstash_certificate }}/{{ logstash_certificate }}.pkcs8.key"
        ssl_certificate => "/usr/share/logstash/config/certificates/{{ logstash_certificate }}/{{ logstash_certificate }}.crt"
    }
}

filter {
    if [message] =~ /^{.*}$/ {
        json {
            source => "message"
            skip_on_invalid_json => true
            remove_field => [ "[event][original]" ]
        }
        if [stack_trace] {
            ruby {
                code => "event.set('message_full', event.get('message') + ':' + 10.chr + event.get('stack_trace'))"
            }
        } else {
            ruby {
                code => "event.set('message_full', event.get('message'))"
            }
        }
    }
    else if [event][dataset] == "system.auth" {
        grok {
            match => { "message" => ["%{SYSLOGTIMESTAMP:[system][auth][timestamp]} %{SYSLOGHOST:[system][auth][hostname]} sshd(?:\[%{POSINT:[system][auth][pid]}\])?: %{DATA:[system][auth][ssh][event]} %{DATA:[system][auth][ssh][method]} for (invalid user )?%{DATA:[system][auth][user]} from %{IPORHOST:[system][auth][ssh][ip]} port %{NUMBER:[system][auth][ssh][port]} ssh2(: %{GREEDYDATA:[system][auth][ssh][signature]})?",
                     "%{SYSLOGTIMESTAMP:[system][auth][timestamp]} %{SYSLOGHOST:[system][auth][hostname]} sshd(?:\[%{POSINT:[system][auth][pid]}\])?: %{DATA:[system][auth][ssh][event]} user %{DATA:[system][auth][user]} from %{IPORHOST:[system][auth][ssh][ip]}",
                     "%{SYSLOGTIMESTAMP:[system][auth][timestamp]} %{SYSLOGHOST:[system][auth][hostname]} sshd(?:\[%{POSINT:[system][auth][pid]}\])?: Did not receive identification string from %{IPORHOST:[system][auth][ssh][dropped_ip]}",
                     "%{SYSLOGTIMESTAMP:[system][auth][timestamp]} %{SYSLOGHOST:[system][auth][hostname]} sudo(?:\[%{POSINT:[system][auth][pid]}\])?: \s*%{DATA:[system][auth][user]} :( %{DATA:[system][auth][sudo][error]} ;)? TTY=%{DATA:[system][auth][sudo][tty]} ; PWD=%{DATA:[system][auth][sudo][pwd]} ; USER=%{DATA:[system][auth][sudo][user]} ; COMMAND=%{GREEDYDATA:[system][auth][sudo][command]}",
                     "%{SYSLOGTIMESTAMP:[system][auth][timestamp]} %{SYSLOGHOST:[system][auth][hostname]} groupadd(?:\[%{POSINT:[system][auth][pid]}\])?: new group: name=%{DATA:system.auth.groupadd.name}, GID=%{NUMBER:system.auth.groupadd.gid}",
                     "%{SYSLOGTIMESTAMP:[system][auth][timestamp]} %{SYSLOGHOST:[system][auth][hostname]} useradd(?:\[%{POSINT:[system][auth][pid]}\])?: new user: name=%{DATA:[system][auth][user][add][name]}, UID=%{NUMBER:[system][auth][user][add][uid]}, GID=%{NUMBER:[system][auth][user][add][gid]}, home=%{DATA:[system][auth][user][add][home]}, shell=%{DATA:[system][auth][user][add][shell]}$",
                     "%{SYSLOGTIMESTAMP:[system][auth][timestamp]} %{SYSLOGHOST:[system][auth][hostname]} %{DATA:[system][auth][program]}(?:\[%{POSINT:[system][auth][pid]}\])?: %{GREEDYMULTILINE:[system][auth][message]}"] }
            pattern_definitions => {
                "GREEDYMULTILINE"=> "(.|\n)*"
            }
            remove_field => [ "message", "[event][original]" ]
        }
    }
    else if [event][dataset] == "postgresql.log" {
        grok {
            match => {
                     "message" => "%{DATESTAMP:timestamp} %{TZ} (\[%{DATA:group_id}\]) (\[?%{DATA:user}\]?@\[?%{DATA:database}\]? )?%{DATA:level}:  %{GREEDYDATA:message_full}" }
            remove_field => [ "message", "[event][original]" ]
        }
    }
    mutate {
      remove_field => [ "[agent]", "[container][id]", "[docker]", "[ecs]", "[host]", "[log]", "[stream]" ]
    }
}

output {
    if [fields][hostname] and [event][dataset] == "system.auth" {
        elasticsearch {
            hosts => ["https://{{ shared_service_elastic_stack_01_hostname }}:{{ service_port_elasticsearch }}"]
            cacert => "/usr/share/logstash/config/certificates/ca/ca.crt"
            user => "{{ elastic_admin_username }}"
            password => "{{ elastic_admin_password }}"

            index => "%{[fields][hostname]}-authlog-%{+YYYY.MM}"

            manage_template => false
       }
    }
    else if [event][dataset] == "system.auth" {
        elasticsearch {
            hosts => ["https://{{ shared_service_elastic_stack_01_hostname }}:{{ service_port_elasticsearch }}"]
            cacert => "/usr/share/logstash/config/certificates/ca/ca.crt"
            user => "{{ elastic_admin_username }}"
            password => "{{ elastic_admin_password }}"

            index => "uncategorized-authlog-%{+YYYY.MM}"

            manage_template => false
       }
    }
    else if [fields][hostname] and [event][dataset] == "system.syslog" {
        elasticsearch {
            hosts => ["https://{{ shared_service_elastic_stack_01_hostname }}:{{ service_port_elasticsearch }}"]
            cacert => "/usr/share/logstash/config/certificates/ca/ca.crt"
            user => "{{ elastic_admin_username }}"
            password => "{{ elastic_admin_password }}"

            index => "%{[fields][hostname]}-syslog-%{+YYYY.MM}"

            manage_template => false
       }
    }
    else if [event][dataset] == "system.syslog" {
        elasticsearch {
            hosts => ["https://{{ shared_service_elastic_stack_01_hostname }}:{{ service_port_elasticsearch }}"]
            cacert => "/usr/share/logstash/config/certificates/ca/ca.crt"
            user => "{{ elastic_admin_username }}"
            password => "{{ elastic_admin_password }}"

            index => "uncategorized-syslog-%{+YYYY.MM}"

            manage_template => false
       }
    }
    else if [fields][hostname] and [event][dataset] == "postgresql.log" {
        elasticsearch {
            hosts => ["https://{{ shared_service_elastic_stack_01_hostname }}:{{ service_port_elasticsearch }}"]
            cacert => "/usr/share/logstash/config/certificates/ca/ca.crt"
            user => "{{ elastic_admin_username }}"
            password => "{{ elastic_admin_password }}"

            index => "%{[fields][hostname]}-postgresql-%{+YYYY.MM}"

            manage_template => false
       }
    }
    else if [fields][hostname] and (([event][dataset] == "mysql.error") or ([event][dataset] == "mysql.slowlog")){
        elasticsearch {
            hosts => ["https://{{ shared_service_elastic_stack_01_hostname }}:{{ service_port_elasticsearch }}"]
            cacert => "/usr/share/logstash/config/certificates/ca/ca.crt"
            user => "{{ elastic_admin_username }}"
            password => "{{ elastic_admin_password }}"

            index => "%{[fields][hostname]}-mysql-%{+YYYY.MM}"

            manage_template => false
       }
    }
    else if [fields][harbor] {
        elasticsearch {
            hosts => ["https://{{ shared_service_elastic_stack_01_hostname }}:{{ service_port_elasticsearch }}"]
            cacert => "/usr/share/logstash/config/certificates/ca/ca.crt"
            user => "{{ elastic_admin_username }}"
            password => "{{ elastic_admin_password }}"

            index => "%{[fields][hostname]}-harbor-%{[fields][harbor-component]}-%{+YYYY.MM}"

            manage_template => false
       }
    }
    else if [kubernetes][namespace] {
  	    if [kubernetes][deployment][name] {
            elasticsearch {
                hosts => ["https://{{ shared_service_elastic_stack_01_hostname }}:{{ service_port_elasticsearch }}"]
                cacert => "/usr/share/logstash/config/certificates/ca/ca.crt"
                user => "{{ elastic_admin_username }}"
                password => "{{ elastic_admin_password }}"

                index => "%{[stage]}-%{[kubernetes][namespace]}-%{[kubernetes][deployment][name]}-%{+YYYY.MM}"

                manage_template => false
            }
        }
        else if [kubernetes][daemonset][name] {
            elasticsearch {
                hosts => ["https://{{ shared_service_elastic_stack_01_hostname }}:{{ service_port_elasticsearch }}"]
                cacert => "/usr/share/logstash/config/certificates/ca/ca.crt"
                user => "{{ elastic_admin_username }}"
                password => "{{ elastic_admin_password }}"

                index => "%{[stage]}-%{[kubernetes][namespace]}-%{[kubernetes][daemonset][name]}-%{+YYYY.MM}"

                manage_template => false
           }
        }
        else if [kubernetes][statefulset][name] {
            elasticsearch {
                hosts => ["https://{{ shared_service_elastic_stack_01_hostname }}:{{ service_port_elasticsearch }}"]
                cacert => "/usr/share/logstash/config/certificates/ca/ca.crt"
                user => "{{ elastic_admin_username }}"
                password => "{{ elastic_admin_password }}"

                index => "%{[stage]}-%{[kubernetes][namespace]}-%{[kubernetes][statefulset][name]}-%{+YYYY.MM}"

                manage_template => false
           }
        }
        else if [kubernetes][container][name] {
            elasticsearch {
                hosts => ["https://{{ shared_service_elastic_stack_01_hostname }}:{{ service_port_elasticsearch }}"]
                cacert => "/usr/share/logstash/config/certificates/ca/ca.crt"
                user => "{{ elastic_admin_username }}"
                password => "{{ elastic_admin_password }}"

                index => "%{[stage]}-%{[kubernetes][namespace]}-%{[kubernetes][container][name]}-%{+YYYY.MM}"

                manage_template => false
           }
        }
        else {
            elasticsearch {
                hosts => ["https://{{ shared_service_elastic_stack_01_hostname }}:{{ service_port_elasticsearch }}"]
                cacert => "/usr/share/logstash/config/certificates/ca/ca.crt"
                user => "{{ elastic_admin_username }}"
                password => "{{ elastic_admin_password }}"

                index => "uncategorized-kubernetes-%{[kubernetes][namespace]}-%{+YYYY.MM}"

                manage_template => false
           }
        }
    }
    else if [fields][pass_tenant_id] and [container][name] and [@metadata][beat] {
        elasticsearch {
            hosts => ["https://{{ shared_service_elastic_stack_01_hostname }}:{{ service_port_elasticsearch }}"]
            cacert => "/usr/share/logstash/config/certificates/ca/ca.crt"
            user => "{{ elastic_admin_username }}"
            password => "{{ elastic_admin_password }}"

            index => "%{[fields][stage]}-%{[fields][pass_tenant_id]}-%{[container][name]}-%{+YYYY.MM}"

            manage_template => false
       }
    }
    else if [container][name] and [@metadata][beat] {
        elasticsearch {
            hosts => ["https://{{ shared_service_elastic_stack_01_hostname }}:{{ service_port_elasticsearch }}"]
            cacert => "/usr/share/logstash/config/certificates/ca/ca.crt"
            user => "{{ elastic_admin_username }}"
            password => "{{ elastic_admin_password }}"

            index => "%{[container][name]}-%{+YYYY.MM}"

            manage_template => false
       }
    }
    else if [@metadata][beat] {
        elasticsearch {
            hosts => ["https://{{ shared_service_elastic_stack_01_hostname }}:{{ service_port_elasticsearch }}"]
            cacert => "/usr/share/logstash/config/certificates/ca/ca.crt"
            user => "{{ elastic_admin_username }}"
            password => "{{ elastic_admin_password }}"

            index => "uncategorized-%{[@metadata][beat]}-%{[@metadata][version]}-%{+YYYY.MM}"

            manage_template => false
       }
    }
    else {
        elasticsearch {
            hosts => ["https://{{ shared_service_elastic_stack_01_hostname }}:{{ service_port_elasticsearch }}"]
            cacert => "/usr/share/logstash/config/certificates/ca/ca.crt"
            user => "{{ elastic_admin_username }}"
            password => "{{ elastic_admin_password }}"

            index => "uncategorized-%{+YYYY.MM}"

            manage_template => false
        }
    }
}