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