keyhan
3435eff256
Bootstrap configures systemd-resolved for *.cluster.local, installs k3s registries.yaml for the internal registry only, removes the legacy external-registry DaemonSet, and aligns Helm REGISTRY_PULL_URL with the in-cluster registry URL. Co-authored-by: Cursor <cursoragent@cursor.com>
1556 lines
58 KiB
TypeScript
1556 lines
58 KiB
TypeScript
import {
|
|
Injectable,
|
|
NotFoundException,
|
|
Logger,
|
|
BadRequestException,
|
|
Inject,
|
|
forwardRef,
|
|
OnModuleDestroy,
|
|
OnModuleInit,
|
|
} from '@nestjs/common';
|
|
import { InjectRepository } from '@nestjs/typeorm';
|
|
import { ConfigService } from '@nestjs/config';
|
|
import { Repository, DataSource, In } from 'typeorm';
|
|
import * as k8s from '@kubernetes/client-node';
|
|
import * as crypto from 'crypto';
|
|
import { Cluster, ClusterHealthStatus } from './entities/cluster.entity';
|
|
import { ClusterPool, PoolStrategy } from './entities/cluster-pool.entity';
|
|
import { ClusterHealth } from './entities/cluster-health.entity';
|
|
import { ClusterAllocationLog } from './entities/cluster-allocation-log.entity';
|
|
import { CreateClusterDto, UpdateClusterDto } from './dto/cluster.dto';
|
|
import { CreateClusterPoolDto, UpdateClusterPoolDto } from './dto/cluster-pool.dto';
|
|
import { ClusterStatus } from '../common/enums';
|
|
import { ElasticsearchService } from '../kubernetes/elasticsearch.service';
|
|
import { RegistryService } from '../kubernetes/registry.service';
|
|
import { CreateApplicationDto } from '../applications/dto/application.dto';
|
|
import { registerKubeconfigNoProxy } from '../common/kubernetes-proxy.util';
|
|
|
|
@Injectable()
|
|
export class ClustersService implements OnModuleInit, OnModuleDestroy {
|
|
private readonly logger = new Logger(ClustersService.name);
|
|
private roundRobinIndex = 0;
|
|
private poolRoundRobinIndices = new Map<string, number>();
|
|
private weightedRoundRobinState = new Map<string, number>();
|
|
private clusterHealthCache = new Map<string, { cluster: Cluster; cachedAt: number }>();
|
|
private healthCheckTimer?: NodeJS.Timeout;
|
|
private readonly cacheTtlMs =
|
|
Number(process.env.CLUSTER_HEALTH_CACHE_TTL_MS || 60_000);
|
|
private readonly overloadedCpuThreshold =
|
|
Number(process.env.CLUSTER_OVERLOADED_CPU_THRESHOLD || 0.85);
|
|
private readonly overloadedMemoryThreshold =
|
|
Number(process.env.CLUSTER_OVERLOADED_MEMORY_THRESHOLD || 0.85);
|
|
private readonly overloadedPodThreshold =
|
|
Number(process.env.CLUSTER_OVERLOADED_POD_THRESHOLD || 0.85);
|
|
|
|
constructor(
|
|
@InjectRepository(Cluster)
|
|
private clustersRepository: Repository<Cluster>,
|
|
@InjectRepository(ClusterPool)
|
|
private poolsRepository: Repository<ClusterPool>,
|
|
@InjectRepository(ClusterHealth)
|
|
private healthRepository: Repository<ClusterHealth>,
|
|
@InjectRepository(ClusterAllocationLog)
|
|
private allocationLogsRepository: Repository<ClusterAllocationLog>,
|
|
private dataSource: DataSource,
|
|
private configService: ConfigService,
|
|
@Inject(forwardRef(() => ElasticsearchService))
|
|
private elasticsearchService: ElasticsearchService,
|
|
@Inject(forwardRef(() => RegistryService))
|
|
private registryService: RegistryService,
|
|
) {}
|
|
|
|
onModuleInit(): void {
|
|
const intervalMs = Number(process.env.CLUSTER_HEALTH_INTERVAL_MS || 120_000);
|
|
this.healthCheckTimer = setInterval(() => {
|
|
this.refreshAllClusterHealth().catch((err) => {
|
|
this.logger.warn(`Periodic cluster health refresh failed: ${err.message}`);
|
|
});
|
|
}, intervalMs);
|
|
this.healthCheckTimer.unref?.();
|
|
}
|
|
|
|
onModuleDestroy(): void {
|
|
if (this.healthCheckTimer) {
|
|
clearInterval(this.healthCheckTimer);
|
|
}
|
|
}
|
|
|
|
/**
|
|
* Test connection to a Kubernetes cluster using its kubeconfig.
|
|
* Calls the /version endpoint to verify the cluster is reachable.
|
|
*/
|
|
async testConnection(kubeconfig: string): Promise<{ connected: boolean; version?: string; error?: string }> {
|
|
try {
|
|
const decryptedKubeconfig = this.decryptKubeconfig(kubeconfig);
|
|
registerKubeconfigNoProxy(decryptedKubeconfig);
|
|
|
|
const kc = new k8s.KubeConfig();
|
|
kc.loadFromString(decryptedKubeconfig);
|
|
|
|
const versionApi = kc.makeApiClient(k8s.VersionApi);
|
|
const result = await versionApi.getCode();
|
|
const info = result.body;
|
|
|
|
this.logger.log(`Cluster connection OK: Kubernetes ${info.gitVersion}`);
|
|
return {
|
|
connected: true,
|
|
version: info.gitVersion,
|
|
};
|
|
} catch (err: any) {
|
|
const message = err?.body?.message || err?.message || 'Unknown connection error';
|
|
this.logger.warn(`Cluster connection failed: ${message}`);
|
|
return {
|
|
connected: false,
|
|
error: message,
|
|
};
|
|
}
|
|
}
|
|
|
|
async create(dto: CreateClusterDto): Promise<Cluster> {
|
|
// Validate kubeconfig by testing actual connection
|
|
const connectionTest = await this.testConnection(dto.kubeconfig);
|
|
if (!connectionTest.connected) {
|
|
throw new BadRequestException(
|
|
`Cannot connect to Kubernetes cluster: ${connectionTest.error}`,
|
|
);
|
|
}
|
|
|
|
if (dto.isDefault === true) {
|
|
const existingDefaults = await this.clustersRepository.find({ where: { isDefault: true } });
|
|
for (const c of existingDefaults) {
|
|
c.isDefault = false;
|
|
await this.clustersRepository.save(c);
|
|
}
|
|
}
|
|
|
|
const cluster = this.clustersRepository.create({
|
|
...dto,
|
|
kubeconfig: this.encryptKubeconfig(dto.kubeconfig),
|
|
status: ClusterStatus.ACTIVE, // Connection verified — mark active
|
|
healthStatus: 'healthy',
|
|
lastHealthCheckedAt: new Date(),
|
|
healthMessage: connectionTest.version ? `Kubernetes ${connectionTest.version}` : 'Connection verified',
|
|
});
|
|
const saved = await this.clustersRepository.save(cluster);
|
|
const usableCluster = this.withDecryptedKubeconfig(saved);
|
|
this.logger.log(`Cluster "${saved.name}" registered (active) — K8s ${connectionTest.version}`);
|
|
|
|
// Bootstrap the cluster with build infrastructure (namespace, registry, SA, etc.)
|
|
this.bootstrapCluster(usableCluster.kubeconfig).catch((err) => {
|
|
this.logger.error(`Failed to bootstrap cluster "${saved.name}": ${err.message}`);
|
|
});
|
|
|
|
// Deploy central logging (Elasticsearch + Kibana) via Helm
|
|
this.elasticsearchService.deploy(saved.id).catch((err) => {
|
|
this.logger.error(`Failed to deploy central logging on "${saved.name}": ${err.message}`);
|
|
});
|
|
|
|
this.recordHealthSnapshot(saved, {
|
|
status: 'healthy',
|
|
message: connectionTest.version ? `Kubernetes ${connectionTest.version}` : 'Connection verified',
|
|
}).catch((err) => this.logger.warn(`Failed to record cluster health: ${err.message}`));
|
|
|
|
return usableCluster;
|
|
}
|
|
|
|
async findAll(): Promise<Cluster[]> {
|
|
return this.clustersRepository.find({
|
|
select: [
|
|
'id',
|
|
'name',
|
|
'description',
|
|
'status',
|
|
'apiServer',
|
|
'region',
|
|
'provider',
|
|
'isDefault',
|
|
'weight',
|
|
'tags',
|
|
'healthStatus',
|
|
'lastHealthCheckedAt',
|
|
'healthMessage',
|
|
'availableResources',
|
|
'createdAt',
|
|
],
|
|
order: { createdAt: 'DESC' },
|
|
});
|
|
}
|
|
|
|
async findOne(id: string): Promise<Cluster> {
|
|
const cluster = await this.clustersRepository.findOne({ where: { id } });
|
|
if (!cluster) {
|
|
throw new NotFoundException('Cluster not found');
|
|
}
|
|
return this.withDecryptedKubeconfig(cluster);
|
|
}
|
|
|
|
async getDefault(): Promise<Cluster> {
|
|
// Prefer active default cluster; fall back to any active cluster
|
|
let cluster = await this.clustersRepository.findOne({
|
|
where: { isDefault: true, status: ClusterStatus.ACTIVE },
|
|
});
|
|
if (!cluster) {
|
|
// Fallback: pick any active cluster and promote it to default
|
|
cluster = await this.clustersRepository.findOne({
|
|
where: { status: ClusterStatus.ACTIVE },
|
|
});
|
|
if (cluster) {
|
|
cluster.isDefault = true;
|
|
await this.clustersRepository.save(cluster);
|
|
this.logger.warn(`No active default cluster — promoted "${cluster.name}" to default`);
|
|
}
|
|
}
|
|
if (!cluster) {
|
|
throw new NotFoundException('No active cluster available');
|
|
}
|
|
return this.withDecryptedKubeconfig(cluster);
|
|
}
|
|
|
|
async update(id: string, dto: UpdateClusterDto): Promise<Cluster> {
|
|
const cluster = await this.findOne(id);
|
|
|
|
// If kubeconfig is being updated, re-test connection
|
|
if (dto.kubeconfig) {
|
|
const connectionTest = await this.testConnection(dto.kubeconfig);
|
|
if (!connectionTest.connected) {
|
|
throw new BadRequestException(
|
|
`Cannot connect to Kubernetes cluster: ${connectionTest.error}`,
|
|
);
|
|
}
|
|
dto.status = ClusterStatus.ACTIVE;
|
|
dto.kubeconfig = this.encryptKubeconfig(dto.kubeconfig);
|
|
this.logger.log(`Cluster "${cluster.name}" kubeconfig updated — connection verified (K8s ${connectionTest.version})`);
|
|
|
|
// Re-bootstrap build infrastructure on the new/updated cluster
|
|
this.bootstrapCluster(this.decryptKubeconfig(dto.kubeconfig)).catch((err: any) => {
|
|
this.logger.error(`Failed to bootstrap cluster "${cluster.name}": ${err.message}`);
|
|
});
|
|
}
|
|
|
|
if (dto.isDefault === true) {
|
|
const existingDefaults = await this.clustersRepository.find({ where: { isDefault: true } });
|
|
for (const c of existingDefaults) {
|
|
if (c.id !== id) {
|
|
c.isDefault = false;
|
|
await this.clustersRepository.save(c);
|
|
}
|
|
}
|
|
}
|
|
Object.assign(cluster, dto);
|
|
cluster.kubeconfig = this.encryptKubeconfig(cluster.kubeconfig);
|
|
const saved = await this.clustersRepository.save(cluster);
|
|
return this.withDecryptedKubeconfig(saved);
|
|
}
|
|
|
|
/**
|
|
* Manually test connectivity to an existing cluster.
|
|
* Updates status to active/inactive based on result.
|
|
*/
|
|
async testClusterById(id: string): Promise<{ connected: boolean; version?: string; error?: string }> {
|
|
const cluster = await this.findOne(id);
|
|
const result = await this.testConnection(cluster.kubeconfig);
|
|
|
|
cluster.status = result.connected ? ClusterStatus.ACTIVE : ClusterStatus.INACTIVE;
|
|
cluster.healthStatus = result.connected ? 'healthy' : 'unhealthy';
|
|
cluster.lastHealthCheckedAt = new Date();
|
|
cluster.healthMessage = result.connected
|
|
? (result.version ? `Kubernetes ${result.version}` : 'Connection verified')
|
|
: result.error || 'Connection failed';
|
|
cluster.kubeconfig = this.encryptKubeconfig(cluster.kubeconfig);
|
|
await this.clustersRepository.save(cluster);
|
|
await this.recordHealthSnapshot(cluster, {
|
|
status: cluster.healthStatus,
|
|
message: cluster.healthMessage,
|
|
});
|
|
this.logger.log(`Cluster "${cluster.name}" test: ${result.connected ? 'ACTIVE' : 'INACTIVE'}`);
|
|
|
|
return result;
|
|
}
|
|
|
|
/**
|
|
* Public cluster list (no sensitive data) — for users to select a cluster.
|
|
*/
|
|
async findAllPublic(): Promise<Pick<Cluster, 'id' | 'name' | 'region' | 'provider' | 'isDefault' | 'status'>[]> {
|
|
return this.clustersRepository.find({
|
|
select: ['id', 'name', 'region', 'provider', 'isDefault', 'status'],
|
|
where: { status: ClusterStatus.ACTIVE },
|
|
order: { isDefault: 'DESC', name: 'ASC' },
|
|
});
|
|
}
|
|
|
|
async selectClusterForApplication(
|
|
dto: CreateApplicationDto,
|
|
userId: string,
|
|
options: {
|
|
excludeClusterIds?: string[];
|
|
applicationId?: string;
|
|
reason?: string;
|
|
} = {},
|
|
): Promise<{ cluster: Cluster; pool?: ClusterPool; allocationLogId: string }> {
|
|
const estimatedRequest = this.estimateApplicationRequest(dto);
|
|
let pool = dto.poolId
|
|
? await this.poolsRepository.findOne({ where: { id: dto.poolId, isActive: true } })
|
|
: null;
|
|
if (dto.poolId && !pool) {
|
|
throw new BadRequestException('Selected cluster pool is not active or does not exist');
|
|
}
|
|
if (!pool) {
|
|
pool = await this.poolsRepository.findOne({
|
|
where: { isActive: true, isDefault: true },
|
|
order: { priority: 'ASC', createdAt: 'ASC' },
|
|
}) || await this.poolsRepository.findOne({
|
|
where: { isActive: true },
|
|
order: { priority: 'ASC', createdAt: 'ASC' },
|
|
});
|
|
}
|
|
|
|
const candidates = await this.getCachedHealthyClusters(pool || undefined, options.excludeClusterIds || []);
|
|
const appCounts = await this.getAppCounts(candidates.map((cluster) => cluster.id));
|
|
const candidateScores: Record<string, any>[] = [];
|
|
const rejectionReasons: Record<string, any>[] = [];
|
|
|
|
for (const cluster of candidates) {
|
|
const rejection = this.getClusterRejectionReason(cluster, estimatedRequest);
|
|
if (rejection) {
|
|
rejectionReasons.push({ clusterId: cluster.id, clusterName: cluster.name, reason: rejection });
|
|
continue;
|
|
}
|
|
|
|
const strategy = this.resolveStrategy(pool?.strategy);
|
|
const appCount = appCounts.get(cluster.id) || 0;
|
|
const score = this.scoreCluster(cluster, estimatedRequest, appCount, strategy, (dto as any).region);
|
|
const resourceMetrics = this.getResourceMetrics(cluster, estimatedRequest, appCount);
|
|
candidateScores.push({
|
|
clusterId: cluster.id,
|
|
clusterName: cluster.name,
|
|
score,
|
|
weight: cluster.weight || 1,
|
|
appCount,
|
|
healthStatus: cluster.healthStatus,
|
|
utilization: resourceMetrics.utilization,
|
|
availableResources: cluster.availableResources || null,
|
|
});
|
|
}
|
|
|
|
if (candidateScores.length === 0) {
|
|
const log = await this.allocationLogsRepository.save(this.allocationLogsRepository.create({
|
|
userId,
|
|
poolId: pool?.id,
|
|
applicationId: options.applicationId,
|
|
selectedClusterId: null,
|
|
strategy: this.resolveStrategy(pool?.strategy),
|
|
estimatedRequest,
|
|
candidateScores,
|
|
rejectionReasons,
|
|
status: 'failed',
|
|
message: options.reason || 'No active healthy cluster has enough estimated capacity',
|
|
}));
|
|
throw new BadRequestException(
|
|
`No active healthy cluster has enough capacity for this application (allocation log: ${log.id})`,
|
|
);
|
|
}
|
|
|
|
candidateScores.sort((a, b) => b.score - a.score);
|
|
const selected = await this.findOne(candidateScores[0].clusterId);
|
|
const log = await this.allocationLogsRepository.save(this.allocationLogsRepository.create({
|
|
userId,
|
|
poolId: pool?.id,
|
|
applicationId: options.applicationId,
|
|
selectedClusterId: selected.id,
|
|
strategy: this.resolveStrategy(pool?.strategy),
|
|
estimatedRequest,
|
|
candidateScores,
|
|
rejectionReasons,
|
|
status: 'success',
|
|
message: options.reason ? `${options.reason}: selected ${selected.name}` : `Selected ${selected.name}`,
|
|
}));
|
|
|
|
this.logger.log(`Allocator selected cluster "${selected.name}" for user ${userId} (score ${candidateScores[0].score.toFixed(2)})`);
|
|
return { cluster: selected, pool: pool || undefined, allocationLogId: log.id };
|
|
}
|
|
|
|
async attachAllocationToApplication(allocationLogId: string, applicationId: string): Promise<void> {
|
|
await this.allocationLogsRepository.update({ id: allocationLogId }, { applicationId });
|
|
}
|
|
|
|
async listAllocationLogs(limit = 100): Promise<ClusterAllocationLog[]> {
|
|
return this.allocationLogsRepository.find({
|
|
relations: ['selectedCluster', 'pool'],
|
|
order: { createdAt: 'DESC' },
|
|
take: limit,
|
|
});
|
|
}
|
|
|
|
async markAllocationFailure(
|
|
applicationId: string,
|
|
clusterId: string | undefined,
|
|
message: string,
|
|
): Promise<void> {
|
|
if (clusterId) {
|
|
await this.clustersRepository.update(clusterId, {
|
|
healthStatus: 'degraded',
|
|
healthMessage: message,
|
|
lastHealthCheckedAt: new Date(),
|
|
});
|
|
this.clusterHealthCache.delete(clusterId);
|
|
}
|
|
|
|
const latest = await this.allocationLogsRepository.findOne({
|
|
where: {
|
|
applicationId,
|
|
...(clusterId ? { selectedClusterId: clusterId } : {}),
|
|
},
|
|
order: { createdAt: 'DESC' },
|
|
});
|
|
|
|
if (latest) {
|
|
await this.allocationLogsRepository.update(latest.id, {
|
|
status: 'failed',
|
|
message,
|
|
});
|
|
}
|
|
}
|
|
|
|
async chooseFallbackClusterForApplication(
|
|
app: CreateApplicationDto & { id: string; userId: string; clusterId?: string; poolId?: string },
|
|
failedClusterIds: string[],
|
|
reason: string,
|
|
): Promise<{ cluster: Cluster; pool?: ClusterPool; allocationLogId: string }> {
|
|
return this.selectClusterForApplication(app, app.userId, {
|
|
excludeClusterIds: failedClusterIds,
|
|
applicationId: app.id,
|
|
reason,
|
|
});
|
|
}
|
|
|
|
/**
|
|
* Get the optimal cluster using load-balancing strategy.
|
|
* Strategy: 'least-apps' — picks the active cluster with fewest deployed applications.
|
|
* Falls back to default cluster if only one active cluster exists.
|
|
*/
|
|
async getOptimalCluster(strategy: 'least-apps' | 'round-robin' = 'least-apps'): Promise<Cluster> {
|
|
const activeClusters = await this.clustersRepository.find({
|
|
where: { status: ClusterStatus.ACTIVE },
|
|
});
|
|
|
|
if (activeClusters.length === 0) {
|
|
throw new NotFoundException('No active clusters available');
|
|
}
|
|
|
|
if (activeClusters.length === 1) {
|
|
return activeClusters[0];
|
|
}
|
|
|
|
if (strategy === 'round-robin') {
|
|
const cluster = activeClusters[this.roundRobinIndex % activeClusters.length];
|
|
this.roundRobinIndex++;
|
|
this.logger.log(`Round-robin selected cluster "${cluster.name}" (index: ${this.roundRobinIndex - 1})`);
|
|
return cluster;
|
|
}
|
|
|
|
// least-apps: count applications per cluster
|
|
const appCounts: { clusterId: string; count: string }[] = await this.dataSource.query(`
|
|
SELECT "clusterId", COUNT(*) as count
|
|
FROM applications
|
|
WHERE "clusterId" IS NOT NULL
|
|
GROUP BY "clusterId"
|
|
`);
|
|
|
|
const countMap = new Map<string, number>();
|
|
for (const row of appCounts) {
|
|
countMap.set(row.clusterId, parseInt(row.count, 10));
|
|
}
|
|
|
|
// Sort by app count ascending (least apps first)
|
|
activeClusters.sort((a, b) => {
|
|
const countA = countMap.get(a.id) || 0;
|
|
const countB = countMap.get(b.id) || 0;
|
|
return countA - countB;
|
|
});
|
|
|
|
const selected = activeClusters[0];
|
|
const selectedCount = countMap.get(selected.id) || 0;
|
|
this.logger.log(`Least-apps selected cluster "${selected.name}" (${selectedCount} apps)`);
|
|
return selected;
|
|
}
|
|
|
|
async delete(id: string): Promise<void> {
|
|
const cluster = await this.findOne(id);
|
|
const wasDefault = cluster.isDefault;
|
|
|
|
// Reassign applications that were on this cluster to another active cluster
|
|
try {
|
|
const replacement = await this.clustersRepository.findOne({
|
|
where: { status: ClusterStatus.ACTIVE, id: undefined as any },
|
|
});
|
|
// Use raw query to exclude the deleted cluster
|
|
const activeReplacement = await this.clustersRepository
|
|
.createQueryBuilder('c')
|
|
.where('c.id != :id', { id })
|
|
.andWhere('c.status = :status', { status: ClusterStatus.ACTIVE })
|
|
.getOne();
|
|
|
|
if (activeReplacement) {
|
|
const result = await this.dataSource.query(
|
|
`UPDATE applications SET "clusterId" = $1 WHERE "clusterId" = $2`,
|
|
[activeReplacement.id, id],
|
|
);
|
|
const count = result?.[1] || 0;
|
|
if (count > 0) {
|
|
this.logger.log(`Reassigned ${count} application(s) from cluster "${cluster.name}" to "${activeReplacement.name}"`);
|
|
}
|
|
} else {
|
|
// No replacement — nullify clusterId so apps aren't orphaned with a dangling FK
|
|
await this.dataSource.query(
|
|
`UPDATE applications SET "clusterId" = NULL WHERE "clusterId" = $1`,
|
|
[id],
|
|
);
|
|
this.logger.warn(`No active replacement cluster — cleared clusterId for apps on "${cluster.name}"`);
|
|
}
|
|
} catch (e: any) {
|
|
this.logger.warn(`Failed to reassign apps from cluster "${cluster.name}": ${e.message}`);
|
|
}
|
|
|
|
await this.clustersRepository.remove(cluster);
|
|
this.logger.log(`Cluster "${cluster.name}" deleted`);
|
|
|
|
// If deleted cluster was default, promote another active cluster
|
|
if (wasDefault) {
|
|
const newDefault = await this.clustersRepository.findOne({
|
|
where: { status: ClusterStatus.ACTIVE },
|
|
});
|
|
if (newDefault) {
|
|
newDefault.isDefault = true;
|
|
await this.clustersRepository.save(newDefault);
|
|
this.logger.log(`Promoted cluster "${newDefault.name}" to default after deleting "${cluster.name}"`);
|
|
}
|
|
}
|
|
}
|
|
|
|
// ─── Cluster Pool Methods ─────────────────────────────────────────
|
|
|
|
async createPool(dto: CreateClusterPoolDto): Promise<ClusterPool> {
|
|
// Validate that all cluster IDs exist
|
|
if (dto.clusterIds.length === 0) {
|
|
throw new BadRequestException('Pool must contain at least one cluster');
|
|
}
|
|
const clusters = await this.clustersRepository.find({
|
|
where: { id: In(dto.clusterIds) },
|
|
});
|
|
if (clusters.length !== dto.clusterIds.length) {
|
|
const foundIds = clusters.map((c) => c.id);
|
|
const missingIds = dto.clusterIds.filter((id) => !foundIds.includes(id));
|
|
throw new BadRequestException(`Clusters not found: ${missingIds.join(', ')}`);
|
|
}
|
|
|
|
if (dto.isDefault === true) {
|
|
await this.poolsRepository.update({ isDefault: true }, { isDefault: false });
|
|
}
|
|
|
|
const pool = this.poolsRepository.create({
|
|
...dto,
|
|
strategy: dto.strategy || 'weighted-resource',
|
|
});
|
|
const saved = await this.poolsRepository.save(pool);
|
|
this.logger.log(`Cluster pool "${saved.name}" created with ${dto.clusterIds.length} clusters (strategy: ${dto.strategy})`);
|
|
return saved;
|
|
}
|
|
|
|
async findAllPools(): Promise<ClusterPool[]> {
|
|
return this.poolsRepository.find({ order: { isDefault: 'DESC', priority: 'ASC', createdAt: 'DESC' } });
|
|
}
|
|
|
|
/**
|
|
* Public pool list — returns pools with resolved cluster names for UI.
|
|
*/
|
|
async findAllPoolsPublic(): Promise<(ClusterPool & { clusters: Pick<Cluster, 'id' | 'name' | 'region' | 'provider' | 'status'>[] })[]> {
|
|
const pools = await this.poolsRepository.find({
|
|
where: { isActive: true },
|
|
order: { isDefault: 'DESC', priority: 'ASC', createdAt: 'DESC' },
|
|
});
|
|
|
|
const allClusterIds = [...new Set(pools.flatMap((p) => p.clusterIds))];
|
|
const clusters = allClusterIds.length > 0
|
|
? await this.clustersRepository.find({
|
|
where: { id: In(allClusterIds) },
|
|
select: ['id', 'name', 'region', 'provider', 'status'],
|
|
})
|
|
: [];
|
|
|
|
const clusterMap = new Map(clusters.map((c) => [c.id, c]));
|
|
|
|
return pools.map((pool) => ({
|
|
...pool,
|
|
clusters: pool.clusterIds
|
|
.map((id) => clusterMap.get(id))
|
|
.filter(Boolean) as Pick<Cluster, 'id' | 'name' | 'region' | 'provider' | 'status'>[],
|
|
}));
|
|
}
|
|
|
|
async findOnePool(id: string): Promise<ClusterPool> {
|
|
const pool = await this.poolsRepository.findOne({ where: { id } });
|
|
if (!pool) {
|
|
throw new NotFoundException('Cluster pool not found');
|
|
}
|
|
return pool;
|
|
}
|
|
|
|
async updatePool(id: string, dto: UpdateClusterPoolDto): Promise<ClusterPool> {
|
|
const pool = await this.findOnePool(id);
|
|
|
|
if (dto.clusterIds && dto.clusterIds.length > 0) {
|
|
const clusters = await this.clustersRepository.find({
|
|
where: { id: In(dto.clusterIds) },
|
|
});
|
|
if (clusters.length !== dto.clusterIds.length) {
|
|
const foundIds = clusters.map((c) => c.id);
|
|
const missingIds = dto.clusterIds.filter((cid) => !foundIds.includes(cid));
|
|
throw new BadRequestException(`Clusters not found: ${missingIds.join(', ')}`);
|
|
}
|
|
}
|
|
|
|
if (dto.isDefault === true) {
|
|
await this.poolsRepository.update({ isDefault: true }, { isDefault: false });
|
|
}
|
|
|
|
Object.assign(pool, dto);
|
|
return this.poolsRepository.save(pool);
|
|
}
|
|
|
|
async deletePool(id: string): Promise<void> {
|
|
const pool = await this.findOnePool(id);
|
|
await this.poolsRepository.remove(pool);
|
|
this.poolRoundRobinIndices.delete(id);
|
|
}
|
|
|
|
/**
|
|
* Get the optimal cluster from a specific pool using the pool's strategy.
|
|
* Only considers ACTIVE clusters within the pool.
|
|
*/
|
|
async getOptimalClusterFromPool(poolId: string): Promise<Cluster> {
|
|
const pool = await this.findOnePool(poolId);
|
|
if (!pool.isActive) {
|
|
throw new BadRequestException(`Pool "${pool.name}" is not active`);
|
|
}
|
|
|
|
const activeClusters = await this.clustersRepository.find({
|
|
where: {
|
|
id: In(pool.clusterIds),
|
|
status: ClusterStatus.ACTIVE,
|
|
},
|
|
});
|
|
|
|
if (activeClusters.length === 0) {
|
|
throw new NotFoundException(`No active clusters in pool "${pool.name}"`);
|
|
}
|
|
|
|
if (activeClusters.length === 1) {
|
|
return activeClusters[0];
|
|
}
|
|
|
|
if (pool.strategy === 'round-robin') {
|
|
const idx = this.poolRoundRobinIndices.get(pool.id) || 0;
|
|
const cluster = activeClusters[idx % activeClusters.length];
|
|
this.poolRoundRobinIndices.set(pool.id, idx + 1);
|
|
this.logger.log(`Pool "${pool.name}" round-robin → cluster "${cluster.name}"`);
|
|
return cluster;
|
|
}
|
|
|
|
// least-apps strategy
|
|
const appCounts: { clusterId: string; count: string }[] = await this.dataSource.query(`
|
|
SELECT "clusterId", COUNT(*) as count
|
|
FROM applications
|
|
WHERE "clusterId" = ANY($1)
|
|
GROUP BY "clusterId"
|
|
`, [pool.clusterIds]);
|
|
|
|
const countMap = new Map<string, number>();
|
|
for (const row of appCounts) {
|
|
countMap.set(row.clusterId, parseInt(row.count, 10));
|
|
}
|
|
|
|
activeClusters.sort((a, b) => {
|
|
return (countMap.get(a.id) || 0) - (countMap.get(b.id) || 0);
|
|
});
|
|
|
|
const selected = activeClusters[0];
|
|
this.logger.log(`Pool "${pool.name}" least-apps → cluster "${selected.name}" (${countMap.get(selected.id) || 0} apps)`);
|
|
return selected;
|
|
}
|
|
|
|
/**
|
|
* Get resource usage for a specific cluster — nodes, total CPU/memory, pod counts.
|
|
*/
|
|
async getClusterResources(id: string, options: { allowCached?: boolean } = {}): Promise<{
|
|
nodes: { name: string; status: string; roles: string; cpuCapacity: string; memoryCapacity: string; cpuAllocatable: string; memoryAllocatable: string; }[];
|
|
totalCpuCapacity: string;
|
|
totalMemoryCapacity: string;
|
|
totalCpuAllocatable: string;
|
|
totalMemoryAllocatable: string;
|
|
totalCpuRequested?: string;
|
|
totalMemoryRequested?: string;
|
|
cpuUtilization?: number;
|
|
memoryUtilization?: number;
|
|
podUtilization?: number;
|
|
podCapacity?: number;
|
|
podCount: number;
|
|
nodeCount: number;
|
|
readyNodeCount?: number;
|
|
appCount: number;
|
|
}> {
|
|
const cluster = await this.findOne(id);
|
|
if (
|
|
options.allowCached !== false &&
|
|
cluster.availableResources &&
|
|
cluster.lastHealthCheckedAt &&
|
|
Date.now() - new Date(cluster.lastHealthCheckedAt).getTime() < this.cacheTtlMs
|
|
) {
|
|
return cluster.availableResources as any;
|
|
}
|
|
|
|
const kc = new k8s.KubeConfig();
|
|
registerKubeconfigNoProxy(cluster.kubeconfig);
|
|
kc.loadFromString(cluster.kubeconfig);
|
|
const coreApi = kc.makeApiClient(k8s.CoreV1Api);
|
|
|
|
try {
|
|
// Get nodes
|
|
const nodesRes = await coreApi.listNode();
|
|
const nodes = nodesRes.body.items.map((node) => {
|
|
const conditions = node.status?.conditions || [];
|
|
const readyCondition = conditions.find((c) => c.type === 'Ready');
|
|
const roles = Object.keys(node.metadata?.labels || {})
|
|
.filter((l) => l.startsWith('node-role.kubernetes.io/'))
|
|
.map((l) => l.replace('node-role.kubernetes.io/', ''))
|
|
.join(', ') || 'worker';
|
|
|
|
return {
|
|
name: node.metadata?.name || 'unknown',
|
|
status: readyCondition?.status === 'True' ? 'Ready' : 'NotReady',
|
|
roles,
|
|
cpuCapacity: node.status?.capacity?.cpu || '0',
|
|
memoryCapacity: node.status?.capacity?.memory || '0',
|
|
cpuAllocatable: node.status?.allocatable?.cpu || '0',
|
|
memoryAllocatable: node.status?.allocatable?.memory || '0',
|
|
};
|
|
});
|
|
|
|
// Total capacity
|
|
let totalCpuCap = 0;
|
|
let totalMemCap = 0;
|
|
let totalCpuAlloc = 0;
|
|
let totalMemAlloc = 0;
|
|
for (const node of nodes) {
|
|
totalCpuCap += this.parseCpuToMillicores(node.cpuCapacity);
|
|
totalMemCap += this.parseMemoryToMi(node.memoryCapacity);
|
|
totalCpuAlloc += this.parseCpuToMillicores(node.cpuAllocatable);
|
|
totalMemAlloc += this.parseMemoryToMi(node.memoryAllocatable);
|
|
}
|
|
|
|
// Get all pods count
|
|
const podsRes = await coreApi.listPodForAllNamespaces();
|
|
const podCount = podsRes.body.items.length;
|
|
let totalCpuRequested = 0;
|
|
let totalMemoryRequested = 0;
|
|
for (const pod of podsRes.body.items) {
|
|
for (const container of pod.spec?.containers || []) {
|
|
totalCpuRequested += this.parseCpuToMillicores(container.resources?.requests?.cpu || '0');
|
|
totalMemoryRequested += this.parseMemoryToMi(container.resources?.requests?.memory || '0');
|
|
}
|
|
}
|
|
|
|
// Get app count for this cluster
|
|
const appCountResult = await this.dataSource.query(
|
|
`SELECT COUNT(*) as count FROM applications WHERE "clusterId" = $1`,
|
|
[id],
|
|
);
|
|
const appCount = parseInt(appCountResult[0]?.count || '0', 10);
|
|
|
|
const readyNodeCount = nodes.filter((node) => node.status === 'Ready').length;
|
|
const podCapacity = nodes.length * 110;
|
|
const resources = {
|
|
nodes,
|
|
totalCpuCapacity: `${totalCpuCap}m`,
|
|
totalMemoryCapacity: `${totalMemCap.toFixed(0)}Mi`,
|
|
totalCpuAllocatable: `${totalCpuAlloc}m`,
|
|
totalMemoryAllocatable: `${totalMemAlloc.toFixed(0)}Mi`,
|
|
totalCpuRequested: `${totalCpuRequested}m`,
|
|
totalMemoryRequested: `${totalMemoryRequested.toFixed(0)}Mi`,
|
|
cpuUtilization: this.utilizationRatio(totalCpuRequested, totalCpuAlloc),
|
|
memoryUtilization: this.utilizationRatio(totalMemoryRequested, totalMemAlloc),
|
|
podUtilization: this.utilizationRatio(podCount, podCapacity),
|
|
podCapacity,
|
|
podCount,
|
|
nodeCount: nodes.length,
|
|
readyNodeCount,
|
|
appCount,
|
|
};
|
|
const healthStatus: ClusterHealthStatus = readyNodeCount === nodes.length && nodes.length > 0
|
|
? 'healthy'
|
|
: readyNodeCount > 0
|
|
? 'degraded'
|
|
: 'unhealthy';
|
|
const healthMessage = `${readyNodeCount}/${nodes.length} nodes ready`;
|
|
|
|
await this.clustersRepository.update(id, {
|
|
healthStatus,
|
|
healthMessage,
|
|
lastHealthCheckedAt: new Date(),
|
|
availableResources: resources as any,
|
|
});
|
|
this.clusterHealthCache.set(id, {
|
|
cluster: {
|
|
...cluster,
|
|
healthStatus,
|
|
healthMessage,
|
|
lastHealthCheckedAt: new Date(),
|
|
availableResources: resources,
|
|
},
|
|
cachedAt: Date.now(),
|
|
});
|
|
await this.recordHealthSnapshot(cluster, { status: healthStatus, message: healthMessage, resources });
|
|
|
|
return resources;
|
|
} catch (err: any) {
|
|
this.logger.error(`Failed to get cluster resources for "${cluster.name}": ${err.message}`);
|
|
await this.clustersRepository.update(id, {
|
|
healthStatus: 'unhealthy',
|
|
healthMessage: err.message,
|
|
lastHealthCheckedAt: new Date(),
|
|
});
|
|
this.clusterHealthCache.delete(id);
|
|
await this.recordHealthSnapshot(cluster, { status: 'unhealthy', message: err.message });
|
|
throw new BadRequestException(`Cannot fetch resources: ${err.message}`);
|
|
}
|
|
}
|
|
|
|
/**
|
|
* Bootstrap a newly-added cluster with the build infrastructure:
|
|
* 1. cloudhost-builds namespace
|
|
* 2. kaniko-builder ServiceAccount
|
|
* 3. Docker Registry Deployment + PVC + Service (ClusterIP) + NodePort Service
|
|
* 4. registry-credentials Secret (for Kaniko docker auth)
|
|
*/
|
|
async bootstrapCluster(kubeconfig: string): Promise<void> {
|
|
const kc = new k8s.KubeConfig();
|
|
registerKubeconfigNoProxy(kubeconfig);
|
|
kc.loadFromString(kubeconfig);
|
|
const coreApi = kc.makeApiClient(k8s.CoreV1Api);
|
|
const appsApi = kc.makeApiClient(k8s.AppsV1Api);
|
|
|
|
const buildNs = this.registryService.getBuildNamespace();
|
|
const saName = this.configService.get<string>('build.serviceAccount') || 'kaniko-builder';
|
|
const registryUrl = this.registryService.getRegistryUrl();
|
|
|
|
this.logger.log(`Bootstrapping cluster — namespace: ${buildNs}`);
|
|
|
|
// ── 1. Namespace ──────────────────────────────────────────────
|
|
try {
|
|
await coreApi.readNamespace(buildNs);
|
|
this.logger.log(`Namespace "${buildNs}" already exists`);
|
|
} catch (err: any) {
|
|
if (err.statusCode === 404 || err.body?.code === 404) {
|
|
await coreApi.createNamespace({ metadata: { name: buildNs } });
|
|
this.logger.log(`Created namespace "${buildNs}"`);
|
|
} else {
|
|
throw err;
|
|
}
|
|
}
|
|
|
|
// ── 2. ServiceAccount for Kaniko ──────────────────────────────
|
|
try {
|
|
await coreApi.readNamespacedServiceAccount(saName, buildNs);
|
|
this.logger.log(`ServiceAccount "${saName}" already exists`);
|
|
} catch (err: any) {
|
|
if (err.statusCode === 404 || err.body?.code === 404) {
|
|
await coreApi.createNamespacedServiceAccount(buildNs, {
|
|
metadata: { name: saName, namespace: buildNs },
|
|
});
|
|
this.logger.log(`Created ServiceAccount "${saName}"`);
|
|
} else {
|
|
throw err;
|
|
}
|
|
}
|
|
|
|
// ── 3. Docker Registry PVC ────────────────────────────────────
|
|
const registryPvcName = 'registry-data';
|
|
try {
|
|
await coreApi.readNamespacedPersistentVolumeClaim(registryPvcName, buildNs);
|
|
this.logger.log(`PVC "${registryPvcName}" already exists`);
|
|
} catch (err: any) {
|
|
if (err.statusCode === 404 || err.body?.code === 404) {
|
|
await coreApi.createNamespacedPersistentVolumeClaim(buildNs, {
|
|
metadata: { name: registryPvcName, namespace: buildNs },
|
|
spec: {
|
|
accessModes: ['ReadWriteOnce'],
|
|
resources: { requests: { storage: '10Gi' } },
|
|
},
|
|
});
|
|
this.logger.log(`Created PVC "${registryPvcName}" (10Gi)`);
|
|
} else {
|
|
throw err;
|
|
}
|
|
}
|
|
|
|
// ── 4. Docker Registry Deployment ─────────────────────────────
|
|
const registryDeployName = 'registry';
|
|
try {
|
|
await appsApi.readNamespacedDeployment(registryDeployName, buildNs);
|
|
this.logger.log(`Deployment "${registryDeployName}" already exists`);
|
|
} catch (err: any) {
|
|
if (err.statusCode === 404 || err.body?.code === 404) {
|
|
await appsApi.createNamespacedDeployment(buildNs, {
|
|
metadata: { name: registryDeployName, namespace: buildNs, labels: { app: 'registry' } },
|
|
spec: {
|
|
replicas: 1,
|
|
selector: { matchLabels: { app: 'registry' } },
|
|
template: {
|
|
metadata: { labels: { app: 'registry' } },
|
|
spec: {
|
|
containers: [{
|
|
name: 'registry',
|
|
image: 'registry:2',
|
|
ports: [{ containerPort: 5000 }],
|
|
env: [{ name: 'REGISTRY_STORAGE_DELETE_ENABLED', value: 'true' }],
|
|
volumeMounts: [{
|
|
name: 'registry-data',
|
|
mountPath: '/var/lib/registry',
|
|
}],
|
|
resources: {
|
|
requests: { cpu: '100m', memory: '128Mi' },
|
|
limits: { cpu: '500m', memory: '512Mi' },
|
|
},
|
|
}],
|
|
volumes: [{
|
|
name: 'registry-data',
|
|
persistentVolumeClaim: { claimName: registryPvcName },
|
|
}],
|
|
},
|
|
},
|
|
},
|
|
});
|
|
this.logger.log(`Created Docker Registry Deployment`);
|
|
} else {
|
|
throw err;
|
|
}
|
|
}
|
|
|
|
// ── 5. Registry ClusterIP Service (Kaniko push + app pull) ───
|
|
const registrySvcName = 'registry';
|
|
try {
|
|
await coreApi.readNamespacedService(registrySvcName, buildNs);
|
|
this.logger.log(`Service "${registrySvcName}" already exists`);
|
|
} catch (err: any) {
|
|
if (err.statusCode === 404 || err.body?.code === 404) {
|
|
await coreApi.createNamespacedService(buildNs, {
|
|
metadata: { name: registrySvcName, namespace: buildNs, labels: { app: 'registry' } },
|
|
spec: {
|
|
type: 'ClusterIP',
|
|
selector: { app: 'registry' },
|
|
ports: [{ port: 5000, targetPort: 5000 as any, protocol: 'TCP' }],
|
|
},
|
|
});
|
|
this.logger.log(`Created Registry ClusterIP Service (port 5000)`);
|
|
} else {
|
|
throw err;
|
|
}
|
|
}
|
|
|
|
// ── 6. Registry NodePort Service (optional host access :30500) ─
|
|
const registryNodePort = 30500;
|
|
const registryNodePortName = 'registry-nodeport';
|
|
try {
|
|
await coreApi.readNamespacedService(registryNodePortName, buildNs);
|
|
this.logger.log(`Service "${registryNodePortName}" already exists`);
|
|
} catch (err: any) {
|
|
if (err.statusCode === 404 || err.body?.code === 404) {
|
|
await coreApi.createNamespacedService(buildNs, {
|
|
metadata: { name: registryNodePortName, namespace: buildNs, labels: { app: 'registry' } },
|
|
spec: {
|
|
type: 'NodePort',
|
|
selector: { app: 'registry' },
|
|
ports: [{
|
|
port: 5000,
|
|
targetPort: 5000 as any,
|
|
nodePort: registryNodePort,
|
|
protocol: 'TCP',
|
|
}],
|
|
},
|
|
});
|
|
this.logger.log(`Created Registry NodePort Service (${registryNodePort} → 5000)`);
|
|
} else {
|
|
throw err;
|
|
}
|
|
}
|
|
|
|
// ── 7. registry-credentials Secret (docker config for Kaniko) ─
|
|
const registrySecretName = 'registry-credentials';
|
|
const dockerConfig = this.registryService.buildDockerConfigJson();
|
|
const kanikoRegistrySecret: k8s.V1Secret = {
|
|
metadata: { name: registrySecretName, namespace: buildNs },
|
|
type: 'kubernetes.io/dockerconfigjson',
|
|
data: {
|
|
'.dockerconfigjson': Buffer.from(dockerConfig).toString('base64'),
|
|
},
|
|
};
|
|
try {
|
|
await coreApi.readNamespacedSecret(registrySecretName, buildNs);
|
|
await coreApi.replaceNamespacedSecret(registrySecretName, buildNs, kanikoRegistrySecret);
|
|
} catch (err: any) {
|
|
if (err.statusCode === 404 || err.body?.code === 404) {
|
|
await coreApi.createNamespacedSecret(buildNs, kanikoRegistrySecret);
|
|
this.logger.log(`Created registry-credentials Secret`);
|
|
} else {
|
|
throw err;
|
|
}
|
|
}
|
|
|
|
await this.ensureNodeClusterDns(coreApi, appsApi);
|
|
await this.ensureK3sRegistryMirrors(appsApi, registryUrl);
|
|
|
|
this.logger.log(`✅ Cluster bootstrap complete — registry: ${registryUrl}`);
|
|
}
|
|
|
|
/**
|
|
* Kubelet/containerd pull images on the host network stack, which uses the node's
|
|
* resolver (often systemd-resolved) — not pod DNS. Forward *.cluster.local to CoreDNS
|
|
* so registry.cloudhost-builds.svc.cluster.local resolves during image pulls.
|
|
*/
|
|
private async ensureNodeClusterDns(
|
|
coreApi: k8s.CoreV1Api,
|
|
appsApi: k8s.AppsV1Api,
|
|
): Promise<void> {
|
|
const dsName = 'cloudhost-node-cluster-dns';
|
|
const namespace = 'kube-system';
|
|
let clusterDnsIp = '10.43.0.10';
|
|
try {
|
|
const dnsSvc = await coreApi.readNamespacedService('kube-dns', namespace);
|
|
clusterDnsIp = dnsSvc.body.spec?.clusterIP || clusterDnsIp;
|
|
} catch (err: any) {
|
|
this.logger.warn(
|
|
`Could not read kube-dns ClusterIP (${err.message}); using ${clusterDnsIp}`,
|
|
);
|
|
}
|
|
|
|
const configureScript = [
|
|
'set -e',
|
|
'CONF=/host/etc/systemd/resolved.conf.d/k8s-cluster-dns.conf',
|
|
'mkdir -p /host/etc/systemd/resolved.conf.d',
|
|
`cat > /tmp/k8s-cluster-dns.conf <<'EOF'`,
|
|
'[Resolve]',
|
|
`DNS=${clusterDnsIp}`,
|
|
'Domains=~cluster.local',
|
|
'EOF',
|
|
'if [ ! -f "$CONF" ] || ! cmp -s /tmp/k8s-cluster-dns.conf "$CONF"; then',
|
|
' cp /tmp/k8s-cluster-dns.conf "$CONF"',
|
|
' echo "Updated k8s-cluster-dns.conf"',
|
|
' if nsenter -t 1 -m -u -i -n -p -- systemctl is-active systemd-resolved >/dev/null 2>&1; then',
|
|
' nsenter -t 1 -m -u -i -n -p -- systemctl restart systemd-resolved',
|
|
' fi',
|
|
'fi',
|
|
'sleep infinity',
|
|
].join('\n');
|
|
|
|
const daemonSet: k8s.V1DaemonSet = {
|
|
metadata: {
|
|
name: dsName,
|
|
namespace,
|
|
labels: { 'app.kubernetes.io/managed-by': 'cloudhost' },
|
|
},
|
|
spec: {
|
|
selector: { matchLabels: { app: dsName } },
|
|
template: {
|
|
metadata: { labels: { app: dsName } },
|
|
spec: {
|
|
hostPID: true,
|
|
tolerations: [{ operator: 'Exists' }],
|
|
containers: [
|
|
{
|
|
name: 'configure',
|
|
image: 'rancher/mirrored-library-busybox:1.36.1',
|
|
command: ['/bin/sh', '-ec'],
|
|
args: [configureScript],
|
|
securityContext: { privileged: true },
|
|
volumeMounts: [{ name: 'etc', mountPath: '/host/etc' }],
|
|
},
|
|
],
|
|
volumes: [
|
|
{
|
|
name: 'etc',
|
|
hostPath: { path: '/etc', type: 'Directory' },
|
|
},
|
|
],
|
|
},
|
|
},
|
|
},
|
|
};
|
|
|
|
try {
|
|
await appsApi.readNamespacedDaemonSet(dsName, namespace);
|
|
await appsApi.replaceNamespacedDaemonSet(dsName, namespace, daemonSet);
|
|
this.logger.log(`Updated DaemonSet "${dsName}" (cluster DNS ${clusterDnsIp})`);
|
|
} catch (err: any) {
|
|
if (err.statusCode === 404 || err.body?.code === 404) {
|
|
await appsApi.createNamespacedDaemonSet(namespace, daemonSet);
|
|
this.logger.log(`Created DaemonSet "${dsName}" (cluster DNS ${clusterDnsIp})`);
|
|
} else {
|
|
throw err;
|
|
}
|
|
}
|
|
}
|
|
|
|
/** In-cluster registry mirror for k3s/containerd (HTTP). Removes legacy external-registry DaemonSet if present. */
|
|
private async ensureK3sRegistryMirrors(
|
|
appsApi: k8s.AppsV1Api,
|
|
registryUrl: string,
|
|
): Promise<void> {
|
|
const namespace = 'kube-system';
|
|
const legacyDs = 'cloudhost-k3s-registry-config';
|
|
try {
|
|
await appsApi.deleteNamespacedDaemonSet(legacyDs, namespace);
|
|
this.logger.log(`Removed legacy DaemonSet "${legacyDs}"`);
|
|
} catch (err: any) {
|
|
if (err.statusCode !== 404 && err.body?.code !== 404) {
|
|
this.logger.warn(`Could not delete legacy DaemonSet "${legacyDs}": ${err.message}`);
|
|
}
|
|
}
|
|
|
|
const { username, password } = this.registryService.getRegistryCredentials();
|
|
const dsName = 'cloudhost-k3s-registry-mirrors';
|
|
const configureScript = [
|
|
'set -e',
|
|
'REG=/host/etc/rancher/k3s/registries.yaml',
|
|
'mkdir -p /host/etc/rancher/k3s',
|
|
'cat > /tmp/cloudhost-registries.yaml <<EOFREG',
|
|
'mirrors:',
|
|
` "${registryUrl}":`,
|
|
' endpoint:',
|
|
` - "http://${registryUrl}"`,
|
|
'configs:',
|
|
` "${registryUrl}":`,
|
|
' auth:',
|
|
` username: ${JSON.stringify(username)}`,
|
|
` password: ${JSON.stringify(password)}`,
|
|
'EOFREG',
|
|
'if [ ! -f "$REG" ] || ! cmp -s /tmp/cloudhost-registries.yaml "$REG"; then',
|
|
' cp /tmp/cloudhost-registries.yaml "$REG"',
|
|
' echo "Updated registries.yaml"',
|
|
' nsenter -t 1 -m -u -i -n -p -- systemctl restart k3s 2>/dev/null || true',
|
|
'fi',
|
|
'sleep infinity',
|
|
].join('\n');
|
|
|
|
const daemonSet: k8s.V1DaemonSet = {
|
|
metadata: {
|
|
name: dsName,
|
|
namespace,
|
|
labels: { 'app.kubernetes.io/managed-by': 'cloudhost' },
|
|
},
|
|
spec: {
|
|
selector: { matchLabels: { app: dsName } },
|
|
template: {
|
|
metadata: { labels: { app: dsName } },
|
|
spec: {
|
|
hostPID: true,
|
|
tolerations: [{ operator: 'Exists' }],
|
|
containers: [
|
|
{
|
|
name: 'configure',
|
|
image: 'rancher/mirrored-library-busybox:1.36.1',
|
|
command: ['/bin/sh', '-ec'],
|
|
args: [configureScript],
|
|
securityContext: { privileged: true },
|
|
volumeMounts: [{ name: 'etc', mountPath: '/host/etc' }],
|
|
},
|
|
],
|
|
volumes: [
|
|
{
|
|
name: 'etc',
|
|
hostPath: { path: '/etc', type: 'Directory' },
|
|
},
|
|
],
|
|
},
|
|
},
|
|
},
|
|
};
|
|
|
|
try {
|
|
await appsApi.readNamespacedDaemonSet(dsName, namespace);
|
|
await appsApi.replaceNamespacedDaemonSet(dsName, namespace, daemonSet);
|
|
this.logger.log(`Updated DaemonSet "${dsName}"`);
|
|
} catch (err: any) {
|
|
if (err.statusCode === 404 || err.body?.code === 404) {
|
|
await appsApi.createNamespacedDaemonSet(namespace, daemonSet);
|
|
this.logger.log(`Created DaemonSet "${dsName}"`);
|
|
} else {
|
|
throw err;
|
|
}
|
|
}
|
|
}
|
|
|
|
private parseCpuToMillicores(cpu: string): number {
|
|
if (!cpu || cpu === '0') return 0;
|
|
if (cpu.endsWith('n')) return parseFloat(cpu) / 1_000_000;
|
|
if (cpu.endsWith('u')) return parseFloat(cpu) / 1_000;
|
|
if (cpu.endsWith('m')) return parseFloat(cpu);
|
|
return parseFloat(cpu) * 1000;
|
|
}
|
|
|
|
private parseMemoryToMi(memory: string): number {
|
|
if (!memory || memory === '0') return 0;
|
|
if (memory.endsWith('Ki')) return parseFloat(memory) / 1024;
|
|
if (memory.endsWith('Mi')) return parseFloat(memory);
|
|
if (memory.endsWith('Gi')) return parseFloat(memory) * 1024;
|
|
if (memory.endsWith('Ti')) return parseFloat(memory) * 1024 * 1024;
|
|
return parseFloat(memory) / (1024 * 1024); // bytes
|
|
}
|
|
|
|
private estimateApplicationRequest(dto: CreateApplicationDto): Record<string, any> {
|
|
const replicas = Math.max(dto.replicas || 1, 1);
|
|
let cpuMillicores = this.parseCpuToMillicores(dto.cpuRequest || '100m') * replicas;
|
|
let memoryMi = this.parseMemoryToMi(dto.memoryRequest || '128Mi') * replicas;
|
|
let storageMi = this.parseStorageToMi(dto.dbStorageSize || '1Gi') + this.parseStorageToMi(dto.appStorageSize || '2Gi');
|
|
|
|
if (dto.databaseType && dto.databaseType !== 'none') {
|
|
cpuMillicores += 100;
|
|
memoryMi += 256;
|
|
}
|
|
|
|
for (const service of ['redis', 'rabbitmq'] as const) {
|
|
const enabled = service === 'redis' ? dto.enableRedis : dto.enableRabbitmq;
|
|
const custom = dto.optionalServiceResources?.[service];
|
|
if (!enabled && !custom) continue;
|
|
|
|
cpuMillicores += this.parseCpuToMillicores(custom?.cpuRequest || custom?.cpuLimit || '100m');
|
|
memoryMi += this.parseMemoryToMi(custom?.memoryRequest || custom?.memoryLimit || '128Mi');
|
|
storageMi += (custom?.storageGi || 1) * 1024;
|
|
}
|
|
|
|
return {
|
|
cpuMillicores,
|
|
memoryMi,
|
|
storageMi,
|
|
replicas,
|
|
podEstimate: replicas + (dto.databaseType && dto.databaseType !== 'none' ? 1 : 0) + (dto.enableRedis ? 1 : 0) + (dto.enableRabbitmq ? 1 : 0),
|
|
};
|
|
}
|
|
|
|
private async getCachedHealthyClusters(
|
|
pool?: ClusterPool,
|
|
excludeClusterIds: string[] = [],
|
|
): Promise<Cluster[]> {
|
|
const query = this.clustersRepository
|
|
.createQueryBuilder('cluster')
|
|
.where('cluster.status = :status', { status: ClusterStatus.ACTIVE });
|
|
|
|
if (pool?.clusterIds?.length) {
|
|
query.andWhere('cluster.id IN (:...clusterIds)', { clusterIds: pool.clusterIds });
|
|
}
|
|
|
|
if (excludeClusterIds.length > 0) {
|
|
query.andWhere('cluster.id NOT IN (:...excludeClusterIds)', { excludeClusterIds });
|
|
}
|
|
|
|
const clusters = await query.getMany();
|
|
const healthy: Cluster[] = [];
|
|
for (const cluster of clusters) {
|
|
const cached = await this.getClusterSnapshot(cluster);
|
|
if (cached.status === ClusterStatus.ACTIVE && cached.healthStatus === 'healthy') {
|
|
healthy.push(cached);
|
|
}
|
|
}
|
|
return healthy;
|
|
}
|
|
|
|
private async getClusterSnapshot(cluster: Cluster): Promise<Cluster> {
|
|
const cached = this.clusterHealthCache.get(cluster.id);
|
|
if (cached && Date.now() - cached.cachedAt < this.cacheTtlMs) {
|
|
return cached.cluster;
|
|
}
|
|
|
|
if (
|
|
cluster.availableResources &&
|
|
cluster.lastHealthCheckedAt &&
|
|
Date.now() - new Date(cluster.lastHealthCheckedAt).getTime() < this.cacheTtlMs
|
|
) {
|
|
this.clusterHealthCache.set(cluster.id, { cluster, cachedAt: Date.now() });
|
|
return cluster;
|
|
}
|
|
|
|
try {
|
|
await this.getClusterResources(cluster.id, { allowCached: false });
|
|
} catch (err: any) {
|
|
this.logger.warn(`Health refresh failed for cluster "${cluster.name}": ${err.message}`);
|
|
}
|
|
|
|
const fresh = await this.clustersRepository.findOne({ where: { id: cluster.id } });
|
|
const snapshot = fresh || cluster;
|
|
this.clusterHealthCache.set(cluster.id, { cluster: snapshot, cachedAt: Date.now() });
|
|
return snapshot;
|
|
}
|
|
|
|
async refreshAllClusterHealth(): Promise<void> {
|
|
const clusters = await this.clustersRepository.find({
|
|
where: [
|
|
{ status: ClusterStatus.ACTIVE },
|
|
{ status: ClusterStatus.MAINTENANCE },
|
|
],
|
|
});
|
|
|
|
for (const cluster of clusters) {
|
|
try {
|
|
await this.getClusterResources(cluster.id, { allowCached: false });
|
|
} catch (err: any) {
|
|
this.logger.warn(`Cluster health refresh failed for "${cluster.name}": ${err.message}`);
|
|
}
|
|
}
|
|
}
|
|
|
|
private getClusterRejectionReason(cluster: Cluster, estimatedRequest: Record<string, any>): string | null {
|
|
if (cluster.status !== ClusterStatus.ACTIVE) {
|
|
return `status=${cluster.status}`;
|
|
}
|
|
|
|
if (cluster.healthStatus !== 'healthy') {
|
|
return `health=${cluster.healthStatus}`;
|
|
}
|
|
|
|
const metrics = this.getResourceMetrics(cluster, estimatedRequest, 0);
|
|
const { available, utilization } = metrics;
|
|
|
|
if (available.cpuMillicores > 0 && available.cpuMillicores < estimatedRequest.cpuMillicores) {
|
|
return `insufficient cpu (${available.cpuMillicores}m < ${estimatedRequest.cpuMillicores}m)`;
|
|
}
|
|
if (available.memoryMi > 0 && available.memoryMi < estimatedRequest.memoryMi) {
|
|
return `insufficient memory (${available.memoryMi}Mi < ${estimatedRequest.memoryMi}Mi)`;
|
|
}
|
|
if (available.pods > 0 && available.pods < estimatedRequest.podEstimate) {
|
|
return `pod pressure (${available.pods} pods available < ${estimatedRequest.podEstimate})`;
|
|
}
|
|
if (utilization.cpu >= this.overloadedCpuThreshold) {
|
|
return `cpu overloaded (${Math.round(utilization.cpu * 100)}%)`;
|
|
}
|
|
if (utilization.memory >= this.overloadedMemoryThreshold) {
|
|
return `memory overloaded (${Math.round(utilization.memory * 100)}%)`;
|
|
}
|
|
if (utilization.pods >= this.overloadedPodThreshold) {
|
|
return `pod overloaded (${Math.round(utilization.pods * 100)}%)`;
|
|
}
|
|
|
|
return null;
|
|
}
|
|
|
|
private scoreCluster(
|
|
cluster: Cluster,
|
|
estimatedRequest: Record<string, any>,
|
|
appCount: number,
|
|
strategy: PoolStrategy,
|
|
desiredRegion?: string,
|
|
): number {
|
|
const metrics = this.getResourceMetrics(cluster, estimatedRequest, appCount);
|
|
const capacityScore = metrics.capacityScore;
|
|
const appPenalty = Math.min(appCount, 100) * 0.75;
|
|
|
|
if (strategy === 'round-robin') {
|
|
return this.nextRoundRobinScore('allocator');
|
|
}
|
|
if (strategy === 'weighted-round-robin') {
|
|
return this.nextWeightedRoundRobinScore(cluster);
|
|
}
|
|
if (strategy === 'least-apps') {
|
|
return 1000 - appPenalty + capacityScore * 100;
|
|
}
|
|
if (strategy === 'least-loaded') {
|
|
return 1000 + capacityScore * 200 - metrics.utilization.average * 100 - appPenalty;
|
|
}
|
|
if (strategy === 'region-based') {
|
|
const regionBoost = desiredRegion && cluster.region === desiredRegion ? 250 : 0;
|
|
return regionBoost + (cluster.weight || 1) * 50 + capacityScore * 150 - appPenalty;
|
|
}
|
|
|
|
return (cluster.weight || 1) * 100 + capacityScore * 100 - appPenalty;
|
|
}
|
|
|
|
private getResourceMetrics(
|
|
cluster: Cluster,
|
|
estimatedRequest: Record<string, any>,
|
|
appCount: number,
|
|
): {
|
|
available: { cpuMillicores: number; memoryMi: number; storageMi: number; pods: number };
|
|
utilization: { cpu: number; memory: number; storage: number; pods: number; appPressure: number; average: number };
|
|
capacityScore: number;
|
|
} {
|
|
const resources = cluster.availableResources || {};
|
|
const cpuCapacity = this.parseCpuToMillicores(resources.totalCpuAllocatable || resources.cpuAllocatable || resources.totalCpuCapacity || '0');
|
|
const memoryCapacity = this.parseMemoryToMi(resources.totalMemoryAllocatable || resources.memoryAllocatable || resources.totalMemoryCapacity || '0');
|
|
const storageCapacity = this.parseStorageToMi(resources.storageAllocatable || resources.storageCapacity || '0');
|
|
const cpuRequested = this.parseCpuToMillicores(resources.totalCpuRequested || resources.cpuRequested || '0');
|
|
const memoryRequested = this.parseMemoryToMi(resources.totalMemoryRequested || resources.memoryRequested || '0');
|
|
const storageUsed = this.parseStorageToMi(resources.storageUsed || '0');
|
|
const podCount = Number(resources.podCount || 0);
|
|
const nodeCount = Number(resources.nodeCount || 0);
|
|
const podCapacity = Number(resources.podCapacity || nodeCount * 110 || 0);
|
|
|
|
const available = {
|
|
cpuMillicores: Math.max(cpuCapacity - cpuRequested, 0),
|
|
memoryMi: Math.max(memoryCapacity - memoryRequested, 0),
|
|
storageMi: storageCapacity > 0 ? Math.max(storageCapacity - storageUsed, 0) : 0,
|
|
pods: podCapacity > 0 ? Math.max(podCapacity - podCount, 0) : 0,
|
|
};
|
|
const utilization = {
|
|
cpu: this.utilizationRatio(cpuRequested, cpuCapacity),
|
|
memory: this.utilizationRatio(memoryRequested, memoryCapacity),
|
|
storage: this.utilizationRatio(storageUsed, storageCapacity),
|
|
pods: this.utilizationRatio(podCount, podCapacity),
|
|
appPressure: Math.min(appCount / 100, 1),
|
|
average: 0,
|
|
};
|
|
utilization.average = (
|
|
utilization.cpu * 0.35 +
|
|
utilization.memory * 0.35 +
|
|
utilization.pods * 0.2 +
|
|
utilization.appPressure * 0.1
|
|
);
|
|
|
|
const capacityScore =
|
|
this.ratioScore(available.cpuMillicores, estimatedRequest.cpuMillicores) * 0.35 +
|
|
this.ratioScore(available.memoryMi, estimatedRequest.memoryMi) * 0.35 +
|
|
this.ratioScore(available.storageMi, estimatedRequest.storageMi) * 0.1 +
|
|
this.ratioScore(available.pods, estimatedRequest.podEstimate) * 0.2;
|
|
|
|
return { available, utilization, capacityScore };
|
|
}
|
|
|
|
private resolveStrategy(strategy?: string): PoolStrategy {
|
|
const strategies: PoolStrategy[] = [
|
|
'round-robin',
|
|
'weighted-round-robin',
|
|
'least-apps',
|
|
'least-loaded',
|
|
'weighted-resource',
|
|
'region-based',
|
|
];
|
|
return strategies.includes(strategy as PoolStrategy) ? strategy as PoolStrategy : 'weighted-resource';
|
|
}
|
|
|
|
private nextRoundRobinScore(scope: string): number {
|
|
const idx = this.poolRoundRobinIndices.get(scope) || 0;
|
|
this.poolRoundRobinIndices.set(scope, idx + 1);
|
|
return 1000 - idx;
|
|
}
|
|
|
|
private nextWeightedRoundRobinScore(cluster: Cluster): number {
|
|
const current = this.weightedRoundRobinState.get(cluster.id) || 0;
|
|
const next = current + Math.max(cluster.weight || 1, 1);
|
|
this.weightedRoundRobinState.set(cluster.id, next);
|
|
return next;
|
|
}
|
|
|
|
private utilizationRatio(used: number, capacity: number): number {
|
|
if (!capacity || capacity <= 0) return 0;
|
|
return Math.min(Math.max(used / capacity, 0), 1);
|
|
}
|
|
|
|
private ratioScore(available: number, required: number): number {
|
|
if (!available) return 0.7;
|
|
if (!required) return 1;
|
|
return Math.min(available / required, 10) / 10;
|
|
}
|
|
|
|
private async getAppCounts(clusterIds: string[]): Promise<Map<string, number>> {
|
|
if (clusterIds.length === 0) {
|
|
return new Map();
|
|
}
|
|
const rows: { clusterId: string; count: string }[] = await this.dataSource.query(`
|
|
SELECT "clusterId", COUNT(*) as count
|
|
FROM applications
|
|
WHERE "clusterId" = ANY($1)
|
|
AND "lifecycleStatus" = 'active'
|
|
GROUP BY "clusterId"
|
|
`, [clusterIds]);
|
|
return new Map(rows.map((row) => [row.clusterId, parseInt(row.count, 10)]));
|
|
}
|
|
|
|
private parseStorageToMi(storage: string): number {
|
|
if (!storage) return 0;
|
|
if (storage.endsWith('Ki')) return parseFloat(storage) / 1024;
|
|
if (storage.endsWith('Mi')) return parseFloat(storage);
|
|
if (storage.endsWith('Gi')) return parseFloat(storage) * 1024;
|
|
if (storage.endsWith('Ti')) return parseFloat(storage) * 1024 * 1024;
|
|
return parseFloat(storage) || 0;
|
|
}
|
|
|
|
private async recordHealthSnapshot(
|
|
cluster: Cluster,
|
|
snapshot: {
|
|
status: ClusterHealthStatus;
|
|
message?: string;
|
|
resources?: Record<string, any>;
|
|
},
|
|
): Promise<void> {
|
|
const resources = snapshot.resources || cluster.availableResources || {};
|
|
await this.healthRepository.save(this.healthRepository.create({
|
|
clusterId: cluster.id,
|
|
status: snapshot.status,
|
|
readyNodes: resources.readyNodeCount || resources.nodeCount || 0,
|
|
nodeCount: resources.nodeCount || 0,
|
|
cpuAllocatable: resources.totalCpuAllocatable || resources.cpuAllocatable || null,
|
|
memoryAllocatable: resources.totalMemoryAllocatable || resources.memoryAllocatable || null,
|
|
podCount: resources.podCount || 0,
|
|
appCount: resources.appCount || 0,
|
|
message: snapshot.message,
|
|
resources,
|
|
}));
|
|
}
|
|
|
|
private encryptKubeconfig(kubeconfig: string): string {
|
|
if (!kubeconfig || kubeconfig.startsWith('enc:v1:')) {
|
|
return kubeconfig;
|
|
}
|
|
const key = this.getKubeconfigEncryptionKey();
|
|
if (!key) {
|
|
return kubeconfig;
|
|
}
|
|
const iv = crypto.randomBytes(12);
|
|
const cipher = crypto.createCipheriv('aes-256-gcm', key, iv);
|
|
const encrypted = Buffer.concat([cipher.update(kubeconfig, 'utf8'), cipher.final()]);
|
|
const tag = cipher.getAuthTag();
|
|
return `enc:v1:${iv.toString('base64')}:${tag.toString('base64')}:${encrypted.toString('base64')}`;
|
|
}
|
|
|
|
private decryptKubeconfig(kubeconfig: string): string {
|
|
if (!kubeconfig?.startsWith('enc:v1:')) {
|
|
return kubeconfig;
|
|
}
|
|
const key = this.getKubeconfigEncryptionKey();
|
|
if (!key) {
|
|
throw new BadRequestException('Kubeconfig is encrypted but CLUSTER_KUBECONFIG_KEY is not configured');
|
|
}
|
|
const [, , ivRaw, tagRaw, encryptedRaw] = kubeconfig.split(':');
|
|
const decipher = crypto.createDecipheriv('aes-256-gcm', key, Buffer.from(ivRaw, 'base64'));
|
|
decipher.setAuthTag(Buffer.from(tagRaw, 'base64'));
|
|
return Buffer.concat([
|
|
decipher.update(Buffer.from(encryptedRaw, 'base64')),
|
|
decipher.final(),
|
|
]).toString('utf8');
|
|
}
|
|
|
|
private withDecryptedKubeconfig(cluster: Cluster): Cluster {
|
|
return {
|
|
...cluster,
|
|
kubeconfig: this.decryptKubeconfig(cluster.kubeconfig),
|
|
};
|
|
}
|
|
|
|
private getKubeconfigEncryptionKey(): Buffer | null {
|
|
const secret = this.configService.get<string>('CLUSTER_KUBECONFIG_KEY') || this.configService.get<string>('cluster.kubeconfigKey');
|
|
if (!secret) {
|
|
return null;
|
|
}
|
|
if (secret.length === 64 && /^[0-9a-f]+$/i.test(secret)) {
|
|
return Buffer.from(secret, 'hex');
|
|
}
|
|
return crypto.createHash('sha256').update(secret).digest();
|
|
}
|
|
}
|