utils/test/dashboard/dashboard/mongo2elastic/main.py

   1 #! /usr/bin/env python
   2
   3 import datetime
   4 import json
   5 import os
   6 import subprocess
   7 import traceback
   8 import uuid
   9
  10 import argparse
  11
  12 from dashboard.common import elastic_access
  13 from dashboard.common import logger_utils
  14 from dashboard.conf import testcases
  15 from dashboard.conf.config import APIConfig
  16 from dashboard.mongo2elastic import format
  17
  18 logger = logger_utils.DashboardLogger('mongo2elastic').get
  19
  20 parser = argparse.ArgumentParser()
  21 parser.add_argument("-c", "--config-file",
  22                     dest='config_file',
  23                     help="Config file location")
  24 parser.add_argument('-ld', '--latest-days',
  25                     default=0,
  26                     type=int,
  27                     metavar='N',
  28                     help='get entries old at most N days from mongodb and'
  29                          ' parse those that are not already in elasticsearch.'
  30                          ' If not present, will get everything from mongodb,'
  31                          ' which is the default')
  32
  33 args = parser.parse_args()
  34 CONF = APIConfig().parse(args.config_file)
  35
  36
  37 tmp_docs_file = './mongo-{}.json'.format(uuid.uuid4())
  38
  39
  40 class DocumentVerification(object):
  41
  42     def __init__(self, doc):
  43         super(DocumentVerification, self).__init__()
  44         self.doc = doc
  45         self.doc_id = doc['_id'] if '_id' in doc else None
  46         self.skip = False
  47
  48     def mandatory_fields_exist(self):
  49         mandatory_fields = ['installer',
  50                             'pod_name',
  51                             'version',
  52                             'case_name',
  53                             'project_name',
  54                             'details',
  55                             'start_date',
  56                             'scenario']
  57         for key, value in self.doc.items():
  58             if key in mandatory_fields:
  59                 if value is None:
  60                     logger.info("Skip testcase '%s' because field "
  61                                 "'%s' missing" % (self.doc_id, key))
  62                     self.skip = True
  63                 else:
  64                     mandatory_fields.remove(key)
  65             else:
  66                 del self.doc[key]
  67
  68         if len(mandatory_fields) > 0:
  69             logger.info("Skip testcase '%s' because field(s) '%s' missing" %
  70                         (self.doc_id, mandatory_fields))
  71             self.skip = True
  72
  73         return self
  74
  75     def modify_start_date(self):
  76         field = 'start_date'
  77         if field in self.doc:
  78             self.doc[field] = self._fix_date(self.doc[field])
  79
  80         return self
  81
  82     def modify_scenario(self):
  83         scenario = 'scenario'
  84         version = 'version'
  85
  86         if (scenario not in self.doc) or \
  87                 (scenario in self.doc and self.doc[scenario] is None):
  88             self.doc[scenario] = self.doc[version]
  89
  90         return self
  91
  92     def is_skip(self):
  93         return self.skip
  94
  95     def _fix_date(self, date_string):
  96         if date_string == 'None':
  97             return None
  98         if isinstance(date_string, dict):
  99             return date_string['$date']
 100         if 'T' not in date_string:
 101             date_string = date_string[:-3].replace(' ', 'T')
 102         if not date_string.endswith('Z'):
 103             date_string += 'Z'
 104
 105         return date_string
 106
 107
 108 class DocumentPublisher(object):
 109
 110     def __init__(self, doc, fmt, exist_docs, creds, elastic_url):
 111         self.doc = doc
 112         self.fmt = fmt
 113         self.creds = creds
 114         self.exist_docs = exist_docs
 115         self.elastic_url = elastic_url
 116         self.is_formatted = True
 117
 118     def format(self):
 119         try:
 120             if self._verify_document() and self.fmt:
 121                 self.is_formatted = vars(format)[self.fmt](self.doc)
 122             else:
 123                 self.is_formatted = False
 124         except Exception:
 125             logger.error("Fail in format testcase[%s]\nerror message: %s" %
 126                          (self.doc, traceback.format_exc()))
 127             self.is_formatted = False
 128         finally:
 129             return self
 130
 131     def publish(self):
 132         if self.is_formatted and self.doc not in self.exist_docs:
 133             self._publish()
 134
 135     def _publish(self):
 136         status, data = elastic_access.publish_docs(
 137             self.elastic_url, self.creds, self.doc)
 138         if status > 300:
 139             logger.error('Publish record[{}] failed, due to [{}]'
 140                          .format(self.doc,
 141                                  json.loads(data)['error']['reason']))
 142
 143     def _fix_date(self, date_string):
 144         if isinstance(date_string, dict):
 145             return date_string['$date']
 146         else:
 147             return date_string[:-3].replace(' ', 'T') + 'Z'
 148
 149     def _verify_document(self):
 150         return not (DocumentVerification(self.doc)
 151                     .modify_start_date()
 152                     .modify_scenario()
 153                     .mandatory_fields_exist()
 154                     .is_skip())
 155
 156
 157 class DocumentsPublisher(object):
 158
 159     def __init__(self, project, case, fmt, days, elastic_url, creds):
 160         self.project = project
 161         self.case = case
 162         self.fmt = fmt
 163         self.days = days
 164         self.elastic_url = elastic_url
 165         self.creds = creds
 166         self.existed_docs = []
 167
 168     def export(self):
 169         if self.days > 0:
 170             past_time = datetime.datetime.today(
 171             ) - datetime.timedelta(days=self.days)
 172             query = '''{{
 173                           "project_name": "{}",
 174                           "case_name": "{}",
 175                           "start_date": {{"$gt" : "{}"}}
 176                         }}'''.format(self.project, self.case, past_time)
 177         else:
 178             query = '''{{
 179                            "project_name": "{}",
 180                            "case_name": "{}"
 181                         }}'''.format(self.project, self.case)
 182         cmd = ['mongoexport',
 183                '--db', 'test_results_collection',
 184                '--collection', 'results',
 185                '--query', '{}'.format(query),
 186                '--out', '{}'.format(tmp_docs_file)]
 187         try:
 188             subprocess.check_call(cmd)
 189             return self
 190         except Exception as err:
 191             logger.error("export mongodb failed: %s" % err)
 192             self._remove()
 193             exit(-1)
 194
 195     def get_exists(self):
 196         if self.days == 0:
 197             body = '''{{
 198                         "query": {{
 199                             "bool": {{
 200                                 "must": [
 201                                     {{ "match": {{ "project_name": "{}" }} }},
 202                                     {{ "match": {{ "case_name": "{}" }} }}
 203                                 ]
 204                             }}
 205                         }}
 206                     }}'''.format(self.project, self.case)
 207         elif self.days > 0:
 208             body = '''{{
 209                        "query": {{
 210                            "bool": {{
 211                                "must": [
 212                                    {{ "match": {{ "project_name": "{}" }} }},
 213                                    {{ "match": {{ "case_name": "{}" }} }}
 214                                ],
 215                                "filter": {{
 216                                    "range": {{
 217                                        "start_date": {{ "gte": "now-{}d" }}
 218                                    }}
 219                                }}
 220                            }}
 221                        }}
 222                    }}'''.format(self.project, self.case, self.days)
 223         else:
 224             raise Exception('Update days must be non-negative')
 225         self.existed_docs = elastic_access.get_docs(
 226             self.elastic_url, self.creds, body)
 227         return self
 228
 229     def publish(self):
 230         fdocs = None
 231         try:
 232             with open(tmp_docs_file) as fdocs:
 233                 for doc_line in fdocs:
 234                     DocumentPublisher(json.loads(doc_line),
 235                                       self.fmt,
 236                                       self.existed_docs,
 237                                       self.creds,
 238                                       self.elastic_url).format().publish()
 239         finally:
 240             if fdocs:
 241                 fdocs.close()
 242             self._remove()
 243
 244     def _remove(self):
 245         if os.path.exists(tmp_docs_file):
 246             os.remove(tmp_docs_file)
 247
 248
 249 def main():
 250     for project, case_dicts in testcases.testcases_yaml.items():
 251         for case_dict in case_dicts:
 252             case = case_dict.get('name')
 253             fmt = testcases.compose_format(case_dict.get('format'))
 254             DocumentsPublisher(project,
 255                                case,
 256                                fmt,
 257                                args.latest_days,
 258                                CONF.index_url,
 259                                CONF.es_creds).export().get_exists().publish()