devops-platform/devops-platform-api
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

feat: 接入 Jenkins 同步 BullMQ 调度

Browse Source 
- deploy-runs: Jenkins 自动同步优先使用 BullMQ 周期调度，并保留内存定时器降级\n- integrations/redis: 复用 BullMQ 连接工厂，避免业务模块散落 Redis 参数\n- README/RTK/settings: 同步 Jenkins 自动同步的 BullMQ 配置说明
This commit is contained in:
湛兮
2026-06-12 07:06:37 +08:00
parent fd472e3346
commit 920ba4dfb7
7 changed files with 233 additions and 21 deletions
Download Patch File Download Diff File Expand all files Collapse all files
.env.example
+1 -1
View File
@@ -10,7 +10,7 @@ DATABASE_URL=mysql://devops_user:devops_password@127.0.0.1:3306/devops_platform
USE_DATABASE_READS=false
PRISMA_CONNECT_ON_BOOT=false
# Redis and queues. Used by health checks and BullMQ-backed notification outbox retry.
# Redis and queues. Used by health checks, Jenkins auto sync, and notification outbox retry.
REDIS_URL=redis://127.0.0.1:6379/0
# Jenkins integration, server-side only.
README.md
+1 -1
View File
@@ -103,7 +103,7 @@ pnpm prisma:seed
- `DATABASE_URL`：MySQL 连接地址。
- `USE_DATABASE_READS`：是否启用 Prisma 数据读写。
- `PRISMA_CONNECT_ON_BOOT`：启动时是否主动连接数据库。
- `REDIS_URL`：Redis 连接地址，`/health` 会执行真实 PING，通知 outbox 重试会优先使用 BullMQ 调度。
- `REDIS_URL`：Redis 连接地址，`/health` 会执行真实 PING，Jenkins 自动同步和通知 outbox 重试会优先使用 BullMQ 调度。
- `JENKINS_BASE_URL`、`JENKINS_USERNAME`、`JENKINS_API_TOKEN`：Jenkins 集成配置。
- `GITEA_BASE_URL`、`GITEA_TOKEN`、`GITEA_WEBHOOK_SECRET`、`GITEA_WEBHOOK_RELAY_BASE_URL`：Gitea 集成与 Jenkins relay 诊断配置。
- `NOTIFICATION_PROVIDER`、`WECOM_WEBHOOK_URL`、`FEISHU_WEBHOOK_URL`、`NOTIFICATION_WEBHOOK_URL`：通知集成配置。
RTK.md
+1 -1
View File
@@ -19,7 +19,7 @@
- `deploy-runs`：发布记录 API、状态模型和 BPMN 步骤摘要。
- `integrations/jenkins`：Jenkins API 客户端边界。
- `integrations/gitea`：Gitea API 客户端边界。
- `integrations/redis`：Redis PING 健康检查和 BullMQ 连接参数边界，通知 outbox 调度复用该模块。
- `integrations/redis`：Redis PING 健康检查和 BullMQ 连接参数边界，Jenkins 自动同步与通知 outbox 调度复用该模块。
- `notifications/wecom`：企微、飞书和通用 webhook 通知边界。
- `agent-config`：Agent 系统级配置、密钥加密和连接测试。
- `agent`：只面向运维工作流的 LLM 代理边界。
src/deploy-runs/deploy-runs.module.ts
+12 -1
View File
@@ -1,7 +1,11 @@
/**
* 发布运行模块集中装配发布执行、Jenkins 同步、项目权限和通知依赖。
*/
import { Module } from '@nestjs/common';
import { AuditModule } from '../audit/audit.module';
import { AuthModule } from '../auth/auth.module';
import { JenkinsModule } from '../integrations/jenkins/jenkins.module';
import { RedisModule } from '../integrations/redis/redis.module';
import { WeComModule } from '../notifications/wecom/wecom.module';
import { ProjectsModule } from '../projects/projects.module';
import { DeployExecutionService } from './deploy-execution.service';
@@ -11,7 +15,14 @@ import { DeployRunsService } from './deploy-runs.service';
import { JenkinsSyncSchedulerService } from './jenkins-sync-scheduler.service';
@Module({
imports: [AuditModule, AuthModule, JenkinsModule, ProjectsModule, WeComModule],
imports: [
AuditModule,
AuthModule,
JenkinsModule,
ProjectsModule,
RedisModule,
WeComModule,
],
controllers: [DeployRunsController],
providers: [
DeployExecutionService,
src/deploy-runs/jenkins-sync-scheduler.service.spec.ts
+7 -2
View File
@@ -1,3 +1,6 @@
/**
* Jenkins 自动同步调度器测试覆盖关闭、成功聚合和失败脱敏审计。
*/
import { ConfigService } from '@nestjs/config';
import { AuditService } from '../audit/audit.service';
import { EnvConfig } from '../config/env.schema';
@@ -23,7 +26,7 @@ describe('JenkinsSyncSchedulerService', () => {
createAudit(),
);
service.onModuleInit();
void service.onModuleInit();
jest.advanceTimersByTime(3000);
expect(sync.syncJenkinsRuns).not.toHaveBeenCalled();
@@ -89,7 +92,9 @@ describe('JenkinsSyncSchedulerService', () => {
});
function createConfig(
values: Partial<Record<keyof EnvConfig, string | number | boolean | undefined>>,
values: Partial<
Record<keyof EnvConfig, string | number | boolean | undefined>
>,
): ConfigService<EnvConfig, true> {
return {
get: jest.fn((key: keyof EnvConfig) => values[key]),
src/deploy-runs/jenkins-sync-scheduler.service.ts
+187 -5
View File
@@ -1,28 +1,55 @@
import { Injectable, OnModuleDestroy, OnModuleInit } from '@nestjs/common';
/**
* Jenkins 自动同步调度服务封装 BullMQ 周期任务和进程内定时器降级路径。
*/
import {
Injectable,
OnModuleDestroy,
OnModuleInit,
Optional,
} from '@nestjs/common';
import { ConfigService } from '@nestjs/config';
import { Queue, Worker } from 'bullmq';
import { AuditService } from '../audit/audit.service';
import { redactSensitive } from '../common/security/redact-sensitive';
import { EnvConfig } from '../config/env.schema';
import { BullMqConnectionFactory } from '../integrations/redis/bullmq-connection.factory';
import { DeployRunsService } from './deploy-runs.service';
type JenkinsAutoSyncJobData = {
source: 'bullmq';
scheduledAt: string;
};
const jenkinsSyncQueueName = 'jenkins-auto-sync';
const jenkinsSyncJobName = 'sync-tick';
const jenkinsSyncSchedulerId = 'jenkins-auto-sync-scheduler';
const bullMqPrefix = 'devops-platform';
const bullMqStartupTimeoutMs = 5000;
/**
* 轻量自动同步器，先复用内存发布单和现有 Jenkins 同步逻辑。
* 后续接 Redis/BullMQ 后，这里会迁移成队列 worker。
* Jenkins 自动同步器优先用 BullMQ 周期调度，Redis 不可用时降级进程内定时器。
* 真正的发布单扫描和状态推进仍复用 DeployRunsService，避免队列层持有业务状态。
*/
@Injectable()
export class JenkinsSyncSchedulerService
implements OnModuleInit, OnModuleDestroy
{
private timer: ReturnType<typeof setInterval> | undefined;
private queue: Queue<JenkinsAutoSyncJobData, void, string> | undefined;
private worker: Worker<JenkinsAutoSyncJobData, void, string> | undefined;
private running = false;
private bullMqFallbackRecorded = false;
private bullMqRuntimeIssueRecorded = false;
constructor(
private readonly config: ConfigService<EnvConfig, true>,
private readonly deployRunsService: DeployRunsService,
private readonly auditService: AuditService,
@Optional()
private readonly bullMqConnectionFactory?: BullMqConnectionFactory,
) {}
onModuleInit(): void {
async onModuleInit(): Promise<void> {
if (!this.config.get('JENKINS_AUTO_SYNC_ENABLED', { infer: true })) {
return;
}
@@ -31,19 +58,121 @@ export class JenkinsSyncSchedulerService
infer: true,
});
await this.startBullMqScheduler(intervalMs);
}
async onModuleDestroy(): Promise<void> {
this.stopMemoryScheduler();
await this.closeBullMqScheduler(false);
}
private async startBullMqScheduler(intervalMs: number): Promise<void> {
const connection =
this.bullMqConnectionFactory?.createConnectionOptions(
'jenkins-auto-sync',
);
if (!connection) {
this.startMemoryScheduler(intervalMs);
return;
}
try {
this.queue = new Queue<JenkinsAutoSyncJobData, void, string>(
jenkinsSyncQueueName,
{
connection,
prefix: bullMqPrefix,
defaultJobOptions: {
removeOnComplete: { count: 20 },
removeOnFail: { count: 20 },
},
},
);
this.worker = new Worker<JenkinsAutoSyncJobData, void, string>(
jenkinsSyncQueueName,
async () => {
await this.runOnce();
},
{
connection,
prefix: bullMqPrefix,
concurrency: 1,
removeOnComplete: { count: 20 },
removeOnFail: { count: 20 },
},
);
this.queue.on('error', (error) => {
void this.recordBullMqIssue('queue', error, 'bullmq_recovering');
});
this.worker.on('error', (error) => {
void this.recordBullMqIssue('worker', error, 'bullmq_recovering');
});
this.worker.on('failed', (_job, error) => {
void this.recordBullMqIssue('job', error, 'bullmq_recovering');
});
await this.withStartupTimeout(
Promise.all([
this.queue.waitUntilReady(),
this.worker.waitUntilReady(),
]),
);
await this.queue.upsertJobScheduler(
jenkinsSyncSchedulerId,
{ every: intervalMs },
{
name: jenkinsSyncJobName,
data: {
source: 'bullmq',
scheduledAt: new Date().toISOString(),
},
opts: {
removeOnComplete: { count: 20 },
removeOnFail: { count: 20 },
},
},
);
void this.runOnce();
} catch (error) {
await this.closeBullMqScheduler(true);
await this.recordBullMqIssue('startup', error, 'memory_timer');
this.startMemoryScheduler(intervalMs);
}
}
private startMemoryScheduler(intervalMs: number): void {
if (this.timer) {
return;
}
this.timer = setInterval(() => {
void this.runOnce();
}, intervalMs);
void this.runOnce();
}
onModuleDestroy(): void {
private stopMemoryScheduler(): void {
if (this.timer) {
clearInterval(this.timer);
this.timer = undefined;
}
}
private async closeBullMqScheduler(forceWorkerClose: boolean): Promise<void> {
const worker = this.worker;
const queue = this.queue;
this.worker = undefined;
this.queue = undefined;
await Promise.all([
worker?.close(forceWorkerClose).catch(() => undefined),
queue?.close().catch(() => undefined),
]);
}
async runOnce(): Promise<void> {
if (this.running) {
return;
@@ -78,4 +207,57 @@ export class JenkinsSyncSchedulerService
this.running = false;
}
}
private async withStartupTimeout<T>(promise: Promise<T>): Promise<T> {
return Promise.race([
promise,
new Promise<T>((_resolve, reject) => {
setTimeout(
() => reject(new Error('BullMQ Jenkins 同步调度器启动超时')),
bullMqStartupTimeoutMs,
);
}),
]);
}
private async recordBullMqIssue(
stage: 'startup' | 'queue' | 'worker' | 'job',
error: unknown,
mode: 'memory_timer' | 'bullmq_recovering',
): Promise<void> {
const recorded =
mode === 'memory_timer'
? this.bullMqFallbackRecorded
: this.bullMqRuntimeIssueRecorded;
if (recorded) {
return;
}
if (mode === 'memory_timer') {
this.bullMqFallbackRecorded = true;
} else {
this.bullMqRuntimeIssueRecorded = true;
}
const message =
error instanceof Error ? error.message : 'BullMQ Jenkins 同步调度异常';
try {
await this.auditService.record({
action:
mode === 'memory_timer'
? 'JENKINS_AUTO_SYNC_SCHEDULER_FALLBACK'
: 'JENKINS_AUTO_SYNC_SCHEDULER_DEGRADED',
resourceType: 'deploy_run',
after: {
stage,
mode,
message: redactSensitive({ message }).message,
},
});
} catch {
// 启动早期数据库可能尚不可用，调度降级不能被审计写入失败阻断。
}
}
}
src/settings/settings.service.ts
+23 -9
View File
@@ -1,3 +1,6 @@
/**
* 系统配置服务只返回集成变量名和配置状态，禁止暴露任何密钥原文。
*/
import { Injectable } from '@nestjs/common';
import { ConfigService } from '@nestjs/config';
import { EnvConfig } from '../config/env.schema';
@@ -32,16 +35,20 @@ const INTEGRATION_DEFINITIONS: IntegrationDefinition[] = [
required: ['JENKINS_BASE_URL', 'JENKINS_USERNAME', 'JENKINS_API_TOKEN'],
optional: ['JENKINS_AUTO_SYNC_ENABLED', 'JENKINS_AUTO_SYNC_INTERVAL_MS'],
variableHelp: {
JENKINS_BASE_URL:
variableHelp('Jenkins 服务端 API 地址，只能由后端服务访问。', {
JENKINS_BASE_URL: variableHelp(
'Jenkins 服务端 API 地址，只能由后端服务访问。',
{
docUrl: 'https://www.jenkins.io/doc/book/using/remote-access-api/',
}),
JENKINS_USERNAME: variableHelp('用于调用 Jenkins Remote API 的服务账号。'),
},
),
JENKINS_USERNAME: variableHelp(
'用于调用 Jenkins Remote API 的服务账号。',
),
JENKINS_API_TOKEN: variableHelp(
'Jenkins 服务账号 API token，只能保存在服务端环境变量或后续加密密钥表。',
),
JENKINS_AUTO_SYNC_ENABLED: variableHelp(
'是否启用轻量自动同步器，自动扫描 queued/running 发布单。',
'是否启用 Jenkins 自动同步器；配置 REDIS_URL 时使用 BullMQ 调度，否则降级进程内定时器。',
{ example: 'true' },
),
JENKINS_AUTO_SYNC_INTERVAL_MS: variableHelp(
@@ -57,7 +64,9 @@ const INTEGRATION_DEFINITIONS: IntegrationDefinition[] = [
optional: ['GITEA_WEBHOOK_SECRET', 'GITEA_WEBHOOK_RELAY_BASE_URL'],
note: 'GITEA_WEBHOOK_SECRET 用于开启 webhook 签名校验；GITEA_WEBHOOK_RELAY_BASE_URL 用于校验仓库 push hook 是否已接入 Jenkins relay。',
variableHelp: {
GITEA_BASE_URL: variableHelp('Gitea 服务端 API 地址，用于读取仓库、分支和 tag。'),
GITEA_BASE_URL: variableHelp(
'Gitea 服务端 API 地址，用于读取仓库、分支和 tag。',
),
GITEA_TOKEN: variableHelp(
'Gitea API token，只能由后端持有，用于读取 refs 和后续仓库元数据。',
{ docUrl: 'https://docs.gitea.com/development/api-usage' },
@@ -108,9 +117,12 @@ const INTEGRATION_DEFINITIONS: IntegrationDefinition[] = [
NOTIFICATION_WEBHOOK_URL: variableHelp(
'内部通知网关或通用 webhook 的服务端地址，适合接自建消息中转服务。',
),
DEVOPS_PUBLIC_URL: variableHelp('通知消息中跳转回 DevOps 平台的公开访问地址。', {
DEVOPS_PUBLIC_URL: variableHelp(
'通知消息中跳转回 DevOps 平台的公开访问地址。',
{
example: 'https://devops.mrzhan.top',
}),
},
),
NOTIFICATION_OUTBOX_RETRY_ENABLED: variableHelp(
'是否启用轻量通知 outbox 重试器，失败通知会按 nextAttemptAt 自动重投。',
{ example: 'true' },
@@ -236,7 +248,9 @@ export class SettingsService {
private toNotificationIntegrationStatus(
definition: IntegrationDefinition,
): IntegrationConfigStatus {
const provider = this.getConfigValue('NOTIFICATION_PROVIDER')?.toLowerCase();
const provider = this.getConfigValue(
'NOTIFICATION_PROVIDER',
)?.toLowerCase();
const providerWebhook = this.notificationWebhookKey(provider);
const required: EnvStringKey[] = providerWebhook
? ['NOTIFICATION_PROVIDER', providerWebhook]