Refactor log processing with Deployment/Pod composite key

Enhanced log processing by introducing a composite key (`Deployment/Pod`) to uniquely identify and manage logs for multiple containers per deployment.

- Added deployment count retrieval in `InitializeDeploymentTimesAsync` to optimize backfilling.
- Updated `ReadNewLinesAsync` and `ProcessEventLogFileAsync` to use the composite key for deployment settings.
- Refactored `ConvertEventRecordToLogLine` to improve clarity and align with the composite key approach.
- Added a new API endpoint (`GET /api/log/counts`) to expose deployment and pod counts.
- Implemented `GetDeploymentCounts` in `LogRepo.cs` to query deployment and pod counts from the database.
- Introduced the `DeploymentCount` class to represent deployment and pod count data.
- Improved logging for better traceability of deployment/pod combinations.
- Cleaned up redundant code and updated comments to reflect the new structure.

These changes improve the granularity, accuracy, and maintainability of log processing.

Author: Timothy Dodd

src/LogMkAgent/Services/LogWatcher.cs

@@ -133,15 +133,28 @@ private async Task InitializeDeploymentTimesAsync(LogApiClient client)
     {
         try
         {
+            // First, get deployment counts to know which deployments have already been backfilled
+            var counts = await client.GetDataAsync<List<DeploymentCount>>("api/log/counts").ConfigureAwait(false);
+            if (counts == null || !counts.Any())
+            {
+                _logger.LogInformation("No existing deployments found, all logs will be processed");
+                return;
+            }
+
+            _logger.LogInformation("Retrieved deployment counts for {Count} deployment/pod combinations", counts.Count);
+
+            // Then, get the latest timestamps for deployments that have logs
             var times = await client.GetDataAsync<List<LatestDeploymentEntry>>("api/log/times").ConfigureAwait(false);
             if (times != null)
             {
-                _logger.LogInformation("Retrieved latest deployment times for {Count} deployments", times.Count);
-                foreach (var pod in times)
+                foreach (var entry in times)
                 {
-                    var settings = GetDeploymentSettings(pod.Deployment);
-                    settings.LastDeploymentTime = pod.TimeStamp;
-                    _logger.LogDebug("Deployment {Deployment} last wrote at {TimeStamp}", pod.Deployment, pod.TimeStamp);
+                    // Use composite key (Deployment/Pod) to handle multiple containers per deployment
+                    var compositeKey = $"{entry.Deployment}/{entry.Pod}";
+                    var settings = GetDeploymentSettings(compositeKey);
+                    settings.LastDeploymentTime = entry.TimeStamp;
+                    _logger.LogDebug("Container {Deployment}/{Pod} last wrote at {TimeStamp}, will skip older logs",
+                        entry.Deployment, entry.Pod, entry.TimeStamp);
                 }
             }
         }
@@ -304,7 +317,9 @@ private async Task ReadNewLinesAsync(PodInfo info, CancellationToken stoppingTok
     {
         var filePath = info.LogPath;
         var podSettings = GetPodSettings(info.PodName);
-        var deploymentSettings = GetDeploymentSettings(info.DeploymentName);
+        // Use composite key (Deployment/Pod) for deployment settings
+        var compositeKey = $"{info.DeploymentName}/{info.PodName}";
+        var deploymentSettings = GetDeploymentSettings(compositeKey);
 
         if (!File.Exists(filePath))
         {
@@ -640,7 +655,10 @@ private async Task ProcessEventLogFileAsync(string filePath, CancellationToken s
             // machine hostname
 
             var machine = Environment.MachineName;
-            var deploymentSettings = GetDeploymentSettings($"{machine}-{logName}");
+            var podName = $"{machine}-{logName}";
+            // Use composite key (Deployment/Pod) - for event logs, deployment = machine
+            var compositeKey = $"{machine}/{podName}";
+            var deploymentSettings = GetDeploymentSettings(compositeKey);
 
             // Handle file rotation/changes
             if (deploymentSettings.FilePath != filePath)
@@ -670,7 +688,7 @@ private async Task ProcessEventLogFileAsync(string filePath, CancellationToken s
                         if (currentRecordId <= lastRecordId)
                             continue;
 
-                        var logLine = ConvertEventRecordToLogLine(eventRecord, logName);
+                        var logLine = ConvertEventRecordToLogLine(eventRecord, podName, machine);
                         if (logLine != null)
                         {
                             // Apply log level filtering
@@ -728,18 +746,15 @@ private async Task ProcessEventLogFileAsync(string filePath, CancellationToken s
         }
     }
 
-    private LogLine? ConvertEventRecordToLogLine(EventRecord eventRecord, string logName)
+    private LogLine? ConvertEventRecordToLogLine(EventRecord eventRecord, string podName, string machine)
     {
         try
         {
 
 
             var logLevel = MapEventLevelToLogLevel(eventRecord.Level);
-            var machine = Environment.MachineName;
             // Build the log message
             var message = BuildEventLogMessage(eventRecord);
-            var logname = eventRecord.LogName ?? logName;
-            var podName = $"{machine}-{logname}";
             var logLine = new LogLine
             {
                 DeploymentName = TruncateString(machine, 50),

src/LogMkApi/Controllers/LogController.cs

@@ -479,6 +479,14 @@ public async Task<IEnumerable<LatestDeploymentEntry>> GetLatestEntryTimes()
         return entries;
     }
 
+    [AllowAnonymous]
+    [HttpGet("counts")]
+    public async Task<IEnumerable<DeploymentCount>> GetDeploymentCounts()
+    {
+        var entries = await _logRepo.GetDeploymentCounts();
+        return entries;
+    }
+
     [AllowAnonymous]
     [HttpGet("settings")]
     public IActionResult GetValidationSettings()

src/LogMkApi/Data/LogRepo.cs

@@ -170,6 +170,19 @@ FROM Log
         }
     }
 
+    public async Task<IEnumerable<DeploymentCount>> GetDeploymentCounts()
+    {
+        using (var db = _dbFactory.CreateConnection())
+        {
+            var query = @"
+    SELECT Deployment, Pod, COUNT(*) AS Count
+    FROM Log
+    GROUP BY Deployment, Pod
+";
+            return await db.QueryAsync<DeploymentCount>(query);
+        }
+    }
+
     public async Task<int> PurgeLogsByDeployment(string deployment, DateTime? startDate)
     {
         using (var db = _dbFactory.CreateConnection())

src/LogMkCommon/DeploymentCount.cs

@@ -0,0 +1,8 @@
+namespace LogMkCommon;
+
+public class DeploymentCount
+{
+    public required string Deployment { get; set; }
+    public required string Pod { get; set; }
+    public required long Count { get; set; }
+}