Inheritance diagram for dc.BatchTasksManagerRealTime.BatchTasksManagerRealTime:

Collaboration diagram for dc.BatchTasksManagerRealTime.BatchTasksManagerRealTime:

Public Member Functions
def	__init__ (self, configParser, connectionBuilderLight=None)

def	on_poll_timeout (self)

def	onEventsHandler (self, event)

def	forkBatch (self, loggingObj, event)

def	sendClientResponse (self, clientRequestEvent, clientResponseObj)

def	prepareDRCERequest (self, eventObj)

def	processDRCERequest (self, taskExecuteRequest)

def	sendToDRCERouter (self, request)

Public Member Functions inherited from app.BaseServerManager.BaseServerManager
def	__init__ (self, poller_manager=None, admin_connection=None, conectionLightBuilder=None, exceptionForward=False, dumpStatVars=True)
	constructor More...

def	addConnection (self, name, connection)

def	setEventHandler (self, eventType, handler)
	set event handler rewrite the current handler for eventType More...

def	send (self, connect_name, event)
	send event More...

def	reply (self, event, reply_event)
	wrapper for sending event in reply for event More...

def	poll (self)
	poll function polling connections receive as multipart msg, the second argument is pickled pyobj More...

def	process (self, event)
	process event call the event handler method that was set by user or on_unhandled_event method if not set More...

def	run (self)

def	is_connection_registered (self, name)
	check is a connection was registered in a instance of BaseServerManager i object More...

def	on_poll_timeout (self)
	function will call every time when ConnectionTimeout exception arrive More...

def	on_unhandled_event (self, event)
	function will call every time when arrive doesn't set handler for event type of event.evenType More...

def	build_poller_list (self)

def	clear_poller (self)

def	onAdminState (self, event)
	onAdminState event handler process admin SHUTDOWN command More...

def	onAdminFetchStatData (self, event)
	onAdminState event handler process admin command More...

def	onAdminSuspend (self, event)
	onAdminState event handler process admin command More...

def	getStatDataFields (self, fields)
	getStatDataFields returns stat data from storage More...

def	getSystemStat (self)
	getSystemStat returns stat data for system indicators: RAMV, RAMR and CPU More...

def	getConfigVarsFields (self, fields)
	getConfigVarsFields returns config vars from storage More...

def	onAdminGetConfigVars (self, event)
	onAdminGetConfigVars event handler process getConfigVars admin command, fill and return config vars array from internal storage More...

def	onAdminSetConfigVars (self, event)
	onAdminSetConfigVars event handler process setConfigVars admin command More...

def	setConfigVars (self, setConfigVars)
	processSetConfigVars sets config vars in storage More...

def	sendAdminReadyEvent (self)
	send ready event to notify adminInterfaceService More...

def	createLogMsg (self, event)
	from string message from event object More...

def	initStatFields (self, connect_name)
	add record in statFields More...

def	updateStatField (self, field_name, value, operation=STAT_FIELDS_OPERATION_ADD)
	update values of stat field - default sum More...

def	processSpecialConfigVars (self, name, value)
	send ready event to notify adminInterfaceService More...

def	getLogLevel (self)
	Get log level from first of existing loggers. More...

def	setLogLevel (self, level)
	Set log level for all loggers. More...

def	saveStatVarsDump (self)
	Save stat vars in json file. More...

def	loadStatVarsDump (self)
	Load stat vars in json file. More...

def	getStatVarsDumpFileName (self)
	Get stat vars file name. More...

def	createDBIDict (self, configParser)

Public Attributes
	serverName

	drceHost

	drcePort

	drceIdGenerator

	drceCommandConvertor

	totalTime

Public Attributes inherited from app.BaseServerManager.BaseServerManager
	dumpStatVars

	poller_manager

	eventBuilder

	exit_flag

	pollTimeout

	connections

	event_handlers

	statFields
	stat fields container More...

	configVars

	exceptionForward

Static Public Attributes
int	DRCE_REDUCER_TTL = 3000000

int	REQUEST_ERROR_OBJECT_TYPE = 1

int	REQUEST_ERROR_URLS_COUNT = 2

int	REQUEST_ERROR_THREADS_NUMBER_EXCEEDED = 3

int	CONFIG_DRCE_REQUEST_ROUTING_DEFAULT = 1

int	CONFIG_BATCH_MAX_ITERATIONS_DEFAULT = 2

string	CONFIG_SERVER = "server"

string	CONFIG_DRCE_STARTER_NAME = "DRCEStarterName"

string	CONFIG_DRCE_HOST = "DRCEHost"

string	CONFIG_DRCE_PORT = "DRCEPort"

string	CONFIG_DRCE_TIMEOUT = "DRCETimeout"

string	CONFIG_DRCE_CRAWLER_APP_NAME = "DRCECrawlerAppName"

string	CONFIG_BATCH_MAX_TIME = "BatchMaxExecutionTime"

string	CONFIG_BATCH_MAX_URLS = "BatchMaxURLs"

string	CONFIG_MAX_THREADS = "MaxThreads"

string	CONFIG_POLLING_TIMEOUT = "PollingTimeout"

string	CONFIG_DRCE_REQUEST_ROUTING = "DRCERequestRouting"

string	CONFIG_BATCH_MAX_ITERATIONS = "BatchMaxIterations"

string	REAL_TIME_CRAWL_THREAD_NAME_PREFIX = 'RtCrawl_'

Static Public Attributes inherited from app.BaseServerManager.BaseServerManager
string	ADMIN_CONNECT_ENDPOINT = "Admin"

string	ADMIN_CONNECT_CLIENT = "Admin"

int	POLL_TIMEOUT_DEFAULT = 3000

int	STAT_FIELDS_OPERATION_ADD = 0

int	STAT_FIELDS_OPERATION_SUB = 1

int	STAT_FIELDS_OPERATION_SET = 2

int	STAT_FIELDS_OPERATION_INIT = 3

string	POLL_TIMEOUT_CONFIG_VAR_NAME = "POLL_TIMEOUT"

string	LOG_LEVEL_CONFIG_VAR_NAME = "LOG_LEVEL"

string	STAT_DUMPS_DEFAULT_DIR = "/tmp/"

string	STAT_DUMPS_DEFAULT_NAME = "%APP_NAME%_%CLASS_NAME%_stat_vars.dump"

dictionary	LOGGERS_NAMES = {APP_CONSTS.LOGGER_NAME, "dc", "dtm", "root", ""}

Detailed Description

Definition at line 51 of file BatchTasksManagerRealTime.py.

Constructor & Destructor Documentation

◆ init()

def dc.BatchTasksManagerRealTime.BatchTasksManagerRealTime.__init__	(	self,
		configParser,
		connectionBuilderLight = `None`
	)

Definition at line 83 of file BatchTasksManagerRealTime.py.

   def __init__(self, configParser, connectionBuilderLight=None):
     super(BatchTasksManagerRealTime, self).__init__()
 
     # Instantiate the connection builder light if not set
     if connectionBuilderLight is None:
       connectionBuilderLight = ConnectionBuilderLight()
 
     # Get configuration settings
     className = self.__class__.__name__
     self.serverName = configParser.get(className, self.CONFIG_SERVER)
 
     # Create connections and raise bind or connect actions for correspondent connection type
     serverConnection = connectionBuilderLight.build(TRANSPORT_CONSTS.SERVER_CONNECT, self.serverName)
     self.setEventHandler(DC_CONSTS.EVENT_TYPES.BATCH, self.onEventsHandler)
 
     # Initialize DRCE API
     self.configVars[self.CONFIG_DRCE_TIMEOUT] = configParser.getint(className, self.CONFIG_DRCE_TIMEOUT)
     self.drceHost = configParser.get(className, self.CONFIG_DRCE_HOST)
     self.drcePort = configParser.get(className, self.CONFIG_DRCE_PORT)
     # self.drceManager = DRCEManager()
     # self.drceManager.activate_host(HostParams(self.drceHost, self.drcePort))
     self.drceIdGenerator = UIDGenerator()
     self.drceCommandConvertor = CommandConvertor()
 
     # Add connections to the polling set
     self.addConnection(self.serverName, serverConnection)
 
     # Max URLs per batch
     self.configVars[self.CONFIG_BATCH_MAX_URLS] = configParser.getint(className, self.CONFIG_BATCH_MAX_URLS)
 
     # Set crawler task app name
     self.configVars[self.CONFIG_DRCE_CRAWLER_APP_NAME] = configParser.get(className, self.CONFIG_DRCE_CRAWLER_APP_NAME)
     self.configVars[self.CONFIG_BATCH_MAX_TIME] = configParser.getint(className, self.CONFIG_BATCH_MAX_TIME)
     self.configVars[self.CONFIG_DRCE_STARTER_NAME] = configParser.get(className, self.CONFIG_DRCE_STARTER_NAME)
     self.configVars[self.CONFIG_MAX_THREADS] = configParser.getint(className, self.CONFIG_MAX_THREADS)
     if configParser.has_option(className, self.CONFIG_DRCE_REQUEST_ROUTING):
       self.configVars[self.CONFIG_DRCE_REQUEST_ROUTING] = configParser.getint(className,
                                                                               self.CONFIG_DRCE_REQUEST_ROUTING)
     else:
       self.configVars[self.CONFIG_DRCE_REQUEST_ROUTING] = self.CONFIG_DRCE_REQUEST_ROUTING_DEFAULT
     if configParser.has_option(className, self.CONFIG_BATCH_MAX_ITERATIONS):
       self.configVars[self.CONFIG_BATCH_MAX_ITERATIONS] = configParser.getint(className,
                                                                               self.CONFIG_BATCH_MAX_ITERATIONS)
     else:
       self.configVars[self.CONFIG_BATCH_MAX_ITERATIONS] = self.CONFIG_BATCH_MAX_ITERATIONS_DEFAULT
 
     # Batches counter init in stat vars
     self.updateStatField(DC_CONSTS.BATCHES_CRAWL_COUNTER_TOTAL_NAME, 0, self.STAT_FIELDS_OPERATION_INIT)
     # Batches that fault processing counter init in stat vars
     self.updateStatField(DC_CONSTS.BATCHES_CRAWL_COUNTER_FAULT_NAME, 0, self.STAT_FIELDS_OPERATION_INIT)
     # Batches urls total counter init in stat vars
     self.updateStatField(DC_CONSTS.BATCHES_CRAWL_COUNTER_URLS_NAME, 0, self.STAT_FIELDS_OPERATION_INIT)
     # Fault batches urls total counter init in stat vars
     self.updateStatField(DC_CONSTS.BATCHES_CRAWL_COUNTER_URLS_FAULT_NAME, 0, self.STAT_FIELDS_OPERATION_INIT)
     # Avg processing time init in stat vars
     self.updateStatField(DC_CONSTS.BATCHES_CRAWL_COUNTER_TIME_AVG_NAME, 0, self.STAT_FIELDS_OPERATION_INIT)
     # Crawling batches real-time threads number init in stat vars
     self.updateStatField(DC_CONSTS.BATCHES_REALTIME_THREADS_NAME, 0, self.STAT_FIELDS_OPERATION_SET)
     # Crawling batches real-time threads created total number init in stat vars
     self.updateStatField(DC_CONSTS.BATCHES_REALTIME_THREADS_CREATED_COUNTER_NAME, 0, self.STAT_FIELDS_OPERATION_SET)
 
     # Total time of processing sum from all request
     self.totalTime = 0
 
     # Set connections poll timeout, defines period of HCE cluster monitoring cycle, msec
     self.configVars[self.POLL_TIMEOUT_CONFIG_VAR_NAME] = configParser.getint(className, self.CONFIG_POLLING_TIMEOUT)
 
 

Member Function Documentation

◆ forkBatch()

def dc.BatchTasksManagerRealTime.BatchTasksManagerRealTime.forkBatch	(	self,
		loggingObj,
		event
	)

Definition at line 244 of file BatchTasksManagerRealTime.py.

   def forkBatch(self, loggingObj, event):
     try:
       global logger  # pylint: disable=W0603
       lock.acquire()
       logger = loggingObj.getLogger(DC_CONSTS.LOGGER_NAME)
       self.updateStatField(DC_CONSTS.BATCHES_REALTIME_THREADS_NAME, 1, self.STAT_FIELDS_OPERATION_ADD)
       lock.release()
       logger.info("THREAD_STARTED")
       logger.debug("event:\n%s", Utils.varDump(event))
       # Set start time
       t = time.time()
       # Set crawlerType to prevent wrong processing
       event.eventObj.crawlerType = EventObjects.Batch.TYPE_REAL_TIME_CRAWLER
       event.eventObj.dbMode = EventObjects.Batch.DB_MODE_R
       if event.eventObj.maxIterations > self.configVars[self.CONFIG_BATCH_MAX_ITERATIONS]:
         event.eventObj.maxIterations = self.configVars[self.CONFIG_BATCH_MAX_ITERATIONS]
       # Prepare request
       lock.acquire()
       taskExecuteRequest = self.prepareDRCERequest(event.eventObj)
       lock.release()
       # Process send request
       clientResponseObj = self.processDRCERequest(taskExecuteRequest)
       logger.debug("ClientResponseObj object:\n" + Utils.varDump(clientResponseObj))
       lock.acquire()
       self.totalTime = self.totalTime + (time.time() - t)
       self.updateStatField(DC_CONSTS.BATCHES_CRAWL_COUNTER_TIME_AVG_NAME,
                            str(self.totalTime / float(1 + self.statFields[DC_CONSTS.BATCHES_CRAWL_COUNTER_TOTAL_NAME])),
                            self.STAT_FIELDS_OPERATION_SET)
       # Send response to client
       self.sendClientResponse(event, clientResponseObj)
       logger.info("THREAD_FINISHED")
       logger.debug("clientResponseObj:\n%s", Utils.varDump(clientResponseObj))
       lock.release()
     except Exception as err:
       msg = "Thread exception:" + str(err)
       lock.acquire()
       logger.error(msg)
       clientResponseObj = EventObjects.ClientResponse()
       clientResponseObj.errorCode = self.REQUEST_ERROR_OBJECT_TYPE
       clientResponseObj.errorMessage = msg
       self.sendClientResponse(event, clientResponseObj)
       lock.release()
     except:  # pylint: disable=W0702
       msg = "Unknown thread exception!"
       lock.acquire()
       logger.error(msg)
       clientResponseObj = EventObjects.ClientResponse()
       clientResponseObj.errorCode = self.REQUEST_ERROR_OBJECT_TYPE
       clientResponseObj.errorMessage = msg
       self.sendClientResponse(event, clientResponseObj)
       lock.release()
     # Decrement of counter of threads
     lock.acquire()
     self.updateStatField(DC_CONSTS.BATCHES_REALTIME_THREADS_NAME, 1, self.STAT_FIELDS_OPERATION_SUB)
     lock.release()
 
 
 

Here is the call graph for this function:

Here is the caller graph for this function:

◆ on_poll_timeout()

def dc.BatchTasksManagerRealTime.BatchTasksManagerRealTime.on_poll_timeout ( self )

Definition at line 154 of file BatchTasksManagerRealTime.py.

   def on_poll_timeout(self):
     lock.acquire()
     # self.updateStatField(DC_CONSTS.BATCHES_REALTIME_THREADS_NAME, threading.active_count(),
     #                     self.STAT_FIELDS_OPERATION_SET)
     # Correct the number of clients to fix some crashes
     # if self.statFields[DC_CONSTS.BATCHES_REALTIME_THREADS_NAME] > threading.active_count():
     #  self.updateStatField(DC_CONSTS.BATCHES_REALTIME_THREADS_NAME, threading.active_count(),
     #                       self.STAT_FIELDS_OPERATION_SET)
     # Calc threads number
     n = 0
     main_thread = threading.currentThread()
     for t in threading.enumerate():
       if t is not main_thread and t.getName().startswith(self.REAL_TIME_CRAWL_THREAD_NAME_PREFIX):
         n += 1
     if self.statFields[DC_CONSTS.BATCHES_REALTIME_THREADS_NAME] > n:
       self.updateStatField(DC_CONSTS.BATCHES_REALTIME_THREADS_NAME, n, self.STAT_FIELDS_OPERATION_SET)
 
     lock.release()
 
 
 

Here is the call graph for this function:

◆ onEventsHandler()

def dc.BatchTasksManagerRealTime.BatchTasksManagerRealTime.onEventsHandler	(	self,
		event
	)

Definition at line 178 of file BatchTasksManagerRealTime.py.

   def onEventsHandler(self, event):
     self.updateStatField(DC_CONSTS.BATCHES_CRAWL_COUNTER_TOTAL_NAME, 1, self.STAT_FIELDS_OPERATION_ADD)
     if isinstance(event.eventObj, EventObjects.Batch):
       if len(event.eventObj.items) > int(self.configVars[self.CONFIG_BATCH_MAX_URLS]):
         clientResponseObj = EventObjects.ClientResponse()
         clientResponseObj.errorCode = self.REQUEST_ERROR_URLS_COUNT
         clientResponseObj.errorMessage = "Wrong requested object type " + str(len(event.eventObj.items)) + \
                                          ", Batch expected."
       else:
         if self.configVars[self.CONFIG_MAX_THREADS] == 0:
           logger.info("Single thread processing started")
           # Process batch in single thread
           self.forkBatch(logging, event)
           return
         else:
           if self.configVars[self.CONFIG_MAX_THREADS] > 0:
             # if self.configVars[self.CONFIG_MAX_THREADS] > threading.active_count():
             if self.configVars[self.CONFIG_MAX_THREADS] > self.statFields[DC_CONSTS.BATCHES_REALTIME_THREADS_NAME]:
               # Process batch in separated thread
               logger.info("Forking new thread")
               self.updateStatField(DC_CONSTS.BATCHES_REALTIME_THREADS_CREATED_COUNTER_NAME, 1,
                                    self.STAT_FIELDS_OPERATION_ADD)
               t1 = threading.Thread(target=self.forkBatch, args=(logging, event,))
               t1.setName(self.REAL_TIME_CRAWL_THREAD_NAME_PREFIX + \
                          str(self.statFields[DC_CONSTS.BATCHES_REALTIME_THREADS_CREATED_COUNTER_NAME]))
               t1.start()
               logger.info("New thread forked")
               # self.updateStatField(DC_CONSTS.BATCHES_REALTIME_THREADS_NAME, threading.active_count(),
               #                     self.STAT_FIELDS_OPERATION_SET)
               return
             else:
               # Return overload error
               clientResponseObj = EventObjects.ClientResponse()
               clientResponseObj.errorCode = self.REQUEST_ERROR_THREADS_NUMBER_EXCEEDED
               lock.acquire()
               clientResponseObj.errorMessage = "Service overloaded, " + \
                                             str(self.statFields[DC_CONSTS.BATCHES_REALTIME_THREADS_NAME]) + " workers."
               logger.error(clientResponseObj.errorMessage)
               lock.release()
           else:
             # Return fake error
             clientResponseObj = EventObjects.ClientResponse()
             clientResponseObj.errorCode = self.REQUEST_ERROR_OBJECT_TYPE
             clientResponseObj.errorMessage = "STUB fake error response"
             logger.info(clientResponseObj.errorMessage)
     else:
       # Return error
       clientResponseObj = EventObjects.ClientResponse()
       clientResponseObj.errorCode = self.REQUEST_ERROR_OBJECT_TYPE
       clientResponseObj.errorMessage = "Wrong requested object type " + type(event.eventObj) + ", Batch expected."
       logger.error(clientResponseObj.errorMessage)
 
     # Send response with error to client
     self.sendClientResponse(event, clientResponseObj)
 
     # self.updateStatField(DC_CONSTS.BATCHES_REALTIME_THREADS_NAME, threading.active_count(),
     #                     self.STAT_FIELDS_OPERATION_SET)
 
 
 

Here is the call graph for this function:

◆ prepareDRCERequest()

def dc.BatchTasksManagerRealTime.BatchTasksManagerRealTime.prepareDRCERequest	(	self,
		eventObj
	)

Definition at line 320 of file BatchTasksManagerRealTime.py.

   def prepareDRCERequest(self, eventObj):
     # Create DRCE task Id
     idGenerator = IDGenerator()
     taskId = ctypes.c_uint32(zlib.crc32(idGenerator.get_connection_uid(), int(time.time()))).value
 
     # Prepare DRCE request object
     taskExecuteStruct = TaskExecuteStruct()
     taskExecuteStruct.command = self.configVars[self.CONFIG_DRCE_CRAWLER_APP_NAME]
     if eventObj.id == 0:
       eventObj.id = taskId
     if eventObj.maxExecutionTime == 0:
       mt = self.configVars[self.CONFIG_BATCH_MAX_TIME]
     else:
       mt = eventObj.maxExecutionTime
       if int(mt) > int(self.configVars[self.CONFIG_DRCE_TIMEOUT] / 1000):
         mt = int(self.configVars[self.CONFIG_DRCE_TIMEOUT]) / 1000
     logger.debug("Custom max DRCE task execution set: %s", str(mt))
     taskExecuteStruct.input = pickle.dumps(eventObj)
     taskExecuteStruct.session = Session(Session.TMODE_SYNC, 0, int(mt) * 1000)
     taskExecuteStruct.session.shell = self.configVars[self.CONFIG_DRCE_STARTER_NAME]
     logger.debug("DRCE taskExecuteStruct:\n" + Utils.varDump(taskExecuteStruct))
 
     # Create DRCE TaskExecuteRequest object
     taskExecuteRequest = TaskExecuteRequest(taskId)
     # Set taskExecuteRequest fields
     taskExecuteRequest.data = taskExecuteStruct
     # Set route as resource-usage balancing if number of items in Batch is 1
     if len(eventObj.items) < 2:
       if self.configVars[self.CONFIG_DRCE_REQUEST_ROUTING] == 5:
         taskExecuteRequest.route = DC_CONSTS.DRCE_REQUEST_ROUTING_RESOURCE_USAGE
       if self.configVars[self.CONFIG_DRCE_REQUEST_ROUTING] == 1:
         taskExecuteRequest.route = DC_CONSTS.DRCE_REQUEST_ROUTING_ROUND_ROBIN
       if self.configVars[self.CONFIG_DRCE_REQUEST_ROUTING] == 0:
         taskExecuteRequest.route = DC_CONSTS.DRCE_REQUEST_ROUTING_MULTICAST
       if self.configVars[self.CONFIG_DRCE_REQUEST_ROUTING] == 4:
         taskExecuteRequest.route = DC_CONSTS.DRCE_REQUEST_ROUTING_RND
 
     logger.debug("DRCE taskExecuteRequest:\n" + Utils.varDump(taskExecuteRequest))
 
     return taskExecuteRequest
 
 
 

Here is the caller graph for this function:

◆ processDRCERequest()

def dc.BatchTasksManagerRealTime.BatchTasksManagerRealTime.processDRCERequest	(	self,
		taskExecuteRequest
	)

Definition at line 366 of file BatchTasksManagerRealTime.py.

   def processDRCERequest(self, taskExecuteRequest):
 
     logger.info("Sending sync task id:" + str(taskExecuteRequest.id) + " to DRCE router!")
     # Send request to DRCE Cluster router
     response = self.sendToDRCERouter(taskExecuteRequest)
     logger.info("Received response on sync task from DRCE router!")
     logger.debug("Response: %s", Utils.varDump(response))
 
     # Create new client response object
     clientResponse = EventObjects.ClientResponse()
 
     # Check response returned
     if response is None:
       clientResponse.errorCode = EventObjects.ClientResponse.STATUS_ERROR_NONE
       clientResponse.errorMessage = "Response error, None returned from DRCE, possible timeout " + \
                                     str(self.configVars[self.CONFIG_DRCE_TIMEOUT]) + " msec!"
       logger.error(clientResponse.errorMessage)
       lock.acquire()
       self.updateStatField(DC_CONSTS.BATCHES_CRAWL_COUNTER_FAULT_NAME, 1, self.STAT_FIELDS_OPERATION_ADD)
       lock.release()
     else:
       if len(response.items) == 0:
         clientResponse.errorCode = EventObjects.ClientResponse.STATUS_ERROR_EMPTY_LIST
         clientResponse.errorMessage = "Response error, empty list returned from DRCE, possible no one node in cluster!"
         logger.error(clientResponse.errorMessage)
         lock.acquire()
         self.updateStatField(DC_CONSTS.BATCHES_CRAWL_COUNTER_FAULT_NAME, 1, self.STAT_FIELDS_OPERATION_ADD)
         lock.release()
       else:
         for item in response.items:
           # New ClientResponseItem object
           clientResponseItem = EventObjects.ClientResponseItem(None)
           # If some error in response item or cli application exit status
           if item.error_code > 0 or item.exit_status > 0:
             clientResponseItem.errorCode = clientResponseItem.STATUS_ERROR_DRCE
             clientResponseItem.errorMessage = "error_message=" + item.error_message + \
                                               ", error_code=" + str(item.error_code) + \
                                               ", exit_status=" + str(item.exit_status) + \
                                               ", stderror=" + str(item.stderror)
             logger.error(clientResponseItem.errorMessage)
             lock.acquire()
             self.updateStatField(DC_CONSTS.BATCHES_CRAWL_COUNTER_URLS_FAULT_NAME, 1, self.STAT_FIELDS_OPERATION_ADD)
             lock.release()
           else:
             # Try to restore serialized response object from dump
             try:
               clientResponseItem.itemObject = pickle.loads(item.stdout)
               if clientResponseItem.itemObject is not None and isinstance(clientResponseItem.itemObject, list):
                 urlContents = len(clientResponseItem.itemObject)
                 lock.acquire()
                 self.updateStatField(DC_CONSTS.BATCHES_CRAWL_COUNTER_URLS_NAME, urlContents,
                                      self.STAT_FIELDS_OPERATION_ADD)
                 lock.release()
             except Exception as e:
               clientResponseItem.errorCode = EventObjects.ClientResponseItem.STATUS_ERROR_RESTORE_OBJECT
               clientResponseItem.errorMessage = EventObjects.ClientResponseItem.MSG_ERROR_RESTORE_OBJECT + "\n" + \
                                                 str(e.message) + "\nstdout=" + str(item.stdout) + \
                                                 ", stderror=" + str(item.stderror)
               logger.error(clientResponseItem.errorMessage)
               lock.acquire()
               self.updateStatField(DC_CONSTS.BATCHES_CRAWL_COUNTER_URLS_FAULT_NAME, 1, self.STAT_FIELDS_OPERATION_ADD)
               lock.release()
           # Set all information fields in any case
           clientResponseItem.id = item.id
           clientResponseItem.host = item.host
           clientResponseItem.port = item.port
           clientResponseItem.node = item.node
           clientResponseItem.time = item.time
           # Add ClientResponseItem object
           clientResponse.itemsList.append(clientResponseItem)
 
     return clientResponse
 
 
 

Here is the call graph for this function:

Here is the caller graph for this function:

◆ sendClientResponse()

def dc.BatchTasksManagerRealTime.BatchTasksManagerRealTime.sendClientResponse	(	self,
		clientRequestEvent,
		clientResponseObj
	)

Definition at line 306 of file BatchTasksManagerRealTime.py.

   def sendClientResponse(self, clientRequestEvent, clientResponseObj):
     # Prepare reply event
     replyEvent = self.eventBuilder.build(DC_CONSTS.EVENT_TYPES.BATCH_RESPONSE, clientResponseObj)
     # Send reply
     self.reply(clientRequestEvent, replyEvent)
     logger.info("Response to client sent")
 
 
 

Here is the call graph for this function:

Here is the caller graph for this function:

◆ sendToDRCERouter()

def dc.BatchTasksManagerRealTime.BatchTasksManagerRealTime.sendToDRCERouter	(	self,
		request
	)

Definition at line 445 of file BatchTasksManagerRealTime.py.

   def sendToDRCERouter(self, request):
     lock.acquire()
     drceManager = DRCEManager()
     drceManager.activate_host(HostParams(self.drceHost, self.drcePort))
     lock.release()
 
     logger.info("DRCE router sending with timeout=" + str(self.configVars[self.CONFIG_DRCE_TIMEOUT]) + \
                 ", host:" + str(self.drceHost) + ", port:" + str(self.drcePort))
     # Try to execute request
     try:
       response = drceManager.process(request, self.configVars[self.CONFIG_DRCE_TIMEOUT], self.DRCE_REDUCER_TTL)
     except (ConnectionTimeout, TransportInternalErr, CommandExecutorErr) as err:
       response = None
       logger.error("DRCE router send error : " + str(err.message))
 
     logger.info("DRCE router sent!")
 
     lock.acquire()
     drceManager.clear_host()
     lock.release()
 
     return response
 

Here is the caller graph for this function:

Member Data Documentation

◆ CONFIG_BATCH_MAX_ITERATIONS

string dc.BatchTasksManagerRealTime.BatchTasksManagerRealTime.CONFIG_BATCH_MAX_ITERATIONS = "BatchMaxIterations"

static

Definition at line 73 of file BatchTasksManagerRealTime.py.

◆ CONFIG_BATCH_MAX_ITERATIONS_DEFAULT

int dc.BatchTasksManagerRealTime.BatchTasksManagerRealTime.CONFIG_BATCH_MAX_ITERATIONS_DEFAULT = 2

static

Definition at line 58 of file BatchTasksManagerRealTime.py.

◆ CONFIG_BATCH_MAX_TIME

string dc.BatchTasksManagerRealTime.BatchTasksManagerRealTime.CONFIG_BATCH_MAX_TIME = "BatchMaxExecutionTime"

static

Definition at line 68 of file BatchTasksManagerRealTime.py.

◆ CONFIG_BATCH_MAX_URLS

string dc.BatchTasksManagerRealTime.BatchTasksManagerRealTime.CONFIG_BATCH_MAX_URLS = "BatchMaxURLs"

static

Definition at line 69 of file BatchTasksManagerRealTime.py.

◆ CONFIG_DRCE_CRAWLER_APP_NAME

string dc.BatchTasksManagerRealTime.BatchTasksManagerRealTime.CONFIG_DRCE_CRAWLER_APP_NAME = "DRCECrawlerAppName"

static

Definition at line 67 of file BatchTasksManagerRealTime.py.

◆ CONFIG_DRCE_HOST

string dc.BatchTasksManagerRealTime.BatchTasksManagerRealTime.CONFIG_DRCE_HOST = "DRCEHost"

static

Definition at line 64 of file BatchTasksManagerRealTime.py.

◆ CONFIG_DRCE_PORT

string dc.BatchTasksManagerRealTime.BatchTasksManagerRealTime.CONFIG_DRCE_PORT = "DRCEPort"

static

Definition at line 65 of file BatchTasksManagerRealTime.py.

◆ CONFIG_DRCE_REQUEST_ROUTING

string dc.BatchTasksManagerRealTime.BatchTasksManagerRealTime.CONFIG_DRCE_REQUEST_ROUTING = "DRCERequestRouting"

static

Definition at line 72 of file BatchTasksManagerRealTime.py.

◆ CONFIG_DRCE_REQUEST_ROUTING_DEFAULT

int dc.BatchTasksManagerRealTime.BatchTasksManagerRealTime.CONFIG_DRCE_REQUEST_ROUTING_DEFAULT = 1

static

Definition at line 57 of file BatchTasksManagerRealTime.py.

◆ CONFIG_DRCE_STARTER_NAME

string dc.BatchTasksManagerRealTime.BatchTasksManagerRealTime.CONFIG_DRCE_STARTER_NAME = "DRCEStarterName"

static

Definition at line 63 of file BatchTasksManagerRealTime.py.

◆ CONFIG_DRCE_TIMEOUT

string dc.BatchTasksManagerRealTime.BatchTasksManagerRealTime.CONFIG_DRCE_TIMEOUT = "DRCETimeout"

static

Definition at line 66 of file BatchTasksManagerRealTime.py.

◆ CONFIG_MAX_THREADS

string dc.BatchTasksManagerRealTime.BatchTasksManagerRealTime.CONFIG_MAX_THREADS = "MaxThreads"

static

Definition at line 70 of file BatchTasksManagerRealTime.py.

◆ CONFIG_POLLING_TIMEOUT

string dc.BatchTasksManagerRealTime.BatchTasksManagerRealTime.CONFIG_POLLING_TIMEOUT = "PollingTimeout"

static

Definition at line 71 of file BatchTasksManagerRealTime.py.

◆ CONFIG_SERVER

string dc.BatchTasksManagerRealTime.BatchTasksManagerRealTime.CONFIG_SERVER = "server"

static

Definition at line 61 of file BatchTasksManagerRealTime.py.

◆ DRCE_REDUCER_TTL

int dc.BatchTasksManagerRealTime.BatchTasksManagerRealTime.DRCE_REDUCER_TTL = 3000000

static

Definition at line 53 of file BatchTasksManagerRealTime.py.

◆ drceCommandConvertor

dc.BatchTasksManagerRealTime.BatchTasksManagerRealTime.drceCommandConvertor

Definition at line 105 of file BatchTasksManagerRealTime.py.

◆ drceHost

dc.BatchTasksManagerRealTime.BatchTasksManagerRealTime.drceHost

Definition at line 100 of file BatchTasksManagerRealTime.py.

◆ drceIdGenerator

dc.BatchTasksManagerRealTime.BatchTasksManagerRealTime.drceIdGenerator

Definition at line 104 of file BatchTasksManagerRealTime.py.

◆ drcePort

dc.BatchTasksManagerRealTime.BatchTasksManagerRealTime.drcePort

Definition at line 101 of file BatchTasksManagerRealTime.py.

◆ REAL_TIME_CRAWL_THREAD_NAME_PREFIX

string dc.BatchTasksManagerRealTime.BatchTasksManagerRealTime.REAL_TIME_CRAWL_THREAD_NAME_PREFIX = 'RtCrawl_'

static

Definition at line 75 of file BatchTasksManagerRealTime.py.

◆ REQUEST_ERROR_OBJECT_TYPE

int dc.BatchTasksManagerRealTime.BatchTasksManagerRealTime.REQUEST_ERROR_OBJECT_TYPE = 1

static

Definition at line 54 of file BatchTasksManagerRealTime.py.

◆ REQUEST_ERROR_THREADS_NUMBER_EXCEEDED

int dc.BatchTasksManagerRealTime.BatchTasksManagerRealTime.REQUEST_ERROR_THREADS_NUMBER_EXCEEDED = 3

static

Definition at line 56 of file BatchTasksManagerRealTime.py.

◆ REQUEST_ERROR_URLS_COUNT

int dc.BatchTasksManagerRealTime.BatchTasksManagerRealTime.REQUEST_ERROR_URLS_COUNT = 2

static

Definition at line 55 of file BatchTasksManagerRealTime.py.

◆ serverName

dc.BatchTasksManagerRealTime.BatchTasksManagerRealTime.serverName

Definition at line 92 of file BatchTasksManagerRealTime.py.

◆ totalTime

dc.BatchTasksManagerRealTime.BatchTasksManagerRealTime.totalTime

Definition at line 145 of file BatchTasksManagerRealTime.py.

The documentation for this class was generated from the following file:

sources/hce/dc/BatchTasksManagerRealTime.py

Public Member Functions

Public Attributes

Static Public Attributes

Detailed Description

Constructor & Destructor Documentation

◆ __init__()

Member Function Documentation

◆ forkBatch()

◆ on_poll_timeout()

◆ onEventsHandler()

◆ prepareDRCERequest()

◆ processDRCERequest()

◆ sendClientResponse()

◆ sendToDRCERouter()

Member Data Documentation

◆ CONFIG_BATCH_MAX_ITERATIONS

◆ CONFIG_BATCH_MAX_ITERATIONS_DEFAULT

◆ CONFIG_BATCH_MAX_TIME

◆ CONFIG_BATCH_MAX_URLS

◆ CONFIG_DRCE_CRAWLER_APP_NAME

◆ CONFIG_DRCE_HOST

◆ CONFIG_DRCE_PORT

◆ CONFIG_DRCE_REQUEST_ROUTING

◆ CONFIG_DRCE_REQUEST_ROUTING_DEFAULT

◆ CONFIG_DRCE_STARTER_NAME

◆ CONFIG_DRCE_TIMEOUT

◆ CONFIG_MAX_THREADS

◆ CONFIG_POLLING_TIMEOUT

◆ CONFIG_SERVER

◆ DRCE_REDUCER_TTL

◆ drceCommandConvertor

◆ drceHost

◆ drceIdGenerator

◆ drcePort

◆ REAL_TIME_CRAWL_THREAD_NAME_PREFIX

◆ REQUEST_ERROR_OBJECT_TYPE

◆ REQUEST_ERROR_THREADS_NUMBER_EXCEEDED

◆ REQUEST_ERROR_URLS_COUNT

◆ serverName

◆ totalTime

◆ init()