Append tes tasks to JSONL

src/TesApi.Web/AzureProxy.cs

@@ -5,6 +5,7 @@
 using System.Collections.Generic;
 using System.IO;
 using System.Linq;
+using System.Text;
 using System.Text.RegularExpressions;
 using System.Threading;
 using System.Threading.Tasks;
@@ -20,6 +21,7 @@
 using Microsoft.Extensions.Logging;
 using Microsoft.Extensions.Options;
 using Microsoft.Rest;
+using Microsoft.WindowsAzure.Storage;
 using Microsoft.WindowsAzure.Storage.Blob;
 using Newtonsoft.Json;
 using Polly;
@@ -672,6 +674,49 @@ public async Task<string> GetStorageAccountKeyAsync(StorageAccountInfo storageAc
             }
         }
 
+        public async Task UploadBlockToTesTasksAppendBlobAsync(Uri containerAbsoluteUri, string content, CancellationToken cancellationToken)
+        {
+            var container = new CloudBlobContainer(containerAbsoluteUri);
+            CloudAppendBlob appendBlob = null;
+            const int maxBlocks = 50000;
+
+            for (int i = 0; i < int.MaxValue; i++)
+            {
+                var blobName = $"testasks{i}.jsonl";
+                appendBlob = container.GetAppendBlobReference(blobName);
+
+                // If the blob exists and has less than the max number of blocks, use it.
+                if (await appendBlob.ExistsAsync(null, null, cancellationToken))
+                {
+                    await appendBlob.FetchAttributesAsync(null, null, null, cancellationToken);
+
+                    if (appendBlob.Properties.AppendBlobCommittedBlockCount < maxBlocks)
+                    {
+                        break;
+                    }
+
+                    continue; // If the blob has max blocks, continue to the next blob.
+                }
+
+                // If the blob does not exist, try creating it.
+                try
+                {
+                    await appendBlob.CreateOrReplaceAsync(null, null, null, cancellationToken);
+                    break; // If successful, break the loop.
+                }
+                catch (StorageException ex) when (ex.RequestInformation?.HttpStatusCode == 409) // Conflict
+                {
+                    // If there's a conflict, another process might have created the blob, so continue the loop.
+                    continue;
+                }
+            }
+
+            // Append the text content to the blob.
+            using var stream = new MemoryStream(Encoding.UTF8.GetBytes(content));
+            await appendBlob.AppendBlockAsync(stream, null, null, null, null, cancellationToken);
+        }
+
+
         /// <inheritdoc/>
         public Task UploadBlobAsync(Uri blobAbsoluteUri, string content, CancellationToken cancellationToken)
             => new CloudBlockBlob(blobAbsoluteUri).UploadTextAsync(content, null, null, null, null, cancellationToken);

src/TesApi.Web/CachingWithRetriesAzureProxy.cs

@@ -229,5 +229,10 @@ public Task DeleteBatchPoolIfExistsAsync(string poolId, CancellationToken cancel
 
         /// <inheritdoc/>
         public Task EnableBatchPoolAutoScaleAsync(string poolId, bool preemptable, TimeSpan interval, IAzureProxy.BatchPoolAutoScaleFormulaFactory formulaFactory, CancellationToken cancellationToken) => azureProxy.EnableBatchPoolAutoScaleAsync(poolId, preemptable, interval, formulaFactory, cancellationToken);
+
+        public Task UploadBlockToTesTasksAppendBlobAsync(Uri containerAbsoluteUri, string content, CancellationToken cancellationToken)
+        {
+            throw new NotImplementedException();
+        }
     }
 }

src/TesApi.Web/IAzureProxy.cs

@@ -324,5 +324,14 @@ public interface IAzureProxy
         /// <param name="currentTarget">Current number of compute nodes.</param>
         /// <returns></returns>
         delegate string BatchPoolAutoScaleFormulaFactory(bool preemptable, int currentTarget);
+
+        /// <summary>
+        /// Appends TES task JSON to an existing blob, or creates a new one
+        /// </summary>
+        /// <param name="containerAbsoluteUri"></param>
+        /// <param name="content"></param>
+        /// <param name="cancellationToken"></param>
+        /// <returns></returns>
+        Task UploadBlockToTesTasksAppendBlobAsync(Uri containerAbsoluteUri, string content, CancellationToken cancellationToken);
     }
 }

src/TesApi.Web/Scheduler.cs

@@ -2,12 +2,14 @@
 // Licensed under the MIT License.
 
 using System;
+using System.Collections.Concurrent;
 using System.Collections.Generic;
 using System.Linq;
 using System.Threading;
 using System.Threading.Tasks;
 using Microsoft.Extensions.Hosting;
 using Microsoft.Extensions.Logging;
+using Newtonsoft.Json;
 using Tes.Extensions;
 using Tes.Models;
 using Tes.Repository;
@@ -25,17 +27,18 @@ public class Scheduler : BackgroundService
         private readonly IBatchScheduler batchScheduler;
         private readonly ILogger<Scheduler> logger;
         private readonly TimeSpan runInterval = TimeSpan.FromSeconds(5);
-
+        private readonly ITesTaskAppender tesTaskAppender;
         /// <summary>
         /// Default constructor
         /// </summary>
         /// <param name="repository">The main TES task database repository implementation</param>
         /// <param name="batchScheduler">The batch scheduler implementation</param>
         /// <param name="logger">The logger instance</param>
-        public Scheduler(IRepository<TesTask> repository, IBatchScheduler batchScheduler, ILogger<Scheduler> logger)
+        public Scheduler(IRepository<TesTask> repository, IBatchScheduler batchScheduler, ITesTaskAppender tesTaskAppender, ILogger<Scheduler> logger)
         {
             this.repository = repository;
             this.batchScheduler = batchScheduler;
+            this.tesTaskAppender = tesTaskAppender;
             this.logger = logger;
         }
 
@@ -199,6 +202,11 @@ private async ValueTask OrchestrateTesTasksOnBatch(CancellationToken stoppingTok
                         }
 
                         await repository.UpdateItemAsync(tesTask, stoppingToken);
+
+                        if (!tesTask.IsActiveState())
+                        {
+                            tesTaskAppender.Append(tesTask);
+                        }
                     }
                 }
                 catch (RepositoryCollisionException exc)

src/TesApi.Web/TesTaskAppender.cs

@@ -0,0 +1,66 @@
+// Copyright (c) Microsoft Corporation.
+// Licensed under the MIT License.
+
+using System;
+using System.Collections.Concurrent;
+using System.Threading;
+using System.Threading.Tasks;
+using Microsoft.Extensions.Hosting;
+using Microsoft.Extensions.Logging;
+using Newtonsoft.Json;
+using Tes.Models;
+
+namespace TesApi.Web
+{
+    /// <inheritdoc/>
+    public interface ITesTaskAppender
+    {
+        /// <inheritdoc/>
+        void Append(TesTask tesTask);
+    }
+
+    /// <inheritdoc/>
+    public class TesTaskAppender : BackgroundService, ITesTaskAppender
+    {
+        private readonly ConcurrentQueue<TesTask> tesTaskJsonAppendQueue = new ConcurrentQueue<TesTask>();
+        private readonly ILogger<TesTaskAppender> logger;
+        private readonly IAzureProxy azureProxy;
+
+        /// <inheritdoc/>
+        public TesTaskAppender(ILogger<TesTaskAppender> logger, IAzureProxy azureProxy)
+        {
+            this.logger = logger;
+            this.azureProxy = azureProxy;
+        }
+
+        /// <inheritdoc/>
+        public void Append(TesTask tesTask)
+        {
+            tesTaskJsonAppendQueue.Enqueue(tesTask);
+        }
+
+        /// <inheritdoc/>
+        protected override async Task ExecuteAsync(CancellationToken stoppingToken)
+        {
+            while (!stoppingToken.IsCancellationRequested)
+            {
+                while (!stoppingToken.IsCancellationRequested && tesTaskJsonAppendQueue.TryDequeue(out TesTask tesTask))
+                {
+                    var jsonl = JsonConvert.SerializeObject(tesTask) + "\n";
+
+                    try
+                    {
+                        // TODO auth, retries, exception handling
+                        await azureProxy.UploadBlockToTesTasksAppendBlobAsync(null, jsonl, stoppingToken);
+                    }
+                    catch (Exception exc)
+                    {
+                        logger.LogError(exc, exc.Message);
+                    }
+                }
+
+                await Task.Delay(TimeSpan.FromSeconds(1));
+            }
+        }
+    }
+}