diff --git a/Sources/pyOpenRPA/Agent/O2A.py b/Sources/pyOpenRPA/Agent/O2A.py index 859f1485..853df131 100644 --- a/Sources/pyOpenRPA/Agent/O2A.py +++ b/Sources/pyOpenRPA/Agent/O2A.py @@ -21,6 +21,11 @@ def O2A_Loop(inGSettings): while inGSettings["O2ADict"]["IsOnlineBool"]: # Send request to the orchestrator server lRequestBody = None + # ConnectionError - CE + lCEPhaseFastTimeLastGoodFloat = time.time() + lCEPhaseFastDurationSecFloat = inGSettings['O2ADict']['ConnectionTimeoutSecFloat'] + lCEPhaseFastRetrySecFloat = inGSettings['O2ADict']['RetryTimeoutSecFloat']/5.0 + lCEPhaseLongRetrySecFloat = inGSettings['O2ADict']['RetryTimeoutSecFloat']*12.0 try: lProtocolStr= "https" if inGSettings["OrchestratorDict"]["IsHTTPSBool"] else "http" lHostStr = inGSettings["OrchestratorDict"]["HostStr"] @@ -28,6 +33,7 @@ def O2A_Loop(inGSettings): lURLStr=f"{lProtocolStr}://{lHostStr}:{lPortInt}/pyOpenRPA/Agent/O2A" lDataDict = { "HostNameUpperStr": inGSettings["AgentDict"]["HostNameUpperStr"], "UserUpperStr": inGSettings["AgentDict"]["UserUpperStr"], "ActivityLastGUIDStr": lActivityLastGUIDStr} lResponse = requests.post(url= lURLStr, cookies = {"AuthToken":inGSettings["OrchestratorDict"]["SuperTokenStr"]}, json=lDataDict, timeout=inGSettings["O2ADict"]["ConnectionTimeoutSecFloat"]) + lCEPhaseFastTimeLastGoodFloat = time.time() if lResponse.status_code != 200: if lL: lL.warning(f"Agent can not connect to Orchestrator. Below the response from the orchestrator:{lResponse}") time.sleep(inGSettings["O2ADict"]["RetryTimeoutSecFloat"]) @@ -53,17 +59,21 @@ def O2A_Loop(inGSettings): else: if lL: lL.debug(f"Empty response from the orchestrator - loop when refresh the connection between Orc and Agent"); except requests.exceptions.ConnectionError as e: - if lL: lL.error(f"O2A Connection error - orchestrator is not available. Sleep for {inGSettings['A2ODict']['RetryTimeoutSecFloat']} s.") - time.sleep(inGSettings["O2ADict"]["RetryTimeoutSecFloat"]) + if time.time() - lCEPhaseFastTimeLastGoodFloat <= lCEPhaseFastDurationSecFloat: + if lL: lL.error(f"O2A Connection error - orchestrator is not available. Sleep for {lCEPhaseFastRetrySecFloat} s.") + time.sleep(lCEPhaseFastRetrySecFloat) + else: + if lL: lL.error(f"O2A Connection error - orchestrator is not available. Sleep for {lCEPhaseLongRetrySecFloat} s.") + time.sleep(lCEPhaseLongRetrySecFloat) except ConnectionResetError as e: - if lL: lL.error(f"O2A Connection reset error - orchestrator is not available. Sleep for {inGSettings['A2ODict']['RetryTimeoutSecFloat']} s.") + if lL: lL.error(f"O2A Connection reset error - orchestrator is not available. Sleep for {inGSettings['O2ADict']['RetryTimeoutSecFloat']} s.") time.sleep(inGSettings["O2ADict"]["RetryTimeoutSecFloat"]) except json.decoder.JSONDecodeError as e: if lL: lL.error(f"O2A JSON decode error - See body of the recieved content from the Orchestrator: {lRequestBody}") time.sleep(inGSettings["O2ADict"]["RetryTimeoutSecFloat"]) except requests.exceptions.Timeout as e: - if lL: lL.exception(f"O2A requests timeout error (no response for long time). Sleep for {inGSettings['A2ODict']['RetryTimeoutSecFloat']} s.") + if lL: lL.exception(f"O2A requests timeout error (no response for long time). Sleep for {inGSettings['O2ADict']['RetryTimeoutSecFloat']} s.") time.sleep(inGSettings["O2ADict"]["RetryTimeoutSecFloat"]) except Exception as e: - if lL: lL.exception(f"O2A Error handler. Sleep for {inGSettings['A2ODict']['RetryTimeoutSecFloat']} s.") + if lL: lL.exception(f"O2A Error handler. Sleep for {inGSettings['O2ADict']['RetryTimeoutSecFloat']} s.") time.sleep(inGSettings["O2ADict"]["RetryTimeoutSecFloat"]) \ No newline at end of file diff --git a/Sources/pyOpenRPA/Agent/__Agent__.py b/Sources/pyOpenRPA/Agent/__Agent__.py index 4656b138..f7d448de 100644 --- a/Sources/pyOpenRPA/Agent/__Agent__.py +++ b/Sources/pyOpenRPA/Agent/__Agent__.py @@ -128,6 +128,9 @@ def OSCMD(inCMDStr, inRunAsyncBool=True, inGSettings = None, inSendOutputToOrche :return: """ lResultStr = "" + # New feature + if inSendOutputToOrchestratorLogsBool == False and inCaptureBool == False: + inCMDStr = f"start {inCMDStr}" # Subdef to listen OS result def _CMDRunAndListenLogs(inCMDStr, inSendOutputToOrchestratorLogsBool, inCMDEncodingStr, inGSettings = None, inCaptureBool = True): lL = inGSettings.get("Logger",None) if type(inGSettings) is dict else None diff --git a/Sources/pyOpenRPA/Orchestrator/Managers/Process.py b/Sources/pyOpenRPA/Orchestrator/Managers/Process.py index 4248fde3..276e4adc 100644 --- a/Sources/pyOpenRPA/Orchestrator/Managers/Process.py +++ b/Sources/pyOpenRPA/Orchestrator/Managers/Process.py @@ -44,6 +44,8 @@ class Process(): mAgentUserNameStr = None mStartPathStr = None mStartCMDStr = None + mStartArgDict = None + mStatusCheckIntervalSecFloat = None mProcessNameWOExeStr = None mStopSafeTimeoutSecFloat = None mStatusStr = None # 0_STOPPED 1_STOPPED_MANUAL 2_STOP_SAFE 3_STOP_SAFE_MANUAL 4_STARTED 5_STARTED_MANUAL @@ -76,7 +78,7 @@ class Process(): return (self.mAgentHostNameStr.upper(), self.mAgentUserNameStr.upper(), self.mProcessNameWOExeStr.upper()) - def __init__(self, inAgentHostNameStr, inAgentUserNameStr, inProcessNameWOExeStr, inStartPathStr=None, inStartCMDStr = None, inStopSafeTimeoutSecFloat=300): + def __init__(self, inAgentHostNameStr, inAgentUserNameStr, inProcessNameWOExeStr, inStartPathStr=None, inStartCMDStr = None, inStopSafeTimeoutSecFloat=300, inStartArgDict=None, inStatusCheckIntervalSecFloat=30): """ Init the class instance. !ATTENTION! Function can raise exception if process with the same (inAgentHostNameStr, inAgentUserNameStr, inProcessNameWOExeStr) is already exists in GSettings (can be restored from previous Orchestrator session). See ProcessInitSafe to sefaty init the instance or restore previous @@ -92,6 +94,7 @@ class Process(): lGS = __Orchestrator__.GSettingsGet() # Check if Process is not exists in GSettings if (inAgentHostNameStr.upper(), inAgentUserNameStr.upper(), inProcessNameWOExeStr.upper()) not in lGS["ManagersProcessDict"]: + self.mStartArgDict = inStartArgDict self.mAgentHostNameStr = inAgentHostNameStr self.mAgentUserNameStr = inAgentUserNameStr self.mStartPathStr = inStartPathStr @@ -101,6 +104,8 @@ class Process(): lGS["ManagersProcessDict"][(inAgentHostNameStr.upper(), inAgentUserNameStr.upper(), inProcessNameWOExeStr.upper())]=self lActivityDict = __Orchestrator__.ProcessorActivityItemCreate(inDef=self.StatusCheck,inArgList=[], inThreadBool=True) __Orchestrator__.ProcessorActivityItemAppend(inActivityItemDict=lActivityDict) + if inStatusCheckIntervalSecFloat is not None: __Orchestrator__.OrchestratorScheduleGet().every(inStatusCheckIntervalSecFloat).seconds.do(Orchestrator.OrchestratorThreadStart,self.StatusCheck) + self.mStatusCheckIntervalSecFloat = inStatusCheckIntervalSecFloat else: raise Exception(f"Managers.Process ({inAgentHostNameStr}, {inAgentUserNameStr}, {inProcessNameWOExeStr}): Can't init the Process instance because it already inited in early (see ProcessInitSafe)") def ManualStopTriggerSet(self, inMSTdTSecFloat: float, inMSTdNInt: int) -> None: @@ -163,7 +168,7 @@ class Process(): if lLogBool == True: self.StatusChangeLog() return self.mStatusStr - def Start(self, inIsManualBool = True) -> str: + def Start(self, inIsManualBool = True, inStartArgDict=None) -> str: """ Manual/Auto start. Manual start will block scheduling execution. To return schedule execution use def Manual2Auto. Will not start if STOP SAFE is now and don't start auto is stopped manual now @@ -177,8 +182,14 @@ class Process(): __Orchestrator__.OrchestratorLoggerGet().warning(lStr) return self.mStatusStr # Send activity item to agent - wait result - if self.mStartPathStr is not None: lCMDStr = f"start {os.path.abspath(self.mStartPathStr)}" - elif self.mStartCMDStr is not None: lCMDStr = f"start {self.mStartCMDStr}" + if self.mStartPathStr is not None: lCMDStr = os.path.abspath(self.mStartPathStr) + elif self.mStartCMDStr is not None: lCMDStr = self.mStartCMDStr + # Append args + if inStartArgDict is not None: self.mStartArgDict = inStartArgDict + if self.mStartArgDict is not None: + for lItemKeyStr in self.mStartArgDict: + lItemValueStr = self.mStartArgDict[lItemKeyStr] + lCMDStr = f"{lCMDStr} {lItemKeyStr} {lItemValueStr}" #import pdb #pdb.set_trace() self.MuteWait() @@ -340,6 +351,14 @@ class Process(): if lWarnSafeBool==True: __Orchestrator__.OrchestratorLoggerGet().warning(f"Managers.Process ({self.mAgentHostNameStr}, {self.mAgentUserNameStr}, {self.mProcessNameWOExeStr}): Safe status has been catched when safe > change saved status to stopped.") return self.mStatusStr + + def StatusCheckIntervalRestore(self): + """Call from orchestrator when init + """ + if self.mStatusCheckIntervalSecFloat is not None: + __Orchestrator__.OrchestratorLoggerGet().info(f"Managers.Process ({self.mAgentHostNameStr}, {self.mAgentUserNameStr}, {self.mProcessNameWOExeStr}): Restore schedule to StatusCheck in interval of {self.mStatusCheckIntervalSecFloat} sec.") + __Orchestrator__.OrchestratorScheduleGet().every(self.mStatusCheckIntervalSecFloat).seconds.do(Orchestrator.OrchestratorThreadStart,self.StatusCheck) + def StatusRestore(self): """ Execute the StatusCheck, after that restore status to the saved state (see StatusSave). Work when orchestrator is restarted. @@ -349,14 +368,15 @@ class Process(): self.StatusCheck() # check current status # Do some action if self.mStatusSavedStr != self.mStatusStr and self.mStatusSavedStr is not None: - lManualBool = False - if "MANUAL" in self.mStatusSavedStr: - lManualBool = True + #lManualBool = False + #if "MANUAL" in self.mStatusSavedStr: + # lManualBool = True if "STOPPED" in self.mStatusSavedStr and "STOPPED" not in self.mStatusStr: - self.StopSafe(inIsManualBool=lManualBool) + self.StopSafe(inIsManualBool=True) if "STARTED" in self.mStatusSavedStr and "STARTED" not in self.mStatusStr: - self.Start(inIsManualBool=lManualBool) + self.Start(inIsManualBool=True) Orchestrator.OrchestratorLoggerGet().info(f"Managers.Process ({self.mAgentHostNameStr}, {self.mAgentUserNameStr}, {self.mProcessNameWOExeStr}): Status has been restored to {self.mStatusSavedStr}") + self.mStatusStr = self.mStatusSavedStr self.mStatusSavedStr = None return self.mStatusStr diff --git a/Sources/pyOpenRPA/Orchestrator/__Orchestrator__.py b/Sources/pyOpenRPA/Orchestrator/__Orchestrator__.py index e65a451b..edbb7c0f 100644 --- a/Sources/pyOpenRPA/Orchestrator/__Orchestrator__.py +++ b/Sources/pyOpenRPA/Orchestrator/__Orchestrator__.py @@ -483,6 +483,9 @@ def OSCMD(inCMDStr, inRunAsyncBool=True, inLogger = None): """ if inLogger is None: inLogger = OrchestratorLoggerGet() lResultStr = "" + # New feature + if inRunAsyncBool == True: + inCMDStr = f"start {inCMDStr}" # Subdef to listen OS result def _CMDRunAndListenLogs(inCMDStr, inLogger): lResultStr = "" @@ -2827,6 +2830,7 @@ def Orchestrator(inGSettings=None, inDumpRestoreBool = True, inRunAsAdministrato # Restore state for process for lProcessKeyTuple in inGSettings["ManagersProcessDict"]: lProcess = inGSettings["ManagersProcessDict"][lProcessKeyTuple] + lProcess.StatusCheckIntervalRestore() lThread = threading.Thread(target= lProcess.StatusRestore) lThread.start()