Improve cluster allocation strategy.

Co-authored-by: Cursor <cursoragent@cursor.com>

backend/src/clusters/clusters.service.ts

@@ -1,11 +1,20 @@
-import { Injectable, NotFoundException, Logger, BadRequestException, Inject, forwardRef } from '@nestjs/common';
+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 } from './entities/cluster-pool.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';
@@ -15,10 +24,21 @@ import { ElasticsearchService } from '../kubernetes/elasticsearch.service';
 import { CreateApplicationDto } from '../applications/dto/application.dto';
 
 @Injectable()
-export class ClustersService {
+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)
@@ -35,6 +55,22 @@ export class ClustersService {
     private elasticsearchService: ElasticsearchService,
   ) {}
 
+  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.
@@ -238,6 +274,11 @@ export class ClustersService {
   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
@@ -256,15 +297,7 @@ export class ClustersService {
       });
     }
 
-    const candidateQuery = this.clustersRepository
-      .createQueryBuilder('cluster')
-      .where('cluster.status = :status', { status: ClusterStatus.ACTIVE });
-
-    if (pool?.clusterIds?.length) {
-      candidateQuery.andWhere('cluster.id IN (:...clusterIds)', { clusterIds: pool.clusterIds });
-    }
-
-    const candidates = await candidateQuery.getMany();
+    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>[] = [];
@@ -276,14 +309,18 @@ export class ClustersService {
         continue;
       }
 
-      const score = this.scoreCluster(cluster, estimatedRequest, appCounts.get(cluster.id) || 0, pool?.strategy || 'weighted-resource');
+      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: appCounts.get(cluster.id) || 0,
+        appCount,
         healthStatus: cluster.healthStatus,
+        utilization: resourceMetrics.utilization,
         availableResources: cluster.availableResources || null,
       });
     }
@@ -292,13 +329,14 @@ export class ClustersService {
       const log = await this.allocationLogsRepository.save(this.allocationLogsRepository.create({
         userId,
         poolId: pool?.id,
+        applicationId: options.applicationId,
         selectedClusterId: null,
-        strategy: pool?.strategy || 'weighted-resource',
+        strategy: this.resolveStrategy(pool?.strategy),
         estimatedRequest,
         candidateScores,
         rejectionReasons,
         status: 'failed',
-        message: 'No active healthy cluster has enough estimated capacity',
+        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})`,
@@ -310,13 +348,14 @@ export class ClustersService {
     const log = await this.allocationLogsRepository.save(this.allocationLogsRepository.create({
       userId,
       poolId: pool?.id,
+      applicationId: options.applicationId,
       selectedClusterId: selected.id,
-      strategy: pool?.strategy || 'weighted-resource',
+      strategy: this.resolveStrategy(pool?.strategy),
       estimatedRequest,
       candidateScores,
       rejectionReasons,
       status: 'success',
-      message: `Selected ${selected.name}`,
+      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)})`);
@@ -335,6 +374,48 @@ export class ClustersService {
     });
   }
 
+  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.
@@ -593,18 +674,32 @@ export class ClustersService {
   /**
    * Get resource usage for a specific cluster — nodes, total CPU/memory, pod counts.
    */
-  async getClusterResources(id: string): Promise<{
+  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();
     kc.loadFromString(cluster.kubeconfig);
@@ -647,6 +742,14 @@ export class ClustersService {
       // 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(
@@ -656,12 +759,19 @@ export class ClustersService {
       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,
@@ -680,6 +790,16 @@ export class ClustersService {
         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;
@@ -690,6 +810,7 @@ export class ClustersService {
         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}`);
     }
@@ -927,30 +1048,106 @@ export class ClustersService {
     };
   }
 
+  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', 'unknown'].includes(cluster.healthStatus)) {
+    if (cluster.healthStatus !== 'healthy') {
       return `health=${cluster.healthStatus}`;
     }
 
-    const resources = cluster.availableResources || {};
-    const cpuFree = this.parseCpuToMillicores(resources.totalCpuAllocatable || resources.cpuAllocatable || '0');
-    const memoryFree = this.parseMemoryToMi(resources.totalMemoryAllocatable || resources.memoryAllocatable || '0');
-    const podCount = Number(resources.podCount || 0);
-    const nodeCount = Number(resources.nodeCount || 0);
-    const podCapacity = Number(resources.podCapacity || nodeCount * 110 || 0);
+    const metrics = this.getResourceMetrics(cluster, estimatedRequest, 0);
+    const { available, utilization } = metrics;
 
-    if (cpuFree > 0 && cpuFree < estimatedRequest.cpuMillicores) {
-      return `insufficient cpu (${cpuFree}m < ${estimatedRequest.cpuMillicores}m)`;
+    if (available.cpuMillicores > 0 && available.cpuMillicores < estimatedRequest.cpuMillicores) {
+      return `insufficient cpu (${available.cpuMillicores}m < ${estimatedRequest.cpuMillicores}m)`;
     }
-    if (memoryFree > 0 && memoryFree < estimatedRequest.memoryMi) {
-      return `insufficient memory (${memoryFree}Mi < ${estimatedRequest.memoryMi}Mi)`;
+    if (available.memoryMi > 0 && available.memoryMi < estimatedRequest.memoryMi) {
+      return `insufficient memory (${available.memoryMi}Mi < ${estimatedRequest.memoryMi}Mi)`;
     }
-    if (podCapacity > 0 && podCount + estimatedRequest.podEstimate > podCapacity) {
-      return `pod pressure (${podCount}/${podCapacity})`;
+    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;
@@ -960,34 +1157,111 @@ export class ClustersService {
     cluster: Cluster,
     estimatedRequest: Record<string, any>,
     appCount: number,
-    strategy: string,
+    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') {
-      const idx = this.poolRoundRobinIndices.get('allocator') || 0;
-      this.poolRoundRobinIndices.set('allocator', idx + 1);
-      return 1000 - idx;
+      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 cpuFree = this.parseCpuToMillicores(resources.totalCpuAllocatable || resources.cpuAllocatable || '0');
-    const memoryFree = this.parseMemoryToMi(resources.totalMemoryAllocatable || resources.memoryAllocatable || '0');
-    const storageFree = this.parseStorageToMi(resources.storageAllocatable || resources.storageFree || '0');
+    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(cpuFree, estimatedRequest.cpuMillicores) * 0.35 +
-      this.ratioScore(memoryFree, estimatedRequest.memoryMi) * 0.35 +
-      this.ratioScore(storageFree, estimatedRequest.storageMi) * 0.15 +
-      (podCapacity > 0 ? Math.max(0, 1 - podCount / podCapacity) : 0.7) * 0.15;
-    const appPenalty = Math.min(appCount, 100) * 0.75;
+      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;
 
-    if (strategy === 'least-apps') {
-      return 1000 - appPenalty + capacityScore * 100;
-    }
+    return { available, utilization, capacityScore };
+  }
 
-    return (cluster.weight || 1) * 100 + capacityScore * 100 - appPenalty;
+  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 {
@@ -1004,6 +1278,7 @@ export class ClustersService {
       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)]));