logstash.conf

# filebeat input
input {
    beats {
        port => "5044"
    }
}

filter {
	# squid logs captured into daily index from filebeat
    if [beat][name] =~ "^squid" or [host][name] =~ "^squid" {
		mutate { 
			add_field => {
			"my_index" => "squid-access-%{+YYYY.MM.dd}"
			}
		}

		grok {
				match => {
				"message" => "%{NUMBER:squid_timestamp}\s+%{NUMBER:response_time} %{IPORHOST:src_ip} %{WORD:squid_request_status}/%{NUMBER:http_status_code} %{NUMBER:reply_size_include_header} %{WORD:http_method} %{NOTSPACE:request_url} %{NOTSPACE:user} %{WORD:squid}/%{IP:server_ip} %{NOTSPACE:content_type}"
				}
		}
		if "_grokparsefailure" in [tags] {
			mutate { remove_tag => ["_grokparsefailure"] }
				grok {
				match => {"message" => "%{NUMBER:squid_timestamp}\s+%{NUMBER:response_time} %{IPORHOST:src_ip} %{WORD:squid_request_status}/%{NUMBER:http_status_code} %{NUMBER:reply_size_include_header} %{WORD:http_method} %{NOTSPACE:request_url} %{NOTSPACE:user} %{WORD:squid}%{GREEDYDATA:tail}"}
			}
		}
		
		mutate { convert => { "response_time" => "integer" } }
		mutate { convert => { "reply_size_include_header" => "integer" } }
    }

	# lancache nginx logs captured into daily index fromm filebeat
    if [beat][name] == "lancache" or [host][name] == "lancache" {

    	mutate {
            add_field => {
    	        "my_index" => "lancache-%{+YYYY.MM.dd}"
            }
        }

        # ACCESS LOG
        if [source] =~ "access" or [log][file][path] =~ "access" {
            mutate { add_tag => "lancache-access" }

            # Extract basic access log fields
            grok {
					# This match is for setups with squid that introduce a cache_ip field and client_ip
					named_captures_only => 'true' 
					match => { "message" => "\[%{WORD:cache_tag}\] %{IP:cache_ip} / %{IP:client_ip} - - \[%{HTTPDATE:timestamp}\] \"%{WORD:http_verb} %{URIPATH:url_path}%{URIPARAM:url_querystring}? HTTP/%{NUMBER:http_version}\" %{NUMBER:http_status_code} %{NUMBER:bytes} \"-\" %{QUOTEDSTRING:user_agent} \"(?:%{WORD:cache_status})\" \"%{HOSTNAME:upstream_server}\" \"%{GREEDYDATA:range}\""}
					}
					
					# this match is for the simplier config without squid introducing cache_ip
					if "_grokparsefailure" in [tags] {
					mutate { remove_tag => ["_grokparsefailure"] }
						grok {
						named_captures_only => 'true' 
						match => { "message" => "\[%{WORD:cache_tag}\] %{IP:client_ip} / - - - \[%{HTTPDATE:timestamp}\] \"%{WORD:http_verb} %{URIPATH:url_path}%{URIPARAM:url_querystring}? HTTP/%{NUMBER:http_version}\" %{NUMBER:http_status_code} %{NUMBER:bytes} %{QUOTEDSTRING:secound0_url} %{QUOTEDSTRING:user_agent} \"(?:%{WORD:cache_status})\" \"%{HOSTNAME:upstream_server}\" \"%{GREEDYDATA:range}\""}
						}
					}	
					
					mutate { convert => { "bytes" => "integer" } }
			
        # ERROR LOG
        } else if [source] =~ "error" or [log][file][path] =~ "error" {
            mutate { add_tag => "lancache-error" }

            grok {
                match => { "message" => "\[%{WORD:cache_tag}\] (?<timestamp>\d{4}/\d{2}/\d{2} \d{2}:\d{2}:\d{2}) \[%{DATA:error_severity}\] %{NOTSPACE} %{NOTSPACE} (?<error_message>(.|\r|\n)*)(?:, client: (?<client_ip>%{IP}|%{HOSTNAME}))(?:, server: %{IPORHOST:server_name})(?:, request: \"%{WORD:http_verb} %{URIPATH:url_path}%{URIPARAM:url_querystring}? HTTP/%{NUMBER:http_version}\")?(?:, host: %{QS:nginx_host})?(?:, referrer: \"%{URI:referrer})?" }
            }

        # UNKNOWN LOG
        } else {
            mutate { add_tag => "lancache-unknown" }
        }

        # STEAM
        if [source] =~ "steam" or [log][file][path] =~ "steam" or [cache_tag] == "steam" {
            mutate { replace => { upstream => "steam" } }

            if "lancache-access" in [tags] {
 
                # STEAM ID
                if [url_querystring] =~ "sid=" {
                    grok {
                      match => { "url_querystring" => "sid=%{INT:steam_id_3}" }
                    }
                }

                # DEPOT ID
                if [url_path] =~ "^\/depot" {
                    grok {
                        match => { "url_path" => "/depot/%{INT:steam_depot_id}/" }
                    }
 
                    # CHUNK ID
                    if [url_path] =~ "^\/depot\/\d*\/chunk\/" {
                        grok {
                            match => { "url_path" => "chunk/%{BASE16NUM:steam_chunk_id}" }
                        }
                    }
 
                    # MANIFEST ID
                    if [url_path] =~ "^\/depot\/\d*\/manifest\/" {
                        grok {
                            match => { "url_path" => "manifest/%{BASE16NUM:steam_manifest_id}" }
                        }
                    }
                }
            }
 
        # BLIZZARD
        } else if [source] =~ "blizzard" or [log][file][path] =~ "blizzard" or [cache_tag] == "blizzard" {
            mutate { replace => { upstream => "blizzard" } }

	    if [url_path] =~ "^\/tpr" {
		    grok {
		        match => { "url_path" => "/tpr/%{DATA:blizzard_game}/%{WORD:blizzard_cdn_type}/%{GREEDYDATA:blizzard_data}" }
		    }
	    }
 
        # RIOT
        } else if [source] =~ "riot" or [log][file][path] =~ "riot" or [cache_tag] == "riot" {
            mutate { replace => { upstream => "riot" } }
 
        # MICROSOFT
        } else if [source] =~ "microsoft" or [log][file][path] =~ "microsoft" or [cache_tag] == "microsoft" {
            mutate { replace => { upstream => "microsoft" } }

        # WSUS
        } else if [source] =~ "wsus" or [log][file][path] =~ "wsus" or [cache_tag] == "wsus" {
            mutate { replace => { upstream => "wsus" } }

        # UPLAY
        } else if [source] =~ "uplay" or [log][file][path] =~ "uplay" or [cache_tag] == "uplay" {
            mutate { replace => { upstream => "uplay" } }

			if [url_path] =~ "^\/uplaypc\/downloads\/" {
				grok {
					match => {"url_path" => "/uplaypc/downloads/%{WORD:uplay_game}/%{GREEDYDATA:uplay_data}"}
				}
			}


        # DAYBREAK
        } else if [source] =~ "daybreak" or [log][file][path] =~ "daybreak" or [cache_tag] == "daybreak" {
            mutate { replace => { upstream => "daybreak" } }
 
        # ORIGIN
        } else if [source] =~ "origin" or [log][file][path] =~ "origin" or [cache_tag] == "origin" {
            mutate { replace => { upstream => "origin" } }

			if [url_path] =~ "^\/eamaster\/s\/shift" {
				grok {
					match => {"url_path" => "/eamaster/s/shift/%{DATA:origin_game}/%{DATA:origin_game_extra}/%{GREEDYDATA:orign_data}"}
				}

				if "_grokparsefailure" in [tags] {
					mutate { remove_tag => ["_grokparsefailure"] }
					grok {
						match => {"url_path" => "/eamaster/s/shift/%{DATA:origin_game}/%{GREEDYDATA:orign_data}"}
					}
				}
			}

        # EPIC
        } else if [source] =~ "epicgames" or [log][file][path] =~ "epicgames" or [cache_tag] == "epicgames" {
            mutate { replace => { upstream => "epicgames" } }

 
        # UNKNOWN
        } else {
            mutate { replace => { upstream => "unknown" } }
        }

    }

    date {
        match => [ "timestamp",  "dd/MMM/yyyy:HH:mm:ss Z"]
        remove_field => [ "timestamp" ]
    }
}

# If you're not using the example docker-compse.yml you might need to change these hosts and/or update the config for the user/password
output {
  elasticsearch {
    hosts    => [ 'elasticsearch' ]
    user => 'elastic'
    password => '${ELASTIC_PASSWORD}'
    ssl => false
    index => "%{my_index}"
  }
}