mohiit1502/Automata
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

feat(core): Add secrets provider reload and refactor (#7277)

Browse Source 
This PR adds a message for queue mode which triggers an external secrets
provider reload inside the workers if the configuration has changed on
the main instance.

It also refactors some of the message handler code to remove cyclic
dependencies, as well as remove unnecessary duplicate redis clients
inside services (thanks to no more cyclic deps)
This commit is contained in:
Michael Auerswald
2023-09-28 12:57:35 +02:00
committed by GitHub
parent a80abad3af
commit 53a7502d20
10 changed files with 190 additions and 86 deletions
Download Patch File Download Diff File Expand all files Collapse all files

47
packages/cli/src/services/orchestration.handler.service.ts Normal file
View File

@@ -0,0 +1,47 @@
import Container, { Service } from 'typedi';
import { RedisService } from './redis.service';
import type { RedisServicePubSubSubscriber } from './redis/RedisServicePubSubSubscriber';
import {
COMMAND_REDIS_CHANNEL,
EVENT_BUS_REDIS_CHANNEL,
WORKER_RESPONSE_REDIS_CHANNEL,
} from './redis/RedisServiceHelper';
import { handleWorkerResponseMessage } from './orchestration/handleWorkerResponseMessage';
import { handleCommandMessage } from './orchestration/handleCommandMessage';
import { MessageEventBus } from '../eventbus/MessageEventBus/MessageEventBus';
@Service()
export class OrchestrationHandlerService {
redisSubscriber: RedisServicePubSubSubscriber;
constructor(readonly redisService: RedisService) {}
async init() {
await this.initSubscriber();
}
async shutdown() {
await this.redisSubscriber?.destroy();
}
private async initSubscriber() {
this.redisSubscriber = await this.redisService.getPubSubSubscriber();
await this.redisSubscriber.subscribeToWorkerResponseChannel();
await this.redisSubscriber.subscribeToCommandChannel();
await this.redisSubscriber.subscribeToEventLog();
this.redisSubscriber.addMessageHandler(
'OrchestrationMessageReceiver',
async (channel: string, messageString: string) => {
if (channel === WORKER_RESPONSE_REDIS_CHANNEL) {
await handleWorkerResponseMessage(messageString);
} else if (channel === COMMAND_REDIS_CHANNEL) {
await handleCommandMessage(messageString);
} else if (channel === EVENT_BUS_REDIS_CHANNEL) {
await Container.get(MessageEventBus).handleRedisEventBusMessage(messageString);
}
},
);
}
}

59
packages/cli/src/services/orchestration.service.ts
View File

@@ -1,10 +1,7 @@
import { Service } from 'typedi';
import { RedisService } from './redis.service';
import type { RedisServicePubSubPublisher } from './redis/RedisServicePubSubPublisher';
import type { RedisServicePubSubSubscriber } from './redis/RedisServicePubSubSubscriber';
import { COMMAND_REDIS_CHANNEL, WORKER_RESPONSE_REDIS_CHANNEL } from './redis/RedisServiceHelper';
import { handleWorkerResponseMessage } from './orchestration/handleWorkerResponseMessage';
import { handleCommandMessage } from './orchestration/handleCommandMessage';
import config from '@/config';
@Service()
export class OrchestrationService {
@@ -12,44 +9,29 @@ export class OrchestrationService {
redisPublisher: RedisServicePubSubPublisher;
redisSubscriber: RedisServicePubSubSubscriber;
get isQueueMode() {
return config.getEnv('executions.mode') === 'queue';
}
constructor(readonly redisService: RedisService) {}
async init() {
await this.initPublisher();
await this.initSubscriber();
this.initialized = true;
}
async shutdown() {
await this.redisPublisher?.destroy();
await this.redisSubscriber?.destroy();
}
private async initPublisher() {
this.redisPublisher = await this.redisService.getPubSubPublisher();
}
private async initSubscriber() {
this.redisSubscriber = await this.redisService.getPubSubSubscriber();
await this.redisSubscriber.subscribeToWorkerResponseChannel();
await this.redisSubscriber.subscribeToCommandChannel();
this.redisSubscriber.addMessageHandler(
'OrchestrationMessageReceiver',
async (channel: string, messageString: string) => {
if (channel === WORKER_RESPONSE_REDIS_CHANNEL) {
await handleWorkerResponseMessage(messageString);
} else if (channel === COMMAND_REDIS_CHANNEL) {
await handleCommandMessage(messageString);
}
},
);
}
async getWorkerStatus(id?: string) {
if (!this.isQueueMode) {
return;
}
if (!this.initialized) {
throw new Error('OrchestrationService not initialized');
}
@@ -60,6 +42,9 @@ export class OrchestrationService {
}
async getWorkerIds() {
if (!this.isQueueMode) {
return;
}
if (!this.initialized) {
throw new Error('OrchestrationService not initialized');
}
@@ -67,4 +52,28 @@ export class OrchestrationService {
command: 'getId',
});
}
async broadcastRestartEventbusAfterDestinationUpdate() {
if (!this.isQueueMode) {
return;
}
if (!this.initialized) {
throw new Error('OrchestrationService not initialized');
}
await this.redisPublisher.publishToCommandChannel({
command: 'restartEventBus',
});
}
async broadcastReloadExternalSecretsProviders() {
if (!this.isQueueMode) {
return;
}
if (!this.initialized) {
throw new Error('OrchestrationService not initialized');
}
await this.redisPublisher.publishToCommandChannel({
command: 'reloadExternalSecretsProviders',
});
}
}

28
packages/cli/src/services/orchestration/handleCommandMessage.ts
View File

@@ -1,14 +1,23 @@
import { LoggerProxy } from 'n8n-workflow';
import { messageToRedisServiceCommandObject } from './helpers';
import config from '@/config';
import { MessageEventBus } from '../../eventbus/MessageEventBus/MessageEventBus';
import { MessageEventBus } from '@/eventbus/MessageEventBus/MessageEventBus';
import Container from 'typedi';
import { ExternalSecretsManager } from '@/ExternalSecrets/ExternalSecretsManager.ee';
import type { N8nInstanceType } from '@/Interfaces';
import { License } from '@/License';
// this function handles commands sent to the MAIN instance. the workers handle their own commands
export async function handleCommandMessage(messageString: string) {
const queueModeId = config.get('redis.queueModeId');
const instanceType = config.get('generic.instanceType') as N8nInstanceType;
const isMainInstance = instanceType === 'main';
const message = messageToRedisServiceCommandObject(messageString);
if (message) {
LoggerProxy.debug(
`RedisCommandHandler(main): Received command message ${message.command} from ${message.senderId}`,
);
if (
message.senderId === queueModeId ||
(message.targets && !message.targets.includes(queueModeId))
@@ -21,16 +30,19 @@ export async function handleCommandMessage(messageString: string) {
}
switch (message.command) {
case 'reloadLicense':
// at this point in time, only a single main instance is supported, thus this
// command _should_ never be caught currently (which is why we log a warning)
LoggerProxy.warn(
'Received command to reload license via Redis, but this should not have happened and is not supported on the main instance yet.',
);
// once multiple main instances are supported, this command should be handled
// await Container.get(License).reload();
if (isMainInstance) {
// at this point in time, only a single main instance is supported, thus this command _should_ never be caught currently
LoggerProxy.error(
'Received command to reload license via Redis, but this should not have happened and is not supported on the main instance yet.',
);
return message;
}
await Container.get(License).reload();
break;
case 'restartEventBus':
await Container.get(MessageEventBus).restart();
case 'reloadExternalSecretsProviders':
await Container.get(ExternalSecretsManager).reloadAllProviders();
default:
break;
}

10
packages/cli/src/services/redis/RedisServiceCommands.ts
View File

@@ -3,7 +3,8 @@ export type RedisServiceCommand =
| 'getId'
| 'restartEventBus'
| 'stopWorker'
| 'reloadLicense';
| 'reloadLicense'
| 'reloadExternalSecretsProviders';
/**
* An object to be sent via Redis pub/sub from the main process to the workers.
@@ -49,6 +50,13 @@ export type RedisServiceWorkerResponseObject = {
error?: string;
};
}
| {
command: 'reloadExternalSecretsProviders';
payload: {
result: 'success' | 'error';
error?: string;
};
}
| {
command: 'stopWorker';
}