filter f_nginx { match("nginx" value("PROGRAM")); };
filter f_nginx_priority_info { level("info"); };

parser p_nginx_message {
  csv-parser(
    dialect(escape-double-char)
    flags(strip-whitespace)
    delimiters(" ")
    quote-pairs('""[]')
    columns("docker", "nginx.client.ip", "nginx.ident", "nginx.auth", "nginx.timestamp", "nginx_request", "nginx.response", "nginx.bytes", "nginx.referrer", "nginx.agent")
  );
};

rewrite r_nginx_docker_header {
  subst(":", " ", value("docker"));
  subst("\/", " ", value("docker"));
  subst('\[', " ", value("docker"));
  subst('\]', " ", value("docker"));
};

parser p_nginx_docker_header {
  csv-parser(
    template("${docker}")
    flags(strip-whitespace)
    delimiters(" ")
    columns("docker.image.name", "docker.container.name", "docker.container.pid")
  );
};

parser p_nginx_request_header {
  csv-parser(
    template("${nginx_request}")
    flags(strip-whitespace)
    delimiters(" ")
    columns("nginx.request_method", "nginx.request_string", "nginx.request_protocol")
  );
};

parser p_nginx_client_ip_geoip2_city {
  geoip2(
    "${nginx.client.ip}",
    prefix( "geoip2.source." )
    database( "/usr/local/share/GeoIP/GeoLite2-City.mmdb" )
  );
};

destination d_nginx_logs {
  elasticsearch-http(
    index("nginx-logs")
    type("")
    user("elastic")
    password("forty6and2")
    url("http://pi501.in.thelinuxpro.net:9200/_bulk")
    template("$(format-json --scope rfc5424 --scope dot-nv-pairs
    --rekey .* --shift 1 --scope nv-pairs
    --exclude DATE @timestamp=${ISODATE})")
    persist-name("d_nginx_logs")
  );
};

log {
  source(s_network_udp);
  filter(f_nginx);
  filter(f_nginx_priority_info);
  parser(p_nginx_message);
  parser(p_nginx_request_header);
  parser(p_nginx_request_header);
  rewrite(r_nginx_docker_header);
  parser(p_nginx_docker_header);
  parser(p_nginx_client_ip_geoip2_city);
  destination(d_nginx_logs);
  flags(final);
};