feat(@xen-orchestra/rest-api): expose events (#9130)

This commit is contained in:
Mathieu
2025-11-04 15:05:06 +01:00
committed by GitHub
parent 08c2923386
commit 7112c33142
13 changed files with 560 additions and 97 deletions

View File

@@ -0,0 +1,81 @@
import type { EventEmitter } from 'node:events'
import type { Subscriber } from '../events/event.class.mjs'
import type { EventType } from '../events/event.type.mjs'
export abstract class Listener {
#subscribers: Map<Subscriber['id'], { fields: '*' | string[]; subscriber: Subscriber }> = new Map()
#eventEmitter: EventEmitter
#eventCallbacks: Map<string, (...args: unknown[]) => void> = new Map()
#watchedEvent: EventType[]
constructor(eventEmitter: EventEmitter, watchedEvent: EventType[]) {
this.#eventEmitter = eventEmitter
this.#watchedEvent = watchedEvent
}
get subscribers() {
return this.#subscribers
}
get eventEmitter() {
return this.#eventEmitter
}
addSubscriber(subscriber: Subscriber, fields: '*' | string[] = '*') {
this.#subscribers.set(subscriber.id, { fields, subscriber })
if (this.#subscribers.size === 1) {
this.#watchedEvent.forEach(event => this.#addEventListener(event))
}
}
removeSubscriber(subscriberId: Subscriber['id']) {
this.#subscribers.delete(subscriberId)
if (this.#subscribers.size === 0) {
this.removeAllEventListeners()
}
}
removeAllEventListeners() {
this.#eventCallbacks.forEach((cb, event) => {
this.#eventEmitter.off(event, cb)
})
}
clear() {
this.removeAllEventListeners()
this.#eventCallbacks.clear()
this.#subscribers.clear()
this.#watchedEvent = []
}
#addEventListener(event: EventType) {
if (this.#eventCallbacks.has(event)) {
return
}
const broadcastAllSubscriber = (...args: unknown[]) => {
this.#subscribers.forEach(conf => {
if (!conf.subscriber.isAlive) {
this.removeSubscriber(conf.subscriber.id)
return
}
const data = this.handleData({ ...conf, event }, ...args)
if (data === undefined) {
return
}
conf.subscriber.broadcast(event, data)
})
}
this.#eventCallbacks.set(event, broadcastAllSubscriber)
this.#eventEmitter.on(event, broadcastAllSubscriber)
}
abstract handleData(
conf: { fields: '*' | string[]; subscriber: Subscriber; event: EventType },
...args: unknown[]
): object | undefined
}

View File

@@ -0,0 +1,165 @@
import pick from 'lodash/pick.js'
import isEqual from 'lodash/isEqual.js'
import { createLogger } from '@xen-orchestra/log'
import type { Response } from 'express'
import { EventEmitter } from 'node:events'
import { noSuchObject } from 'xo-common/api-errors.js'
import { Listener } from '../abstract-classes/listener.mjs'
import type { CollectionEventType, EventType, SubscriberId, XapiXoListenerType } from './event.type.mjs'
import type { XapiXoRecord, XoAlarm } from '@vates/types'
import type { AlarmService } from '../alarms/alarm.service.mjs'
const log = createLogger('xo:rest-api:event-service')
export class Subscriber {
#id: SubscriberId
#manager: SubscriberManager
#connection: Response
#isAlive: boolean
get id() {
return this.#id
}
get isAlive() {
return this.#isAlive
}
get connection() {
return this.#connection
}
constructor(res: Response, manager: SubscriberManager) {
this.#id = crypto.randomUUID() as SubscriberId
const headers = new Headers({
'Content-Type': 'text/event-stream',
Connection: 'keep-alive',
'Cache-Control': 'no-cache, no-transform',
})
res.setHeaders(headers)
res.on('close', () => this.clear())
manager.addSubscriber(this)
this.#connection = res
this.#manager = manager
this.#isAlive = true
}
broadcast(event: EventType | 'init', data: object) {
if (!this.#isAlive) {
log.warn('broadcast called on a subscriber that is not alive, but still in memory! Force clear and do nothing')
this.clear()
return
}
this.#connection.write(`event:${event}\n`)
this.#connection.write(`data:${JSON.stringify(data)}\n\n`)
}
clear() {
this.#isAlive = false
this.#manager.removeSubscriber(this.id)
}
}
export class XapiXoListener extends Listener {
#type: XapiXoListenerType
#alarmService?: AlarmService
constructor(type: XapiXoListenerType, eventEmitter: EventEmitter, alarmService?: AlarmService) {
super(eventEmitter, ['add', 'update', 'remove'])
this.#type = type
this.#alarmService = alarmService
}
handleData<T extends Exclude<XapiXoRecord, XoAlarm>>(
{ fields, event }: { fields: '*' | string[]; subscriber: Subscriber; event: CollectionEventType },
object: T | undefined,
previousObj?: T
): Partial<XapiXoRecord> | undefined {
let _object: Partial<XapiXoRecord> | undefined = object
let _prevObject: Partial<XapiXoRecord> | undefined = previousObj
if (this.#type === 'alarm') {
if (object?.type === 'message' && this.#alarmService?.isAlarm(object)) {
_object = this.#alarmService.parseAlarm(object)
}
if (previousObj?.type === 'message' && this.#alarmService?.isAlarm(previousObj)) {
_prevObject = this.#alarmService.parseAlarm(previousObj)
}
}
if (fields !== '*') {
if (_object !== undefined) {
_object = pick(_object, fields)
}
if (_prevObject !== undefined) {
_prevObject = pick(_prevObject, fields)
}
}
if (event === 'update' && fields !== '*' && isEqual(_object, _prevObject)) {
// if no changes from user perspective, don't send update
return
}
// if _object === undefined, this means we are on a remove event, so _prevObject will not be undefined
return _object ?? _prevObject!
}
}
export class PingListener extends Listener {
#intervalId: NodeJS.Timeout
constructor() {
super(new EventEmitter(), ['ping'])
this.#intervalId = setInterval(() => {
this.eventEmitter.emit('ping')
}, 1000 * 30)
}
handleData(): { ping: number } {
return { ping: Date.now() }
}
clear(): void {
clearInterval(this.#intervalId)
super.clear()
}
}
export class SubscriberManager {
#subscribers: Map<string, Subscriber> = new Map()
get subscribers() {
return this.#subscribers
}
getSubscriber(id: SubscriberId) {
const subscriber = this.#subscribers.get(id)
if (subscriber === undefined) {
throw noSuchObject(id, 'event')
}
return subscriber
}
addSubscriber(subscriber: Subscriber) {
this.#subscribers.set(subscriber.id, subscriber)
}
removeSubscriber(id: string) {
this.#subscribers.delete(id)
}
clear() {
this.#subscribers.forEach(subscriber => {
if (!subscriber.connection.closed) {
subscriber.connection.destroy()
}
this.#subscribers.delete(subscriber.id)
})
}
}

View File

@@ -0,0 +1,109 @@
import {
Body,
Controller,
Example,
Delete,
Get,
Middlewares,
Path,
Post,
Request,
Response,
Route,
Security,
SuccessResponse,
Tags,
} from 'tsoa'
import {
badRequestResp,
createdResp,
noContentResp,
notFoundResp,
unauthorizedResp,
} from '../open-api/common/response.common.mjs'
import { provide } from 'inversify-binding-decorators'
import { inject } from 'inversify'
import { EventService } from './event.service.mjs'
import type { AuthenticatedRequest } from '../helpers/helper.type.mjs'
import { json } from 'express'
import type { SubscriberId, XapiXoListenerType } from './event.type.mjs'
import { addSubscription } from '../open-api/oa-examples/event.oa-example.mjs'
@Route('events')
@Security('*')
@Response(badRequestResp.status, badRequestResp.description)
@Response(unauthorizedResp.status, unauthorizedResp.description)
@Tags('events')
@provide(EventController)
export class EventController extends Controller {
#eventService: EventService
constructor(@inject(EventService) eventService: EventService) {
super()
this.#eventService = eventService
}
/**
* Opens an SSE (Server-Sent Events) connection.
*
* By default, there are no active subscriptions in the stream.
* To add subscriptions, use the following endpoint:
*
* POST /rest/v0/events/:id/subscriptions
*
*
* Events you will receive:
* - **init**: The first event you will receive.
* Data: the connection ID.
*
* - **ping**: A simple event used to keep the connection alive between the server and the client.
* Data: the event timestamp.
*
* - **add**: Triggered when an object has been added.
* Data: the added object.
*
* - **update**: Triggered when an object has been updated.
* Data: the updated object.
*
* - **remove**: Triggered when an object has been removed.
* Data: the removed object.
*/
@Get('')
@SuccessResponse(200, 'OK')
openSseConnection(@Request() req: AuthenticatedRequest) {
this.#eventService.createSseSubscriber(req.res)
}
/**
* Add a subscription
*
* @example id "0d8b28c6-e9bf-4c9d-a382-3c9e0d7cfbff"
* @example body {"collection": "VM", "fields": ["id", "name_label"]}
*/
@Example(addSubscription)
@Post('{id}/subscriptions')
@Middlewares(json())
@SuccessResponse(createdResp.status, createdResp.description)
@Response(notFoundResp.status, notFoundResp.description)
addSubscription(
@Path() id: string,
@Body() body: { collection: XapiXoListenerType; fields?: '*' | string[] }
): { id: string } {
this.#eventService.addListenerFor(id as SubscriberId, { ...body, type: body.collection })
return { id: body.collection }
}
/**
* Remove a subscription
*
* @example id "0d8b28c6-e9bf-4c9d-a382-3c9e0d7cfbff"
* @example subscriptionId: "VM"
*/
@Delete('{id}/subscriptions/{subscriptionId}')
@SuccessResponse(noContentResp.status, noContentResp.description)
@Response(notFoundResp.status, notFoundResp.description)
removeSubscription(@Path() id: string, @Path() subscriptionId: XapiXoListenerType): void {
this.#eventService.removeListenerFor(id as SubscriberId, subscriptionId)
}
}

View File

@@ -0,0 +1,71 @@
import { createLogger } from '@xen-orchestra/log'
import type { Response } from 'express'
import type { Listener } from '../abstract-classes/listener.mjs'
import { PingListener, Subscriber, SubscriberManager, XapiXoListener } from './event.class.mjs'
import type { ListenerType, SubscriberId } from './event.type.mjs'
import { AlarmService } from '../alarms/alarm.service.mjs'
import type { RestApi } from '../rest-api/rest-api.mjs'
const log = createLogger('xo:rest-api:event-service')
export class EventService {
#alarmService: AlarmService
#restApi: RestApi
#listeners: Map<string, Listener> = new Map()
#subscriberManager = new SubscriberManager()
constructor(restApi: RestApi) {
process.on('SIGTERM', () => {
log.debug(`SIGTERM received, close all SSE subscribers (nb: ${this.#subscriberManager.subscribers.size})`)
this.#subscriberManager.clear()
this.#listeners.forEach(l => l.clear())
this.#listeners.clear()
})
this.#restApi = restApi
this.#alarmService = restApi.ioc.get(AlarmService)
}
#getListener(type: ListenerType): Listener {
if (!this.#listeners.has(type)) {
if (type === 'ping') {
this.#listeners.set(type, new PingListener())
} else {
const isAlarm = type === 'alarm'
// alarm is purely XO-related; it doesn't exist at the XAPI level.
// alarm is a message with parsed values. So, in the case of an alarm listener, it listens for message collection.
const ee = this.#restApi.xoApp.objects.allIndexes.type.getEventEmitterByType(isAlarm ? 'message' : type)
this.#listeners.set(type, new XapiXoListener(type, ee, isAlarm ? this.#alarmService : undefined))
}
}
return this.#listeners.get(type)!
}
createSseSubscriber(res: Response): SubscriberId {
const subscriber = new Subscriber(res, this.#subscriberManager)
subscriber.broadcast('init', { id: subscriber.id })
this.addListenerFor(subscriber.id, { type: 'ping' })
log.debug(`new SSE subscriber added: ${subscriber.id}`)
log.debug(`nb subscriber: ${this.#subscriberManager.subscribers.size}`)
return subscriber.id
}
addListenerFor(id: SubscriberId, { fields, type }: { fields?: '*' | string[]; type: ListenerType }) {
const subscriber = this.#subscriberManager.getSubscriber(id)
const listener = this.#getListener(type)
listener.addSubscriber(subscriber, fields)
}
removeListenerFor(id: SubscriberId, type: ListenerType): void {
const subscriber = this.#subscriberManager.getSubscriber(id)
const listener = this.#getListener(type)
listener.removeSubscriber(subscriber.id)
}
}

View File

@@ -0,0 +1,11 @@
import type { Branded, XapiXoRecord } from '@vates/types'
export type SubscriberId = Branded<'Subscriber'>
export type XapiXoListenerType = XapiXoRecord['type'] | 'alarm'
export type ListenerType = XapiXoListenerType | 'ping'
export type CollectionEventType = 'add' | 'update' | 'remove'
export type EventType = CollectionEventType | 'ping'

View File

@@ -13,6 +13,7 @@ import { VdiService } from '../vdis/vdi.service.mjs'
import { UserService } from '../users/user.service.mjs'
import { BackupJobService } from '../backup-jobs/backup-job.service.mjs'
import { BackupLogService } from '../backup-logs/backup-log.service.mjs'
import { EventService } from '../events/event.service.mjs'
const iocContainer = new Container()
@@ -99,6 +100,14 @@ export function setupContainer(xoApp: XoApp) {
return new BackupLogService()
})
.inSingletonScope()
iocContainer
.bind(EventService)
.toDynamicValue(ctx => {
const restApi = ctx.container.get(RestApi)
return new EventService(restApi)
})
.inSingletonScope()
}
export { iocContainer }

View File

@@ -0,0 +1,3 @@
export const addSubscription = {
id: 'VM'
}

View File

@@ -77,6 +77,15 @@ export type XoApp = {
getOptional(path: string): Record<string, string> | undefined
getOptionalDuration(path: string): number | undefined
}
objects: EventEmitter & {
allIndexes: {
type: {
getEventEmitterByType(type: XapiXoRecord['type']): EventEmitter
}
}
}
tasks: EventEmitter & {
abort(id: XoTask['id'], reason?: string): Promise<void>
clearLogs(): Promise<void>

View File

@@ -11,6 +11,10 @@
> Users must be able to say: “Nice enhancement, I'm eager to test it”
- [REST API] Expose `GET /rest/v0/events` to open an SSE connection (PR [#9130](https://github.com/vatesfr/xen-orchestra/pull/9130))
- [REST API] Expose `POST /rest/v0/events/:id/subscriptions` to add a subscription in the SSE connection (PR [#9130](https://github.com/vatesfr/xen-orchestra/pull/9130))
- [REST API] Expose `DELETE /rest/v0/events/:id/subscriptions` to remove a subscription in the SSE connection (PR [#9130](https://github.com/vatesfr/xen-orchestra/pull/9130))
- **XO 6:**
### Bug fixes
@@ -33,4 +37,7 @@
<!--packages-start-->
- @xen-orchestra/rest-api minor
- xo-collection minor
<!--packages-end-->

View File

@@ -110,6 +110,10 @@ export class Collection extends EventEmitter {
return this._size
}
get allIndexes() {
return this._indexes
}
// -----------------------------------------------------------------
// Manipulation
// -----------------------------------------------------------------

View File

@@ -1,47 +1,53 @@
import iteratee from 'lodash/iteratee'
import { EventEmitter } from 'node:events'
import clearObject from './clear-object'
import isEmpty from './is-empty'
import NotImplemented from './not-implemented'
import { ACTION_ADD, ACTION_UPDATE, ACTION_REMOVE } from './collection'
// ===================================================================
export class Index {
constructor(computeHash) {
if (computeHash) {
this.computeHash = iteratee(computeHash)
#indexType
constructor(indexType) {
if (indexType === undefined) {
throw new Error('indexType must be passed')
}
this._itemsByHash = Object.create(null)
this._keysToHash = Object.create(null)
this.#indexType = typeof indexType === 'function' ? indexType : iteratee(indexType)
this._itemsByType = Object.create(null)
this._keysToType = Object.create(null)
this._eeByType = Object.create(null)
// Bound versions of listeners.
this._onAdd = this._onAdd.bind(this)
this._onUpdate = this._onUpdate.bind(this)
this._onRemove = this._onRemove.bind(this)
this.getIndexType = this.getIndexType.bind(this)
}
// This method is used to compute the hash under which an item must
// be saved.
computeHash(value, key) {
throw new NotImplemented('this method must be overridden')
}
// Remove empty items lists.
sweep() {
const { _itemsByHash: itemsByHash } = this
for (const hash in itemsByHash) {
if (isEmpty(itemsByHash[hash])) {
delete itemsByHash[hash]
}
}
getIndexType(object) {
return this.#indexType(object)
}
// -----------------------------------------------------------------
get items() {
return this._itemsByHash
return this._itemsByType
}
getEventEmitterByType(type) {
if (type === undefined) {
throw new Error('type is required')
}
if (this._eeByType[type] === undefined) {
this._eeByType[type] = new EventEmitter()
}
return this._eeByType[type]
}
// -----------------------------------------------------------------
@@ -63,65 +69,69 @@ export class Index {
collection.removeListener(ACTION_UPDATE, this._onUpdate)
collection.removeListener(ACTION_REMOVE, this._onRemove)
clearObject(this._itemsByHash)
clearObject(this._keysToHash)
clearObject(this._itemsByType)
clearObject(this._keysToType)
}
// -----------------------------------------------------------------
_onAdd(items) {
const { computeHash, _itemsByHash: itemsByHash, _keysToHash: keysToHash } = this
const { getIndexType, _itemsByType: itemsByType, _keysToType: keysToType } = this
for (const key in items) {
const value = items[key]
const hash = computeHash(value, key)
const type = getIndexType(value)
if (hash != null) {
;(itemsByHash[hash] ||
if (type != null) {
;(itemsByType[type] ||
// FIXME: We do not use objects without prototype for now
// because it breaks Angular in xo-web, change it back when
// this is fixed.
(itemsByHash[hash] = {}))[key] = value
(itemsByType[type] = {}))[key] = value
keysToHash[key] = hash
keysToType[key] = type
this.getEventEmitterByType(type)?.emit('add', value)
}
}
}
_onUpdate(items) {
const { computeHash, _itemsByHash: itemsByHash, _keysToHash: keysToHash } = this
const { getIndexType, _itemsByType: itemsByType, _keysToType: keysToType } = this
for (const key in items) {
const value = items[key]
const prev = keysToHash[key]
const hash = computeHash(value, key)
const prevType = keysToType[key]
const previousObj = itemsByType[prevType]?.[key]
const type = getIndexType(value)
// Removes item from the previous hash's list if any.
if (prev != null) delete itemsByHash[prev][key]
// Removes item from the previous type's list if any.
if (prevType != null) delete itemsByType[prevType][key]
// Inserts item into the new hash's list if any.
if (hash != null) {
;(itemsByHash[hash] ||
// Inserts item into the new type's list if any.
if (type != null) {
;(itemsByType[type] ||
// FIXME: idem: change back to Object.create(null)
(itemsByHash[hash] = {}))[key] = value
(itemsByType[type] = {}))[key] = value
keysToHash[key] = hash
keysToType[key] = type
this.getEventEmitterByType(type)?.emit('update', value, previousObj)
} else {
delete keysToHash[key]
delete keysToType[key]
}
}
}
_onRemove(items) {
const { _itemsByHash: itemsByHash, _keysToHash: keysToHash } = this
const { _itemsByType: itemsByType, _keysToType: keysToType } = this
for (const key in items) {
const prev = keysToHash[key]
const prev = keysToType[key]
if (prev != null) {
delete itemsByHash[prev][key]
delete keysToHash[key]
this.getEventEmitterByType(prev)?.emit('remove', undefined, itemsByType[prev][key])
delete itemsByType[prev][key]
delete keysToType[key]
}
}
}

View File

@@ -158,23 +158,4 @@ describe('Index', function () {
})
})
})
describe('#sweep()', function () {
it('removes empty items lists', function () {
col.remove(item2)
return waitTicks().then(() => {
byGroup.sweep()
assert.deepEqual(JSON.parse(JSON.stringify(col.indexes)), {
byGroup: {
foo: {
[item1.id]: item1,
[item3.id]: item3,
},
},
})
})
})
})
})

View File

@@ -1,36 +1,39 @@
import iteratee from 'lodash/iteratee'
import clearObject from './clear-object'
import NotImplemented from './not-implemented'
import { ACTION_ADD, ACTION_UPDATE, ACTION_REMOVE } from './collection'
// ===================================================================
export class UniqueIndex {
constructor(computeHash) {
if (computeHash) {
this.computeHash = iteratee(computeHash)
#indexRef
constructor(indexRef) {
if (indexRef === undefined) {
throw new Error('indexRef must be passed')
}
this._itemByHash = Object.create(null)
this._keysToHash = Object.create(null)
this.#indexRef = typeof indexRef === 'function' ? indexRef : iteratee(indexRef)
this._itemByRef = Object.create(null)
this._keysToRef = Object.create(null)
// Bound versions of listeners.
this._onAdd = this._onAdd.bind(this)
this._onUpdate = this._onUpdate.bind(this)
this._onRemove = this._onRemove.bind(this)
this.getIndexRef = this.getIndexRef.bind(this)
}
// This method is used to compute the hash under which an item must
// be saved.
computeHash(value, key) {
throw new NotImplemented('this method must be overridden')
getIndexRef(object) {
return this.#indexRef(object)
}
// -----------------------------------------------------------------
get items() {
return this._itemByHash
return this._itemByRef
}
// -----------------------------------------------------------------
@@ -52,57 +55,57 @@ export class UniqueIndex {
collection.removeListener(ACTION_UPDATE, this._onUpdate)
collection.removeListener(ACTION_REMOVE, this._onRemove)
clearObject(this._itemByHash)
clearObject(this._keysToHash)
clearObject(this._itemByRef)
clearObject(this._keysToRef)
}
// -----------------------------------------------------------------
_onAdd(items) {
const { computeHash, _itemByHash: itemByHash, _keysToHash: keysToHash } = this
const { getIndexRef, _itemByRef: itemByRef, _keysToRef: keysToRef } = this
for (const key in items) {
const value = items[key]
const hash = computeHash(value, key)
const ref = getIndexRef(value)
if (hash != null) {
itemByHash[hash] = value
keysToHash[key] = hash
if (ref != null) {
itemByRef[ref] = value
keysToRef[key] = ref
}
}
}
_onUpdate(items) {
const { computeHash, _itemByHash: itemByHash, _keysToHash: keysToHash } = this
const { getIndexRef, _itemByRef: itemByRef, _keysToRef: keysToRef } = this
for (const key in items) {
const value = items[key]
const prev = keysToHash[key]
const hash = computeHash(value, key)
const prevRef = keysToRef[key]
const ref = getIndexRef(value)
// Removes item from the previous hash's list if any.
if (prev != null) delete itemByHash[prev]
// Removes item from the previous ref's list if any.
if (prevRef != null) delete itemByRef[prevRef]
// Inserts item into the new hash's list if any.
if (hash != null) {
itemByHash[hash] = value
keysToHash[key] = hash
// Inserts item into the new ref's list if any.
if (ref != null) {
itemByRef[ref] = value
keysToRef[key] = ref
} else {
delete keysToHash[key]
delete keysToRef[key]
}
}
}
_onRemove(items) {
const { _itemByHash: itemByHash, _keysToHash: keysToHash } = this
const { _itemByRef: itemByRef, _keysToRef: keysToRef } = this
for (const key in items) {
const prev = keysToHash[key]
if (prev != null) {
delete itemByHash[prev]
delete keysToHash[key]
const prevRef = keysToRef[key]
if (prevRef != null) {
delete itemByRef[prevRef]
delete keysToRef[key]
}
}
}