44 * See License.AGPL.txt in the project root for license information.
55 */
66
7- import {
8- Disposable ,
9- DisposableCollection ,
10- HeadlessWorkspaceEvent ,
11- PrebuildWithStatus ,
12- WorkspaceInstance ,
13- } from "@gitpod/gitpod-protocol" ;
14- import { log } from "@gitpod/gitpod-protocol/lib/util/logging" ;
7+ import { HeadlessWorkspaceEvent , PrebuildWithStatus , WorkspaceInstance } from "@gitpod/gitpod-protocol" ;
158import { TraceContext } from "@gitpod/gitpod-protocol/lib/util/tracing" ;
16- import { inject , injectable } from "inversify" ;
17- import { MessageBusIntegration } from "../workspace/messagebus-integration" ;
18- import { getExperimentsClientForBackend } from "@gitpod/gitpod-protocol/lib/experiments/configcat-server" ;
199
2010export interface PrebuildUpdateListener {
2111 ( ctx : TraceContext , evt : PrebuildWithStatus ) : void ;
@@ -26,176 +16,3 @@ export interface HeadlessWorkspaceEventListener {
2616export interface WorkspaceInstanceUpdateListener {
2717 ( ctx : TraceContext , instance : WorkspaceInstance ) : void ;
2818}
29-
30- export const LocalMessageBroker = Symbol ( "LocalMessageBroker" ) ;
31- export interface LocalMessageBroker {
32- start ( ) : Promise < void > ;
33-
34- stop ( ) : Promise < void > ;
35-
36- listenForPrebuildUpdates ( projectId : string , listener : PrebuildUpdateListener ) : Disposable ;
37-
38- listenForPrebuildUpdatableEvents ( listener : HeadlessWorkspaceEventListener ) : Disposable ;
39-
40- listenForWorkspaceInstanceUpdates ( userId : string , listener : WorkspaceInstanceUpdateListener ) : Disposable ;
41- }
42-
43- /**
44- * With our current code we basically create O(ws*p) queues for every user (ws = nr of websockets connections,
45- * p = nr of projects). This already breaks production regularly, as each queue consumes memory (and in
46- * consequence: CPU) on the messagebus. This is unnecessary in 90% of the cases, as we don't use that updates anyways.
47- *
48- * The core issue here is two-fold:
49- * 1) we create a lot of websocket connections: 1 for each dashboard tab, 2 for each workspace (frontend + supervisor)
50- * 2) we currently create new queues for each websocket connection: because of how the contracts around messages between server and dashboard evolved this is not trivial to change.
51- *
52- * To mitigate this (and also to pave the path for a switch to a less complicated message distribution technology) we
53- * introduce this LocalMessageBroker. It is meant to be a 100% backwards compatible replacement so we don't need to touch the
54- * dashboard/server logic w.r.t. to expected messages.
55- * It creates one queue per topic with rabbitmq, and does the distribution internally (per/to server instance).
56- */
57- @injectable ( )
58- export class LocalRabbitMQBackedMessageBroker implements LocalMessageBroker {
59- static readonly UNDEFINED_KEY = "undefined" ;
60-
61- @inject ( MessageBusIntegration ) protected readonly messageBusIntegration : MessageBusIntegration ;
62-
63- protected prebuildUpdateListeners : Map < string , PrebuildUpdateListener [ ] > = new Map ( ) ;
64- protected headlessWorkspaceEventListeners : Map < string , HeadlessWorkspaceEventListener [ ] > = new Map ( ) ;
65- protected workspaceInstanceUpdateListeners : Map < string , WorkspaceInstanceUpdateListener [ ] > = new Map ( ) ;
66-
67- protected readonly disposables = new DisposableCollection ( ) ;
68-
69- async start ( ) {
70- this . disposables . push (
71- this . messageBusIntegration . listenForPrebuildUpdates (
72- undefined ,
73- async ( ctx : TraceContext , update : PrebuildWithStatus ) => {
74- const enabled = await this . isRedisPubSubByTypeEnabled ( "prebuild" ) ;
75- if ( enabled ) {
76- log . debug ( "[messagebus] Prebuild listener is disabled through feature flag" ) ;
77- return ;
78- }
79- TraceContext . setOWI ( ctx , { workspaceId : update . info . buildWorkspaceId } ) ;
80-
81- const listeners = this . prebuildUpdateListeners . get ( update . info . projectId ) || [ ] ;
82- for ( const l of listeners ) {
83- try {
84- l ( ctx , update ) ;
85- } catch ( err ) {
86- TraceContext . setError ( ctx , err ) ;
87- log . error (
88- { userId : update . info . userId , workspaceId : update . info . buildWorkspaceId } ,
89- "listenForPrebuildUpdates" ,
90- err ,
91- { projectId : update . info . projectId , prebuildId : update . info . id } ,
92- ) ;
93- }
94- }
95- } ,
96- ) ,
97- ) ;
98- this . disposables . push (
99- this . messageBusIntegration . listenForPrebuildUpdatableQueue (
100- async ( ctx : TraceContext , evt : HeadlessWorkspaceEvent ) => {
101- const enabled = await this . isRedisPubSubByTypeEnabled ( "prebuild-updatable" ) ;
102- if ( enabled ) {
103- log . debug ( "[messagebus] Prebuild updatable listener is disabled through feature flag" ) ;
104- return ;
105- }
106- TraceContext . setOWI ( ctx , { workspaceId : evt . workspaceID } ) ;
107-
108- const listeners =
109- this . headlessWorkspaceEventListeners . get ( LocalRabbitMQBackedMessageBroker . UNDEFINED_KEY ) || [ ] ;
110- for ( const l of listeners ) {
111- try {
112- l ( ctx , evt ) ;
113- } catch ( err ) {
114- TraceContext . setError ( ctx , err ) ;
115- log . error ( { workspaceId : evt . workspaceID } , "listenForPrebuildUpdatableQueue" , err ) ;
116- }
117- }
118- } ,
119- ) ,
120- ) ;
121- this . disposables . push (
122- this . messageBusIntegration . listenForWorkspaceInstanceUpdates (
123- undefined ,
124- async ( ctx : TraceContext , instance : WorkspaceInstance , userId : string | undefined ) => {
125- const enabled = await this . isRedisPubSubByTypeEnabled ( "workspace-instance" ) ;
126- if ( enabled ) {
127- log . debug ( "[messagebus] Workspace instance listner is disabled through feature flag" ) ;
128- return ;
129- }
130-
131- TraceContext . setOWI ( ctx , { userId, instanceId : instance . id } ) ;
132-
133- if ( ! userId ) {
134- return ;
135- }
136-
137- const listeners = this . workspaceInstanceUpdateListeners . get ( userId ) || [ ] ;
138- for ( const l of listeners ) {
139- try {
140- l ( ctx , instance ) ;
141- } catch ( err ) {
142- TraceContext . setError ( ctx , err ) ;
143- log . error ( { userId, instanceId : instance . id } , "listenForWorkspaceInstanceUpdates" , err ) ;
144- }
145- }
146- } ,
147- ) ,
148- ) ;
149- }
150-
151- async stop ( ) {
152- this . disposables . dispose ( ) ;
153- }
154-
155- listenForPrebuildUpdates ( projectId : string , listener : PrebuildUpdateListener ) : Disposable {
156- return this . doRegister ( projectId , listener , this . prebuildUpdateListeners ) ;
157- }
158-
159- listenForPrebuildUpdatableEvents ( listener : HeadlessWorkspaceEventListener ) : Disposable {
160- // we're being cheap here in re-using a map where it just needs to be a plain array.
161- return this . doRegister (
162- LocalRabbitMQBackedMessageBroker . UNDEFINED_KEY ,
163- listener ,
164- this . headlessWorkspaceEventListeners ,
165- ) ;
166- }
167-
168- listenForWorkspaceInstanceUpdates ( userId : string , listener : WorkspaceInstanceUpdateListener ) : Disposable {
169- return this . doRegister ( userId , listener , this . workspaceInstanceUpdateListeners ) ;
170- }
171-
172- protected doRegister < L > ( key : string , listener : L , listenersStore : Map < string , L [ ] > ) : Disposable {
173- let listeners = listenersStore . get ( key ) ;
174- if ( listeners === undefined ) {
175- listeners = [ ] ;
176- listenersStore . set ( key , listeners ) ;
177- }
178- listeners . push ( listener ) ;
179- return Disposable . create ( ( ) => {
180- const ls = listeners ! ;
181- const idx = ls . findIndex ( ( l ) => l === listener ) ;
182- if ( idx !== - 1 ) {
183- ls . splice ( idx , 1 ) ;
184- }
185- if ( ls . length === 0 ) {
186- listenersStore . delete ( key ) ;
187- }
188- } ) ;
189- }
190-
191- private async isRedisPubSubByTypeEnabled (
192- type : "workspace-instance" | "prebuild" | "prebuild-updatable" ,
193- ) : Promise < boolean > {
194- const enabledTypes = await getExperimentsClientForBackend ( ) . getValueAsync (
195- "enableRedisPubSubByUpdateType" ,
196- "none" ,
197- { } ,
198- ) ;
199- return enabledTypes . indexOf ( type ) >= 0 ;
200- }
201- }
0 commit comments