0ebbb172f0
- Автоматичне створення streams при старті worker - Перевірка наявності streams перед створенням - Підтримка всіх 4 streams (MM_ONLINE, MM_OFFLINE, MM_WRITE, MM_EVENTS) Це вирішує проблему з DNS в K8s Job
121 lines
3.9 KiB
Python
121 lines
3.9 KiB
Python
#!/usr/bin/env python3
|
|
"""
|
|
Worker Daemon для Memory Module
|
|
Реєстрація capabilities, підписка на NATS streams, виконання jobs
|
|
"""
|
|
|
|
import asyncio
|
|
import os
|
|
import signal
|
|
import sys
|
|
from typing import Optional
|
|
|
|
from worker.registry import CapabilityRegistry
|
|
from worker.nats_client import NATSClient
|
|
from worker.job_executor import JobExecutor
|
|
from worker.metrics import MetricsExporter
|
|
from worker.stream_creator import StreamCreator
|
|
|
|
|
|
class WorkerDaemon:
|
|
def __init__(self):
|
|
self.node_id = os.getenv("NODE_ID", "unknown")
|
|
self.tier = os.getenv("TIER", "C")
|
|
self.region = os.getenv("REGION", "unknown")
|
|
self.nats_url = os.getenv("NATS_URL", "nats://nats-client.nats:4222")
|
|
self.postgres_url = os.getenv("CAPABILITY_REGISTRY", "")
|
|
|
|
self.registry: Optional[CapabilityRegistry] = None
|
|
self.nats_client: Optional[NATSClient] = None
|
|
self.job_executor: Optional[JobExecutor] = None
|
|
self.metrics: Optional[MetricsExporter] = None
|
|
|
|
self.running = False
|
|
|
|
async def start(self):
|
|
"""Запуск worker daemon"""
|
|
print(f"🚀 Worker Daemon запускається...")
|
|
print(f" Node ID: {self.node_id}")
|
|
print(f" Tier: {self.tier}")
|
|
print(f" NATS URL: {self.nats_url}")
|
|
|
|
# Ініціалізація компонентів
|
|
self.registry = CapabilityRegistry(self.postgres_url, self.node_id, self.tier, self.region)
|
|
self.nats_client = NATSClient(self.nats_url)
|
|
self.job_executor = JobExecutor(self.node_id, self.tier)
|
|
self.metrics = MetricsExporter(port=9090)
|
|
|
|
# Реєстрація capabilities
|
|
await self.registry.register()
|
|
|
|
# Підключення до NATS
|
|
await self.nats_client.connect()
|
|
|
|
# Створення streams якщо не існують
|
|
stream_creator = StreamCreator(self.nats_url)
|
|
await stream_creator.connect()
|
|
await stream_creator.create_streams_if_not_exist()
|
|
await stream_creator.disconnect()
|
|
|
|
# Підписка на streams
|
|
await self.nats_client.subscribe_streams(self.job_executor)
|
|
|
|
# Запуск metrics server
|
|
await self.metrics.start()
|
|
|
|
# Heartbeat loop
|
|
self.running = True
|
|
asyncio.create_task(self.heartbeat_loop())
|
|
|
|
print("✅ Worker Daemon запущено")
|
|
|
|
async def heartbeat_loop(self):
|
|
"""Heartbeat кожні 30 секунд"""
|
|
while self.running:
|
|
await asyncio.sleep(30)
|
|
if self.registry:
|
|
await self.registry.update_heartbeat()
|
|
|
|
async def stop(self):
|
|
"""Зупинка worker daemon"""
|
|
print("🛑 Зупинка Worker Daemon...")
|
|
self.running = False
|
|
|
|
if self.nats_client:
|
|
await self.nats_client.disconnect()
|
|
if self.registry:
|
|
await self.registry.unregister()
|
|
if self.metrics:
|
|
await self.metrics.stop()
|
|
|
|
print("✅ Worker Daemon зупинено")
|
|
|
|
def setup_signal_handlers(self):
|
|
"""Налаштування обробників сигналів"""
|
|
def signal_handler(sig, frame):
|
|
print(f"\n📡 Отримано сигнал {sig}")
|
|
asyncio.create_task(self.stop())
|
|
sys.exit(0)
|
|
|
|
signal.signal(signal.SIGINT, signal_handler)
|
|
signal.signal(signal.SIGTERM, signal_handler)
|
|
|
|
|
|
async def main():
|
|
daemon = WorkerDaemon()
|
|
daemon.setup_signal_handlers()
|
|
|
|
try:
|
|
await daemon.start()
|
|
# Чекаємо поки працює
|
|
while daemon.running:
|
|
await asyncio.sleep(1)
|
|
except KeyboardInterrupt:
|
|
pass
|
|
finally:
|
|
await daemon.stop()
|
|
|
|
|
|
if __name__ == "__main__":
|
|
asyncio.run(main())
|