mongo_logs.cfg

input {
	stdin {
		tags => [ "tech_talk" ]	
	}
}

filter {
	# Initial grok to remove the syslog timestamp
	grok {
		match => [ "message" , "%{SYSLOGBASE} %{GREEDYDATA:message}" ]
		overwrite => [ "message" ]
	}

	# If the log comes from mongo, then fall into this conditional
	if [program] =~ "mongos" {
		# Mutate the timestamp field into syslog_timestamp.
		# We will populate 'timestamp' with the value from the mongo process
		mutate {
			rename => [ "timestamp" , "syslog_timestamp" ]
			add_tag => "mongos"
		}
		# Use the %MONGO_LOG pattern to pull out the timestamp. 
		# Overwrite the message abd timestamp fields
		# Add a new field
		grok {
			match => [ "message" , "%{MONGO_LOG}" ]
			match => [ "message" , "%{SYSLOGTIMESTAMP:timestamp} %{GREEDYDATA:message}" ]
			overwrite => [ "message" , "timestamp" ]
			add_field => { "mongo_timestamp" => "%{timestamp}" }
			add_tag => [ "%{component}" ]
		}

		if [component] =~ "conn" {
			mutate {
				replace => [ "component", "conn" ]
			}
		}
		# Capture the following lines
		#
		# Jan 14 12:50:55 lonapp0-2vz74 mongos.23074[2420]: Wed Jan 14 12:50:55.254 [mongosMain] connection accepted from 10.183.224.66:53539 #4364679 (12 connections now open)
		#
		#

		if [component] == "conn" {
			if [message] =~ "end connection" {
				mutate {
					add_tag => [ "end_connection" ]
				}
			}
			if [message] =~ "authenticate" {
				grok {
					match => [ "message" , "authenticate db: %{WORD:authdb} { %{GREEDYDATA:message} }"]
					overwrite => [ "message" ]
					add_tag => [ "authenticate" ]
				}
				kv {
					value_split => ":"	
					field_split => ","
					trim => " \""
				}
			}
		}

		if [component] == "mongosMain" {
			grok {
				break_on_match => false
				match => ["message" , "connection accepted from %{IP:connect_from}:%{POSINT:connect_from_port} %{DATA} \(%{POSINT:open_conns} connections now open\)"]
			}
			geoip {
				source => "connect_from"
			}
		}

		if [component] == "Balancer" {

		}
		
	}#if

	if [program] =~ "mongod" {
                # Mutate the timestamp field into syslog_timestamp.
                # We will populate 'timestamp' with the value from the mongo process
                mutate {
                        rename => [ "timestamp" , "syslog_timestamp" ]
                        add_tag => "mongod"
                }
		grok {
                        match => [ "message" , "%{MONGO_LOG}" ]
                        match => [ "message" , "%{SYSLOGTIMESTAMP:timestamp} \[%{DATA:component}\] %{GREEDYDATA:message}" ]
                        match => [ "message" , "%{SYSLOGTIMESTAMP:timestamp} %{GREEDYDATA:message}" ]
                        match => [ "message" , "%{GREEDYDATA:message}" ]
                        overwrite => [ "message" , "timestamp" ]
                        add_field => { "mongo_timestamp" => "%{timestamp}" }
			add_tag => [ "%{component}" ]
			tag_on_failure => [ "_grokparsefailure","_gp1"]
                }

		if [message] =~ "dbexit" { 
			mutate { 
				add_tag => ["dbexit"] 
				# There is no timestamp set now, use the syslog one.
				replace => [ "timestamp", "%{syslog_timestamp}" ]
			}
		}		

		if [component] =~ "conn" {
			mutate {
				replace => [ "component", "conn" ]
			}
		}

		if [component] == "initandlisten" {
			grok {
                                match => [ "message" , "connection accepted from %{IP:connect_from}:%{POSINT:connect_from_port} %{DATA} \(%{POSINT:open_conns:int} connections now open\)"]
                        	match => [ "message" , "%{GREEDYDATA:message}" ]
				overwrite => [ "message" , "timestamp" ]
                        }
			if [message] =~ "MongoDB starting" {
				grok {
					match => [ "message" , "MongoDB starting : pid=%{POSINT} port=%{POSINT:port:int} dbpath=%{DATA:dbpath} %{DATA} host=%{DATA:host}" ]
					add_tag => [ "startup" ]
				}
			}
		}

		if [component] == "conn" {
			if [message] =~ "end connection" {
				mutate {
					add_tag => [ "end_connection" ]
				}
			}
			if [message] =~ "authenticate db" {
				grok {
					match => [ "message" , "authenticate db: %{WORD:authdb} { %{GREEDYDATA:message} }"]
					overwrite => [ "message" ]
					add_tag => [ "authenticate" ]
				}
				kv {
					value_split => ":"	
					field_split => ","
					trim => " \""
				}
			}
			if [message] =~ "nreturned" {
				grok {
					match => [ "message", "%{MONGO_SLOWQUERY}"]
					add_tag => [ "query" ]
				}
			}
		}
	}

	date {
		match => [ "timestamp", "MMM  d HH:mm:ss", "MMM  d HH:mm:ss.SSS" ,"MMM dd HH:mm:ss.SSS", "MMM DD HH:mm:ss" ]
	}
}

output {
	#if "_grokparsefailure" in [tags] {
	#	stdout { 
	#		codec => rubydebug 
	#	}
	#}
	elasticsearch_http {
		host => "134.213.133.52"
		replication=> "async"
	}
}