import { getLogger } from "./logger"; import { Rabbit } from "./rabbit"; import { randomUUID } from "crypto"; import { WrapperRegistry, WrapperInfo } from "./wrapperRegistry"; import { ServerRegistry } from "./serverRegistry"; import { WrapperRegisterMessage, WrapperHeartbeatMessage, WrapperStatusMessage, ServerStartCommand, ServerStopCommand } from "server-manager-shared"; type WrapperMessage = | WrapperRegisterMessage | WrapperHeartbeatMessage | WrapperStatusMessage; const logger = getLogger("WrapperManager"); export class WrapperManager { constructor( private rabbit: Rabbit, private registry: WrapperRegistry, private serverRegistry: ServerRegistry ){ } async start(){ const queue = "wrapper.register"; await this.rabbit.channel.assertQueue( queue ); logger.info( "Warte auf Wrapper..." ); this.rabbit.channel.consume( queue, async msg => { if(!msg) return; const data = JSON.parse( msg.content.toString() ) as WrapperMessage; switch(data.type){ case "wrapper.register": await this.register( data ); break; case "wrapper.heartbeat": this.heartbeat( data ); break; case "wrapper.status": logger.debug( "Status update received from wrapper-id", data.wrapperId ); this.registry.updateStatus( data ); this.serverRegistry.syncWrapperStatus( data ); break; default: logger.warn( "Unbekannte Nachricht:", data ); } this.rabbit.channel.ack( msg ); } ); // Offline Check setInterval(()=>{ this.registry.checkOffline( 30000 ); },5000); } async register(data: WrapperRegisterMessage){ let id = data.wrapperId; /* Neue Wrapper bekommen eine ID */ if(!id){ id = "wrp-" + randomUUID() .slice(0,8); logger.debug( "Neue Wrapper ID:", id ); } const existing = this.registry.get( id ); /* Wrapper war schon bekannt */ if(existing){ existing.online = true; existing.lastHeartbeat = Date.now(); existing.hostname = data.hostname; existing.cpu = data.cpu; existing.ram = data.ram; existing.ramFree = data.ram; // servers are updated via wrapper.status, not wrapper.register this.registry.update(existing); } /* Neuer Wrapper */ else { const commandQueue = `wrapper.${id}.commands`; await this.rabbit.channel.assertQueue( commandQueue ); const wrapper:WrapperInfo = { id, hostname: data.hostname, commandQueue, cpu: data.cpu, ram: data.ram, ramFree: data.ram, servers: [], online:true, lastHeartbeat: Date.now() }; this.registry.add( wrapper ); } /* Antwort zurück an Wrapper */ const wrapper = this.registry.get( id ); if(data.replyQueue){ this.rabbit.channel.sendToQueue( data.replyQueue, Buffer.from( JSON.stringify({ type: "wrapper.registered", wrapperId: id, commandQueue: wrapper?.commandQueue }) ) ); } } heartbeat(data: WrapperHeartbeatMessage){ this.registry.updateHeartbeat( data.wrapperId ); logger.debug( "Heartbeat:", data.wrapperId ); } sendCommand( wrapperId:string, command: ServerStartCommand | ServerStopCommand ){ const wrapper = this.registry.get( wrapperId ); if(!wrapper){ throw new Error( "Wrapper nicht gefunden" ); } this.rabbit.channel.sendToQueue( wrapper.commandQueue, Buffer.from( JSON.stringify( command ) ) ); } getWrappers(){ return this.registry.getAll(); } listWrappers(){ const wrappers = this.registry.getAll(); logger.info( "Wrapper Übersicht ausgegeben" ); if(wrappers.length===0){ logger.info( "Keine Wrapper registriert" ); return; } for(const wrapper of wrappers){ logger.info(` ID: ${wrapper.id} Host: ${wrapper.hostname} CPU: ${wrapper.cpu} Cores RAM: ${wrapper.ram} MB RAM frei: ${wrapper.ramFree} MB Server: ${wrapper.servers.length} Online: ${wrapper.online} Heartbeat: ${new Date( wrapper.lastHeartbeat ).toLocaleString()} ---------------------- `); } } }