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