# Provide stream processing for Monasca log metrics.

input {
    kafka {
        zk_connect => "{% raw %}{{ monasca_zookeeper_servers }}{% endraw %}"
        topic_id => "{% raw %}{{ monasca_transformed_logs_topic }}{% endraw %}"
        group_id => "log-metric"
        consumer_id => "monasca_log_metrics"
    }
}

filter {
    # Drop everything without a log_level
    if ![log][dimensions][log_level] {
        drop {
        }
    }

    # Override log level for specific log metrics
    if 'infra.haproxy.http' in [log][dimensions][tag] {
        mutate {
            replace => { "[log][dimensions][log_level]" => "haproxy_tr" }
        }
    }

    # Convert logs to metrics
    if [log][dimensions][log_level] in [ "error", "warning" ] {
        mutate {
            add_field => { "[metric][name]" => "log.%{[log][dimensions][log_level]}" }
            add_field => { "[metric][value]" => 1 }
            add_field => { "[metric][dimensions][programname]" => "%{[log][dimensions][programname]}" }
            add_field => { "[metric][dimensions][hostname]" => "%{[log][dimensions][Hostname]}" }
        }
    } else if "haproxy_tr" in [log][dimensions][log_level] {
        mutate {
            add_field => { "[metric][name]" => "haproxy_logs_time_to_request" }
            add_field => { "[metric][value]" => "%{[log][dimensions][tr]}" }
            add_field => { "[metric][dimensions][frontend_name]" => "%{[log][dimensions][frontend_name]}" }
            add_field => { "[metric][dimensions][status_code]" => "%{[log][dimensions][status_code]}" }
            add_field => { "[metric][dimensions][http_req]" => "%{[log][dimensions][http_req]}" }
        }
    } else {
        drop {
        }
    }

    # Common operations to new metrics
    mutate {
        add_field => { "[metric][dimensions][service]" => "logmetrics" }
        convert => { "[metric][value]" => "float" }
    }

    # Convert the timestamp of the event to milliseconds since epoch.
    ruby {
        code => "event['metric']['timestamp'] = event['@timestamp'].to_i * 1000"
    }

    # Clean up any fields which aren't required from the new metric
    mutate {
        remove_field => ["[metric][meta]",
                         "creation_time",
                         "log",
                         "@version",
                         "@timestamp"]
    }
}

output {
  kafka {
    bootstrap_servers => "{% raw %}{{ monasca_kafka_servers }}{% endraw %}"
    topic_id => "{% raw %}{{ monasca_metrics_topic }}{% endraw %}"
    client_id => "monasca_log_metrics"
  }
}