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 common import logger_utils, elastic_access
  14 from conf import testcases
  15 from conf.config import APIConfig
  16 from 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 DocumentPublisher:
  40
  41     def __init__(self, doc, fmt, exist_docs, creds, to):
  42         self.doc = doc
  43         self.fmt = fmt
  44         self.creds = creds
  45         self.exist_docs = exist_docs
  46         self.to = to
  47         self.is_formatted = True
  48
  49     def format(self):
  50         try:
  51             if self._verify_document() and self.fmt:
  52                 self.is_formatted = vars(format)[self.fmt](self.doc)
  53             else:
  54                 self.is_formatted = False
  55         except Exception:
  56             logger.error("Fail in format testcase[%s]\nerror message: %s" %
  57                          (self.doc, traceback.format_exc()))
  58             self.is_formatted = False
  59         finally:
  60             return self
  61
  62     def publish(self):
  63         if self.is_formatted and self.doc not in self.exist_docs:
  64             self._publish()
  65
  66     def _publish(self):
  67         status, data = elastic_access.publish_json(self.doc, self.creds, self.to)
  68         if status > 300:
  69             logger.error('Publish record[{}] failed, due to [{}]'
  70                          .format(self.doc, json.loads(data)['error']['reason']))
  71
  72     def _fix_date(self, date_string):
  73         if isinstance(date_string, dict):
  74             return date_string['$date']
  75         else:
  76             return date_string[:-3].replace(' ', 'T') + 'Z'
  77
  78     def _verify_document(self):
  79         """
  80         Mandatory fields:
  81             installer
  82             pod_name
  83             version
  84             case_name
  85             date
  86             project
  87             details
  88
  89             these fields must be present and must NOT be None
  90
  91         Optional fields:
  92             description
  93
  94             these fields will be preserved if the are NOT None
  95         """
  96         mandatory_fields = ['installer',
  97                             'pod_name',
  98                             'version',
  99                             'case_name',
 100                             'project_name',
 101                             'details']
 102         mandatory_fields_to_modify = {'start_date': self._fix_date}
 103         fields_to_swap_or_add = {'scenario': 'version'}
 104         if '_id' in self.doc:
 105             mongo_id = self.doc['_id']
 106         else:
 107             mongo_id = None
 108         optional_fields = ['description']
 109         for key, value in self.doc.items():
 110             if key in mandatory_fields:
 111                 if value is None:
 112                     # empty mandatory field, invalid input
 113                     logger.info("Skipping testcase with mongo _id '{}' because the testcase was missing value"
 114                                 " for mandatory field '{}'".format(mongo_id, key))
 115                     return False
 116                 else:
 117                     mandatory_fields.remove(key)
 118             elif key in mandatory_fields_to_modify:
 119                 if value is None:
 120                     # empty mandatory field, invalid input
 121                     logger.info("Skipping testcase with mongo _id '{}' because the testcase was missing value"
 122                                 " for mandatory field '{}'".format(mongo_id, key))
 123                     return False
 124                 else:
 125                     self.doc[key] = mandatory_fields_to_modify[key](value)
 126                     del mandatory_fields_to_modify[key]
 127             elif key in fields_to_swap_or_add:
 128                 if value is None:
 129                     swapped_key = fields_to_swap_or_add[key]
 130                     swapped_value = self.doc[swapped_key]
 131                     logger.info("Swapping field '{}' with value None for '{}' with value '{}'.".format(key, swapped_key,
 132                                                                                                        swapped_value))
 133                     self.doc[key] = swapped_value
 134                     del fields_to_swap_or_add[key]
 135                 else:
 136                     del fields_to_swap_or_add[key]
 137             elif key in optional_fields:
 138                 if value is None:
 139                     # empty optional field, remove
 140                     del self.doc[key]
 141                 optional_fields.remove(key)
 142             else:
 143                 # unknown field
 144                 del self.doc[key]
 145
 146         if len(mandatory_fields) > 0:
 147             # some mandatory fields are missing
 148             logger.info("Skipping testcase with mongo _id '{}' because the testcase was missing"
 149                         " mandatory field(s) '{}'".format(mongo_id, mandatory_fields))
 150             return False
 151         elif len(mandatory_fields_to_modify) > 0:
 152             # some mandatory fields are missing
 153             logger.info("Skipping testcase with mongo _id '{}' because the testcase was missing"
 154                         " mandatory field(s) '{}'".format(mongo_id, mandatory_fields_to_modify.keys()))
 155             return False
 156         else:
 157             if len(fields_to_swap_or_add) > 0:
 158                 for key, swap_key in fields_to_swap_or_add.iteritems():
 159                     self.doc[key] = self.doc[swap_key]
 160
 161             return True
 162
 163
 164 class DocumentsPublisher:
 165
 166     def __init__(self, project, case, fmt, days, elastic_url, creds, to):
 167         self.project = project
 168         self.case = case
 169         self.fmt = fmt
 170         self.days = days
 171         self.elastic_url = elastic_url
 172         self.creds = creds
 173         self.to = to
 174         self.existed_docs = []
 175
 176     def export(self):
 177         if self.days > 0:
 178             past_time = datetime.datetime.today() - datetime.timedelta(days=self.days)
 179             query = '''{{
 180                           "project_name": "{}",
 181                           "case_name": "{}",
 182                           "start_date": {{"$gt" : "{}"}}
 183                         }}'''.format(self.project, self.case, past_time)
 184         else:
 185             query = '''{{
 186                            "project_name": "{}",
 187                            "case_name": "{}"
 188                         }}'''.format(self.project, self.case)
 189         cmd = ['mongoexport',
 190                '--db', 'test_results_collection',
 191                '--collection', 'results',
 192                '--query', '{}'.format(query),
 193                '--out', '{}'.format(tmp_docs_file)]
 194         try:
 195             subprocess.check_call(cmd)
 196             return self
 197         except Exception, err:
 198             logger.error("export mongodb failed: %s" % err)
 199             self._remove()
 200             exit(-1)
 201
 202     def get_existed_docs(self):
 203         self.existed_docs = elastic_access.get_elastic_docs_by_days(self.elastic_url, self.creds, self.days)
 204         return self
 205
 206     def publish(self):
 207         try:
 208             with open(tmp_docs_file) as fdocs:
 209                 for doc_line in fdocs:
 210                     DocumentPublisher(json.loads(doc_line),
 211                                       self.fmt,
 212                                       self.existed_docs,
 213                                       self.creds,
 214                                       self.to).format().publish()
 215         finally:
 216             fdocs.close()
 217             self._remove()
 218
 219     def _remove(self):
 220         if os.path.exists(tmp_docs_file):
 221             os.remove(tmp_docs_file)
 222
 223
 224 def main():
 225     base_elastic_url = urlparse.urljoin(CONF.elastic_url, '/test_results/mongo2elastic')
 226     to = CONF.destination
 227     days = args.latest_days
 228     es_creds = CONF.elastic_creds
 229
 230     if to == 'elasticsearch':
 231         to = base_elastic_url
 232
 233     for project, case_dicts in testcases.testcases_yaml.items():
 234         for case_dict in case_dicts:
 235             case = case_dict.get('name')
 236             fmt = testcases.compose_format(case_dict.get('format'))
 237             DocumentsPublisher(project,
 238                                case,
 239                                fmt,
 240                                days,
 241                                base_elastic_url,
 242                                es_creds,
 243                                to).export().get_existed_docs().publish()