Add telemetry for chunk-dedup build artifacts uploads (#3375)

* Add telemetry for chunk-dedup build artifacts uploads

* fix tests

* better design

* cleanup

* Don't need this anymore

* cleanup

* cleanup

* fix test

* tryparse

src/Agent.Plugins/Artifact/FileContainerProvider.cs

@@ -2,7 +2,6 @@
 // Licensed under the MIT License.
 
 using Agent.Sdk;
-using Agent.Sdk.Blob;
 using Agent.Plugins.PipelineArtifact.Telemetry;
 using BuildXL.Cache.ContentStore.Hashing;
 using Microsoft.TeamFoundation.Build.WebApi;

src/Agent.Plugins/Artifact/FileShareProvider.cs

@@ -11,7 +11,6 @@
 using System.Threading.Tasks.Dataflow;
 using Agent.Plugins.PipelineArtifact.Telemetry;
 using Agent.Sdk;
-using Agent.Sdk.Blob;
 using Microsoft.TeamFoundation.Build.WebApi;
 using Microsoft.VisualStudio.Services.Agent.Blob;
 using Microsoft.VisualStudio.Services.Agent.Util;

src/Agent.Plugins/PipelineArtifact/Telemetry/PipelineArtifactActionRecord.cs

@@ -3,7 +3,7 @@
 
 using System;
 using Agent.Sdk;
-using Agent.Sdk.Blob;
+using Microsoft.VisualStudio.Services.Agent.Blob;
 using Microsoft.VisualStudio.Services.Content.Common.Telemetry;
 using Microsoft.VisualStudio.Services.BlobStore.WebApi;
 

src/Agent.Worker/Build/BuildArtifactActionRecord.cs

@@ -2,7 +2,7 @@
 // Licensed under the MIT License.
 
 using System;
-using Agent.Sdk.Blob;
+using Microsoft.VisualStudio.Services.Agent.Blob;
 using Microsoft.VisualStudio.Services.Content.Common.Telemetry;
 
 namespace Microsoft.VisualStudio.Services.Agent.Worker.Build

src/Agent.Worker/Build/FileContainerServer.cs

@@ -15,12 +15,12 @@
 using Microsoft.VisualStudio.Services.WebApi;
 using System.Net.Http;
 using System.Net;
-using Agent.Sdk.Blob;
 using BuildXL.Cache.ContentStore.Hashing;
+using Microsoft.TeamFoundation.DistributedTask.WebApi;
 using Microsoft.VisualStudio.Services.BlobStore.WebApi;
-using Microsoft.VisualStudio.Services.Content.Common;
-using Microsoft.VisualStudio.Services.BlobStore.Common;
+using Microsoft.VisualStudio.Services.Agent.Worker.Telemetry;
 using Microsoft.VisualStudio.Services.BlobStore.Common.Telemetry;
+using Microsoft.VisualStudio.Services.WebPlatform;
 
 namespace Microsoft.VisualStudio.Services.Agent.Worker.Build
 {
@@ -32,7 +32,7 @@ public class FileContainerServer
         private readonly FileContainerHttpClient _fileContainerHttpClient;
         private readonly VssConnection _connection;
         private DedupStoreClient _dedupClient;
-        private BlobStoreClientTelemetry _blobTelemetry;
+        private BlobStoreClientTelemetryTfs _blobTelemetry;
 
         private CancellationTokenSource _uploadCancellationTokenSource;
         private TaskCompletionSource<int> _uploadFinished;
@@ -204,6 +204,20 @@ private async Task<UploadResult> ParallelUploadAsync(IAsyncCommandContext contex
             _uploadFinished.TrySetResult(0);
             await uploadMonitor;
 
+            // report telemetry
+            if (uploadToBlob)
+            {
+                if (!Guid.TryParse(context.GetVariableValueOrDefault(WellKnownDistributedTaskVariables.PlanId), out var planId))
+                {
+                    planId = Guid.Empty;
+                }
+                if (!Guid.TryParse(context.GetVariableValueOrDefault(WellKnownDistributedTaskVariables.JobId), out var jobId))
+                {
+                    jobId = Guid.Empty;
+                }
+                await _blobTelemetry.CommitTelemetry(planId, jobId);
+            }
+
             return uploadResult;
         }
 
@@ -325,7 +339,7 @@ private async Task<UploadResult> UploadAsync(IAsyncCommandContext context, int u
         {
             var verbose = String.Equals(context.GetVariableValueOrDefault("system.debug"), "true", StringComparison.InvariantCultureIgnoreCase);
 
-            return await BlobStoreUtils.UploadToBlobStore<BuildArtifactActionRecord>(verbose, itemPath, (level, uri, type) =>
+            return await BlobStoreUtils.UploadToBlobStore(verbose, itemPath, (level, uri, type) =>
                 new BuildArtifactActionRecord(level, uri, type, nameof(UploadToBlobStore), context), (str) => context.Output(str), _dedupClient, _blobTelemetry, cancellationToken);
         }
 

src/Microsoft.VisualStudio.Services.Agent/Blob/BlobStoreClientTelemetryTfs.cs

@@ -0,0 +1,37 @@
+// Copyright (c) Microsoft Corporation.
+// Licensed under the MIT License.
+
+using System;
+using System.Threading.Tasks;
+using Microsoft.VisualStudio.Services.WebApi;
+using Microsoft.VisualStudio.Services.Content.Common.Tracing;
+using Microsoft.VisualStudio.Services.Content.Common.Telemetry;
+using Microsoft.VisualStudio.Services.BlobStore.Common.Telemetry;
+
+namespace Microsoft.VisualStudio.Services.Agent.Blob
+{
+    public class BlobStoreClientTelemetryTfs : BlobStoreClientTelemetry
+    {
+        private CustomerIntelligenceTelemetrySender _ciSender;
+
+        public BlobStoreClientTelemetryTfs(IAppTraceSource tracer, Uri baseAddress, VssConnection connection) 
+            : base(tracer, baseAddress)
+        {
+            _ciSender = new CustomerIntelligenceTelemetrySender(connection);
+            this.senders.Add(_ciSender);
+        }
+
+        // for testing
+        public BlobStoreClientTelemetryTfs(IAppTraceSource tracer, Uri baseAddress, VssConnection connection, ITelemetrySender sender)
+            : base(tracer, baseAddress, sender)
+        {
+            _ciSender = new CustomerIntelligenceTelemetrySender(connection);
+            this.senders.Add(_ciSender);
+        }
+
+        public async Task CommitTelemetry(Guid planId, Guid jobId)
+        {
+            await _ciSender.CommitTelemetry(planId, jobId);
+        }
+    }
+}

src/Microsoft.VisualStudio.Services.Agent/Blob/BlobStoreUtils.cs

@@ -26,14 +26,14 @@ namespace Microsoft.VisualStudio.Services.Agent.Blob
     /// </summary>
     public static class BlobStoreUtils
     {
-        public static async Task<(DedupIdentifier dedupId, ulong length)> UploadToBlobStore<T>(
+        public static async Task<(DedupIdentifier dedupId, ulong length)> UploadToBlobStore(
             bool verbose,
             string itemPath,
-            Func<TelemetryInformationLevel, Uri, string, T> telemetryRecordFactory,
+            Func<TelemetryInformationLevel, Uri, string, BlobStoreTelemetryRecord> telemetryRecordFactory,
             Action<string> traceOutput,
             DedupStoreClient dedupClient,
             BlobStoreClientTelemetry clientTelemetry,
-            CancellationToken cancellationToken) where T : BlobStoreTelemetryRecord
+            CancellationToken cancellationToken)
         {
             // Create chunks and identifier
             var chunk = await ChunkerHelper.CreateFromFileAsync(FileSystem.Instance, itemPath, cancellationToken, false);
@@ -49,13 +49,14 @@ public static class BlobStoreUtils
             var uploadSession = dedupClient.CreateUploadSession(keepUntilRef, tracer, FileSystem.Instance);
 
             // Upload the chunks
-            var uploadRecord = clientTelemetry.CreateRecord<T>(telemetryRecordFactory);
+            var uploadRecord = clientTelemetry.CreateRecord<BlobStoreTelemetryRecord>(telemetryRecordFactory);
             await clientTelemetry.MeasureActionAsync(
                 record: uploadRecord,
                 actionAsync: async () => await AsyncHttpRetryHelper.InvokeAsync(
                         async () =>
                         {
-                            return await uploadSession.UploadAsync(rootNode, new Dictionary<DedupIdentifier, string>(){ [dedupId] = itemPath }, cancellationToken);
+                            await uploadSession.UploadAsync(rootNode, new Dictionary<DedupIdentifier, string>(){ [dedupId] = itemPath }, cancellationToken);
+                            return uploadSession.UploadStatistics;
                         },
                         maxRetries: 3,
                         tracer: tracer,
@@ -65,19 +66,5 @@ await clientTelemetry.MeasureActionAsync(
             );
             return (dedupId, rootNode.TransitiveContentBytes);
         }
-
-        public static async Task<(DedupIdentifier dedupId, ulong length)> UploadToBlobStore<T>(
-            bool verbose,
-            string itemPath,
-            Func<TelemetryInformationLevel, Uri, string, T> telemetryRecordFactory,
-            Action<string> traceOutput,
-            VssConnection connection,
-            CancellationToken cancellationToken) where T : BlobStoreTelemetryRecord
-        {
-            var (dedupClient, clientTelemetry) = await DedupManifestArtifactClientFactory.Instance
-                    .CreateDedupClientAsync(verbose, traceOutput, connection, cancellationToken);
-
-            return await UploadToBlobStore<T>(verbose, itemPath, telemetryRecordFactory, traceOutput, dedupClient, clientTelemetry, cancellationToken);
-        }
     }
 }

src/Microsoft.VisualStudio.Services.Agent/Blob/CustomerIntelligenceTelemetrySender.cs

@@ -0,0 +1,85 @@
+// Copyright (c) Microsoft Corporation.
+// Licensed under the MIT License.
+
+using System;
+using System.Collections.Generic;
+using System.Threading.Tasks;
+using Microsoft.VisualStudio.Services.WebApi;
+using Microsoft.VisualStudio.Services.Agent.Util;
+using Microsoft.VisualStudio.Services.Content.Common.Telemetry;
+using Microsoft.VisualStudio.Services.CustomerIntelligence.WebApi;
+using Microsoft.VisualStudio.Services.WebPlatform;
+
+namespace Microsoft.VisualStudio.Services.Agent.Blob
+{
+    public class CustomerIntelligenceTelemetrySender : ITelemetrySender
+    {
+        private CustomerIntelligenceHttpClient _ciClient;
+
+        private long _chunksUploaded = 0;
+        private long _compressionBytesSaved = 0;
+        private long _dedupUploadBytesSaved = 0;
+        private long _logicalContentBytesUploaded = 0;
+        private long _physicalContentBytesUploaded = 0;
+        private long _totalNumberOfChunks = 0;
+
+        public CustomerIntelligenceTelemetrySender(VssConnection connection)
+        {
+            ArgUtil.NotNull(connection, nameof(connection));
+            _ciClient = connection.GetClient<CustomerIntelligenceHttpClient>();
+        }
+
+        // Not used by the interface. We just want to capture successful telemetry for dedup analytics
+        public void StartSender()
+        {
+        }
+        public void StopSender()
+        {
+        }
+        public void SendErrorTelemetry(ErrorTelemetryRecord errorTelemetry)
+        {
+        }
+        public void SendRecord(TelemetryRecord record)
+        {
+        }
+
+        public void SendActionTelemetry(ActionTelemetryRecord actionTelemetry)
+        {
+            if (actionTelemetry is IDedupRecord dedupRecord)
+            {
+                var uploadStats = dedupRecord.UploadStatistics;
+                if (uploadStats != null)
+                {
+                    this._chunksUploaded += uploadStats.ChunksUploaded;
+                    this._compressionBytesSaved += uploadStats.CompressionBytesSaved;
+                    this._dedupUploadBytesSaved += uploadStats.DedupUploadBytesSaved;
+                    this._logicalContentBytesUploaded += uploadStats.LogicalContentBytesUploaded;
+                    this._physicalContentBytesUploaded += uploadStats.PhysicalContentBytesUploaded;
+                    this._totalNumberOfChunks += uploadStats.TotalNumberOfChunks;
+                }
+            }
+        }
+
+        public async Task CommitTelemetry(Guid planId, Guid jobId)
+        {
+            var ciData = new Dictionary<string, object>();
+            ciData.Add("PlanId", planId);
+            ciData.Add("JobId", jobId);
+
+            ciData.Add("ChunksUploaded", this._chunksUploaded);
+            ciData.Add("CompressionBytesSaved", this._compressionBytesSaved);
+            ciData.Add("DedupUploadBytesSaved", this._dedupUploadBytesSaved);
+            ciData.Add("LogicalContentBytesUploaded", this._logicalContentBytesUploaded);
+            ciData.Add("PhysicalContentBytesUploaded", this._physicalContentBytesUploaded);
+            ciData.Add("TotalNumberOfChunks", this._totalNumberOfChunks);
+
+            var ciEvent = new CustomerIntelligenceEvent
+            {
+                Area = "AzurePipelinesAgent",
+                Feature = "BuildArtifacts",
+                Properties = ciData
+            };
+            await _ciClient.PublishEventsAsync(new [] { ciEvent });
+        }
+    }
+}

src/Microsoft.VisualStudio.Services.Agent/Blob/DedupManifestArtifactClientFactory.cs

@@ -22,7 +22,7 @@ public interface IDedupManifestArtifactClientFactory
             CancellationToken cancellationToken);
 
 
-        Task<(DedupStoreClient client, BlobStoreClientTelemetry telemetry)> CreateDedupClientAsync(
+        Task<(DedupStoreClient client, BlobStoreClientTelemetryTfs telemetry)> CreateDedupClientAsync(
             bool verbose,
             Action<string> traceOutput,
             VssConnection connection,
@@ -70,7 +70,7 @@ private DedupManifestArtifactClientFactory()
             return (new DedupManifestArtifactClient(telemetry, client, tracer), telemetry);
         }
 
-        public async Task<(DedupStoreClient client, BlobStoreClientTelemetry telemetry)> CreateDedupClientAsync(
+        public async Task<(DedupStoreClient client, BlobStoreClientTelemetryTfs telemetry)> CreateDedupClientAsync(
             bool verbose,
             Action<string> traceOutput,
             VssConnection connection,
@@ -97,7 +97,7 @@ private DedupManifestArtifactClientFactory()
                 cancellationToken: cancellationToken,
                 continueOnCapturedContext: false);
 
-            var telemetry = new BlobStoreClientTelemetry(tracer, dedupStoreHttpClient.BaseAddress);
+            var telemetry = new BlobStoreClientTelemetryTfs(tracer, dedupStoreHttpClient.BaseAddress, connection);
             var client = new DedupStoreClient(dedupStoreHttpClient, 192); // TODO
             return (client, telemetry);
         }

src/Microsoft.VisualStudio.Services.Agent/Blob/IDedupRecord.cs

@@ -0,0 +1,12 @@
+// Copyright (c) Microsoft Corporation.
+// Licensed under the MIT License.
+
+using Microsoft.VisualStudio.Services.BlobStore.Common.Telemetry;
+
+namespace Microsoft.VisualStudio.Services.Agent.Blob
+{
+    public interface IDedupRecord
+    {
+        DedupUploadStatistics UploadStatistics { get; }
+    }
+}

src/Agent.Sdk/Blob/PipelineTelemetryRecord.cs → src/Microsoft.VisualStudio.Services.Agent/Blob/PipelineTelemetryRecord.cs

@@ -7,16 +7,17 @@
 using Microsoft.VisualStudio.Services.BlobStore.Common.Telemetry;
 using Microsoft.TeamFoundation.DistributedTask.WebApi;
 
-namespace Agent.Sdk.Blob
+namespace Microsoft.VisualStudio.Services.Agent.Blob
 {
     /// <summary>
     /// Generic telemetry record for use with Pipeline events.
     /// </summary>
-    public abstract class PipelineTelemetryRecord : BlobStoreTelemetryRecord
+    public abstract class PipelineTelemetryRecord : BlobStoreTelemetryRecord, IDedupRecord
     {
         public Guid PlanId { get; private set; }
         public Guid JobId { get; private set; }
         public Guid TaskInstanceId { get; private set; }
+        public DedupUploadStatistics UploadStatistics { get; private set; }
 
         public PipelineTelemetryRecord(
             TelemetryInformationLevel level, 
@@ -31,5 +32,29 @@ public PipelineTelemetryRecord(
             JobId = new Guid(context.GetVariableValueOrDefault(WellKnownDistributedTaskVariables.JobId) ?? Guid.Empty.ToString());
             TaskInstanceId = new Guid(context.GetVariableValueOrDefault(WellKnownDistributedTaskVariables.TaskInstanceId) ?? Guid.Empty.ToString());
         }
+
+        public PipelineTelemetryRecord(
+            TelemetryInformationLevel level, 
+            Uri baseAddress, 
+            string eventNamePrefix, 
+            string eventNameSuffix, 
+            Guid planId,
+            Guid jobId,
+            Guid taskInstanceId,
+            uint attemptNumber = 1)
+            : base(level, baseAddress, eventNamePrefix, eventNameSuffix, attemptNumber)
+        {
+            PlanId = planId;
+            JobId = jobId;
+            TaskInstanceId = taskInstanceId;
+        }
+
+        protected override void SetMeasuredActionResult<T>(T value)
+        {
+            if (value is DedupUploadStatistics upStats)
+            {
+                UploadStatistics = upStats;
+            }
+        }
     }
 }

src/Microsoft.VisualStudio.Services.Agent/Blob/TimelineRecordAttachmentTelemetryRecord.cs

@@ -12,15 +12,18 @@ namespace Microsoft.VisualStudio.Services.Agent.Blob
     /// <summary>
     /// Generic telemetry record for use with timeline record events.
     /// </summary>
-    public class TimelineRecordAttachmentTelemetryRecord : BlobStoreTelemetryRecord
+    public class TimelineRecordAttachmentTelemetryRecord : PipelineTelemetryRecord
     {
         public TimelineRecordAttachmentTelemetryRecord(
             TelemetryInformationLevel level, 
             Uri baseAddress, 
             string eventNamePrefix, 
             string eventNameSuffix,
+            Guid planId,
+            Guid jobId,
+            Guid taskInstanceId,
             uint attemptNumber = 1)
-            : base(level, baseAddress, eventNamePrefix, eventNameSuffix, attemptNumber)
+            : base(level, baseAddress, eventNamePrefix, eventNameSuffix, planId, jobId, taskInstanceId, attemptNumber)
         {
         }
     }