Update the package hash and length in DB after blob copy (#364)

Progress NuGet/Engineering#1190

Author: joelverhagen

src/NuGet.Services.Validation.Orchestrator/Configuration/ConfigurationValidator.cs

@@ -102,9 +102,7 @@ private void CheckUnknownValidators()
         {
             foreach (var validatorItem in _configuration.Validations)
             {
-                // This method will throw if the validator does not exist.
-                var validatorType = _validatorProvider.GetValidatorType(validatorItem.Name);
-                if (validatorType == null)
+                if (!_validatorProvider.IsValidator(validatorItem.Name))
                 {
                     throw new ConfigurationErrorsException("Validator implementation not found for " + validatorItem.Name);
                 }
@@ -142,7 +140,7 @@ private void CheckForCyclesAndParallelProcessors()
             var processorNames = _configuration
                 .Validations
                 .Select(x => x.Name)
-                .Where(x => typeof(IProcessor).IsAssignableFrom(_validatorProvider.GetValidatorType(x)))
+                .Where(x => _validatorProvider.IsProcessor(x))
                 .ToList();
 
             TopologicalSort.Validate(_configuration.Validations, processorNames);

src/NuGet.Services.Validation.Orchestrator/IValidationPackageFileService.cs

@@ -2,13 +2,21 @@
 // Licensed under the Apache License, Version 2.0. See License.txt in the project root for license information.
 
 using System;
+using System.IO;
 using System.Threading.Tasks;
 using NuGetGallery;
 
 namespace NuGet.Services.Validation.Orchestrator
 {
     public interface IValidationPackageFileService : ICorePackageFileService
     {
+        /// <summary>
+        /// Download the package content from the packages container to a temporary location on disk.
+        /// </summary>
+        /// <param name="package">The package metadata.</param>
+        /// <returns>The package stream.</returns>
+        Task<Stream> DownloadPackageFileToDiskAsync(Package package);
+
         /// <summary>
         /// Copy a package from the validation container to a location specific for the validation set. This allows the
         /// validation set to have its own copy of the package to mutate (via <see cref="IProcessor"/>) and validate.

src/NuGet.Services.Validation.Orchestrator/IValidatorProvider.cs

@@ -1,16 +1,15 @@
 // Copyright (c) .NET Foundation. All rights reserved.
 // Licensed under the Apache License, Version 2.0. See License.txt in the project root for license information.
 
-using System;
-
 namespace NuGet.Services.Validation.Orchestrator
 {
     /// <summary>
     /// Interface for the class that provides validator instances by their name
     /// </summary>
     public interface IValidatorProvider
     {
-        Type GetValidatorType(string validatorName);
+        bool IsValidator(string validatorName);
+        bool IsProcessor(string validatorName);
         IValidator GetValidator(string validatorName);
     }
 }

src/NuGet.Services.Validation.Orchestrator/Job.cs

@@ -3,8 +3,9 @@
 
 using System;
 using System.Collections.Generic;
-using System.Linq;
 using System.Net;
+using System.Net.Http;
+using System.Reflection;
 using System.Threading.Tasks;
 using AnglicanGeek.MarkdownMailer;
 using Autofac;
@@ -18,12 +19,12 @@
 using Microsoft.WindowsAzure.Storage;
 using NuGet.Jobs;
 using NuGet.Jobs.Configuration;
+using NuGet.Jobs.Validation;
 using NuGet.Jobs.Validation.Common;
 using NuGet.Jobs.Validation.PackageSigning.Messages;
 using NuGet.Jobs.Validation.PackageSigning.Storage;
 using NuGet.Services.Configuration;
 using NuGet.Services.KeyVault;
-using NuGet.Services.Logging;
 using NuGet.Services.ServiceBus;
 using NuGet.Services.Validation.Orchestrator.Telemetry;
 using NuGet.Services.Validation.PackageCertificates;
@@ -48,6 +49,7 @@ public class Job : JobBase
         private const string ServiceBusConfigurationSectionName = "ServiceBus";
         private const string SmtpConfigurationSectionName = "Smtp";
         private const string EmailConfigurationSectionName = "Email";
+        private const string PackageDownloadTimeoutName = "PackageDownloadTimeout";
 
         private const string VcsBindingKey = VcsSectionName;
         private const string PackageVerificationTopicClientBindingKey = "PackageVerificationTopicClient";
@@ -192,7 +194,7 @@ private void ConfigureJobServices(IServiceCollection services, IConfigurationRoo
                 });
             services.AddTransient<NuGetGallery.ICoreFileStorageService, NuGetGallery.CloudBlobCoreFileStorageService>();
             services.AddTransient<IValidationPackageFileService, ValidationPackageFileService>();
-            services.AddTransient<IValidationOutcomeProcessor, ValidationOutcomeProcessor>();
+            services.AddTransient<IPackageDownloader, PackageDownloader>();
             services.AddTransient<IPackageStatusProcessor, PackageStatusProcessor>();
             services.AddTransient<IValidationSetProvider, ValidationSetProvider>();
             services.AddTransient<IValidationSetProcessor, ValidationSetProcessor>();
@@ -230,8 +232,27 @@ private void ConfigureJobServices(IServiceCollection services, IConfigurationRoo
             services.AddTransient<ICoreMessageServiceConfiguration, CoreMessageServiceConfiguration>();
             services.AddTransient<ICoreMessageService, CoreMessageService>();
             services.AddTransient<IMessageService, MessageService>();
+            services.AddTransient<ICommonTelemetryService, CommonTelemetryService>();
             services.AddTransient<ITelemetryService, TelemetryService>();
             services.AddSingleton(new TelemetryClient());
+            services.AddTransient<IValidationOutcomeProcessor, ValidationOutcomeProcessor>();
+            services.AddSingleton(p =>
+            {
+                var assembly = Assembly.GetEntryAssembly();
+                var assemblyName = assembly.GetName().Name;
+                var assemblyVersion = assembly.GetCustomAttribute<AssemblyInformationalVersionAttribute>()?.InformationalVersion ?? "0.0.0";
+
+                var client = new HttpClient(new WebRequestHandler
+                {
+                    AllowPipelining = true,
+                    AutomaticDecompression = (DecompressionMethods.GZip | DecompressionMethods.Deflate),
+                });
+
+                client.Timeout = configurationRoot.GetValue<TimeSpan>(PackageDownloadTimeoutName);
+                client.DefaultRequestHeaders.Add("User-Agent", $"{assemblyName}/{assemblyVersion}");
+
+                return client;
+            });
         }
 
         private static IServiceProvider CreateProvider(IServiceCollection services)

src/NuGet.Services.Validation.Orchestrator/PackageStatusProcessor.cs

@@ -2,28 +2,34 @@
 // Licensed under the Apache License, Version 2.0. See License.txt in the project root for license information.
 
 using System;
+using System.Diagnostics;
+using System.Linq;
 using System.Threading.Tasks;
 using Microsoft.Extensions.Logging;
 using NuGet.Services.Validation.Orchestrator.Telemetry;
 using NuGetGallery;
+using NuGetGallery.Packaging;
 
 namespace NuGet.Services.Validation.Orchestrator
 {
     public class PackageStatusProcessor : IPackageStatusProcessor
     {
         private readonly ICorePackageService _galleryPackageService;
         private readonly IValidationPackageFileService _packageFileService;
+        private readonly IValidatorProvider _validatorProvider;
         private readonly ITelemetryService _telemetryService;
         private readonly ILogger<PackageStatusProcessor> _logger;
 
         public PackageStatusProcessor(
             ICorePackageService galleryPackageService,
             IValidationPackageFileService packageFileService,
+            IValidatorProvider validatorProvider,
             ITelemetryService telemetryService,
             ILogger<PackageStatusProcessor> logger)
         {
             _galleryPackageService = galleryPackageService ?? throw new ArgumentNullException(nameof(galleryPackageService));
             _packageFileService = packageFileService ?? throw new ArgumentNullException(nameof(packageFileService));
+            _validatorProvider = validatorProvider ?? throw new ArgumentNullException(nameof(validatorProvider));
             _telemetryService = telemetryService ?? throw new ArgumentNullException(nameof(telemetryService));
             _logger = logger ?? throw new ArgumentNullException(nameof(logger));
         }
@@ -127,8 +133,15 @@ private async Task MoveFileToPublicStorageAndMarkPackageAsAvailable(PackageValid
                 package.NormalizedVersion,
                 validationSet.ValidationTrackingId);
 
+            // If the validation set contains any processors, we must use the copy of the package that is specific to
+            // this validation set. We can't use the original validation package because it does not have any of the
+            // changes that the processors made. If the validation set package does not exist for some reason and there
+            // are processors in the validation set, this indicates a bug and an exception will be thrown by the copy
+            // operation below. This will cause the validation queue message to eventually dead-letter at which point
+            // the on-call person should investigate.
             bool copied;
-            if (await _packageFileService.DoesValidationSetPackageExistAsync(validationSet))
+            if (validationSet.PackageValidations.Any(x => _validatorProvider.IsProcessor(x.Type)) ||
+                await _packageFileService.DoesValidationSetPackageExistAsync(validationSet))
             {
                 copied = await CopyAsync(
                     validationSet,
@@ -150,6 +163,39 @@ private async Task MoveFileToPublicStorageAndMarkPackageAsAvailable(PackageValid
                     x => _packageFileService.CopyValidationPackageToPackageFileAsync(x.PackageId, x.PackageNormalizedVersion));
             }
 
+            // Use whatever package made it into the packages container. This is what customers will consume so the DB
+            // record must match.
+            using (var packageStream = await _packageFileService.DownloadPackageFileToDiskAsync(package))
+            {
+                var stopwatch = Stopwatch.StartNew();
+                var hash = CryptographyService.GenerateHash(packageStream, CoreConstants.Sha512HashAlgorithmId);
+                _telemetryService.TrackDurationToHashPackage(
+                    stopwatch.Elapsed,
+                    package.PackageRegistration.Id,
+                    package.NormalizedVersion,
+                    CoreConstants.Sha512HashAlgorithmId,
+                    packageStream.GetType().FullName);
+
+                var streamMetadata = new PackageStreamMetadata
+                {
+                    Size = packageStream.Length,
+                    Hash = hash,
+                    HashAlgorithm = CoreConstants.Sha512HashAlgorithmId,
+                };
+
+                // We don't immediately commit here. Later, we will commit these changes as well as the new package
+                // status as part of the same transaction.
+                if (streamMetadata.Size != package.PackageFileSize
+                    || streamMetadata.Hash != package.Hash
+                    || streamMetadata.HashAlgorithm != package.HashAlgorithm)
+                {
+                    await _galleryPackageService.UpdatePackageStreamMetadataAsync(
+                        package,
+                        streamMetadata,
+                        commitChanges: false);
+                }
+            }
+
             _logger.LogInformation("Marking package {PackageId} {PackageVersion}, validation set {ValidationSetId} as {PackageStatus} in DB",
                 package.PackageRegistration.Id,
                 package.NormalizedVersion,
@@ -158,6 +204,7 @@ private async Task MoveFileToPublicStorageAndMarkPackageAsAvailable(PackageValid
 
             try
             {
+                // Make the package available and commit any other pending changes (e.g. updated hash).
                 await UpdatePackageStatusAsync(package, PackageStatus.Available);
             }
             catch (Exception e)
@@ -171,8 +218,9 @@ private async Task MoveFileToPublicStorageAndMarkPackageAsAvailable(PackageValid
                     validationSet.ValidationTrackingId);
 
                 // If this execution was not the one to copy the package, then don't delete the package on failure.
-                // This prevents the (unlikely) case where two actors attempt the DB update, one suceeds and one fails.
-                // We don't want an available package record with nothing in the packages container!
+                // This prevents a missing passing in the (unlikely) case where two actors attempt the DB update, one
+                // succeeds and one fails. We don't want an available package record with nothing in the packages
+                // container!
                 if (copied)
                 {
                     await _packageFileService.DeletePackageFileAsync(package.PackageRegistration.Id, package.Version);
@@ -200,9 +248,10 @@ private async Task<bool> CopyAsync(PackageValidationSet validationSet, Package p
             catch (InvalidOperationException)
             {
                 // The package already exists in the packages container. This can happen if the DB commit below fails
-                // and this flow is retried. We assume that the package content has not changed. Today there is no way
-                // for the content to change. Hard deletes (the one way a package ID and version can get different
-                // content) delete from both the packages and validating container so this can't be a mismatch.
+                // and this flow is retried or another validation set for the package completed first. Either way, we
+                // will later attempt to use the hash from the package in the packages container (the destination).
+                // In other words, we don't care which copy wins, but the DB record must match the package that ends
+                // up in the packages container.
                 _logger.LogInformation(
                     "Package already exists in packages container for {PackageId} {PackageVersion}, validation set {ValidationSetId}",
                     package.PackageRegistration.Id,

src/NuGet.Services.Validation.Orchestrator/Telemetry/ITelemetryService.cs

@@ -91,5 +91,10 @@ public interface ITelemetryService
         /// in the public container.
         /// </summary>
         void TrackMissingNupkgForAvailablePackage(string packageId, string normalizedVersion, string validationTrackingId);
+
+        /// <summary>
+        /// A metric to of how long it took to hash a package.
+        /// </summary>
+        void TrackDurationToHashPackage(TimeSpan duration, string packageId, string normalizedVersion, string hashAlgorithm, string streamType);
     }
 }

src/NuGet.Services.Validation.Orchestrator/Telemetry/TelemetryService.cs

@@ -23,6 +23,7 @@ public class TelemetryService : ITelemetryService
         private const string ClientValidationIssue = Prefix + "ClientValidationIssue";
         private const string MissingPackageForValidationMessage = Prefix + "MissingPackageForValidationMessage";
         private const string MissingNupkgForAvailablePackage = Prefix + "MissingNupkgForAvailablePackage";
+        private const string DurationToHashPackageSeconds = Prefix + "DurationToHashPackageSeconds";
 
         private const string FromStatus = "FromStatus";
         private const string ToStatus = "ToStatus";
@@ -33,6 +34,9 @@ public class TelemetryService : ITelemetryService
         private const string PackageId = "PackageId";
         private const string NormalizedVersion = "NormalizedVersion";
         private const string ValidationTrackingId = "ValidationTrackingId";
+        private const string PackageSize = "PackageSize";
+        private const string HashAlgorithm = "HashAlgorithm";
+        private const string StreamType = "StreamType";
 
         private readonly TelemetryClient _telemetryClient;
 
@@ -41,6 +45,21 @@ public TelemetryService(TelemetryClient telemetryClient)
             _telemetryClient = telemetryClient ?? throw new ArgumentNullException(nameof(telemetryClient));
         }
 
+        public void TrackDurationToHashPackage(TimeSpan duration, string packageId, string normalizedVersion, string hashAlgorithm, string streamType)
+        {
+            _telemetryClient.TrackMetric(
+                DurationToHashPackageSeconds,
+                duration.TotalSeconds,
+                new Dictionary<string, string>
+                {
+                    { PackageId, packageId },
+                    { NormalizedVersion, normalizedVersion },
+                    { PackageSize, PackageSize.ToString() },
+                    { HashAlgorithm, hashAlgorithm },
+                    { StreamType, streamType },
+                });
+        }
+
         public void TrackDurationToValidationSetCreation(TimeSpan duration)
         {
             _telemetryClient.TrackMetric(

src/NuGet.Services.Validation.Orchestrator/ValidationPackageFileService.cs

@@ -2,25 +2,38 @@
 // Licensed under the Apache License, Version 2.0. See License.txt in the project root for license information.
 
 using System;
+using System.IO;
+using System.Threading;
 using System.Threading.Tasks;
 using Microsoft.Extensions.Logging;
+using NuGet.Jobs.Validation;
 using NuGetGallery;
 
 namespace NuGet.Services.Validation.Orchestrator
 {
     public class ValidationPackageFileService : CorePackageFileService, IValidationPackageFileService
     {
         private readonly ICoreFileStorageService _fileStorageService;
+        private readonly IPackageDownloader _packageDownloader;
         private readonly ILogger<ValidationPackageFileService> _logger;
 
         public ValidationPackageFileService(
             ICoreFileStorageService fileStorageService,
+            IPackageDownloader packageDownloader,
             ILogger<ValidationPackageFileService> logger) : base(fileStorageService)
         {
             _fileStorageService = fileStorageService ?? throw new ArgumentNullException(nameof(fileStorageService));
+            _packageDownloader = packageDownloader ?? throw new ArgumentNullException(nameof(packageDownloader));
             _logger = logger ?? throw new ArgumentNullException(nameof(logger));
         }
 
+        public async Task<Stream> DownloadPackageFileToDiskAsync(Package package)
+        {
+            var packageUri = await GetPackageReadUriAsync(package);
+
+            return await _packageDownloader.DownloadAsync(packageUri, CancellationToken.None);
+        }
+
         public Task CopyValidationPackageForValidationSetAsync(PackageValidationSet validationSet)
         {
             var srcFileName = BuildFileName(