# This config file is used to generate Monasca metrics from logs ingested
# by the Monasca Log API. Alarms and notifications can then be set to alert
# users when a particular log message is ingested. This example config file
# generates metrics for all logs which have a log level which isn't
# debug, trace or info level. A user may want to apply additional logic to
# generate special metrics when particular log messages are ingested. For
# example, if HAProxy fails over, file system corruption is detected or
# other scenarios of interest.

input {
    kafka {
        bootstrap_servers => "{{ monasca_kafka_servers }}"
        topics => ["{{ monasca_transformed_logs_topic }}"]
        group_id => "log_metrics"
        consumer_threads => "{{ monasca_log_pipeline_threads }}"
        codec => json
    }
}

filter {
    # Drop everything we don't want to create metrics for.
    if ![log][dimensions][log_level] or [log][dimensions][log_level] in [ "debug", "trace", "info" ] {
        drop {
        }
    }

    # Generate a metric name based on the program and log level
    mutate {
        add_field => { "[metric][name]" => "log.%{[log][dimensions][programname]}.%{[log][dimensions][log_level]}" }
    }

    # Form the metric structure.
    mutate {
        add_field => { "[metric][value]" => 1 }
        rename => { "[log][dimensions]" => "[metric][dimensions]" }
        rename => { "[metric][dimensions][Hostname]" => "[metric][dimensions][hostname]" }
        rename => { "[metric][dimensions][programname]" => "[metric][dimensions][service]" }
    }
    mutate {
         convert => { "[metric][value]" => "float" }
    }

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

    # Clean up any fields which aren't required from the new metric to save space
    mutate {
        remove_field => ["[metric][dimensions][log_level]",
                         "[metric][dimensions][domain_id]",
                         "[metric][dimensions][user_id]",
                         "[metric][dimensions][tenant_id]",
                         "[metric][dimensions][project_domain]",
                         "[metric][dimensions][tag]",
                         "[metric][dimensions][Logger]",
                         "[metric][dimensions][Pid]",
                         "[metric][dimensions][user_domain]",
                         "[metric][dimensions][request_id]",
                         "[metric][dimensions][python_module]",
                         "[metric][meta]",
                         "creation_time",
                         "log",
                         "@version",
                         "@timestamp"]
    }
}

output {
    kafka {
        codec => json
        bootstrap_servers => "{{ monasca_kafka_servers }}"
        topic_id => "{{ monasca_metrics_topic }}"
    }
}