From fd27821bebe91bfc68ba981b5b56fb5e63c6b4e1 Mon Sep 17 00:00:00 2001 From: Eric Allam Date: Thu, 4 Jul 2024 16:20:21 +0100 Subject: [PATCH] Implement ability to disable org concurrency --- apps/webapp/app/v3/marqs/index.server.ts | 93 ++++++++++++++++--- .../app/v3/marqs/marqsKeyProducer.server.ts | 9 +- apps/webapp/app/v3/marqs/types.ts | 1 + 3 files changed, 89 insertions(+), 14 deletions(-) diff --git a/apps/webapp/app/v3/marqs/index.server.ts b/apps/webapp/app/v3/marqs/index.server.ts index ddec419ed..670ee6a0c 100644 --- a/apps/webapp/app/v3/marqs/index.server.ts +++ b/apps/webapp/app/v3/marqs/index.server.ts @@ -231,7 +231,7 @@ export class MarQS { const messageQueue = await this.#getRandomQueueFromParentQueue( parentQueue, this.options.envQueuePriorityStrategy, - (queue) => this.#calculateMessageQueueCapacities(queue), + (queue) => this.#calculateMessageQueueCapacities(queue, { checkForDisabled: false }), env.id ); @@ -345,7 +345,7 @@ export class MarQS { const messageQueue = await this.#getRandomQueueFromParentQueue( parentQueue, this.options.queuePriorityStrategy, - (queue) => this.#calculateMessageQueueCapacities(queue), + (queue) => this.#calculateMessageQueueCapacities(queue, { checkForDisabled: true }), consumerId ); @@ -737,7 +737,7 @@ export class MarQS { return queueScores; } - async #calculateMessageQueueCapacities(queue: string) { + async #calculateMessageQueueCapacities(queue: string, options?: { checkForDisabled?: boolean }) { return await this.#callCalculateMessageCapacities({ currentConcurrencyKey: this.keys.currentConcurrencyKeyFromQueue(queue), currentEnvConcurrencyKey: this.keys.envCurrentConcurrencyKeyFromQueue(queue), @@ -745,6 +745,9 @@ export class MarQS { concurrencyLimitKey: this.keys.concurrencyLimitKeyFromQueue(queue), envConcurrencyLimitKey: this.keys.envConcurrencyLimitKeyFromQueue(queue), orgConcurrencyLimitKey: this.keys.orgConcurrencyLimitKeyFromQueue(queue), + disabledConcurrencyLimitKey: options?.checkForDisabled + ? this.keys.disabledConcurrencyLimitKeyFromQueue(queue) + : undefined, }); } @@ -1102,6 +1105,7 @@ export class MarQS { concurrencyLimitKey, envConcurrencyLimitKey, orgConcurrencyLimitKey, + disabledConcurrencyLimitKey, }: { currentConcurrencyKey: string; currentEnvConcurrencyKey: string; @@ -1109,17 +1113,30 @@ export class MarQS { concurrencyLimitKey: string; envConcurrencyLimitKey: string; orgConcurrencyLimitKey: string; + disabledConcurrencyLimitKey: string | undefined; }): Promise { - const capacities = await this.redis.calculateMessageQueueCapacities( - currentConcurrencyKey, - currentEnvConcurrencyKey, - currentOrgConcurrencyKey, - concurrencyLimitKey, - envConcurrencyLimitKey, - orgConcurrencyLimitKey, - String(this.options.defaultEnvConcurrency), - String(this.options.defaultOrgConcurrency) - ); + const capacities = disabledConcurrencyLimitKey + ? await this.redis.calculateMessageQueueCapacitiesWithDisabling( + currentConcurrencyKey, + currentEnvConcurrencyKey, + currentOrgConcurrencyKey, + concurrencyLimitKey, + envConcurrencyLimitKey, + orgConcurrencyLimitKey, + disabledConcurrencyLimitKey, + String(this.options.defaultEnvConcurrency), + String(this.options.defaultOrgConcurrency) + ) + : await this.redis.calculateMessageQueueCapacities( + currentConcurrencyKey, + currentEnvConcurrencyKey, + currentOrgConcurrencyKey, + concurrencyLimitKey, + envConcurrencyLimitKey, + orgConcurrencyLimitKey, + String(this.options.defaultEnvConcurrency), + String(this.options.defaultOrgConcurrency) + ); const queueCurrent = Number(capacities[0]); const envLimit = Number(capacities[3]); @@ -1419,6 +1436,43 @@ redis.call('ZADD', visibilityQueue, newVisibilityTimeout, messageId) `, }); + this.redis.defineCommand("calculateMessageQueueCapacitiesWithDisabling", { + numberOfKeys: 7, + lua: ` +-- Keys: currentConcurrencyKey, currentEnvConcurrencyKey, currentOrgConcurrencyKey, concurrencyLimitKey, envConcurrencyLimitKey, orgConcurrencyLimitKey, disabledConcurrencyLimitKey +local currentConcurrencyKey = KEYS[1] +local currentEnvConcurrencyKey = KEYS[2] +local currentOrgConcurrencyKey = KEYS[3] +local concurrencyLimitKey = KEYS[4] +local envConcurrencyLimitKey = KEYS[5] +local orgConcurrencyLimitKey = KEYS[6] +local disabledConcurrencyLimitKey = KEYS[7] + +-- Args defaultEnvConcurrencyLimit, defaultOrgConcurrencyLimit +local defaultEnvConcurrencyLimit = tonumber(ARGV[1]) +local defaultOrgConcurrencyLimit = tonumber(ARGV[2]) + +local currentOrgConcurrency = tonumber(redis.call('SCARD', currentOrgConcurrencyKey) or '0') + +-- Check if disabledConcurrencyLimitKey exists +local orgConcurrencyLimit +if redis.call('EXISTS', disabledConcurrencyLimitKey) == 1 then + orgConcurrencyLimit = 0 +else + orgConcurrencyLimit = tonumber(redis.call('GET', orgConcurrencyLimitKey) or defaultOrgConcurrencyLimit) +end + +local currentEnvConcurrency = tonumber(redis.call('SCARD', currentEnvConcurrencyKey) or '0') +local envConcurrencyLimit = tonumber(redis.call('GET', envConcurrencyLimitKey) or defaultEnvConcurrencyLimit) + +local currentConcurrency = tonumber(redis.call('SCARD', currentConcurrencyKey) or '0') +local concurrencyLimit = redis.call('GET', concurrencyLimitKey) + +-- Return current capacity and concurrency limits for the queue, env, org +return { currentConcurrency, concurrencyLimit, currentEnvConcurrency, envConcurrencyLimit, currentOrgConcurrency, orgConcurrencyLimit } + `, + }); + this.redis.defineCommand("calculateMessageQueueCapacities", { numberOfKeys: 6, lua: ` @@ -1580,6 +1634,19 @@ declare module "ioredis" { callback?: Callback ): Result; + calculateMessageQueueCapacitiesWithDisabling( + currentConcurrencyKey: string, + currentEnvConcurrencyKey: string, + currentOrgConcurrencyKey: string, + concurrencyLimitKey: string, + envConcurrencyLimitKey: string, + orgConcurrencyLimitKey: string, + disabledConcurrencyLimitKey: string, + defaultEnvConcurrencyLimit: string, + defaultOrgConcurrencyLimit: string, + callback?: Callback + ): Result; + updateGlobalConcurrencyLimits( envConcurrencyLimitKey: string, orgConcurrencyLimitKey: string, diff --git a/apps/webapp/app/v3/marqs/marqsKeyProducer.server.ts b/apps/webapp/app/v3/marqs/marqsKeyProducer.server.ts index a044db773..f40bff66c 100644 --- a/apps/webapp/app/v3/marqs/marqsKeyProducer.server.ts +++ b/apps/webapp/app/v3/marqs/marqsKeyProducer.server.ts @@ -5,6 +5,7 @@ const constants = { SHARED_QUEUE: "sharedQueue", CURRENT_CONCURRENCY_PART: "currentConcurrency", CONCURRENCY_LIMIT_PART: "concurrency", + DISABLED_CONCURRENCY_LIMIT_PART: "disabledConcurrency", ENV_PART: "env", ORG_PART: "org", QUEUE_PART: "queue", @@ -13,7 +14,7 @@ const constants = { } as const; export class MarQSShortKeyProducer implements MarQSKeyProducer { - constructor(private _prefix: string) { } + constructor(private _prefix: string) {} sharedQueueScanPattern() { return `${this._prefix}*${constants.SHARED_QUEUE}`; @@ -89,6 +90,12 @@ export class MarQSShortKeyProducer implements MarQSKeyProducer { ); } + disabledConcurrencyLimitKeyFromQueue(queue: string) { + const orgId = this.normalizeQueue(queue).split(":")[1]; + + return `${constants.ORG_PART}:${orgId}:${constants.DISABLED_CONCURRENCY_LIMIT_PART}`; + } + orgConcurrencyLimitKeyFromQueue(queue: string) { const orgId = this.normalizeQueue(queue).split(":")[1]; diff --git a/apps/webapp/app/v3/marqs/types.ts b/apps/webapp/app/v3/marqs/types.ts index f76c07a11..305d6a14e 100644 --- a/apps/webapp/app/v3/marqs/types.ts +++ b/apps/webapp/app/v3/marqs/types.ts @@ -37,6 +37,7 @@ export interface MarQSKeyProducer { queue: string, concurrencyKey?: string ): string; + disabledConcurrencyLimitKeyFromQueue(queue: string): string; orgConcurrencyLimitKeyFromQueue(queue: string): string; orgCurrentConcurrencyKeyFromQueue(queue: string): string; envConcurrencyLimitKeyFromQueue(queue: string): string;