Initial commit - Fully working architecture

logstash/indexer/pipeline/kafka_elasticsearch.conf

@@ -0,0 +1,94 @@
+input {
+	kafka {
+        codec =>  json{}
+        bootstrap_servers => "kafka:9092"
+        topics => ["nginx-access", "random", "apache"]
+        client_id => "logstash_indexer_1"
+	}
+}
+
+filter {
+	if [type] == "nginx-access" {
+		grok {
+			match => [ "message" , "%{COMBINEDAPACHELOG}+%{GREEDYDATA:extra_fields}"]
+			overwrite => [ "message" ]
+		}
+		mutate {
+			convert => ["response", "integer"]
+			convert => ["bytes", "integer"]
+			convert => ["responsetime", "float"]
+		}
+		geoip {
+			source => "clientip"
+			target => "geoip"
+			add_tag => [ "nginx-geoip" ]
+		}
+		date {
+			match => [ "timestamp" , "dd/MMM/YYYY:HH:mm:ss Z" ]
+			remove_field => [ "timestamp" ]
+		}
+		useragent {
+			source => "agent"
+		}
+	}	
+	if [type] == "random" {
+		grok {
+			match => [ "message" , "(?<timestamp>%{YEAR}[./-]%{MONTHNUM}[./-]%{MONTHDAY}[- ]%{TIME}) %{NUMBER:pid} %{GREEDYDATA:filename} %{NUMBER:line} %{GREEDYDATA:logger} %{LOGLEVEL:severity} %{GREEDYDATA:quote}"]
+			overwrite => [ "message" ]
+		}
+		date {
+			match => [ "timestamp", "YYYY-MM-dd HH:mm:ss,SSS"]
+			remove_field => [ "timestamp" ]
+		}
+	}
+	if [type] == "apache" {
+		grok {
+			match => [ "message" , "%{COMBINEDAPACHELOG}"]
+			overwrite => [ "message" ]
+		}
+		mutate {
+			convert => ["response", "integer"]
+			convert => ["bytes", "integer"]
+			convert => ["responsetime", "float"]
+		}
+		geoip {
+			source => "clientip"
+			target => "geoip"
+			add_tag => [ "apache-geoip" ]
+		}
+		date {
+			match => [ "timestamp", "dd/MMM/yyyy:HH:mm:ss Z" ]
+			remove_field => [ "timestamp" ]
+		}
+	}
+}
+
+output {
+	if [type] == "nginx-access" {
+        elasticsearch {
+			hosts => ["elasticsearch:9200"]
+			index => "nginx-%{+YYYYMM}"
+        }
+        stdout {
+          codec => rubydebug
+        }
+	}
+	if [type] == "random" {
+        elasticsearch {
+			hosts => ["elasticsearch:9200"]
+			index => "random-%{+YYYYMM}"
+        }
+        stdout {
+          codec => rubydebug
+        }
+	}
+	if [type] == "apache" {
+        elasticsearch {
+			hosts => ["elasticsearch:9200"]
+			index => "apache-%{+YYYYMM}"
+        }
+        stdout {
+          codec => rubydebug
+        }
+	}
+}

logstash/logstash.yml

@@ -0,0 +1,2 @@
+http.host: "0.0.0.0"
+xpack.monitoring.enabled: false

logstash/shipper/pipeline/beat_kafka.conf

@@ -0,0 +1,15 @@
+input {
+  beats {
+    port => 5044
+  }
+}
+output {
+  kafka {
+    codec => json
+    bootstrap_servers => "kafka:9092"
+	topic_id => "%{type}"
+  }
+  stdout {
+    codec => rubydebug
+  }
+}