import threading
import time
from motor.common.resources import RegisterMsg, ReregisterMsg, HeartbeatMsg
from motor.common.http.http_client import ConnectionMode, SafeHTTPSClient
from motor.common.logger import get_logger
from motor.common.logger.rate_limited_logger import RateLimitedLogger
from motor.config.controller import ControllerConfig
from motor.config.node_manager import NodeManagerConfig
logger = get_logger(__name__)
_rl = RateLimitedLogger(logger)
_HEARTBEAT_CLIENT: SafeHTTPSClient | None = None
_HEARTBEAT_CLIENT_LOCK = threading.Lock()
_HEARTBEAT_REQUEST_LOCK = threading.Lock()
_HEARTBEAT_TIMEOUT = 5
_HEARTBEAT_RETRY_DELAYS = (1, 2)
def _get_heartbeat_client() -> SafeHTTPSClient:
"""Return the shared long-lived heartbeat HTTP client, creating it on first use."""
global _HEARTBEAT_CLIENT
if _HEARTBEAT_CLIENT is not None:
return _HEARTBEAT_CLIENT
with _HEARTBEAT_CLIENT_LOCK:
if _HEARTBEAT_CLIENT is not None:
return _HEARTBEAT_CLIENT
client_args = ControllerApiClient._generate_client_args()
_HEARTBEAT_CLIENT = SafeHTTPSClient(
mode=ConnectionMode.LONG,
timeout=_HEARTBEAT_TIMEOUT,
**client_args,
)
logger.info(
"Heartbeat HTTP client created (Keep-Alive, timeout=%ds) → %s",
_HEARTBEAT_TIMEOUT,
client_args.get("address", "unknown"),
)
return _HEARTBEAT_CLIENT
def _reset_heartbeat_client() -> None:
"""Close and discard the shared heartbeat client so the next call creates a fresh one."""
global _HEARTBEAT_CLIENT
with _HEARTBEAT_CLIENT_LOCK:
if _HEARTBEAT_CLIENT is not None:
try:
_HEARTBEAT_CLIENT.close()
except Exception:
pass
_HEARTBEAT_CLIENT = None
class ControllerApiClient:
controller_config = ControllerConfig.from_json()
nodemanager_config = NodeManagerConfig.from_json()
@staticmethod
def register(register_msg: RegisterMsg) -> bool:
client_args = {}
try:
client_args = ControllerApiClient._generate_client_args()
with SafeHTTPSClient(timeout=15, **client_args) as client:
response = client.post("/controller/register", register_msg.model_dump())
except Exception as e:
logger.error(
"Exception occurred while register to controller at %s: %s", client_args.get("address", "unknown"), e
)
return False
if not isinstance(response, dict):
logger.error("Invalid register response from controller: %s", response)
return False
if error := response.get("error"):
logger.warning("Register rejected by controller: %s", error)
return False
logger.info("Register success!")
return True
@staticmethod
def re_register(re_register_msg: ReregisterMsg):
client_args = {}
try:
client_args = ControllerApiClient._generate_client_args()
with SafeHTTPSClient(timeout=15, **client_args) as client:
_ = client.post("/controller/reregister", re_register_msg.model_dump())
logger.info("Register success!")
return True
except Exception as e:
logger.error(
"Exception occurred while reregister to controller at %s: %s", client_args.get("address", "unknown"), e
)
return False
@staticmethod
def report_heartbeat(heartbeat_msg: HeartbeatMsg):
"""Send heartbeat to Controller over a persistent Keep-Alive connection.
On TCP-level failure the stale client is discarded and up to two retries
with back-off (1 s, 2 s) are attempted before propagating the exception.
The shared Keep-Alive client is guarded by a request lock so that
concurrent callers cannot interleave requests on the same connection
or reset the client while another thread is using it.
"""
last_error: Exception | None = None
total_attempts = 1 + len(_HEARTBEAT_RETRY_DELAYS)
for attempt in range(total_attempts):
with _HEARTBEAT_REQUEST_LOCK:
try:
client = _get_heartbeat_client()
response = client.post("/controller/heartbeat", heartbeat_msg.model_dump())
except Exception as exc:
last_error = exc
_reset_heartbeat_client()
else:
_rl.record_success("node_manager.controller.report_heartbeat")
_rl.emit_periodic(
"node_manager.controller.report_heartbeat",
"NodeManager->Controller report_heartbeat periodic summary: succeeded {count} times in last 60s",
level="DEBUG",
)
logger.debug(
"Heartbeat success (attempt %d/%d), response: %s",
attempt + 1,
total_attempts,
response,
)
return
if attempt < len(_HEARTBEAT_RETRY_DELAYS):
delay = _HEARTBEAT_RETRY_DELAYS[attempt]
logger.debug(
"Heartbeat attempt %d/%d failed (%s), retrying in %ds...",
attempt + 1,
total_attempts,
last_error,
delay,
)
time.sleep(delay)
raise last_error
@staticmethod
def report_software_fault(fault_data: dict):
"""Report a software fault to the Controller.
Args:
fault_data: dict with keys: exception_type, exception_message,
engine_id, engine_status, pod_ip, additional_info
"""
client_args = {}
try:
client_args = ControllerApiClient._generate_client_args()
with SafeHTTPSClient(timeout=15, **client_args) as client:
response = client.post("/controller/report_software_fault", fault_data)
logger.debug("Software fault reported successfully, response: %s", response)
return True
except Exception as e:
logger.error(
"Exception occurred while reporting software fault to controller at %s: %s",
client_args.get("address", "unknown"),
e,
)
return False
@classmethod
def _generate_client_args(cls) -> dict[str, str]:
api_config = cls.controller_config.api_config
tls_config = cls.nodemanager_config.mgmt_tls_config
address = f"{api_config.controller_api_dns}:{api_config.controller_api_port}"
client_ars = {"address": f"{address}", "tls_config": tls_config}
return client_ars