input {
  tcp {
    port => 5000
    codec => json
  }
}

filter {
  if [logger_name] =~ /cvat.client/ {
    # 1. Decode the event from json in 'message' field
    # 2. Remove unnecessary field from it
    # 3. Type it as client
    json {
      source => "message"
    }

    date {
      match => ["timestamp", "UNIX", "UNIX_MS"]
      remove_field => "timestamp"
    }

    if [event] == "Send exception" {
      aggregate {
        task_id => "%{userid}_%{application}_%{message}_%{filename}_%{line}"
        code => "
          require 'time'

          map['userid'] ||= event.get('userid');
          map['application'] ||= event.get('application');
          map['error'] ||= event.get('message');
          map['filename'] ||= event.get('filename');
          map['line'] ||= event.get('line');
          map['task'] ||= event.get('task');

          map['error_count'] ||= 0;
          map['error_count'] += 1;

          map['aggregated_stack'] ||= '';
          map['aggregated_stack'] += event.get('stack') + '\n\n\n';"

        timeout => 3600
        timeout_tags => ['aggregated_exception']
        push_map_as_event_on_timeout => true
      }
    }

    prune {
      blacklist_names => ["level", "host", "logger_name", "message", "path",
        "port", "stack_info"]
    }

    mutate {
      replace => { "type" => "client" }
    }
  } else if [logger_name] =~ /cvat.server/ {
    # 1. Remove 'logger_name' field and create 'task' field
    # 2. Remove unnecessary field from it
    # 3. Type it as server
    if [logger_name] =~ /cvat\.server\.task_[0-9]+/ {
      mutate {
        rename => { "logger_name" => "task" }
        gsub => [ "task", "cvat.server.task_", "" ]
      }

      # Need to split the mutate because otherwise the conversion
      # doesn't work.
      mutate {
        convert => { "task" => "integer" }
      }
    }

    prune {
      blacklist_names => ["host", "port", "stack_info"]
    }

    mutate {
      replace => { "type" => "server" }
    }
  }
}

output {
  stdout {
    codec => rubydebug
  }

  if [type] == "client" {
    elasticsearch {
      hosts => ["elasticsearch:9200"]
      index => "cvat.client"
    }
  } else if [type] == "server" {
    elasticsearch {
      hosts => ["elasticsearch:9200"]
      index => "cvat.server"
    }
  }
}