feat(opsserver): introduce OpsServer (TypedRequest API) and new lightweight web UI; replace legacy Angular UI and add typed interfaces
This commit is contained in:
219
ts/opsserver/handlers/logs.handler.ts
Normal file
219
ts/opsserver/handlers/logs.handler.ts
Normal file
@@ -0,0 +1,219 @@
|
||||
import * as plugins from '../../plugins.ts';
|
||||
import { logger } from '../../logging.ts';
|
||||
import type { OpsServer } from '../classes.opsserver.ts';
|
||||
import * as interfaces from '../../../ts_interfaces/index.ts';
|
||||
import { requireValidIdentity } from '../helpers/guards.ts';
|
||||
|
||||
export class LogsHandler {
|
||||
public typedrouter = new plugins.typedrequest.TypedRouter();
|
||||
|
||||
constructor(private opsServerRef: OpsServer) {
|
||||
this.opsServerRef.typedrouter.addTypedRouter(this.typedrouter);
|
||||
this.registerHandlers();
|
||||
}
|
||||
|
||||
private registerHandlers(): void {
|
||||
// Service log stream
|
||||
this.typedrouter.addTypedHandler(
|
||||
new plugins.typedrequest.TypedHandler<interfaces.requests.IReq_GetServiceLogStream>(
|
||||
'getServiceLogStream',
|
||||
async (dataArg) => {
|
||||
await requireValidIdentity(this.opsServerRef.adminHandler, dataArg);
|
||||
|
||||
const service = this.opsServerRef.oneboxRef.database.getServiceByName(dataArg.serviceName);
|
||||
if (!service) {
|
||||
throw new plugins.typedrequest.TypedResponseError('Service not found');
|
||||
}
|
||||
|
||||
const virtualStream = new plugins.typedrequest.VirtualStream<Uint8Array>();
|
||||
const encoder = new TextEncoder();
|
||||
|
||||
// Get container and start streaming in background
|
||||
(async () => {
|
||||
try {
|
||||
let container = await this.opsServerRef.oneboxRef.docker.getContainerById(service.containerID!);
|
||||
if (!container) {
|
||||
// Try finding by service label
|
||||
const containers = await this.opsServerRef.oneboxRef.docker.listAllContainers();
|
||||
const serviceContainer = containers.find((c: any) => {
|
||||
const labels = c.Labels || {};
|
||||
return labels['com.docker.swarm.service.id'] === service.containerID;
|
||||
});
|
||||
if (serviceContainer) {
|
||||
container = await this.opsServerRef.oneboxRef.docker.getContainerById(serviceContainer.Id);
|
||||
}
|
||||
}
|
||||
|
||||
if (!container) {
|
||||
virtualStream.sendData(encoder.encode(JSON.stringify({ error: 'Container not found' })));
|
||||
return;
|
||||
}
|
||||
|
||||
const logStream = await container.streamLogs({
|
||||
stdout: true,
|
||||
stderr: true,
|
||||
timestamps: true,
|
||||
tail: 100,
|
||||
});
|
||||
|
||||
let buffer = new Uint8Array(0);
|
||||
|
||||
logStream.on('data', (chunk: Uint8Array) => {
|
||||
// Append to buffer
|
||||
const newBuffer = new Uint8Array(buffer.length + chunk.length);
|
||||
newBuffer.set(buffer);
|
||||
newBuffer.set(chunk, buffer.length);
|
||||
buffer = newBuffer;
|
||||
|
||||
// Process Docker multiplexed frames
|
||||
while (buffer.length >= 8) {
|
||||
const frameSize = (buffer[4] << 24) | (buffer[5] << 16) | (buffer[6] << 8) | buffer[7];
|
||||
if (buffer.length < 8 + frameSize) break;
|
||||
|
||||
const frameData = buffer.slice(8, 8 + frameSize);
|
||||
try {
|
||||
virtualStream.sendData(frameData);
|
||||
} catch {
|
||||
logStream.destroy();
|
||||
return;
|
||||
}
|
||||
buffer = buffer.slice(8 + frameSize);
|
||||
}
|
||||
});
|
||||
|
||||
logStream.on('error', (error: Error) => {
|
||||
logger.error(`Log stream error for ${dataArg.serviceName}: ${error.message}`);
|
||||
});
|
||||
} catch (error) {
|
||||
logger.error(`Failed to start log stream: ${error}`);
|
||||
}
|
||||
})();
|
||||
|
||||
return { logStream: virtualStream as any };
|
||||
},
|
||||
),
|
||||
);
|
||||
|
||||
// Platform service log stream
|
||||
this.typedrouter.addTypedHandler(
|
||||
new plugins.typedrequest.TypedHandler<interfaces.requests.IReq_GetPlatformServiceLogStream>(
|
||||
'getPlatformServiceLogStream',
|
||||
async (dataArg) => {
|
||||
await requireValidIdentity(this.opsServerRef.adminHandler, dataArg);
|
||||
|
||||
const platformService = this.opsServerRef.oneboxRef.database.getPlatformServiceByType(
|
||||
dataArg.serviceType,
|
||||
);
|
||||
if (!platformService || !platformService.containerId) {
|
||||
throw new plugins.typedrequest.TypedResponseError('Platform service has no container');
|
||||
}
|
||||
|
||||
const virtualStream = new plugins.typedrequest.VirtualStream<Uint8Array>();
|
||||
|
||||
(async () => {
|
||||
try {
|
||||
const container = await this.opsServerRef.oneboxRef.docker.getContainerById(
|
||||
platformService.containerId!,
|
||||
);
|
||||
if (!container) return;
|
||||
|
||||
const logStream = await container.streamLogs({
|
||||
stdout: true,
|
||||
stderr: true,
|
||||
timestamps: true,
|
||||
tail: 100,
|
||||
});
|
||||
|
||||
let buffer = new Uint8Array(0);
|
||||
|
||||
logStream.on('data', (chunk: Uint8Array) => {
|
||||
const newBuffer = new Uint8Array(buffer.length + chunk.length);
|
||||
newBuffer.set(buffer);
|
||||
newBuffer.set(chunk, buffer.length);
|
||||
buffer = newBuffer;
|
||||
|
||||
while (buffer.length >= 8) {
|
||||
const frameSize = (buffer[4] << 24) | (buffer[5] << 16) | (buffer[6] << 8) | buffer[7];
|
||||
if (buffer.length < 8 + frameSize) break;
|
||||
const frameData = buffer.slice(8, 8 + frameSize);
|
||||
try {
|
||||
virtualStream.sendData(frameData);
|
||||
} catch {
|
||||
logStream.destroy();
|
||||
return;
|
||||
}
|
||||
buffer = buffer.slice(8 + frameSize);
|
||||
}
|
||||
});
|
||||
} catch (error) {
|
||||
logger.error(`Failed to start platform log stream: ${error}`);
|
||||
}
|
||||
})();
|
||||
|
||||
return { logStream: virtualStream as any };
|
||||
},
|
||||
),
|
||||
);
|
||||
|
||||
// Network log stream
|
||||
this.typedrouter.addTypedHandler(
|
||||
new plugins.typedrequest.TypedHandler<interfaces.requests.IReq_GetNetworkLogStream>(
|
||||
'getNetworkLogStream',
|
||||
async (dataArg) => {
|
||||
await requireValidIdentity(this.opsServerRef.adminHandler, dataArg);
|
||||
|
||||
const virtualStream = new plugins.typedrequest.VirtualStream<Uint8Array>();
|
||||
const encoder = new TextEncoder();
|
||||
const clientId = crypto.randomUUID();
|
||||
|
||||
// Create a mock WebSocket-like object for the CaddyLogReceiver
|
||||
const mockSocket = {
|
||||
readyState: 1, // WebSocket.OPEN
|
||||
send: (data: string) => {
|
||||
try {
|
||||
virtualStream.sendData(encoder.encode(data));
|
||||
} catch {
|
||||
this.opsServerRef.oneboxRef.caddyLogReceiver.removeClient(clientId);
|
||||
}
|
||||
},
|
||||
};
|
||||
|
||||
const filter = dataArg.filter || {};
|
||||
this.opsServerRef.oneboxRef.caddyLogReceiver.addClient(
|
||||
clientId,
|
||||
mockSocket as any,
|
||||
filter,
|
||||
);
|
||||
|
||||
return { logStream: virtualStream as any };
|
||||
},
|
||||
),
|
||||
);
|
||||
|
||||
// Event stream (general updates)
|
||||
this.typedrouter.addTypedHandler(
|
||||
new plugins.typedrequest.TypedHandler<interfaces.requests.IReq_GetEventStream>(
|
||||
'getEventStream',
|
||||
async (dataArg) => {
|
||||
await requireValidIdentity(this.opsServerRef.adminHandler, dataArg);
|
||||
|
||||
const virtualStream = new plugins.typedrequest.VirtualStream<Uint8Array>();
|
||||
const encoder = new TextEncoder();
|
||||
|
||||
// Send initial connection message
|
||||
virtualStream.sendData(
|
||||
encoder.encode(
|
||||
JSON.stringify({
|
||||
type: 'connected',
|
||||
message: 'Connected to Onebox event stream',
|
||||
timestamp: Date.now(),
|
||||
}),
|
||||
),
|
||||
);
|
||||
|
||||
return { eventStream: virtualStream as any };
|
||||
},
|
||||
),
|
||||
);
|
||||
}
|
||||
}
|
||||
Reference in New Issue
Block a user