Status - add second layer of aggregation for incidents (#563)

Author: Scott Bommarito

src/StatusAggregator/Cursor.cs src/StatusAggregator/Collector/Cursor.cssrc/StatusAggregator/Cursor.cs renamed to src/StatusAggregator/Collector/Cursor.cs

@@ -1,4 +1,4 @@
-// Copyright (c) .NET Foundation. All rights reserved.
+// 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;
@@ -8,10 +8,13 @@
 using NuGet.Services.Status.Table;
 using StatusAggregator.Table;
 
-namespace StatusAggregator
+namespace StatusAggregator.Collector
 {
     public class Cursor : ICursor
     {
+        private readonly ITableWrapper _table;
+        private readonly ILogger<Cursor> _logger;
+
         public Cursor(
             ITableWrapper table,
             ILogger<Cursor> logger)
@@ -20,16 +23,13 @@ public Cursor(
             _logger = logger ?? throw new ArgumentNullException(nameof(logger));
         }
 
-        private readonly ITableWrapper _table;
-
-        private readonly ILogger<Cursor> _logger;
-
         public async Task<DateTime> Get(string name)
         {
+            name = name ?? throw new ArgumentNullException(nameof(name));
+
             using (_logger.Scope("Fetching cursor with name {CursorName}.", name))
             {
-                var cursor = await _table.RetrieveAsync<CursorEntity>(
-                    CursorEntity.DefaultPartitionKey, name);
+                var cursor = await _table.RetrieveAsync<CursorEntity>(name);
 
                 DateTime value;
                 if (cursor == null)
@@ -50,11 +50,13 @@ public async Task<DateTime> Get(string name)
 
         public Task Set(string name, DateTime value)
         {
+            name = name ?? throw new ArgumentNullException(nameof(name));
+
             using (_logger.Scope("Updating cursor with name {CursorName} to {CursorValue}.", name, value))
             {
                 var cursorEntity = new CursorEntity(name, value);
                 return _table.InsertOrReplaceAsync(cursorEntity);
             }
         }
     }
-}
+}

src/StatusAggregator/Collector/EntityCollector.cs

@@ -0,0 +1,36 @@
+// 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;
+using System.Threading.Tasks;
+
+namespace StatusAggregator.Collector
+{
+    public class EntityCollector : IEntityCollector
+    {
+        private readonly ICursor _cursor;
+        private readonly IEntityCollectorProcessor _processor;
+
+        public EntityCollector(
+            ICursor cursor,
+            IEntityCollectorProcessor processor)
+        {
+            _cursor = cursor ?? throw new ArgumentNullException(nameof(cursor));
+            _processor = processor ?? throw new ArgumentNullException(nameof(processor));
+        }
+
+        public string Name => _processor.Name;
+
+        public async Task<DateTime> FetchLatest()
+        {
+            var lastCursor = await _cursor.Get(Name);
+            var nextCursor = await _processor.FetchSince(lastCursor);
+            if (nextCursor.HasValue)
+            {
+                await _cursor.Set(Name, nextCursor.Value);
+            }
+
+            return nextCursor ?? lastCursor;
+        }
+    }
+}

src/StatusAggregator/Collector/ICursor.cs

@@ -0,0 +1,24 @@
+// 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;
+using System.Threading.Tasks;
+
+namespace StatusAggregator.Collector
+{
+    /// <summary>
+    /// Persists a <see cref="DateTime"/> that represents that job's current progress by a string.
+    /// </summary>
+    public interface ICursor
+    {
+        /// <summary>
+        /// Gets the current progress of the job by <paramref name="name"/>.
+        /// </summary>
+        Task<DateTime> Get(string name);
+
+        /// <summary>
+        /// Saves <paramref name="value"/> as the current progress of the job by <paramref name="name"/>.
+        /// </summary>
+        Task Set(string name, DateTime value);
+    }
+}

src/StatusAggregator/Collector/IEntityCollector.cs

@@ -0,0 +1,24 @@
+// 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;
+using System.Threading.Tasks;
+
+namespace StatusAggregator.Collector
+{
+    /// <summary>
+    /// Collects new entities from a source.
+    /// </summary>
+    public interface IEntityCollector
+    {
+        /// <summary>
+        /// A unique name for this collector.
+        /// </summary>
+        string Name { get; }
+
+        /// <summary>
+        /// Fetches all new entities from the source and returns the newest timestamp found.
+        /// </summary>
+        Task<DateTime> FetchLatest();
+    }
+}

src/StatusAggregator/Collector/IEntityCollectorProcessor.cs

@@ -0,0 +1,24 @@
+// 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;
+using System.Threading.Tasks;
+
+namespace StatusAggregator.Collector
+{
+    /// <summary>
+    /// Used by <see cref="IEntityCollector"/> to fetch the latest entities from a source.
+    /// </summary>
+    public interface IEntityCollectorProcessor
+    {
+        /// <summary>
+        /// A unique name for this processor.
+        /// </summary>
+        string Name { get; }
+
+        /// <summary>
+        /// Fetches all entities from the source newer than <paramref name="cursor"/> and returns the latest timestamp found.
+        /// </summary>
+        Task<DateTime?> FetchSince(DateTime cursor);
+    }
+}

src/StatusAggregator/IncidentUpdater.cs …ctor/IncidentEntityCollectorProcessor.cssrc/StatusAggregator/IncidentUpdater.cs renamed to src/StatusAggregator/Collector/IncidentEntityCollectorProcessor.cs src/StatusAggregator/IncidentUpdater.cs renamed to src/StatusAggregator/Collector/IncidentEntityCollectorProcessor.cs

@@ -1,68 +1,49 @@
-// Copyright (c) .NET Foundation. All rights reserved.
+// 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 Microsoft.Extensions.Logging;
 using NuGet.Jobs.Extensions;
 using NuGet.Services.Incidents;
 using NuGet.Services.Status.Table;
+using StatusAggregator.Factory;
 using StatusAggregator.Parse;
-using StatusAggregator.Table;
 using System;
 using System.Linq;
 using System.Threading.Tasks;
 
-namespace StatusAggregator
+namespace StatusAggregator.Collector
 {
-    public class IncidentUpdater : IIncidentUpdater
+    /// <summary>
+    /// Fetches new <see cref="IncidentEntity"/>s using an <see cref="IIncidentApiClient"/>.
+    /// </summary>
+    public class IncidentEntityCollectorProcessor : IEntityCollectorProcessor
     {
-        private readonly ITableWrapper _table;
+        public const string IncidentsCollectorName = "incidents";
+        
         private readonly IAggregateIncidentParser _aggregateIncidentParser;
         private readonly IIncidentApiClient _incidentApiClient;
-        private readonly IIncidentFactory _incidentFactory;
-        private readonly ILogger<IncidentUpdater> _logger;
+        private readonly IComponentAffectingEntityFactory<IncidentEntity> _incidentFactory;
+        private readonly ILogger<IncidentEntityCollectorProcessor> _logger;
 
         private readonly string _incidentApiTeamId;
 
-        public IncidentUpdater(
-            ITableWrapper table,
+        public IncidentEntityCollectorProcessor(
             IIncidentApiClient incidentApiClient,
             IAggregateIncidentParser aggregateIncidentParser,
-            IIncidentFactory incidentFactory,
+            IComponentAffectingEntityFactory<IncidentEntity> incidentFactory,
             StatusAggregatorConfiguration configuration,
-            ILogger<IncidentUpdater> logger)
+            ILogger<IncidentEntityCollectorProcessor> logger)
         {
-            _table = table ?? throw new ArgumentNullException(nameof(table));
             _incidentApiClient = incidentApiClient ?? throw new ArgumentNullException(nameof(incidentApiClient));
             _aggregateIncidentParser = aggregateIncidentParser ?? throw new ArgumentNullException(nameof(aggregateIncidentParser));
             _incidentFactory = incidentFactory ?? throw new ArgumentNullException(nameof(incidentFactory));
             _incidentApiTeamId = configuration?.TeamId ?? throw new ArgumentNullException(nameof(configuration));
             _logger = logger ?? throw new ArgumentNullException(nameof(logger));
         }
 
-        public async Task RefreshActiveIncidents()
-        {
-            using (_logger.Scope("Refreshing active incidents."))
-            {
-                var activeIncidentEntities = _table
-                    .CreateQuery<IncidentEntity>()
-                    .Where(i => i.PartitionKey == IncidentEntity.DefaultPartitionKey && i.IsActive)
-                    .ToList();
-
-                _logger.LogInformation("Refreshing {ActiveIncidentsCount} active incidents.", activeIncidentEntities.Count());
-                foreach (var activeIncidentEntity in activeIncidentEntities)
-                {
-                    using (_logger.Scope("Refreshing active incident '{IncidentRowKey}'.", activeIncidentEntity.RowKey))
-                    {
-                        var activeIncident = await _incidentApiClient.GetIncident(activeIncidentEntity.IncidentApiId);
-                        activeIncidentEntity.MitigationTime = activeIncident.MitigationData?.Date;
-                        _logger.LogInformation("Updated mitigation time of active incident to {MitigationTime}", activeIncidentEntity.MitigationTime);
-                        await _table.InsertOrReplaceAsync(activeIncidentEntity);
-                    }
-                }
-            }
-        }
+        public string Name => IncidentsCollectorName;
 
-        public async Task<DateTime?> FetchNewIncidents(DateTime cursor)
+        public async Task<DateTime?> FetchSince(DateTime cursor)
         {
             using (_logger.Scope("Fetching all new incidents since {Cursor}.", cursor))
             {
@@ -73,12 +54,19 @@ public async Task RefreshActiveIncidents()
                     .Where(i => i.CreateDate > cursor)
                     .ToList();
 
+                _logger.LogInformation("Found {IncidentCount} incidents to parse.", incidents.Count);
                 var parsedIncidents = incidents
-                    .SelectMany(i => _aggregateIncidentParser.ParseIncident(i))
+                    .SelectMany(_aggregateIncidentParser.ParseIncident)
                     .ToList();
-                foreach (var parsedIncident in parsedIncidents.OrderBy(i => i.CreationTime))
+
+                _logger.LogInformation("Parsed {ParsedIncidentCount} incidents.", parsedIncidents.Count);
+                foreach (var parsedIncident in parsedIncidents.OrderBy(i => i.StartTime))
                 {
-                    await _incidentFactory.CreateIncident(parsedIncident);
+                    using (_logger.Scope("Creating incident for parsed incident with ID {ParsedIncidentID} affecting {ParsedIncidentPath} at {ParsedIncidentStartTime} with status {ParsedIncidentStatus}.",
+                        parsedIncident.Id, parsedIncident.AffectedComponentPath, parsedIncident.StartTime, parsedIncident.AffectedComponentStatus))
+                    {
+                        await _incidentFactory.CreateAsync(parsedIncident);
+                    }
                 }
 
                 return incidents.Any() ? incidents.Max(i => i.CreateDate) : (DateTime?)null;
@@ -97,4 +85,4 @@ private string GetRecentIncidentsQuery(DateTime cursor)
             return query;
         }
     }
-}
+}

…ator/Manual/ManualStatusChangeUpdater.cs …/ManualStatusChangeCollectorProcessor.cssrc/StatusAggregator/Manual/ManualStatusChangeUpdater.cs renamed to src/StatusAggregator/Collector/ManualStatusChangeCollectorProcessor.cs src/StatusAggregator/Manual/ManualStatusChangeUpdater.cs renamed to src/StatusAggregator/Collector/ManualStatusChangeCollectorProcessor.cs

@@ -1,43 +1,49 @@
-// Copyright (c) .NET Foundation. All rights reserved.
+// 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 Microsoft.Extensions.Logging;
 using NuGet.Jobs.Extensions;
 using NuGet.Services.Status.Table.Manual;
+using StatusAggregator.Manual;
 using StatusAggregator.Table;
 using System;
 using System.Linq;
 using System.Threading.Tasks;
 
-namespace StatusAggregator.Manual
+namespace StatusAggregator.Collector
 {
-    public class ManualStatusChangeUpdater : IManualStatusChangeUpdater
+    /// <summary>
+    /// Fetches new <see cref="ManualStatusChangeEntity"/>s using an <see cref="ITableWrapper"/>.
+    /// </summary>
+    public class ManualStatusChangeCollectorProcessor : IEntityCollectorProcessor
     {
+        public const string ManualCollectorNamePrefix = "manual";
+
         private readonly ITableWrapper _table;
         private readonly IManualStatusChangeHandler _handler;
-        private readonly ILogger<ManualStatusChangeUpdater> _logger;
+        private readonly ILogger<ManualStatusChangeCollectorProcessor> _logger;
 
-        public ManualStatusChangeUpdater(
+        public ManualStatusChangeCollectorProcessor(
             string name,
             ITableWrapper table,
             IManualStatusChangeHandler handler,
-            ILogger<ManualStatusChangeUpdater> logger)
+            ILogger<ManualStatusChangeCollectorProcessor> logger)
         {
-            Name = name;
+            Name = ManualCollectorNamePrefix + 
+                (name ?? throw new ArgumentNullException(nameof(name)));
             _table = table ?? throw new ArgumentNullException(nameof(table));
             _handler = handler ?? throw new ArgumentNullException(nameof(handler));
             _logger = logger ?? throw new ArgumentNullException(nameof(logger));
         }
 
         public string Name { get; }
 
-        public async Task<DateTime?> ProcessNewManualChanges(DateTime cursor)
+        public async Task<DateTime?> FetchSince(DateTime cursor)
         {
             using (_logger.Scope("Processing manual status changes."))
             {
                 var manualChangesQuery = _table
-                    .CreateQuery<ManualStatusChangeEntity>()
-                    .Where(c => c.PartitionKey == ManualStatusChangeEntity.DefaultPartitionKey);
+                    .CreateQuery<ManualStatusChangeEntity>();
 
                 // Table storage throws on queries with DateTime values that are too low.
                 // If we are fetching manual changes for the first time, don't filter on the timestamp.
@@ -49,7 +55,7 @@ public ManualStatusChangeUpdater(
                 var manualChanges = manualChangesQuery.ToList();
 
                 _logger.LogInformation("Processing {ManualChangesCount} manual status changes.", manualChanges.Count());
-                foreach (var manualChange in manualChanges)
+                foreach (var manualChange in manualChanges.OrderBy(m => m.Timestamp))
                 {
                     await _handler.Handle(_table, manualChange);
                 }
@@ -58,4 +64,4 @@ public ManualStatusChangeUpdater(
             }
         }
     }
-}
+}

src/StatusAggregator/Container/ContainerWrapper.cs

@@ -0,0 +1,29 @@
+// 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.Threading.Tasks;
+using Microsoft.WindowsAzure.Storage.Blob;
+
+namespace StatusAggregator.Container
+{
+    public class ContainerWrapper : IContainerWrapper
+    {
+        private readonly CloudBlobContainer _container;
+
+        public ContainerWrapper(CloudBlobContainer container)
+        {
+            _container = container;
+        }
+
+        public Task CreateIfNotExistsAsync()
+        {
+            return _container.CreateIfNotExistsAsync();
+        }
+
+        public Task SaveBlobAsync(string name, string contents)
+        {
+            var blob = _container.GetBlockBlobReference(name);
+            return blob.UploadTextAsync(contents);
+        }
+    }
+}