diff --git a/CHANGELOG.md b/CHANGELOG.md index dc84398d..0264def0 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -17,6 +17,7 @@ - Fixed assertion error on closing or invalidating a link with messages forwarded from a virtual link. - Fixed host hanging if stopped while an instance is in the process of creating a save during startup. - Fixed `instance.auto_start` option. +- Fixed host log spammed with errors when plugins broadcast messages to instances with the plugin disabled. ## Version 2.0.0-alpha.14 diff --git a/packages/controller/index.ts b/packages/controller/index.ts index 5218034f..caebb912 100644 --- a/packages/controller/index.ts +++ b/packages/controller/index.ts @@ -1,7 +1,9 @@ import { bootstrap } from "./controller"; export { default as Controller } from "./src/Controller"; export { default as ControllerUser } from "./src/ControllerUser"; +export { default as ControllerRouter } from "./src/ControllerRouter"; export { default as ControlConnection } from "./src/ControlConnection"; +export { default as HostConnection } from "./src/HostConnection"; export { default as InstanceInfo } from "./src/InstanceInfo"; export { default as BaseControllerPlugin } from "./src/BaseControllerPlugin"; export { default as UserManager } from "./src/UserManager"; diff --git a/packages/controller/src/BaseConnection.ts b/packages/controller/src/BaseConnection.ts index 493ebbde..3ebac51e 100644 --- a/packages/controller/src/BaseConnection.ts +++ b/packages/controller/src/BaseConnection.ts @@ -6,7 +6,6 @@ import type WsServerConnector from "./WsServerConnector"; import * as lib from "@clusterio/lib"; const { logger } = lib; -import ControllerRouter from "./ControllerRouter"; import * as routes from "./routes"; @@ -24,7 +23,7 @@ export default class BaseConnection extends lib.Link { public _controller: Controller ) { super(connector); - this.router = new ControllerRouter(this._controller); + this.router = _controller.router; for (let [Request, handler] of this._controller._registeredRequests) { this.handle(Request, handler); } for (let [Request, handler] of this._controller._fallbackedRequests) { this.fallbackRequest(Request, handler); } for (let [Event, handler] of this._controller._registeredEvents) { this.handle(Event, handler); } diff --git a/packages/controller/src/Controller.ts b/packages/controller/src/Controller.ts index 8930f854..3e81c0c1 100644 --- a/packages/controller/src/Controller.ts +++ b/packages/controller/src/Controller.ts @@ -26,6 +26,7 @@ import WsServer from "./WsServer"; import HostConnection from "./HostConnection"; import HostInfo from "./HostInfo"; import BaseControllerPlugin from "./BaseControllerPlugin"; +import ControllerRouter from "./ControllerRouter"; const endpointDurationSummary = new Summary( "clusterio_controller_http_endpoint_duration_seconds", @@ -74,6 +75,7 @@ export default class Controller { /** WebSocket server */ wsServer: WsServer; + router = new ControllerRouter(this); trustedProxies: BlockList; debugEvents: events.EventEmitter = new events.EventEmitter(); private _events: events.EventEmitter = new events.EventEmitter(); @@ -1374,7 +1376,11 @@ export default class Controller { } } else if (dst.id === lib.Address.instance || dst.id === lib.Address.host) { + const plugin = event.constructor.plugin; for (let hostConnection of this.wsServer.hostConnections.values()) { + if (plugin && !hostConnection.plugins.has(plugin)) { + continue; + } hostConnection.sendEvent(event, dst); } diff --git a/packages/controller/src/ControllerRouter.ts b/packages/controller/src/ControllerRouter.ts index 84420595..4679312d 100644 --- a/packages/controller/src/ControllerRouter.ts +++ b/packages/controller/src/ControllerRouter.ts @@ -7,8 +7,6 @@ import ControlConnection from "./ControlConnection"; const { logger } = lib; export default class ControllerRouter { - pendingRequests: Map = new Map(); - constructor( public controller: Controller ) {} @@ -16,12 +14,14 @@ export default class ControllerRouter { /** * @param origin - Source link of the message. * @param message - Link the message originated from. + * @param entry - message entry if message is a request or an event. * @param hasFallback - true if fallback handling is available. * @returns true if the message was handled */ forwardMessage( origin: ControlConnection | HostConnection, message: lib.MessageRoutable, + entry: lib.RequestEntry | lib.EventEntry | undefined, hasFallback: boolean ): boolean { if (!["request", "response", "responseError", "event"].includes(message.type)) { @@ -32,7 +32,7 @@ export default class ControllerRouter { let nextHop: ControlConnection | HostConnection | undefined; let msg: string = ""; if (dst.type === lib.Address.broadcast) { - this.broadcastMessage(origin, message); + this.broadcastMessage(origin, message, entry as lib.EventEntry); return true; } else if (dst.type === lib.Address.host) { nextHop = this.controller.wsServer.hostConnections.get(dst.id); @@ -96,13 +96,17 @@ export default class ControllerRouter { broadcastMessage( origin: lib.Link, message: lib.MessageRoutable, + entry: lib.EventEntry, ) { let dst = message.dst; if (message.type !== "event") { this.warnUnrouted(message, `Unexpected broadcast of ${message.type}`); - } else if (dst.id === lib.Address.host || dst.id === lib.Address.instance) { + return; + } + const plugin = entry.Event.plugin; + if (dst.id === lib.Address.host || dst.id === lib.Address.instance) { for (let hostConnection of this.controller.wsServer.hostConnections.values()) { - if (hostConnection !== origin) { + if (hostConnection !== origin && (!plugin || hostConnection.plugins.has(plugin))) { hostConnection.connector.send(message); } } @@ -126,5 +130,3 @@ export default class ControllerRouter { logger.warn(msg ? `${baseMsg}: ${msg}.` : `${baseMsg}.`); } } - -module.exports = ControllerRouter; diff --git a/packages/host/index.ts b/packages/host/index.ts index 2b01bc21..f39ce536 100644 --- a/packages/host/index.ts +++ b/packages/host/index.ts @@ -1,6 +1,9 @@ import { bootstrap } from "./host"; export { default as BaseHostPlugin } from "./src/BaseHostPlugin"; export { default as BaseInstancePlugin } from "./src/BaseInstancePlugin"; +export { default as Host, HostRouter } from "./src/Host"; +export { default as Instance } from "./src/Instance"; +export { default as InstanceConnection } from "./src/InstanceConnection"; if (module === require.main) { bootstrap(); diff --git a/packages/host/src/Host.ts b/packages/host/src/Host.ts index 06a23809..b1b9dd41 100644 --- a/packages/host/src/Host.ts +++ b/packages/host/src/Host.ts @@ -82,7 +82,7 @@ const instanceStartingMessages = new Set([ lib.InstanceExportDataRequest.name, ]); -class HostRouter { +export class HostRouter { constructor( public host: Host ) { } @@ -93,16 +93,22 @@ class HostRouter { * @param origin - * Link the message originated from. * @param message - Message to process. + * @param entry - message entry if message is a request or an event. * @param hasFallback - is fallback available * @returns true if the message was handled, false if fallback * is requested */ - forwardMessage(origin: lib.Link, message: lib.MessageRoutable, hasFallback: boolean) { + forwardMessage( + origin: lib.Link, + message: lib.MessageRoutable, + entry: lib.RequestEntry | lib.EventEntry | undefined, + hasFallback: boolean, + ) { let dst = message.dst; let nextHop: lib.Link | undefined; let msg: string | undefined; if (dst.type === lib.Address.broadcast) { - this.broadcastMessage(origin, message); + this.broadcastMessage(origin, message, entry as lib.EventEntry); return true; } else if ( dst.type === lib.Address.controller @@ -153,27 +159,36 @@ class HostRouter { return true; } - broadcastMessage(origin: lib.Link, message: lib.MessageRoutable) { + broadcastMessage( + origin: lib.Link, + message: lib.MessageRoutable, + entry: lib.EventEntry, + ) { let dst = message.dst; if (message.type !== "event") { this.warnUnrouted(message, `Unexpected broadcast of ${message.type}`); - } else if (dst.id === lib.Address.host || dst.id === lib.Address.instance) { + return; + } + const plugin = entry.Event.plugin; + if (dst.id === lib.Address.instance) { for (let instanceConnection of this.host.instanceConnections.values()) { - if (instanceConnection !== origin) { + if (instanceConnection !== origin && (!plugin || instanceConnection.plugins.has(plugin))) { instanceConnection.connector.send(message); } } - if (this.host !== origin) { + if (this.host !== origin && (!plugin || this.host.serverPlugins.has(plugin))) { this.host.connector.send(message); } } else if (dst.id === lib.Address.control) { if (this.host !== origin) { - this.host.connector.send(message); + if (!plugin || this.host.serverPlugins.has(plugin)) { + this.host.connector.send(message); + } } else { - logger.warn(`Received control broadcast of ${(message as lib.MessageEvent).name} from master.`); + logger.warn(`Received control broadcast of ${(message as lib.MessageEvent).name} from controller.`); } } else { - this.warnUnrouted(message, `Unexpected broacdast target ${dst.id}`); + this.warnUnrouted(message, `Unexpected broadcast target ${dst.id}`); } } diff --git a/packages/lib/src/link/link.ts b/packages/lib/src/link/link.ts index 14b7d1ec..520e5fe4 100644 --- a/packages/lib/src/link/link.ts +++ b/packages/lib/src/link/link.ts @@ -72,7 +72,12 @@ export type RequestHandler = (request: Req, src: libData.Address, dst: export type EventHandler = (event: T, src: libData.Address, dst: libData.Address) => Promise; interface Router { - forwardMessage(origin: Link, message: MessageRoutable, fallback: boolean): boolean, + forwardMessage( + origin: Link, + message: MessageRoutable, + entry: RequestEntry | EventEntry | undefined, + fallback: boolean, + ): boolean, } // Some definitions for the terminology used here: @@ -343,7 +348,7 @@ export class Link { let fallback = message.type === "request" ? this._requestFallbacks.get((entry as RequestEntry).Request) : undefined ; - if (this.router.forwardMessage(this, message, Boolean(fallback))) { + if (this.router.forwardMessage(this, message, entry, Boolean(fallback))) { return; } if (!fallback) { diff --git a/test/integration/routing.js b/test/integration/routing.js new file mode 100644 index 00000000..6881f079 --- /dev/null +++ b/test/integration/routing.js @@ -0,0 +1,422 @@ +"use strict"; +const assert = require("assert").strict; + +const lib = require("@clusterio/lib"); +const { Host, Instance, InstanceConnection } = require("@clusterio/host"); +const { ControlConnection, Controller, HostConnection, InstanceInfo } = require("@clusterio/controller"); + +const addr = lib.Address.fromShorthand; + +class Control extends lib.Link { } + +/** + * @param {Controller} controller + * @param {number} hostId + */ +function connectHost(controller, hostId, plugins) { + const [controllerSide, hostSide] = lib.VirtualConnector.makePair(addr("controller"), addr({ hostId })); + const registerData = new lib.RegisterHostData("", "0.0.0", hostId, plugins); + const hostConnection = new HostConnection(registerData, controllerSide, controller, "host-a.test"); + controller.wsServer.hostConnections.set(hostId, hostConnection); + const hostConfig = new lib.HostConfig("host"); + hostConfig.set("host.id", hostId); + return new Host(hostSide, "invalid", hostConfig, undefined, []); +} + +/** + * @param {Controller} controller + * @param {Host} host + * @param {number} instanceId + */ +function connectInstance(controller, host, instanceId) { + const hostAddress = addr({ hostId: host.config.get("host.id")}); + const instanceAddress = addr({ instanceId }); + const [instanceSide, hostSide] = lib.VirtualConnector.makePair(instanceAddress, hostAddress); + const instanceConfig = new lib.InstanceConfig("host"); + instanceConfig.set("instance.id", instanceId); + instanceConfig.set("instance.assigned_host", host.config.get("host.id")); + host.instanceInfos.set(instanceId, { path: "invalid", config: instanceConfig }); + controller.instances.set(instanceId, new InstanceInfo(instanceConfig, "stopped")); + const instance = new Instance(host, instanceSide, "invalid", "invalid", instanceConfig); + const instanceConnection = new InstanceConnection(hostSide, host, instance); + host.instanceConnections.set(instanceId, instanceConnection); + return instance; +} + +/** + * @param {Controller} controller + * @param {number} hostId + */ +function connectControl(controller, controlId) { + const [controllerSide, controlSide] = lib.VirtualConnector.makePair(addr("controller"), addr({ controlId })); + const registerData = new lib.RegisterControlData("", "0.0.0"); + let user = controller.userManager.users.get("test"); + if (!user) { + user = controller.userManager.createUser("test"); + } + const controlConnection = new ControlConnection(registerData, controllerSide, controller, user, controlId); + controller.wsServer.controlConnections.set(controlId, controlConnection); + return new Control(controlSide); +} + +class TestRequest { + static type = "request"; + static src = ["controller", "host", "instance", "control"]; + static dst = ["controller", "host", "instance", "control"]; + static permission = null; +} + +class TestEvent { + static type = "event"; + static src = ["controller", "host", "instance", "control"]; + static dst = ["controller", "host", "instance", "control"]; + static permission = null; +} + +class PluginEvent { + static type = "event"; + static src = ["controller", "host", "instance", "control"]; + static dst = ["controller", "host", "instance", "control"]; + static permission = null; + static plugin = "test"; +} + +lib.Link.register(TestRequest); +lib.Link.register(TestEvent); +lib.Link.register(PluginEvent); + +describe("Integration of link routing", function() { + /** @type {Controller} */ + let controller; + /** @type {Host} */ + let hostA; + /** @type {Host} */ + let hostB; + /** @type {Instance} */ + let instanceA1; + /** @type {Instance} */ + let instanceA2; + /** @type {Instance} */ + let instanceB1; + /** @type {Control} */ + let controlA; + /** @type {Control} */ + let controlB; + + beforeEach(function() { + controller = new Controller( + {}, + [], + "invalid", + new lib.ControllerConfig("controller") + ); + hostA = connectHost(controller, 10, []); + hostB = connectHost(controller, 20, []); + instanceA1 = connectInstance(controller, hostA, 11); + instanceA2 = connectInstance(controller, hostA, 12); + instanceB1 = connectInstance(controller, hostB, 21); + controlA = connectControl(controller, 100); + controlB = connectControl(controller, 101); + }); + + /** @returns {Controller | Host | Instance} */ + function get(name) { + return { + controller, + hostA, + hostB, + instanceA1, + instanceA2, + instanceB1, + controlA, + controlB, + }[name]; + } + + // Combinations to try. Equivalent paths have been removed. + const parties = [ + ["controller", ["hostA", "instanceA1", "controlA"]], + ["hostA", ["controller", "hostB", "instanceA1", "instanceB1", "controlA"]], + ["instanceA1", ["controller", "hostA", "hostB", "instanceA2", "instanceB1", "controlA"]], + ["controlA", ["controller", "hostA", "instanceA1", "controlB"]], + ]; + + function handle(dst, Message, handler) { + if (dst instanceof Controller) { + for (const host of dst.wsServer.hostConnections.values()) { + host.handle(Message, handler); + } + for (const control of dst.wsServer.controlConnections.values()) { + control.handle(Message, handler); + } + } else if (dst instanceof Host) { + for (const instance of dst.instanceConnections.values()) { + instance.handle(Message, handler); + } + dst.handle(Message, handler); + } else { + dst.handle(Message, handler); + } + } + + /** + * @param {Controller | Host | Instance} link + * @param {Address} dst + */ + function send(link, dst, message) { + // This logic exists becasue .sendTo doesn't forward all kinds of + // messages to the correct location from all places. Idealy this + // would not be neccessary and .sendTo would do the right thing. + if (link instanceof Host) { + const instance = link.instanceConnections.get(dst.id); + if (instance) { + return instance.sendTo(dst, message); + } + } + return link.sendTo(dst, message); + } + + describe("request routing", function() { + for (const [srcName, dstNames] of parties) { + it(`should route from ${srcName}`, async function() { + for (const dstName of dstNames) { + let called; + const dst = get(dstName); + handle(dst, TestRequest, async () => { called = true; }); + const dstAddr = dst instanceof Controller ? addr("controller") : dst.connector.src; + await send(get(srcName), dstAddr, new TestRequest()); + assert(called, `${dstName} handler was not called`); + } + }); + } + }); + + describe("event routing", function() { + for (const [srcName, dstNames] of parties) { + it(`should route from ${srcName}`, async function() { + for (const dstName of dstNames) { + let called; + const dst = get(dstName); + handle(dst, TestEvent, async () => { called = true; }); + const dstAddr = dst instanceof Controller ? addr("controller") : dst.connector.src; + send(get(srcName), dstAddr, new TestEvent()); + assert(called, `${dstName} handler was not called`); + } + }); + } + }); + + /** + * @param {Controller | Host | Instance | Control} link + * @param {"allHosts" | "allInstances" | "allControls"} dst + */ + function bcast(link, dst, message) { + // This logic exists becasue .sendTo doesn't forward all kinds of + // messages to the correct location from all places. Idealy this + // would not be neccessary and .sendTo would do the right thing. + if (link instanceof Host) { + const plugin = message.constructor.plugin; + if (dst === "allInstances") { + for (const instance of link.instanceConnections.values()) { + if (plugin && !instance.plugins.has(plugin)) { + continue; + } + instance.sendTo(dst, message); + } + } + if (plugin && !link.serverPlugins.has(plugin)) { + return; + } + } + link.sendTo(dst, message); + } + + function bhandle(srcName, dst, Event, canReach = (dstB) => true) { + const asserts = []; + const dsts = { + allHosts: ["hostA", "hostB"], + allInstances: ["instanceA1", "instanceA2", "instanceB1"], + allControls: ["controlA", "controlB"], + }; + for (const dstName of dsts[dst]) { + if (srcName === dstName) { continue; } + // TODO remove when broadcasting from instance to hosts is fixed, see #575 + if (srcName === "instanceA1" && dstName === "hostB") { continue; } + let called; + handle(get(dstName), Event, async () => { called = true; }); + const should = canReach(dstName); + asserts.push(() => assert( + !should ^ called, + `${dstName} handler was ${should ? "not " : ""}called from ${srcName}`, + )); + } + return asserts; + } + + describe("event broadcast", function() { + for (const srcName of ["controller", "hostA", "instanceA1", "controlA"]) { + it(`should broadcast from ${srcName}`, async function() { + for (const dstName of ["allHosts", "allInstances", "allControls"]) { + const dstAsserts = bhandle(srcName, dstName, TestEvent); + bcast(get(srcName), dstName, new TestEvent()); + for (const dstAssert of dstAsserts) { + dstAssert(); + } + } + }); + } + }); + + const graph = { + controller: ["hostA", "hostB", "controlA", "controlB"], + hostA: ["controller", "instanceA1", "instanceA2"], + hostB: ["controller", "instanceB1"], + instanceA1: ["hostA"], + instanceA2: ["hostA"], + instanceB1: ["hostB"], + controlA: ["controller"], + controlB: ["controller"], + }; + + // Returns true if the path from src to dst goes through via + function goesVia(src, dst, via) { + // Places we've been to + const checked = new Set(); + // Places left to check + const paths = new Set([src]); + while (paths.size) { + const [candidate] = paths; + paths.delete(candidate); + if (candidate === via) { + continue; + } + if (checked.has(candidate)) { + continue; + } + if (candidate === dst) { + return false; + } + checked.add(candidate); + for (const newCandidate of graph[candidate]) { + paths.add(newCandidate); + } + } + return true; + } + assert(goesVia("instanceA1", "instanceB1", "controller")); + assert(!goesVia("instanceA1", "instanceA2", "controller")); + assert(goesVia("instanceA1", "hostB", "hostA")); + assert(!goesVia("instanceA1", "hostB", "controlA")); + assert(goesVia("controlA", "controlB", "controller")); + assert(goesVia("controlA", "hostA", "controller")); + + function addPluginExceptFor(exName) { + const ex = get(exName); + const exAddr = ex instanceof Controller ? addr("controller") : ex.connector.src; + for (const name of [ + "controller", "hostA", "hostB", "instanceA1", "instanceA2", "instanceB1", "controlA", "controlB", + ]) { + const party = get(name); + if (party instanceof Controller) { + for (const host of party.wsServer.hostConnections.values()) { + if (!host.connector.dst.equals(exAddr)) { + host.plugins.set("test", "0.0.0"); + } + } + for (const control of party.wsServer.controlConnections.values()) { + if (!control.connector.dst.equals(exAddr)) { + // No way to set plugins + } + } + } else if (party instanceof Host) { + for (const instance of party.instanceConnections.values()) { + if (!instance.connector.dst.equals(exAddr)) { + instance.plugins.set("test", "0.0.0"); + } + } + if (!party.connector.dst.equals(exAddr)) { + party.serverPlugins.set("test", "0.0.0"); + } + } else if (party instanceof Instance) { + // Currently does not handle plugins + } else if (party instanceof Control) { + // Currently does not handle plugins + } + } + } + + // When plugin events are broadcast the party sending the event should + // filter it out if the next hop it is forwarded to does not have plugin + // the event was defined in loaded. + describe("event broadcast with controller missing plugin", function() { + for (const srcName of ["hostA", "instanceA1", "controlA"]) { + it(`should broadcast from ${srcName}`, async function() { + if (srcName === "controlA") { + this.skip(); // Not implemented + } + addPluginExceptFor("controller"); + for (const dstName of ["allHosts", "allInstances", "allControls"]) { + const dstAsserts = bhandle( + srcName, dstName, PluginEvent, dstB => !goesVia(srcName, dstB, "controller"), + ); + bcast(get(srcName), dstName, new PluginEvent()); + for (const dstAssert of dstAsserts) { + dstAssert(); + } + } + }); + } + }); + + describe("event broadcast with host missing plugin", function() { + for (const srcName of ["controller", "hostA", "instanceA1", "controlA"]) { + it(`should broadcast from ${srcName}`, async function() { + addPluginExceptFor("hostB"); + for (const dstName of ["allHosts", "allInstances", "allControls"]) { + const dstAsserts = bhandle( + srcName, dstName, PluginEvent, dstB => !goesVia(srcName, dstB, "hostB"), + ); + bcast(get(srcName), dstName, new PluginEvent()); + for (const dstAssert of dstAsserts) { + dstAssert(); + } + } + }); + } + }); + + describe("event broadcast with instance missing plugin", function() { + for (const srcName of ["controller", "hostA", "instanceA1", "controlA"]) { + it(`should broadcast from ${srcName}`, async function() { + addPluginExceptFor("instanceA2"); + for (const dstName of ["allHosts", "allInstances", "allControls"]) { + const dstAsserts = bhandle( + srcName, dstName, PluginEvent, dstB => !goesVia(srcName, dstB, "instanceA2"), + ); + bcast(get(srcName), dstName, new PluginEvent()); + for (const dstAssert of dstAsserts) { + dstAssert(); + } + } + }); + } + }); + + describe("event broadcast with control missing plugin", function() { + for (const srcName of ["controller", "hostA", "instanceA1", "controlA"]) { + it(`should broadcast from ${srcName}`, async function() { + this.skip(); // Currently not implemented. + addPluginExceptFor("controlB"); + for (const dstName of ["allHosts", "allInstances", "allControls"]) { + const dstAsserts = bhandle( + srcName, dstName, PluginEvent, dstB => !goesVia(srcName, dstB, "controlB"), + ); + bcast(get(srcName), dstName, new PluginEvent()); + for (const dstAssert of dstAsserts) { + dstAssert(); + } + } + }); + } + }); +});