606 lines
18 KiB
JavaScript
606 lines
18 KiB
JavaScript
/**
|
|
* Dependency Manager - Service dependency tracking with ordered restart chains
|
|
*
|
|
* Manages directed acyclic graph (DAG) of service dependencies. Services can
|
|
* declare which other services they depend on, and this manager provides:
|
|
* - Full dependency graph inspection
|
|
* - Topological ordering for safe restart chains
|
|
* - Circular dependency detection
|
|
* - Health-aware restart with per-service polling
|
|
*
|
|
* Dependencies are stored directly on service objects in services.json:
|
|
* { id, name, ..., dependsOn: ['service-id-1', 'service-id-2'] }
|
|
*
|
|
* @module dependency-manager
|
|
*/
|
|
|
|
const EventEmitter = require('events');
|
|
|
|
/** Maximum seconds to wait for a single container to become healthy after restart */
|
|
const HEALTH_CHECK_TIMEOUT_MS = 30_000;
|
|
|
|
/** Interval between container health polls */
|
|
const HEALTH_CHECK_INTERVAL_MS = 1_000;
|
|
|
|
/**
|
|
* @typedef {Object} ServiceNode
|
|
* @property {string} serviceId
|
|
* @property {string} name
|
|
* @property {string|null} containerId
|
|
*/
|
|
|
|
/**
|
|
* @typedef {Object} DependencyEdge
|
|
* @property {string} from - The service that depends
|
|
* @property {string} to - The service being depended upon
|
|
*/
|
|
|
|
/**
|
|
* @typedef {Object} DependencyGraph
|
|
* @property {ServiceNode[]} nodes
|
|
* @property {DependencyEdge[]} edges
|
|
*/
|
|
|
|
/**
|
|
* @typedef {Object} DependencyStatusEntry
|
|
* @property {string} serviceId
|
|
* @property {string} name
|
|
* @property {boolean} isUp
|
|
* @property {string} [error]
|
|
*/
|
|
|
|
/**
|
|
* DependencyManager — tracks service dependencies and orchestrates ordered restarts.
|
|
*
|
|
* Events emitted:
|
|
* - `dependency-restart-start` ({ serviceId, chain: string[] })
|
|
* - `dependency-restart-progress` ({ serviceId, currentServiceId, index, total })
|
|
* - `dependency-restart-complete` ({ serviceId, chain: string[], results: Array })
|
|
* - `dependency-restart-failed` ({ serviceId, failedServiceId, error, chain: string[] })
|
|
*
|
|
* @extends EventEmitter
|
|
*/
|
|
class DependencyManager extends EventEmitter {
|
|
/**
|
|
* @param {Object} ctx - Application context
|
|
* @param {Object} ctx.servicesStateManager - StateManager for services.json
|
|
* @param {Object} ctx.docker - Docker context ({ client: Dockerode })
|
|
* @param {Object} ctx.notification - NotificationManager instance
|
|
* @param {Object} ctx.log - Logger instance
|
|
*/
|
|
constructor(ctx) {
|
|
super();
|
|
/** @private */
|
|
this.ctx = ctx;
|
|
/** @private */
|
|
this._servicesStateManager = ctx.servicesStateManager;
|
|
/** @private */
|
|
this._docker = ctx.docker;
|
|
/** @private */
|
|
this._notification = ctx.notification;
|
|
/** @private */
|
|
this._log = ctx.log || console;
|
|
}
|
|
|
|
// ---------------------------------------------------------------------------
|
|
// Core helpers
|
|
// ---------------------------------------------------------------------------
|
|
|
|
/**
|
|
* Load all services from the state manager.
|
|
* @private
|
|
* @returns {Promise<Object[]>}
|
|
*/
|
|
async _loadServices() {
|
|
const data = await this._servicesStateManager.read();
|
|
return Array.isArray(data) ? data : (data.services || []);
|
|
}
|
|
|
|
/**
|
|
* Find a single service by ID.
|
|
* @private
|
|
* @param {string} serviceId
|
|
* @returns {Promise<Object|null>}
|
|
*/
|
|
async _findService(serviceId) {
|
|
const services = await this._loadServices();
|
|
return services.find(s => s.id === serviceId) || null;
|
|
}
|
|
|
|
// ---------------------------------------------------------------------------
|
|
// Graph queries
|
|
// ---------------------------------------------------------------------------
|
|
|
|
/**
|
|
* Return the full dependency graph for visualisation.
|
|
*
|
|
* @returns {Promise<DependencyGraph>}
|
|
*/
|
|
async getDependencyGraph() {
|
|
const services = await this._loadServices();
|
|
|
|
const nodes = services.map(s => ({
|
|
serviceId: s.id,
|
|
name: s.name,
|
|
containerId: s.containerId || null,
|
|
}));
|
|
|
|
const edges = [];
|
|
for (const service of services) {
|
|
const deps = service.dependsOn || [];
|
|
for (const depId of deps) {
|
|
edges.push({ from: service.id, to: depId });
|
|
}
|
|
}
|
|
|
|
return { nodes, edges };
|
|
}
|
|
|
|
/**
|
|
* Return the services that depend on the given service (reverse deps).
|
|
*
|
|
* @param {string} serviceId
|
|
* @returns {Promise<Object[]>} Services whose `dependsOn` includes `serviceId`.
|
|
*/
|
|
async getDependents(serviceId) {
|
|
const services = await this._loadServices();
|
|
return services.filter(s => (s.dependsOn || []).includes(serviceId));
|
|
}
|
|
|
|
/**
|
|
* Return the direct dependencies for a service.
|
|
*
|
|
* @param {string} serviceId
|
|
* @returns {Promise<Object[]>} Services that `serviceId` depends on.
|
|
*/
|
|
async getDependencies(serviceId) {
|
|
const services = await this._loadServices();
|
|
const service = services.find(s => s.id === serviceId);
|
|
if (!service) return [];
|
|
const depIds = service.dependsOn || [];
|
|
return services.filter(s => depIds.includes(s.id));
|
|
}
|
|
|
|
// ---------------------------------------------------------------------------
|
|
// Topological sort
|
|
// ---------------------------------------------------------------------------
|
|
|
|
/**
|
|
* Build an adjacency list for the current dependency graph.
|
|
* Edge direction: service → its dependencies (i.e. what it depends on).
|
|
*
|
|
* @private
|
|
* @param {Object[]} services
|
|
* @returns {Map<string, string[]>}
|
|
*/
|
|
_buildAdjacencyList(services) {
|
|
const adj = new Map();
|
|
for (const service of services) {
|
|
adj.set(service.id, (service.dependsOn || []).slice());
|
|
}
|
|
return adj;
|
|
}
|
|
|
|
/**
|
|
* DFS-based topological sort with cycle detection (white/gray/black coloring).
|
|
*
|
|
* Returns services in restart order: dependencies first, dependents last.
|
|
* The target service is included at the end.
|
|
*
|
|
* @private
|
|
* @param {string} serviceId - Target service (will be last in the result).
|
|
* @param {Object[]} services - All services.
|
|
* @param {Map<string, string[]>} adj - Adjacency list (service → deps).
|
|
* @returns {string[]} Ordered service IDs for restart.
|
|
* @throws {Error} If a circular dependency is detected.
|
|
*/
|
|
_topologicalSort(serviceId, services, adj) {
|
|
// Collect only the reachable sub-graph from serviceId
|
|
const visited = new Set();
|
|
const reachable = new Set();
|
|
|
|
const collectReachable = (id) => {
|
|
if (reachable.has(id)) return;
|
|
reachable.add(id);
|
|
for (const dep of (adj.get(id) || [])) {
|
|
collectReachable(dep);
|
|
}
|
|
};
|
|
collectReachable(serviceId);
|
|
|
|
// DFS topological sort on the reachable sub-graph
|
|
const WHITE = 0, GRAY = 1, BLACK = 2;
|
|
const color = new Map();
|
|
for (const id of reachable) color.set(id, WHITE);
|
|
|
|
const result = [];
|
|
|
|
const dfs = (id) => {
|
|
if (color.get(id) === BLACK) return;
|
|
if (color.get(id) === GRAY) {
|
|
throw new Error(`Circular dependency detected involving service "${id}"`);
|
|
}
|
|
color.set(id, GRAY);
|
|
for (const dep of (adj.get(id) || [])) {
|
|
dfs(dep);
|
|
}
|
|
color.set(id, BLACK);
|
|
result.push(id);
|
|
};
|
|
|
|
// Visit the target last so it ends up at the end of the result
|
|
// Actually, we want deps *first* then the target.
|
|
// The DFS naturally puts deps before dependents, so starting from
|
|
// serviceId will place it last (which is correct for restart order).
|
|
dfs(serviceId);
|
|
|
|
return result;
|
|
}
|
|
|
|
/**
|
|
* Get the topologically ordered restart chain for a service.
|
|
*
|
|
* The returned array lists all services that must be restarted,
|
|
* starting with leaf dependencies and ending with the target service.
|
|
*
|
|
* @param {string} serviceId - The service to build the chain for.
|
|
* @returns {Promise<string[]>} Ordered service IDs.
|
|
* @throws {Error} If `serviceId` doesn't exist or a circular dependency is found.
|
|
*/
|
|
async getOrderedRestartChain(serviceId) {
|
|
const services = await this._loadServices();
|
|
const service = services.find(s => s.id === serviceId);
|
|
if (!service) {
|
|
throw new Error(`Service "${serviceId}" not found`);
|
|
}
|
|
|
|
const adj = this._buildAdjacencyList(services);
|
|
return this._topologicalSort(serviceId, services, adj);
|
|
}
|
|
|
|
// ---------------------------------------------------------------------------
|
|
// Validation
|
|
// ---------------------------------------------------------------------------
|
|
|
|
/**
|
|
* Validate a proposed set of dependencies for a service.
|
|
*
|
|
* Checks:
|
|
* - All referenced service IDs exist.
|
|
* - Adding these dependencies would not create a circular dependency.
|
|
* - A service cannot depend on itself.
|
|
*
|
|
* @param {string} serviceId - The service to set dependencies on.
|
|
* @param {string[]} dependsOn - Proposed dependency IDs.
|
|
* @returns {Promise<{ valid: boolean, errors: string[] }>}
|
|
*/
|
|
async validateDependencies(serviceId, dependsOn) {
|
|
const errors = [];
|
|
|
|
if (!Array.isArray(dependsOn)) {
|
|
return { valid: false, errors: ['dependsOn must be an array'] };
|
|
}
|
|
|
|
const services = await this._loadServices();
|
|
const allIds = new Set(services.map(s => s.id));
|
|
|
|
// Service must exist
|
|
if (!allIds.has(serviceId)) {
|
|
return { valid: false, errors: [`Service "${serviceId}" not found`] };
|
|
}
|
|
|
|
// Self-dependency
|
|
if (dependsOn.includes(serviceId)) {
|
|
errors.push(`Service "${serviceId}" cannot depend on itself`);
|
|
}
|
|
|
|
// Existence check
|
|
for (const depId of dependsOn) {
|
|
if (!allIds.has(depId)) {
|
|
errors.push(`Dependency service "${depId}" does not exist`);
|
|
}
|
|
}
|
|
|
|
if (errors.length > 0) {
|
|
return { valid: false, errors };
|
|
}
|
|
|
|
// Circular dependency check: temporarily set the proposed dependsOn
|
|
// and attempt a topological sort.
|
|
const tempServices = services.map(s => {
|
|
if (s.id === serviceId) {
|
|
return { ...s, dependsOn: dependsOn.slice() };
|
|
}
|
|
return { ...s };
|
|
});
|
|
|
|
const adj = this._buildAdjacencyList(tempServices);
|
|
|
|
// Check every node for cycles with the new edges
|
|
try {
|
|
const WHITE = 0, GRAY = 1, BLACK = 2;
|
|
const color = new Map();
|
|
for (const s of tempServices) color.set(s.id, WHITE);
|
|
|
|
const dfs = (id) => {
|
|
if (color.get(id) === BLACK) return;
|
|
if (color.get(id) === GRAY) {
|
|
throw new Error(`Circular dependency detected involving service "${id}"`);
|
|
}
|
|
color.set(id, GRAY);
|
|
for (const dep of (adj.get(id) || [])) {
|
|
dfs(dep);
|
|
}
|
|
color.set(id, BLACK);
|
|
};
|
|
|
|
for (const s of tempServices) {
|
|
if (color.get(s.id) === WHITE) {
|
|
dfs(s.id);
|
|
}
|
|
}
|
|
} catch (err) {
|
|
errors.push(err.message);
|
|
}
|
|
|
|
return { valid: errors.length === 0, errors };
|
|
}
|
|
|
|
// ---------------------------------------------------------------------------
|
|
// Health status
|
|
// ---------------------------------------------------------------------------
|
|
|
|
/**
|
|
* Get the current container status for a service and all its transitive dependencies.
|
|
*
|
|
* @param {string} serviceId
|
|
* @returns {Promise<DependencyStatusEntry[]>}
|
|
* @throws {Error} If `serviceId` doesn't exist.
|
|
*/
|
|
async getDependencyStatus(serviceId) {
|
|
const services = await this._loadServices();
|
|
const service = services.find(s => s.id === serviceId);
|
|
if (!service) {
|
|
throw new Error(`Service "${serviceId}" not found`);
|
|
}
|
|
|
|
// Collect all transitive dependencies via BFS
|
|
const serviceMap = new Map(services.map(s => [s.id, s]));
|
|
const visited = new Set();
|
|
const queue = [serviceId];
|
|
const allRelated = [];
|
|
|
|
while (queue.length > 0) {
|
|
const currentId = queue.shift();
|
|
if (visited.has(currentId)) continue;
|
|
visited.add(currentId);
|
|
|
|
const svc = serviceMap.get(currentId);
|
|
if (!svc) continue;
|
|
|
|
allRelated.push(svc);
|
|
|
|
for (const depId of (svc.dependsOn || [])) {
|
|
if (!visited.has(depId)) {
|
|
queue.push(depId);
|
|
}
|
|
}
|
|
}
|
|
|
|
// Query container status for each
|
|
const results = [];
|
|
for (const svc of allRelated) {
|
|
const entry = {
|
|
serviceId: svc.id,
|
|
name: svc.name,
|
|
isUp: false,
|
|
};
|
|
|
|
if (!svc.containerId) {
|
|
entry.error = 'No container associated with this service';
|
|
results.push(entry);
|
|
continue;
|
|
}
|
|
|
|
try {
|
|
const container = this._docker.client.getContainer(svc.containerId);
|
|
const info = await container.inspect();
|
|
entry.isUp = info.State?.Running === true;
|
|
} catch (err) {
|
|
entry.error = err.message || 'Unable to inspect container';
|
|
}
|
|
|
|
results.push(entry);
|
|
}
|
|
|
|
return results;
|
|
}
|
|
|
|
// ---------------------------------------------------------------------------
|
|
// Restart with dependencies
|
|
// ---------------------------------------------------------------------------
|
|
|
|
/**
|
|
* Wait for a container to report as running after a restart.
|
|
*
|
|
* @private
|
|
* @param {string} containerId
|
|
* @param {number} [timeoutMs=30000]
|
|
* @returns {Promise<boolean>} `true` if healthy, `false` if timed out.
|
|
*/
|
|
async _waitForContainerHealthy(containerId, timeoutMs = HEALTH_CHECK_TIMEOUT_MS) {
|
|
const start = Date.now();
|
|
while (Date.now() - start < timeoutMs) {
|
|
try {
|
|
const container = this._docker.client.getContainer(containerId);
|
|
const info = await container.inspect();
|
|
if (info.State?.Running === true) {
|
|
return true;
|
|
}
|
|
} catch {
|
|
// Container might not be inspectable during restart — keep polling
|
|
}
|
|
await new Promise(r => setTimeout(r, HEALTH_CHECK_INTERVAL_MS));
|
|
}
|
|
return false;
|
|
}
|
|
|
|
/**
|
|
* Restart a service and all its dependencies in topological order.
|
|
*
|
|
* Emits progress events and sends a notification on completion/failure.
|
|
* This method is designed to be called from the route handler and
|
|
* **does not throw** — errors are reported via events and notifications.
|
|
*
|
|
* @param {string} serviceId - Target service to restart (with deps).
|
|
* @returns {Promise<{ success: boolean, chain: string[], results: Array }>}
|
|
*/
|
|
async restartWithDependencies(serviceId) {
|
|
const service = await this._findService(serviceId);
|
|
if (!service) {
|
|
const err = new Error(`Service "${serviceId}" not found`);
|
|
this.emit('dependency-restart-failed', {
|
|
serviceId,
|
|
failedServiceId: serviceId,
|
|
error: err.message,
|
|
chain: [],
|
|
});
|
|
throw err;
|
|
}
|
|
|
|
let chain;
|
|
try {
|
|
chain = await this.getOrderedRestartChain(serviceId);
|
|
} catch (err) {
|
|
this.emit('dependency-restart-failed', {
|
|
serviceId,
|
|
failedServiceId: serviceId,
|
|
error: err.message,
|
|
chain: [],
|
|
});
|
|
throw err;
|
|
}
|
|
|
|
const services = await this._loadServices();
|
|
const serviceMap = new Map(services.map(s => [s.id, s]));
|
|
|
|
this._log.info('dependency', 'Starting dependency restart chain', {
|
|
serviceId,
|
|
chain,
|
|
});
|
|
|
|
this.emit('dependency-restart-start', { serviceId, chain });
|
|
|
|
const results = [];
|
|
const total = chain.length;
|
|
|
|
for (let i = 0; i < total; i++) {
|
|
const currentId = chain[i];
|
|
const svc = serviceMap.get(currentId);
|
|
|
|
this.emit('dependency-restart-progress', {
|
|
serviceId,
|
|
currentServiceId: currentId,
|
|
index: i,
|
|
total,
|
|
});
|
|
|
|
if (!svc || !svc.containerId) {
|
|
const msg = !svc
|
|
? `Service "${currentId}" not found in state`
|
|
: `Service "${currentId}" has no container — skipping restart`;
|
|
this._log.warn('dependency', msg);
|
|
results.push({ serviceId: currentId, restarted: false, skipped: true, reason: msg });
|
|
continue;
|
|
}
|
|
|
|
try {
|
|
const container = this._docker.client.getContainer(svc.containerId);
|
|
this._log.info('dependency', `Restarting container for service "${currentId}"`, {
|
|
containerId: svc.containerId,
|
|
});
|
|
await container.restart();
|
|
|
|
// Wait for it to come back up
|
|
const healthy = await this._waitForContainerHealthy(svc.containerId);
|
|
if (!healthy) {
|
|
const msg = `Container for service "${currentId}" did not become healthy within ${HEALTH_CHECK_TIMEOUT_MS / 1000}s`;
|
|
this._log.warn('dependency', msg);
|
|
results.push({ serviceId: currentId, restarted: true, healthy: false, error: msg });
|
|
|
|
// Abort chain — dependency didn't come back
|
|
this.emit('dependency-restart-failed', {
|
|
serviceId,
|
|
failedServiceId: currentId,
|
|
error: msg,
|
|
chain,
|
|
});
|
|
await this._notifyRestartResult(serviceId, false, chain, results, currentId);
|
|
return { success: false, chain, results };
|
|
}
|
|
|
|
this._log.info('dependency', `Service "${currentId}" is healthy after restart`);
|
|
results.push({ serviceId: currentId, restarted: true, healthy: true });
|
|
} catch (err) {
|
|
const msg = err.message || 'Unknown error during restart';
|
|
this._log.error('dependency', `Failed to restart service "${currentId}"`, {
|
|
error: msg,
|
|
});
|
|
results.push({ serviceId: currentId, restarted: false, error: msg });
|
|
|
|
this.emit('dependency-restart-failed', {
|
|
serviceId,
|
|
failedServiceId: currentId,
|
|
error: msg,
|
|
chain,
|
|
});
|
|
await this._notifyRestartResult(serviceId, false, chain, results, currentId);
|
|
return { success: false, chain, results };
|
|
}
|
|
}
|
|
|
|
this.emit('dependency-restart-complete', { serviceId, chain, results });
|
|
await this._notifyRestartResult(serviceId, true, chain, results);
|
|
return { success: true, chain, results };
|
|
}
|
|
|
|
/**
|
|
* Send a notification about the restart result.
|
|
*
|
|
* @private
|
|
* @param {string} serviceId
|
|
* @param {boolean} success
|
|
* @param {string[]} chain
|
|
* @param {Array} results
|
|
* @param {string} [failedServiceId]
|
|
*/
|
|
async _notifyRestartResult(serviceId, success, chain, results, failedServiceId) {
|
|
if (!this._notification) return;
|
|
|
|
try {
|
|
if (success) {
|
|
await this._notification.send('dependency-restart-complete', {
|
|
text: `✅ Dependency restart chain completed for "${serviceId}". Restarted: ${chain.join(' → ')}`,
|
|
serviceId,
|
|
chain,
|
|
results,
|
|
});
|
|
} else {
|
|
await this._notification.send('dependency-restart-failed', {
|
|
text: `❌ Dependency restart chain failed for "${serviceId}" at "${failedServiceId}". Chain: ${chain.join(' → ')}`,
|
|
serviceId,
|
|
failedServiceId,
|
|
chain,
|
|
results,
|
|
});
|
|
}
|
|
} catch (err) {
|
|
this._log.error('dependency', 'Failed to send restart notification', {
|
|
error: err.message,
|
|
});
|
|
}
|
|
}
|
|
}
|
|
|
|
module.exports = DependencyManager;
|