implements reconnection in ogagent for oglive
ogagent/pipeline/head This commit looks good
Details
ogagent/pipeline/head This commit looks good
Details
parent
f50d5a1197
commit
70a9d4ad51
|
|
@ -5,6 +5,12 @@ All notable changes to this project will be documented in this file.
|
||||||
The format is based on [Keep a Changelog](https://keepachangelog.com/en/1.1.0/),
|
The format is based on [Keep a Changelog](https://keepachangelog.com/en/1.1.0/),
|
||||||
and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0.html).
|
and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0.html).
|
||||||
|
|
||||||
|
## [8.5.3] - 2026-04-30
|
||||||
|
|
||||||
|
### Fixed
|
||||||
|
|
||||||
|
- Reconnect to ogcore after a connection loss in oglive: the monitoring thread now detects when contact is restored and re-sends `InclusionCliente` so ogcore marks the client as OG_LIVE and refreshes its token
|
||||||
|
|
||||||
## [8.5.2] - 2026-04-28
|
## [8.5.2] - 2026-04-28
|
||||||
|
|
||||||
### Fixed
|
### Fixed
|
||||||
|
|
|
||||||
|
|
@ -149,6 +149,13 @@ class ogAdmClientWorker (ogLiveWorker):
|
||||||
logger.error ('Ha ocurrido algún problema al procesar la trama recibida')
|
logger.error ('Ha ocurrido algún problema al procesar la trama recibida')
|
||||||
break
|
break
|
||||||
|
|
||||||
|
def _on_reconnect(self):
|
||||||
|
logger.info('Re-registering with ogcore after reconnection')
|
||||||
|
if self.inclusionCliente():
|
||||||
|
self._registered = True
|
||||||
|
else:
|
||||||
|
logger.warning('Re-registration failed, will retry on next ping')
|
||||||
|
|
||||||
def inclusionCliente (self):
|
def inclusionCliente (self):
|
||||||
cfg = self.LeeConfiguracion()
|
cfg = self.LeeConfiguracion()
|
||||||
if not cfg:
|
if not cfg:
|
||||||
|
|
|
||||||
|
|
@ -67,6 +67,7 @@ class ThreadWithResult (threading.Thread):
|
||||||
class ogLiveWorker(ServerWorker):
|
class ogLiveWorker(ServerWorker):
|
||||||
thread_list = {}
|
thread_list = {}
|
||||||
thread_lock = threading.Lock()
|
thread_lock = threading.Lock()
|
||||||
|
_registered = False # Whether ogcore has acknowledged our registration
|
||||||
|
|
||||||
tbErroresScripts = [
|
tbErroresScripts = [
|
||||||
"Se han generado errores desconocidos. No se puede continuar la ejecución de este módulo", ## 0
|
"Se han generado errores desconocidos. No se puede continuar la ejecución de este módulo", ## 0
|
||||||
|
|
@ -147,6 +148,10 @@ class ogLiveWorker(ServerWorker):
|
||||||
"Error desconocido",
|
"Error desconocido",
|
||||||
]
|
]
|
||||||
|
|
||||||
|
def _on_reconnect(self):
|
||||||
|
"""Hook called by mon() when connection to ogcore is restored. Override in subclasses."""
|
||||||
|
pass
|
||||||
|
|
||||||
def notifier (self, job_id, result):
|
def notifier (self, job_id, result):
|
||||||
result['job_id'] = job_id
|
result['job_id'] = job_id
|
||||||
self.REST.sendMessage ('clients/status/webhook', result)
|
self.REST.sendMessage ('clients/status/webhook', result)
|
||||||
|
|
@ -261,8 +266,14 @@ class ogLiveWorker(ServerWorker):
|
||||||
'iph': self.IPlocal,
|
'iph': self.IPlocal,
|
||||||
'timestamp': int (time.time()),
|
'timestamp': int (time.time()),
|
||||||
}
|
}
|
||||||
#logger.debug (f'about to send ping ({body})')
|
result = self.REST.sendMessage('clients/status/webhook', body)
|
||||||
self.REST.sendMessage ('clients/status/webhook', body)
|
if result is None:
|
||||||
|
if self._registered:
|
||||||
|
logger.debug('Lost connection to ogcore')
|
||||||
|
self._registered = False
|
||||||
|
elif not self._registered:
|
||||||
|
logger.info('Connection to ogcore restored, re-registering')
|
||||||
|
self._on_reconnect()
|
||||||
|
|
||||||
def interfaceAdmin (self, method, parametros=[]):
|
def interfaceAdmin (self, method, parametros=[]):
|
||||||
if method not in ['ConsolaRemota', 'procesaCache']:
|
if method not in ['ConsolaRemota', 'procesaCache']:
|
||||||
|
|
@ -537,7 +548,8 @@ class ogLiveWorker(ServerWorker):
|
||||||
logger.warning ('onActivation failed')
|
logger.warning ('onActivation failed')
|
||||||
return
|
return
|
||||||
|
|
||||||
threading.Thread (name='monitoring_thread', target=self.mon, daemon=True).start()
|
self._registered = True
|
||||||
|
threading.Thread(name='monitoring_thread', target=self.mon, daemon=True).start()
|
||||||
|
|
||||||
def _long_running_job (self, name, f, args):
|
def _long_running_job (self, name, f, args):
|
||||||
any_job_running = False
|
any_job_running = False
|
||||||
|
|
|
||||||
Loading…
Reference in New Issue