EventObjects.py

Go to the documentation of this file.

# coding: utf-8
'''
HCE project, Python bindings, Distributed Tasks Manager application.
Event objects definitions.

@package: dtm
@author bgv bgv.hce@gmail.com
@link: http://hierarchical-cluster-engine.com/
@copyright: Copyright © 2013-2014 IOIX Ukraine
@license: http://hierarchical-cluster-engine.com/license/
@since: 0.1
'''


import ctypes
from app.Utils import JsonSerializable
from app.Utils import getHash
from datetime import datetime
import json
import zlib
import time
import logging
import dtm.Constants as DTM_CONSTS
import app.SQLCriterions

from transport.IDGenerator import IDGenerator

logger = logging.getLogger(DTM_CONSTS.LOGGER_NAME)


class Task(object):

  FILE_ACTION_CREATE_BEFORE = 1
  FILE_ACTION_DELETE_BEFORE = 2
  FILE_ACTION_READ_AFTER = 4
  FILE_ACTION_DELETE_AFTER = 8
  FILE_ACTION_BASE64_ENCODED = 2147483648

  STRATEGY_DATE = "DATE"
  STRATEGY_DATE_MAX = "DATE_MAX"
  STRATEGY_DATE_SHIFT = "DATE_SHIFT"
  STRATEGY_CPU = "CPU"
  STRATEGY_RAM_FREE = "RAM_FREE"
  STRATEGY_RAM = "RAM"
  STRATEGY_DISK_FREE = "DISK_FREE"
  STRATEGY_DISK = "DISK"
  STRATEGY_TIME = "TIME"
  STRATEGY_THREADS = "THREADS"
  STRATEGY_SDELAY = "SDELAY"
  STRATEGY_RDELAY = "RDELAY"
  STRATEGY_RETRY = "RETRY"
  STRATEGY_PRIORITY = "PRIORITY"
  STRATEGY_CPU_LOAD_MAX = "CPU_LOAD_MAX"
  STRATEGY_IO_WAIT_MAX = "IO_WAIT_MAX"
  STRATEGY_autoCleanupFields = "autoCleanupFields"

  STRATEGY_AUTOCLEANUP_TTL = "TTL"
  STRATEGY_AUTOCLEANUP_DELETE_TYPE = "DeleteType"
  STRATEGY_AUTOCLEANUP_DELETE_RETRIES = "DeleteRetries"
  STRATEGY_AUTOCLEANUP_SSTATE = "State"

  TASK_MODE_SYNCH = 1
  TASK_MODE_ASYNCH = 2

  TIME_MAX_DEFAULT = 60000

  TASK_TYPE_SHELL = 0
  TASK_TYPE_SSH = 1

  TYPE_DEFAULT = 0

  
  def __init__(self):
    
    self.id = None
    
    self.command = None
    
    self.input = None
    
    self.files = []
    
    self.session = {}
    
    self.strategy = {}
    
    self.limits = {}
    
    self.autoCleanupFields = {}
    
    self.type = self.TYPE_DEFAULT
    
    self.name = ""


  
  def setSessionVar(self, sessionVarName, sessionVarValue):
    self.session[sessionVarName] = sessionVarValue


  
  def setStrategyVar(self, strategyVarName, strategyVarValue):
    self.strategy[strategyVarName] = strategyVarValue


  
  def setFile(self, fileItem):
    self.files.append(fileItem)


  
  def setLimitsVar(self, limitsVarName, limitsVarValue):
    self.limits[limitsVarName] = limitsVarValue




class NewTask(Task):

  
  def __init__(self, taskCommandLine, taskId=None, name=None):
    super(NewTask, self).__init__()
    
    if taskId is None:
      idGenerator = IDGenerator()
      #self.id = ctypes.c_ulong(zlib.crc32(idGenerator.get_connection_uid())).value
      #self.id = ctypes.c_uint32(zlib.crc32(idGenerator.get_connection_uid(), int(time.time()))).value
      self.id = getHash(idGenerator.get_connection_uid())
    else:
      self.id = taskId
    if name is not None:
      self.name = name
    
    self.command = taskCommandLine
    
    self.input = ""
    #Strategy priority init
    self.setStrategyVar(self.STRATEGY_PRIORITY, 1)
    #Session init
    self.setSessionVar("time_max", self.TIME_MAX_DEFAULT)
    self.setSessionVar("tmode", self.TASK_MODE_SYNCH)
    self.setSessionVar("type", self.TASK_TYPE_SHELL)
    self.setSessionVar("home_directory", "")
    self.setSessionVar("port", 0)
    self.setSessionVar("user", "")
    self.setSessionVar("password", "")
    self.setSessionVar("shell", "")
    self.setSessionVar("environment", {})
    self.setSessionVar("timeout", 0)


  def toJSON(self):
    return json.dumps(self, default=lambda o: o.__dict__, sort_keys=True, indent=4)




class UpdateTask(Task):

  
  def __init__(self, taskId):
    super(UpdateTask, self).__init__()
    
    self.id = taskId




class UpdateTaskFields(object):

  
  def __init__(self, taskId):
    
    self.id = taskId
    
    self.fields = {}




class GetTaskManagerFields(object):

  
  def __init__(self, taskId):
    
    self.id = taskId




class TaskManagerFields(object):

  
  def __init__(self, taskId):
    
    self.id = taskId
    
    self.fields = {}


  def toJSON(self):
    return json.dumps(self, default=lambda o: o.__dict__, sort_keys=True, indent=4)




class TasksStatus(object):


  def __init__(self):
    self.taskManagerFieldsList = []


  def toJSON(self):
    return json.dumps(self, default=lambda o: o.__dict__, sort_keys=True, indent=4)




class GetTasksStatus(object):

  LOG_STRATEGY = "CHECK_LOG"
  CHECK_LOG_YES = 1

  FILTER_CDATE_FROM = "CDATE_FROM"
  FILTER_CDATE_TO = "CDATE_TO"
  FILTER_SDATE_FROM = "SDATE_FROM"
  FILTER_SDATE_TO = "SDATE_TO"
  FILTER_RDATE_FROM = "RDATE_FROM"
  FILTER_RDATE_TO = "RDATE_TO"
  FILTER_FDATE_FROM = "FDATE_FROM"
  FILTER_FDATE_TO = "FDATE_TO"
  FILTER_INPROGRESS_TIME_FROM = "INPROGRESS_TIME_FROM"
  FILTER_INPROGRESS_TIME_TO = "INPROGRESS_TIME_TO"
  FILTER_INPROGRESS_TIME_MAX_FROM = "INPROGRESS_TIME_MAX_FROM"
  FILTER_INPROGRESS_TIME_MAX_TO = "INPROGRESS_TIME_MAX_TO"
  FILTER_INPROGRESS = "INPROGRESS"
  FILTER_SCHEDULED = "SCHEDULED"
  FILTER_RUNNING = "RUNNING"
  FILTER_FINISHED = "FINISHED"
  FILTER_TERMINATED = "TERMINATED"
  FILTER_CRASHED = "CRASHED"
  FILTER_RRAM_FROM = "RRAM_FROM"
  FILTER_RRAM_TO = "RRAM_TO"
  FILTER_VRAM_FROM = "VRAM_FROM"
  FILTER_VRAM_TO = "VRAM_TO"
  FILTER_CPU_FROM = "CPU_FROM"
  FILTER_CPU_TO = "CPU_TO"

  
  def __init__(self, idsList):
    
    self.ids = idsList
    
    self.filters = {}
    
    self.strategy = {}


  
  def setFilterVar(self, filterVarName, filterVarValue):
    self.filters[filterVarName] = filterVarValue


  
  def setStrategyVar(self, strategyVarName, strategyVarValue):
    self.strategy[strategyVarName] = strategyVarValue




class CheckTaskState(object):

  TYPE_SIMPLE = 1
  TYPE_FULL = 2

  
  def __init__(self, taskId, checkType=TYPE_SIMPLE):
    
    self.id = taskId
    
    self.type = checkType

  def toJSON(self):
    return json.dumps(self, default=lambda o: o.__dict__, sort_keys=True, indent=4)




class FetchTasksResults(object):

  TYPE_DELETE = 1
  TYPE_SAVE = 2

  
  def __init__(self, taskId, fetchType=TYPE_DELETE):
    
    self.id = taskId
    
    self.type = fetchType

  def toJSON(self):
    return json.dumps(self, default=lambda o: o.__dict__, sort_keys=True, indent=4)




class FetchTasksResultsFromCache(object):

  
  def __init__(self, taskId):
    
    self.ids = taskId




class DeleteTask(Task):

  GROUP_DELETE = -1

  RESPONSE_CODE_DBI_ERROR = 1
  RESPONSE_CODE_UNKNOWN_ERROR = 2
  RESPONSE_CODE_DRCE_ERROR = 3

  TERMINATE_ALG_DEFAULT = 1
  TERMINATE_ALG_CUSTOM = 2

  TERMINATE_DELAY_DEFAULT = 1000
  TERMINATE_REPEAT_DEFAULT = 3

  TERMINATE_SIGTERM = 15
  TERMINATE_SIGKILL = 9

  ACTION_DELETE_TASK_DATA = 0
  ACTION_TERMINATE_TASK_AND_DELETE_DATA = 1
  ACTION_TERMINATE_TASK_ONLY = 2
  ACTION_DELETE_ON_DTM = 3

  DEFAULT_TASK_NAME = "DELETE"

  
  def __init__(self, deleteTaskId, taskId=None):
    super(DeleteTask, self).__init__()
    
    if taskId is None:
      idGenerator = IDGenerator()
      self.id = ctypes.c_uint32(zlib.crc32(idGenerator.get_connection_uid(), int(time.time()))).value
    else:
      self.id = taskId
    #Strategy priority init
    self.setStrategyVar(self.STRATEGY_PRIORITY, 1)
    #The task Id to delete, used for EE request TERMINATE task
    self.deleteTaskId = deleteTaskId
    
    self.alg = self.TERMINATE_ALG_DEFAULT
    
    self.delay = self.TERMINATE_DELAY_DEFAULT
    
    self.repeat = self.TERMINATE_REPEAT_DEFAULT
    
    self.signal = self.TERMINATE_SIGTERM
    
    self.host = ""
    
    self.port = 0
    # Action that will be performed on EEManager, default only delete task's data on EE and delete task on DTM
    # - this is possible only for tasks in proper state (that is not in active states)
    self.action = self.ACTION_DELETE_TASK_DATA
    
    self.strategy = {}
    #TODO: possible find the way to define decorator
    #@property
    #def id(self):
    #  return self.id

    #@id.setter
    #def id(self, value):
    #  self.deleteTaskId = value
    self.name = self.DEFAULT_TASK_NAME

  def toJSON(self):
    return json.dumps(self, default=lambda o: o.__dict__, sort_keys=True, indent=4)




class ExecuteTask(object):

  
  def __init__(self, taskId):
    
    self.id = taskId




class GetScheduledTasks(object):

  
  def __init__(self, timeSlotSize):
    
    self.timeSlot = timeSlotSize




class GetScheduledTasksResponse(object):

  
  def __init__(self, idsList):
    
    self.ids = idsList




class GeneralResponse(object):

  #General valid response value
  ERROR_OK = 0

  
  def __init__(self, errorCode=ERROR_OK, errorMessage=""):
    
    self.errorCode = errorCode
    
    self.errorMessage = errorMessage
    
    self.statuses = []

  def toJSON(self):
    return json.dumps(self, default=lambda o: o.__dict__, sort_keys=True, indent=4)




class DeleteTaskData(object):

  
  def __init__(self, taskId):
    
    self.id = taskId




class FetchEEResponseData(object):

  
  def __init__(self, taskId):
    
    self.id = taskId




class DeleteEEResponseData(object):

  
  def __init__(self, taskId):
    
    self.id = taskId




class FetchTaskData(object):

  
  def __init__(self, taskId):
    
    self.id = taskId




class EEResponseData(object):

  #Error codes
  ERROR_CODE_TIMEOUT = -1
  ERROR_CODE_OK = 0
  #ERROR_CODE_TASK_NOT_FOUND = -2
  ERROR_CODE_TASK_NOT_FOUND = 3
  ERROR_CODE_BAD_ID = 14

  #Error messages
  ERROR_MESSAGE_OK = ""
  ERROR_MESSAGE_TIMEOUT = "Request timeout reached!"
  ERROR_MESSAGE_TASK_NOT_FOUND = "Task not found in queue!"
  #Requests types
  REQUEST_TYPE_SET = 0
  REQUEST_TYPE_CHECK = 1
  REQUEST_TYPE_DELETE = 2
  REQUEST_TYPE_GET = 3
  #Task states
  TASK_STATE_FINISHED = 0
  TASK_STATE_IN_PROGRESS = 1
  TASK_STATE_NEW = 2
  TASK_STATE_NOT_FOUND = 3
  TASK_STATE_TERMINATED = 4
  TASK_STATE_CRASHED = 5
  TASK_STATE_SET_ERROR = 6
  TASK_STATE_UNDEFINED = 7
  TASK_STATE_TERMINATED_BY_DRCE_TTL = 11
  TASK_STATE_SCHEDULED_TO_DELETE = 100
  TASK_STATE_DELETED = 101
  TASK_STATE_NEW_DATA_STORED = 102
  TASK_STATE_NEW_SCHEDULED = 103
  TASK_STATE_CLEANED = 104
  TASK_STATE_NEW_JUST_CREATED = 105
  TASK_STATE_SCHEDULE_TRIES_EXCEEDED = 106
  TASK_STATE_RUN_TRIES_EXCEEDED = 107

  
  def __init__(self, taskId):
    
    self.id = taskId
    
    self.type = self.REQUEST_TYPE_SET
    
    self.errorCode = self.ERROR_CODE_OK
    
    self.errorMessage = self.ERROR_MESSAGE_OK
    
    self.requestTime = 0
    
    self.state = self.TASK_STATE_FINISHED
    
    self.pId = 0
    
    self.stdout = ""
    
    self.stderr = ""
    
    self.exitStatus = 0
    
    self.taskTime = 0
    
    self.nodeName = ""
    
    self.nodeHost = ""
    
    self.nodePort = 0
    
    self.files = []
    
    self.fields = {}

  def toJSON(self):
    return json.dumps(self, default=lambda o: o.__dict__, sort_keys=True, indent=4)


  
  def setFile(self, fileItem):
    self.files.append(fileItem)




class ScheduledTask(object):

  STATE_PLANNED = 1
  STATE_INPROGRESS = 2
  STATE_CLOSED = 3

  
  def __init__(self, taskId, rTime, rTimeMax, state, priority):
    
    self.id = taskId
    
    self.rTime = rTime
    
    self.rTimeMax = rTimeMax
    
    self.state = state
    
    self.priority = priority




class UpdateScheduledTasks(object):

  
  def __init__(self, scheduledTasks=None):
    
    self.tasks = []
    if scheduledTasks is not None:
      self.tasks = scheduledTasks




class GetScheduledTask(object):

  
  def __init__(self):
    
    self.taskIdMin = None
    
    self.taskIdMax = None
    
    self.rTimeMin = None
    
    self.rTimeMax = None
    
    self.state = ScheduledTask.STATE_PLANNED
    #The priority min
    self.priorityMin = None
    
    self.priorityMax = None




class Resource(object):

  STATE_ACTIVE = 0
  STATE_UNDEFINED = 1
  STATE_INACTIVE = 2

  
  def __init__(self, nodeId):
    
    self.nodeId = nodeId
    
    self.nodeName = ""
    
    self.host = ""
    
    self.port = 0
    
    self.cpu = 0
    
    self.io = 0
    
    self.ramRU = 0
    
    self.ramVU = 0
    
    self.ramR = 0
    
    self.ramV = 0
    
    self.swap = 0
    
    self.swapU = 0
    
    self.disk = 0
    
    self.diskU = 0
    
    self.state = self.STATE_UNDEFINED
    
    self.uDate = datetime.now()
    #Number of CPU cores
    self.cpuCores = 0
    #Number of run threads
    self.threads = 0
    #Number of run processes
    self.processes = 0




class ResourcesAVG(object):

  
  def __init__(self):
    
    self.cpu = 0
    
    self.io = 0
    
    self.ramRU = 0
    
    self.ramVU = 0
    
    self.ramR = 0
    
    self.ramV = 0
    
    self.swap = 0
    
    self.disk = 0
    
    self.uDate = 0
    #Number of CPU cores
    self.cpuCores = 0
    #Number of run threads
    self.threads = 0
    #Number of run processes
    self.processes = 0




class AdminStatData(object):

  FIELD_CLIENTS_LIST = "clients"

  
  def __init__(self, className, statFieldsDic=None):
    
    self.className = className
    
    self.fields = {}
    if statFieldsDic is not None:
      self.fields = statFieldsDic


  def toJSON(self):
    return json.dumps(self, default=lambda o: o.__dict__, sort_keys=True, indent=4)




class AdminConfigVars(object):

  
  def __init__(self, className, configFieldsList=None):
    
    self.className = className
    
    self.fields = {}
    if configFieldsList is not None:
      self.fields = configFieldsList


  def toJSON(self):
    return json.dumps(self, default=lambda o: o.__dict__, sort_keys=True, indent=4)




class AdminState(object):

  STATE_NOP = 0
  STATE_READY = 1
  STATE_SHUTDOWN = 2
  STATE_SUSPEND = 3
  STATE_TRACE = 4
  STATE_TRANSACTION_ROLLBACK = 6
  STATE_ERROR = 7

  
  def __init__(self, className, command):
    
    self.className = className
    
    self.command = command


  def toJSON(self):
    return json.dumps(self, default=lambda o: o.__dict__, sort_keys=True, indent=4)




class AdminSuspend(object):

  SUSPEND = 0
  RUN = 1

  
  def __init__(self, suspendType=SUSPEND):
    
    self.suspendType = suspendType


  def isSuspend(self):
    return self.suspendType == self.SUSPEND


  def toJSON(self):
    return json.dumps(self, default=lambda o: o.__dict__, sort_keys=True, indent=4)




class DeleteTaskResults(object):

  TASK_STATE_ERROR = 2000
  TASK_STATE_ERROR_MESSAGE = "Wrong task state to cleanup task's data!"
  TASK_NOT_FOUND_ERROR = 2001
  TASK_NOT_FOUND_ERROR_MESSAGE = "Task not found!"
  EMPRY_RAW_ERROR = 2002
  EMPRY_RAW_ERROR_MESSAGE = "Empty json raw!"
  DRCE_ERROR = 2003
  DRCE_ERROR_MESSAGE = "Some DRCE Error!"


  
  def __init__(self, taskId):
    
    self.id = taskId




class FetchAvailabelTaskIds(JsonSerializable):

  TABLE_NAME_DEFAULT = "task_back_log_scheme"

  
  def __init__(self, fetchNum, fetchAdditionalFields=False, criterions=None, tableName=TABLE_NAME_DEFAULT):
    super(FetchAvailabelTaskIds, self).__init__()
    self.criterions = criterions
    self.fetchNum = fetchNum
    self.fetchAdditionalFields = fetchAdditionalFields
    self.tableName = tableName
    self.fillCriterions()


  
  def fillCriterions(self):
    if self.criterions is None:
      self.criterions = {}
    if app.SQLCriterions.CRITERION_WHERE not in self.criterions or \
    self.criterions[app.SQLCriterions.CRITERION_WHERE] is None:
      self.criterions[app.SQLCriterions.CRITERION_WHERE] = "deleteTaskId = 0"
    if app.SQLCriterions.CRITERION_ORDER not in self.criterions or \
    self.criterions[app.SQLCriterions.CRITERION_ORDER] is None:
      self.criterions[app.SQLCriterions.CRITERION_ORDER] = "rDate"
    if app.SQLCriterions.CRITERION_LIMIT not in self.criterions or \
    self.criterions[app.SQLCriterions.CRITERION_LIMIT] is None:
      self.criterions[app.SQLCriterions.CRITERION_LIMIT] = self.fetchNum




class AvailableTaskIds(JsonSerializable):
  
  def __init__(self, ids, tasks=None):
    super(AvailableTaskIds, self).__init__()
    self.ids = ids
    self.tasks = tasks




class System(object):
  
  def __init__(self, stype=0, data=None):
    self.type = stype
    self.data = data


  def toJSON(self):
    return json.dumps(self, default=lambda o: o.__dict__, sort_keys=True, indent=4)



from dbi.EventObjects import CustomRequest
from dbi.EventObjects import CustomResponse