diff --git a/CHANGELOG.md b/CHANGELOG.md index 9c99a8dc..119579cc 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -5,6 +5,10 @@ The format is based on [Keep a Changelog](https://keepachangelog.com/en/1.0.0/), and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0.html). ## [Unreleased] +- **Changed:** Global `FileSystemService` and `ComposeService` singletons refactored into node-aware instances. +- **Added:** `IFileAdapter`, `LocalFileAdapter`, and `SSHFileAdapter` to abstract all filesystem interactions for remote node support. +- **Changed:** `MonitorService` now evaluates limits, fetches metrics, and detects container crashes across all registered nodes concurrently. +- **Added:** Node Context Middleware in Express API to dynamically extract `x-node-id` headers and parse WebSocket query parameters. - **Added:** Remote Nodes Foundation (Strategy B) — `nodes` table in SQLite with auto-seeded default local node. - **Added:** `NodeRegistry` service for managing multiple Docker daemon connections (local socket + TCP). - **Added:** Node management API endpoints: list, get, create, update, delete, and test connection. diff --git a/backend/package-lock.json b/backend/package-lock.json index 0f543f2b..3e10f5dd 100644 --- a/backend/package-lock.json +++ b/backend/package-lock.json @@ -23,6 +23,8 @@ "express": "^5.2.1", "jsonwebtoken": "^9.0.3", "node-pty": "^1.1.0", + "ssh2": "^1.17.0", + "ssh2-sftp-client": "^12.1.0", "systeminformation": "^5.31.1", "ws": "^8.19.0", "yaml": "^2.8.2" @@ -33,6 +35,8 @@ "@types/cookie-parser": "^1.4.10", "@types/jsonwebtoken": "^9.0.10", "@types/node": "^25.3.0", + "@types/ssh2": "^1.15.5", + "@types/ssh2-sftp-client": "^9.0.6", "@types/yaml": "^1.9.6", "nodemon": "^3.1.13", "ts-node": "^10.9.2", @@ -414,6 +418,16 @@ "@types/node": "^18.11.18" } }, + "node_modules/@types/ssh2-sftp-client": { + "version": "9.0.6", + "resolved": "https://registry.npmjs.org/@types/ssh2-sftp-client/-/ssh2-sftp-client-9.0.6.tgz", + "integrity": "sha512-4+KvXO/V77y9VjI2op2T8+RCGI/GXQAwR0q5Qkj/EJ5YSeyKszqZP6F8i3H3txYoBqjc7sgorqyvBP3+w1EHyg==", + "dev": true, + "license": "MIT", + "dependencies": { + "@types/ssh2": "^1.0.0" + } + }, "node_modules/@types/ssh2/node_modules/@types/node": { "version": "18.19.130", "resolved": "https://registry.npmjs.org/@types/node/-/node-18.19.130.tgz", @@ -764,6 +778,12 @@ "integrity": "sha512-zRpUiDwd/xk6ADqPMATG8vc9VPrkck7T07OIx0gnjmJAnHnTVXNQG3vfvWNuiZIkwu9KrKdA1iJKfsfTVxE6NA==", "license": "BSD-3-Clause" }, + "node_modules/buffer-from": { + "version": "1.1.2", + "resolved": "https://registry.npmjs.org/buffer-from/-/buffer-from-1.1.2.tgz", + "integrity": "sha512-E+XQCRwSbaaiChtv6k6Dwgc+bx+Bs6vuKJHHl5kox/BaKbhiXzqQOwK4cO22yElGp2OCmjwVhT3HmxgyPGnJfQ==", + "license": "MIT" + }, "node_modules/buildcheck": { "version": "0.0.7", "resolved": "https://registry.npmjs.org/buildcheck/-/buildcheck-0.0.7.tgz", @@ -933,6 +953,21 @@ "yaml": "^2.x" } }, + "node_modules/concat-stream": { + "version": "2.0.0", + "resolved": "https://registry.npmjs.org/concat-stream/-/concat-stream-2.0.0.tgz", + "integrity": "sha512-MWufYdFw53ccGjCA+Ol7XJYpAlW6/prSMzuPOTRnJGcGzuhLn4Scrz7qf6o8bROZ514ltazcIFJZevcfbo0x7A==", + "engines": [ + "node >= 6.0" + ], + "license": "MIT", + "dependencies": { + "buffer-from": "^1.0.0", + "inherits": "^2.0.3", + "readable-stream": "^3.0.2", + "typedarray": "^0.0.6" + } + }, "node_modules/content-disposition": { "version": "1.0.1", "resolved": "https://registry.npmjs.org/content-disposition/-/content-disposition-1.0.1.tgz", @@ -2647,6 +2682,23 @@ "nan": "^2.23.0" } }, + "node_modules/ssh2-sftp-client": { + "version": "12.1.0", + "resolved": "https://registry.npmjs.org/ssh2-sftp-client/-/ssh2-sftp-client-12.1.0.tgz", + "integrity": "sha512-f8+EylryHXPg+pMHYqtK+jPOPXJTBDOdgwOPYNS7mJTkk5bPsLQfOFUVl99PQvUkE5MkbyR5jThBlfeQMXcvrA==", + "license": "Apache-2.0", + "dependencies": { + "concat-stream": "^2.0.0", + "ssh2": "^1.16.0" + }, + "engines": { + "node": ">=18.20.4" + }, + "funding": { + "type": "individual", + "url": "https://square.link/u/4g7sPflL" + } + }, "node_modules/statuses": { "version": "2.0.2", "resolved": "https://registry.npmjs.org/statuses/-/statuses-2.0.2.tgz", @@ -2875,6 +2927,12 @@ "node": ">= 0.6" } }, + "node_modules/typedarray": { + "version": "0.0.6", + "resolved": "https://registry.npmjs.org/typedarray/-/typedarray-0.0.6.tgz", + "integrity": "sha512-/aCDEGatGvZ2BIk+HmLf4ifCJFwvKFNb9/JeZPMulfgFracn9QFcAf5GO8B/mweUjSoblS5In0cWhqpfs/5PQA==", + "license": "MIT" + }, "node_modules/typescript": { "version": "5.9.3", "resolved": "https://registry.npmjs.org/typescript/-/typescript-5.9.3.tgz", diff --git a/backend/package.json b/backend/package.json index b323ee98..a452faac 100644 --- a/backend/package.json +++ b/backend/package.json @@ -19,6 +19,8 @@ "@types/cookie-parser": "^1.4.10", "@types/jsonwebtoken": "^9.0.10", "@types/node": "^25.3.0", + "@types/ssh2": "^1.15.5", + "@types/ssh2-sftp-client": "^9.0.6", "@types/yaml": "^1.9.6", "nodemon": "^3.1.13", "ts-node": "^10.9.2", @@ -39,6 +41,8 @@ "express": "^5.2.1", "jsonwebtoken": "^9.0.3", "node-pty": "^1.1.0", + "ssh2": "^1.17.0", + "ssh2-sftp-client": "^12.1.0", "systeminformation": "^5.31.1", "ws": "^8.19.0", "yaml": "^2.8.2" diff --git a/backend/src/index.ts b/backend/src/index.ts index e09ee973..e04e874f 100644 --- a/backend/src/index.ts +++ b/backend/src/index.ts @@ -30,11 +30,7 @@ const execAsync = promisify(exec); const app = express(); const PORT = 3000; -// FileSystemService for stack management -const fileSystemService = new FileSystemService(); - -// ComposeService for stack operations -const composeService = new ComposeService(); +// FileSystemService and ComposeService are instantiated per-request via .getInstance(nodeId) // Cookie settings const COOKIE_NAME = 'sencho_token'; @@ -60,10 +56,29 @@ app.use(cors({ app.use(express.json()); app.use(cookieParser()); -// Extend Express Request type for user -declare module 'express' { - interface Request { - user?: { username: string }; +// Node Context Middleware +const nodeContextMiddleware = (req: Request, res: Response, next: NextFunction) => { + const nodeIdHeader = req.headers['x-node-id'] as string; + const nodeIdQuery = req.query.nodeId as string; + if (nodeIdHeader) { + req.nodeId = parseInt(nodeIdHeader, 10); + } else if (nodeIdQuery) { + req.nodeId = parseInt(nodeIdQuery, 10); + } else { + req.nodeId = NodeRegistry.getInstance().getDefaultNodeId(); + } + next(); +}; + +app.use(nodeContextMiddleware); + +// Extend Express Request type for user and node +declare global { + namespace Express { + interface Request { + user?: { username: string }; + nodeId: number; + } } } @@ -279,16 +294,22 @@ server.on('upgrade', async (req, socket, head) => { // Check if this is a stack logs WebSocket request const url = req.url || ''; - const logsMatch = url.match(/^\/api\/stacks\/([^/]+)\/logs$/); - const hostConsoleMatch = url.match(/^\/api\/system\/host-console/); + const parsedUrl = new URL(url, `http://${req.headers.host || 'localhost'}`); + const pathname = parsedUrl.pathname; + + const logsMatch = pathname.match(/^\/api\/stacks\/([^/]+)\/logs$/); + const hostConsoleMatch = pathname.match(/^\/api\/system\/host-console/); if (logsMatch) { // Dedicated stack logs WebSocket - uses Supervisor loop for persistent logs const logsWss = new WebSocket.Server({ noServer: true }); logsWss.handleUpgrade(req, socket, head, (ws) => { const stackName = decodeURIComponent(logsMatch[1]); + const nodeIdParam = parsedUrl.searchParams.get('nodeId'); + const nodeId = nodeIdParam ? parseInt(nodeIdParam, 10) : NodeRegistry.getInstance().getDefaultNodeId(); + try { - composeService.streamLogs(stackName, ws); + ComposeService.getInstance(nodeId).streamLogs(stackName, ws); } catch (error) { console.error('Failed to stream logs:', error); if (ws.readyState === WebSocket.OPEN) { @@ -299,15 +320,20 @@ server.on('upgrade', async (req, socket, head) => { } else if (hostConsoleMatch) { const hostConsoleWss = new WebSocket.Server({ noServer: true }); hostConsoleWss.handleUpgrade(req, socket, head, (ws) => { - let targetDirectory = fileSystemService.getBaseDir(); + let targetDirectory = ''; try { const reqUrl = new URL(req.url || '', `http://${req.headers.host || 'localhost'}`); + const nodeIdParam = reqUrl.searchParams.get('nodeId'); + const nodeId = nodeIdParam ? parseInt(nodeIdParam, 10) : NodeRegistry.getInstance().getDefaultNodeId(); + targetDirectory = FileSystemService.getInstance(nodeId).getBaseDir(); + const stackParam = reqUrl.searchParams.get('stack'); if (stackParam) { targetDirectory = path.join(targetDirectory, stackParam); } } catch (e) { // ignore parsing error, fallback to base dir + targetDirectory = FileSystemService.getInstance(NodeRegistry.getInstance().getDefaultNodeId()).getBaseDir(); } try { HostTerminalService.spawnTerminal(ws, targetDirectory); @@ -347,12 +373,14 @@ wss.on('connection', (ws) => { if (data.action === 'connectTerminal') { terminalWs = ws; } else if (data.action === 'streamStats') { - const dockerController = DockerController.getInstance(); + const nodeId = data.nodeId ? parseInt(data.nodeId, 10) : NodeRegistry.getInstance().getDefaultNodeId(); + const dockerController = DockerController.getInstance(nodeId); dockerController.streamStats(data.containerId, ws); } else if (data.action === 'execContainer') { // Handle container exec for bash access // Input, resize, and cleanup are handled inside execContainer's closure - const dockerController = DockerController.getInstance(); + const nodeId = data.nodeId ? parseInt(data.nodeId, 10) : NodeRegistry.getInstance().getDefaultNodeId(); + const dockerController = DockerController.getInstance(nodeId); dockerController.execContainer(data.containerId, ws); } } catch (error) { @@ -365,7 +393,7 @@ wss.on('connection', (ws) => { app.get('/api/containers', async (req: Request, res: Response) => { try { - const dockerController = DockerController.getInstance(); + const dockerController = DockerController.getInstance(req.nodeId); const containers = await dockerController.getRunningContainers(); res.json(containers); } catch (error) { @@ -377,7 +405,7 @@ app.get('/api/containers', async (req: Request, res: Response) => { app.get('/api/stacks', async (req: Request, res: Response) => { try { - const stacks = await fileSystemService.getStacks(); + const stacks = await FileSystemService.getInstance(req.nodeId).getStacks(); res.json(stacks); } catch (error) { res.status(500).json({ error: 'Failed to fetch stacks' }); @@ -387,7 +415,7 @@ app.get('/api/stacks', async (req: Request, res: Response) => { app.get('/api/stacks/:stackName', async (req: Request, res: Response) => { try { const stackName = req.params.stackName as string; - const content = await fileSystemService.getStackContent(stackName); + const content = await FileSystemService.getInstance(req.nodeId).getStackContent(stackName); res.send(content); } catch (error) { res.status(500).json({ error: 'Failed to read stack' }); @@ -406,7 +434,7 @@ app.put('/api/stacks/:stackName', async (req: Request, res: Response) => { console.error('Content is not a string:', content); return res.status(400).json({ error: 'Content must be a string' }); } - await fileSystemService.saveStackContent(stackName, content); + await FileSystemService.getInstance(req.nodeId).saveStackContent(stackName, content); console.log('Stack saved successfully:', stackName); res.json({ message: 'Stack saved successfully' }); } catch (error) { @@ -416,8 +444,9 @@ app.put('/api/stacks/:stackName', async (req: Request, res: Response) => { }); // Helper: resolve all env file paths dynamically from compose.yaml's env_file field -async function resolveAllEnvFilePaths(stackName: string): Promise { - const stackDir = path.join(fileSystemService.getBaseDir(), stackName); +async function resolveAllEnvFilePaths(nodeId: number, stackName: string): Promise { + const fsService = FileSystemService.getInstance(nodeId); + const stackDir = path.join(fsService.getBaseDir(), stackName); const defaultEnvPath = path.join(stackDir, '.env'); try { @@ -427,7 +456,7 @@ async function resolveAllEnvFilePaths(stackName: string): Promise { for (const file of composeFiles) { try { - composeContent = await fsPromises.readFile(path.join(stackDir, file), 'utf-8'); + composeContent = await fsService.readFile(path.join(stackDir, file), 'utf-8'); break; } catch { // Try next file @@ -479,7 +508,7 @@ async function resolveAllEnvFilePaths(stackName: string): Promise { app.get('/api/stacks/:stackName/envs', async (req: Request, res: Response) => { try { const stackName = req.params.stackName as string; - const envPaths = await resolveAllEnvFilePaths(stackName); + const envPaths = await resolveAllEnvFilePaths(req.nodeId, stackName); res.json({ envFiles: envPaths }); } catch (error) { res.status(500).json({ error: 'Failed to resolve env files' }); @@ -490,7 +519,7 @@ app.get('/api/stacks/:stackName/env', async (req: Request, res: Response) => { try { const stackName = req.params.stackName as string; const requestedFile = req.query.file as string | undefined; - const envPaths = await resolveAllEnvFilePaths(stackName); + const envPaths = await resolveAllEnvFilePaths(req.nodeId, stackName); let envPath = envPaths[0]; // Fallback to the first @@ -503,13 +532,15 @@ app.get('/api/stacks/:stackName/env', async (req: Request, res: Response) => { } } + const fsService = FileSystemService.getInstance(req.nodeId); + try { - await fsPromises.access(envPath); + await fsService.access(envPath); } catch { return res.status(404).json({ error: 'Env file not found' }); } - const content = await fsPromises.readFile(envPath, 'utf-8'); + const content = await fsService.readFile(envPath, 'utf-8'); res.send(content); } catch (error) { console.error('Failed to read env file:', error); @@ -529,7 +560,7 @@ app.put('/api/stacks/:stackName/env', async (req: Request, res: Response) => { } const requestedFile = req.query.file as string | undefined; - const envPaths = await resolveAllEnvFilePaths(stackName); + const envPaths = await resolveAllEnvFilePaths(req.nodeId, stackName); let envPath = envPaths[0]; // Fallback @@ -541,7 +572,8 @@ app.put('/api/stacks/:stackName/env', async (req: Request, res: Response) => { } } - await fsPromises.writeFile(envPath, content, 'utf-8'); + const fsService = FileSystemService.getInstance(req.nodeId); + await fsService.writeFile(envPath, content, 'utf-8'); res.json({ message: 'Env file saved successfully' }); } catch (error) { console.error('Failed to save env file:', error); @@ -558,7 +590,7 @@ app.post('/api/stacks', async (req: Request, res: Response) => { if (!/^[a-zA-Z0-9-]+$/.test(stackName)) { return res.status(400).json({ error: 'Stack name can only contain alphanumeric characters and hyphens' }); } - await fileSystemService.createStack(stackName); + await FileSystemService.getInstance(req.nodeId).createStack(stackName); res.json({ message: 'Stack created successfully', name: stackName }); } catch (error: any) { if (error.message && error.message.includes('already exists')) { @@ -574,13 +606,13 @@ app.delete('/api/stacks/:name', async (req: Request, res: Response) => { try { // Stage 1: Tell Docker to clean up ghost networks/containers try { - await composeService.downStack(stackName); + await ComposeService.getInstance(req.nodeId).downStack(stackName); } catch (downErr) { console.warn(`[Teardown] Docker down failed or nothing to clean up for ${stackName}`); } // Stage 2: Obliterate the files - await fileSystemService.deleteStack(stackName); + await FileSystemService.getInstance(req.nodeId).deleteStack(stackName); res.json({ success: true }); } catch (error: any) { @@ -591,7 +623,7 @@ app.delete('/api/stacks/:name', async (req: Request, res: Response) => { app.get('/api/stacks/:stackName/containers', async (req: Request, res: Response) => { try { const stackName = req.params.stackName as string; - const dockerController = DockerController.getInstance(); + const dockerController = DockerController.getInstance(req.nodeId); const containers = await dockerController.getContainersByStack(stackName); res.json(containers); } catch (error) { @@ -602,7 +634,7 @@ app.get('/api/stacks/:stackName/containers', async (req: Request, res: Response) app.get('/api/containers/:id/logs', async (req: Request, res: Response) => { try { const id = req.params.id as string; - const dockerController = DockerController.getInstance(); + const dockerController = DockerController.getInstance(req.nodeId); // Pass both req and res so we can listen for the client disconnect await dockerController.streamContainerLogs(id, req, res); } catch (error) { @@ -613,7 +645,7 @@ app.get('/api/containers/:id/logs', async (req: Request, res: Response) => { app.post('/api/containers/:id/start', async (req: Request, res: Response) => { try { const id = req.params.id as string; - const dockerController = DockerController.getInstance(); + const dockerController = DockerController.getInstance(req.nodeId); await dockerController.startContainer(id); res.json({ message: 'Container started' }); } catch (error) { @@ -624,7 +656,7 @@ app.post('/api/containers/:id/start', async (req: Request, res: Response) => { app.post('/api/containers/:id/stop', async (req: Request, res: Response) => { try { const id = req.params.id as string; - const dockerController = DockerController.getInstance(); + const dockerController = DockerController.getInstance(req.nodeId); await dockerController.stopContainer(id); res.json({ message: 'Container stopped' }); } catch (error) { @@ -635,7 +667,7 @@ app.post('/api/containers/:id/stop', async (req: Request, res: Response) => { app.post('/api/containers/:id/restart', async (req: Request, res: Response) => { try { const id = req.params.id as string; - const dockerController = DockerController.getInstance(); + const dockerController = DockerController.getInstance(req.nodeId); await dockerController.restartContainer(id); res.json({ message: 'Container restarted' }); } catch (error) { @@ -647,7 +679,7 @@ app.post('/api/containers/:id/restart', async (req: Request, res: Response) => { app.post('/api/stacks/:stackName/deploy', async (req: Request, res: Response) => { try { const stackName = req.params.stackName as string; - await composeService.deployStack(stackName, terminalWs || undefined); + await ComposeService.getInstance(req.nodeId).deployStack(stackName, terminalWs || undefined); res.json({ message: 'Deployed successfully' }); } catch (error: any) { console.error('Failed to deploy stack:', error); @@ -658,7 +690,7 @@ app.post('/api/stacks/:stackName/deploy', async (req: Request, res: Response) => app.post('/api/stacks/:stackName/down', async (req: Request, res: Response) => { try { const stackName = req.params.stackName as string; - await composeService.runCommand(stackName, 'down', terminalWs || undefined); + await ComposeService.getInstance(req.nodeId).runCommand(stackName, 'down', terminalWs || undefined); res.json({ status: 'Command started' }); } catch (error) { res.status(500).json({ error: 'Failed to start command' }); @@ -668,7 +700,7 @@ app.post('/api/stacks/:stackName/down', async (req: Request, res: Response) => { app.post('/api/stacks/:stackName/restart', async (req: Request, res: Response) => { try { const stackName = req.params.stackName as string; - const dockerController = DockerController.getInstance(); + const dockerController = DockerController.getInstance(req.nodeId); const containers = await dockerController.getContainersByStack(stackName); if (!containers || containers.length === 0) { @@ -686,7 +718,7 @@ app.post('/api/stacks/:stackName/restart', async (req: Request, res: Response) = app.post('/api/stacks/:stackName/stop', async (req: Request, res: Response) => { try { const stackName = req.params.stackName as string; - const dockerController = DockerController.getInstance(); + const dockerController = DockerController.getInstance(req.nodeId); const containers = await dockerController.getContainersByStack(stackName); if (!containers || containers.length === 0) { @@ -704,7 +736,7 @@ app.post('/api/stacks/:stackName/stop', async (req: Request, res: Response) => { app.post('/api/stacks/:stackName/start', async (req: Request, res: Response) => { try { const stackName = req.params.stackName as string; - const dockerController = DockerController.getInstance(); + const dockerController = DockerController.getInstance(req.nodeId); const containers = await dockerController.getContainersByStack(stackName); if (!containers || containers.length === 0) { @@ -724,7 +756,7 @@ app.post('/api/stacks/:stackName/update', async (req: Request, res: Response) => try { const stackName = req.params.stackName as string; // Await update completion - await composeService.updateStack(stackName, terminalWs || undefined); + await ComposeService.getInstance(req.nodeId).updateStack(stackName, terminalWs || undefined); res.json({ status: 'Update completed' }); } catch (error) { res.status(500).json({ error: 'Failed to update' }); @@ -749,7 +781,7 @@ app.post('/api/convert', async (req: Request, res: Response) => { // Get all containers stats for dashboard app.get('/api/stats', async (req: Request, res: Response) => { try { - const dockerController = DockerController.getInstance(); + const dockerController = DockerController.getInstance(req.nodeId); const containers = await dockerController.getRunningContainers(); const allContainers = await dockerController.getAllContainers(); @@ -774,7 +806,7 @@ app.get('/api/metrics/historical', async (req: Request, res: Response) => { app.get('/api/logs/global', async (req: Request, res: Response) => { try { - const dockerController = DockerController.getInstance(); + const dockerController = DockerController.getInstance(req.nodeId); const containers = await dockerController.getRunningContainers(); const allLogs: any[] = []; @@ -870,7 +902,7 @@ app.get('/api/logs/global/stream', async (req: Request, res: Response) => { res.setHeader('Connection', 'keep-alive'); res.flushHeaders(); - const dockerController = DockerController.getInstance(); + const dockerController = DockerController.getInstance(req.nodeId); const streams: NodeJS.ReadableStream[] = []; try { @@ -1136,8 +1168,8 @@ app.post('/api/notifications/test', async (req: Request, res: Response) => { app.get('/api/system/orphans', async (req: Request, res: Response) => { try { - const knownStacks = await fileSystemService.getStacks(); - const dockerController = DockerController.getInstance(); + const knownStacks = await FileSystemService.getInstance(req.nodeId).getStacks(); + const dockerController = DockerController.getInstance(req.nodeId); const orphans = await dockerController.getOrphanContainers(knownStacks); res.json(orphans); } catch (error) { @@ -1152,7 +1184,7 @@ app.post('/api/system/prune/orphans', async (req: Request, res: Response) => { if (!Array.isArray(containerIds)) { return res.status(400).json({ error: 'containerIds must be an array' }); } - const dockerController = DockerController.getInstance(); + const dockerController = DockerController.getInstance(req.nodeId); const results = await dockerController.removeContainers(containerIds); res.json({ results }); } catch (error) { @@ -1168,7 +1200,7 @@ app.post('/api/system/prune/system', async (req: Request, res: Response) => { return res.status(400).json({ error: 'Invalid prune target' }); } - const dockerController = DockerController.getInstance(); + const dockerController = DockerController.getInstance(req.nodeId); const result = await dockerController.pruneSystem(target); res.json({ message: 'Prune completed', ...result }); @@ -1180,7 +1212,7 @@ app.post('/api/system/prune/system', async (req: Request, res: Response) => { app.get('/api/system/docker-df', async (req: Request, res: Response) => { try { - const dockerController = DockerController.getInstance(); + const dockerController = DockerController.getInstance(req.nodeId); const df = await dockerController.getDiskUsage(); res.json(df); } catch (error) { @@ -1191,7 +1223,7 @@ app.get('/api/system/docker-df', async (req: Request, res: Response) => { app.get('/api/system/images', async (req: Request, res: Response) => { try { - const dockerController = DockerController.getInstance(); + const dockerController = DockerController.getInstance(req.nodeId); const images = await dockerController.getImages(); res.json(images); } catch (error) { @@ -1202,7 +1234,7 @@ app.get('/api/system/images', async (req: Request, res: Response) => { app.get('/api/system/volumes', async (req: Request, res: Response) => { try { - const dockerController = DockerController.getInstance(); + const dockerController = DockerController.getInstance(req.nodeId); const volumes = await dockerController.getVolumes(); res.json(volumes); } catch (error) { @@ -1213,7 +1245,7 @@ app.get('/api/system/volumes', async (req: Request, res: Response) => { app.get('/api/system/networks', async (req: Request, res: Response) => { try { - const dockerController = DockerController.getInstance(); + const dockerController = DockerController.getInstance(req.nodeId); const networks = await dockerController.getNetworks(); res.json(networks); } catch (error) { @@ -1226,7 +1258,7 @@ app.post('/api/system/images/delete', async (req: Request, res: Response) => { try { const { id } = req.body; if (!id) return res.status(400).json({ error: 'ID is required' }); - const dockerController = DockerController.getInstance(); + const dockerController = DockerController.getInstance(req.nodeId); await dockerController.removeImage(id); res.json({ success: true, message: 'Image deleted' }); } catch (error: any) { @@ -1239,7 +1271,7 @@ app.post('/api/system/volumes/delete', async (req: Request, res: Response) => { try { const { id } = req.body; if (!id) return res.status(400).json({ error: 'ID is required' }); - const dockerController = DockerController.getInstance(); + const dockerController = DockerController.getInstance(req.nodeId); await dockerController.removeVolume(id); res.json({ success: true, message: 'Volume deleted' }); } catch (error: any) { @@ -1252,7 +1284,7 @@ app.post('/api/system/networks/delete', async (req: Request, res: Response) => { try { const { id } = req.body; if (!id) return res.status(400).json({ error: 'ID is required' }); - const dockerController = DockerController.getInstance(); + const dockerController = DockerController.getInstance(req.nodeId); await dockerController.removeNetwork(id); res.json({ success: true, message: 'Network deleted' }); } catch (error: any) { @@ -1280,7 +1312,7 @@ app.post('/api/templates/deploy', async (req: Request, res: Response) => { return res.status(400).json({ error: 'stackName and template are required' }); } - const stackPath = path.join(fileSystemService.getBaseDir(), stackName); + const stackPath = path.join(FileSystemService.getInstance(req.nodeId).getBaseDir(), stackName); if (fs.existsSync(stackPath)) { return res.status(409).json({ error: `A stack directory named '${stackName}' already exists. Please choose a different Stack Name.`, @@ -1289,23 +1321,23 @@ app.post('/api/templates/deploy', async (req: Request, res: Response) => { } // 1. Create stack directory - await fileSystemService.createStack(stackName); + await FileSystemService.getInstance(req.nodeId).createStack(stackName); // 2. Generate compose YAML and save const composeYaml = templateService.generateComposeFromTemplate(template); - await fileSystemService.saveStackContent(stackName, composeYaml); + await FileSystemService.getInstance(req.nodeId).saveStackContent(stackName, composeYaml); // 3. Generate env string and save to default .env if (envVars) { const envString = templateService.generateEnvString(envVars); - const stackDir = path.join(fileSystemService.getBaseDir(), stackName); + const stackDir = path.join(FileSystemService.getInstance(req.nodeId).getBaseDir(), stackName); const defaultEnvPath = path.join(stackDir, '.env'); await fsPromises.writeFile(defaultEnvPath, envString, 'utf-8'); } // 4. Deploy the stack with atomic rollback try { - await composeService.deployStack(stackName, terminalWs || undefined); + await ComposeService.getInstance(req.nodeId).deployStack(stackName, terminalWs || undefined); res.json({ success: true, message: 'Template deployed successfully' }); } catch (deployError: any) { const rawError = deployError.message || String(deployError); @@ -1316,14 +1348,14 @@ app.post('/api/templates/deploy', async (req: Request, res: Response) => { if (shouldRollback) { try { // Stage 1: Tell Docker to clean up ghost networks/containers - await composeService.downStack(stackName); + await ComposeService.getInstance(req.nodeId).downStack(stackName); } catch (downErr) { console.error("Rollback Stage 1 (Docker down) failed:", downErr); } try { // Stage 2: Obliterate the files - await fileSystemService.deleteStack(stackName); + await FileSystemService.getInstance(req.nodeId).deleteStack(stackName); } catch (fsErr) { console.error("Rollback Stage 2 (File deletion) failed:", fsErr); } @@ -1473,7 +1505,8 @@ async function startServer() { try { // Run migration before starting server console.log('Running stack migration check...'); - await fileSystemService.migrateFlatToDirectory(); + const defaultFsService = FileSystemService.getInstance(NodeRegistry.getInstance().getDefaultNodeId()); + await defaultFsService.migrateFlatToDirectory(); console.log('Migration check completed'); } catch (error) { console.error('Migration failed:', error); diff --git a/backend/src/services/ComposeService.ts b/backend/src/services/ComposeService.ts index 1caf2e32..442981b4 100644 --- a/backend/src/services/ComposeService.ts +++ b/backend/src/services/ComposeService.ts @@ -4,97 +4,54 @@ import path from 'path'; import WebSocket from 'ws'; import DockerController from './DockerController'; import { LogFormatter } from './LogFormatter'; +import { NodeRegistry } from './NodeRegistry'; +import { Client as SSHClient } from 'ssh2'; const execAsync = promisify(exec); export class ComposeService { private baseDir: string; + private nodeId: number; - constructor() { - this.baseDir = process.env.COMPOSE_DIR || '/app/compose'; + constructor(nodeId?: number) { + this.nodeId = nodeId ?? NodeRegistry.getInstance().getDefaultNodeId(); + this.baseDir = NodeRegistry.getInstance().getComposeDir(this.nodeId); + } + + public static getInstance(nodeId?: number): ComposeService { + return new ComposeService(nodeId); } /** - * Run docker compose up or down command - * CRITICAL: cwd is set to the stack directory so relative paths in compose files - * resolve correctly inside the isolated stack folder + * Universal command execution (Local or Remote SSH) */ - async runCommand(stackName: string, action: 'down' | 'start' | 'stop' | 'restart', ws?: WebSocket): Promise { - const stackDir = path.join(this.baseDir, stackName); + private async executeCommand( + command: string, + args: string[], + cwd: string, + ws?: WebSocket, + throwOnError = true + ): Promise { + const node = NodeRegistry.getInstance().getNode(this.nodeId); + if (!node) throw new Error(`Node ${this.nodeId} not found`); - // Run docker compose from within the stack directory - // This ensures relative paths (e.g., ./data:/config) resolve correctly - const args = ['compose', action]; - - return new Promise((resolve, reject) => { - const child = spawn('docker', args, { - cwd: stackDir, // CRITICAL: Set working directory to stack folder - env: { - ...process.env, - PATH: process.env.PATH || '/usr/local/sbin:/usr/local/bin:/usr/sbin:/usr/bin:/sbin:/bin' - } - }); - - if (ws) { - child.stdout.on('data', (data: Buffer) => { - ws.send(data.toString()); - }); - - child.stderr.on('data', (data: Buffer) => { - ws.send(data.toString()); - }); - - child.on('close', (code: number | null) => { - ws.send(`Command exited with code ${code}\n`); - if (code === 0) resolve(); - else reject(new Error(`Command exited with code ${code}`)); - }); - - child.on('error', (error: Error) => { - console.error(`Docker Compose Error for ${stackName}:`, error.message); - ws.send(`Error: ${error.message}\n`); - reject(error); - }); - } else { - // Without WS, just wait for resolution - let stderr = ''; - child.stdout.on('data', () => { }); // Drain stdout to prevent pipe buffer from blocking the process - child.stderr.on('data', (data: Buffer) => { stderr += data.toString(); }); - - child.on('close', (code: number | null) => { - if (code === 0) resolve(); - else reject(new Error(`Command failed with code ${code}. Stderr: ${stderr}`)); - }); - - child.on('error', (error: Error) => { - console.error(`Docker Compose Error for ${stackName}:`, error.message); - reject(error); - }); - } - }); - } - - /** - * Deploy stack: executes up -d --remove-orphans and awaits completion. - */ - async deployStack(stackName: string, ws?: WebSocket): Promise { - const stackDir = path.join(this.baseDir, stackName); - - try { - const dockerController = DockerController.getInstance(); - const legacyContainers = await dockerController.getContainersByStack(stackName); - if (legacyContainers && legacyContainers.length > 0) { - if (ws) ws.send(`=== Cleaning up existing containers for clean deployment ===\n`); - await dockerController.removeContainers(legacyContainers.map(c => c.Id)); - } - } catch (e) { - console.warn(`Failed to clean up legacy containers for ${stackName}:`, e); + if (node.type === 'local' || !node.host) { + return this.executeLocal(command, args, cwd, ws, throwOnError); + } else { + return this.executeRemote(node, command, args, cwd, ws, throwOnError); } + } - await new Promise((resolve, reject) => { - const args = ['compose', 'up', '-d', '--remove-orphans']; - const child = spawn('docker', args, { - cwd: stackDir, + private async executeLocal( + command: string, + args: string[], + cwd: string, + ws?: WebSocket, + throwOnError = true + ): Promise { + return new Promise((resolve, reject) => { + const child = spawn(command, args, { + cwd, env: { ...process.env, PATH: process.env.PATH || '/usr/local/sbin:/usr/local/bin:/usr/sbin:/usr/bin:/sbin:/bin' @@ -103,46 +60,117 @@ export class ComposeService { let errorLog = ''; - if (ws) { - child.stdout.on('data', (data: Buffer) => { - const text = data.toString(); - errorLog += text; + const onData = (data: Buffer) => { + const text = data.toString(); + errorLog += text; + if (ws && ws.readyState === WebSocket.OPEN) { ws.send(text); - }); - child.stderr.on('data', (data: Buffer) => { - const text = data.toString(); - errorLog += text; - ws.send(text); - }); - child.on('close', (code: number | null) => { + } + }; + + child.stdout.on('data', onData); + child.stderr.on('data', onData); + + child.on('close', (code: number | null) => { + if (ws && ws.readyState === WebSocket.OPEN) { ws.send(`Command exited with code ${code}\n`); - if (code === 0) resolve(); - else reject(new Error(errorLog.trim() || `Command failed with code ${code}`)); - }); - } else { - child.stdout.on('data', (data: Buffer) => { - errorLog += data.toString(); - }); - child.stderr.on('data', (data: Buffer) => { - errorLog += data.toString(); - }); - child.on('close', (code: number | null) => { - if (code === 0) resolve(); - else reject(new Error(errorLog.trim() || `Command failed with code ${code}`)); - }); - } + } + if (code === 0) resolve(); + else if (throwOnError) reject(new Error(errorLog.trim() || `Command failed with code ${code}`)); + else resolve(); + }); child.on('error', (error: Error) => { - console.error(`Docker Compose Deploy Error for ${stackName}:`, error.message); - if (ws) ws.send(`Error: ${error.message}\n`); - reject(error); + if (ws && ws.readyState === WebSocket.OPEN) { + ws.send(`Error: ${error.message}\n`); + } + if (throwOnError) reject(error); + else resolve(); }); }); + } + + private async executeRemote( + node: any, + command: string, + args: string[], + cwd: string, + ws?: WebSocket, + throwOnError = true + ): Promise { + return new Promise((resolve, reject) => { + const conn = new SSHClient(); + let errorLog = ''; + + conn.on('ready', () => { + const cmdString = `cd "${cwd}" && ${command} ${args.map(a => `"${a}"`).join(' ')}`; + + conn.exec(cmdString, (err, stream) => { + if (err) { + conn.end(); + if (throwOnError) reject(err); else resolve(); + return; + } + + const onData = (data: any) => { + const text = data.toString(); + errorLog += text; + if (ws && ws.readyState === WebSocket.OPEN) { + ws.send(text); + } + }; + + stream.on('data', onData).stderr.on('data', onData); + + stream.on('close', (code: any, signal: any) => { + conn.end(); + if (ws && ws.readyState === WebSocket.OPEN) { + ws.send(`Command exited with code ${code}\n`); + } + if (code === 0) resolve(); + else if (throwOnError) reject(new Error(errorLog.trim() || `Command failed with code ${code}`)); + else resolve(); + }); + }); + }).on('error', (err) => { + if (ws && ws.readyState === WebSocket.OPEN) ws.send(`SSH Error: ${err.message}\n`); + if (throwOnError) reject(err); else resolve(); + }).connect({ + host: node.host, + port: node.port || 22, + username: node.ssh_user!, + password: node.ssh_password, + privateKey: node.ssh_key, + readyTimeout: 10000, + }); + }); + } + + async runCommand(stackName: string, action: 'down' | 'start' | 'stop' | 'restart', ws?: WebSocket): Promise { + const stackDir = path.join(this.baseDir, stackName); + await this.executeCommand('docker', ['compose', action], stackDir, ws); + } + + async deployStack(stackName: string, ws?: WebSocket): Promise { + const stackDir = path.join(this.baseDir, stackName); + + try { + const dockerController = DockerController.getInstance(this.nodeId); + const legacyContainers = await dockerController.getContainersByStack(stackName); + if (legacyContainers && legacyContainers.length > 0) { + if (ws && ws.readyState === WebSocket.OPEN) ws.send(`=== Cleaning up existing containers for clean deployment ===\n`); + await dockerController.removeContainers(legacyContainers.map((c: any) => c.Id)); + } + } catch (e) { + console.warn(`Failed to clean up legacy containers for ${stackName}:`, e); + } + + await this.executeCommand('docker', ['compose', 'up', '-d', '--remove-orphans'], stackDir, ws); // Post-Deploy Health Probe await new Promise(resolve => setTimeout(resolve, 3000)); - const dockerController = DockerController.getInstance(); + const dockerController = DockerController.getInstance(this.nodeId); const containers = await dockerController.getDocker().listContainers({ all: true, filters: { label: [`com.docker.compose.project=${stackName}`] } @@ -156,35 +184,25 @@ export class ComposeService { if (exitCode !== 0) { const logs = await container.logs({ stdout: true, stderr: true, tail: 50 }); - // Strip Docker log multiplex headers if they exist let logStr = logs.toString('utf-8'); throw new Error(`CONTAINER_CRASHED\nExit Code: ${exitCode}\n${logStr}`); } } } - } - /** - * Stream docker logs for a stack via WebSocket. - * Supervisors: Fetches containers via DockerController and spawns `docker logs -f` for each. - * Automatically re-spawns on process exit if WebSocket is still OPEN, creating a persistent stream. - * Kills the child processes when the WebSocket closes. - */ streamLogs(stackName: string, ws: WebSocket) { let isClosed = false; let isFirstRun = true; let isWaitingForActivity = false; - ws.on('close', () => { - isClosed = true; - }); + ws.on('close', () => { isClosed = true; }); const startStream = async () => { if (isClosed || ws.readyState !== WebSocket.OPEN) return; try { - const dockerController = DockerController.getInstance(); + const dockerController = DockerController.getInstance(this.nodeId); const containers = await dockerController.getContainersByStack(stackName); if (!containers || containers.length === 0) { @@ -196,9 +214,8 @@ export class ComposeService { return; } - const runningContainers = containers.filter(c => c.State === 'running'); + const runningContainers = containers.filter((c: any) => c.State === 'running'); - // If not first run and no containers are running, we poll to wait for activity if (!isFirstRun && runningContainers.length === 0) { if (!isWaitingForActivity) { ws.send(`\r\n\x1b[33m[Sencho] Log stream ended. Waiting for container activity...\x1b[0m\r\n`); @@ -208,20 +225,19 @@ export class ComposeService { return; } - // On first run, we stream all containers to dump history. - // On subsequent runs, we only attach to running containers to avoid immediate exit loop. const containersToLog = isFirstRun ? containers : runningContainers; isFirstRun = false; - isWaitingForActivity = false; // Reset since we are tracking active elements + isWaitingForActivity = false; let activeProcesses = 0; let streamEndedHandled = false; - const childProcesses: ReturnType[] = []; + + let clientConnections: any[] = []; + let localProcesses: ReturnType[] = []; const onWsClose = () => { - childProcesses.forEach(cp => { - try { cp.kill(); } catch { /* ignore */ } - }); + localProcesses.forEach(cp => { try { cp.kill(); } catch {} }); + clientConnections.forEach(conn => { try { conn.end(); } catch {} }); }; ws.on('close', onWsClose); @@ -231,35 +247,25 @@ export class ComposeService { if (activeProcesses <= 0 && !streamEndedHandled) { streamEndedHandled = true; ws.removeListener('close', onWsClose); - if (!isClosed && ws.readyState === WebSocket.OPEN) { setTimeout(startStream, 1000); } } }; + const node = NodeRegistry.getInstance().getNode(this.nodeId); + const isRemote = (node && node.type === 'remote' && node.host); + for (const container of containersToLog) { const containerName = container.Names?.[0]?.replace(/^\//, '') || container.Id; - const child = spawn('docker', ['logs', '-f', '--tail', '100', containerName], { - env: { - ...process.env, - PATH: process.env.PATH || '/usr/local/sbin:/usr/local/bin:/usr/sbin:/usr/bin:/sbin:/bin' - } - }); - activeProcesses++; - childProcesses.push(child); - let lineBuffer = ''; - const sendOutput = (data: Buffer) => { + const sendOutput = (data: Buffer | any) => { if (ws.readyState === WebSocket.OPEN) { lineBuffer += data.toString(); const lines = lineBuffer.split(/\r?\n/); - - // The last element is either an incomplete line or empty string lineBuffer = lines.pop() || ''; - for (const line of lines) { const formattedLine = LogFormatter.process(line); ws.send(formattedLine + '\r\n'); @@ -267,20 +273,61 @@ export class ComposeService { } }; - child.stdout.on('data', sendOutput); - child.stderr.on('data', sendOutput); - child.on('error', handleProcessEnd); - child.on('close', () => { - // Flush any remaining partial line before ending - if (lineBuffer && ws.readyState === WebSocket.OPEN) { - const formattedLine = LogFormatter.process(lineBuffer); - ws.send(formattedLine + '\r\n'); - lineBuffer = ''; + const flushBuffer = () => { + if (lineBuffer && ws.readyState === WebSocket.OPEN) { + const formattedLine = LogFormatter.process(lineBuffer); + ws.send(formattedLine + '\r\n'); + lineBuffer = ''; } - handleProcessEnd(); - }); - } + } + if (isRemote) { + const conn = new SSHClient(); + clientConnections.push(conn); + conn.on('ready', () => { + conn.exec(`docker logs -f --tail 100 "${containerName}"`, { + env: { + ...process.env, + PATH: process.env.PATH || '/usr/local/sbin:/usr/local/bin:/usr/sbin:/usr/bin:/sbin:/bin' + } + }, (err, stream) => { + if (err) { + conn.end(); + handleProcessEnd(); + return; + } + stream.on('data', sendOutput).stderr.on('data', sendOutput); + stream.on('close', () => { + flushBuffer(); + conn.end(); + handleProcessEnd(); + }); + }); + }).on('error', handleProcessEnd).connect({ + host: node!.host, + port: node!.port || 22, + username: node!.ssh_user!, + password: node!.ssh_password, + privateKey: node!.ssh_key, + readyTimeout: 10000, + }); + } else { + const child = spawn('docker', ['logs', '-f', '--tail', '100', containerName], { + env: { + ...process.env, + PATH: process.env.PATH || '/usr/local/sbin:/usr/local/bin:/usr/sbin:/usr/bin:/sbin:/bin' + } + }); + localProcesses.push(child); + child.stdout.on('data', sendOutput); + child.stderr.on('data', sendOutput); + child.on('error', handleProcessEnd); + child.on('close', () => { + flushBuffer(); + handleProcessEnd(); + }); + } + } } catch (err) { if (!isClosed && ws.readyState === WebSocket.OPEN) { if (!isWaitingForActivity) { @@ -295,114 +342,35 @@ export class ComposeService { startStream(); } - /** - * Update stack: pull images first, then recreate containers - * CRITICAL: cwd is set to the stack directory so relative paths resolve correctly - */ async updateStack(stackName: string, ws?: WebSocket): Promise { const stackDir = path.join(this.baseDir, stackName); - const sendOutput = (data: string) => { - if (ws && ws.readyState === WebSocket.OPEN) { - ws.send(data); - } + if (ws && ws.readyState === WebSocket.OPEN) ws.send(data); }; try { - const dockerController = DockerController.getInstance(); + const dockerController = DockerController.getInstance(this.nodeId); const legacyContainers = await dockerController.getContainersByStack(stackName); if (legacyContainers && legacyContainers.length > 0) { sendOutput(`=== Cleaning up existing containers for clean update ===\n`); - await dockerController.removeContainers(legacyContainers.map(c => c.Id)); + await dockerController.removeContainers(legacyContainers.map((c: any) => c.Id)); } } catch (e) { console.warn(`Failed to clean up legacy containers for ${stackName}:`, e); } - // Step 1: Pull images sendOutput('=== Pulling latest images ===\n'); - await new Promise((resolve, reject) => { - const pullProcess = spawn('docker', ['compose', 'pull'], { - cwd: stackDir, - env: { - ...process.env, - PATH: process.env.PATH || '/usr/local/sbin:/usr/local/bin:/usr/sbin:/usr/bin:/sbin:/bin' - } - }); + await this.executeCommand('docker', ['compose', 'pull'], stackDir, ws); - pullProcess.stdout.on('data', (data: Buffer) => { - sendOutput(data.toString()); - }); - - pullProcess.stderr.on('data', (data: Buffer) => { - sendOutput(data.toString()); - }); - - pullProcess.on('close', (code: number | null) => { - if (code === 0) { - sendOutput('=== Images pulled successfully ===\n'); - resolve(); - } else { - sendOutput(`=== Pull failed with code ${code} ===\n`); - reject(new Error(`Pull failed with code ${code}`)); - } - }); - - pullProcess.on('error', (error: Error) => { - console.error(`Docker Compose Pull Error for ${stackName}:`, error.message); - sendOutput(`Pull error: ${error.message}\n`); - reject(error); - }); - }); - - // Step 2: Recreate containers with new images sendOutput('=== Recreating containers ===\n'); - await new Promise((resolve, reject) => { - const upProcess = spawn('docker', ['compose', 'up', '-d', '--remove-orphans'], { - cwd: stackDir, - env: { - ...process.env, - PATH: process.env.PATH || '/usr/local/sbin:/usr/local/bin:/usr/sbin:/usr/bin:/sbin:/bin' - } - }); - - let errorLog = ''; - - upProcess.stdout.on('data', (data: Buffer) => { - const text = data.toString(); - errorLog += text; - sendOutput(text); - }); - - upProcess.stderr.on('data', (data: Buffer) => { - const text = data.toString(); - errorLog += text; - sendOutput(text); - }); - - upProcess.on('close', (code: number | null) => { - if (code === 0) { - sendOutput('=== Stack updated successfully ===\n'); - resolve(); - } else { - sendOutput(`=== Update failed with code ${code} ===\n`); - reject(new Error(errorLog.trim() || `Up failed with code ${code}`)); - } - }); - - upProcess.on('error', (error: Error) => { - console.error(`Docker Compose Up Error for ${stackName}:`, error.message); - sendOutput(`Update error: ${error.message}\n`); - reject(error); - }); - }); + await this.executeCommand('docker', ['compose', 'up', '-d', '--remove-orphans'], stackDir, ws); + sendOutput('=== Stack updated successfully ===\n'); } public async downStack(stackName: string): Promise { const stackPath = path.join(this.baseDir, stackName); try { - // Run down to clean up any partially created networks or containers - await execAsync(`docker compose -f compose.yaml down`, { cwd: stackPath }); + await this.executeCommand('docker', ['compose', 'down'], stackPath, undefined, false); } catch (error) { console.warn(`[Teardown] Docker down failed or nothing to clean up for ${stackName}`); } diff --git a/backend/src/services/DatabaseService.ts b/backend/src/services/DatabaseService.ts index 7d632652..dc28f68b 100644 --- a/backend/src/services/DatabaseService.ts +++ b/backend/src/services/DatabaseService.ts @@ -26,7 +26,7 @@ export interface StackAlert { } export interface Node { - id?: number; + id: number; name: string; type: 'local' | 'remote'; host: string; @@ -35,6 +35,12 @@ export interface Node { is_default: boolean; status: 'online' | 'offline' | 'unknown'; created_at: number; + ssh_user?: string; + ssh_password?: string; + ssh_key?: string; + tls_ca?: string; + tls_cert?: string; + tls_key?: string; } export interface NotificationHistory { @@ -134,8 +140,20 @@ export class DatabaseService { status TEXT NOT NULL DEFAULT 'unknown', created_at INTEGER NOT NULL ); + ); `); + // Apply migrations safely (ignore if columns already exist) + const maybeAddCol = (table: string, col: string, def: string) => { + try { this.db.prepare(`ALTER TABLE ${table} ADD COLUMN ${col} ${def}`).run(); } catch (e) { /* ignore */ } + }; + maybeAddCol('nodes', 'ssh_user', "TEXT DEFAULT ''"); + maybeAddCol('nodes', 'ssh_password', "TEXT DEFAULT ''"); + maybeAddCol('nodes', 'ssh_key', "TEXT DEFAULT ''"); + maybeAddCol('nodes', 'tls_ca', "TEXT DEFAULT ''"); + maybeAddCol('nodes', 'tls_cert', "TEXT DEFAULT ''"); + maybeAddCol('nodes', 'tls_key', "TEXT DEFAULT ''"); + // Initialize default global settings if they don't exist const stmt = this.db.prepare('INSERT OR IGNORE INTO global_settings (key, value) VALUES (?, ?)'); stmt.run('host_cpu_limit', '90'); @@ -351,7 +369,7 @@ export class DatabaseService { this.db.prepare('UPDATE nodes SET is_default = 0').run(); } const stmt = this.db.prepare( - 'INSERT INTO nodes (name, type, host, port, compose_dir, is_default, status, created_at) VALUES (?, ?, ?, ?, ?, ?, ?, ?)' + 'INSERT INTO nodes (name, type, host, port, compose_dir, is_default, status, created_at, ssh_user, ssh_password, ssh_key, tls_ca, tls_cert, tls_key) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)' ); const result = stmt.run( node.name, @@ -361,7 +379,13 @@ export class DatabaseService { node.compose_dir, node.is_default ? 1 : 0, 'unknown', - Date.now() + Date.now(), + node.ssh_user || '', + node.ssh_password || '', + node.ssh_key || '', + node.tls_ca || '', + node.tls_cert || '', + node.tls_key || '' ); return result.lastInsertRowid as number; } @@ -385,6 +409,12 @@ export class DatabaseService { if (updates.compose_dir !== undefined) { fields.push('compose_dir = ?'); values.push(updates.compose_dir); } if (updates.is_default !== undefined) { fields.push('is_default = ?'); values.push(updates.is_default ? 1 : 0); } if (updates.status !== undefined) { fields.push('status = ?'); values.push(updates.status); } + if (updates.ssh_user !== undefined) { fields.push('ssh_user = ?'); values.push(updates.ssh_user); } + if (updates.ssh_password !== undefined) { fields.push('ssh_password = ?'); values.push(updates.ssh_password); } + if (updates.ssh_key !== undefined) { fields.push('ssh_key = ?'); values.push(updates.ssh_key); } + if (updates.tls_ca !== undefined) { fields.push('tls_ca = ?'); values.push(updates.tls_ca); } + if (updates.tls_cert !== undefined) { fields.push('tls_cert = ?'); values.push(updates.tls_cert); } + if (updates.tls_key !== undefined) { fields.push('tls_key = ?'); values.push(updates.tls_key); } if (fields.length === 0) return; diff --git a/backend/src/services/DockerController.ts b/backend/src/services/DockerController.ts index 17832227..dccac8e4 100644 --- a/backend/src/services/DockerController.ts +++ b/backend/src/services/DockerController.ts @@ -6,23 +6,23 @@ import path from 'path'; import fs from 'fs/promises'; import * as yaml from 'yaml'; +import { NodeRegistry } from './NodeRegistry'; + const execAsync = promisify(exec); const COMPOSE_DIR = process.env.COMPOSE_DIR || '/app/compose'; class DockerController { - private static instance: DockerController; private docker: Docker; + private nodeId: number; - private constructor() { - // Use default constructor to support both Windows named pipes and Linux sockets - this.docker = new Docker(); + private constructor(nodeId: number) { + this.nodeId = nodeId; + this.docker = NodeRegistry.getInstance().getDocker(nodeId); } - public static getInstance(): DockerController { - if (!DockerController.instance) { - DockerController.instance = new DockerController(); - } - return DockerController.instance; + public static getInstance(nodeId?: number): DockerController { + const id = nodeId ?? NodeRegistry.getInstance().getDefaultNodeId(); + return new DockerController(id); } public getDocker(): Docker { @@ -533,7 +533,8 @@ let lastNetSum = { rx: 0, tx: 0, timestamp: Date.now() }; export const updateGlobalDockerNetwork = async () => { try { - const dockerController = DockerController.getInstance(); + const nodeId = NodeRegistry.getInstance().getDefaultNodeId(); + const dockerController = DockerController.getInstance(nodeId); const containers = await dockerController.getRunningContainers(); const statsResults = await Promise.allSettled( diff --git a/backend/src/services/FileSystemService.ts b/backend/src/services/FileSystemService.ts index 15dc2cbd..6bfbe781 100644 --- a/backend/src/services/FileSystemService.ts +++ b/backend/src/services/FileSystemService.ts @@ -1,11 +1,29 @@ -import { promises as fs, Dirent } from 'fs'; import path from 'path'; +import { IFileAdapter } from './fs/IFileAdapter'; +import { LocalFileAdapter } from './fs/LocalFileAdapter'; +import { SSHFileAdapter } from './fs/SSHFileAdapter'; +import { NodeRegistry } from './NodeRegistry'; export class FileSystemService { private baseDir: string; + private adapter: IFileAdapter; + private nodeId: number; - constructor() { + constructor(nodeId?: number) { this.baseDir = process.env.COMPOSE_DIR || '/app/compose'; + this.nodeId = nodeId ?? NodeRegistry.getInstance().getDefaultNodeId(); + + const node = NodeRegistry.getInstance().getNode(this.nodeId); + + if (!node || node.type === 'local' || !node.host) { + this.adapter = new LocalFileAdapter(); + } else { + this.adapter = new SSHFileAdapter(node); + } + } + + public static getInstance(nodeId?: number): FileSystemService { + return new FileSystemService(nodeId); } /** @@ -16,7 +34,7 @@ export class FileSystemService { for (const file of composeFiles) { try { - await fs.access(path.join(dir, file)); + await this.adapter.access(path.join(dir, file)); return true; } catch { // Continue checking other options @@ -37,7 +55,7 @@ export class FileSystemService { for (const file of composeFiles) { const filePath = path.join(stackDir, file); try { - await fs.access(filePath); + await this.adapter.access(filePath); return filePath; } catch { // Continue checking other options @@ -53,7 +71,7 @@ export class FileSystemService { */ async getStacks(): Promise { try { - const items = await fs.readdir(this.baseDir, { withFileTypes: true }); + const items = await this.adapter.readdir(this.baseDir, { withFileTypes: true }); const stackNames: string[] = []; for (const item of items) { @@ -80,7 +98,7 @@ export class FileSystemService { async getStackContent(stackName: string): Promise { try { const filePath = await this.getComposeFilePath(stackName); - return await fs.readFile(filePath, 'utf-8'); + return await this.adapter.readFile(filePath, 'utf-8'); } catch (error) { console.error('Error reading stack content:', error); throw new Error(`Failed to read stack: ${stackName}`); @@ -98,7 +116,7 @@ export class FileSystemService { console.log('Saving to path:', filePath); try { - await fs.writeFile(filePath, content, 'utf-8'); + await this.adapter.writeFile(filePath, content, 'utf-8'); console.log('File written successfully'); } catch (error) { console.error('Error writing file:', error); @@ -112,20 +130,34 @@ export class FileSystemService { async envExists(stackName: string): Promise { const envPath = path.join(this.baseDir, stackName, '.env'); try { - await fs.access(envPath); + await this.adapter.access(envPath); return true; } catch { return false; } } + // Proxy to adapter read/write operations for use in other services and generic routes + + async readFile(filePath: string, encoding: BufferEncoding = 'utf-8'): Promise { + return this.adapter.readFile(filePath, encoding); + } + + async writeFile(filePath: string, content: string, encoding: BufferEncoding = 'utf-8'): Promise { + return this.adapter.writeFile(filePath, content, encoding); + } + + async access(filePath: string): Promise { + return this.adapter.access(filePath); + } + /** * Get the content of a stack's .env file */ async getEnvContent(stackName: string): Promise { const envPath = path.join(this.baseDir, stackName, '.env'); try { - return await fs.readFile(envPath, 'utf-8'); + return await this.adapter.readFile(envPath, 'utf-8'); } catch (error) { console.error('Error reading env file:', error); throw new Error(`Failed to read env file for stack: ${stackName}`); @@ -140,7 +172,7 @@ export class FileSystemService { console.log('Saving env to path:', envPath); try { - await fs.writeFile(envPath, content, 'utf-8'); + await this.adapter.writeFile(envPath, content, 'utf-8'); console.log('Env file written successfully'); } catch (error) { console.error('Error writing env file:', error); @@ -161,7 +193,7 @@ export class FileSystemService { // Check if directory already exists try { - await fs.access(stackDir); + await this.adapter.access(stackDir); throw new Error(`Stack "${stackName}" already exists`); } catch (error: any) { if (error.message.includes('already exists')) { @@ -171,7 +203,7 @@ export class FileSystemService { } // Create the directory - await fs.mkdir(stackDir, { recursive: true }); + await this.adapter.mkdir(stackDir, { recursive: true }); // Write boilerplate compose.yaml const composePath = path.join(stackDir, 'compose.yaml'); @@ -183,7 +215,7 @@ export class FileSystemService { restart: always `; try { - await fs.writeFile(composePath, boilerplate, 'utf-8'); + await this.adapter.writeFile(composePath, boilerplate, 'utf-8'); console.log('Stack created successfully:', stackName); } catch (error) { console.error('Error creating stack:', error); @@ -198,7 +230,7 @@ export class FileSystemService { const stackDir = path.join(this.baseDir, stackName); try { - await fs.rm(stackDir, { recursive: true, force: true }); + await this.adapter.rm(stackDir, { recursive: true, force: true }); console.log('Stack deleted successfully:', stackName); } catch (error: any) { if (error.code !== 'ENOENT') { @@ -223,14 +255,14 @@ export class FileSystemService { try { // Ensure base directory exists try { - await fs.access(this.baseDir); + await this.adapter.access(this.baseDir); } catch { console.log('Creating compose directory:', this.baseDir); - await fs.mkdir(this.baseDir, { recursive: true }); + await this.adapter.mkdir(this.baseDir, { recursive: true }); return; // No files to migrate in a new directory } - const items = await fs.readdir(this.baseDir, { withFileTypes: true }); + const items = await this.adapter.readdir(this.baseDir, { withFileTypes: true }); for (const item of items) { // Only process .yml/.yaml files (skip directories and other files) @@ -242,7 +274,7 @@ export class FileSystemService { // Check if target directory already exists try { - await fs.access(stackDir); + await this.adapter.access(stackDir); console.log(`Skipping migration for "${stackName}": directory already exists`); continue; } catch { @@ -252,19 +284,19 @@ export class FileSystemService { console.log(`Migrating stack: ${stackName}`); // Create the stack directory - await fs.mkdir(stackDir, { recursive: true }); + await this.adapter.mkdir(stackDir, { recursive: true }); // Move compose file to new location (standardize on compose.yaml) const oldComposePath = path.join(this.baseDir, item.name); const newComposePath = path.join(stackDir, 'compose.yaml'); - await fs.rename(oldComposePath, newComposePath); + await this.adapter.rename(oldComposePath, newComposePath); // Move env file if it exists (old pattern: stackname.env) const oldEnvPath = path.join(this.baseDir, `${stackName}.env`); const newEnvPath = path.join(stackDir, '.env'); try { - await fs.access(oldEnvPath); - await fs.rename(oldEnvPath, newEnvPath); + await this.adapter.access(oldEnvPath); + await this.adapter.rename(oldEnvPath, newEnvPath); console.log(`Migrated env file for: ${stackName}`); } catch { // No env file to migrate, that's fine diff --git a/backend/src/services/MonitorService.ts b/backend/src/services/MonitorService.ts index 793f60b4..1905ae0a 100644 --- a/backend/src/services/MonitorService.ts +++ b/backend/src/services/MonitorService.ts @@ -118,31 +118,30 @@ export class MonitorService { // 2. Global Crash Detect if (settings['global_crash'] === '1') { try { - const docker = DockerController.getInstance(); - const containers = await docker.getAllContainers(); - for (const c of containers) { - if (c.State === 'exited' || String(c.Status).includes('unhealthy')) { - // Basic deduplication could be added, but for now we just dispatch. - // Usually, users will restart or remove the container. - // To prevent massive spam, we check if it exited in the last 30 secs - if (c.State === 'exited') { - // Check if it exited recently - if (c.Status.includes('seconds ago')) { - // Extract the exit code from the status string (e.g., "Exited (143) 5 seconds ago") - const match = c.Status.match(/Exited \((\d+)\)/i); - const exitCode = match ? parseInt(match[1], 10) : null; - - // 0: Success, 137: SIGKILL (Force Stop), 143: SIGTERM (Graceful Stop), 255: Docker Daemon Stop - const intentionalExitCodes = [0, 137, 143, 255]; - - // Only alert if we found a code AND it is not an intentional stop - if (exitCode !== null && !intentionalExitCodes.includes(exitCode)) { - await notifier.dispatchAlert('error', `Container Crash Detected: ${c.Names[0]} exited unexpectedly (Code: ${exitCode}).`); + const nodes = DatabaseService.getInstance().getNodes(); + for (const node of nodes) { + if (!node.id) continue; + try { + const docker = DockerController.getInstance(node.id); + const containers = await docker.getAllContainers(); + for (const c of containers) { + if (c.State === 'exited' || String(c.Status).includes('unhealthy')) { + if (c.State === 'exited') { + if (c.Status.includes('seconds ago')) { + const match = c.Status.match(/Exited \((\d+)\)/i); + const exitCode = match ? parseInt(match[1], 10) : null; + const intentionalExitCodes = [0, 137, 143, 255]; + if (exitCode !== null && !intentionalExitCodes.includes(exitCode)) { + await notifier.dispatchAlert('error', `[Node: ${node.name}] Container Crash Detected: ${c.Names[0]} exited unexpectedly (Code: ${exitCode}).`); + } + } + } else if (String(c.Status).includes('unhealthy')) { + await notifier.dispatchAlert('error', `[Node: ${node.name}] Healthcheck Failed: Container ${c.Names[0]} is unhealthy.`); } } - } else if (String(c.Status).includes('unhealthy')) { - await notifier.dispatchAlert('error', `Healthcheck Failed: Container ${c.Names[0]} is unhealthy.`); } + } catch (err) { + console.error(`Error checking crashes on node ${node.name}`, err); } } } catch (e) { @@ -203,90 +202,98 @@ export class MonitorService { private async evaluateStackAlerts(db: DatabaseService) { const alerts = db.getStackAlerts(); - const docker = DockerController.getInstance(); + const nodes = db.getNodes(); - try { - const containers = await docker.getRunningContainers(); - for (const container of containers) { - const stackName = container.Labels?.['com.docker.compose.project'] || 'system'; + for (const node of nodes) { + if (!node.id) continue; + try { + const docker = DockerController.getInstance(node.id); + const containers = await docker.getRunningContainers(); + for (const container of containers) { + const stackName = container.Labels?.['com.docker.compose.project'] || 'system'; - try { - const rawStats = await docker.getContainerStatsStream(container.Id); - const stats = JSON.parse(rawStats); + try { + const rawStats = await docker.getContainerStatsStream(container.Id); + const stats = JSON.parse(rawStats); - const metrics = { - cpu_percent: this.calculateCpuPercent(stats), - memory_percent: this.calculateMemoryPercent(stats), - memory_mb: (stats.memory_stats?.usage || 0) / (1024 * 1024), - net_rx: this.calculateNetwork(stats, 'rx'), - net_tx: this.calculateNetwork(stats, 'tx'), - restart_count: 0 // Simplification since ContainerInfo doesn't have it natively - }; + const metrics = { + cpu_percent: this.calculateCpuPercent(stats), + memory_percent: this.calculateMemoryPercent(stats), + memory_mb: (stats.memory_stats?.usage || 0) / (1024 * 1024), + net_rx: this.calculateNetwork(stats, 'rx'), + net_tx: this.calculateNetwork(stats, 'tx'), + restart_count: 0 // Simplification since ContainerInfo doesn't have it natively + }; - db.addContainerMetric({ - container_id: container.Id, - stack_name: stackName, - cpu_percent: metrics.cpu_percent || 0, - memory_mb: metrics.memory_mb || 0, - net_rx_mb: metrics.net_rx || 0, - net_tx_mb: metrics.net_tx || 0, - timestamp: Date.now() - }); + db.addContainerMetric({ + container_id: container.Id, + stack_name: stackName, + cpu_percent: metrics.cpu_percent || 0, + memory_mb: metrics.memory_mb || 0, + net_rx_mb: metrics.net_rx || 0, + net_tx_mb: metrics.net_tx || 0, + timestamp: Date.now() + }); - const stackAlerts = alerts.filter(a => a.stack_name === stackName); - for (const rule of stackAlerts) { - const ruleId = rule.id!; - const currentValue = metrics[rule.metric as keyof typeof metrics]; + const stackAlerts = alerts.filter(a => a.stack_name === stackName); + for (const rule of stackAlerts) { + const ruleId = rule.id!; + const currentValue = metrics[rule.metric as keyof typeof metrics]; - if (currentValue === undefined) continue; + if (currentValue === undefined) continue; - const isBreaching = this.evaluateCondition(currentValue, rule.operator, rule.threshold); + const isBreaching = this.evaluateCondition(currentValue, rule.operator, rule.threshold); - if (isBreaching) { - if (!this.activeBreaches.has(ruleId)) { - this.activeBreaches.set(ruleId, { breachStartedAt: Date.now() }); - } + if (isBreaching) { + if (!this.activeBreaches.has(ruleId)) { + this.activeBreaches.set(ruleId, { breachStartedAt: Date.now() }); + } - const breachState = this.activeBreaches.get(ruleId)!; - const durationMs = Date.now() - breachState.breachStartedAt; - const requiredDurationMs = rule.duration_mins * 60 * 1000; + const breachState = this.activeBreaches.get(ruleId)!; + const durationMs = Date.now() - breachState.breachStartedAt; + const requiredDurationMs = rule.duration_mins * 60 * 1000; - if (durationMs >= requiredDurationMs) { - // Duration met! Check cooldown - const timeSinceLastFired = Date.now() - (rule.last_fired_at || 0); - const requiredCooldownMs = rule.cooldown_mins * 60 * 1000; + if (durationMs >= requiredDurationMs) { + // Duration met! Check cooldown + const timeSinceLastFired = Date.now() - (rule.last_fired_at || 0); + const requiredCooldownMs = rule.cooldown_mins * 60 * 1000; - if (timeSinceLastFired >= requiredCooldownMs) { - // Formatted Alert Message - const { name: metricName, unit } = getMetricDetails(rule.metric); - const operatorPhrase = getOperatorPhrase(rule.operator); + if (timeSinceLastFired >= requiredCooldownMs) { + // Formatted Alert Message + const { name: metricName, unit } = getMetricDetails(rule.metric); + const operatorPhrase = getOperatorPhrase(rule.operator); - const safeCurrent = typeof currentValue === 'number' ? Number(currentValue.toFixed(2)) : currentValue; - const safeThreshold = typeof rule.threshold === 'number' ? Number(rule.threshold.toFixed(2)) : rule.threshold; + const safeCurrent = typeof currentValue === 'number' ? Number(currentValue.toFixed(2)) : currentValue; + const safeThreshold = typeof rule.threshold === 'number' ? Number(rule.threshold.toFixed(2)) : rule.threshold; - const message = `The **${metricName}** for **${rule.stack_name}** ${operatorPhrase} **${safeThreshold}${unit}** (Currently: ${safeCurrent}${unit}).`; + const message = `[Node: ${node.name}] The **${metricName}** for **${rule.stack_name}** ${operatorPhrase} **${safeThreshold}${unit}** (Currently: ${safeCurrent}${unit}).`; - await NotificationService.getInstance().dispatchAlert( - 'warning', - message - ); + await NotificationService.getInstance().dispatchAlert( + 'warning', + message + ); - // Update last fired - db.updateStackAlertLastFired(ruleId, Date.now()); + // Update last fired + db.updateStackAlertLastFired(ruleId, Date.now()); + } + } + } else { + // Rule isn't breaching anymore, reset tracker + if (this.activeBreaches.has(ruleId)) { + this.activeBreaches.delete(ruleId); } } - } else { - // Rule isn't breaching anymore, reset tracker - if (this.activeBreaches.has(ruleId)) { - this.activeBreaches.delete(ruleId); - } } + } catch (e) { + console.error(`Error parsing stats for container ${container.Id} on node ${node.name}`, e); } - } catch (e) { - console.error(`Error parsing stats for container ${container.Id}`, e); } + } catch (err) { + console.error(`Error fetching containers for node ${node.name}`, err); } + } + try { db.cleanupOldMetrics(24); } catch (e) { } } diff --git a/backend/src/services/NodeRegistry.ts b/backend/src/services/NodeRegistry.ts index b5eb33af..186ad63d 100644 --- a/backend/src/services/NodeRegistry.ts +++ b/backend/src/services/NodeRegistry.ts @@ -66,6 +66,14 @@ export class NodeRegistry { return defaultNode?.id || 1; } + /** + * Get a node configuration by its ID. + */ + public getNode(nodeId: number): Node | undefined { + const db = DatabaseService.getInstance(); + return db.getNode(nodeId); + } + /** * Create a Docker client based on node configuration. * - Local nodes: use the default socket (Docker autodetects) diff --git a/backend/src/services/fs/IFileAdapter.ts b/backend/src/services/fs/IFileAdapter.ts new file mode 100644 index 00000000..9f8bd331 --- /dev/null +++ b/backend/src/services/fs/IFileAdapter.ts @@ -0,0 +1,9 @@ +export interface IFileAdapter { + access(filePath: string): Promise; + readdir(dirPath: string, options?: any): Promise; + readFile(filePath: string, encoding: string): Promise; + writeFile(filePath: string, content: string, encoding: string): Promise; + mkdir(dirPath: string, options?: any): Promise; + rm(targetPath: string, options?: any): Promise; + rename(oldPath: string, newPath: string): Promise; +} diff --git a/backend/src/services/fs/LocalFileAdapter.ts b/backend/src/services/fs/LocalFileAdapter.ts new file mode 100644 index 00000000..3b63a1b7 --- /dev/null +++ b/backend/src/services/fs/LocalFileAdapter.ts @@ -0,0 +1,27 @@ +import { promises as fs } from 'fs'; +import { IFileAdapter } from './IFileAdapter'; + +export class LocalFileAdapter implements IFileAdapter { + async access(filePath: string): Promise { + return fs.access(filePath); + } + async readdir(dirPath: string, options?: any): Promise { + return fs.readdir(dirPath, options); + } + async readFile(filePath: string, encoding: any): Promise { + const raw = await fs.readFile(filePath, encoding); + return typeof raw === 'string' ? raw : raw.toString(encoding || 'utf-8'); + } + async writeFile(filePath: string, content: string, encoding: any): Promise { + await fs.writeFile(filePath, content, encoding); + } + async mkdir(dirPath: string, options?: any): Promise { + await fs.mkdir(dirPath, options); + } + async rm(targetPath: string, options?: any): Promise { + return fs.rm(targetPath, options); + } + async rename(oldPath: string, newPath: string): Promise { + return fs.rename(oldPath, newPath); + } +} diff --git a/backend/src/services/fs/SSHFileAdapter.ts b/backend/src/services/fs/SSHFileAdapter.ts new file mode 100644 index 00000000..653fb1e4 --- /dev/null +++ b/backend/src/services/fs/SSHFileAdapter.ts @@ -0,0 +1,122 @@ +import Client from 'ssh2-sftp-client'; +import { IFileAdapter } from './IFileAdapter'; +import { Node } from '../DatabaseService'; + +export class SSHFileAdapter implements IFileAdapter { + private node: Node; + + constructor(node: Node) { + this.node = node; + } + + private async getClient() { + const sftp = new Client(); + await sftp.connect({ + host: this.node.host, + port: this.node.port || 22, + username: this.node.ssh_user!, + password: this.node.ssh_password, + privateKey: this.node.ssh_key, + readyTimeout: 10000, + }); + return sftp; + } + + async access(filePath: string): Promise { + const sftp = await this.getClient(); + try { + const exists = await sftp.exists(filePath); + if (!exists) throw Object.assign(new Error(), { code: 'ENOENT' }); + } finally { + await sftp.end(); + } + } + + async readdir(dirPath: string, options?: any): Promise { + const sftp = await this.getClient(); + try { + const list = await sftp.list(dirPath); + if (options?.withFileTypes) { + return list.map((item: any) => ({ + name: item.name, + isDirectory: () => item.type === 'd', + isFile: () => item.type === '-', + })); + } + return list.map((item: any) => item.name); + } catch(err: any) { + if(err.code === 2 || err.message.includes('No such file')) throw Object.assign(new Error(), { code: 'ENOENT' }); + throw err; + } finally { + await sftp.end(); + } + } + + async readFile(filePath: string, encoding: any): Promise { + const sftp = await this.getClient(); + try { + const buffer = await sftp.get(filePath); + if (Buffer.isBuffer(buffer)) { + return buffer.toString(encoding as BufferEncoding); + } + return buffer as unknown as string; + } catch(err: any) { + if(err.code === 2 || err.message.includes('No such file')) throw Object.assign(new Error(), { code: 'ENOENT' }); + throw err; + } finally { + await sftp.end(); + } + } + + async writeFile(filePath: string, content: string, encoding: any): Promise { + const sftp = await this.getClient(); + try { + await sftp.put(Buffer.from(content, encoding as BufferEncoding), filePath); + } finally { + await sftp.end(); + } + } + + async mkdir(dirPath: string, options?: any): Promise { + const sftp = await this.getClient(); + try { + const exists = await sftp.exists(dirPath); + if (!exists) { + await sftp.mkdir(dirPath, options?.recursive); + } + } finally { + await sftp.end(); + } + } + + async rm(targetPath: string, options?: any): Promise { + const sftp = await this.getClient(); + try { + const type = await sftp.exists(targetPath); + if (type === 'd') { + await sftp.rmdir(targetPath, options?.recursive); + } else if (type === '-') { + await sftp.delete(targetPath); + } else if (!options?.force) { + throw Object.assign(new Error(), { code: 'ENOENT' }); + } + } catch(err: any) { + if(err.code === 2 || err.message.includes('No such file')) { + if (options?.force) return; + throw Object.assign(new Error(), { code: 'ENOENT' }); + } + throw err; + } finally { + await sftp.end(); + } + } + + async rename(oldPath: string, newPath: string): Promise { + const sftp = await this.getClient(); + try { + await sftp.rename(oldPath, newPath); + } finally { + await sftp.end(); + } + } +} diff --git a/frontend/src/components/BashExecModal.tsx b/frontend/src/components/BashExecModal.tsx index e69e5be1..bf3a8806 100644 --- a/frontend/src/components/BashExecModal.tsx +++ b/frontend/src/components/BashExecModal.tsx @@ -118,10 +118,10 @@ export default function BashExecModal({ isOpen, onClose, containerId, containerN wsRef.current = ws; ws.onopen = () => { - // Kick off the exec session ws.send(JSON.stringify({ action: 'execContainer', containerId: containerId, + nodeId: localStorage.getItem('sencho-active-node') || undefined })); setIsConnected(true); diff --git a/frontend/src/components/EditorLayout.tsx b/frontend/src/components/EditorLayout.tsx index aba71516..09d0e0b4 100644 --- a/frontend/src/components/EditorLayout.tsx +++ b/frontend/src/components/EditorLayout.tsx @@ -193,7 +193,11 @@ export default function EditorLayout() { const wsProtocol = window.location.protocol === 'https:' ? 'wss:' : 'ws:'; const ws = new WebSocket(`${wsProtocol}//${window.location.host}`); wsMap[container.Id] = ws; - ws.onopen = () => ws.send(JSON.stringify({ action: 'streamStats', containerId: container.Id })); + ws.onopen = () => ws.send(JSON.stringify({ + action: 'streamStats', + containerId: container.Id, + nodeId: localStorage.getItem('sencho-active-node') || undefined + })); ws.onmessage = (event) => { try { const data = JSON.parse(event.data); diff --git a/frontend/src/components/GlobalObservabilityView.tsx b/frontend/src/components/GlobalObservabilityView.tsx index 3f860b4c..73205fe8 100644 --- a/frontend/src/components/GlobalObservabilityView.tsx +++ b/frontend/src/components/GlobalObservabilityView.tsx @@ -73,7 +73,8 @@ export function GlobalObservabilityView() { useEffect(() => { if (devMode) { // SSE mode - const eventSource = new EventSource('/api/logs/global/stream'); + const activeNodeId = localStorage.getItem('sencho-active-node') || ''; + const eventSource = new EventSource(`/api/logs/global/stream?nodeId=${activeNodeId}`); eventSource.onmessage = (event) => { try { diff --git a/frontend/src/components/LogViewer.tsx b/frontend/src/components/LogViewer.tsx index 5eb227e6..13125f7e 100644 --- a/frontend/src/components/LogViewer.tsx +++ b/frontend/src/components/LogViewer.tsx @@ -27,7 +27,8 @@ export function LogViewer({ containerId, containerName, isOpen, onClose }: LogVi setLogs([]); setIsConnected(false); - const eventSource = new EventSource(`/api/containers/${containerId}/logs`); + const activeNodeId = localStorage.getItem('sencho-active-node') || ''; + const eventSource = new EventSource(`/api/containers/${containerId}/logs?nodeId=${activeNodeId}`); eventSource.onopen = () => setIsConnected(true); diff --git a/frontend/src/components/Terminal.tsx b/frontend/src/components/Terminal.tsx index 97098135..031724b1 100644 --- a/frontend/src/components/Terminal.tsx +++ b/frontend/src/components/Terminal.tsx @@ -125,11 +125,12 @@ export default function TerminalComponent({ stackName }: TerminalComponentProps) const wsProtocol = window.location.protocol === 'https:' ? 'wss:' : 'ws:'; const cleanStackName = stackName?.replace(/\.(yml|yaml)$/, ''); + const activeNodeId = localStorage.getItem('sencho-active-node') || ''; // If a stackName is provided, connect to the dedicated logs WebSocket // Otherwise, fall back to the generic terminal WebSocket const wsUrl = cleanStackName - ? `${wsProtocol}//${window.location.host}/api/stacks/${cleanStackName}/logs` + ? `${wsProtocol}//${window.location.host}/api/stacks/${cleanStackName}/logs${activeNodeId ? `?nodeId=${activeNodeId}` : ''}` : `${wsProtocol}//${window.location.host}`; const ws = new WebSocket(wsUrl); diff --git a/frontend/src/lib/api.ts b/frontend/src/lib/api.ts index cfd18bfa..d73683be 100644 --- a/frontend/src/lib/api.ts +++ b/frontend/src/lib/api.ts @@ -5,10 +5,13 @@ export async function apiFetch( options: RequestInit = {} ): Promise { const url = `${API_BASE}${endpoint}`; + const activeNodeId = localStorage.getItem('sencho-active-node'); + const defaultOptions: RequestInit = { credentials: 'include', headers: { 'Content-Type': 'application/json', + ...(activeNodeId ? { 'x-node-id': activeNodeId } : {}), ...options.headers, }, };