WEBDIENSTE/huly-platform
1
0
Fork 0
mirror of https://github.com/hcengineering/platform.git synced 2026-08-31 04:09:43 +02:00
Code Issues Packages Projects Releases Wiki Activity

Rework API

Browse Source
This commit is contained in:
Andrey Sobolev
2025-09-15 20:25:13 +07:00
parent 4916937d13
commit 4712fa6c29
26 changed files with 613 additions and 368 deletions
Download Patch File Download Diff File Expand all files Collapse all files
.vscode/launch.json
Vendored
+3 -2
View File
@@ -20,7 +20,8 @@
"request": "launch",
"args": ["src/index.ts"],
"env": {
"PORT": "37371"
"PORT": "37371",
"DEVELOPMENT": "true"
},
"runtimeArgs": ["--nolazy", "-r", "ts-node/register"],
"runtimeVersion": "22",
@@ -34,7 +35,7 @@
"name": "Debug Tool",
"type": "node",
"request": "launch",
"args": ["src/index.ts", "list-agents"],
"args": ["src/index.ts", "bench-agent"],
"env": {
"NETWORK_HOST": "localhost:37371"
},
packages/backrpc/src/client.ts
+9 -2
View File
@@ -68,12 +68,13 @@ export class BackRPCClient<ClientT extends string = ClientId> {
this.observer = new zmq.Observer(this.dealer)
this.dealer.connect(`tcp://${host}:${port}`)
this.observer.on('connect', (data) => {
console.log('Connected to server', data)
void this.sendHello().catch((err) => {
console.error('Failed to send hello', err)
})
})
this.dealer.connect(`tcp://${host}:${port}`)
void this.start().catch((err) => {
console.error('Failed to start BackRPCClient', err)
})
@@ -101,7 +102,7 @@ export class BackRPCClient<ClientT extends string = ClientId> {
}
async checkAlive (): Promise<void> {
await this.doSend([backrpcOperations.ping, this.clientId as string, '', ''])
await this.doSend([backrpcOperations.ping, '', '', ''])
}
private async sendHello (): Promise<void> {
@@ -199,6 +200,10 @@ export class BackRPCClient<ClientT extends string = ClientId> {
}
private async resendRequests (): Promise<void> {
if (this.closed) {
console.error('Client is closed, cannot resend requests')
return
}
for (const [reqId, req] of Array.from(this.requests.entries())) {
try {
await this.doSend([backrpcOperations.request, reqId, JSON.stringify([req.method, req.params])])
@@ -214,6 +219,8 @@ export class BackRPCClient<ClientT extends string = ClientId> {
}
}
rCount = 0
async request<T>(method: string, params: any): Promise<T> {
if (this.serverId instanceof Promise) {
await this.serverId
packages/backrpc/src/server.ts
+8 -2
View File
@@ -66,9 +66,15 @@ export class BackRPCServer<ClientT extends string = ClientId> {
private readonly handlers: BackRPCServerHandler<ClientT>,
private readonly tickMgr: TickManager,
readonly host: string = '*',
private readonly port: number = 0
private readonly port: number = 0,
private readonly options: zmq.SocketOptions<zmq.Router> = {}
) {
this.router = new zmq.Router({ context })
this.router = new zmq.Router({
...options,
context,
// linger: 0,
tcpKeepalive: 1
})
this.stopTick = this.tickMgr.register(() => {
void this.checkAlive().catch((err) => {
packages/client/src/agent.ts
+2 -2
View File
@@ -2,7 +2,7 @@ import {
agentDirectRef,
type ClientUuid,
type ContainerConnection,
type ContainerEvent,
type NetworkEvent,
type ContainerUuid,
type NetworkAgent,
type TickManager
@@ -42,7 +42,7 @@ export class NetworkAgentServer implements BackRPCServerHandler<ClientUuid> {
await this.agent.onAgentUpdate?.()
}
async onContainerUpdate (event: ContainerEvent): Promise<void> {
async onContainerUpdate (event: NetworkEvent): Promise<void> {
// Handle container update
}
packages/client/src/client.ts
+123 -68
View File
@@ -1,25 +1,28 @@
import { BackRPCClient, type BackRPCResponseSend } from '@hcengineering/network-backrpc'
import {
agentDirectRef,
EndpointKind,
parseEndpointRef,
type AgentEndpointRef,
type AgentRecord,
type AgentRecordInfo,
type AgentUuid,
type ClientUuid,
type ContainerConnection,
type ContainerEndpointRef,
type ContainerEvent,
type NetworkEvent,
type ContainerKind,
type ContainerRecord,
type ContainerReference,
type ContainerRequest,
type ContainerUpdateListener,
type NetworkUpdateListener,
type ContainerUuid,
type GetOptions,
type NetworkAgent,
type NetworkClient,
type TickManager
type TickManager,
NetworkEventKind
} from '@hcengineering/network-core'
import { v4 as uuidv4 } from 'uuid'
import { ContainerConnectionImpl, NetworkDirectConnectionImpl, RoutedNetworkAgentConnectionImpl } from './agent'
import { BackRPCClient, type BackRPCResponseSend } from '@hcengineering/network-backrpc'
import { agentDirectRef, EndpointKind, parseEndpointRef } from '@hcengineering/network-core'
import { opNames } from './types'
interface ClientAgentRecord {
@@ -31,8 +34,7 @@ interface ClientAgentRecord {
class ContainerReferenceImpl implements ContainerReference {
constructor (
readonly uuid: ContainerUuid,
private readonly client: NetworkClientImpl,
private readonly _request: ContainerRequest
private readonly client: NetworkClientImpl
) {}
get endpoint (): ContainerEndpointRef {
@@ -57,16 +59,16 @@ class ContainerReferenceImpl implements ContainerReference {
if (conn !== undefined) {
return conn
}
const endpoint = await this.client.getContainerRef(this.uuid, this._request)
conn = this.client.establishConnection(this.uuid, endpoint)
conn = this.client.establishConnection(this.uuid, this.endpoint)
await conn.connect()
return conn
}
}
interface ContainereRef {
interface ContainerRef {
ref: ContainerReference
request: ContainerRequest
kind: ContainerKind
request: GetOptions
endpoint: ContainerEndpointRef
}
@@ -86,9 +88,10 @@ export class NetworkClientImpl implements NetworkClient {
containerConnections = new Map<ContainerUuid, ContainerConnectionImpl>()
agentConnections = new Map<AgentEndpointRef, RoutedNetworkAgentConnectionImpl<ClientUuid>>()
containerListeners: ContainerUpdateListener[] = []
cid: number = 0
containerListeners = new Map<number, NetworkUpdateListener>()
references = new Map<ContainerUuid, ContainereRef>()
references = new Map<ContainerUuid, ContainerRef>()
registered: boolean = false
@@ -124,6 +127,9 @@ export class NetworkClientImpl implements NetworkClient {
}
async close (): Promise<void> {
for (const refs of this.references.values()) {
await refs.ref.close()
}
for (const directConn of this.containerConnections.values()) {
await directConn.close()
}
@@ -166,12 +172,12 @@ export class NetworkClientImpl implements NetworkClient {
}
}
async onEvent (event: ContainerEvent): Promise<void> {
async onEvent (event: NetworkEvent): Promise<void> {
// Handle container events
// In case of container stopped, agent stopped or endpoint changed, we need to update direct connections to be re-established.
await this.handleConnectionUpdates(event)
for (const listener of this.containerListeners) {
for (const listener of this.containerListeners.values()) {
try {
await listener(event)
} catch (error) {
@@ -180,32 +186,52 @@ export class NetworkClientImpl implements NetworkClient {
}
}
async handleRefUpdate (
uuid: ContainerUuid,
endpoint: ContainerEndpointRef | Promise<ContainerEndpointRef>
): Promise<void> {
async handleRefUpdate (uuid: ContainerUuid, endpoint: ContainerEndpointRef): Promise<void> {
const ref = this.references.get(uuid)
if (ref !== undefined) {
const refEndpoint = endpoint instanceof Promise ? await endpoint : endpoint
const conn = this.containerConnections.get(ref.ref.uuid)
if (conn !== undefined && ref.endpoint !== refEndpoint) {
conn.setConnection(this.establishConnection(ref.ref.uuid, refEndpoint))
if (conn !== undefined && ref.endpoint !== endpoint) {
conn.setConnection(this.establishConnection(ref.ref.uuid, endpoint))
} else {
ref.endpoint = refEndpoint
ref.endpoint = endpoint
}
}
}
async onRegister (): Promise<void> {
// TODO: Add retry in container requests on re-connect to new network.
for (const [uuid, ref] of this.references.entries()) {
await this.handleRefUpdate(uuid, await this.retryGetContainerRef(uuid, ref.request))
async handleNewContainer (oldUuid: ContainerUuid, uuid: ContainerUuid, endpoint: ContainerEndpointRef): Promise<void> {
const ref = this.references.get(oldUuid)
this.references.delete(oldUuid)
if (ref !== undefined) {
const conn = this.containerConnections.get(oldUuid)
this.containerConnections.delete(oldUuid)
if (conn !== undefined) {
this.containerConnections.set(uuid, conn)
if (ref.endpoint !== endpoint) {
conn.setConnection(this.establishConnection(uuid, endpoint))
}
}
ref.ref.uuid = uuid
ref.endpoint = endpoint
this.references.set(uuid, ref)
}
}
async onRegister (): Promise<void> {
this.registered = true
// We need to re-register all our managed agents
// We need to re-register all our managed agents, since we could provide containers we request to our selfs
for (const agent of this._agents.values()) {
await this.doRegister(agent.agent)
}
for (const [uuid, ref] of this.references.entries()) {
const [newUuid, newEndpoint] = await this.retryGetContainerRef(ref.kind, ref.request)
if (uuid !== newUuid) {
await this.handleNewContainer(uuid, newUuid, newEndpoint)
} else {
await this.handleRefUpdate(uuid, newEndpoint)
}
}
}
/**
@@ -258,23 +284,27 @@ export class NetworkClientImpl implements NetworkClient {
this._agents.get(agent.uuid)?.resolve()
}
async agents (): Promise<AgentRecord[]> {
async agents (): Promise<AgentRecordInfo[]> {
// Return actual list of agents
return await this.client.request<AgentRecord[]>(opNames.getAgents, {})
return await this.client.request<AgentRecordInfo[]>(opNames.getAgents, {})
}
async kinds (): Promise<ContainerKind[]> {
return await this.client.request<ContainerKind[]>(opNames.getKinds, {})
}
async get (uuid: ContainerUuid, request: ContainerRequest): Promise<ContainerReference> {
const existing = this.references.get(uuid)
if (existing !== undefined) {
return existing.ref
async get (kind: ContainerKind, request: GetOptions): Promise<ContainerReference> {
// TODO: Wait for all pending requests to finish
if (request.uuid !== undefined) {
const existing = this.references.get(request.uuid)
if (existing !== undefined) {
return existing.ref
}
}
const endpoint = await this.getContainerRef(uuid, request)
const ref: ContainerReference = new ContainerReferenceImpl(uuid, this, request)
this.references.set(uuid, { ref, request, endpoint })
const [uuid, endpoint] = await this.retryGetContainerRef(kind, request)
const ref: ContainerReference = new ContainerReferenceImpl(uuid, this)
this.references.set(uuid, { kind, ref, request, endpoint })
return ref
}
@@ -325,38 +355,59 @@ export class NetworkClientImpl implements NetworkClient {
return conn
}
async handleConnectionUpdates (event: ContainerEvent): Promise<void> {
async handleConnectionUpdates (event: NetworkEvent): Promise<void> {
// Handle connection updates
for (const updated of event.updated) {
await this.handleRefUpdate(updated.uuid, updated.endpoint)
}
for (const deleted of event.deleted) {
await this.handleRefUpdate(deleted.uuid, deleted.endpoint)
}
}
async getContainerRef (uuid: ContainerUuid, request: ContainerRequest): Promise<ContainerEndpointRef> {
return await this.client.request<ContainerEndpointRef>(opNames.getContainer, { uuid, request })
}
async retryGetContainerRef (uuid: ContainerUuid, request: ContainerRequest): Promise<ContainerEndpointRef> {
let waitTimeout: number = 1
while (true) {
try {
const ref = await this.getContainerRef(uuid, request)
if (waitTimeout > 1) {
console.log(`Successfully got container ref for ${uuid} after ${waitTimeout - 1} retries.`)
}
return ref
} catch (err) {
console.warn(`Error getting container ref for ${uuid}. Will retry...`)
await this.tickMgr.waitTick(waitTimeout)
waitTimeout++
for (const e of event.containers ?? []) {
if (e.event === NetworkEventKind.removed || e.event === NetworkEventKind.updated) {
await this.handleRefUpdate(e.container.uuid, e.container.endpoint)
}
}
}
private async getContainerRef (
kind: ContainerKind,
request: GetOptions
): Promise<[ContainerUuid, ContainerEndpointRef]> {
return await this.client.request<[ContainerUuid, ContainerEndpointRef]>(opNames.getContainer, { kind, request })
}
async retryGetContainerRef (kind: ContainerKind, request: GetOptions): Promise<[ContainerUuid, ContainerEndpointRef]> {
let waitTimeout: number = 1
let earlyRetry = (): void => {}
const stop = this.onUpdate(async (event) => {
// We agent is appear with a required kind, we can retry immediately
if (event.agents.some((e) => e.event === NetworkEventKind.added && e.kinds.includes(kind))) {
waitTimeout = 0
earlyRetry()
}
})
try {
while (true) {
try {
const ref = await this.getContainerRef(kind, request)
if (waitTimeout > 1) {
console.log(`Successfully got container ref for ${kind} after ${waitTimeout - 1} retries.`)
}
return ref
} catch (err) {
console.warn(`Error getting container ref for ${kind}. Will retry...`)
await Promise.any([
this.tickMgr.waitTick(waitTimeout),
new Promise<void>((resolve) => {
earlyRetry = resolve
})
])
if (waitTimeout < this.tickMgr.tps * 5) {
waitTimeout++
}
}
}
} finally {
stop()
}
}
async release (uuid: ContainerUuid): Promise<void> {
await this.client.request<any>(opNames.releaseContainer, { uuid })
}
@@ -372,7 +423,11 @@ export class NetworkClientImpl implements NetworkClient {
return await this.client.request<any>(opNames.sendContainer, [target, operation, data])
}
onContainerUpdate (listener: ContainerUpdateListener): void {
this.containerListeners.push(listener)
onUpdate (listener: NetworkUpdateListener): () => void {
const cid = this.cid++
this.containerListeners.set(cid, listener)
return () => {
this.containerListeners.delete(cid)
}
}
}
packages/client/src/index.ts
+2 -1
View File
@@ -17,7 +17,6 @@ export * from './client'
export * from './types'
const tickMgr = new TickManagerImpl(timeouts.pingInterval * 2)
tickMgr.start()
export function shutdownNetworkTickMgr (): void {
tickMgr.stop()
@@ -30,6 +29,7 @@ process.on('exit', () => {
export function createNetworkClient (url: string): NetworkClient {
const [host, portStr] = url.split(':')
const port = portStr != null ? parseInt(portStr, 10) : 3737
tickMgr.start()
return new NetworkClientImpl(host, port, tickMgr)
}
@@ -42,6 +42,7 @@ export async function createAgent (
const [host, portStr] = endpointUrl.split(':')
const port = portStr != null ? parseInt(portStr, 10) : 3738
tickMgr.start()
const server = new NetworkAgentServer(tickMgr, host, '*', port)
await server.start(agent)
packages/client/src/types.ts
+12 -12
View File
@@ -1,16 +1,16 @@
export const opNames = {
// NetworkOperations
register: 'r',
unregister: 'u',
getAgents: 'a',
getKinds: 'k',
listContainers: 'l',
getContainer: 'g',
releaseContainer: 'r',
sendContainer: 's',
register: 'ar',
unregister: 'au',
getAgents: 'al',
getKinds: 'ak',
listContainers: 'cl',
getContainer: 'gg',
releaseContainer: 'cr',
sendContainer: 'cs',
// Agent operations
containerUpdate: 'c',
terminate: 't',
connect: 'c!',
disconnect: 'd'
containerUpdate: 'cu',
terminate: 'ct',
connect: 'cc',
disconnect: 'cd'
}
packages/core/src/__test__/network.spec.ts
+8 -4
View File
@@ -1,6 +1,6 @@
import { AgentImpl } from '../agent'
import type { AgentUuid, ClientUuid, ContainerEndpointRef, ContainerKind } from '../api/types'
import type { Container } from '../containers'
import type { AgentUuid, ClientUuid, ContainerEndpointRef, ContainerKind, GetOptions } from '../api/types'
import { containerUuid, type Container } from '../containers'
import { NetworkImpl } from '../network'
import { TickManagerImpl } from '../utils'
@@ -55,12 +55,16 @@ describe('network tests', () => {
const network = new NetworkImpl(tickManager)
const agent1 = new AgentImpl(agents.agent1, {
[kinds.session]: () => Promise.resolve([new DummyContainer(), '' as ContainerEndpointRef])
[kinds.session]: async (opt: GetOptions) => ({
uuid: opt.uuid ?? containerUuid(),
container: new DummyContainer(),
endpoint: '' as ContainerEndpointRef
})
})
await agent1.register(network)
expect((network as any)._agents.size).toBe(1)
const s1 = await network.get(agents.agent1 as any, 's1' as any, { kind: kinds.session })
const s1 = await network.get(agents.agent1 as any, kinds.session, { uuid: 's1' as any })
expect(s1).toBeDefined()
})
})
packages/core/src/agent.ts
+80 -30
View File
@@ -6,7 +6,7 @@ import type {
ContainerEndpointRef,
ContainerKind,
ContainerRecord,
ContainerRequest,
GetOptions,
ContainerUuid
} from './api/types'
import type { Container, ContainerFactory } from './containers'
@@ -17,14 +17,25 @@ interface ContainerRecordImpl {
endpoint: ContainerEndpointRef
kind: ContainerKind
labels?: string[]
lastVisit: number
}
interface AgentPendingContainer {
kind: ContainerKind
options: GetOptions
promise: Promise<ContainerRecordImpl>
}
export class AgentImpl implements NetworkAgent {
// Own, managed containers
private readonly _byId = new Map<ContainerUuid, ContainerRecordImpl | Promise<ContainerRecordImpl>>()
private readonly _byId = new Map<ContainerUuid, ContainerRecordImpl>()
private readonly _byKind = new Map<ContainerKind, Map<ContainerUuid, ContainerRecordImpl>>()
private readonly _containers = new Map<ContainerUuid, ContainerRecordImpl>()
// Containers pending startup
pidCounter: number = 0
private readonly pendings = new Map<number, AgentPendingContainer>()
endpoint?: AgentEndpointRef | undefined
@@ -49,7 +60,7 @@ export class AgentImpl implements NetworkAgent {
}
async list (kind?: ContainerKind): Promise<ContainerRecord[]> {
return Array.from(this._containers.values())
return Array.from(kind !== undefined ? (this._byKind.get(kind)?.values() ?? []) : this._byId.values())
.filter((it) => !(it instanceof Promise) && (kind === undefined || it.kind === kind))
.map((it) => ({
agentId: this.uuid,
@@ -64,47 +75,86 @@ export class AgentImpl implements NetworkAgent {
return Object.keys(this.factory) as ContainerKind[]
}
async getContainerImpl (uuid: ContainerUuid): Promise<ContainerRecordImpl | undefined> {
let current = this._byId.get(uuid)
if (current instanceof Promise) {
current = await current
this._byId.set(uuid, current)
selectContainer (kind: ContainerKind, labels?: string[]): ContainerRecordImpl | undefined {
const list = this._byKind.get(kind)
if (list !== undefined) {
let l = list.values()
if (labels !== undefined) {
l = l.filter((it) => it.labels !== undefined && labels.every((l) => (it.labels ?? []).includes(l)))
}
return l.next()?.value
}
return current
}
async getContainer (uuid: ContainerUuid): Promise<Container | undefined> {
return (await this.getContainerImpl(uuid))?.container
return this._byId.get(uuid)?.container
}
async get (uuid: ContainerUuid, request: ContainerRequest): Promise<ContainerEndpointRef> {
const current = await this.getContainerImpl(uuid)
if (current !== undefined) {
return current.endpoint
async get (kind: ContainerKind, options: GetOptions): Promise<[ContainerUuid, ContainerEndpointRef]> {
// If uuid is fixed, we must return that one
if (options.uuid !== undefined) {
const current = this._byId.get(options.uuid)
if (current !== undefined) {
return [current.uuid, current.endpoint]
}
// Check pending requests
for (const p of this.pendings.values()) {
if (p.kind === kind && p.options.uuid === options.uuid) {
const containerImpl = await p.promise
return [containerImpl.uuid, containerImpl.endpoint]
}
}
} else {
// If we have one with kind, return it.
const existing = this.selectContainer(kind, options.labels)
if (existing !== undefined) {
return [existing.uuid, existing.endpoint]
}
}
let container: ContainerRecordImpl | Promise<ContainerRecordImpl> = this.factory[request.kind](uuid, request).then(
(r) => ({
container: r[0],
endpoint: r[1],
kind: request.kind,
lastVisit: Date.now(),
uuid
})
)
this._byId.set(uuid, container)
container = await container
this._containers.set(uuid, container)
this._byId.set(uuid, container)
// Check pending container by kind
for (const p of this.pendings.values()) {
if (
p.kind === kind &&
(options.labels === undefined ||
(p.options.labels !== undefined && options.labels.every((l) => (p.options.labels ?? []).includes(l))))
) {
const containerImpl = await p.promise
return [containerImpl.uuid, containerImpl.endpoint]
}
}
return container.endpoint
const pendingId = this.pidCounter++
const container = this.factory[kind](options).then(({ uuid, container, endpoint }) => ({
container,
endpoint,
kind,
lastVisit: Date.now(),
uuid
}))
this.pendings.set(pendingId, { kind, promise: container, options })
const containerImpl = await container
this.pendings.delete(pendingId)
this._byId.set(containerImpl.uuid, containerImpl)
const byKind = this._byKind.get(kind)
if (byKind !== undefined) {
byKind.set(containerImpl.uuid, containerImpl)
} else {
this._byKind.set(kind, new Map([[containerImpl.uuid, containerImpl]]))
}
return [containerImpl.uuid, containerImpl.endpoint]
}
async terminate (uuid: ContainerUuid): Promise<void> {
const current = this._byId.get(uuid)
if (current !== undefined) {
this._containers.delete(uuid)
this._byId.delete(uuid)
this._byKind.get(current.kind)?.delete(uuid)
if (current instanceof Promise) {
await (await current).container.terminate() // Await promise before terminating
} else {
packages/core/src/api/agent.ts
+7 -5
View File
@@ -3,11 +3,11 @@ import type {
AgentEndpointRef,
AgentUuid,
ContainerEndpointRef,
ContainerEvent,
NetworkEvent,
ContainerKind,
ContainerRecord,
ContainerRequest,
ContainerUuid
ContainerUuid,
GetOptions
} from './types'
export interface AgentRecord {
@@ -21,6 +21,8 @@ export interface AgentRecord {
containers: ContainerRecord[]
kinds: ContainerKind[]
}
export type AgentRecordInfo = Omit<AgentRecord, 'containers'> & { containers: number }
/**
* Interface to Huly Agent on agent.
*/
@@ -35,13 +37,13 @@ export interface NetworkAgent {
kinds: ContainerKind[]
// event handled from agent to network events.
onUpdate?: (event: ContainerEvent) => Promise<void>
onUpdate?: (event: NetworkEvent) => Promise<void>
// Send agent update info to network, if applicable.
onAgentUpdate?: () => Promise<void>
// Get/Start of required container kind on agent
get: (uuid: ContainerUuid, request: ContainerRequest) => Promise<ContainerEndpointRef>
get: (kind: ContainerKind, options: GetOptions) => Promise<[ContainerUuid, ContainerEndpointRef]>
// A low level reference to container
getContainer: (uuid: ContainerUuid) => Promise<Container | undefined>
packages/core/src/api/client.ts
+9 -8
View File
@@ -1,14 +1,14 @@
import type { AgentRecord, NetworkAgent } from './agent'
import type { AgentRecordInfo, NetworkAgent } from './agent'
import type {
ContainerEndpointRef,
ContainerEvent,
NetworkEvent,
ContainerKind,
ContainerRecord,
ContainerRequest,
ContainerUuid
ContainerUuid,
GetOptions
} from './types'
export type ContainerUpdateListener = (event: ContainerEvent) => Promise<void>
export type NetworkUpdateListener = (event: NetworkEvent) => Promise<void>
/**
* Interface to Huly network.
@@ -24,7 +24,7 @@ export interface NetworkClient {
*/
register: (agent: NetworkAgent) => Promise<void>
agents: () => Promise<AgentRecord[]>
agents: () => Promise<AgentRecordInfo[]>
// A full uniq set of supported container kinds.
kinds: () => Promise<ContainerKind[]>
@@ -33,7 +33,7 @@ export interface NetworkClient {
* Get/Start of required container kind on agent
* Will start a required container on agent, if not already started.
*/
get: (uuid: ContainerUuid, request: ContainerRequest) => Promise<ContainerReference>
get: (kind: ContainerKind, options: GetOptions) => Promise<ContainerReference>
list: (kind?: ContainerKind) => Promise<ContainerRecord[]>
@@ -41,7 +41,8 @@ export interface NetworkClient {
request: (target: ContainerUuid, operation: string, data?: any) => Promise<any>
// Register on container update listener
onContainerUpdate: (listener: ContainerUpdateListener) => void
// Return unsubscribe function
onUpdate: (listener: NetworkUpdateListener) => () => void
// We could wait for a connection for a time period.
// If timeout === 0, we wait indefinitely.
packages/core/src/api/network.ts
+8 -8
View File
@@ -1,13 +1,13 @@
import type { AgentRecord, NetworkAgent } from './agent'
import type { AgentRecord, AgentRecordInfo, NetworkAgent } from './agent'
import type {
ContainerEndpointRef,
AgentUuid,
ClientUuid,
ContainerEndpointRef,
NetworkEvent,
ContainerKind,
ContainerUuid,
ContainerRequest,
ContainerRecord,
ContainerEvent
ContainerUuid,
GetOptions
} from './types'
/**
@@ -27,7 +27,7 @@ export interface Network {
// Mark an agent as alive (updates lastSeen timestamp)
ping: (agentId: AgentUuid | ClientUuid) => void
agents: () => Promise<AgentRecord[]>
agents: () => Promise<AgentRecordInfo[]>
// A full uniq set of supported container kinds.
kinds: () => Promise<ContainerKind[]>
@@ -36,7 +36,7 @@ export interface Network {
* Get/Start of required container kind on agent
* Will start a required container on agent, if not already started.
*/
get: (client: ClientUuid, uuid: ContainerUuid, request: ContainerRequest) => Promise<ContainerEndpointRef>
get: (client: ClientUuid, kind: ContainerKind, options: GetOptions) => Promise<[ContainerUuid, ContainerEndpointRef]>
/**
* Release a container for a client, if container is not used anymore it will be shutdown with a shutdown delay.
@@ -52,7 +52,7 @@ export interface Network {
}
export interface NetworkWithClients {
addClient: (clientUuid: ClientUuid, onContainer?: (event: ContainerEvent) => Promise<void>) => void
addClient: (clientUuid: ClientUuid, onContainer?: (event: NetworkEvent) => Promise<void>) => void
removeClient: (clientUuid: ClientUuid) => void
// When client is registering agent.
packages/core/src/api/timeouts.ts
+2 -1
View File
@@ -1,4 +1,5 @@
export const timeouts = {
aliveTimeout: 3, // seconds - timeout for detecting dead agents
aliveTimeout: 3, // seconds - timeout for detecting dead agents/clients
unusedContainerTimeout: 5, // 10 seconds for container to be terminated because of being unused
pingInterval: 1 // seconds - how often to ping agents
}
packages/core/src/api/types.ts
+11 -6
View File
@@ -17,14 +17,19 @@ export interface ContainerRecord {
labels?: string[]
}
export interface ContainerEvent {
added: ContainerRecord[]
deleted: ContainerRecord[]
updated: ContainerRecord[]
export enum NetworkEventKind {
added = 0,
updated = 1,
removed = 2
}
export interface NetworkEvent {
agents: { id: AgentUuid, kinds: ContainerKind[], event: NetworkEventKind }[]
containers: { container: ContainerRecord, event: NetworkEventKind }[]
}
export interface ContainerRequest {
kind: ContainerKind
export interface GetOptions {
uuid?: ContainerUuid
extra?: Record<string, any> // Extra parameters for container start
labels?: string[]
packages/core/src/containers.ts
+6 -9
View File
@@ -1,4 +1,5 @@
import type { ClientUuid, ContainerEndpointRef, ContainerRecord, ContainerRequest, ContainerUuid } from './api/types'
import type { ClientUuid, ContainerEndpointRef, ContainerUuid, GetOptions } from './api/types'
import { v4 as uuidv4 } from 'uuid'
export interface Container {
request: (operation: string, data?: any, clientId?: ClientUuid) => Promise<any>
@@ -15,13 +16,9 @@ export interface Container {
}
export type ContainerFactory = (
uuid: ContainerUuid,
request: ContainerRequest
) => Promise<[Container, ContainerEndpointRef]>
request: GetOptions
) => Promise<{ uuid: ContainerUuid, container: Container, endpoint: ContainerEndpointRef }>
export interface ContainerRecordImpl {
record: ContainerRecord
endpoint: ContainerEndpointRef | Promise<ContainerEndpointRef>
clients: Set<ClientUuid>
export function containerUuid (): ContainerUuid {
return uuidv4() as ContainerUuid
}
packages/core/src/network.ts
+238 -165
View File
@@ -1,23 +1,32 @@
import type { AgentRecord, NetworkAgent } from './api/agent'
import type { AgentRecord, AgentRecordInfo, NetworkAgent } from './api/agent'
import type { Network, NetworkWithClients } from './api/network'
import { timeouts } from './api/timeouts'
import { NetworkEventKind } from './api/types'
import type {
AgentEndpointRef,
AgentUuid,
ClientUuid,
ContainerEndpointRef,
ContainerEvent,
NetworkEvent,
ContainerKind,
ContainerRecord,
ContainerRequest,
ContainerUuid
ContainerUuid,
GetOptions
} from './api/types'
import type { TickManager } from './api/utils'
import type { ContainerRecordImpl } from './containers'
export interface ContainerRecordImpl {
record: ContainerRecord
endpoint: ContainerEndpointRef
agent: AgentRecordImpl
clients: Set<ClientUuid>
}
interface AgentRecordImpl {
id: AgentUuid
api: NetworkAgent
containers: Map<ContainerUuid, ContainerRecordImpl>
containers: Set<ContainerUuid>
endpoint?: AgentEndpointRef
kinds: ContainerKind[]
@@ -27,10 +36,18 @@ interface AgentRecordImpl {
interface ClientRecordImpl {
lastSeen: number
containers: Set<ContainerUuid>
onContainer?: (event: ContainerEvent) => Promise<void>
onContainer?: (event: NetworkEvent) => Promise<void>
agents: Set<AgentUuid>
}
interface PendingContainer {
agent: AgentUuid
kind: ContainerKind
options: GetOptions
promise: Promise<ContainerRecordImpl>
clients: Set<ClientUuid>
}
/**
* Server network implementation.
*/
@@ -39,13 +56,17 @@ export class NetworkImpl implements Network, NetworkWithClients {
private readonly _agents = new Map<AgentUuid, AgentRecordImpl>()
private readonly _containers = new Map<ContainerUuid, AgentUuid>()
private readonly _containers = new Map<ContainerUuid, ContainerRecordImpl>()
private pidCounter: number = 0
private readonly pending = new Map<number, PendingContainer>()
private readonly _clients = new Map<ClientUuid, ClientRecordImpl>()
private readonly _orphanedContainers = new Map<ContainerEndpointRef, ContainerRecordImpl>()
private readonly _orphanedContainers = new Map<ContainerUuid, { container: ContainerRecordImpl, time: number }>()
private eventQueue: ContainerEvent[] = []
private eventQueue: NetworkEvent[] = []
private readonly stopTick?: () => void
@@ -64,15 +85,13 @@ export class NetworkImpl implements Network, NetworkWithClients {
this.stopTick?.()
}
async agents (): Promise<AgentRecord[]> {
async agents (): Promise<AgentRecordInfo[]> {
return Array.from(
this._agents.values().map(({ api, containers }) => ({
agentId: api.uuid,
endpoint: api.endpoint,
kinds: api.kinds,
containers: Array.from(containers.values())
.filter((it) => !(it.endpoint instanceof Promise))
.map(({ record, endpoint }) => ({ ...record, endpoint: endpoint as ContainerEndpointRef }))
containers: containers.size
}))
)
}
@@ -87,99 +106,100 @@ export class NetworkImpl implements Network, NetworkWithClients {
}
async list (kind?: ContainerKind): Promise<ContainerRecord[]> {
return Array.from(this._agents.values())
.flatMap((it) => Array.from(it.containers.values()))
return Array.from(this._containers.values())
.filter((it) => kind === undefined || it.record.kind === kind)
.map((it) => it.record)
}
async request (target: ContainerUuid, operation: string, data?: any): Promise<any> {
const agentId = this._containers.get(target)
if (agentId === undefined) {
const container = this._containers.get(target)
if (container === undefined) {
throw new Error(`Container ${target} not found`)
}
const agent = this._agents.get(agentId)
if (agent === undefined) {
throw new Error(`Agent ${agentId} not found for container ${target}`)
}
const container = agent.containers.get(target)
if (container === undefined) {
throw new Error(`Container ${target} not registered on agent ${agentId}`)
}
return await agent.api.request(target, operation, data)
return await container.agent?.api.request(target, operation, data)
}
async register (record: AgentRecord, agent: NetworkAgent): Promise<ContainerUuid[]> {
const containers: ContainerRecord[] = record.containers
const newContainers = new Map<ContainerUuid, ContainerRecordImpl>(
containers.map((record) => [
const newContainers: ContainerRecord[] = record.containers
const newContainersMap = new Map<ContainerUuid, ContainerRecordImpl>(
newContainers.map((record) => [
record.uuid,
{
record,
request: { kind: record.kind },
endpoint: record.endpoint,
clients: new Set<ClientUuid>([])
clients: new Set<ClientUuid>([]),
agent: null as any // Temporarily
}
])
)
const containerEvent: ContainerEvent = {
added: [],
deleted: [],
updated: []
}
// Register agent record
const oldAgent = this._agents.get(record.agentId)
if (oldAgent !== undefined) {
// In case re-register or reconnect is happened.
// Check if some of container changed endpoints.
for (const rec of containers) {
const oldRec = oldAgent.containers.get(rec.uuid)
if (oldRec !== undefined) {
if (oldRec.record.endpoint !== rec.endpoint) {
oldRec.endpoint = rec.endpoint // Update endpoint
containerEvent.updated.push(rec)
}
}
}
// Handle remove of containers
for (const oldC of oldAgent.containers.values()) {
if (newContainers.get(oldC.record.uuid) === undefined) {
containerEvent.deleted.push(oldC.record)
this._containers.delete(oldC.record.uuid) // Remove from active container registry
}
}
}
const containersToShutdown: ContainerUuid[] = []
// Update active container registry.
for (const rec of containers) {
const oldAgentId = this._containers.get(rec.uuid)
if (oldAgentId === undefined) {
containerEvent.added.push(rec)
const containerImpl = newContainers.get(rec.uuid)
if (containerImpl !== undefined) {
this._orphanedContainers.set(rec.endpoint, containerImpl)
}
this._containers.set(rec.uuid, record.agentId)
}
if (oldAgentId !== undefined && oldAgentId !== record.agentId) {
containersToShutdown.push(rec.uuid)
}
const containerEvent: NetworkEvent = {
agents: [],
containers: []
}
// update agent record
this._agents.set(record.agentId, {
const agentRecord: AgentRecordImpl = {
id: record.agentId,
api: agent,
containers: newContainers,
containers: new Set(),
endpoint: record.endpoint,
kinds: record.kinds,
lastSeen: this.tickManager.now()
})
}
const oldAgent = this._agents.get(record.agentId)
this._agents.set(record.agentId, agentRecord)
const containersToShutdown: ContainerUuid[] = []
// Find removed containers
for (const cid of oldAgent?.containers ?? []) {
if (!newContainersMap.has(cid)) {
const container = this._containers.get(cid)
if (container !== undefined) {
containerEvent.containers.push({
container: container.record,
event: NetworkEventKind.removed
})
this._containers.delete(cid)
}
}
}
// Update active container registry.
for (const containerImpl of newContainersMap.values()) {
containerImpl.agent = agentRecord
const existingContainer = this._containers.get(containerImpl.record.uuid)
if (existingContainer === undefined) {
containerEvent.containers.push({
container: containerImpl.record,
event: NetworkEventKind.added
})
this._containers.set(containerImpl.record.uuid, containerImpl)
agentRecord.containers.add(containerImpl.record.uuid)
} else {
if (existingContainer.agent.id !== record.agentId) {
// Container already started on different agent, need to shutdown old one.
containersToShutdown.push(containerImpl.record.uuid)
}
if (existingContainer.record.endpoint !== containerImpl.record.endpoint) {
containerEvent.containers.push({
container: containerImpl.record,
event: NetworkEventKind.updated
})
}
}
}
containerEvent.agents.push({
id: agent.uuid,
kinds: record.kinds,
event: oldAgent !== undefined ? NetworkEventKind.updated : NetworkEventKind.added
})
this.eventQueue.push(containerEvent)
// Send notification to all agents about containers update.
@@ -191,7 +211,7 @@ export class NetworkImpl implements Network, NetworkWithClients {
if (agent === undefined) {
return // Fine to ignore
}
await this.processDeadAgent(agentId)
await this.processAgentRemove(agentId)
}
async sendEvents (): Promise<void> {
@@ -202,27 +222,38 @@ export class NetworkImpl implements Network, NetworkWithClients {
}
// Combine events
const finalEvent: ContainerEvent = {
added: [],
deleted: [],
updated: []
}
const agents = new Map<AgentUuid, NetworkEvent['agents'][0]>()
const containers = new Map<ContainerUuid, NetworkEvent['containers'][0]>()
for (const event of events) {
finalEvent.added.push(...event.added)
finalEvent.deleted.push(...event.deleted)
finalEvent.updated.push(...event.updated)
for (const agent of event.agents) {
if (agent.event === NetworkEventKind.removed) {
agents.delete(agent.id)
} else {
agents.set(agent.id, agent)
}
}
for (const container of event.containers) {
if (container.event === NetworkEventKind.removed) {
containers.delete(container.container.uuid)
} else {
containers.set(container.container.uuid, container)
}
}
}
// Skip deleted events.
const deletedIds = finalEvent.deleted.map((c) => c.uuid)
finalEvent.added = finalEvent.added.filter((c) => !deletedIds.includes(c.uuid))
finalEvent.updated = finalEvent.updated.filter((c) => !deletedIds.includes(c.uuid))
const finalEvent: NetworkEvent = {
agents: Array.from(agents.values()),
containers: Array.from(containers.values())
}
for (const [clientUuid, client] of Object.entries(this._clients)) {
for (const [clientUuid, client] of this._clients.entries()) {
if (client.onContainer !== undefined) {
try {
// We should not block on broadcast to clients.
void client.onContainer(finalEvent)
void client.onContainer?.(finalEvent)?.catch((err) => {
console.error(`Error in client ${clientUuid} onContainer callback:`, err)
})
} catch (err: any) {
console.error(`Error in client ${clientUuid} onContainer callback:`, err)
}
@@ -230,7 +261,7 @@ export class NetworkImpl implements Network, NetworkWithClients {
}
}
addClient (clientUuid: ClientUuid, onContainer?: (event: ContainerEvent) => Promise<void>): void {
addClient (clientUuid: ClientUuid, onContainer?: (event: NetworkEvent) => Promise<void>): void {
const info = this._clients.get(clientUuid) ?? {
lastSeen: this.tickManager.now(),
containers: new Set(),
@@ -268,77 +299,111 @@ export class NetworkImpl implements Network, NetworkWithClients {
}
}
async get (clientUuid: ClientUuid, uuid: ContainerUuid, request: ContainerRequest): Promise<ContainerEndpointRef> {
async get (
clientUuid: ClientUuid,
kind: ContainerKind,
options: GetOptions
): Promise<[ContainerUuid, ContainerEndpointRef]> {
let client = this._clients.get(clientUuid)
if (client === undefined) {
client = { lastSeen: this.tickManager.now(), containers: new Set(), agents: new Set() }
this._clients.set(clientUuid, client)
}
client.containers.add(uuid)
const record = await this.getContainer(uuid, request, [clientUuid])
if (record.endpoint instanceof Promise) {
return await record.endpoint
}
return record.endpoint
const record: ContainerRecordImpl = await this.getContainer(kind, options, [clientUuid])
client.containers.add(record.record.uuid)
return [record.record.uuid, record.endpoint]
}
async getContainer (
uuid: ContainerUuid,
request: ContainerRequest,
clients: ClientUuid[]
): Promise<ContainerRecordImpl> {
const existing = this._containers.get(uuid)
if (existing !== undefined) {
const agent = this._agents.get(existing)
const containerImpl = agent?.containers?.get(uuid)
if (containerImpl !== undefined) {
if (!(containerImpl.endpoint instanceof Promise)) {
this._orphanedContainers.delete(containerImpl.endpoint)
}
async getContainer (kind: ContainerKind, options: GetOptions, clients: ClientUuid[]): Promise<ContainerRecordImpl> {
// Reuse existing container if uuid is provided and container exists
if (options.uuid !== undefined) {
const existing = this._containers.get(options.uuid)
if (existing !== undefined) {
this._orphanedContainers.delete(existing.record.uuid)
for (const cl of clients) {
containerImpl.clients.add(cl)
existing.clients.add(cl)
}
return existing
}
// Find if container is pending starting for our it
for (const p of this.pending.values()) {
if (p.kind === kind && p.options.uuid === options.uuid) {
// Add to pendings list to properly track orphaned
for (const cl of clients) {
p.clients.add(cl)
}
const containerImpl = await p.promise
for (const cl of clients) {
containerImpl.clients.add(cl)
}
return containerImpl
}
}
} else {
// Check if we have a container pending if no uuid is provided
for (const p of this.pending.values()) {
if (
p.kind === kind &&
(options.labels === undefined ||
(p.options.labels !== undefined && options.labels.every((l) => (p.options.labels ?? []).includes(l))))
) {
// Add to pendings list to properly track orphaned
for (const cl of clients) {
p.clients.add(cl)
}
const containerImpl = await p.promise
for (const cl of clients) {
containerImpl.clients.add(cl)
}
return containerImpl
}
return containerImpl
}
}
// Select agent using round/robin and register it in agent
const suitableAgents = Array.from(this._agents.values().filter((it) => it.kinds.includes(request.kind)))
const suitableAgents = Array.from(this._agents.values().filter((it) => it.kinds.includes(kind)))
if (suitableAgents.length === 0) {
throw new Error(`No suitable agents found for container ${uuid}`)
throw new Error(`No suitable agents found for container ${kind}`)
}
const agent = Array.from(suitableAgents)[++this.idx % suitableAgents.length]
const record: ContainerRecordImpl = {
const record: Promise<ContainerRecordImpl> = agent.api.get(kind, options).then(([uuid, endpoint]) => ({
agent,
record: {
uuid,
agentId: agent.api.uuid,
kind: request.kind,
kind,
lastVisit: this.tickManager.now(),
endpoint: '' as ContainerEndpointRef, // Placeholder, will be updated later
labels: request.labels,
extra: request.extra
labels: options.labels,
extra: options.extra
},
clients: new Set(clients),
endpoint: agent.api.get(uuid, request)
}
agent.containers.set(uuid, record)
this._containers.set(uuid, agent.api.uuid)
endpoint
}))
const pid = ++this.pidCounter
this.pending.set(pid, { agent: agent.id, kind, options, promise: record, clients: new Set(clients) })
// Wait for endpoint to be established
try {
const endpointRef = await record.endpoint
record.endpoint = endpointRef
const recordImpl = await record
agent.containers.add(recordImpl.record.uuid)
this.eventQueue.push({
added: [record.record],
deleted: [],
updated: []
agents: [],
containers: [{ container: recordImpl.record, event: NetworkEventKind.added }]
})
return record
// TODO: What if container started with same id?
this._containers.set(recordImpl.record.uuid, recordImpl)
return recordImpl
} catch (err: any) {
this._containers.delete(uuid) // Remove from active container registry
throw new Error(`Failed to get endpoint for container ${uuid}: ${err.message}`)
throw new Error(`Failed to get endpoint for container ${kind}: ${err.message}`)
} finally {
this.pending.delete(pid)
}
}
@@ -348,29 +413,26 @@ export class NetworkImpl implements Network, NetworkWithClients {
const existing = this._containers.get(uuid)
if (existing !== undefined) {
const agent = this._agents.get(existing)
const containerImpl = agent?.containers?.get(uuid)
if (containerImpl !== undefined) {
containerImpl.clients.delete(client)
if (containerImpl.clients.size === 0 && !(containerImpl.endpoint instanceof Promise)) {
this._orphanedContainers.set(containerImpl.endpoint, containerImpl)
}
existing.clients.delete(client)
if (existing.clients.size === 0) {
this._orphanedContainers.set(existing.record.uuid, {
container: existing,
time: this.tickManager.now()
})
}
}
}
async terminate (container: ContainerRecordImpl): Promise<void> {
this._containers.delete(container.record.uuid) // Remove from active container registry
this._orphanedContainers.delete(container.record.endpoint)
this._orphanedContainers.delete(container.record.uuid)
this.eventQueue.push({
added: [],
deleted: [container.record],
updated: []
agents: [],
containers: [{ container: container.record, event: NetworkEventKind.removed }]
})
const agent = this._agents.get(container.record.agentId)
agent?.containers.delete(container.record.uuid)
container.agent.containers.delete(container.record.uuid)
await agent?.api.terminate(container.record.uuid)
await container.agent.api.terminate(container.record.uuid)
}
/**
@@ -414,49 +476,60 @@ export class NetworkImpl implements Network, NetworkWithClients {
// Remove dead agents and their containers
for (const agentId of deadAgents) {
await this.processDeadAgent(agentId)
await this.processAgentRemove(agentId)
}
// Handle termination of orphaned containers
for (const container of [...this._orphanedContainers.values()]) {
void this.terminate(container).catch((err) => {
console.error(`Failed to terminate orphaned container ${container.record.uuid}: ${err.message}`)
})
for (const { container, time } of [...this._orphanedContainers.values()]) {
if (now - time > timeouts.unusedContainerTimeout * 1000) {
void this.terminate(container).catch((err) => {
console.error(`Failed to terminate orphaned container ${container.record.uuid}: ${err.message}`)
})
}
}
}
/**
* Remove a dead agent and clean up its containers
*/
private async processDeadAgent (agentId: AgentUuid): Promise<void> {
private async processAgentRemove (agentId: AgentUuid): Promise<void> {
const agent = this._agents.get(agentId)
if (agent == null) {
return
}
console.log(`Removing dead agent ${agentId} and its ${agent.containers.size} containers`)
console.log(`Removing agent ${agentId} and its ${agent.containers.size} containers`)
// Collect containers to remove
const affectedContainers: ContainerRecordImpl[] = []
for (const [containerId, containerRecord] of agent.containers.entries()) {
affectedContainers.push(containerRecord)
for (const containerId of agent.containers.values()) {
const c = this._containers.get(containerId)
if (c !== undefined) {
affectedContainers.push(c)
}
this._containers.delete(containerId)
}
// We need to clean pending ones
for (const [pid, p] of this.pending.entries()) {
if (p.agent === agentId) {
this.pending.delete(pid)
}
}
// Remove agent
this._agents.delete(agentId)
const containerEvent: ContainerEvent = {
added: [],
deleted: [],
updated: []
const containerEvent: NetworkEvent = {
agents: [{ id: agentId, kinds: agent.kinds, event: NetworkEventKind.removed }],
containers: []
}
// We need to add requests for all used containers
for (const container of affectedContainers) {
this._orphanedContainers.delete(container.record.endpoint)
this._orphanedContainers.delete(container.record.uuid)
// We just send container is deleted, so clients should re-request whem again.
containerEvent.deleted.push(container.record)
containerEvent.containers.push({ container: container.record, event: NetworkEventKind.removed })
}
if (affectedContainers.length > 0) {
this.eventQueue.push(containerEvent)
packages/core/src/utils.ts
+7
View File
@@ -29,6 +29,8 @@ export class TickManagerImpl implements TickManager {
tickListeners = new Map<number, (() => void)[]>()
started: boolean = false
constructor (readonly tps: number) {
if (tps > 1000 || tps < 1) {
throw new Error('Ticks per second has an invalid value: must be >= 1 && <= 1000')
@@ -88,6 +90,10 @@ export class TickManagerImpl implements TickManager {
stop: () => void = () => {}
start (): void {
if (this.started) {
return
}
this.started = true
const to = setInterval(
() => {
this.tick().catch((err) => {
@@ -97,6 +103,7 @@ export class TickManagerImpl implements TickManager {
Math.round(1000 / this.tps)
)
this.stop = () => {
this.started = false
clearInterval(to)
}
}
packages/server/src/__test__/network.spec.ts
+16 -7
View File
@@ -2,6 +2,7 @@ import {
AgentImpl,
composeCID,
containerOnAgentEndpointRef,
containerUuid,
EndpointKind,
NetworkImpl,
parseEndpointRef,
@@ -29,13 +30,19 @@ const kinds = {
function createAgent1 (tickMgr: TickManagerImpl, networkClient: NetworkClient): AgentImpl {
const agent: AgentImpl = new AgentImpl(agents.agent1, {
[kinds.session]: async (uuid) => {
return [new DummySessionContainer(), containerOnAgentEndpointRef(agent.endpoint as AgentEndpointRef, uuid)]
[kinds.session]: async (options) => {
const uuid = options.uuid ?? containerUuid()
return {
uuid,
container: new DummySessionContainer(),
endpoint: containerOnAgentEndpointRef(agent.endpoint as AgentEndpointRef, uuid)
}
},
[kinds.workspace]: async (uuid) => {
[kinds.workspace]: async (options) => {
const uuid = options.uuid ?? containerUuid()
const container = new DummyWorkspaceContainer(uuid, agent.uuid, networkClient)
const endpoint = await container.start(tickMgr)
return [container, endpoint]
return { uuid, container, endpoint }
}
})
return agent
@@ -59,6 +66,7 @@ describe('check network server is working fine', () => {
await networkClient.close()
await network.close()
tickMgr.stop()
})
it('check routed connection and requests', async () => {
@@ -83,10 +91,10 @@ describe('check network server is working fine', () => {
const _agents = await networkClient.agents()
expect(_agents.length).toEqual(1)
expect(_agents[0].agentId).toEqual(agents.agent1)
expect(_agents[0].containers.length).toEqual(0)
expect(_agents[0].containers).toEqual(0)
// Start a new container and check if messaging works
const containerRef = await networkClient.get(composeCID('session', 'user1'), { kind: kinds.session })
const containerRef = await networkClient.get(kinds.session, { uuid: composeCID('session', 'user1') })
const data = parseEndpointRef(containerRef.endpoint)
expect(data.kind).toEqual(EndpointKind.routed)
@@ -113,8 +121,9 @@ describe('check network server is working fine', () => {
expect(events.length).toEqual(1)
expect(events[0]).toEqual('event')
await agentServer.close()
await networkClient.close()
await agentServer.close()
await network.close()
tickMgr.stop()
})
})
packages/server/src/server.ts
+8 -6
View File
@@ -6,7 +6,7 @@ import {
type ContainerEndpointRef,
type ContainerKind,
type ContainerRecord,
type ContainerRequest,
type GetOptions,
type ContainerUuid,
type Network,
type NetworkAgent,
@@ -25,8 +25,8 @@ class AgentCallbackHandler implements NetworkAgent {
readonly client: ClientUuid
) {}
async get (uuid: ContainerUuid, request: ContainerRequest): Promise<ContainerEndpointRef> {
return await this.rpcServer.request(this.client, opNames.getContainer, [this.uuid, [uuid, request]])
async get (kind: ContainerKind, request: GetOptions): Promise<[ContainerUuid, ContainerEndpointRef]> {
return await this.rpcServer.request(this.client, opNames.getContainer, [this.uuid, [kind, request]])
}
async list (kind?: ContainerKind): Promise<ContainerRecord[]> {
@@ -92,9 +92,9 @@ export class NetworkServer implements BackRPCServerHandler<ClientUuid> {
break
}
case opNames.getContainer: {
const uuid: ContainerUuid = params.uuid
const request: ContainerRequest = params.request
await send(await this.network.get(client, uuid, request))
const kind: ContainerKind = params.kind
const request: GetOptions = params.request
await send(await this.network.get(client, kind, request))
break
}
case opNames.releaseContainer: {
@@ -164,6 +164,7 @@ export class NetworkServer implements BackRPCServerHandler<ClientUuid> {
new AgentCallbackHandler(server, agentUuid, endpoint, kinds, client)
)
this.network.mapAgent(client, agentUuid)
console.log('Client registered agent', agentUuid, 'with containers', containers.length, 'kinds', kinds)
await send(res)
}
@@ -175,6 +176,7 @@ export class NetworkServer implements BackRPCServerHandler<ClientUuid> {
const agentUuid: AgentUuid = params.uuid
await this.network.unregister(agentUuid)
this.network.unmapAgent(client, agentUuid)
console.log('Client unregistered agent', agentUuid)
await send('ok')
}
}
pods/network-pod/Dockerfile
+1 -1
View File
@@ -1,4 +1,4 @@
FROM hardcoreeng/base:v20250310
FROM hardcoreeng/base-slim:latest
WORKDIR /usr/src/app
pods/network-tool/Dockerfile
+1 -1
View File
@@ -1,4 +1,4 @@
FROM hardcoreeng/base:v20250310
FROM hardcoreeng/base-slim:latest
WORKDIR /usr/src/app
pods/network-tool/bench_cli.md
+17
View File
@@ -0,0 +1,17 @@
## Agent
```bash
rushx run bench-agent -n localhost:37371
```
## List agents
```bash
rushx run list-agents -n localhost:37371
```
# Client
```bash
rushx run bench -n localhost:37371
```
pods/network-tool/package.json
+1
View File
@@ -22,6 +22,7 @@
"docker:abuild": "docker build -t hardcoreeng/network-tool . --platform=linux/arm64 && ../../common/scripts/docker_tag_push.sh hardcoreeng/network-tool",
"docker:staging": "../../common/scripts/docker_tag.sh hardcoreeng/network-tool staging",
"docker:push": "../../common/scripts/docker_tag.sh hardcoreeng/network-tool",
"run": "(rushx bundle || true) && node bundle/bundle.js",
"build:watch": "compile",
"format": "format src",
"test": "jest --passWithNoTests --silent --forceExit",
pods/network-tool/src/agents.ts
+1 -4
View File
@@ -13,10 +13,7 @@ export function registerAgentOperations (): void {
const agents = await client.agents()
console.log(`Active agents: ${agents.length}`)
for (const agent of agents) {
console.log(` - Agent: ${agent.agentId} at ${agent.endpoint} with ${agent.containers.length} containers\n`)
for (const container of agent.containers) {
console.log(` - Container: ${container.uuid} kind: ${container.kind} endpoint: ${container.endpoint}\n`)
}
console.log(` - Agent: ${agent.agentId} at ${agent.endpoint} with ${agent.containers} containers\n`)
}
await client.close()
pods/network-tool/src/benchmark.ts
+15 -4
View File
@@ -1,12 +1,14 @@
import { createAgent, createNetworkClient } from '@hcengineering/network-client'
import {
containerOnAgentEndpointRef,
containerUuid,
type AgentEndpointRef,
type ClientUuid,
type Container,
type ContainerKind,
type ContainerReference,
type ContainerUuid
type ContainerUuid,
type GetOptions
} from '@hcengineering/network-core'
import { program } from 'commander'
import { addShutdownHandler, tickManager } from './utils'
@@ -53,9 +55,14 @@ export function registerBenchmark (): void {
const client = createNetworkClient(network)
const { agent, server } = await createAgent(cmd.endpoint, {
[benchmarkContainer]: async (uuid: ContainerUuid) => {
[benchmarkContainer]: async (options: GetOptions) => {
console.log('Starting bench container')
return [new BenchmarkContainer(uuid), containerOnAgentEndpointRef(agent.endpoint as AgentEndpointRef, uuid)]
const uuid = options.uuid ?? containerUuid()
return {
uuid,
container: new BenchmarkContainer(uuid),
endpoint: containerOnAgentEndpointRef(agent.endpoint as AgentEndpointRef, uuid)
}
}
})
@@ -82,12 +89,16 @@ export function registerBenchmark (): void {
.option('-e, --exit <exit>', 'Exit after <exit>. If 1 exit on end', '0')
.action(async (cmd: { network: string, count: number, requests: number, exit: number }) => {
const network = process.env.NETWORK_HOST ?? cmd.network
console.log('Benchmark agent')
const client = createNetworkClient(network)
console.log('Connected to network')
const st = tickManager.now()
const containers: ContainerReference[] = []
for (let i = 0; i < cmd.count; i++) {
const container = await client.get(`benchmark-${i}` as ContainerUuid, { kind: benchmarkContainer })
console.log('request container:' + i)
const container = await client.get(benchmarkContainer, { uuid: `benchmark-${i}` as ContainerUuid })
console.log('container obtained', container.endpoint)
containers.push(container)
}
pods/network-tool/src/request.ts
+18 -20
View File
@@ -1,13 +1,19 @@
import { createNetworkClient } from '@hcengineering/network-client'
import type { ContainerEvent, ContainerKind, ContainerUuid } from '@hcengineering/network-core'
import {
type NetworkEvent,
type ContainerKind,
type ContainerUuid,
NetworkEventKind
} from '@hcengineering/network-core'
import { program } from 'commander'
import { addShutdownHandler, tickManager } from './utils'
export function registerRequest (): void {
program
.command('request <kind> <uuid>')
.command('request <kind>')
.description('Connect to network and request a container undefinitely')
.option('-n, --network <network>', 'Network address', 'localhost:3737')
.option('-u, --uuid <uuid>', 'Container UUID to request')
.option(
'-l, --label <label>',
'Label to add to the container. Allow multiple or ; separated',
@@ -17,27 +23,19 @@ export function registerRequest (): void {
},
[]
)
.action(async (kind: string, uuid: string, cmd: { network: string, label: string[] }) => {
.action(async (kind: string, cmd: { network: string, uuid?: string, label: string[] }) => {
const network = process.env.NETWORK_HOST ?? cmd.network
const client = createNetworkClient(network)
const container = await client.get(uuid as ContainerUuid, { kind: kind as ContainerKind, labels: cmd.label })
const container = await client.get(kind as ContainerKind, { uuid: cmd.uuid as ContainerUuid, labels: cmd.label })
console.log(`Requested container ${uuid} of kind ${kind}`)
client.onContainerUpdate(async (container: ContainerEvent) => {
for (const added of container.added) {
if (added.uuid === uuid) {
console.log(`Container started: ${added.uuid} kind: ${added.kind} endpoint: ${added.endpoint}`)
}
}
for (const removed of container.deleted) {
if (removed.uuid === uuid) {
console.log(`Container removed: ${removed.uuid}`)
}
}
for (const updated of container.updated) {
if (updated.uuid === uuid) {
console.log(`Container updated: ${updated.uuid} endpoint: ${updated.endpoint}`)
console.log(`Requested container ${cmd.uuid} of kind ${kind}`)
client.onUpdate(async (event: NetworkEvent) => {
for (const added of event.containers) {
if (added.container.uuid === container.uuid) {
console.log(
`Container ${NetworkEventKind[added.event]}: ${added.container.uuid} kind: ${added.container.kind} endpoint: ${added.container.endpoint}`
)
}
}
})
@@ -45,7 +43,7 @@ export function registerRequest (): void {
// Every 5 second print a status of our requested container
const stop = tickManager.register(() => {
if (container !== undefined) {
console.log(`Container alive at for ${uuid}: ${container.endpoint}`)
console.log(`Container alive at for ${cmd.uuid}: ${container.endpoint}`)
}
}, 5)
addShutdownHandler(async () => {