Add application migration workflow.
Co-authored-by: Cursor <cursoragent@cursor.com>
This commit is contained in:
keyhan
@@ -0,0 +1,51 @@
|
||||
import { Body, Controller, Get, Param, Post, Query, Request, UseGuards } from '@nestjs/common';
|
||||
import { AuthGuard } from '@nestjs/passport';
|
||||
import { ApiBearerAuth, ApiOperation, ApiTags } from '@nestjs/swagger';
|
||||
import { ApplicationMigrationsService } from './application-migrations.service';
|
||||
import { CreateApplicationMigrationDto } from './dto/application-migration.dto';
|
||||
import { RolesGuard } from '../common/guards/roles.guard';
|
||||
import { Roles } from '../common/decorators/roles.decorator';
|
||||
import { UserRole } from '../common/enums';
|
||||
|
||||
@ApiTags('Application Migrations')
|
||||
@ApiBearerAuth()
|
||||
@Controller('application-migrations')
|
||||
@UseGuards(AuthGuard('jwt'), RolesGuard)
|
||||
@Roles(UserRole.ADMIN)
|
||||
export class ApplicationMigrationsController {
|
||||
constructor(private readonly migrationsService: ApplicationMigrationsService) {}
|
||||
|
||||
@Post('applications/:applicationId')
|
||||
@ApiOperation({ summary: 'Queue application migration to another cluster (Super Admin)' })
|
||||
async create(
|
||||
@Param('applicationId') applicationId: string,
|
||||
@Request() req: any,
|
||||
@Body() dto: CreateApplicationMigrationDto,
|
||||
) {
|
||||
return this.migrationsService.create(applicationId, req.user.id, dto);
|
||||
}
|
||||
|
||||
@Get()
|
||||
@ApiOperation({ summary: 'List application migration jobs (Super Admin)' })
|
||||
async list(@Query('applicationId') applicationId?: string) {
|
||||
return this.migrationsService.list(applicationId);
|
||||
}
|
||||
|
||||
@Get(':id')
|
||||
@ApiOperation({ summary: 'Get application migration details (Super Admin)' })
|
||||
async findOne(@Param('id') id: string) {
|
||||
return this.migrationsService.findOne(id);
|
||||
}
|
||||
|
||||
@Get(':id/events')
|
||||
@ApiOperation({ summary: 'Get application migration event log (Super Admin)' })
|
||||
async events(@Param('id') id: string) {
|
||||
return this.migrationsService.events(id);
|
||||
}
|
||||
|
||||
@Post(':id/retry')
|
||||
@ApiOperation({ summary: 'Retry failed application migration (Super Admin)' })
|
||||
async retry(@Param('id') id: string, @Request() req: any) {
|
||||
return this.migrationsService.retry(id, req.user.id);
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,25 @@
|
||||
import { Module, forwardRef } from '@nestjs/common';
|
||||
import { BullModule } from '@nestjs/bull';
|
||||
import { TypeOrmModule } from '@nestjs/typeorm';
|
||||
import { ApplicationMigrationsController } from './application-migrations.controller';
|
||||
import { ApplicationMigrationsProcessor } from './application-migrations.processor';
|
||||
import { ApplicationMigrationsService } from './application-migrations.service';
|
||||
import { ApplicationMigrationJob } from './entities/application-migration-job.entity';
|
||||
import { ApplicationMigrationEvent } from './entities/application-migration-event.entity';
|
||||
import { ApplicationsModule } from '../applications/applications.module';
|
||||
import { ClustersModule } from '../clusters/clusters.module';
|
||||
import { KubernetesModule } from '../kubernetes/kubernetes.module';
|
||||
|
||||
@Module({
|
||||
imports: [
|
||||
TypeOrmModule.forFeature([ApplicationMigrationJob, ApplicationMigrationEvent]),
|
||||
BullModule.registerQueue({ name: 'application-migrations' }),
|
||||
forwardRef(() => ApplicationsModule),
|
||||
ClustersModule,
|
||||
KubernetesModule,
|
||||
],
|
||||
controllers: [ApplicationMigrationsController],
|
||||
providers: [ApplicationMigrationsService, ApplicationMigrationsProcessor],
|
||||
exports: [ApplicationMigrationsService],
|
||||
})
|
||||
export class ApplicationMigrationsModule {}
|
||||
@@ -0,0 +1,13 @@
|
||||
import { Process, Processor } from '@nestjs/bull';
|
||||
import { Job } from 'bull';
|
||||
import { ApplicationMigrationsService } from './application-migrations.service';
|
||||
|
||||
@Processor('application-migrations')
|
||||
export class ApplicationMigrationsProcessor {
|
||||
constructor(private readonly migrationsService: ApplicationMigrationsService) {}
|
||||
|
||||
@Process('run')
|
||||
async run(job: Job<{ migrationId: string }>): Promise<void> {
|
||||
await this.migrationsService.process(job.data.migrationId);
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,263 @@
|
||||
import { BadRequestException, Injectable, NotFoundException } from '@nestjs/common';
|
||||
import { InjectQueue } from '@nestjs/bull';
|
||||
import { InjectRepository } from '@nestjs/typeorm';
|
||||
import { Queue } from 'bull';
|
||||
import { Repository } from 'typeorm';
|
||||
import { ApplicationMigrationJob } from './entities/application-migration-job.entity';
|
||||
import { ApplicationMigrationEvent } from './entities/application-migration-event.entity';
|
||||
import { CreateApplicationMigrationDto } from './dto/application-migration.dto';
|
||||
import { ApplicationsService } from '../applications/applications.service';
|
||||
import { ClustersService } from '../clusters/clusters.service';
|
||||
import { KubernetesService } from '../kubernetes/kubernetes.service';
|
||||
import { ClusterStatus } from '../common/enums';
|
||||
|
||||
@Injectable()
|
||||
export class ApplicationMigrationsService {
|
||||
constructor(
|
||||
@InjectRepository(ApplicationMigrationJob)
|
||||
private readonly jobsRepository: Repository<ApplicationMigrationJob>,
|
||||
@InjectRepository(ApplicationMigrationEvent)
|
||||
private readonly eventsRepository: Repository<ApplicationMigrationEvent>,
|
||||
@InjectQueue('application-migrations')
|
||||
private readonly queue: Queue,
|
||||
private readonly applicationsService: ApplicationsService,
|
||||
private readonly clustersService: ClustersService,
|
||||
private readonly kubernetesService: KubernetesService,
|
||||
) {}
|
||||
|
||||
async create(
|
||||
applicationId: string,
|
||||
requestedBy: string,
|
||||
dto: CreateApplicationMigrationDto,
|
||||
): Promise<ApplicationMigrationJob> {
|
||||
const app = await this.applicationsService.findOne(applicationId);
|
||||
if (!app.clusterId) {
|
||||
throw new BadRequestException('Application does not have a source cluster assignment');
|
||||
}
|
||||
if (app.clusterId === dto.targetClusterId) {
|
||||
throw new BadRequestException('Target cluster must be different from the source cluster');
|
||||
}
|
||||
|
||||
const target = await this.clustersService.findOne(dto.targetClusterId);
|
||||
if (target.status !== ClusterStatus.ACTIVE || target.healthStatus !== 'healthy') {
|
||||
throw new BadRequestException('Target cluster must be active and healthy before migration');
|
||||
}
|
||||
|
||||
const activeJob = await this.jobsRepository.findOne({
|
||||
where: [
|
||||
{ applicationId, status: 'queued' },
|
||||
{ applicationId, status: 'running' },
|
||||
{ applicationId, status: 'rolling_back' },
|
||||
],
|
||||
order: { createdAt: 'DESC' },
|
||||
});
|
||||
if (activeJob) {
|
||||
throw new BadRequestException('A migration is already queued or running for this application');
|
||||
}
|
||||
|
||||
const job = await this.jobsRepository.save(this.jobsRepository.create({
|
||||
applicationId,
|
||||
requestedBy,
|
||||
sourceClusterId: app.clusterId,
|
||||
targetClusterId: dto.targetClusterId,
|
||||
status: 'queued',
|
||||
maxAttempts: 3,
|
||||
metadata: {
|
||||
note: dto.note,
|
||||
migrateStorage: dto.migrateStorage !== false,
|
||||
},
|
||||
}));
|
||||
await this.log(job.id, 'queued', 'Migration job queued', { targetClusterId: dto.targetClusterId });
|
||||
await this.queue.add('run', { migrationId: job.id }, {
|
||||
attempts: 1,
|
||||
removeOnComplete: true,
|
||||
removeOnFail: false,
|
||||
});
|
||||
return this.findOne(job.id);
|
||||
}
|
||||
|
||||
async list(applicationId?: string): Promise<ApplicationMigrationJob[]> {
|
||||
return this.jobsRepository.find({
|
||||
where: applicationId ? { applicationId } : {},
|
||||
relations: ['application', 'sourceCluster', 'targetCluster'],
|
||||
order: { createdAt: 'DESC' },
|
||||
take: 100,
|
||||
});
|
||||
}
|
||||
|
||||
async findOne(id: string): Promise<ApplicationMigrationJob> {
|
||||
const job = await this.jobsRepository.findOne({
|
||||
where: { id },
|
||||
relations: ['application', 'sourceCluster', 'targetCluster'],
|
||||
});
|
||||
if (!job) {
|
||||
throw new NotFoundException('Migration job not found');
|
||||
}
|
||||
return job;
|
||||
}
|
||||
|
||||
async events(id: string): Promise<ApplicationMigrationEvent[]> {
|
||||
await this.findOne(id);
|
||||
return this.eventsRepository.find({
|
||||
where: { migrationId: id },
|
||||
order: { createdAt: 'ASC' },
|
||||
});
|
||||
}
|
||||
|
||||
async retry(id: string, requestedBy: string): Promise<ApplicationMigrationJob> {
|
||||
const job = await this.findOne(id);
|
||||
if (!['failed', 'rolled_back'].includes(job.status)) {
|
||||
throw new BadRequestException('Only failed or rolled back migrations can be retried');
|
||||
}
|
||||
if (job.attempts >= job.maxAttempts) {
|
||||
throw new BadRequestException('Migration retry limit reached');
|
||||
}
|
||||
|
||||
await this.jobsRepository.update(id, {
|
||||
status: 'queued',
|
||||
errorMessage: null,
|
||||
currentStep: 'queued',
|
||||
completedAt: null,
|
||||
metadata: {
|
||||
...(job.metadata || {}),
|
||||
retriedBy: requestedBy,
|
||||
retriedAt: new Date().toISOString(),
|
||||
} as any,
|
||||
});
|
||||
await this.log(id, 'queued', 'Migration retry queued', { requestedBy });
|
||||
await this.queue.add('run', { migrationId: id }, {
|
||||
attempts: 1,
|
||||
removeOnComplete: true,
|
||||
removeOnFail: false,
|
||||
});
|
||||
return this.findOne(id);
|
||||
}
|
||||
|
||||
async process(migrationId: string): Promise<void> {
|
||||
const job = await this.findOne(migrationId);
|
||||
const app = await this.applicationsService.findOne(job.applicationId);
|
||||
const sourceClusterId = job.sourceClusterId;
|
||||
const targetClusterId = job.targetClusterId;
|
||||
const sourceApp = { ...app, clusterId: sourceClusterId };
|
||||
const targetApp = { ...app, clusterId: targetClusterId };
|
||||
|
||||
await this.updateJob(job.id, 'running', 'validate-target', { attempts: job.attempts + 1, startedAt: new Date() });
|
||||
|
||||
try {
|
||||
await this.validateTargetCluster(targetClusterId);
|
||||
await this.log(job.id, 'validate-target', 'Target cluster is active and healthy', { targetClusterId });
|
||||
|
||||
await this.updateJob(job.id, 'running', 'transfer-secrets-configs');
|
||||
await this.kubernetesService.prepareApplicationMigration(sourceApp as any, targetClusterId, {
|
||||
migrateStorage: job.metadata?.migrateStorage !== false,
|
||||
log: (step, message, metadata) => this.log(job.id, step, message, metadata),
|
||||
});
|
||||
|
||||
if (!app.latestImageTag) {
|
||||
throw new BadRequestException('Application must have a deployed image before migration');
|
||||
}
|
||||
|
||||
await this.updateJob(job.id, 'running', 'deploy-target');
|
||||
await this.log(job.id, 'deploy-target', 'Deploying application on target cluster before cutover');
|
||||
const targetResources = await this.kubernetesService.deployApplication(targetApp as any, app.latestImageTag);
|
||||
|
||||
await this.updateJob(job.id, 'running', 'health-check');
|
||||
await this.kubernetesService.waitForApplicationReady(targetApp as any, 600_000);
|
||||
await this.log(job.id, 'health-check', 'Target deployment is healthy');
|
||||
|
||||
await this.updateJob(job.id, 'running', 'update-ingress');
|
||||
await this.kubernetesService.updateIngress(targetApp as any);
|
||||
await this.log(job.id, 'update-ingress', 'Ingress/load balancer updated on target cluster');
|
||||
|
||||
await this.updateJob(job.id, 'running', 'cutover');
|
||||
await this.applicationsService.updateClusterAssignment(app.id, targetClusterId, app.poolId);
|
||||
await this.log(job.id, 'cutover', 'Application traffic cut over to target cluster', {
|
||||
sourceClusterId,
|
||||
targetClusterId,
|
||||
});
|
||||
|
||||
await this.updateJob(job.id, 'running', 'cleanup-source');
|
||||
await this.kubernetesService.deleteApplication(sourceApp as any);
|
||||
await this.log(job.id, 'cleanup-source', 'Source cluster deployment removed after successful migration');
|
||||
|
||||
await this.jobsRepository.update(job.id, {
|
||||
status: 'completed',
|
||||
currentStep: 'completed',
|
||||
completedAt: new Date(),
|
||||
metadata: {
|
||||
...(job.metadata || {}),
|
||||
targetResources,
|
||||
},
|
||||
});
|
||||
await this.log(job.id, 'completed', 'Migration completed successfully');
|
||||
} catch (error: any) {
|
||||
await this.handleFailure(job.id, sourceApp, targetApp, error);
|
||||
throw error;
|
||||
}
|
||||
}
|
||||
|
||||
private async validateTargetCluster(targetClusterId: string): Promise<void> {
|
||||
const target = await this.clustersService.findOne(targetClusterId);
|
||||
if (target.status !== ClusterStatus.ACTIVE || target.healthStatus !== 'healthy') {
|
||||
throw new BadRequestException('Target cluster is not active and healthy');
|
||||
}
|
||||
}
|
||||
|
||||
private async handleFailure(jobId: string, sourceApp: any, targetApp: any, error: any): Promise<void> {
|
||||
const message = error?.message || 'Migration failed';
|
||||
await this.jobsRepository.update(jobId, {
|
||||
status: 'rolling_back',
|
||||
currentStep: 'rollback',
|
||||
errorMessage: message,
|
||||
});
|
||||
await this.log(jobId, 'rollback', 'Migration failed; rolling back target resources', { error: message }, 'error');
|
||||
|
||||
try {
|
||||
await this.kubernetesService.deleteApplication(targetApp);
|
||||
await this.applicationsService.updateClusterAssignment(sourceApp.id, sourceApp.clusterId, sourceApp.poolId);
|
||||
await this.jobsRepository.update(jobId, {
|
||||
status: 'rolled_back',
|
||||
currentStep: 'rolled_back',
|
||||
completedAt: new Date(),
|
||||
});
|
||||
await this.log(jobId, 'rolled_back', 'Rollback completed; source cluster assignment preserved');
|
||||
} catch (rollbackError: any) {
|
||||
await this.jobsRepository.update(jobId, {
|
||||
status: 'failed',
|
||||
currentStep: 'failed',
|
||||
completedAt: new Date(),
|
||||
errorMessage: `${message}; rollback failed: ${rollbackError.message}`,
|
||||
});
|
||||
await this.log(jobId, 'failed', `Rollback failed: ${rollbackError.message}`, undefined, 'error');
|
||||
}
|
||||
}
|
||||
|
||||
private async updateJob(
|
||||
id: string,
|
||||
status: ApplicationMigrationJob['status'],
|
||||
currentStep: string,
|
||||
extra: Partial<ApplicationMigrationJob> = {},
|
||||
): Promise<void> {
|
||||
await this.jobsRepository.update(id, {
|
||||
status,
|
||||
currentStep,
|
||||
...extra,
|
||||
});
|
||||
}
|
||||
|
||||
private async log(
|
||||
migrationId: string,
|
||||
step: string,
|
||||
message: string,
|
||||
metadata?: Record<string, any>,
|
||||
level: 'info' | 'warn' | 'error' = 'info',
|
||||
): Promise<void> {
|
||||
await this.eventsRepository.save(this.eventsRepository.create({
|
||||
migrationId,
|
||||
step,
|
||||
message,
|
||||
metadata,
|
||||
level,
|
||||
}));
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,18 @@
|
||||
import { ApiProperty, ApiPropertyOptional } from '@nestjs/swagger';
|
||||
import { IsBoolean, IsOptional, IsString, IsUUID } from 'class-validator';
|
||||
|
||||
export class CreateApplicationMigrationDto {
|
||||
@ApiProperty({ description: 'Destination Kubernetes cluster ID' })
|
||||
@IsUUID()
|
||||
targetClusterId: string;
|
||||
|
||||
@ApiPropertyOptional({ description: 'Optional operator note for audit logs' })
|
||||
@IsOptional()
|
||||
@IsString()
|
||||
note?: string;
|
||||
|
||||
@ApiPropertyOptional({ description: 'Copy PVC definitions and attempt provider-neutral storage preparation' })
|
||||
@IsOptional()
|
||||
@IsBoolean()
|
||||
migrateStorage?: boolean;
|
||||
}
|
||||
@@ -0,0 +1,37 @@
|
||||
import {
|
||||
Column,
|
||||
CreateDateColumn,
|
||||
Entity,
|
||||
JoinColumn,
|
||||
ManyToOne,
|
||||
PrimaryGeneratedColumn,
|
||||
} from 'typeorm';
|
||||
import { ApplicationMigrationJob } from './application-migration-job.entity';
|
||||
|
||||
@Entity('application_migration_events')
|
||||
export class ApplicationMigrationEvent {
|
||||
@PrimaryGeneratedColumn('uuid')
|
||||
id: string;
|
||||
|
||||
@Column()
|
||||
migrationId: string;
|
||||
|
||||
@ManyToOne(() => ApplicationMigrationJob, { onDelete: 'CASCADE' })
|
||||
@JoinColumn({ name: 'migrationId' })
|
||||
migration: ApplicationMigrationJob;
|
||||
|
||||
@Column()
|
||||
step: string;
|
||||
|
||||
@Column({ default: 'info' })
|
||||
level: 'info' | 'warn' | 'error';
|
||||
|
||||
@Column()
|
||||
message: string;
|
||||
|
||||
@Column({ type: 'jsonb', nullable: true })
|
||||
metadata: Record<string, any>;
|
||||
|
||||
@CreateDateColumn()
|
||||
createdAt: Date;
|
||||
}
|
||||
@@ -0,0 +1,79 @@
|
||||
import {
|
||||
Column,
|
||||
CreateDateColumn,
|
||||
Entity,
|
||||
JoinColumn,
|
||||
ManyToOne,
|
||||
PrimaryGeneratedColumn,
|
||||
UpdateDateColumn,
|
||||
} from 'typeorm';
|
||||
import { Application } from '../../applications/entities/application.entity';
|
||||
import { Cluster } from '../../clusters/entities/cluster.entity';
|
||||
|
||||
export type ApplicationMigrationStatus =
|
||||
| 'queued'
|
||||
| 'running'
|
||||
| 'completed'
|
||||
| 'failed'
|
||||
| 'rolling_back'
|
||||
| 'rolled_back';
|
||||
|
||||
@Entity('application_migration_jobs')
|
||||
export class ApplicationMigrationJob {
|
||||
@PrimaryGeneratedColumn('uuid')
|
||||
id: string;
|
||||
|
||||
@Column()
|
||||
applicationId: string;
|
||||
|
||||
@ManyToOne(() => Application, { onDelete: 'CASCADE' })
|
||||
@JoinColumn({ name: 'applicationId' })
|
||||
application: Application;
|
||||
|
||||
@Column()
|
||||
requestedBy: string;
|
||||
|
||||
@Column({ type: 'uuid' })
|
||||
sourceClusterId: string;
|
||||
|
||||
@ManyToOne(() => Cluster, { onDelete: 'RESTRICT' })
|
||||
@JoinColumn({ name: 'sourceClusterId' })
|
||||
sourceCluster: Cluster;
|
||||
|
||||
@Column({ type: 'uuid' })
|
||||
targetClusterId: string;
|
||||
|
||||
@ManyToOne(() => Cluster, { onDelete: 'RESTRICT' })
|
||||
@JoinColumn({ name: 'targetClusterId' })
|
||||
targetCluster: Cluster;
|
||||
|
||||
@Column({ default: 'queued' })
|
||||
status: ApplicationMigrationStatus;
|
||||
|
||||
@Column({ default: 0 })
|
||||
attempts: number;
|
||||
|
||||
@Column({ default: 3 })
|
||||
maxAttempts: number;
|
||||
|
||||
@Column({ type: 'varchar', nullable: true })
|
||||
currentStep: string | null;
|
||||
|
||||
@Column({ type: 'varchar', nullable: true })
|
||||
errorMessage: string | null;
|
||||
|
||||
@Column({ type: 'jsonb', nullable: true })
|
||||
metadata: Record<string, any> | null;
|
||||
|
||||
@Column({ type: 'timestamptz', nullable: true })
|
||||
startedAt: Date | null;
|
||||
|
||||
@Column({ type: 'timestamptz', nullable: true })
|
||||
completedAt: Date | null;
|
||||
|
||||
@CreateDateColumn()
|
||||
createdAt: Date;
|
||||
|
||||
@UpdateDateColumn()
|
||||
updatedAt: Date;
|
||||
}
|
||||
Reference in New Issue
Block a user