import time import os import subprocess import re import json import sys import argparse from kafka import KafkaProducer print("Python executable:", sys.executable) # Colores ANSI COLOR_GREEN = '\033[0;32m' COLOR_RED = '\033[0;31m' COLOR_GRAY = '\033[1;30m' COLOR_RESET = '\033[0m' #CONFIG_FILE = "coinfeed.conf" CONFIG_FILE = os.path.abspath(os.path.join(os.path.dirname(__file__), '..', 'coinfeed.conf')) ENV_FILE = os.path.abspath(os.path.join(os.path.dirname(__file__), '..', '.env')) LOGS_DIR = "/var/log/trading_modules" # Directorio para los logs # Variables globales para el topic de Kafka y la lista de posiciones DOCKER_ID = None DOCKER_KAKFA_TOPIC = None DOCKER_KAKFA_SERVER = None KAFKA_PRODUCTER = None last_positions = {} def parse_config_line(line): line = line.strip().strip('[]') try: props = dict( (kv.split('=')[0].strip(), kv.split('=')[1].strip().strip("'")) for kv in line.split(',') if '=' in kv ) except Exception: return None return props def initialize_kafka_producer(): global KAFKA_PRODUCTER KAFKA_PRODUCTER = KafkaProducer( bootstrap_servers=DOCKER_KAKFA_SERVER, value_serializer=lambda v: json.dumps(v).encode('utf-8') ) def get_service_name(service_type, props): global DOCKER_ID if service_type == "coinfeed": exchange = props.get("exchange", "").lower() symbol = props.get("symbol", "") interval = props.get("interval", "") return f"docker-{DOCKER_ID}_trading_module_{service_type}_{exchange}_{symbol}_{interval}.service" elif service_type == "nodejs": return f"docker-{DOCKER_ID}_trading_module_{service_type}.service" elif service_type == "monitor": return f"docker-{DOCKER_ID}_trading_module_{service_type}.service" return None def check_service_status(service_name): try: result = subprocess.run( ['systemctl', 'is-active', service_name], stdout=subprocess.PIPE, stderr=subprocess.DEVNULL, text=True ) return result.stdout.strip() except Exception: return "unknown" def read_docker_settings(): global DOCKER_ID global DOCKER_KAKFA_TOPIC global DOCKER_KAKFA_SERVER try: with open(ENV_FILE, 'r') as f: print(f"{COLOR_GRAY}.env File Abierto Listo para leer: {COLOR_RESET}") for line in f: line = line.strip() if not line or line.startswith('#'): # Ignorar líneas vacías o comentarios continue key, _, value = line.partition('=') key = key.strip() value = value.strip().strip('"').strip("'") # Eliminar comillas si las hay # print(f"{COLOR_GRAY}.env File Abierto Listo para leer: {key} = {value}{COLOR_RESET}") if key == "DOCKER_ID": DOCKER_ID = value print(f"{COLOR_GREEN}Encontrado DOCKER_ID en configuración: {DOCKER_ID}{COLOR_RESET}") elif key == "DOCKER_KAKFA_SERVER": DOCKER_KAKFA_SERVER = value print(f"{COLOR_GREEN}Encontrado DOCKER_KAKFA_SERVER en configuración: {DOCKER_KAKFA_SERVER}{COLOR_RESET}") elif key == "DOCKER_KAFKA_TOPIC": DOCKER_KAKFA_TOPIC = value print(f"{COLOR_GREEN}Encontrado DOCKER_KAFKA_TOPIC en configuración: {DOCKER_KAKFA_TOPIC}{COLOR_RESET}") if DOCKER_ID and DOCKER_KAKFA_TOPIC: return True else: print(f"{COLOR_RED} ⚠️ read_docker_settings : Faltan valores requeridos en el archivo .env{COLOR_RESET}") except FileNotFoundError: print(f"{COLOR_RED} ⚠️ read_docker_settings : No se encontró el archivo {ENV_FILE}{COLOR_RESET}") return False def get_log_mtime(path): try: return time.strftime('%Y-%m-%d %H:%M:%S', time.localtime(os.path.getmtime(path))) except Exception: return None def tail_last_lines(path, num_lines=10): try: with open(path, 'rb') as f: f.seek(0, os.SEEK_END) end = f.tell() lines = [] block = -1 while len(lines) <= num_lines and abs(block * 1024) < end: f.seek(block * 1024, os.SEEK_END) lines = f.read().splitlines() block -= 1 return [l.decode(errors='ignore') for l in lines[-num_lines:]] except: return [] def print_status_and_send_kafka(kafka_topic): os.system('clear' if os.name == 'posix' else 'cls') print(f"=== Estado de servicios : docker-{DOCKER_ID} ===") servicio_list = [] try: with open(CONFIG_FILE, 'r') as f: for line in f: line = line.strip() if not line or not line.startswith('['): continue props = parse_config_line(line) if not props: continue service_type = props.get("service_type") if service_type not in ["coinfeed", "nodejs", "monitor"]: continue service_name = get_service_name(service_type, props) if not service_name: continue status = check_service_status(service_name) if status == "active": color = COLOR_GREEN elif status in ["inactive", "failed"]: color = COLOR_RED elif status == "unknown": color = COLOR_GRAY status = "not found" else: color = COLOR_GRAY if service_type == "coinfeed": # log_filename = f"{props.get('exchange')}-{props.get('symbol')}-{props.get('interval')}.log" log_filename = f"docker-{DOCKER_ID}_trading_module_{service_type}_{props.get('exchange')}_{props.get('symbol')}_{props.get('interval')}.log" elif service_type == "nodejs": # log_filename = "nodejs.log" log_filename = f"docker-{DOCKER_ID}_trading_module_{service_type}.log" elif service_type == "monitor": # log_filename = "monitor.log" log_filename = f"docker-{DOCKER_ID}_trading_module_{service_type}.log" else: log_filename = None log_path = os.path.join(LOGS_DIR, log_filename) if log_filename else None log_status = "NO_LOG" log_lastmod = None log_last10 = [] last_pos = last_positions.get(log_path, 0) if log_path else 0 if log_path and os.path.exists(log_path): current_size = os.path.getsize(log_path) if current_size < last_pos: last_pos = 0 # Log reiniciado with open(log_path, "r") as logf: logf.seek(last_pos) nuevos_logs = logf.read() last_positions[log_path] = logf.tell() if "error" in nuevos_logs.lower(): log_status = "ERROR" else: log_status = "OK" log_lastmod = get_log_mtime(log_path) log_last10 = tail_last_lines(log_path) entry = { "service_name": service_name, "service_type": service_type, "status": status, "log_status": log_status, "log_lastmod": log_lastmod, "log_last10lines": log_last10 } if service_type == "coinfeed": entry.update({ "exchange": props.get("exchange"), "symbol": props.get("symbol"), "interval": props.get("interval") }) servicio_list.append(entry) print(f"{color}{service_name} => {status}{COLOR_RESET}") print(" ⏩ Listo para enviar a Kafka") mensaje = { "notification_type": "service_status", "servicioList": servicio_list, "timestamp": time.strftime('%Y-%m-%d %H:%M:%S', time.localtime()) } try: KAFKA_PRODUCTER.send(kafka_topic, mensaje).get(timeout=10) print(f"{COLOR_GREEN} ✅ Mensaje enviado correctamente a Kafka : Topic = {kafka_topic} 👌 {COLOR_RESET}") except Exception as e: print(f"{COLOR_RED} ❌ Error al enviar mensaje a Kafka: {e} 👎 {COLOR_RESET}") print(f"Fecha y hora: {time.strftime('%Y-%m-%d %H:%M:%S')}\n") except FileNotFoundError: print(f"{COLOR_RED}⚠️ print_status_and_send_kafka : No se encontró el archivo {CONFIG_FILE}{COLOR_RESET}") def main(): """ Main function to parse arguments, read configuration settings, and monitor the health status of services in real-time. """ parser = argparse.ArgumentParser( description="Script to monitor the health status of services and send updates to a Kafka topic." ) parser.add_argument('--kafkamonitor', type=str, default='', help='Kafka Topcic to monitor') args = parser.parse_args() # Leer la configuración 'settings' y obtener el 'DOCKER_ID' if not read_docker_settings(): print(f"{COLOR_RED}Error al leer la configuración '.env'.{COLOR_RESET}") sys.exit(1) # Inicializar el productor de Kafka initialize_kafka_producer() while True: dockerkafkatopic = f"docker-{DOCKER_ID}-{DOCKER_KAKFA_TOPIC}" print_status_and_send_kafka(dockerkafkatopic) print("Actualizando en 60 segundos... (Ctrl+C para salir)") time.sleep(60) if __name__ == "__main__": main()