input {
	http {
		port => "9111"
		codec => json_lines
		additional_codecs => { "application/json" => "json_lines" }
		threads => "1000"
		ssl => true
		keystore => "/etc/logstash/logstash-ssl.jks"
		keystore_password => "changeit"

		user => "logstashuser"
		password => ""

	}
	http {
		port => "9112"
		codec => json_lines
		additional_codecs => { "application/json" => "json_lines" }
		threads => "1000"
		type => redacted
		ssl => true
		keystore => "/etc/logstash/logstash-ssl.jks"
		keystore_password => "changeit"

	}
}

filter {
	# elasticsearch bulk format alternates 'index' or 'create' lines with records
    # there could also be 'delete' and 'update', but logqueue doesn't send those
    # we might want to consider throwing error just in case
    if [create] or [index] or [delete] or [update] {
        drop{}
    }

    # Push the time stamp we will use for our S3 filenames (or test file names) into metadata
    ruby {
		code => "event.set('[@metadata][TimeInFileName]', event.sprintf('%{+YYYY-MM-dd\'T\'HH-mm}'))"
	}

	if "redacted" in [type] {
		metrics {
			meter => [ "_events" ]
			add_tag => [ "metric", "_events" ]
		}
	} else {
		# Track logstash performance
		metrics {
			meter => [ "es_events" ]
			add_tag => [ "metric", "es_events" ]
		}
		if ![LogIndex] {
			mutate {
				add_field => { "LogIndex" => "health" }
			}
		}

		if ![LogType] {
			mutate {
				add_field => { "LogType" => "healthcheck" }
			}
		}

		if [LogIndex] == "security" {
			metrics {
				meter => [ "security_events" ]
				add_tag => [ "metric", "security_events" ]
			}
			mutate {
				add_field => { "Tags" => "filter_processed_time:%{[@metadata]TimeInFileName}" }
			}
		}
	}
	mutate {
		remove_field => [ "@timestamp", "@version", host, headers]
	}

}

output {
	if "metric" in [tags] {
		if "es_events" in [tags] {
			stdout {
				codec => line {
					format => "%{[es_events][count]} elastic logs processed. Current rate (1m) : %{[es_events][rate_1m]}"
				}
			}
		}
		if "security_events" in [tags] {
			stdout {
				codec => line {
					format => "%{[security_events][count]} security logs processed. Current rate (1m) : %{[security_events][rate_1m]}"
				}
			}
		}
		if "_events" in [tags] {
			stdout {
				codec => line {
					format => "%{[_events][count]} logs processed. Current rate (1m) : %{[_events][rate_1m]}"
				}
			}
		}
	} else if "redacted" in [type] {
#       file {
#    	   path => "conf\%{[@metadata]TimeInFileName}.log"
#       }
		s3 {

				bucket => "security-logs"
				size_file => "2048000"
				time_file => "5"
				prefix => "DI/"
				codec => "json_lines"
				tags => "redacted"
				canned_acl => "bucket_owner_full_control"
		}
	} else {
		if [LogIndex] == "security" {
#           file {
#       	    path => "conf\security_%{[@metadata]TimeInFileName}.log"
#           }
			s3 {

				bucket => "security-logs"
				size_file => "2048000"
				time_file => "5"
				prefix => "DI/"
				codec => "json_lines"
				tags => "security"
				canned_acl => "bucket_owner_full_control"
			}
		}

		elasticsearch {
				action => "index"
				hosts => ["lb.aws..net:9200"]
				ssl => "true"
				# can either use truststore and truststore_password or just cacert
				#truststrore => "${ES_JKS_TRUST_STORE}"
				#truststrore_password => "${ES_JKS_TRUST_STORE_PASSWORD}"
				cacert => "/etc/ssl/certs/ca-bundle.crt"
				index => "%{LogIndex}"
				document_type => "%{LogType}"
				manage_template => "false"
				user => ""
				password => ""
				#workers => 4
		}
#		stdout { codec  => rubydebug{metadata => true}}
	}
}