michael/scrypted
1
0
Fork 0
mirror of https://github.com/koush/scrypted.git synced 2026-09-17 09:10:38 +01:00
Code Issues Actions Packages Projects Releases Wiki Activity

server: wip cluster

Browse Source
This commit is contained in:
Koushik Dutta
2024-11-10 20:24:03 -08:00
parent 68183775db
commit d93f05a228
8 changed files with 272 additions and 45 deletions
Download Patch File Download Diff File Expand all files Collapse all files

62
server/.vscode/launch.json vendored
View File

@@ -39,5 +39,67 @@
// "DEBUG": "*",
}
},
{
"autoAttachChildProcesses": false,
"type": "node",
"request": "launch",
"name": "Launch Cluster Server",
"skipFiles": [
"<node_internals>/**"
],
"preLaunchTask": "npm: build",
"program": "${workspaceFolder}/bin/scrypted-serve",
"runtimeArgs": [
"--trace-warnings",
"--nolazy",
],
"sourceMaps": true,
"resolveSourceMapLocations": [
"${workspaceFolder}/**",
"!**/node_modules/**"
],
"outFiles": [
"${workspaceFolder}/**/*.js"
],
"env": {
"SCRYPTED_CLUSTER_MODE": "server",
"SCRYPTED_CLUSTER_SERVER": "192.168.2.124",
"SCRYPTED_CLUSTER_SECRET": "swordfish",
"SCRYPTED_CAN_RESTART": "true",
"SCRYPTED_VOLUME": "/Users/koush/.scrypted-cluster/volume-server",
}
},
{
"autoAttachChildProcesses": false,
"type": "node",
"request": "launch",
"name": "Launch Cluster Client",
"skipFiles": [
"<node_internals>/**"
],
"preLaunchTask": "npm: build",
"program": "${workspaceFolder}/bin/scrypted-serve",
"runtimeArgs": [
"--trace-warnings",
"--nolazy",
],
"sourceMaps": true,
"resolveSourceMapLocations": [
"${workspaceFolder}/**",
"!**/node_modules/**"
],
"outFiles": [
"${workspaceFolder}/**/*.js"
],
"env": {
"SCRYPTED_CLUSTER_MODE": "client",
"SCRYPTED_CLUSTER_SERVER": "192.168.2.124",
"SCRYPTED_CLUSTER_SECRET": "swordfish",
"SCRYPTED_CAN_RESTART": "true",
"SCRYPTED_VOLUME": "/Users/koush/.scrypted-cluster/volume-client",
}
},
]
}

9
server/src/ip.ts
View File

@@ -1,10 +1,11 @@
import os from 'os';
import net from 'net';
export const loopbackList = new net.BlockList();
loopbackList.addSubnet('127.0.0.0', 8);
loopbackList.addAddress('::1', 'ipv6');
const unusableList = new net.BlockList();
// loopback
unusableList.addSubnet('127.0.0.0', 8);
unusableList.addAddress('::1', 'ipv6');
// link local
unusableList.addSubnet('169.254.0.0', 16);
@@ -56,7 +57,7 @@ function isUsableNetworkAddress(address: string) {
if (type === 'ipv6' && privateList.check(address, type))
return false;
return !unusableList.check(address, type);
return !unusableList.check(address, type) && !loopbackList.check(address, type);
}
catch (e) {
return false;

4
server/src/runtime.ts
View File

@@ -43,6 +43,7 @@ import { getNpmPackageInfo, PluginComponent } from './services/plugin';
import { ServiceControl } from './services/service-control';
import { UsersService } from './services/users';
import { getState, ScryptedStateManager, setState } from './state';
import { ClusterWorker } from './scrypted-cluster';
interface DeviceProxyPair {
handler: PluginDeviceProxyHandler;
@@ -59,7 +60,8 @@ interface HttpPluginData {
export class ScryptedRuntime extends PluginHttp<HttpPluginData> {
clusterId = crypto.randomBytes(3).toString('hex');
clusterSecret = crypto.randomBytes(16).toString('hex');
clusterSecret = process.env.SCRYPTED_CLUSTER_SECRET || crypto.randomBytes(16).toString('hex');
clusterWorkers = new Set<ClusterWorker>();
plugins: { [id: string]: PluginHost } = {};
pluginDevices: { [id: string]: PluginDevice } = {};
devices: { [id: string]: DeviceProxyPair } = {};

12
server/src/scrypted-cluster-main.ts Normal file
View File

@@ -0,0 +1,12 @@
import { install as installSourceMapSupport } from 'source-map-support';
import { startClusterClient } from './scrypted-cluster';
installSourceMapSupport({
environment: 'node',
});
async function start(mainFilename: string) {
startClusterClient(mainFilename);
}
export default start;

151
server/src/scrypted-cluster.ts Normal file
View File

@@ -0,0 +1,151 @@
import os from 'os';
import net from 'net';
import tls from 'tls';
import type { createSelfSignedCertificate } from './cert';
import { computeClusterObjectHash } from './cluster/cluster-hash';
import { ClusterObject } from './cluster/connect-rpc-object';
import { loopbackList } from './ip';
import { RpcPeer } from './rpc';
import { createRpcDuplexSerializer } from './rpc-serializer';
import type { ScryptedRuntime } from './runtime';
import { SCRYPTED_CLUSTER_WORKERS } from './server-settings';
import { sleep } from './sleep';
import { once } from 'events';
export interface ClusterWorkerProperties {
labels: string[];
}
export interface ClusterWorker extends ClusterWorkerProperties {
peer: RpcPeer;
}
export function getScryptedClusterMode(): ['server' | 'client', string, number] {
const mode = process.env.SCRYPTED_CLUSTER_MODE as 'server' | 'client';
if (!mode)
return;
if (!['server', 'client'].includes(mode))
throw new Error('SCRYPTED_CLUSTER_MODE must be set to either "server" or "client".');
const [server, sport] = process.env.SCRYPTED_CLUSTER_SERVER?.split(':') || [];
const port = parseInt(sport) || 10556;
if (!net.isIP(server)) {
if (server)
throw new Error('SCRYPTED_CLUSTER_SERVER is set but is not a valid IP address.');
if (process.env.SCRYPTED_CLUSTER_SECRET)
throw new Error('SCRYPTED_CLUSTER_SECRET is set but SCRYPTED_CLUSTER_SERVER is not set.');
return;
}
if (!process.env.SCRYPTED_CLUSTER_SECRET)
throw new Error('SCRYPTED_CLUSTER_SERVER is set but SCRYPTED_CLUSTER_SECRET is not set.');
return [mode, server, port];
}
export function startClusterClient(mainFilename: string) {
let labels = process.env.SCRYPTED_CLUSTER_LABELS?.split(',') || [];
labels.push(process.arch, process.platform, os.hostname());
labels = [...new Set(labels)];
const secret = process.env.SCRYPTED_CLUSTER_SECRET;
const clusterMode = getScryptedClusterMode();
const [, host, port] = clusterMode;
for (let i = 0; i < SCRYPTED_CLUSTER_WORKERS; i++) {
(async () => {
while (true) {
const backoff = sleep(10000);
try {
const client = tls.connect({
host,
port,
rejectUnauthorized: false,
});
const serializer = createRpcDuplexSerializer(client);
const peer = new RpcPeer('cluster-remote', 'cluster-host', (message, reject, serializationContext) => {
serializer.sendMessage(message, reject, serializationContext);
});
serializer.setupRpcPeer(peer);
client.on('data', data => serializer.onData(data));
client.on('error', e => {
peer.kill(e);
});
client.on('close', () => {
peer.kill('cluster server closed');
});
const connectForkWorker = await peer.getParam('connectForkWorker');
const auth: ClusterObject = {
address: client.localAddress,
port: client.localPort,
id: undefined,
proxyId: undefined,
sourceKey: undefined,
sha256: undefined,
};
auth.sha256 = computeClusterObjectHash(auth, secret);
const properties: ClusterWorkerProperties = {
labels,
};
await connectForkWorker(auth, properties);
console.warn('worker ready');
}
catch (e) {
}
await backoff;
}
})();
}
}
export function createClusterServer(runtime: ScryptedRuntime, certificate: ReturnType<typeof createSelfSignedCertificate>) {
const server = tls.createServer({
key: certificate.serviceKey,
cert: certificate.certificate,
}, (socket) => {
const serializer = createRpcDuplexSerializer(socket);
const peer = new RpcPeer('cluster-host', 'cluster-remote', (message, reject, serializationContext) => {
serializer.sendMessage(message, reject, serializationContext);
});
serializer.setupRpcPeer(peer);
socket.on('data', data => serializer.onData(data));
socket.on('error', e => {
peer.kill(e);
});
socket.on('close', () => {
peer.kill('cluster client closed');
});
peer.killed.then(() => {
socket.destroy();
});
peer.params['connectForkWorker'] = async (auth: ClusterObject, properties: ClusterWorkerProperties) => {
try {
const sha256 = computeClusterObjectHash(auth, runtime.clusterSecret);
if (sha256 !== auth.sha256)
throw new Error('cluster object hash mismatch');
// the remote address may be ipv6 prefixed so use a fuzzy match.
// eg ::ffff:192.168.2.124
if (auth.port !== socket.remotePort || !socket.remoteAddress.endsWith(auth.address))
throw new Error('cluster object address mismatch');
const worker: ClusterWorker = {
...properties,
peer,
};
runtime.clusterWorkers.add(worker);
peer.killed.then(() => {
runtime.clusterWorkers.delete(worker);
});
socket.on('close', () => {
runtime.clusterWorkers.delete(worker);
});
}
catch (e) {
peer.kill(e);
}
}
});
return server;
}

40
server/src/scrypted-main-exports.ts
View File

@@ -8,10 +8,10 @@ import v8 from 'v8';
import vm from 'vm';
import { PluginError } from './plugin/plugin-error';
import { getScryptedVolume } from './plugin/plugin-volume';
import { isNodePluginWorkerProcess } from './plugin/runtime/node-fork-worker';
import { RPCResultError, startPeriodicGarbageCollection } from './rpc';
import type { Runtime } from './scrypted-server-main';
import { isNodePluginWorkerProcess } from './plugin/runtime/node-fork-worker';
import { getScryptedClusterMode } from './scrypted-cluster';
function start(mainFilename: string, options?: {
onRuntimeCreated?: (runtime: Runtime) => Promise<void>,
@@ -28,8 +28,8 @@ function start(mainFilename: string, options?: {
globalThis.gc = vm.runInNewContext("gc");
}
if (!semver.gte(process.version, '16.0.0')) {
throw new Error('"node" version out of date. Please update node to v16 or higher.')
if (!semver.gte(process.version, '18.0.0')) {
throw new Error('"node" version out of date. Please update node to v18 or higher.')
}
// Node 17 changes the dns resolution order to return the record order.
@@ -53,20 +53,26 @@ function start(mainFilename: string, options?: {
const start = require('./scrypted-plugin-main').default;
return start(mainFilename);
}
// unhandled rejections are allowed if they are from a rpc/plugin call.
process.on('unhandledRejection', error => {
if (error?.constructor !== RPCResultError && error?.constructor !== PluginError) {
console.error('fatal error', error);
throw error;
}
console.warn('unhandled rejection of RPC Result', error);
});
dotenv.config({
path: path.join(getScryptedVolume(), '.env'),
});
const clusterMode = getScryptedClusterMode();
if (clusterMode?.[0] === 'client') {
const start = require('./scrypted-cluster-main').default;
return start(mainFilename);
}
else {
// unhandled rejections are allowed if they are from a rpc/plugin call.
process.on('unhandledRejection', error => {
if (error?.constructor !== RPCResultError && error?.constructor !== PluginError) {
console.error('fatal error', error);
throw error;
}
console.warn('unhandled rejection of RPC Result', error);
});
dotenv.config({
path: path.join(getScryptedVolume(), '.env'),
});
const start = require('./scrypted-server-main').default;
return start(mainFilename, options);
}

38
server/src/scrypted-server-main.ts
View File

@@ -11,38 +11,24 @@ import net from 'net';
import os from 'os';
import path from 'path';
import process from 'process';
import semver from 'semver';
import { install as installSourceMapSupport } from 'source-map-support';
import { createSelfSignedCertificate, CURRENT_SELF_SIGNED_CERTIFICATE_VERSION } from './cert';
import { Plugin, ScryptedUser, Settings } from './db-types';
import { getUsableNetworkAddresses } from './ip';
import Level from './level';
import { PluginError } from './plugin/plugin-error';
import { getScryptedVolume } from './plugin/plugin-volume';
import { RPCResultError } from './rpc';
import { ScryptedRuntime } from './runtime';
import { SCRYPTED_DEBUG_PORT, SCRYPTED_INSECURE_PORT, SCRYPTED_SECURE_PORT } from './server-settings';
import { getNpmPackageInfo } from './services/plugin';
import { setScryptedUserPassword, UsersService } from './services/users';
import { sleep } from './sleep';
import { ONE_DAY_MILLISECONDS, UserToken } from './usertoken';
import { createClusterServer, getScryptedClusterMode } from './scrypted-cluster';
export type Runtime = ScryptedRuntime;
if (!semver.gte(process.version, '18.0.0')) {
throw new Error('"node" version out of date. Please update node to v18 or higher.')
}
process.on('unhandledRejection', error => {
if (error?.constructor !== RPCResultError && error?.constructor !== PluginError) {
console.error('pending crash', error);
throw error;
}
console.warn('unhandled rejection of RPC Result', error);
});
async function listenServerPort(env: string, port: number, server: any) {
server.listen(port);
async function listenServerPort(env: string, port: number, server: http.Server | https.Server | net.Server, hostname?: string) {
server.listen(port, hostname);
try {
await once(server, 'listening');
}
@@ -474,7 +460,7 @@ async function start(mainFilename: string, options?: {
debugServer.on('connection', resolve);
});
waitDebug.catch(() => {});
waitDebug.catch(() => { });
workerInspectPort = Math.round(Math.random() * 10000) + 30000;
try {
@@ -737,6 +723,12 @@ async function start(mainFilename: string, options?: {
await listenServerPort('SCRYPTED_SECURE_PORT', SCRYPTED_SECURE_PORT, secure);
await listenServerPort('SCRYPTED_INSECURE_PORT', SCRYPTED_INSECURE_PORT, insecure);
const clusterMode = getScryptedClusterMode();
if (clusterMode?.[0] === 'server') {
const clusterServer = createClusterServer(scrypted, keyPair);
await listenServerPort('SCRYPTED_CLUSTER_SERVER', clusterMode[2], clusterServer);
}
console.log('#######################################################');
console.log(`Scrypted Volume : ${volumeDir}`);
console.log(`Scrypted Server (Local) : https://localhost:${SCRYPTED_SECURE_PORT}/`);
@@ -746,13 +738,13 @@ async function start(mainFilename: string, options?: {
console.log(`Version: : ${await scrypted.info.getVersion()}`);
console.log('#######################################################');
console.log('Scrypted insecure http service port:', SCRYPTED_INSECURE_PORT);
console.log('Ports can be changed with environment variables.')
console.log('https: $SCRYPTED_SECURE_PORT')
console.log('http : $SCRYPTED_INSECURE_PORT')
console.log('Certificate can be modified via tls.createSecureContext options in')
console.log('Ports can be changed with environment variables.');
console.log('https: $SCRYPTED_SECURE_PORT');
console.log('http : $SCRYPTED_INSECURE_PORT');
console.log('Certificate can be modified via tls.createSecureContext options in');
console.log('JSON file located at SCRYPTED_HTTPS_OPTIONS_FILE environment variable:');
console.log('export SCRYPTED_HTTPS_OPTIONS_FILE=/path/to/options.json');
console.log('https://nodejs.org/api/tls.html#tlscreatesecurecontextoptions')
console.log('https://nodejs.org/api/tls.html#tlscreatesecurecontextoptions');
console.log('#######################################################');
return scrypted;

1
server/src/server-settings.ts
View File

@@ -3,6 +3,7 @@ import { getUsableNetworkAddresses } from './ip';
export const SCRYPTED_INSECURE_PORT = parseInt(process.env.SCRYPTED_INSECURE_PORT) || 11080;
export const SCRYPTED_SECURE_PORT = parseInt(process.env.SCRYPTED_SECURE_PORT) || 10443;
export const SCRYPTED_DEBUG_PORT = parseInt(process.env.SCRYPTED_DEBUG_PORT) || 10081;
export const SCRYPTED_CLUSTER_WORKERS = parseInt(process.env.SCRYPTED_CLUSTER_WORKERS) || 32;
export function getIpAddress(): string {
return getUsableNetworkAddresses()[0];