# -*- coding: utf-8 -*-

# Copyright 2010 - 2017 RhodeCode GmbH and the AppEnlight project authors
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
#   http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.

import argparse
import datetime
import logging
import copy

import sqlalchemy as sa
import elasticsearch.exceptions
import elasticsearch.helpers

from collections import defaultdict
from pyramid.paster import setup_logging
from pyramid.paster import bootstrap
from appenlight.models import DBSession, Datastores, metadata
from appenlight.lib import get_callable
from appenlight.models.report_group import ReportGroup
from appenlight.models.report import Report
from appenlight.models.report_stat import ReportStat
from appenlight.models.log import Log
from appenlight.models.slow_call import SlowCall
from appenlight.models.metric import Metric

log = logging.getLogger(__name__)

tables = {
    "slow_calls_p_": [],
    "reports_stats_p_": [],
    "reports_p_": [],
    "reports_groups_p_": [],
    "logs_p_": [],
    "metrics_p_": [],
}


def detect_tables(table_prefix):
    found_tables = []
    db_tables_query = """
    SELECT tablename FROM pg_tables WHERE tablename NOT LIKE 'pg_%' AND
    tablename NOT LIKE 'sql_%' ORDER BY tablename ASC;"""

    for table in DBSession.execute(db_tables_query).fetchall():
        tablename = table.tablename
        if tablename.startswith(table_prefix):
            t = sa.Table(
                tablename, metadata, autoload=True, autoload_with=DBSession.bind.engine
            )
            found_tables.append(t)
    return found_tables


def main():
    """
    Recreates Elasticsearch indexes
    Performs reindex of whole db to Elasticsearch

    """

    # need parser twice because we first need to load ini file
    # bootstrap pyramid and then load plugins
    pre_parser = argparse.ArgumentParser(
        description="Reindex AppEnlight data", add_help=False
    )
    pre_parser.add_argument(
        "-c", "--config", required=True, help="Configuration ini file of application"
    )
    pre_parser.add_argument("-h", "--help", help="Show help", nargs="?")
    pre_parser.add_argument(
        "-t", "--types", nargs="+", help="Which parts of database should get reindexed"
    )
    args = pre_parser.parse_args()

    config_uri = args.config
    setup_logging(config_uri)
    log.setLevel(logging.INFO)
    env = bootstrap(config_uri)
    parser = argparse.ArgumentParser(description="Reindex AppEnlight data")
    choices = {
        "reports": "appenlight.scripts.reindex_elasticsearch:reindex_reports",
        "logs": "appenlight.scripts.reindex_elasticsearch:reindex_logs",
        "metrics": "appenlight.scripts.reindex_elasticsearch:reindex_metrics",
        "slow_calls": "appenlight.scripts.reindex_elasticsearch:reindex_slow_calls",
        "template": "appenlight.scripts.reindex_elasticsearch:update_template",
    }
    for k, v in env["registry"].appenlight_plugins.items():
        if v.get("fulltext_indexer"):
            choices[k] = v["fulltext_indexer"]
    parser.add_argument(
        "-t",
        "--types",
        nargs="*",
        choices=["all"] + list(choices.keys()),
        default=[],
        help="Which parts of database should get reindexed",
    )
    parser.add_argument(
        "-c", "--config", required=True, help="Configuration ini file of application"
    )
    args = parser.parse_args()

    if "all" in args.types:
        args.types = list(choices.keys())

    print("Selected types to reindex: {}".format(args.types))

    log.info("settings {}".format(args.types))

    if "template" in args.types:
        get_callable(choices["template"])()
        args.types.remove("template")
    for selected in args.types:
        get_callable(choices[selected])()


def update_template():
    try:
        Datastores.es.indices.delete_template("rcae_reports")
    except elasticsearch.exceptions.NotFoundError as e:
        log.error(e)

    try:
        Datastores.es.indices.delete_template("rcae_logs")
    except elasticsearch.exceptions.NotFoundError as e:
        log.error(e)
    try:
        Datastores.es.indices.delete_template("rcae_slow_calls")
    except elasticsearch.exceptions.NotFoundError as e:
        log.error(e)
    try:
        Datastores.es.indices.delete_template("rcae_metrics")
    except elasticsearch.exceptions.NotFoundError as e:
        log.error(e)
    log.info("updating elasticsearch template")
    tag_templates = [
        {
            "values": {
                "path_match": "tags.*",
                "mapping": {
                    "type": "object",
                    "properties": {
                        "values": {
                            "type": "text",
                            "analyzer": "tag_value",
                            "fields": {
                                "keyword": {"type": "keyword", "ignore_above": 256}
                            },
                        },
                        "numeric_values": {"type": "float"},
                    },
                },
            }
        }
    ]

    shared_analysis = {
        "analyzer": {
            "url_path": {
                "type": "custom",
                "char_filter": [],
                "tokenizer": "path_hierarchy",
                "filter": [],
            },
            "tag_value": {
                "type": "custom",
                "char_filter": [],
                "tokenizer": "keyword",
                "filter": ["lowercase"],
            },
        }
    }

    shared_log_mapping = {
        "_all": {"enabled": False},
        "dynamic_templates": tag_templates,
        "properties": {
            "pg_id": {"type": "keyword", "index": True},
            "delete_hash": {"type": "keyword", "index": True},
            "resource_id": {"type": "integer"},
            "timestamp": {"type": "date"},
            "permanent": {"type": "boolean"},
            "request_id": {"type": "keyword", "index": True},
            "log_level": {"type": "text", "analyzer": "simple"},
            "message": {"type": "text", "analyzer": "simple"},
            "namespace": {
                "type": "text",
                "fields": {"keyword": {"type": "keyword", "ignore_above": 256}},
            },
            "tags": {"type": "object"},
            "tag_list": {
                "type": "text",
                "analyzer": "tag_value",
                "fields": {"keyword": {"type": "keyword", "ignore_above": 256}},
            },
        },
    }

    report_schema = {
        "template": "rcae_r_*",
        "settings": {
            "index": {
                "refresh_interval": "5s",
                "translog": {"sync_interval": "5s", "durability": "async"},
            },
            "number_of_shards": 5,
            "analysis": shared_analysis,
        },
        "mappings": {
            "report": {
                "_all": {"enabled": False},
                "dynamic_templates": tag_templates,
                "properties": {
                    "type": {"type": "keyword", "index": True},
                    # report group
                    "group_id": {"type": "keyword", "index": True},
                    "resource_id": {"type": "integer"},
                    "priority": {"type": "integer"},
                    "error": {"type": "text", "analyzer": "simple"},
                    "read": {"type": "boolean"},
                    "occurences": {"type": "integer"},
                    "fixed": {"type": "boolean"},
                    "first_timestamp": {"type": "date"},
                    "last_timestamp": {"type": "date"},
                    "average_duration": {"type": "float"},
                    "summed_duration": {"type": "float"},
                    "public": {"type": "boolean"},
                    # report
                    "report_id": {"type": "keyword", "index": True},
                    "http_status": {"type": "integer"},
                    "ip": {"type": "keyword", "index": True},
                    "url_domain": {"type": "text", "analyzer": "simple"},
                    "url_path": {"type": "text", "analyzer": "url_path"},
                    "report_type": {"type": "integer"},
                    "start_time": {"type": "date"},
                    "request_id": {"type": "keyword", "index": True},
                    "end_time": {"type": "date"},
                    "duration": {"type": "float"},
                    "tags": {"type": "object"},
                    "tag_list": {
                        "type": "text",
                        "analyzer": "tag_value",
                        "fields": {"keyword": {"type": "keyword", "ignore_above": 256}},
                    },
                    "extra": {"type": "object"},
                    # report stats
                    "report_stat_id": {"type": "keyword", "index": True},
                    "timestamp": {"type": "date"},
                    "permanent": {"type": "boolean"},
                    "log_level": {"type": "text", "analyzer": "simple"},
                    "message": {"type": "text", "analyzer": "simple"},
                    "namespace": {
                        "type": "text",
                        "fields": {"keyword": {"type": "keyword", "ignore_above": 256}},
                    },
                    "join_field": {
                        "type": "join",
                        "relations": {"report_group": ["report", "report_stat"]},
                    },
                },
            }
        },
    }

    Datastores.es.indices.put_template("rcae_reports", body=report_schema)

    logs_mapping = copy.deepcopy(shared_log_mapping)
    logs_mapping["properties"]["log_id"] = logs_mapping["properties"]["pg_id"]
    del logs_mapping["properties"]["pg_id"]

    log_template = {
        "template": "rcae_l_*",
        "settings": {
            "index": {
                "refresh_interval": "5s",
                "translog": {"sync_interval": "5s", "durability": "async"},
            },
            "number_of_shards": 5,
            "analysis": shared_analysis,
        },
        "mappings": {"log": logs_mapping},
    }

    Datastores.es.indices.put_template("rcae_logs", body=log_template)

    slow_call_mapping = copy.deepcopy(shared_log_mapping)
    slow_call_mapping["properties"]["slow_call_id"] = slow_call_mapping["properties"][
        "pg_id"
    ]
    del slow_call_mapping["properties"]["pg_id"]

    slow_call_template = {
        "template": "rcae_sc_*",
        "settings": {
            "index": {
                "refresh_interval": "5s",
                "translog": {"sync_interval": "5s", "durability": "async"},
            },
            "number_of_shards": 5,
            "analysis": shared_analysis,
        },
        "mappings": {"log": slow_call_mapping},
    }

    Datastores.es.indices.put_template("rcae_slow_calls", body=slow_call_template)

    metric_mapping = copy.deepcopy(shared_log_mapping)
    metric_mapping["properties"]["metric_id"] = metric_mapping["properties"]["pg_id"]
    del metric_mapping["properties"]["pg_id"]

    metrics_template = {
        "template": "rcae_m_*",
        "settings": {
            "index": {
                "refresh_interval": "5s",
                "translog": {"sync_interval": "5s", "durability": "async"},
            },
            "number_of_shards": 5,
            "analysis": shared_analysis,
        },
        "mappings": {"log": metric_mapping},
    }

    Datastores.es.indices.put_template("rcae_metrics", body=metrics_template)

    uptime_metric_mapping = copy.deepcopy(shared_log_mapping)
    uptime_metric_mapping["properties"]["uptime_id"] = uptime_metric_mapping[
        "properties"
    ]["pg_id"]
    del uptime_metric_mapping["properties"]["pg_id"]

    uptime_metrics_template = {
        "template": "rcae_uptime_ce_*",
        "settings": {
            "index": {
                "refresh_interval": "5s",
                "translog": {"sync_interval": "5s", "durability": "async"},
            },
            "number_of_shards": 5,
            "analysis": shared_analysis,
        },
        "mappings": {"log": shared_log_mapping},
    }

    Datastores.es.indices.put_template(
        "rcae_uptime_metrics", body=uptime_metrics_template
    )


def reindex_reports():
    reports_groups_tables = detect_tables("reports_groups_p_")
    try:
        Datastores.es.indices.delete("`rcae_r_*")
    except elasticsearch.exceptions.NotFoundError as e:
        log.error(e)

    log.info("reindexing report groups")
    i = 0
    task_start = datetime.datetime.now()
    for partition_table in reports_groups_tables:
        conn = DBSession.connection().execution_options(stream_results=True)
        result = conn.execute(partition_table.select())
        while True:
            chunk = result.fetchmany(2000)
            if not chunk:
                break
            es_docs = defaultdict(list)
            for row in chunk:
                i += 1
                item = ReportGroup(**dict(list(row.items())))
                d_range = item.partition_id
                es_docs[d_range].append(item.es_doc())
            if es_docs:
                name = partition_table.name
                log.info("round {}, {}".format(i, name))
                for k, v in es_docs.items():
                    to_update = {"_index": k, "_type": "report"}
                    [i.update(to_update) for i in v]
                    elasticsearch.helpers.bulk(Datastores.es, v)

    log.info("total docs {} {}".format(i, datetime.datetime.now() - task_start))

    i = 0
    log.info("reindexing reports")
    task_start = datetime.datetime.now()
    reports_tables = detect_tables("reports_p_")
    for partition_table in reports_tables:
        conn = DBSession.connection().execution_options(stream_results=True)
        result = conn.execute(partition_table.select())
        while True:
            chunk = result.fetchmany(2000)
            if not chunk:
                break
            es_docs = defaultdict(list)
            for row in chunk:
                i += 1
                item = Report(**dict(list(row.items())))
                d_range = item.partition_id
                es_docs[d_range].append(item.es_doc())
            if es_docs:
                name = partition_table.name
                log.info("round {}, {}".format(i, name))
                for k, v in es_docs.items():
                    to_update = {"_index": k, "_type": "report"}
                    [i.update(to_update) for i in v]
                    elasticsearch.helpers.bulk(Datastores.es, v)

    log.info("total docs {} {}".format(i, datetime.datetime.now() - task_start))

    log.info("reindexing reports stats")
    i = 0
    task_start = datetime.datetime.now()
    reports_stats_tables = detect_tables("reports_stats_p_")
    for partition_table in reports_stats_tables:
        conn = DBSession.connection().execution_options(stream_results=True)
        result = conn.execute(partition_table.select())
        while True:
            chunk = result.fetchmany(2000)
            if not chunk:
                break
            es_docs = defaultdict(list)
            for row in chunk:
                rd = dict(list(row.items()))
                # remove legacy columns
                # TODO: remove the column later
                rd.pop("size", None)
                item = ReportStat(**rd)
                i += 1
                d_range = item.partition_id
                es_docs[d_range].append(item.es_doc())
            if es_docs:
                name = partition_table.name
                log.info("round  {}, {}".format(i, name))
                for k, v in es_docs.items():
                    to_update = {"_index": k, "_type": "report"}
                    [i.update(to_update) for i in v]
                    elasticsearch.helpers.bulk(Datastores.es, v)

    log.info("total docs {} {}".format(i, datetime.datetime.now() - task_start))


def reindex_logs():
    try:
        Datastores.es.indices.delete("rcae_l_*")
    except elasticsearch.exceptions.NotFoundError as e:
        log.error(e)

    # logs
    log.info("reindexing logs")
    i = 0
    task_start = datetime.datetime.now()
    log_tables = detect_tables("logs_p_")
    for partition_table in log_tables:
        conn = DBSession.connection().execution_options(stream_results=True)
        result = conn.execute(partition_table.select())
        while True:
            chunk = result.fetchmany(2000)
            if not chunk:
                break
            es_docs = defaultdict(list)

            for row in chunk:
                i += 1
                item = Log(**dict(list(row.items())))
                d_range = item.partition_id
                es_docs[d_range].append(item.es_doc())
            if es_docs:
                name = partition_table.name
                log.info("round  {}, {}".format(i, name))
                for k, v in es_docs.items():
                    to_update = {"_index": k, "_type": "log"}
                    [i.update(to_update) for i in v]
                    elasticsearch.helpers.bulk(Datastores.es, v)

    log.info("total docs {} {}".format(i, datetime.datetime.now() - task_start))


def reindex_metrics():
    try:
        Datastores.es.indices.delete("rcae_m_*")
    except elasticsearch.exceptions.NotFoundError as e:
        log.error(e)

    log.info("reindexing applications metrics")
    i = 0
    task_start = datetime.datetime.now()
    metric_tables = detect_tables("metrics_p_")
    for partition_table in metric_tables:
        conn = DBSession.connection().execution_options(stream_results=True)
        result = conn.execute(partition_table.select())
        while True:
            chunk = result.fetchmany(2000)
            if not chunk:
                break
            es_docs = defaultdict(list)
            for row in chunk:
                i += 1
                item = Metric(**dict(list(row.items())))
                d_range = item.partition_id
                es_docs[d_range].append(item.es_doc())
            if es_docs:
                name = partition_table.name
                log.info("round  {}, {}".format(i, name))
                for k, v in es_docs.items():
                    to_update = {"_index": k, "_type": "log"}
                    [i.update(to_update) for i in v]
                    elasticsearch.helpers.bulk(Datastores.es, v)

    log.info("total docs {} {}".format(i, datetime.datetime.now() - task_start))


def reindex_slow_calls():
    try:
        Datastores.es.indices.delete("rcae_sc_*")
    except elasticsearch.exceptions.NotFoundError as e:
        log.error(e)

    log.info("reindexing slow calls")
    i = 0
    task_start = datetime.datetime.now()
    slow_calls_tables = detect_tables("slow_calls_p_")
    for partition_table in slow_calls_tables:
        conn = DBSession.connection().execution_options(stream_results=True)
        result = conn.execute(partition_table.select())
        while True:
            chunk = result.fetchmany(2000)
            if not chunk:
                break
            es_docs = defaultdict(list)
            for row in chunk:
                i += 1
                item = SlowCall(**dict(list(row.items())))
                d_range = item.partition_id
                es_docs[d_range].append(item.es_doc())
            if es_docs:
                name = partition_table.name
                log.info("round  {}, {}".format(i, name))
                for k, v in es_docs.items():
                    to_update = {"_index": k, "_type": "log"}
                    [i.update(to_update) for i in v]
                    elasticsearch.helpers.bulk(Datastores.es, v)

    log.info("total docs {} {}".format(i, datetime.datetime.now() - task_start))


if __name__ == "__main__":
    main()

f18e8ef0f648007dfc5296ccd709867cf3712803
<mask token>


def detect_tables(table_prefix):
    found_tables = []
    db_tables_query = """
    SELECT tablename FROM pg_tables WHERE tablename NOT LIKE 'pg_%' AND
    tablename NOT LIKE 'sql_%' ORDER BY tablename ASC;"""
    for table in DBSession.execute(db_tables_query).fetchall():
        tablename = table.tablename
        if tablename.startswith(table_prefix):
            t = sa.Table(tablename, metadata, autoload=True, autoload_with=
                DBSession.bind.engine)
            found_tables.append(t)
    return found_tables


def main():
    """
    Recreates Elasticsearch indexes
    Performs reindex of whole db to Elasticsearch

    """
    pre_parser = argparse.ArgumentParser(description=
        'Reindex AppEnlight data', add_help=False)
    pre_parser.add_argument('-c', '--config', required=True, help=
        'Configuration ini file of application')
    pre_parser.add_argument('-h', '--help', help='Show help', nargs='?')
    pre_parser.add_argument('-t', '--types', nargs='+', help=
        'Which parts of database should get reindexed')
    args = pre_parser.parse_args()
    config_uri = args.config
    setup_logging(config_uri)
    log.setLevel(logging.INFO)
    env = bootstrap(config_uri)
    parser = argparse.ArgumentParser(description='Reindex AppEnlight data')
    choices = {'reports':
        'appenlight.scripts.reindex_elasticsearch:reindex_reports', 'logs':
        'appenlight.scripts.reindex_elasticsearch:reindex_logs', 'metrics':
        'appenlight.scripts.reindex_elasticsearch:reindex_metrics',
        'slow_calls':
        'appenlight.scripts.reindex_elasticsearch:reindex_slow_calls',
        'template': 'appenlight.scripts.reindex_elasticsearch:update_template'}
    for k, v in env['registry'].appenlight_plugins.items():
        if v.get('fulltext_indexer'):
            choices[k] = v['fulltext_indexer']
    parser.add_argument('-t', '--types', nargs='*', choices=['all'] + list(
        choices.keys()), default=[], help=
        'Which parts of database should get reindexed')
    parser.add_argument('-c', '--config', required=True, help=
        'Configuration ini file of application')
    args = parser.parse_args()
    if 'all' in args.types:
        args.types = list(choices.keys())
    print('Selected types to reindex: {}'.format(args.types))
    log.info('settings {}'.format(args.types))
    if 'template' in args.types:
        get_callable(choices['template'])()
        args.types.remove('template')
    for selected in args.types:
        get_callable(choices[selected])()


def update_template():
    try:
        Datastores.es.indices.delete_template('rcae_reports')
    except elasticsearch.exceptions.NotFoundError as e:
        log.error(e)
    try:
        Datastores.es.indices.delete_template('rcae_logs')
    except elasticsearch.exceptions.NotFoundError as e:
        log.error(e)
    try:
        Datastores.es.indices.delete_template('rcae_slow_calls')
    except elasticsearch.exceptions.NotFoundError as e:
        log.error(e)
    try:
        Datastores.es.indices.delete_template('rcae_metrics')
    except elasticsearch.exceptions.NotFoundError as e:
        log.error(e)
    log.info('updating elasticsearch template')
    tag_templates = [{'values': {'path_match': 'tags.*', 'mapping': {'type':
        'object', 'properties': {'values': {'type': 'text', 'analyzer':
        'tag_value', 'fields': {'keyword': {'type': 'keyword',
        'ignore_above': 256}}}, 'numeric_values': {'type': 'float'}}}}}]
    shared_analysis = {'analyzer': {'url_path': {'type': 'custom',
        'char_filter': [], 'tokenizer': 'path_hierarchy', 'filter': []},
        'tag_value': {'type': 'custom', 'char_filter': [], 'tokenizer':
        'keyword', 'filter': ['lowercase']}}}
    shared_log_mapping = {'_all': {'enabled': False}, 'dynamic_templates':
        tag_templates, 'properties': {'pg_id': {'type': 'keyword', 'index':
        True}, 'delete_hash': {'type': 'keyword', 'index': True},
        'resource_id': {'type': 'integer'}, 'timestamp': {'type': 'date'},
        'permanent': {'type': 'boolean'}, 'request_id': {'type': 'keyword',
        'index': True}, 'log_level': {'type': 'text', 'analyzer': 'simple'},
        'message': {'type': 'text', 'analyzer': 'simple'}, 'namespace': {
        'type': 'text', 'fields': {'keyword': {'type': 'keyword',
        'ignore_above': 256}}}, 'tags': {'type': 'object'}, 'tag_list': {
        'type': 'text', 'analyzer': 'tag_value', 'fields': {'keyword': {
        'type': 'keyword', 'ignore_above': 256}}}}}
    report_schema = {'template': 'rcae_r_*', 'settings': {'index': {
        'refresh_interval': '5s', 'translog': {'sync_interval': '5s',
        'durability': 'async'}}, 'number_of_shards': 5, 'analysis':
        shared_analysis}, 'mappings': {'report': {'_all': {'enabled': False
        }, 'dynamic_templates': tag_templates, 'properties': {'type': {
        'type': 'keyword', 'index': True}, 'group_id': {'type': 'keyword',
        'index': True}, 'resource_id': {'type': 'integer'}, 'priority': {
        'type': 'integer'}, 'error': {'type': 'text', 'analyzer': 'simple'},
        'read': {'type': 'boolean'}, 'occurences': {'type': 'integer'},
        'fixed': {'type': 'boolean'}, 'first_timestamp': {'type': 'date'},
        'last_timestamp': {'type': 'date'}, 'average_duration': {'type':
        'float'}, 'summed_duration': {'type': 'float'}, 'public': {'type':
        'boolean'}, 'report_id': {'type': 'keyword', 'index': True},
        'http_status': {'type': 'integer'}, 'ip': {'type': 'keyword',
        'index': True}, 'url_domain': {'type': 'text', 'analyzer': 'simple'
        }, 'url_path': {'type': 'text', 'analyzer': 'url_path'},
        'report_type': {'type': 'integer'}, 'start_time': {'type': 'date'},
        'request_id': {'type': 'keyword', 'index': True}, 'end_time': {
        'type': 'date'}, 'duration': {'type': 'float'}, 'tags': {'type':
        'object'}, 'tag_list': {'type': 'text', 'analyzer': 'tag_value',
        'fields': {'keyword': {'type': 'keyword', 'ignore_above': 256}}},
        'extra': {'type': 'object'}, 'report_stat_id': {'type': 'keyword',
        'index': True}, 'timestamp': {'type': 'date'}, 'permanent': {'type':
        'boolean'}, 'log_level': {'type': 'text', 'analyzer': 'simple'},
        'message': {'type': 'text', 'analyzer': 'simple'}, 'namespace': {
        'type': 'text', 'fields': {'keyword': {'type': 'keyword',
        'ignore_above': 256}}}, 'join_field': {'type': 'join', 'relations':
        {'report_group': ['report', 'report_stat']}}}}}}
    Datastores.es.indices.put_template('rcae_reports', body=report_schema)
    logs_mapping = copy.deepcopy(shared_log_mapping)
    logs_mapping['properties']['log_id'] = logs_mapping['properties']['pg_id']
    del logs_mapping['properties']['pg_id']
    log_template = {'template': 'rcae_l_*', 'settings': {'index': {
        'refresh_interval': '5s', 'translog': {'sync_interval': '5s',
        'durability': 'async'}}, 'number_of_shards': 5, 'analysis':
        shared_analysis}, 'mappings': {'log': logs_mapping}}
    Datastores.es.indices.put_template('rcae_logs', body=log_template)
    slow_call_mapping = copy.deepcopy(shared_log_mapping)
    slow_call_mapping['properties']['slow_call_id'] = slow_call_mapping[
        'properties']['pg_id']
    del slow_call_mapping['properties']['pg_id']
    slow_call_template = {'template': 'rcae_sc_*', 'settings': {'index': {
        'refresh_interval': '5s', 'translog': {'sync_interval': '5s',
        'durability': 'async'}}, 'number_of_shards': 5, 'analysis':
        shared_analysis}, 'mappings': {'log': slow_call_mapping}}
    Datastores.es.indices.put_template('rcae_slow_calls', body=
        slow_call_template)
    metric_mapping = copy.deepcopy(shared_log_mapping)
    metric_mapping['properties']['metric_id'] = metric_mapping['properties'][
        'pg_id']
    del metric_mapping['properties']['pg_id']
    metrics_template = {'template': 'rcae_m_*', 'settings': {'index': {
        'refresh_interval': '5s', 'translog': {'sync_interval': '5s',
        'durability': 'async'}}, 'number_of_shards': 5, 'analysis':
        shared_analysis}, 'mappings': {'log': metric_mapping}}
    Datastores.es.indices.put_template('rcae_metrics', body=metrics_template)
    uptime_metric_mapping = copy.deepcopy(shared_log_mapping)
    uptime_metric_mapping['properties']['uptime_id'] = uptime_metric_mapping[
        'properties']['pg_id']
    del uptime_metric_mapping['properties']['pg_id']
    uptime_metrics_template = {'template': 'rcae_uptime_ce_*', 'settings':
        {'index': {'refresh_interval': '5s', 'translog': {'sync_interval':
        '5s', 'durability': 'async'}}, 'number_of_shards': 5, 'analysis':
        shared_analysis}, 'mappings': {'log': shared_log_mapping}}
    Datastores.es.indices.put_template('rcae_uptime_metrics', body=
        uptime_metrics_template)


def reindex_reports():
    reports_groups_tables = detect_tables('reports_groups_p_')
    try:
        Datastores.es.indices.delete('`rcae_r_*')
    except elasticsearch.exceptions.NotFoundError as e:
        log.error(e)
    log.info('reindexing report groups')
    i = 0
    task_start = datetime.datetime.now()
    for partition_table in reports_groups_tables:
        conn = DBSession.connection().execution_options(stream_results=True)
        result = conn.execute(partition_table.select())
        while True:
            chunk = result.fetchmany(2000)
            if not chunk:
                break
            es_docs = defaultdict(list)
            for row in chunk:
                i += 1
                item = ReportGroup(**dict(list(row.items())))
                d_range = item.partition_id
                es_docs[d_range].append(item.es_doc())
            if es_docs:
                name = partition_table.name
                log.info('round {}, {}'.format(i, name))
                for k, v in es_docs.items():
                    to_update = {'_index': k, '_type': 'report'}
                    [i.update(to_update) for i in v]
                    elasticsearch.helpers.bulk(Datastores.es, v)
    log.info('total docs {} {}'.format(i, datetime.datetime.now() - task_start)
        )
    i = 0
    log.info('reindexing reports')
    task_start = datetime.datetime.now()
    reports_tables = detect_tables('reports_p_')
    for partition_table in reports_tables:
        conn = DBSession.connection().execution_options(stream_results=True)
        result = conn.execute(partition_table.select())
        while True:
            chunk = result.fetchmany(2000)
            if not chunk:
                break
            es_docs = defaultdict(list)
            for row in chunk:
                i += 1
                item = Report(**dict(list(row.items())))
                d_range = item.partition_id
                es_docs[d_range].append(item.es_doc())
            if es_docs:
                name = partition_table.name
                log.info('round {}, {}'.format(i, name))
                for k, v in es_docs.items():
                    to_update = {'_index': k, '_type': 'report'}
                    [i.update(to_update) for i in v]
                    elasticsearch.helpers.bulk(Datastores.es, v)
    log.info('total docs {} {}'.format(i, datetime.datetime.now() - task_start)
        )
    log.info('reindexing reports stats')
    i = 0
    task_start = datetime.datetime.now()
    reports_stats_tables = detect_tables('reports_stats_p_')
    for partition_table in reports_stats_tables:
        conn = DBSession.connection().execution_options(stream_results=True)
        result = conn.execute(partition_table.select())
        while True:
            chunk = result.fetchmany(2000)
            if not chunk:
                break
            es_docs = defaultdict(list)
            for row in chunk:
                rd = dict(list(row.items()))
                rd.pop('size', None)
                item = ReportStat(**rd)
                i += 1
                d_range = item.partition_id
                es_docs[d_range].append(item.es_doc())
            if es_docs:
                name = partition_table.name
                log.info('round  {}, {}'.format(i, name))
                for k, v in es_docs.items():
                    to_update = {'_index': k, '_type': 'report'}
                    [i.update(to_update) for i in v]
                    elasticsearch.helpers.bulk(Datastores.es, v)
    log.info('total docs {} {}'.format(i, datetime.datetime.now() - task_start)
        )


def reindex_logs():
    try:
        Datastores.es.indices.delete('rcae_l_*')
    except elasticsearch.exceptions.NotFoundError as e:
        log.error(e)
    log.info('reindexing logs')
    i = 0
    task_start = datetime.datetime.now()
    log_tables = detect_tables('logs_p_')
    for partition_table in log_tables:
        conn = DBSession.connection().execution_options(stream_results=True)
        result = conn.execute(partition_table.select())
        while True:
            chunk = result.fetchmany(2000)
            if not chunk:
                break
            es_docs = defaultdict(list)
            for row in chunk:
                i += 1
                item = Log(**dict(list(row.items())))
                d_range = item.partition_id
                es_docs[d_range].append(item.es_doc())
            if es_docs:
                name = partition_table.name
                log.info('round  {}, {}'.format(i, name))
                for k, v in es_docs.items():
                    to_update = {'_index': k, '_type': 'log'}
                    [i.update(to_update) for i in v]
                    elasticsearch.helpers.bulk(Datastores.es, v)
    log.info('total docs {} {}'.format(i, datetime.datetime.now() - task_start)
        )


def reindex_metrics():
    try:
        Datastores.es.indices.delete('rcae_m_*')
    except elasticsearch.exceptions.NotFoundError as e:
        log.error(e)
    log.info('reindexing applications metrics')
    i = 0
    task_start = datetime.datetime.now()
    metric_tables = detect_tables('metrics_p_')
    for partition_table in metric_tables:
        conn = DBSession.connection().execution_options(stream_results=True)
        result = conn.execute(partition_table.select())
        while True:
            chunk = result.fetchmany(2000)
            if not chunk:
                break
            es_docs = defaultdict(list)
            for row in chunk:
                i += 1
                item = Metric(**dict(list(row.items())))
                d_range = item.partition_id
                es_docs[d_range].append(item.es_doc())
            if es_docs:
                name = partition_table.name
                log.info('round  {}, {}'.format(i, name))
                for k, v in es_docs.items():
                    to_update = {'_index': k, '_type': 'log'}
                    [i.update(to_update) for i in v]
                    elasticsearch.helpers.bulk(Datastores.es, v)
    log.info('total docs {} {}'.format(i, datetime.datetime.now() - task_start)
        )


def reindex_slow_calls():
    try:
        Datastores.es.indices.delete('rcae_sc_*')
    except elasticsearch.exceptions.NotFoundError as e:
        log.error(e)
    log.info('reindexing slow calls')
    i = 0
    task_start = datetime.datetime.now()
    slow_calls_tables = detect_tables('slow_calls_p_')
    for partition_table in slow_calls_tables:
        conn = DBSession.connection().execution_options(stream_results=True)
        result = conn.execute(partition_table.select())
        while True:
            chunk = result.fetchmany(2000)
            if not chunk:
                break
            es_docs = defaultdict(list)
            for row in chunk:
                i += 1
                item = SlowCall(**dict(list(row.items())))
                d_range = item.partition_id
                es_docs[d_range].append(item.es_doc())
            if es_docs:
                name = partition_table.name
                log.info('round  {}, {}'.format(i, name))
                for k, v in es_docs.items():
                    to_update = {'_index': k, '_type': 'log'}
                    [i.update(to_update) for i in v]
                    elasticsearch.helpers.bulk(Datastores.es, v)
    log.info('total docs {} {}'.format(i, datetime.datetime.now() - task_start)
        )


<mask token>

<mask token>


def detect_tables(table_prefix):
    found_tables = []
    db_tables_query = """
    SELECT tablename FROM pg_tables WHERE tablename NOT LIKE 'pg_%' AND
    tablename NOT LIKE 'sql_%' ORDER BY tablename ASC;"""
    for table in DBSession.execute(db_tables_query).fetchall():
        tablename = table.tablename
        if tablename.startswith(table_prefix):
            t = sa.Table(tablename, metadata, autoload=True, autoload_with=
                DBSession.bind.engine)
            found_tables.append(t)
    return found_tables


def main():
    """
    Recreates Elasticsearch indexes
    Performs reindex of whole db to Elasticsearch

    """
    pre_parser = argparse.ArgumentParser(description=
        'Reindex AppEnlight data', add_help=False)
    pre_parser.add_argument('-c', '--config', required=True, help=
        'Configuration ini file of application')
    pre_parser.add_argument('-h', '--help', help='Show help', nargs='?')
    pre_parser.add_argument('-t', '--types', nargs='+', help=
        'Which parts of database should get reindexed')
    args = pre_parser.parse_args()
    config_uri = args.config
    setup_logging(config_uri)
    log.setLevel(logging.INFO)
    env = bootstrap(config_uri)
    parser = argparse.ArgumentParser(description='Reindex AppEnlight data')
    choices = {'reports':
        'appenlight.scripts.reindex_elasticsearch:reindex_reports', 'logs':
        'appenlight.scripts.reindex_elasticsearch:reindex_logs', 'metrics':
        'appenlight.scripts.reindex_elasticsearch:reindex_metrics',
        'slow_calls':
        'appenlight.scripts.reindex_elasticsearch:reindex_slow_calls',
        'template': 'appenlight.scripts.reindex_elasticsearch:update_template'}
    for k, v in env['registry'].appenlight_plugins.items():
        if v.get('fulltext_indexer'):
            choices[k] = v['fulltext_indexer']
    parser.add_argument('-t', '--types', nargs='*', choices=['all'] + list(
        choices.keys()), default=[], help=
        'Which parts of database should get reindexed')
    parser.add_argument('-c', '--config', required=True, help=
        'Configuration ini file of application')
    args = parser.parse_args()
    if 'all' in args.types:
        args.types = list(choices.keys())
    print('Selected types to reindex: {}'.format(args.types))
    log.info('settings {}'.format(args.types))
    if 'template' in args.types:
        get_callable(choices['template'])()
        args.types.remove('template')
    for selected in args.types:
        get_callable(choices[selected])()


def update_template():
    try:
        Datastores.es.indices.delete_template('rcae_reports')
    except elasticsearch.exceptions.NotFoundError as e:
        log.error(e)
    try:
        Datastores.es.indices.delete_template('rcae_logs')
    except elasticsearch.exceptions.NotFoundError as e:
        log.error(e)
    try:
        Datastores.es.indices.delete_template('rcae_slow_calls')
    except elasticsearch.exceptions.NotFoundError as e:
        log.error(e)
    try:
        Datastores.es.indices.delete_template('rcae_metrics')
    except elasticsearch.exceptions.NotFoundError as e:
        log.error(e)
    log.info('updating elasticsearch template')
    tag_templates = [{'values': {'path_match': 'tags.*', 'mapping': {'type':
        'object', 'properties': {'values': {'type': 'text', 'analyzer':
        'tag_value', 'fields': {'keyword': {'type': 'keyword',
        'ignore_above': 256}}}, 'numeric_values': {'type': 'float'}}}}}]
    shared_analysis = {'analyzer': {'url_path': {'type': 'custom',
        'char_filter': [], 'tokenizer': 'path_hierarchy', 'filter': []},
        'tag_value': {'type': 'custom', 'char_filter': [], 'tokenizer':
        'keyword', 'filter': ['lowercase']}}}
    shared_log_mapping = {'_all': {'enabled': False}, 'dynamic_templates':
        tag_templates, 'properties': {'pg_id': {'type': 'keyword', 'index':
        True}, 'delete_hash': {'type': 'keyword', 'index': True},
        'resource_id': {'type': 'integer'}, 'timestamp': {'type': 'date'},
        'permanent': {'type': 'boolean'}, 'request_id': {'type': 'keyword',
        'index': True}, 'log_level': {'type': 'text', 'analyzer': 'simple'},
        'message': {'type': 'text', 'analyzer': 'simple'}, 'namespace': {
        'type': 'text', 'fields': {'keyword': {'type': 'keyword',
        'ignore_above': 256}}}, 'tags': {'type': 'object'}, 'tag_list': {
        'type': 'text', 'analyzer': 'tag_value', 'fields': {'keyword': {
        'type': 'keyword', 'ignore_above': 256}}}}}
    report_schema = {'template': 'rcae_r_*', 'settings': {'index': {
        'refresh_interval': '5s', 'translog': {'sync_interval': '5s',
        'durability': 'async'}}, 'number_of_shards': 5, 'analysis':
        shared_analysis}, 'mappings': {'report': {'_all': {'enabled': False
        }, 'dynamic_templates': tag_templates, 'properties': {'type': {
        'type': 'keyword', 'index': True}, 'group_id': {'type': 'keyword',
        'index': True}, 'resource_id': {'type': 'integer'}, 'priority': {
        'type': 'integer'}, 'error': {'type': 'text', 'analyzer': 'simple'},
        'read': {'type': 'boolean'}, 'occurences': {'type': 'integer'},
        'fixed': {'type': 'boolean'}, 'first_timestamp': {'type': 'date'},
        'last_timestamp': {'type': 'date'}, 'average_duration': {'type':
        'float'}, 'summed_duration': {'type': 'float'}, 'public': {'type':
        'boolean'}, 'report_id': {'type': 'keyword', 'index': True},
        'http_status': {'type': 'integer'}, 'ip': {'type': 'keyword',
        'index': True}, 'url_domain': {'type': 'text', 'analyzer': 'simple'
        }, 'url_path': {'type': 'text', 'analyzer': 'url_path'},
        'report_type': {'type': 'integer'}, 'start_time': {'type': 'date'},
        'request_id': {'type': 'keyword', 'index': True}, 'end_time': {
        'type': 'date'}, 'duration': {'type': 'float'}, 'tags': {'type':
        'object'}, 'tag_list': {'type': 'text', 'analyzer': 'tag_value',
        'fields': {'keyword': {'type': 'keyword', 'ignore_above': 256}}},
        'extra': {'type': 'object'}, 'report_stat_id': {'type': 'keyword',
        'index': True}, 'timestamp': {'type': 'date'}, 'permanent': {'type':
        'boolean'}, 'log_level': {'type': 'text', 'analyzer': 'simple'},
        'message': {'type': 'text', 'analyzer': 'simple'}, 'namespace': {
        'type': 'text', 'fields': {'keyword': {'type': 'keyword',
        'ignore_above': 256}}}, 'join_field': {'type': 'join', 'relations':
        {'report_group': ['report', 'report_stat']}}}}}}
    Datastores.es.indices.put_template('rcae_reports', body=report_schema)
    logs_mapping = copy.deepcopy(shared_log_mapping)
    logs_mapping['properties']['log_id'] = logs_mapping['properties']['pg_id']
    del logs_mapping['properties']['pg_id']
    log_template = {'template': 'rcae_l_*', 'settings': {'index': {
        'refresh_interval': '5s', 'translog': {'sync_interval': '5s',
        'durability': 'async'}}, 'number_of_shards': 5, 'analysis':
        shared_analysis}, 'mappings': {'log': logs_mapping}}
    Datastores.es.indices.put_template('rcae_logs', body=log_template)
    slow_call_mapping = copy.deepcopy(shared_log_mapping)
    slow_call_mapping['properties']['slow_call_id'] = slow_call_mapping[
        'properties']['pg_id']
    del slow_call_mapping['properties']['pg_id']
    slow_call_template = {'template': 'rcae_sc_*', 'settings': {'index': {
        'refresh_interval': '5s', 'translog': {'sync_interval': '5s',
        'durability': 'async'}}, 'number_of_shards': 5, 'analysis':
        shared_analysis}, 'mappings': {'log': slow_call_mapping}}
    Datastores.es.indices.put_template('rcae_slow_calls', body=
        slow_call_template)
    metric_mapping = copy.deepcopy(shared_log_mapping)
    metric_mapping['properties']['metric_id'] = metric_mapping['properties'][
        'pg_id']
    del metric_mapping['properties']['pg_id']
    metrics_template = {'template': 'rcae_m_*', 'settings': {'index': {
        'refresh_interval': '5s', 'translog': {'sync_interval': '5s',
        'durability': 'async'}}, 'number_of_shards': 5, 'analysis':
        shared_analysis}, 'mappings': {'log': metric_mapping}}
    Datastores.es.indices.put_template('rcae_metrics', body=metrics_template)
    uptime_metric_mapping = copy.deepcopy(shared_log_mapping)
    uptime_metric_mapping['properties']['uptime_id'] = uptime_metric_mapping[
        'properties']['pg_id']
    del uptime_metric_mapping['properties']['pg_id']
    uptime_metrics_template = {'template': 'rcae_uptime_ce_*', 'settings':
        {'index': {'refresh_interval': '5s', 'translog': {'sync_interval':
        '5s', 'durability': 'async'}}, 'number_of_shards': 5, 'analysis':
        shared_analysis}, 'mappings': {'log': shared_log_mapping}}
    Datastores.es.indices.put_template('rcae_uptime_metrics', body=
        uptime_metrics_template)


def reindex_reports():
    reports_groups_tables = detect_tables('reports_groups_p_')
    try:
        Datastores.es.indices.delete('`rcae_r_*')
    except elasticsearch.exceptions.NotFoundError as e:
        log.error(e)
    log.info('reindexing report groups')
    i = 0
    task_start = datetime.datetime.now()
    for partition_table in reports_groups_tables:
        conn = DBSession.connection().execution_options(stream_results=True)
        result = conn.execute(partition_table.select())
        while True:
            chunk = result.fetchmany(2000)
            if not chunk:
                break
            es_docs = defaultdict(list)
            for row in chunk:
                i += 1
                item = ReportGroup(**dict(list(row.items())))
                d_range = item.partition_id
                es_docs[d_range].append(item.es_doc())
            if es_docs:
                name = partition_table.name
                log.info('round {}, {}'.format(i, name))
                for k, v in es_docs.items():
                    to_update = {'_index': k, '_type': 'report'}
                    [i.update(to_update) for i in v]
                    elasticsearch.helpers.bulk(Datastores.es, v)
    log.info('total docs {} {}'.format(i, datetime.datetime.now() - task_start)
        )
    i = 0
    log.info('reindexing reports')
    task_start = datetime.datetime.now()
    reports_tables = detect_tables('reports_p_')
    for partition_table in reports_tables:
        conn = DBSession.connection().execution_options(stream_results=True)
        result = conn.execute(partition_table.select())
        while True:
            chunk = result.fetchmany(2000)
            if not chunk:
                break
            es_docs = defaultdict(list)
            for row in chunk:
                i += 1
                item = Report(**dict(list(row.items())))
                d_range = item.partition_id
                es_docs[d_range].append(item.es_doc())
            if es_docs:
                name = partition_table.name
                log.info('round {}, {}'.format(i, name))
                for k, v in es_docs.items():
                    to_update = {'_index': k, '_type': 'report'}
                    [i.update(to_update) for i in v]
                    elasticsearch.helpers.bulk(Datastores.es, v)
    log.info('total docs {} {}'.format(i, datetime.datetime.now() - task_start)
        )
    log.info('reindexing reports stats')
    i = 0
    task_start = datetime.datetime.now()
    reports_stats_tables = detect_tables('reports_stats_p_')
    for partition_table in reports_stats_tables:
        conn = DBSession.connection().execution_options(stream_results=True)
        result = conn.execute(partition_table.select())
        while True:
            chunk = result.fetchmany(2000)
            if not chunk:
                break
            es_docs = defaultdict(list)
            for row in chunk:
                rd = dict(list(row.items()))
                rd.pop('size', None)
                item = ReportStat(**rd)
                i += 1
                d_range = item.partition_id
                es_docs[d_range].append(item.es_doc())
            if es_docs:
                name = partition_table.name
                log.info('round  {}, {}'.format(i, name))
                for k, v in es_docs.items():
                    to_update = {'_index': k, '_type': 'report'}
                    [i.update(to_update) for i in v]
                    elasticsearch.helpers.bulk(Datastores.es, v)
    log.info('total docs {} {}'.format(i, datetime.datetime.now() - task_start)
        )


def reindex_logs():
    try:
        Datastores.es.indices.delete('rcae_l_*')
    except elasticsearch.exceptions.NotFoundError as e:
        log.error(e)
    log.info('reindexing logs')
    i = 0
    task_start = datetime.datetime.now()
    log_tables = detect_tables('logs_p_')
    for partition_table in log_tables:
        conn = DBSession.connection().execution_options(stream_results=True)
        result = conn.execute(partition_table.select())
        while True:
            chunk = result.fetchmany(2000)
            if not chunk:
                break
            es_docs = defaultdict(list)
            for row in chunk:
                i += 1
                item = Log(**dict(list(row.items())))
                d_range = item.partition_id
                es_docs[d_range].append(item.es_doc())
            if es_docs:
                name = partition_table.name
                log.info('round  {}, {}'.format(i, name))
                for k, v in es_docs.items():
                    to_update = {'_index': k, '_type': 'log'}
                    [i.update(to_update) for i in v]
                    elasticsearch.helpers.bulk(Datastores.es, v)
    log.info('total docs {} {}'.format(i, datetime.datetime.now() - task_start)
        )


def reindex_metrics():
    try:
        Datastores.es.indices.delete('rcae_m_*')
    except elasticsearch.exceptions.NotFoundError as e:
        log.error(e)
    log.info('reindexing applications metrics')
    i = 0
    task_start = datetime.datetime.now()
    metric_tables = detect_tables('metrics_p_')
    for partition_table in metric_tables:
        conn = DBSession.connection().execution_options(stream_results=True)
        result = conn.execute(partition_table.select())
        while True:
            chunk = result.fetchmany(2000)
            if not chunk:
                break
            es_docs = defaultdict(list)
            for row in chunk:
                i += 1
                item = Metric(**dict(list(row.items())))
                d_range = item.partition_id
                es_docs[d_range].append(item.es_doc())
            if es_docs:
                name = partition_table.name
                log.info('round  {}, {}'.format(i, name))
                for k, v in es_docs.items():
                    to_update = {'_index': k, '_type': 'log'}
                    [i.update(to_update) for i in v]
                    elasticsearch.helpers.bulk(Datastores.es, v)
    log.info('total docs {} {}'.format(i, datetime.datetime.now() - task_start)
        )


def reindex_slow_calls():
    try:
        Datastores.es.indices.delete('rcae_sc_*')
    except elasticsearch.exceptions.NotFoundError as e:
        log.error(e)
    log.info('reindexing slow calls')
    i = 0
    task_start = datetime.datetime.now()
    slow_calls_tables = detect_tables('slow_calls_p_')
    for partition_table in slow_calls_tables:
        conn = DBSession.connection().execution_options(stream_results=True)
        result = conn.execute(partition_table.select())
        while True:
            chunk = result.fetchmany(2000)
            if not chunk:
                break
            es_docs = defaultdict(list)
            for row in chunk:
                i += 1
                item = SlowCall(**dict(list(row.items())))
                d_range = item.partition_id
                es_docs[d_range].append(item.es_doc())
            if es_docs:
                name = partition_table.name
                log.info('round  {}, {}'.format(i, name))
                for k, v in es_docs.items():
                    to_update = {'_index': k, '_type': 'log'}
                    [i.update(to_update) for i in v]
                    elasticsearch.helpers.bulk(Datastores.es, v)
    log.info('total docs {} {}'.format(i, datetime.datetime.now() - task_start)
        )


if __name__ == '__main__':
    main()

<mask token>
log = logging.getLogger(__name__)
tables = {'slow_calls_p_': [], 'reports_stats_p_': [], 'reports_p_': [],
    'reports_groups_p_': [], 'logs_p_': [], 'metrics_p_': []}


def detect_tables(table_prefix):
    found_tables = []
    db_tables_query = """
    SELECT tablename FROM pg_tables WHERE tablename NOT LIKE 'pg_%' AND
    tablename NOT LIKE 'sql_%' ORDER BY tablename ASC;"""
    for table in DBSession.execute(db_tables_query).fetchall():
        tablename = table.tablename
        if tablename.startswith(table_prefix):
            t = sa.Table(tablename, metadata, autoload=True, autoload_with=
                DBSession.bind.engine)
            found_tables.append(t)
    return found_tables


def main():
    """
    Recreates Elasticsearch indexes
    Performs reindex of whole db to Elasticsearch

    """
    pre_parser = argparse.ArgumentParser(description=
        'Reindex AppEnlight data', add_help=False)
    pre_parser.add_argument('-c', '--config', required=True, help=
        'Configuration ini file of application')
    pre_parser.add_argument('-h', '--help', help='Show help', nargs='?')
    pre_parser.add_argument('-t', '--types', nargs='+', help=
        'Which parts of database should get reindexed')
    args = pre_parser.parse_args()
    config_uri = args.config
    setup_logging(config_uri)
    log.setLevel(logging.INFO)
    env = bootstrap(config_uri)
    parser = argparse.ArgumentParser(description='Reindex AppEnlight data')
    choices = {'reports':
        'appenlight.scripts.reindex_elasticsearch:reindex_reports', 'logs':
        'appenlight.scripts.reindex_elasticsearch:reindex_logs', 'metrics':
        'appenlight.scripts.reindex_elasticsearch:reindex_metrics',
        'slow_calls':
        'appenlight.scripts.reindex_elasticsearch:reindex_slow_calls',
        'template': 'appenlight.scripts.reindex_elasticsearch:update_template'}
    for k, v in env['registry'].appenlight_plugins.items():
        if v.get('fulltext_indexer'):
            choices[k] = v['fulltext_indexer']
    parser.add_argument('-t', '--types', nargs='*', choices=['all'] + list(
        choices.keys()), default=[], help=
        'Which parts of database should get reindexed')
    parser.add_argument('-c', '--config', required=True, help=
        'Configuration ini file of application')
    args = parser.parse_args()
    if 'all' in args.types:
        args.types = list(choices.keys())
    print('Selected types to reindex: {}'.format(args.types))
    log.info('settings {}'.format(args.types))
    if 'template' in args.types:
        get_callable(choices['template'])()
        args.types.remove('template')
    for selected in args.types:
        get_callable(choices[selected])()


def update_template():
    try:
        Datastores.es.indices.delete_template('rcae_reports')
    except elasticsearch.exceptions.NotFoundError as e:
        log.error(e)
    try:
        Datastores.es.indices.delete_template('rcae_logs')
    except elasticsearch.exceptions.NotFoundError as e:
        log.error(e)
    try:
        Datastores.es.indices.delete_template('rcae_slow_calls')
    except elasticsearch.exceptions.NotFoundError as e:
        log.error(e)
    try:
        Datastores.es.indices.delete_template('rcae_metrics')
    except elasticsearch.exceptions.NotFoundError as e:
        log.error(e)
    log.info('updating elasticsearch template')
    tag_templates = [{'values': {'path_match': 'tags.*', 'mapping': {'type':
        'object', 'properties': {'values': {'type': 'text', 'analyzer':
        'tag_value', 'fields': {'keyword': {'type': 'keyword',
        'ignore_above': 256}}}, 'numeric_values': {'type': 'float'}}}}}]
    shared_analysis = {'analyzer': {'url_path': {'type': 'custom',
        'char_filter': [], 'tokenizer': 'path_hierarchy', 'filter': []},
        'tag_value': {'type': 'custom', 'char_filter': [], 'tokenizer':
        'keyword', 'filter': ['lowercase']}}}
    shared_log_mapping = {'_all': {'enabled': False}, 'dynamic_templates':
        tag_templates, 'properties': {'pg_id': {'type': 'keyword', 'index':
        True}, 'delete_hash': {'type': 'keyword', 'index': True},
        'resource_id': {'type': 'integer'}, 'timestamp': {'type': 'date'},
        'permanent': {'type': 'boolean'}, 'request_id': {'type': 'keyword',
        'index': True}, 'log_level': {'type': 'text', 'analyzer': 'simple'},
        'message': {'type': 'text', 'analyzer': 'simple'}, 'namespace': {
        'type': 'text', 'fields': {'keyword': {'type': 'keyword',
        'ignore_above': 256}}}, 'tags': {'type': 'object'}, 'tag_list': {
        'type': 'text', 'analyzer': 'tag_value', 'fields': {'keyword': {
        'type': 'keyword', 'ignore_above': 256}}}}}
    report_schema = {'template': 'rcae_r_*', 'settings': {'index': {
        'refresh_interval': '5s', 'translog': {'sync_interval': '5s',
        'durability': 'async'}}, 'number_of_shards': 5, 'analysis':
        shared_analysis}, 'mappings': {'report': {'_all': {'enabled': False
        }, 'dynamic_templates': tag_templates, 'properties': {'type': {
        'type': 'keyword', 'index': True}, 'group_id': {'type': 'keyword',
        'index': True}, 'resource_id': {'type': 'integer'}, 'priority': {
        'type': 'integer'}, 'error': {'type': 'text', 'analyzer': 'simple'},
        'read': {'type': 'boolean'}, 'occurences': {'type': 'integer'},
        'fixed': {'type': 'boolean'}, 'first_timestamp': {'type': 'date'},
        'last_timestamp': {'type': 'date'}, 'average_duration': {'type':
        'float'}, 'summed_duration': {'type': 'float'}, 'public': {'type':
        'boolean'}, 'report_id': {'type': 'keyword', 'index': True},
        'http_status': {'type': 'integer'}, 'ip': {'type': 'keyword',
        'index': True}, 'url_domain': {'type': 'text', 'analyzer': 'simple'
        }, 'url_path': {'type': 'text', 'analyzer': 'url_path'},
        'report_type': {'type': 'integer'}, 'start_time': {'type': 'date'},
        'request_id': {'type': 'keyword', 'index': True}, 'end_time': {
        'type': 'date'}, 'duration': {'type': 'float'}, 'tags': {'type':
        'object'}, 'tag_list': {'type': 'text', 'analyzer': 'tag_value',
        'fields': {'keyword': {'type': 'keyword', 'ignore_above': 256}}},
        'extra': {'type': 'object'}, 'report_stat_id': {'type': 'keyword',
        'index': True}, 'timestamp': {'type': 'date'}, 'permanent': {'type':
        'boolean'}, 'log_level': {'type': 'text', 'analyzer': 'simple'},
        'message': {'type': 'text', 'analyzer': 'simple'}, 'namespace': {
        'type': 'text', 'fields': {'keyword': {'type': 'keyword',
        'ignore_above': 256}}}, 'join_field': {'type': 'join', 'relations':
        {'report_group': ['report', 'report_stat']}}}}}}
    Datastores.es.indices.put_template('rcae_reports', body=report_schema)
    logs_mapping = copy.deepcopy(shared_log_mapping)
    logs_mapping['properties']['log_id'] = logs_mapping['properties']['pg_id']
    del logs_mapping['properties']['pg_id']
    log_template = {'template': 'rcae_l_*', 'settings': {'index': {
        'refresh_interval': '5s', 'translog': {'sync_interval': '5s',
        'durability': 'async'}}, 'number_of_shards': 5, 'analysis':
        shared_analysis}, 'mappings': {'log': logs_mapping}}
    Datastores.es.indices.put_template('rcae_logs', body=log_template)
    slow_call_mapping = copy.deepcopy(shared_log_mapping)
    slow_call_mapping['properties']['slow_call_id'] = slow_call_mapping[
        'properties']['pg_id']
    del slow_call_mapping['properties']['pg_id']
    slow_call_template = {'template': 'rcae_sc_*', 'settings': {'index': {
        'refresh_interval': '5s', 'translog': {'sync_interval': '5s',
        'durability': 'async'}}, 'number_of_shards': 5, 'analysis':
        shared_analysis}, 'mappings': {'log': slow_call_mapping}}
    Datastores.es.indices.put_template('rcae_slow_calls', body=
        slow_call_template)
    metric_mapping = copy.deepcopy(shared_log_mapping)
    metric_mapping['properties']['metric_id'] = metric_mapping['properties'][
        'pg_id']
    del metric_mapping['properties']['pg_id']
    metrics_template = {'template': 'rcae_m_*', 'settings': {'index': {
        'refresh_interval': '5s', 'translog': {'sync_interval': '5s',
        'durability': 'async'}}, 'number_of_shards': 5, 'analysis':
        shared_analysis}, 'mappings': {'log': metric_mapping}}
    Datastores.es.indices.put_template('rcae_metrics', body=metrics_template)
    uptime_metric_mapping = copy.deepcopy(shared_log_mapping)
    uptime_metric_mapping['properties']['uptime_id'] = uptime_metric_mapping[
        'properties']['pg_id']
    del uptime_metric_mapping['properties']['pg_id']
    uptime_metrics_template = {'template': 'rcae_uptime_ce_*', 'settings':
        {'index': {'refresh_interval': '5s', 'translog': {'sync_interval':
        '5s', 'durability': 'async'}}, 'number_of_shards': 5, 'analysis':
        shared_analysis}, 'mappings': {'log': shared_log_mapping}}
    Datastores.es.indices.put_template('rcae_uptime_metrics', body=
        uptime_metrics_template)


def reindex_reports():
    reports_groups_tables = detect_tables('reports_groups_p_')
    try:
        Datastores.es.indices.delete('`rcae_r_*')
    except elasticsearch.exceptions.NotFoundError as e:
        log.error(e)
    log.info('reindexing report groups')
    i = 0
    task_start = datetime.datetime.now()
    for partition_table in reports_groups_tables:
        conn = DBSession.connection().execution_options(stream_results=True)
        result = conn.execute(partition_table.select())
        while True:
            chunk = result.fetchmany(2000)
            if not chunk:
                break
            es_docs = defaultdict(list)
            for row in chunk:
                i += 1
                item = ReportGroup(**dict(list(row.items())))
                d_range = item.partition_id
                es_docs[d_range].append(item.es_doc())
            if es_docs:
                name = partition_table.name
                log.info('round {}, {}'.format(i, name))
                for k, v in es_docs.items():
                    to_update = {'_index': k, '_type': 'report'}
                    [i.update(to_update) for i in v]
                    elasticsearch.helpers.bulk(Datastores.es, v)
    log.info('total docs {} {}'.format(i, datetime.datetime.now() - task_start)
        )
    i = 0
    log.info('reindexing reports')
    task_start = datetime.datetime.now()
    reports_tables = detect_tables('reports_p_')
    for partition_table in reports_tables:
        conn = DBSession.connection().execution_options(stream_results=True)
        result = conn.execute(partition_table.select())
        while True:
            chunk = result.fetchmany(2000)
            if not chunk:
                break
            es_docs = defaultdict(list)
            for row in chunk:
                i += 1
                item = Report(**dict(list(row.items())))
                d_range = item.partition_id
                es_docs[d_range].append(item.es_doc())
            if es_docs:
                name = partition_table.name
                log.info('round {}, {}'.format(i, name))
                for k, v in es_docs.items():
                    to_update = {'_index': k, '_type': 'report'}
                    [i.update(to_update) for i in v]
                    elasticsearch.helpers.bulk(Datastores.es, v)
    log.info('total docs {} {}'.format(i, datetime.datetime.now() - task_start)
        )
    log.info('reindexing reports stats')
    i = 0
    task_start = datetime.datetime.now()
    reports_stats_tables = detect_tables('reports_stats_p_')
    for partition_table in reports_stats_tables:
        conn = DBSession.connection().execution_options(stream_results=True)
        result = conn.execute(partition_table.select())
        while True:
            chunk = result.fetchmany(2000)
            if not chunk:
                break
            es_docs = defaultdict(list)
            for row in chunk:
                rd = dict(list(row.items()))
                rd.pop('size', None)
                item = ReportStat(**rd)
                i += 1
                d_range = item.partition_id
                es_docs[d_range].append(item.es_doc())
            if es_docs:
                name = partition_table.name
                log.info('round  {}, {}'.format(i, name))
                for k, v in es_docs.items():
                    to_update = {'_index': k, '_type': 'report'}
                    [i.update(to_update) for i in v]
                    elasticsearch.helpers.bulk(Datastores.es, v)
    log.info('total docs {} {}'.format(i, datetime.datetime.now() - task_start)
        )


def reindex_logs():
    try:
        Datastores.es.indices.delete('rcae_l_*')
    except elasticsearch.exceptions.NotFoundError as e:
        log.error(e)
    log.info('reindexing logs')
    i = 0
    task_start = datetime.datetime.now()
    log_tables = detect_tables('logs_p_')
    for partition_table in log_tables:
        conn = DBSession.connection().execution_options(stream_results=True)
        result = conn.execute(partition_table.select())
        while True:
            chunk = result.fetchmany(2000)
            if not chunk:
                break
            es_docs = defaultdict(list)
            for row in chunk:
                i += 1
                item = Log(**dict(list(row.items())))
                d_range = item.partition_id
                es_docs[d_range].append(item.es_doc())
            if es_docs:
                name = partition_table.name
                log.info('round  {}, {}'.format(i, name))
                for k, v in es_docs.items():
                    to_update = {'_index': k, '_type': 'log'}
                    [i.update(to_update) for i in v]
                    elasticsearch.helpers.bulk(Datastores.es, v)
    log.info('total docs {} {}'.format(i, datetime.datetime.now() - task_start)
        )


def reindex_metrics():
    try:
        Datastores.es.indices.delete('rcae_m_*')
    except elasticsearch.exceptions.NotFoundError as e:
        log.error(e)
    log.info('reindexing applications metrics')
    i = 0
    task_start = datetime.datetime.now()
    metric_tables = detect_tables('metrics_p_')
    for partition_table in metric_tables:
        conn = DBSession.connection().execution_options(stream_results=True)
        result = conn.execute(partition_table.select())
        while True:
            chunk = result.fetchmany(2000)
            if not chunk:
                break
            es_docs = defaultdict(list)
            for row in chunk:
                i += 1
                item = Metric(**dict(list(row.items())))
                d_range = item.partition_id
                es_docs[d_range].append(item.es_doc())
            if es_docs:
                name = partition_table.name
                log.info('round  {}, {}'.format(i, name))
                for k, v in es_docs.items():
                    to_update = {'_index': k, '_type': 'log'}
                    [i.update(to_update) for i in v]
                    elasticsearch.helpers.bulk(Datastores.es, v)
    log.info('total docs {} {}'.format(i, datetime.datetime.now() - task_start)
        )


def reindex_slow_calls():
    try:
        Datastores.es.indices.delete('rcae_sc_*')
    except elasticsearch.exceptions.NotFoundError as e:
        log.error(e)
    log.info('reindexing slow calls')
    i = 0
    task_start = datetime.datetime.now()
    slow_calls_tables = detect_tables('slow_calls_p_')
    for partition_table in slow_calls_tables:
        conn = DBSession.connection().execution_options(stream_results=True)
        result = conn.execute(partition_table.select())
        while True:
            chunk = result.fetchmany(2000)
            if not chunk:
                break
            es_docs = defaultdict(list)
            for row in chunk:
                i += 1
                item = SlowCall(**dict(list(row.items())))
                d_range = item.partition_id
                es_docs[d_range].append(item.es_doc())
            if es_docs:
                name = partition_table.name
                log.info('round  {}, {}'.format(i, name))
                for k, v in es_docs.items():
                    to_update = {'_index': k, '_type': 'log'}
                    [i.update(to_update) for i in v]
                    elasticsearch.helpers.bulk(Datastores.es, v)
    log.info('total docs {} {}'.format(i, datetime.datetime.now() - task_start)
        )


if __name__ == '__main__':
    main()

import argparse
import datetime
import logging
import copy
import sqlalchemy as sa
import elasticsearch.exceptions
import elasticsearch.helpers
from collections import defaultdict
from pyramid.paster import setup_logging
from pyramid.paster import bootstrap
from appenlight.models import DBSession, Datastores, metadata
from appenlight.lib import get_callable
from appenlight.models.report_group import ReportGroup
from appenlight.models.report import Report
from appenlight.models.report_stat import ReportStat
from appenlight.models.log import Log
from appenlight.models.slow_call import SlowCall
from appenlight.models.metric import Metric
log = logging.getLogger(__name__)
tables = {'slow_calls_p_': [], 'reports_stats_p_': [], 'reports_p_': [],
    'reports_groups_p_': [], 'logs_p_': [], 'metrics_p_': []}


def detect_tables(table_prefix):
    found_tables = []
    db_tables_query = """
    SELECT tablename FROM pg_tables WHERE tablename NOT LIKE 'pg_%' AND
    tablename NOT LIKE 'sql_%' ORDER BY tablename ASC;"""
    for table in DBSession.execute(db_tables_query).fetchall():
        tablename = table.tablename
        if tablename.startswith(table_prefix):
            t = sa.Table(tablename, metadata, autoload=True, autoload_with=
                DBSession.bind.engine)
            found_tables.append(t)
    return found_tables


def main():
    """
    Recreates Elasticsearch indexes
    Performs reindex of whole db to Elasticsearch

    """
    pre_parser = argparse.ArgumentParser(description=
        'Reindex AppEnlight data', add_help=False)
    pre_parser.add_argument('-c', '--config', required=True, help=
        'Configuration ini file of application')
    pre_parser.add_argument('-h', '--help', help='Show help', nargs='?')
    pre_parser.add_argument('-t', '--types', nargs='+', help=
        'Which parts of database should get reindexed')
    args = pre_parser.parse_args()
    config_uri = args.config
    setup_logging(config_uri)
    log.setLevel(logging.INFO)
    env = bootstrap(config_uri)
    parser = argparse.ArgumentParser(description='Reindex AppEnlight data')
    choices = {'reports':
        'appenlight.scripts.reindex_elasticsearch:reindex_reports', 'logs':
        'appenlight.scripts.reindex_elasticsearch:reindex_logs', 'metrics':
        'appenlight.scripts.reindex_elasticsearch:reindex_metrics',
        'slow_calls':
        'appenlight.scripts.reindex_elasticsearch:reindex_slow_calls',
        'template': 'appenlight.scripts.reindex_elasticsearch:update_template'}
    for k, v in env['registry'].appenlight_plugins.items():
        if v.get('fulltext_indexer'):
            choices[k] = v['fulltext_indexer']
    parser.add_argument('-t', '--types', nargs='*', choices=['all'] + list(
        choices.keys()), default=[], help=
        'Which parts of database should get reindexed')
    parser.add_argument('-c', '--config', required=True, help=
        'Configuration ini file of application')
    args = parser.parse_args()
    if 'all' in args.types:
        args.types = list(choices.keys())
    print('Selected types to reindex: {}'.format(args.types))
    log.info('settings {}'.format(args.types))
    if 'template' in args.types:
        get_callable(choices['template'])()
        args.types.remove('template')
    for selected in args.types:
        get_callable(choices[selected])()


def update_template():
    try:
        Datastores.es.indices.delete_template('rcae_reports')
    except elasticsearch.exceptions.NotFoundError as e:
        log.error(e)
    try:
        Datastores.es.indices.delete_template('rcae_logs')
    except elasticsearch.exceptions.NotFoundError as e:
        log.error(e)
    try:
        Datastores.es.indices.delete_template('rcae_slow_calls')
    except elasticsearch.exceptions.NotFoundError as e:
        log.error(e)
    try:
        Datastores.es.indices.delete_template('rcae_metrics')
    except elasticsearch.exceptions.NotFoundError as e:
        log.error(e)
    log.info('updating elasticsearch template')
    tag_templates = [{'values': {'path_match': 'tags.*', 'mapping': {'type':
        'object', 'properties': {'values': {'type': 'text', 'analyzer':
        'tag_value', 'fields': {'keyword': {'type': 'keyword',
        'ignore_above': 256}}}, 'numeric_values': {'type': 'float'}}}}}]
    shared_analysis = {'analyzer': {'url_path': {'type': 'custom',
        'char_filter': [], 'tokenizer': 'path_hierarchy', 'filter': []},
        'tag_value': {'type': 'custom', 'char_filter': [], 'tokenizer':
        'keyword', 'filter': ['lowercase']}}}
    shared_log_mapping = {'_all': {'enabled': False}, 'dynamic_templates':
        tag_templates, 'properties': {'pg_id': {'type': 'keyword', 'index':
        True}, 'delete_hash': {'type': 'keyword', 'index': True},
        'resource_id': {'type': 'integer'}, 'timestamp': {'type': 'date'},
        'permanent': {'type': 'boolean'}, 'request_id': {'type': 'keyword',
        'index': True}, 'log_level': {'type': 'text', 'analyzer': 'simple'},
        'message': {'type': 'text', 'analyzer': 'simple'}, 'namespace': {
        'type': 'text', 'fields': {'keyword': {'type': 'keyword',
        'ignore_above': 256}}}, 'tags': {'type': 'object'}, 'tag_list': {
        'type': 'text', 'analyzer': 'tag_value', 'fields': {'keyword': {
        'type': 'keyword', 'ignore_above': 256}}}}}
    report_schema = {'template': 'rcae_r_*', 'settings': {'index': {
        'refresh_interval': '5s', 'translog': {'sync_interval': '5s',
        'durability': 'async'}}, 'number_of_shards': 5, 'analysis':
        shared_analysis}, 'mappings': {'report': {'_all': {'enabled': False
        }, 'dynamic_templates': tag_templates, 'properties': {'type': {
        'type': 'keyword', 'index': True}, 'group_id': {'type': 'keyword',
        'index': True}, 'resource_id': {'type': 'integer'}, 'priority': {
        'type': 'integer'}, 'error': {'type': 'text', 'analyzer': 'simple'},
        'read': {'type': 'boolean'}, 'occurences': {'type': 'integer'},
        'fixed': {'type': 'boolean'}, 'first_timestamp': {'type': 'date'},
        'last_timestamp': {'type': 'date'}, 'average_duration': {'type':
        'float'}, 'summed_duration': {'type': 'float'}, 'public': {'type':
        'boolean'}, 'report_id': {'type': 'keyword', 'index': True},
        'http_status': {'type': 'integer'}, 'ip': {'type': 'keyword',
        'index': True}, 'url_domain': {'type': 'text', 'analyzer': 'simple'
        }, 'url_path': {'type': 'text', 'analyzer': 'url_path'},
        'report_type': {'type': 'integer'}, 'start_time': {'type': 'date'},
        'request_id': {'type': 'keyword', 'index': True}, 'end_time': {
        'type': 'date'}, 'duration': {'type': 'float'}, 'tags': {'type':
        'object'}, 'tag_list': {'type': 'text', 'analyzer': 'tag_value',
        'fields': {'keyword': {'type': 'keyword', 'ignore_above': 256}}},
        'extra': {'type': 'object'}, 'report_stat_id': {'type': 'keyword',
        'index': True}, 'timestamp': {'type': 'date'}, 'permanent': {'type':
        'boolean'}, 'log_level': {'type': 'text', 'analyzer': 'simple'},
        'message': {'type': 'text', 'analyzer': 'simple'}, 'namespace': {
        'type': 'text', 'fields': {'keyword': {'type': 'keyword',
        'ignore_above': 256}}}, 'join_field': {'type': 'join', 'relations':
        {'report_group': ['report', 'report_stat']}}}}}}
    Datastores.es.indices.put_template('rcae_reports', body=report_schema)
    logs_mapping = copy.deepcopy(shared_log_mapping)
    logs_mapping['properties']['log_id'] = logs_mapping['properties']['pg_id']
    del logs_mapping['properties']['pg_id']
    log_template = {'template': 'rcae_l_*', 'settings': {'index': {
        'refresh_interval': '5s', 'translog': {'sync_interval': '5s',
        'durability': 'async'}}, 'number_of_shards': 5, 'analysis':
        shared_analysis}, 'mappings': {'log': logs_mapping}}
    Datastores.es.indices.put_template('rcae_logs', body=log_template)
    slow_call_mapping = copy.deepcopy(shared_log_mapping)
    slow_call_mapping['properties']['slow_call_id'] = slow_call_mapping[
        'properties']['pg_id']
    del slow_call_mapping['properties']['pg_id']
    slow_call_template = {'template': 'rcae_sc_*', 'settings': {'index': {
        'refresh_interval': '5s', 'translog': {'sync_interval': '5s',
        'durability': 'async'}}, 'number_of_shards': 5, 'analysis':
        shared_analysis}, 'mappings': {'log': slow_call_mapping}}
    Datastores.es.indices.put_template('rcae_slow_calls', body=
        slow_call_template)
    metric_mapping = copy.deepcopy(shared_log_mapping)
    metric_mapping['properties']['metric_id'] = metric_mapping['properties'][
        'pg_id']
    del metric_mapping['properties']['pg_id']
    metrics_template = {'template': 'rcae_m_*', 'settings': {'index': {
        'refresh_interval': '5s', 'translog': {'sync_interval': '5s',
        'durability': 'async'}}, 'number_of_shards': 5, 'analysis':
        shared_analysis}, 'mappings': {'log': metric_mapping}}
    Datastores.es.indices.put_template('rcae_metrics', body=metrics_template)
    uptime_metric_mapping = copy.deepcopy(shared_log_mapping)
    uptime_metric_mapping['properties']['uptime_id'] = uptime_metric_mapping[
        'properties']['pg_id']
    del uptime_metric_mapping['properties']['pg_id']
    uptime_metrics_template = {'template': 'rcae_uptime_ce_*', 'settings':
        {'index': {'refresh_interval': '5s', 'translog': {'sync_interval':
        '5s', 'durability': 'async'}}, 'number_of_shards': 5, 'analysis':
        shared_analysis}, 'mappings': {'log': shared_log_mapping}}
    Datastores.es.indices.put_template('rcae_uptime_metrics', body=
        uptime_metrics_template)


def reindex_reports():
    reports_groups_tables = detect_tables('reports_groups_p_')
    try:
        Datastores.es.indices.delete('`rcae_r_*')
    except elasticsearch.exceptions.NotFoundError as e:
        log.error(e)
    log.info('reindexing report groups')
    i = 0
    task_start = datetime.datetime.now()
    for partition_table in reports_groups_tables:
        conn = DBSession.connection().execution_options(stream_results=True)
        result = conn.execute(partition_table.select())
        while True:
            chunk = result.fetchmany(2000)
            if not chunk:
                break
            es_docs = defaultdict(list)
            for row in chunk:
                i += 1
                item = ReportGroup(**dict(list(row.items())))
                d_range = item.partition_id
                es_docs[d_range].append(item.es_doc())
            if es_docs:
                name = partition_table.name
                log.info('round {}, {}'.format(i, name))
                for k, v in es_docs.items():
                    to_update = {'_index': k, '_type': 'report'}
                    [i.update(to_update) for i in v]
                    elasticsearch.helpers.bulk(Datastores.es, v)
    log.info('total docs {} {}'.format(i, datetime.datetime.now() - task_start)
        )
    i = 0
    log.info('reindexing reports')
    task_start = datetime.datetime.now()
    reports_tables = detect_tables('reports_p_')
    for partition_table in reports_tables:
        conn = DBSession.connection().execution_options(stream_results=True)
        result = conn.execute(partition_table.select())
        while True:
            chunk = result.fetchmany(2000)
            if not chunk:
                break
            es_docs = defaultdict(list)
            for row in chunk:
                i += 1
                item = Report(**dict(list(row.items())))
                d_range = item.partition_id
                es_docs[d_range].append(item.es_doc())
            if es_docs:
                name = partition_table.name
                log.info('round {}, {}'.format(i, name))
                for k, v in es_docs.items():
                    to_update = {'_index': k, '_type': 'report'}
                    [i.update(to_update) for i in v]
                    elasticsearch.helpers.bulk(Datastores.es, v)
    log.info('total docs {} {}'.format(i, datetime.datetime.now() - task_start)
        )
    log.info('reindexing reports stats')
    i = 0
    task_start = datetime.datetime.now()
    reports_stats_tables = detect_tables('reports_stats_p_')
    for partition_table in reports_stats_tables:
        conn = DBSession.connection().execution_options(stream_results=True)
        result = conn.execute(partition_table.select())
        while True:
            chunk = result.fetchmany(2000)
            if not chunk:
                break
            es_docs = defaultdict(list)
            for row in chunk:
                rd = dict(list(row.items()))
                rd.pop('size', None)
                item = ReportStat(**rd)
                i += 1
                d_range = item.partition_id
                es_docs[d_range].append(item.es_doc())
            if es_docs:
                name = partition_table.name
                log.info('round  {}, {}'.format(i, name))
                for k, v in es_docs.items():
                    to_update = {'_index': k, '_type': 'report'}
                    [i.update(to_update) for i in v]
                    elasticsearch.helpers.bulk(Datastores.es, v)
    log.info('total docs {} {}'.format(i, datetime.datetime.now() - task_start)
        )


def reindex_logs():
    try:
        Datastores.es.indices.delete('rcae_l_*')
    except elasticsearch.exceptions.NotFoundError as e:
        log.error(e)
    log.info('reindexing logs')
    i = 0
    task_start = datetime.datetime.now()
    log_tables = detect_tables('logs_p_')
    for partition_table in log_tables:
        conn = DBSession.connection().execution_options(stream_results=True)
        result = conn.execute(partition_table.select())
        while True:
            chunk = result.fetchmany(2000)
            if not chunk:
                break
            es_docs = defaultdict(list)
            for row in chunk:
                i += 1
                item = Log(**dict(list(row.items())))
                d_range = item.partition_id
                es_docs[d_range].append(item.es_doc())
            if es_docs:
                name = partition_table.name
                log.info('round  {}, {}'.format(i, name))
                for k, v in es_docs.items():
                    to_update = {'_index': k, '_type': 'log'}
                    [i.update(to_update) for i in v]
                    elasticsearch.helpers.bulk(Datastores.es, v)
    log.info('total docs {} {}'.format(i, datetime.datetime.now() - task_start)
        )


def reindex_metrics():
    try:
        Datastores.es.indices.delete('rcae_m_*')
    except elasticsearch.exceptions.NotFoundError as e:
        log.error(e)
    log.info('reindexing applications metrics')
    i = 0
    task_start = datetime.datetime.now()
    metric_tables = detect_tables('metrics_p_')
    for partition_table in metric_tables:
        conn = DBSession.connection().execution_options(stream_results=True)
        result = conn.execute(partition_table.select())
        while True:
            chunk = result.fetchmany(2000)
            if not chunk:
                break
            es_docs = defaultdict(list)
            for row in chunk:
                i += 1
                item = Metric(**dict(list(row.items())))
                d_range = item.partition_id
                es_docs[d_range].append(item.es_doc())
            if es_docs:
                name = partition_table.name
                log.info('round  {}, {}'.format(i, name))
                for k, v in es_docs.items():
                    to_update = {'_index': k, '_type': 'log'}
                    [i.update(to_update) for i in v]
                    elasticsearch.helpers.bulk(Datastores.es, v)
    log.info('total docs {} {}'.format(i, datetime.datetime.now() - task_start)
        )


def reindex_slow_calls():
    try:
        Datastores.es.indices.delete('rcae_sc_*')
    except elasticsearch.exceptions.NotFoundError as e:
        log.error(e)
    log.info('reindexing slow calls')
    i = 0
    task_start = datetime.datetime.now()
    slow_calls_tables = detect_tables('slow_calls_p_')
    for partition_table in slow_calls_tables:
        conn = DBSession.connection().execution_options(stream_results=True)
        result = conn.execute(partition_table.select())
        while True:
            chunk = result.fetchmany(2000)
            if not chunk:
                break
            es_docs = defaultdict(list)
            for row in chunk:
                i += 1
                item = SlowCall(**dict(list(row.items())))
                d_range = item.partition_id
                es_docs[d_range].append(item.es_doc())
            if es_docs:
                name = partition_table.name
                log.info('round  {}, {}'.format(i, name))
                for k, v in es_docs.items():
                    to_update = {'_index': k, '_type': 'log'}
                    [i.update(to_update) for i in v]
                    elasticsearch.helpers.bulk(Datastores.es, v)
    log.info('total docs {} {}'.format(i, datetime.datetime.now() - task_start)
        )


if __name__ == '__main__':
    main()

# -*- coding: utf-8 -*-

# Copyright 2010 - 2017 RhodeCode GmbH and the AppEnlight project authors
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
#   http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.

import argparse
import datetime
import logging
import copy

import sqlalchemy as sa
import elasticsearch.exceptions
import elasticsearch.helpers

from collections import defaultdict
from pyramid.paster import setup_logging
from pyramid.paster import bootstrap
from appenlight.models import DBSession, Datastores, metadata
from appenlight.lib import get_callable
from appenlight.models.report_group import ReportGroup
from appenlight.models.report import Report
from appenlight.models.report_stat import ReportStat
from appenlight.models.log import Log
from appenlight.models.slow_call import SlowCall
from appenlight.models.metric import Metric

log = logging.getLogger(__name__)

tables = {
    "slow_calls_p_": [],
    "reports_stats_p_": [],
    "reports_p_": [],
    "reports_groups_p_": [],
    "logs_p_": [],
    "metrics_p_": [],
}


def detect_tables(table_prefix):
    found_tables = []
    db_tables_query = """
    SELECT tablename FROM pg_tables WHERE tablename NOT LIKE 'pg_%' AND
    tablename NOT LIKE 'sql_%' ORDER BY tablename ASC;"""

    for table in DBSession.execute(db_tables_query).fetchall():
        tablename = table.tablename
        if tablename.startswith(table_prefix):
            t = sa.Table(
                tablename, metadata, autoload=True, autoload_with=DBSession.bind.engine
            )
            found_tables.append(t)
    return found_tables


def main():
    """
    Recreates Elasticsearch indexes
    Performs reindex of whole db to Elasticsearch

    """

    # need parser twice because we first need to load ini file
    # bootstrap pyramid and then load plugins
    pre_parser = argparse.ArgumentParser(
        description="Reindex AppEnlight data", add_help=False
    )
    pre_parser.add_argument(
        "-c", "--config", required=True, help="Configuration ini file of application"
    )
    pre_parser.add_argument("-h", "--help", help="Show help", nargs="?")
    pre_parser.add_argument(
        "-t", "--types", nargs="+", help="Which parts of database should get reindexed"
    )
    args = pre_parser.parse_args()

    config_uri = args.config
    setup_logging(config_uri)
    log.setLevel(logging.INFO)
    env = bootstrap(config_uri)
    parser = argparse.ArgumentParser(description="Reindex AppEnlight data")
    choices = {
        "reports": "appenlight.scripts.reindex_elasticsearch:reindex_reports",
        "logs": "appenlight.scripts.reindex_elasticsearch:reindex_logs",
        "metrics": "appenlight.scripts.reindex_elasticsearch:reindex_metrics",
        "slow_calls": "appenlight.scripts.reindex_elasticsearch:reindex_slow_calls",
        "template": "appenlight.scripts.reindex_elasticsearch:update_template",
    }
    for k, v in env["registry"].appenlight_plugins.items():
        if v.get("fulltext_indexer"):
            choices[k] = v["fulltext_indexer"]
    parser.add_argument(
        "-t",
        "--types",
        nargs="*",
        choices=["all"] + list(choices.keys()),
        default=[],
        help="Which parts of database should get reindexed",
    )
    parser.add_argument(
        "-c", "--config", required=True, help="Configuration ini file of application"
    )
    args = parser.parse_args()

    if "all" in args.types:
        args.types = list(choices.keys())

    print("Selected types to reindex: {}".format(args.types))

    log.info("settings {}".format(args.types))

    if "template" in args.types:
        get_callable(choices["template"])()
        args.types.remove("template")
    for selected in args.types:
        get_callable(choices[selected])()


def update_template():
    try:
        Datastores.es.indices.delete_template("rcae_reports")
    except elasticsearch.exceptions.NotFoundError as e:
        log.error(e)

    try:
        Datastores.es.indices.delete_template("rcae_logs")
    except elasticsearch.exceptions.NotFoundError as e:
        log.error(e)
    try:
        Datastores.es.indices.delete_template("rcae_slow_calls")
    except elasticsearch.exceptions.NotFoundError as e:
        log.error(e)
    try:
        Datastores.es.indices.delete_template("rcae_metrics")
    except elasticsearch.exceptions.NotFoundError as e:
        log.error(e)
    log.info("updating elasticsearch template")
    tag_templates = [
        {
            "values": {
                "path_match": "tags.*",
                "mapping": {
                    "type": "object",
                    "properties": {
                        "values": {
                            "type": "text",
                            "analyzer": "tag_value",
                            "fields": {
                                "keyword": {"type": "keyword", "ignore_above": 256}
                            },
                        },
                        "numeric_values": {"type": "float"},
                    },
                },
            }
        }
    ]

    shared_analysis = {
        "analyzer": {
            "url_path": {
                "type": "custom",
                "char_filter": [],
                "tokenizer": "path_hierarchy",
                "filter": [],
            },
            "tag_value": {
                "type": "custom",
                "char_filter": [],
                "tokenizer": "keyword",
                "filter": ["lowercase"],
            },
        }
    }

    shared_log_mapping = {
        "_all": {"enabled": False},
        "dynamic_templates": tag_templates,
        "properties": {
            "pg_id": {"type": "keyword", "index": True},
            "delete_hash": {"type": "keyword", "index": True},
            "resource_id": {"type": "integer"},
            "timestamp": {"type": "date"},
            "permanent": {"type": "boolean"},
            "request_id": {"type": "keyword", "index": True},
            "log_level": {"type": "text", "analyzer": "simple"},
            "message": {"type": "text", "analyzer": "simple"},
            "namespace": {
                "type": "text",
                "fields": {"keyword": {"type": "keyword", "ignore_above": 256}},
            },
            "tags": {"type": "object"},
            "tag_list": {
                "type": "text",
                "analyzer": "tag_value",
                "fields": {"keyword": {"type": "keyword", "ignore_above": 256}},
            },
        },
    }

    report_schema = {
        "template": "rcae_r_*",
        "settings": {
            "index": {
                "refresh_interval": "5s",
                "translog": {"sync_interval": "5s", "durability": "async"},
            },
            "number_of_shards": 5,
            "analysis": shared_analysis,
        },
        "mappings": {
            "report": {
                "_all": {"enabled": False},
                "dynamic_templates": tag_templates,
                "properties": {
                    "type": {"type": "keyword", "index": True},
                    # report group
                    "group_id": {"type": "keyword", "index": True},
                    "resource_id": {"type": "integer"},
                    "priority": {"type": "integer"},
                    "error": {"type": "text", "analyzer": "simple"},
                    "read": {"type": "boolean"},
                    "occurences": {"type": "integer"},
                    "fixed": {"type": "boolean"},
                    "first_timestamp": {"type": "date"},
                    "last_timestamp": {"type": "date"},
                    "average_duration": {"type": "float"},
                    "summed_duration": {"type": "float"},
                    "public": {"type": "boolean"},
                    # report
                    "report_id": {"type": "keyword", "index": True},
                    "http_status": {"type": "integer"},
                    "ip": {"type": "keyword", "index": True},
                    "url_domain": {"type": "text", "analyzer": "simple"},
                    "url_path": {"type": "text", "analyzer": "url_path"},
                    "report_type": {"type": "integer"},
                    "start_time": {"type": "date"},
                    "request_id": {"type": "keyword", "index": True},
                    "end_time": {"type": "date"},
                    "duration": {"type": "float"},
                    "tags": {"type": "object"},
                    "tag_list": {
                        "type": "text",
                        "analyzer": "tag_value",
                        "fields": {"keyword": {"type": "keyword", "ignore_above": 256}},
                    },
                    "extra": {"type": "object"},
                    # report stats
                    "report_stat_id": {"type": "keyword", "index": True},
                    "timestamp": {"type": "date"},
                    "permanent": {"type": "boolean"},
                    "log_level": {"type": "text", "analyzer": "simple"},
                    "message": {"type": "text", "analyzer": "simple"},
                    "namespace": {
                        "type": "text",
                        "fields": {"keyword": {"type": "keyword", "ignore_above": 256}},
                    },
                    "join_field": {
                        "type": "join",
                        "relations": {"report_group": ["report", "report_stat"]},
                    },
                },
            }
        },
    }

    Datastores.es.indices.put_template("rcae_reports", body=report_schema)

    logs_mapping = copy.deepcopy(shared_log_mapping)
    logs_mapping["properties"]["log_id"] = logs_mapping["properties"]["pg_id"]
    del logs_mapping["properties"]["pg_id"]

    log_template = {
        "template": "rcae_l_*",
        "settings": {
            "index": {
                "refresh_interval": "5s",
                "translog": {"sync_interval": "5s", "durability": "async"},
            },
            "number_of_shards": 5,
            "analysis": shared_analysis,
        },
        "mappings": {"log": logs_mapping},
    }

    Datastores.es.indices.put_template("rcae_logs", body=log_template)

    slow_call_mapping = copy.deepcopy(shared_log_mapping)
    slow_call_mapping["properties"]["slow_call_id"] = slow_call_mapping["properties"][
        "pg_id"
    ]
    del slow_call_mapping["properties"]["pg_id"]

    slow_call_template = {
        "template": "rcae_sc_*",
        "settings": {
            "index": {
                "refresh_interval": "5s",
                "translog": {"sync_interval": "5s", "durability": "async"},
            },
            "number_of_shards": 5,
            "analysis": shared_analysis,
        },
        "mappings": {"log": slow_call_mapping},
    }

    Datastores.es.indices.put_template("rcae_slow_calls", body=slow_call_template)

    metric_mapping = copy.deepcopy(shared_log_mapping)
    metric_mapping["properties"]["metric_id"] = metric_mapping["properties"]["pg_id"]
    del metric_mapping["properties"]["pg_id"]

    metrics_template = {
        "template": "rcae_m_*",
        "settings": {
            "index": {
                "refresh_interval": "5s",
                "translog": {"sync_interval": "5s", "durability": "async"},
            },
            "number_of_shards": 5,
            "analysis": shared_analysis,
        },
        "mappings": {"log": metric_mapping},
    }

    Datastores.es.indices.put_template("rcae_metrics", body=metrics_template)

    uptime_metric_mapping = copy.deepcopy(shared_log_mapping)
    uptime_metric_mapping["properties"]["uptime_id"] = uptime_metric_mapping[
        "properties"
    ]["pg_id"]
    del uptime_metric_mapping["properties"]["pg_id"]

    uptime_metrics_template = {
        "template": "rcae_uptime_ce_*",
        "settings": {
            "index": {
                "refresh_interval": "5s",
                "translog": {"sync_interval": "5s", "durability": "async"},
            },
            "number_of_shards": 5,
            "analysis": shared_analysis,
        },
        "mappings": {"log": shared_log_mapping},
    }

    Datastores.es.indices.put_template(
        "rcae_uptime_metrics", body=uptime_metrics_template
    )


def reindex_reports():
    reports_groups_tables = detect_tables("reports_groups_p_")
    try:
        Datastores.es.indices.delete("`rcae_r_*")
    except elasticsearch.exceptions.NotFoundError as e:
        log.error(e)

    log.info("reindexing report groups")
    i = 0
    task_start = datetime.datetime.now()
    for partition_table in reports_groups_tables:
        conn = DBSession.connection().execution_options(stream_results=True)
        result = conn.execute(partition_table.select())
        while True:
            chunk = result.fetchmany(2000)
            if not chunk:
                break
            es_docs = defaultdict(list)
            for row in chunk:
                i += 1
                item = ReportGroup(**dict(list(row.items())))
                d_range = item.partition_id
                es_docs[d_range].append(item.es_doc())
            if es_docs:
                name = partition_table.name
                log.info("round {}, {}".format(i, name))
                for k, v in es_docs.items():
                    to_update = {"_index": k, "_type": "report"}
                    [i.update(to_update) for i in v]
                    elasticsearch.helpers.bulk(Datastores.es, v)

    log.info("total docs {} {}".format(i, datetime.datetime.now() - task_start))

    i = 0
    log.info("reindexing reports")
    task_start = datetime.datetime.now()
    reports_tables = detect_tables("reports_p_")
    for partition_table in reports_tables:
        conn = DBSession.connection().execution_options(stream_results=True)
        result = conn.execute(partition_table.select())
        while True:
            chunk = result.fetchmany(2000)
            if not chunk:
                break
            es_docs = defaultdict(list)
            for row in chunk:
                i += 1
                item = Report(**dict(list(row.items())))
                d_range = item.partition_id
                es_docs[d_range].append(item.es_doc())
            if es_docs:
                name = partition_table.name
                log.info("round {}, {}".format(i, name))
                for k, v in es_docs.items():
                    to_update = {"_index": k, "_type": "report"}
                    [i.update(to_update) for i in v]
                    elasticsearch.helpers.bulk(Datastores.es, v)

    log.info("total docs {} {}".format(i, datetime.datetime.now() - task_start))

    log.info("reindexing reports stats")
    i = 0
    task_start = datetime.datetime.now()
    reports_stats_tables = detect_tables("reports_stats_p_")
    for partition_table in reports_stats_tables:
        conn = DBSession.connection().execution_options(stream_results=True)
        result = conn.execute(partition_table.select())
        while True:
            chunk = result.fetchmany(2000)
            if not chunk:
                break
            es_docs = defaultdict(list)
            for row in chunk:
                rd = dict(list(row.items()))
                # remove legacy columns
                # TODO: remove the column later
                rd.pop("size", None)
                item = ReportStat(**rd)
                i += 1
                d_range = item.partition_id
                es_docs[d_range].append(item.es_doc())
            if es_docs:
                name = partition_table.name
                log.info("round  {}, {}".format(i, name))
                for k, v in es_docs.items():
                    to_update = {"_index": k, "_type": "report"}
                    [i.update(to_update) for i in v]
                    elasticsearch.helpers.bulk(Datastores.es, v)

    log.info("total docs {} {}".format(i, datetime.datetime.now() - task_start))


def reindex_logs():
    try:
        Datastores.es.indices.delete("rcae_l_*")
    except elasticsearch.exceptions.NotFoundError as e:
        log.error(e)

    # logs
    log.info("reindexing logs")
    i = 0
    task_start = datetime.datetime.now()
    log_tables = detect_tables("logs_p_")
    for partition_table in log_tables:
        conn = DBSession.connection().execution_options(stream_results=True)
        result = conn.execute(partition_table.select())
        while True:
            chunk = result.fetchmany(2000)
            if not chunk:
                break
            es_docs = defaultdict(list)

            for row in chunk:
                i += 1
                item = Log(**dict(list(row.items())))
                d_range = item.partition_id
                es_docs[d_range].append(item.es_doc())
            if es_docs:
                name = partition_table.name
                log.info("round  {}, {}".format(i, name))
                for k, v in es_docs.items():
                    to_update = {"_index": k, "_type": "log"}
                    [i.update(to_update) for i in v]
                    elasticsearch.helpers.bulk(Datastores.es, v)

    log.info("total docs {} {}".format(i, datetime.datetime.now() - task_start))


def reindex_metrics():
    try:
        Datastores.es.indices.delete("rcae_m_*")
    except elasticsearch.exceptions.NotFoundError as e:
        log.error(e)

    log.info("reindexing applications metrics")
    i = 0
    task_start = datetime.datetime.now()
    metric_tables = detect_tables("metrics_p_")
    for partition_table in metric_tables:
        conn = DBSession.connection().execution_options(stream_results=True)
        result = conn.execute(partition_table.select())
        while True:
            chunk = result.fetchmany(2000)
            if not chunk:
                break
            es_docs = defaultdict(list)
            for row in chunk:
                i += 1
                item = Metric(**dict(list(row.items())))
                d_range = item.partition_id
                es_docs[d_range].append(item.es_doc())
            if es_docs:
                name = partition_table.name
                log.info("round  {}, {}".format(i, name))
                for k, v in es_docs.items():
                    to_update = {"_index": k, "_type": "log"}
                    [i.update(to_update) for i in v]
                    elasticsearch.helpers.bulk(Datastores.es, v)

    log.info("total docs {} {}".format(i, datetime.datetime.now() - task_start))


def reindex_slow_calls():
    try:
        Datastores.es.indices.delete("rcae_sc_*")
    except elasticsearch.exceptions.NotFoundError as e:
        log.error(e)

    log.info("reindexing slow calls")
    i = 0
    task_start = datetime.datetime.now()
    slow_calls_tables = detect_tables("slow_calls_p_")
    for partition_table in slow_calls_tables:
        conn = DBSession.connection().execution_options(stream_results=True)
        result = conn.execute(partition_table.select())
        while True:
            chunk = result.fetchmany(2000)
            if not chunk:
                break
            es_docs = defaultdict(list)
            for row in chunk:
                i += 1
                item = SlowCall(**dict(list(row.items())))
                d_range = item.partition_id
                es_docs[d_range].append(item.es_doc())
            if es_docs:
                name = partition_table.name
                log.info("round  {}, {}".format(i, name))
                for k, v in es_docs.items():
                    to_update = {"_index": k, "_type": "log"}
                    [i.update(to_update) for i in v]
                    elasticsearch.helpers.bulk(Datastores.es, v)

    log.info("total docs {} {}".format(i, datetime.datetime.now() - task_start))


if __name__ == "__main__":
    main()
