Azure · hallipr · Jul 16, 2024 · Jul 13, 2024
@@ -2,6 +2,7 @@
 using System.Collections.Generic;
 using System.Linq;
 using System.Threading.Tasks;
+using Azure.Sdk.Tools.PipelineWitness.AzurePipelines;
 using Azure.Sdk.Tools.PipelineWitness.Configuration;
 using Azure.Storage.Blobs;
 using Microsoft.Extensions.Logging.Abstractions;
@@ -52,7 +53,7 @@ public async Task BasicBlobProcessInvokesSuccessfully()
             Assert.True(recentBuilds.Count > 0);
             int targetBuildId = recentBuilds.First().Id;
 
-            BlobUploadProcessor processor = new(logger: new NullLogger<BlobUploadProcessor>(),
+            AzurePipelinesProcessor processor = new(logger: new NullLogger<AzurePipelinesProcessor>(),
                 blobServiceClient: blobServiceClient,
                 vssConnection: this.visualStudioConnection,
                 options: Options.Create<PipelineWitnessSettings>(this.testSettings));
@@ -67,7 +68,7 @@ public async Task BasicBlobProcessInvokesSuccessfully()
         [InlineData(0, 10000, 0)]
         public void TestBatching(int startingNumber, int batchSize, int expectedBatchNumber)
         {
-            int numberOfBatches = BlobUploadProcessor.CalculateBatches(startingNumber, batchSize);
+            int numberOfBatches = AzurePipelinesProcessor.CalculateBatches(startingNumber, batchSize);
 
             Assert.Equal(expectedBatchNumber, numberOfBatches);
         }

@@ -6,7 +6,7 @@ namespace Azure.Sdk.Tools.PipelineWitness.Tests
 {
     public class TestLogger : ILogger
     {
-        internal List<object> Logs { get; } = new List<object>();
+        internal List<object> Logs { get; } = [];
 
         public IDisposable BeginScope<TState>(TState state)
         {

@@ -3,6 +3,7 @@
   <PropertyGroup>
     <TargetFramework>net8.0</TargetFramework>
     <UserSecretsId>bc5587e8-3503-4e1a-816c-1e219e4047f6</UserSecretsId>
+    <EnforceCodeStyleInBuild>True</EnforceCodeStyleInBuild>
   </PropertyGroup>
 
   <ItemGroup>

@@ -0,0 +1,48 @@
+using System;
+using System.Threading;
+using System.Threading.Tasks;
+using Azure.Sdk.Tools.PipelineWitness.Configuration;
+using Azure.Sdk.Tools.PipelineWitness.Services;
+using Azure.Sdk.Tools.PipelineWitness.Services.WorkTokens;
+using Microsoft.Extensions.Logging;
+using Microsoft.Extensions.Options;
+
+namespace Azure.Sdk.Tools.PipelineWitness.AzurePipelines
+{
+    public class AzurePipelinesBuildDefinitionWorker : PeriodicLockingBackgroundService
+    {
+        private readonly ILogger<AzurePipelinesBuildDefinitionWorker> logger;
+        private readonly AzurePipelinesProcessor runProcessor;
+        private readonly IOptions<PipelineWitnessSettings> options;
+
+        public AzurePipelinesBuildDefinitionWorker(
+            ILogger<AzurePipelinesBuildDefinitionWorker> logger,
+            AzurePipelinesProcessor runProcessor,
+            IAsyncLockProvider asyncLockProvider,
+            IOptions<PipelineWitnessSettings> options)
+            : base(
+                  logger,
+                  asyncLockProvider,
+                  options.Value.BuildDefinitionWorker)
+        {
+            this.logger = logger;
+            this.runProcessor = runProcessor;
+            this.options = options;
+        }
+
+        protected override async Task ProcessAsync(CancellationToken cancellationToken)
+        {
+            var settings = this.options.Value;
+            foreach (string project in settings.Projects)
+            {
+                await this.runProcessor.UploadBuildDefinitionBlobsAsync(settings.Account, project);
+            }
+        }
+
+        protected override Task ProcessExceptionAsync(Exception ex)
+        {
+            this.logger.LogError(ex, "Error processing build definitions");
+            return Task.CompletedTask;
+        }
+    }
+}
@@ -6,11 +6,11 @@
 using System.Linq;
 using System.Net;
 using System.Text;
-using System.Text.RegularExpressions;
+using System.Threading;
 using System.Threading.Tasks;
 
 using Azure.Sdk.Tools.PipelineWitness.Configuration;
-using Azure.Sdk.Tools.PipelineWitness.Services;
+using Azure.Sdk.Tools.PipelineWitness.Utilities;
 using Azure.Storage.Blobs;
 using Azure.Storage.Blobs.Models;
 
@@ -24,10 +24,10 @@
 using Newtonsoft.Json.Converters;
 using Newtonsoft.Json.Serialization;
 
-namespace Azure.Sdk.Tools.PipelineWitness
+namespace Azure.Sdk.Tools.PipelineWitness.AzurePipelines
 {
     [SuppressMessage("Style", "IDE0037:Use inferred member name", Justification = "Explicit member names are added to json export objects for clarity")]
-    public class BlobUploadProcessor
+    public class AzurePipelinesProcessor
     {
         private const string BuildsContainerName = "builds";
         private const string BuildLogLinesContainerName = "buildloglines";
@@ -39,14 +39,15 @@ public class BlobUploadProcessor
         private const int ApiBatchSize = 10000;
 
         private const string TimeFormat = @"yyyy-MM-dd\THH:mm:ss.fffffff\Z";
+
         private static readonly JsonSerializerSettings jsonSettings = new()
         {
             ContractResolver = new CamelCasePropertyNamesContractResolver(),
             Converters = { new StringEnumConverter(new CamelCaseNamingStrategy()) },
             Formatting = Formatting.None,
         };
 
-        private readonly ILogger<BlobUploadProcessor> logger;
+        private readonly ILogger<AzurePipelinesProcessor> logger;
         private readonly TestResultsHttpClient testResultsClient;
         private readonly BuildHttpClient buildClient;
         private readonly BlobContainerClient buildLogLinesContainerClient;
@@ -57,18 +58,15 @@ public class BlobUploadProcessor
         private readonly BlobContainerClient buildDefinitionsContainerClient;
         private readonly BlobContainerClient pipelineOwnersContainerClient;
         private readonly IOptions<PipelineWitnessSettings> options;
-        private readonly Dictionary<string, int?> cachedDefinitionRevisions = new();
+        private readonly Dictionary<string, int?> cachedDefinitionRevisions = [];
 
-        public BlobUploadProcessor(
-            ILogger<BlobUploadProcessor> logger,
+        public AzurePipelinesProcessor(
+            ILogger<AzurePipelinesProcessor> logger,
             BlobServiceClient blobServiceClient,
             VssConnection vssConnection,
             IOptions<PipelineWitnessSettings> options)
         {
-            if (blobServiceClient == null)
-            {
-                throw new ArgumentNullException(nameof(blobServiceClient));
-            }
+            ArgumentNullException.ThrowIfNull(blobServiceClient);
 
             this.logger = logger ?? throw new ArgumentNullException(nameof(logger));
             this.options = options ?? throw new ArgumentNullException(nameof(options));
@@ -81,10 +79,7 @@ public BlobUploadProcessor(
             this.buildDefinitionsContainerClient = blobServiceClient.GetBlobContainerClient(BuildDefinitionsContainerName);
             this.pipelineOwnersContainerClient = blobServiceClient.GetBlobContainerClient(PipelineOwnersContainerName);
 
-            if (vssConnection == null)
-            {
-                throw new ArgumentNullException(nameof(vssConnection));
-            }
+            ArgumentNullException.ThrowIfNull(vssConnection);
 
             this.buildClient = vssConnection.GetClient<EnhancedBuildHttpClient>();
             this.testResultsClient = vssConnection.GetClient<TestResultsHttpClient>();
@@ -151,8 +146,6 @@ public async Task UploadBuildBlobsAsync(string account, Guid projectId, int buil
                 return;
             }
 
-            await UploadBuildBlobAsync(account, build);
-
             await UploadTestRunBlobsAsync(account, build);
 
             Timeline timeline = await this.buildClient.GetBuildTimelineAsync(projectId, buildId);
@@ -185,13 +178,24 @@ public async Task UploadBuildBlobsAsync(string account, Guid projectId, int buil
             {
                 await UploadPipelineOwnersBlobAsync(account, build, timeline);
             }
+
+            // We upload the build blob last. This allows us to use the existence of the blob as a signal that build processing is complete.
+            await UploadBuildBlobAsync(account, build);
+        }
+
+        public string GetBuildBlobName(Build build)
+        {
+            long changeTime = ((DateTimeOffset)build.LastChangedDate).ToUnixTimeSeconds();
+            string blobName = $"{build.Project.Name}/{build.FinishTime:yyyy/MM/dd}/{build.Id}-{changeTime}.jsonl".ToLower();
+
+            return blobName;
         }
 
         private async Task UploadPipelineOwnersBlobAsync(string account, Build build, Timeline timeline)
         {
             try
             {
-                string blobPath = $"{build.Project.Name}/{build.FinishTime:yyyy/MM/dd}/{build.Id}-{timeline.ChangeId}.jsonl";
+                string blobPath = $"{build.Project.Name}/{build.FinishTime:yyyy/MM/dd}/{build.Id}-{timeline.ChangeId}.jsonl".ToLower();
                 BlobClient blobClient = this.pipelineOwnersContainerClient.GetBlobClient(blobPath);
 
                 if (await blobClient.ExistsAsync())
@@ -314,7 +318,7 @@ public async Task UploadBuildDefinitionBlobsAsync(string account, string project
 
         private async Task UploadBuildDefinitionBlobAsync(string account, BuildDefinition definition)
         {
-            string blobPath = $"{definition.Project.Name}/{definition.Id}-{definition.Revision}.jsonl";
+            string blobPath = $"{definition.Project.Name}/{definition.Id}-{definition.Revision}.jsonl".ToLower();
 
             try
             {
@@ -390,7 +394,7 @@ private List<BuildLogInfo> GetBuildLogInfos(Build build, Timeline timeline, List
         {
             Dictionary<int, BuildLog> logsById = logs.ToDictionary(l => l.Id);
 
-            List<BuildLogInfo> buildLogInfos = new();
+            List<BuildLogInfo> buildLogInfos = [];
 
             foreach (BuildLog log in logs)
             {
@@ -441,9 +445,7 @@ private async Task UploadBuildBlobAsync(string account, Build build)
         {
             try
             {
-                long changeTime = ((DateTimeOffset)build.LastChangedDate).ToUnixTimeSeconds();
-                string blobPath = $"{build.Project.Name}/{build.FinishTime:yyyy/MM/dd}/{build.Id}-{changeTime}.jsonl";
-                BlobClient blobClient = this.buildsContainerClient.GetBlobClient(blobPath);
+                BlobClient blobClient = this.buildsContainerClient.GetBlobClient(GetBuildBlobName(build));
 
                 if (await blobClient.ExistsAsync())
                 {
@@ -493,7 +495,7 @@ private async Task UploadBuildBlobAsync(string account, Build build)
                     SourceBranch = build.SourceBranch,
                     SourceVersion = build.SourceVersion,
                     Status = build.Status,
-                    Tags = build.Tags?.Any() == true ? JsonConvert.SerializeObject(build.Tags, jsonSettings) : null,
+                    Tags = build.Tags?.Count > 0 ? JsonConvert.SerializeObject(build.Tags, jsonSettings) : null,
                     Url = $"https://dev.azure.com/{account}/{build.Project!.Name}/_build/results?buildId={build.Id}",
                     ValidationResults = build.ValidationResults,
                     EtlIngestDate = DateTime.UtcNow,
@@ -522,7 +524,7 @@ private async Task UploadTimelineBlobAsync(string account, Build build, Timeline
                     return;
                 }
 
-                string blobPath = $"{build.Project.Name}/{build.FinishTime:yyyy/MM/dd}/{build.Id}-{timeline.ChangeId}.jsonl";
+                string blobPath = $"{build.Project.Name}/{build.FinishTime:yyyy/MM/dd}/{build.Id}-{timeline.ChangeId}.jsonl".ToLower();
                 BlobClient blobClient = this.buildTimelineRecordsContainerClient.GetBlobClient(blobPath);
 
                 if (await blobClient.ExistsAsync())
@@ -570,7 +572,7 @@ private async Task UploadTimelineBlobAsync(string account, Build build, Timeline
                             TaskName = record.Task?.Name,
                             TaskVersion = record.Task?.Version,
                             Type = record.RecordType,
-                            Issues = record.Issues?.Any() == true
+                            Issues = record.Issues?.Count > 0
                                 ? JsonConvert.SerializeObject(record.Issues, jsonSettings)
                                 : null,
                             EtlIngestDate = DateTime.UtcNow,
@@ -596,7 +598,7 @@ private async Task UploadLogLinesBlobAsync(string account, Build build, BuildLog
             {
                 // we don't use FinishTime in the logs blob path to prevent duplicating logs when processing retries.
                 // i.e.  logs with a given buildid/logid are immutable and retries only add new logs.
-                string blobPath = $"{build.Project.Name}/{build.QueueTime:yyyy/MM/dd}/{build.Id}-{log.LogId}.jsonl";
+                string blobPath = $"{build.Project.Name}/{build.QueueTime:yyyy/MM/dd}/{build.Id}-{log.LogId}.jsonl".ToLower();
                 BlobClient blobClient = this.buildLogLinesContainerClient.GetBlobClient(blobPath);
 
                 if (await blobClient.ExistsAsync())
@@ -631,20 +633,10 @@ private async Task UploadLogLinesBlobAsync(string account, Build build, BuildLog
                         lineNumber += 1;
                         characterCount += line.Length;
 
-                        // log lines usually follow the format:
-                        // 2022-03-30T21:38:38.7007903Z Downloading task: AzureKeyVault (1.200.0)
-                        // Sometimes, there's no leading timestamp, so we'll use the last timestamp we saw.
-                        Match match = Regex.Match(line, @"^([^Z]{20,28}Z) (.*)$");
-
-                        DateTimeOffset timestamp = match.Success
-                            ? DateTime.ParseExact(match.Groups[1].Value, TimeFormat, null,
-                                System.Globalization.DateTimeStyles.AssumeUniversal).ToUniversalTime()
-                            : lastTimeStamp;
+                        var (timestamp, message) = StringUtilities.ParseLogLine(line, lastTimeStamp);
 
                         lastTimeStamp = timestamp;
 
-                        string message = match.Success ? match.Groups[2].Value : line;
-
                         await blobWriter.WriteLineAsync(JsonConvert.SerializeObject(new
                         {
                             OrganizationName = account,
@@ -682,7 +674,6 @@ private async Task UploadTestRunBlobsAsync(string account, Build build)
             try
             {
                 string continuationToken = string.Empty;
-                int[] buildIds = new[] { build.Id };
 
                 DateTime minLastUpdatedDate = build.QueueTime!.Value.AddHours(-1);
                 DateTime maxLastUpdatedDate = build.FinishTime!.Value.AddHours(1);
@@ -705,7 +696,7 @@ private async Task UploadTestRunBlobsAsync(string account, Build build)
                             rangeStart,
                             rangeEnd,
                             continuationToken: continuationToken,
-                            buildIds: buildIds
+                            buildIds: [ build.Id ]
                         );
 
                         foreach (TestRun testRun in page)
@@ -731,7 +722,7 @@ private async Task UploadTestRunBlobAsync(string account, Build build, TestRun t
         {
             try
             {
-                string blobPath = $"{build.Project.Name}/{testRun.CompletedDate:yyyy/MM/dd}/{testRun.Id}.jsonl";
+                string blobPath = $"{build.Project.Name}/{testRun.CompletedDate:yyyy/MM/dd}/{testRun.Id}.jsonl".ToLower();
                 BlobClient blobClient = this.testRunsContainerClient.GetBlobClient(blobPath);
 
                 if (await blobClient.ExistsAsync())
@@ -814,7 +805,7 @@ private async Task UploadTestRunResultBlobAsync(string account, Build build, Tes
         {
             try
             {
-                string blobPath = $"{build.Project.Name}/{testRun.CompletedDate:yyyy/MM/dd}/{testRun.Id}.jsonl";
+                string blobPath = $"{build.Project.Name}/{testRun.CompletedDate:yyyy/MM/dd}/{testRun.Id}.jsonl".ToLower();
                 BlobClient blobClient = this.testResultsContainerClient.GetBlobClient(blobPath);
 
                 if (await blobClient.ExistsAsync())
@@ -824,7 +815,7 @@ private async Task UploadTestRunResultBlobAsync(string account, Build build, Tes
                 }
 
                 StringBuilder builder = new();
-                int batchCount = BlobUploadProcessor.CalculateBatches(testRun.TotalTests, batchSize: ApiBatchSize);
+                int batchCount = AzurePipelinesProcessor.CalculateBatches(testRun.TotalTests, batchSize: ApiBatchSize);
 
                 for (int batchMultiplier = 0; batchMultiplier < batchCount; batchMultiplier++)
                 {

@@ -0,0 +1,27 @@
+using System.Text.Json;
+using System.Threading.Tasks;
+using Azure.Sdk.Tools.PipelineWitness.Configuration;
+using Azure.Storage.Queues;
+using Azure.Storage.Queues.Models;
+using Microsoft.Extensions.Logging;
+using Microsoft.Extensions.Options;
+
+namespace Azure.Sdk.Tools.PipelineWitness.AzurePipelines;
+
+public class BuildCompleteQueue
+{
+    private readonly ILogger<BuildCompleteQueue> logger;
+    private readonly QueueClient queueClient;
+
+    public BuildCompleteQueue(ILogger<BuildCompleteQueue> logger, QueueServiceClient queueServiceClient, IOptions<PipelineWitnessSettings> options)
+    {
+        this.logger = logger;
+        this.queueClient = queueServiceClient.GetQueueClient(options.Value.BuildCompleteQueueName);
+    }
+
+    public async Task EnqueueMessageAsync(BuildCompleteQueueMessage message)
+    {
+        SendReceipt response = await this.queueClient.SendMessageAsync(JsonSerializer.Serialize(message));
+        this.logger.LogDebug("Message added to queue with id {MessageId}", response.MessageId);
+    }
+}
@@ -0,0 +1,13 @@
+using System;
+
+namespace Azure.Sdk.Tools.PipelineWitness.AzurePipelines
+{
+    public class BuildCompleteQueueMessage
+    {
+        public string Account { get; set; }
+
+        public Guid ProjectId { get; set; }
+
+        public int BuildId { get; set; }
+    }
+}