elasticsearch: bump to ES 6.x

.travis.yml

@@ -1,43 +1,46 b''
 language: python
 
 dist: xenial
 
 notifications:
   on_success: change
   on_failure: always
 
 matrix:
   include:
   - python: 3.5
-    env: TOXENV=py35
+    env: TOXENV=py35 ES_VERSION=6.6.2 ES_DOWNLOAD_URL=https://artifacts.elastic.co/downloads/elasticsearch/elasticsearch-oss-${ES_VERSION}.tar.gz
   - python: 3.6
-    env: TOXENV=py36
+    env: TOXENV=py36 ES_VERSION=6.6.2 ES_DOWNLOAD_URL=https://artifacts.elastic.co/downloads/elasticsearch/elasticsearch-oss-${ES_VERSION}.tar.gz
     addons:
       postgresql: "9.6"
   - python: 3.6
-    env: TOXENV=py36 PGPORT=5432
+    env: TOXENV=py36 PGPORT=5432 ES_VERSION=6.6.2 ES_DOWNLOAD_URL=https://artifacts.elastic.co/downloads/elasticsearch/elasticsearch-oss-${ES_VERSION}.tar.gz
     addons:
       postgresql: "10"
       apt:
         packages:
         - postgresql-10
         - postgresql-client-10
 
 install:
+- wget ${ES_DOWNLOAD_URL}
+- tar -xzf elasticsearch-${ES_VERSION}.tar.gz
+- ./elasticsearch-${ES_VERSION}/bin/elasticsearch &
 - travis_retry pip install -U setuptools pip tox
 
 script:
 - travis_retry tox
 
 services:
 - postgresql
 - elasticsearch
 - redis
 
 before_script:
   - psql -c "create user test with encrypted password 'test';" -U postgres
   - psql -c 'create database appenlight_test owner test;' -U postgres
 
 after_success:
 - pip install coveralls
 - coveralls

backend/src/appenlight/scripts/reindex_elasticsearch.py

@@ -1,577 +1,572 b''
 # -*- 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"},
-                "mapping": {"single_type": True}
+                "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"},
-                "mapping": {"single_type": True}
             },
             "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"},
-                "mapping": {"single_type": True}
             },
             "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"},
-                "mapping": {"single_type": True}
             },
             "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"},
-                "mapping": {"single_type": True}
             },
             "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()