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