seperate create and delete queues

EthDevOps · Apr 17, 2024 · f5213c8 · f5213c8
1 parent 0b332c0
commit f5213c8
Show file tree

Hide file tree

Showing 4 changed files with 39 additions and 36 deletions.
diff --git a/RunnerTask.cs → CreateRunnerTask.cs b/RunnerTask.cs → CreateRunnerTask.cs
@@ -1,12 +1,16 @@
 namespace GithubActionsOrchestrator;
 
-public record RunnerTask
+public record CreateRunnerTask
 {
     public string Arch { get; set; }
     public string Size { get; set; }
     public string RunnerToken { get; set; }
     public string OrgName { get; set; }
     public int RetryCount { get; set; }
-    public RunnerAction Action { get; set; }
     public long ServerId { get; set; }
-}
+}
+public record DeleteRunnerTask
+{
+    public int RetryCount { get; set; }
+    public long ServerId { get; set; }
+}
diff --git a/PoolManager.cs b/PoolManager.cs
@@ -14,13 +14,15 @@ public class PoolManager : BackgroundService
     private static readonly Gauge QueueSize = Metrics
         .CreateGauge("github_queue", "Number of queued runner tasks");
 
-    public ConcurrentQueue<RunnerTask?> Tasks { get; }
+    public ConcurrentQueue<CreateRunnerTask?> CreateTasks { get; }
+    public ConcurrentQueue<DeleteRunnerTask?> DeleteTasks { get; }
 
     public PoolManager(CloudController cc, ILogger<PoolManager> logger)
     {
         _cc = cc;
         _logger = logger;
-        Tasks = new ConcurrentQueue<RunnerTask?>();
+        CreateTasks = new ConcurrentQueue<CreateRunnerTask?>();
+        DeleteTasks = new ConcurrentQueue<DeleteRunnerTask?>();
     }
 
     protected override async Task ExecuteAsync(CancellationToken stoppingToken)
@@ -108,9 +110,8 @@ protected override async Task ExecuteAsync(CancellationToken stoppingToken)
                 for (int i = 0; i < missingCt; i++)
                 {
                     // Queue VM creation
-                    Tasks.Enqueue(new RunnerTask
+                    CreateTasks.Enqueue(new CreateRunnerTask
                     {
-                        Action = RunnerAction.Create,
                         Arch = arch,
                         Size = pool.Size,
                         RunnerToken = runnerToken,
@@ -127,26 +128,33 @@ protected override async Task ExecuteAsync(CancellationToken stoppingToken)
 
         while (!stoppingToken.IsCancellationRequested)
         {
-            QueueSize.Set(Tasks.Count);
-            if (Tasks.TryDequeue(out RunnerTask? task))
+            QueueSize.Set(CreateTasks.Count + DeleteTasks.Count);
+
+            if (DeleteTasks.TryDequeue(out DeleteRunnerTask? dtask))
             {
-                _logger.LogInformation($"Current Queue length: {Tasks.Count}");
-                if (task != null)
+                _logger.LogInformation($"Current Queue length: C:{CreateTasks.Count} D:{DeleteTasks.Count}");
+                if (dtask != null)
                 {
-                    bool success = false;
-                    switch (task.Action)
+                    bool success = await DeleteRunner(dtask);
+                    if (!success)
                     {
-                        case RunnerAction.Create:
-                            success = await CreateRunner(task);
-                            break;
-                        case RunnerAction.Delete:
-                            success = await DeleteRunner(task);
-                            break;
+                        // Deletion didn't succeed. Let's hold processing runners for a minute
+                        _logger.LogWarning("Encountered a problem deleting runners. Will hold queue processing for 1 minute.");
+                        await Task.Delay(TimeSpan.FromMinutes(1), stoppingToken);
                     }
+                }
+            }
+
+            if (CreateTasks.TryDequeue(out CreateRunnerTask? task))
+            {
+                _logger.LogInformation($"Current Queue length: C:{CreateTasks.Count} D:{DeleteTasks.Count}");
+                if (task != null)
+                {
+                    bool success = await CreateRunner(task);
                     if (!success)
                     {
                         // Creation didn't succeed. Let's hold of creating new runners for a minute
-                        _logger.LogWarning("Encountered a problem creating runners. Will hold creation for 1 minute.");
+                        _logger.LogWarning("Encountered a problem creating runners. Will hold queue processing for 1 minute.");
                         await Task.Delay(TimeSpan.FromMinutes(1), stoppingToken);
                     }
                 }
@@ -156,7 +164,7 @@ protected override async Task ExecuteAsync(CancellationToken stoppingToken)
         }
     }
 
-    private async Task<bool> DeleteRunner(RunnerTask rt)
+    private async Task<bool> DeleteRunner(DeleteRunnerTask rt)
     {
         try
         {
@@ -168,12 +176,12 @@ private async Task<bool> DeleteRunner(RunnerTask rt)
             _logger.LogError(
                 $"Unable to delete runner [{rt.ServerId} | Retry: {rt.RetryCount}]: {ex.Message}");
             rt.RetryCount += 1;
-            Tasks.Enqueue(rt);
+            DeleteTasks.Enqueue(rt);
             return false;
         }
     }
 
-    private async Task<bool> CreateRunner(RunnerTask rt)
+    private async Task<bool> CreateRunner(CreateRunnerTask rt)
     {
         try
         {
@@ -186,7 +194,7 @@ private async Task<bool> CreateRunner(RunnerTask rt)
         {
             _logger.LogError($"Unable to create runner [{rt.Size} on {rt.Arch} | Retry: {rt.RetryCount}]: {ex.Message}");
             rt.RetryCount += 1;
-            Tasks.Enqueue(rt);
+            CreateTasks.Enqueue(rt);
             return false;
         }
     }

diff --git a/Program.cs b/Program.cs
@@ -196,10 +196,9 @@ private static void JobCompleted(ILogger<Program> logger, long jobId, CloudContr
         }
         else
         {
-            poolMgr.Tasks.Enqueue(new RunnerTask
+            poolMgr.DeleteTasks.Enqueue(new DeleteRunnerTask
             {
-                Action = RunnerAction.Delete,
-                ServerId = vm.Id
+                ServerId = vm.Id,
             });
             ProcessedJobCount.Labels(vm.OrgName, vm.Size).Inc();
 
@@ -265,9 +264,8 @@ private static async Task JobQueued(ILogger<Program> logger, string? repoName, L
             return;
         }
 
-        poolMgr.Tasks.Enqueue(new RunnerTask
+        poolMgr.CreateTasks.Enqueue(new CreateRunnerTask
         {
-            Action = RunnerAction.Create,
             Arch = arch,
             Size = size,
             RunnerToken = runnerToken,

diff --git a/RunnerAction.cs b/RunnerAction.cs