untracked instances cause groups to throttle (#119)

* untracked instances cause groups to throttle

flag on groups for throttle behavior on/off
audit adds function for querying untracked by group
instance launcher checks untrack counts

* get untracked from metrics loop not audit

* only throttle if untracked count exceeds threshold

* global max for scale up threshold

Author: aaronkvanmeerten

src/app.ts

@@ -143,22 +143,24 @@ const autoscaleProcessor = new AutoscaleProcessor({
     audit: audit,
 });
 
+const metricsLoop = new MetricsLoop({
+    redisClient: redisClient,
+    metricsTTL: config.ServiceLevelMetricsTTL,
+    instanceGroupManager: instanceGroupManager,
+    instanceTracker: instanceTracker,
+    ctx: initCtx,
+});
+
 const instanceLauncher = new InstanceLauncher({
+    maxThrottleThreshold: config.MaxThrottleThreshold,
     instanceTracker: instanceTracker,
     cloudManager: cloudManager,
     instanceGroupManager: instanceGroupManager,
     lockManager: lockManager,
     redisClient,
     shutdownManager,
     audit: audit,
-});
-
-const metricsLoop = new MetricsLoop({
-    redisClient: redisClient,
-    metricsTTL: config.ServiceLevelMetricsTTL,
-    instanceGroupManager: instanceGroupManager,
-    instanceTracker: instanceTracker,
-    ctx: initCtx,
+    metricsLoop,
 });
 
 const groupReportGenerator = new GroupReportGenerator({

src/config.ts

@@ -44,6 +44,7 @@ const env = envalid.cleanEnv(process.env, {
     SHUTDOWN_TTL_SEC: envalid.num({ default: 86400 }), // default 1 day
     SHUTDOWN_STATUS_TTL_SEC: envalid.num({ default: 600 }), // default 10 minutes
     AUDIT_TTL_SEC: envalid.num({ default: 172800 }), // default 2 day
+    MAX_THROTTLE_THRESHOLD: envalid.num({ default: 40 }), // default max of 40 untracked per group to throttle scale up
     GROUP_RELATED_DATA_TTL_SEC: envalid.num({ default: 172800 }), // default 2 day; keep group related data max 2 days after the group is deleted or no action is performed on it
     GROUP_LOCK_TTL_MS: envalid.num({ default: 180000 }), // time in ms
     GROUP_JOBS_CREATION_INTERVAL_SEC: envalid.num({ default: 30 }), // with what interval this instance should try producing jobs for group processing
@@ -127,6 +128,8 @@ export default {
     GroupRelatedDataTTL: env.GROUP_RELATED_DATA_TTL_SEC,
     // group processing lock
     GroupLockTTLMs: env.GROUP_LOCK_TTL_MS,
+    // group untracked threshold
+    MaxThrottleThreshold: env.MAX_THROTTLE_THRESHOLD,
     // queue jobs producers
     GroupJobsCreationIntervalSec: env.GROUP_JOBS_CREATION_INTERVAL_SEC,
     SanityJobsCreationIntervalSec: env.SANITY_JOBS_CREATION_INTERVAL_SEC,

src/handlers.ts

@@ -17,6 +17,7 @@ interface InstanceGroupScalingActivitiesRequest {
     enableAutoScale?: boolean;
     enableLaunch?: boolean;
     enableScheduler?: boolean;
+    enableUntrackedThrottle?: boolean;
 }
 
 export interface InstanceGroupDesiredValuesRequest {
@@ -189,6 +190,10 @@ class Handlers {
                 if (scalingActivitiesRequest.enableScheduler != null) {
                     instanceGroup.enableScheduler = scalingActivitiesRequest.enableScheduler;
                 }
+                if (scalingActivitiesRequest.enableUntrackedThrottle != null) {
+                    instanceGroup.enableUntrackedThrottle = scalingActivitiesRequest.enableUntrackedThrottle;
+                }
+
                 await this.instanceGroupManager.upsertInstanceGroup(req.context, instanceGroup);
                 res.status(200);
                 res.send({ save: 'OK' });

src/instance_group.ts

@@ -25,6 +25,7 @@ export interface InstanceGroup {
     enableAutoScale: boolean;
     enableLaunch: boolean;
     enableScheduler: boolean;
+    enableUntrackedThrottle: boolean;
     gracePeriodTTLSec: number;
     protectedTTLSec: number;
     scalingOptions: ScalingOptions;

src/instance_launcher.ts

@@ -7,6 +7,7 @@ import { Context } from './context';
 import * as promClient from 'prom-client';
 import ShutdownManager from './shutdown_manager';
 import Audit from './audit';
+import MetricsLoop from './metrics_loop';
 
 const instancesLaunchedCounter = new promClient.Counter({
     name: 'autoscaling_instance_launched_total',
@@ -27,23 +28,27 @@ const instanceErrorsCounter = new promClient.Counter({
 });
 
 export interface InstanceLauncherOptions {
+    maxThrottleThreshold?: number;
     instanceTracker: InstanceTracker;
     cloudManager: CloudManager;
     instanceGroupManager: InstanceGroupManager;
     lockManager: LockManager;
     redisClient: Redis.Redis;
     shutdownManager: ShutdownManager;
     audit: Audit;
+    metricsLoop: MetricsLoop;
 }
 
 export default class InstanceLauncher {
+    private maxThrottleThreshold = 40;
     private instanceTracker: InstanceTracker;
     private instanceGroupManager: InstanceGroupManager;
     private cloudManager: CloudManager;
     private redisClient: Redis.Redis;
     private lockManager: LockManager;
     private shutdownManager: ShutdownManager;
     private audit: Audit;
+    private metricsLoop: MetricsLoop;
 
     constructor(options: InstanceLauncherOptions) {
         this.instanceTracker = options.instanceTracker;
@@ -53,6 +58,11 @@ export default class InstanceLauncher {
         this.redisClient = options.redisClient;
         this.shutdownManager = options.shutdownManager;
         this.audit = options.audit;
+        this.metricsLoop = options.metricsLoop;
+
+        if (options.maxThrottleThreshold) {
+            this.maxThrottleThreshold = options.maxThrottleThreshold;
+        }
 
         this.launchOrShutdownInstancesByGroup = this.launchOrShutdownInstancesByGroup.bind(this);
     }
@@ -79,6 +89,39 @@ export default class InstanceLauncher {
 
                 const actualScaleUpQuantity =
                     Math.min(group.scalingOptions.maxDesired, group.scalingOptions.desiredCount) - count;
+
+                // if untracked throttle enabled, only scale up if there aren't too many untracked instances
+                if (group.enableUntrackedThrottle == null || group.enableUntrackedThrottle == true) {
+                    // use desired scaleUpQuantity to ensure we only scale up this many (plus one) until the previous batch are ready
+                    // ensure a maximum threshold from config (default of 40, much higher than ever seen except in cases in which throttling is desired)
+                    const untrackedThrottleThreshold = Math.min(
+                        group.scalingOptions.maxDesired + 1,
+                        this.maxThrottleThreshold,
+                    );
+                    const untrackedCount = await this.metricsLoop.getUnTrackedCount(group.name);
+                    // only allow scale up if untracked count is less than the threshold
+                    const allowedScaleUp = untrackedCount < untrackedThrottleThreshold;
+
+                    ctx.logger.debug(
+                        `[Launcher] Scaling throttle check for group ${groupName} with ${count} instances.`,
+                        { actualScaleUpQuantity, untrackedThrottleThreshold, untrackedCount, allowedScaleUp },
+                    );
+                    if (!allowedScaleUp) {
+                        // not allow to scale at all, error out here
+                        ctx.logger.error(
+                            `[Launcher] Scaling throttle launch of ALL new instances for group ${groupName} with ${count} instances.`,
+                            { untrackedCount, actualScaleUpQuantity, allowedScaleUp },
+                        );
+                        throw new Error(
+                            `[Launcher] Scaling throttled, failed to launch ALL new instances for group ${groupName}`,
+                        );
+                    } else {
+                        ctx.logger.debug(`[Launcher] Scaling throttle check passed for group ${groupName}.`);
+                    }
+                } else {
+                    ctx.logger.debug(`[Launcher] Scaling throttle disabled for group ${groupName}.`);
+                }
+
                 const scaleDownProtected = await this.instanceGroupManager.isScaleDownProtected(group.name);
                 const scaleUpCount = await this.cloudManager.scaleUp(
                     ctx,