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 uuid
  17 import traceback
  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", MANUAL_SCALE_VNF="manual scale vnf",
  33                 HEAL_VNF="heal vnf")
  34
  35
  36 class JobUtil(object):
  37     def __init__(self):
  38         pass
  39
  40     @staticmethod
  41     def __gen_job_id(job_name):
  42         return "%s-%s" % (job_name if job_name else "UnknownJob", uuid.uuid1())
  43
  44     @staticmethod
  45     def query_job_status(job_id, index_id=-1):
  46         # logger.info("Query job status, jobid =[%s], responseid [%d]" % (job_id, index_id))
  47         jobs = []
  48         if index_id < 0:
  49             row = JobStatusModel.objects.filter(jobid=job_id).order_by("-indexid").first()
  50             if row:
  51                 jobs.append(row)
  52         else:
  53             [jobs.append(job) for job in JobStatusModel.objects.filter(jobid=job_id).order_by("-indexid")
  54              if job.indexid > index_id]
  55
  56         # logger.info("Query job status, rows=%s" % str(jobs))
  57         return jobs
  58
  59     @staticmethod
  60     def is_job_exists(job_id):
  61         jobs = JobModel.objects.filter(jobid=job_id)
  62         return len(jobs) > 0
  63
  64     @staticmethod
  65     def create_job(inst_type, jobaction, inst_id, user='', job_id=None, res_name=''):
  66         if job_id is None:
  67             job_id = JobUtil.__gen_job_id(
  68                 '%s-%s-%s' % (str(inst_type).replace(' ', '_'), str(jobaction).replace(' ', '_'), str(inst_id)))
  69         job = JobModel()
  70         job.jobid = job_id
  71         job.jobtype = inst_type
  72         job.jobaction = jobaction
  73         job.resid = str(inst_id)
  74         job.status = JOB_STATUS.PROCESSING
  75         job.user = user
  76         job.starttime = datetime.datetime.now().strftime('%Y-%m-%d %X')
  77         job.progress = 0
  78         job.resname = res_name
  79         logger.debug("create a new job, jobid=%s, jobtype=%s, jobaction=%s, resid=%s, status=%d" %
  80                      (job.jobid, job.jobtype, job.jobaction, job.resid, job.status))
  81         job.save()
  82         return job_id
  83
  84     @staticmethod
  85     def clear_job(job_id):
  86         [job.delete() for job in JobModel.objects.filter(jobid=job_id)]
  87         logger.debug("Clear job, job_id=%s" % job_id)
  88
  89     @staticmethod
  90     def add_job_status(job_id, progress, status_decs, error_code=""):
  91         jobs = JobModel.objects.filter(jobid=job_id)
  92         if not jobs:
  93             logger.error("Job[%s] is not exists, please create job first." % job_id)
  94             raise Exception("Job[%s] is not exists." % job_id)
  95         try:
  96             int_progress = int(progress)
  97             job_status = JobStatusModel()
  98             jobstatuslist = JobStatusModel.objects.filter(jobid=job_id)
  99             indexid = 0
 100             for jobstatus in jobstatuslist:
 101                 if jobstatus.indexid > indexid:
 102                     indexid = jobstatus.indexid
 103             job_status.indexid = indexid + 1
 104             job_status.jobid = job_id
 105             job_status.status = "processing"
 106             job_status.progress = int_progress
 107
 108             if job_status.progress == 0:
 109                 job_status.status = "started"
 110             elif job_status.progress == 100:
 111                 job_status.status = "finished"
 112             elif job_status.progress == 101:
 113                 job_status.status = "partly_finished"
 114             elif job_status.progress > 101:
 115                 job_status.status = "error"
 116
 117             if error_code == "255":
 118                 job_status.status = "error"
 119
 120             job_status.descp = status_decs
 121             # job_status.errcode = error_code
 122             job_status.errcode = error_code if error_code else "0"
 123             job_status.addtime = datetime.datetime.now().strftime('%Y-%m-%d %X')
 124             job_status.save()
 125             logger.debug("Add a new job status, jobid=%s, indexid=%d,"
 126                          " status=%s, description=%s, progress=%d, errcode=%s, addtime=%r" %
 127                          (job_status.jobid, job_status.indexid, job_status.status, job_status.descp,
 128                           job_status.progress, job_status.errcode, job_status.addtime))
 129
 130             job = jobs[0]
 131             job.progress = int_progress
 132             if job_status.progress >= 100:
 133                 job.status = JOB_STATUS.FINISH
 134                 job.endtime = datetime.datetime.now().strftime('%Y-%m-%d %X')
 135             job.save()
 136             logger.debug("update job, jobid=%s, progress=%d" % (job_status.jobid, int_progress))
 137         except:
 138             logger.error(traceback.format_exc())
 139
 140     @staticmethod
 141     def clear_job_status(job_id):
 142         [job.delete() for job in JobStatusModel.objects.filter(jobid=job_id)]
 143         logger.debug("Clear job status, job_id=%s" % job_id)
 144
 145     @staticmethod
 146     def get_unfinished_jobs(url_prefix, inst_id, inst_type):
 147         jobs = JobModel.objects.filter(resid=inst_id, jobtype=inst_type, status=JOB_STATUS.PROCESSING)
 148         progresses = reduce(lambda content, job: content + [url_prefix + "/" + job.jobid], jobs, [])
 149         return progresses