Add code owner processing to PipelineWitness (#3742)

Author: yeojunh

tools/pipeline-witness/Azure.Sdk.Tools.PipelineWitness/BlobUploadProcessor.cs

@@ -1,9 +1,12 @@
-namespace Azure.Sdk.Tools.PipelineWitness
+using System.Diagnostics;
+
+namespace Azure.Sdk.Tools.PipelineWitness
 {
     using System;
     using System.Collections.Generic;
     using System.Collections.Immutable;
     using System.IO;
+    using System.IO.Compression;
     using System.Linq;
     using System.Net;
     using System.Text;
@@ -32,6 +35,7 @@ public class BlobUploadProcessor
         private const string BuildTimelineRecordsContainerName = "buildtimelinerecords";
         private const string BuildDefinitionsContainerName = "builddefinitions";
         private const string BuildFailuresContainerName = "buildfailures";
+        private const string PipelineOwnersContainerName = "pipelineowners";
         private const string TestRunsContainerName = "testruns";
 
         private const string TimeFormat = @"yyyy-MM-dd\THH:mm:ss.fffffff\Z";
@@ -52,6 +56,7 @@ public class BlobUploadProcessor
         private readonly BlobContainerClient testRunsContainerClient;
         private readonly BlobContainerClient buildDefinitionsContainerClient;
         private readonly BlobContainerClient buildFailuresContainerClient;
+        private readonly BlobContainerClient pipelineOwnersContainerClient; 
         private readonly IOptions<PipelineWitnessSettings> options;
         private readonly Dictionary<string, int?> cachedDefinitionRevisions = new();
         private readonly IFailureAnalyzer failureAnalyzer;        
@@ -82,6 +87,7 @@ public BlobUploadProcessor(
             this.buildFailuresContainerClient = blobServiceClient.GetBlobContainerClient(BuildFailuresContainerName);
             this.testRunsContainerClient = blobServiceClient.GetBlobContainerClient(TestRunsContainerName);
             this.buildDefinitionsContainerClient = blobServiceClient.GetBlobContainerClient(BuildDefinitionsContainerName);
+            this.pipelineOwnersContainerClient = blobServiceClient.GetBlobContainerClient(PipelineOwnersContainerName);
             this.failureAnalyzer = failureAnalyzer;
         }
 
@@ -176,6 +182,119 @@ public async Task UploadBuildBlobsAsync(string account, Guid projectId, int buil
             {
                 await UploadLogLinesBlobAsync(account, build, log);
             }
+
+            if (build.Definition.Id == options.Value.PipelineOwnersDefinitionId)
+            {
+                await UploadPipelineOwnersBlobAsync(account, build, timeline); 
+            }
+        }
+
+        private async Task UploadPipelineOwnersBlobAsync(string account, Build build, Timeline timeline)
+        {
+            try
+            {
+                var blobPath = $"{build.Project.Name}/{build.FinishTime:yyyy/MM/dd}/{build.Id}-{timeline.ChangeId}.jsonl";
+                var blobClient = this.pipelineOwnersContainerClient.GetBlobClient(blobPath);
+
+                if (await blobClient.ExistsAsync())
+                {
+                    this.logger.LogInformation("Skipping existing build failure blob for build {BuildId}", build.Id);
+                    return;
+                }
+
+                var owners = await GetOwnersFromBuildArtifactAsync(build);
+
+                if (owners == null)
+                {
+                    // no need to log anything here. GetOwnersFromBuildArtifactAsync logs a warning before returning null;
+                    return;
+                }
+
+                this.logger.LogInformation("Creating owners blob for build {DefinitionId} change {ChangeId}", build.Id, timeline.ChangeId);
+
+                var stringBuilder = new StringBuilder();
+
+                foreach (var owner in owners)
+                {
+                    var contentLine = JsonConvert.SerializeObject(new
+                    {
+                        OrganizationName = account,
+                        BuildDefinitionId = owner.Key,
+                        Owners = owner.Value,
+                        Timestamp = new DateTimeOffset(build.FinishTime.Value).ToUniversalTime(),
+                        EtlIngestDate = DateTimeOffset.UtcNow
+                    }, jsonSettings);
+
+                    stringBuilder.AppendLine(contentLine);
+                }
+
+                await blobClient.UploadAsync(new BinaryData(stringBuilder.ToString()));
+            }
+            catch (RequestFailedException ex) when (ex.Status == (int)HttpStatusCode.Conflict)
+            {
+                this.logger.LogInformation("Ignoring exception from existing owners blob for build {BuildId}", build.Id);
+            }
+            catch (Exception ex)
+            {
+                this.logger.LogError(ex, "Error processing owners artifact from build {BuildId}", build.Id);
+                throw;
+            }
+        }
+
+        private async Task<Dictionary<int, string[]>> GetOwnersFromBuildArtifactAsync(Build build)
+        {
+            var artifactName = this.options.Value.PipelineOwnersArtifactName;
+            var filePath = this.options.Value.PipelineOwnersFilePath;
+
+            try
+            {
+                await using var artifactStream = await this.buildClient.GetArtifactContentZipAsync(build.Project.Id, build.Id, artifactName);
+                using var zip = new ZipArchive(artifactStream);
+
+                var fileEntry = zip.GetEntry(filePath);
+
+                if (fileEntry == null)
+                {
+                    this.logger.LogWarning("Artifact {ArtifactName} in build {BuildId} didn't contain the expected file {FilePath}", artifactName, build.Id, filePath);
+                    return null;
+                }
+
+                await using var contentStream = fileEntry.Open();
+                using var contentReader = new StreamReader(contentStream);
+                var content = await contentReader.ReadToEndAsync();
+
+                if (string.IsNullOrEmpty(content))
+                {
+                    this.logger.LogWarning("The file {filePath} in artifact {ArtifactName} in build {BuildId} contained no content", filePath, artifactName, build.Id);
+                    return null;
+                }
+
+                var ownersDictionary = JsonConvert.DeserializeObject<Dictionary<int, string[]>>(content);
+
+                if (ownersDictionary == null)
+                {
+                    this.logger.LogWarning("The file {filePath} in artifact {ArtifactName} in build {BuildId} contained a null json object", filePath, artifactName, build.Id);
+                }
+
+                return ownersDictionary;
+            }
+            catch (ArtifactNotFoundException ex)
+            {
+                this.logger.LogWarning(ex, "Build {BuildId} did not contain the expected artifact {ArtifactName}", build.Id, artifactName);
+            }
+            catch (InvalidDataException ex)
+            {
+                this.logger.LogWarning(ex, "Unable to read ZIP contents from artifact {ArtifactName} in build {BuildId}", artifactName, build.Id);
+
+                // rethrow the exception so the queue message will be retried.
+                throw;
+            }
+            catch (JsonSerializationException ex)
+            {
+                this.logger.LogWarning(ex, "Problem deserializing JSON from artifact {ArtifactName} in build {BuildId}", artifactName, build.Id);
+            }
+
+            return null;
         }
 
         private async Task UploadBuildFailureBlobAsync(string account, Build build, Timeline timeline)
@@ -217,7 +336,8 @@ private async Task UploadBuildFailureBlobAsync(string account, Build build, Time
                         RecordId = failure.Record.Id,
                         BuildTimelineId = timeline.Id,
                         ErrorClassification = failure.Classification,
-                     }, jsonSettings);
+                        EtlIngestDate = DateTimeOffset.UtcNow
+                    }, jsonSettings);
                      stringBuilder.AppendLine(contentLine);
                 }
 
@@ -241,16 +361,16 @@ public async Task UploadBuildDefinitionBlobsAsync(string account, string project
             foreach (var definition in definitions)
             {
                 var cacheKey = $"{definition.Project.Id}:{definition.Id}";
-                
+
                 if (!this.cachedDefinitionRevisions.TryGetValue(cacheKey, out var cachedRevision) || cachedRevision != definition.Revision)
-                { 
+                {
                     await UploadBuildDefinitionBlobAsync(account, definition);
                 }
 
                 this.cachedDefinitionRevisions[cacheKey] = definition.Revision;
             }
         }
-
+        
         private async Task UploadBuildDefinitionBlobAsync(string account, BuildDefinition definition)
         {
             var blobPath = $"{definition.Project.Name}/{definition.Id}-{definition.Revision}.jsonl";

tools/pipeline-witness/Azure.Sdk.Tools.PipelineWitness/PipelineWitnessSettings.cs

@@ -65,5 +65,20 @@ public class PipelineWitnessSettings
         /// Gets or sets the number of concurrent build complete queue workers to register
         /// </summary>
         public int BuildCompleteWorkerCount { get; set; } = 1;
+
+        /// <summary>
+        /// Gets or sets the artifact name used by the pipeline owners extraction build 
+        /// </summary>
+        public string PipelineOwnersArtifactName { get; set; }
+
+        /// <summary>
+        /// Gets or sets the file name used by the pipeline owners extraction build 
+        /// </summary>
+        public string PipelineOwnersFilePath { get; set; }
+
+        /// <summary>
+        /// Gets or sets the definition id of the pipeline owners extraction build 
+        /// </summary>
+        public int PipelineOwnersDefinitionId { get; set; }
     }
 }

tools/pipeline-witness/Azure.Sdk.Tools.PipelineWitness/Services/EnhancedBuildHttpClient.cs

@@ -0,0 +1,68 @@
+using System;
+using System.IO;
+using System.Net.Http;
+using System.Threading;
+using System.Threading.Tasks;
+
+using Microsoft.TeamFoundation.Build.WebApi;
+using Microsoft.VisualStudio.Services.Common;
+
+namespace Azure.Sdk.Tools.PipelineWitness.Services;
+
+public class EnhancedBuildHttpClient : BuildHttpClient
+
+{
+    public EnhancedBuildHttpClient(Uri baseUrl, VssCredentials credentials)
+        : base(baseUrl, credentials)
+    {}
+
+    public EnhancedBuildHttpClient(Uri baseUrl, VssCredentials credentials, VssHttpRequestSettings settings)
+        : base(baseUrl, credentials, settings)
+    {}
+
+    public EnhancedBuildHttpClient(Uri baseUrl, VssCredentials credentials, params DelegatingHandler[] handlers)
+        : base(baseUrl, credentials, handlers)
+    {}
+
+    public EnhancedBuildHttpClient(Uri baseUrl, VssCredentials credentials, VssHttpRequestSettings settings, params DelegatingHandler[] handlers)
+        : base(baseUrl, credentials, settings, handlers)
+    {}
+
+    public EnhancedBuildHttpClient(Uri baseUrl, HttpMessageHandler pipeline, bool disposeHandler)
+        : base(baseUrl, pipeline, disposeHandler)
+    {}
+
+    public override async Task<Stream> GetArtifactContentZipAsync(
+        Guid project,
+        int buildId,
+        string artifactName,
+        object userState = null,
+        CancellationToken cancellationToken = default)
+    {
+        var artifact = await base.GetArtifactAsync(project, buildId, artifactName, userState, cancellationToken);
+        return await GetArtifactContentZipAsync(artifact, cancellationToken);
+    }
+
+    public override async Task<Stream> GetArtifactContentZipAsync(
+        string project,
+        int buildId,
+        string artifactName,
+        object userState = null,
+        CancellationToken cancellationToken = default)
+    {
+        var artifact = await base.GetArtifactAsync(project, buildId, artifactName, userState, cancellationToken);
+        return await GetArtifactContentZipAsync(artifact, cancellationToken);
+    }
+
+    private async Task<Stream> GetArtifactContentZipAsync(BuildArtifact artifact, CancellationToken cancellationToken)
+    {
+        var downloadUrl = artifact?.Resource?.DownloadUrl;
+        if (string.IsNullOrWhiteSpace(downloadUrl))
+        {
+            throw new InvalidArtifactDataException("Artifact contained no download url");
+        }
+
+        var responseStream = await Client.GetStreamAsync(downloadUrl, cancellationToken);
+        return responseStream;
+    }
+}

tools/pipeline-witness/Azure.Sdk.Tools.PipelineWitness/Startup.cs

@@ -49,7 +49,7 @@ public static void Configure(WebApplicationBuilder builder)
             });
 
             builder.Services.AddSingleton(provider => provider.GetRequiredService<VssConnection>().GetClient<ProjectHttpClient>());
-            builder.Services.AddSingleton(provider => provider.GetRequiredService<VssConnection>().GetClient<BuildHttpClient>());
+            builder.Services.AddSingleton<BuildHttpClient>(provider => provider.GetRequiredService<VssConnection>().GetClient<EnhancedBuildHttpClient>());
             builder.Services.AddSingleton(provider => provider.GetRequiredService<VssConnection>().GetClient<TestResultsHttpClient>());
 
             builder.Services.AddLogging();

tools/pipeline-witness/Azure.Sdk.Tools.PipelineWitness/appsettings.json

@@ -20,6 +20,9 @@
     "MaxDequeueCount": 5,
     "Account": "azure-sdk",
     "Projects": [ "internal", "playground", "public" ],
-    "BuildDefinitionLoopPeriod": "00:05:00"
+    "BuildDefinitionLoopPeriod": "00:05:00",
+    "PipelineOwnersArtifactName": "pipelineOwners",
+    "PipelineOwnersFilePath": "pipelineOwners/pipelineOwners.json",
+    "PipelineOwnersDefinitionId": 5112
   }
 }