catalog/pub/utils/jobutil.py

   1 # Copyright 2017 ZTE Corporation.
   2 #
   3 # Licensed under the Apache License, Version 2.0 (the "License");
   4 # you may not use this file except in compliance with the License.
   5 # You may obtain a copy of the License at
   6 #
   7 #         http://www.apache.org/licenses/LICENSE-2.0
   8 #
   9 # Unless required by applicable law or agreed to in writing, software
  10 # distributed under the License is distributed on an "AS IS" BASIS,
  11 # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
  12 # See the License for the specific language governing permissions and
  13 # limitations under the License.
  14 import datetime
  15 import logging
  16 import traceback
  17 import uuid
  18 from functools import reduce
  19
  20 from catalog.pub.database.models import JobStatusModel, JobModel
  21
  22 logger = logging.getLogger(__name__)
  23
  24
  25 def enum(**enums):
  26     return type('Enum', (), enums)
  27
  28
  29 JOB_STATUS = enum(PROCESSING=0, FINISH=1)
  30 JOB_MODEL_STATUS = enum(STARTED='started', PROCESSING='processing', FINISHED='finished', ERROR='error',
  31                         TIMEOUT='timeout')
  32 JOB_TYPE = enum(CREATE_VNF="create vnf", TERMINATE_VNF="terminate vnf", GRANT_VNF="grant vnf",
  33                 MANUAL_SCALE_VNF="manual scale vnf",
  34                 HEAL_VNF="heal vnf")
  35 JOB_ERROR_CODE = enum(NORMAL=0, PACKAGE_EXIST=1, CATALOG_EXCEPTION=2, SYSTEM_ERROR=3)
  36
  37
  38 class JobUtil(object):
  39     def __init__(self):
  40         pass
  41
  42     @staticmethod
  43     def __gen_job_id(job_name):
  44         return "%s-%s" % (job_name if job_name else "UnknownJob", uuid.uuid1())
  45
  46     @staticmethod
  47     def query_job_status(job_id, index_id=-1):
  48         # logger.info("Query job status, jobid =[%s], responseid [%d]" % (job_id, index_id))
  49         jobs = []
  50         if index_id < 0:
  51             row = JobStatusModel.objects.filter(jobid=job_id).order_by("-indexid").first()
  52             if row:
  53                 jobs.append(row)
  54         else:
  55             [jobs.append(job) for job in JobStatusModel.objects.filter(jobid=job_id).order_by("-indexid")
  56              if job.indexid > index_id]
  57
  58         # logger.info("Query job status, rows=%s" % str(jobs))
  59         return jobs
  60
  61     @staticmethod
  62     def is_job_exists(job_id):
  63         jobs = JobModel.objects.filter(jobid=job_id)
  64         return len(jobs) > 0
  65
  66     @staticmethod
  67     def create_job(inst_type, jobaction, inst_id, user='', job_id=None, res_name=''):
  68         if job_id is None:
  69             job_id = JobUtil.__gen_job_id(
  70                 '%s-%s-%s' % (str(inst_type).replace(' ', '_'), str(jobaction).replace(' ', '_'), str(inst_id)))
  71         job = JobModel()
  72         job.jobid = job_id
  73         job.jobtype = inst_type
  74         job.jobaction = jobaction
  75         job.resid = str(inst_id)
  76         job.status = JOB_STATUS.PROCESSING
  77         job.user = user
  78         job.starttime = datetime.datetime.now().strftime('%Y-%m-%d %X')
  79         job.progress = 0
  80         job.resname = res_name
  81         logger.debug("create a new job, jobid=%s, jobtype=%s, jobaction=%s, resid=%s, status=%d" %
  82                      (job.jobid, job.jobtype, job.jobaction, job.resid, job.status))
  83         job.save()
  84         return job_id
  85
  86     @staticmethod
  87     def clear_job(job_id):
  88         [job.delete() for job in JobModel.objects.filter(jobid=job_id)]
  89         logger.debug("Clear job, job_id=%s" % job_id)
  90
  91     @staticmethod
  92     def add_job_status(job_id, progress, status_decs, error_code=""):
  93         jobs = JobModel.objects.filter(jobid=job_id)
  94         if not jobs:
  95             logger.error("Job[%s] is not exists, please create job first." % job_id)
  96             raise Exception("Job[%s] is not exists." % job_id)
  97         try:
  98             int_progress = int(progress)
  99             job_status = JobStatusModel()
 100             jobstatuslist = JobStatusModel.objects.filter(jobid=job_id)
 101             indexid = 0
 102             for jobstatus in jobstatuslist:
 103                 if jobstatus.indexid > indexid:
 104                     indexid = jobstatus.indexid
 105             job_status.indexid = indexid + 1
 106             job_status.jobid = job_id
 107             job_status.status = "processing"
 108             job_status.progress = int_progress
 109
 110             if job_status.progress == 0:
 111                 job_status.status = "started"
 112             elif job_status.progress == 100:
 113                 job_status.status = "finished"
 114             elif job_status.progress == 101:
 115                 job_status.status = "partly_finished"
 116             elif job_status.progress > 101:
 117                 job_status.status = "error"
 118
 119             if error_code == "255":
 120                 job_status.status = "error"
 121
 122             job_status.descp = status_decs
 123             # job_status.errcode = error_code
 124             job_status.errcode = error_code if error_code else "0"
 125             job_status.addtime = datetime.datetime.now().strftime('%Y-%m-%d %X')
 126             job_status.save()
 127             logger.debug("Add a new job status, jobid=%s, indexid=%d,"
 128                          " status=%s, description=%s, progress=%d, errcode=%s, addtime=%r" %
 129                          (job_status.jobid, job_status.indexid, job_status.status, job_status.descp,
 130                           job_status.progress, job_status.errcode, job_status.addtime))
 131
 132             job = jobs[0]
 133             job.progress = int_progress
 134             if job_status.progress >= 100:
 135                 job.status = JOB_STATUS.FINISH
 136                 job.endtime = datetime.datetime.now().strftime('%Y-%m-%d %X')
 137             job.save()
 138             logger.debug("update job, jobid=%s, progress=%d" % (job_status.jobid, int_progress))
 139         except:
 140             logger.error(traceback.format_exc())
 141
 142     @staticmethod
 143     def clear_job_status(job_id):
 144         [job.delete() for job in JobStatusModel.objects.filter(jobid=job_id)]
 145         logger.debug("Clear job status, job_id=%s" % job_id)
 146
 147     @staticmethod
 148     def get_unfinished_jobs(url_prefix, inst_id, inst_type):
 149         jobs = JobModel.objects.filter(resid=inst_id, jobtype=inst_type, status=JOB_STATUS.PROCESSING)
 150         progresses = reduce(lambda content, job: content + [url_prefix + "/" + job.jobid], jobs, [])
 151         return progresses