Refactoring ZeroToOneTargetScaler (Azure#53263)

Author: alrod

sdk/storage/Microsoft.Azure.WebJobs.Extensions.Storage.Blobs/src/Listeners/BlobLogListener.cs

@@ -4,14 +4,14 @@
 using System;
 using System.Collections.Generic;
 using System.Globalization;
+using System.IO;
 using System.Linq;
 using System.Text;
 using System.Threading;
 using System.Threading.Tasks;
 using Azure.Storage.Blobs;
 using Azure.Storage.Blobs.Models;
 using Azure.Storage.Blobs.Specialized;
-using Microsoft.Azure.WebJobs.Host.Timers;
 using Microsoft.Extensions.Logging;
 
 namespace Microsoft.Azure.WebJobs.Extensions.Storage.Blobs.Listeners
@@ -77,6 +77,72 @@ public async Task<IEnumerable<BlobWithContainer<BlobBaseClient>>> GetRecentBlobW
             return blobs;
         }
 
+        public async Task<StorageAnalyticsLogEntry> GetFirstLogEntryWithWritesAsync(string containerName, CancellationToken cancellationToken, int hoursWindow = DefaultScanHoursWindow)
+        {
+            if (hoursWindow <= 0)
+            {
+                return null;
+            }
+
+            DateTime hourCursor = DateTime.UtcNow;
+            BlobContainerClient containerClient = _blobClient.GetBlobContainerClient(LogContainer);
+
+            for (int hourIndex = 0; hourIndex < hoursWindow; hourIndex++)
+            {
+                cancellationToken.ThrowIfCancellationRequested();
+
+                string prefix = GetSearchPrefix("blob", hourCursor, hourCursor);
+
+                await foreach (BlobItem blob in containerClient
+                    .GetBlobsAsync(traits: BlobTraits.Metadata, prefix: prefix, states: BlobStates.None, cancellationToken: cancellationToken)
+                    .ConfigureAwait(false))
+                {
+                    cancellationToken.ThrowIfCancellationRequested();
+
+                    if (blob.Metadata is not null &&
+                        blob.Metadata.TryGetValue(LogType, out string logType) &&
+                        !string.IsNullOrEmpty(logType) &&
+                        logType.IndexOf("write", StringComparison.OrdinalIgnoreCase) >= 0)
+                    {
+                        BlobClient logBlobClient = containerClient.GetBlobClient(blob.Name);
+
+                        using (Stream stream = await logBlobClient.OpenReadAsync(options: default, cancellationToken: cancellationToken).ConfigureAwait(false))
+                        {
+                            using (StreamReader reader = new StreamReader(stream))
+                            {
+                                int lineNumber = 0;
+                                while (!reader.EndOfStream)
+                                {
+                                    cancellationToken.ThrowIfCancellationRequested();
+                                    string line = await reader.ReadLineAsync().ConfigureAwait(false);
+                                    lineNumber++;
+
+                                    if (line != null)
+                                    {
+                                        var entry = _parser.ParseLine(line, logBlobClient.Name, lineNumber.ToString(CultureInfo.InvariantCulture));
+                                        if (entry != null && entry.IsBlobWrite)
+                                        {
+                                            var path = entry.ToBlobPath();
+                                            if (path != null &&
+                                                string.Equals(path.ContainerName, containerName, StringComparison.OrdinalIgnoreCase))
+                                            {
+                                                // If we found a valid write entry, we can stop searching.
+                                                return entry;
+                                            }
+                                        }
+                                    }
+                                }
+                            }
+                        }
+                    }
+                }
+
+                hourCursor = hourCursor.AddHours(-1);
+            }
+
+            return null;
+        }
+
         internal static IEnumerable<BlobPath> GetPathsForValidBlobWrites(IEnumerable<StorageAnalyticsLogEntry> entries)
         {
             IEnumerable<BlobPath> parsedBlobPaths = from entry in entries

sdk/storage/Microsoft.Azure.WebJobs.Extensions.Storage.Blobs/src/Listeners/BlobScalerMonitorProvider.cs

@@ -93,28 +93,9 @@ public async Task<IEnumerable<StorageAnalyticsLogEntry>> ParseLogAsync(BlobBaseC
                             break;
                         }
 
-                        Version version = TryParseVersion(line);
-                        if (version == null)
+                        var entry = ParseLine(line, blob.Name, lineNumber.ToString(CultureInfo.CurrentCulture));
+                        if (entry != null)
                         {
-                            string message = String.Format(CultureInfo.CurrentCulture,
-                                "Unable to detect a version of log entry on line {1} of Storage Analytics log file '{0}'.",
-                                blob.Name, lineNumber);
-
-                            _logger.LogWarning(message);
-                        }
-
-                        if (version == supportedVersion)
-                        {
-                            StorageAnalyticsLogEntry entry = TryParseLogEntry(line);
-                            if (entry == null)
-                            {
-                                string message = String.Format(CultureInfo.CurrentCulture,
-                                    "Unable to parse the log entry on line {1} of Storage Analytics log file '{0}'.",
-                                    blob.Name, lineNumber);
-
-                                _logger.LogWarning(message);
-                            }
-
                             entries.Add(entry);
                         }
                     }
@@ -174,5 +155,33 @@ from Match m in _compiledRegex.Matches(line)
 
             return StorageAnalyticsLogEntry.TryParse(fieldsArray);
         }
+
+        internal StorageAnalyticsLogEntry ParseLine(string line, string blobName, string lineNumber)
+        {
+            Version version = TryParseVersion(line);
+            if (version == null)
+            {
+                string message = String.Format(CultureInfo.CurrentCulture,
+                    "Unable to detect a version of log entry on line {1} of Storage Analytics log file '{0}'.",
+                    blobName, lineNumber);
+
+                _logger.LogWarning(message);
+            }
+
+            if (version == supportedVersion)
+            {
+                StorageAnalyticsLogEntry entry = TryParseLogEntry(line);
+                if (entry == null)
+                {
+                    string message = String.Format(CultureInfo.CurrentCulture,
+                        "Unable to parse the log entry on line {1} of Storage Analytics log file '{0}'.",
+                        blobName, lineNumber);
+
+                    _logger.LogWarning(message);
+                }
+                return entry;
+            }
+            return null;
+        }
     }
 }