gistfile1.txt

# CLEANUP PREVIOUS DATA
POST _transform/stat_customers_with_keys/_stop?force=true
DELETE _transform/stat_customers_with_keys
DELETE /stat_customers_with_keys

---------

# CREATE INDEX
PUT stat_customers_with_keys
{
    "mappings" : {
        "properties" : {
            "credits" : {"type" : "double"},
            "requests": {"type" : "double"},
            "error_codes.4xx": {"type" : "double"},
            "error_codes.5xx": {"type" : "double"},
            "keys": {
                "type": "nested",
                "properties": {
                    "name": {"type": "keyword"},
                    "requests": {"type": "double"}
                }
            },
            "timestamp": {"type" : "date"},
            "customer": {"type" : "keyword"}
        }
    },
    "settings" : {
        "index" : {
            "number_of_shards" : "1",
            "auto_expand_replicas" : "0-all"
        }
    }
}

# CREATE TRANSFORM
PUT _transform/stat_customers_with_keys
{
    "description": "Credits and keys per customer",
    "dest": {"index": "stat_customers_with_keys"},
    "source": {
        "index": "stats-*"
    },
    "frequency": "1h",
    "sync": {
        "time": {
            "field": "timestamp",
            "delay": "90m"
        }
    },
    "settings": {
        "align_checkpoints": false
    },
    "pivot": {
        "aggs": {
            "credits": { "sum": { "field": "credits" } },
            "requests": { "value_count": { "field": "timestamp" }},
            "error_codes": {
                "scripted_metric": {
                    "init_script": "state.responses = [:]",
                    "map_script": """
                        def key_name = 'missing';
                        if (doc['error.code'].size() != 0) {
                            def error_code = doc['error.code'].value;
                            if (error_code >= 400 && error_code <= 499) {
                                key_name = '4xx';
                            }
                            else if (error_code >= 500 && error_code <= 599) {
                                key_name = '5xx';
                            }
                            else {
                                key_name = error_code;
                            }
                        }

                        if (!state.responses.containsKey(key_name)) {
                            state.responses[key_name] = 0;
                        }
                        state.responses[key_name] += 1;
                        """,
                    "combine_script": "state.responses",
                    "reduce_script": """
                        def counts = [:];
                        for (responses in states) {
                            for (key in responses.keySet()) {
                                if (key == 'missing') {
                                    continue;
                                }
                                if (!counts.containsKey(key)) {
                                    counts[key] = 0;
                                }
                                counts[key] += responses[key];
                            }
                        }
                        return counts;
                    """
                }
            },
            "keys": {
                "scripted_metric": {
                    "init_script": "state.keys = [];",
                    "map_script": "state.keys.add(doc.key.value)",
                    "combine_script": "return state.keys",
                    "reduce_script": """
                        def key_counts = [:];
                        for (keys in states) {
                            for (key in keys) {
                                if (!key_counts.containsKey(key)) {
                                    key_counts[key] = 0;
                                }
                                key_counts[key] += 1;
                            }
                        }
                        def results = [];
                        for (key in key_counts.keySet()) {
                            def count = key_counts[key];
                            // results.add({"name": key, "requests": count})
                            Map result = new HashMap();
                            result.put("name", key);
                            result.put("requests", count);
                            results.add(result);
                        }
                        return results;
                    """
                }
            }
        },
        "group_by": {
            "timestamp": {"date_histogram": {"field": "timestamp", "fixed_interval": "1h"}},
            "customer": {"terms": {"field":"customer"}}
        }
    }
}

# START TRANSFORM
POST _transform/stat_customers_with_keys/_start