Enhance log validation and caching mechanisms

- Introduced `LogCacheService` for caching pod metadata, recent logs,
  and tracking backfill grace periods to optimize validation and
  prevent duplicate entries.
- Added conditional timestamp validation based on pod existence
  and backfill periods, with stricter real-time rules for existing pods.
- Implemented duplicate detection using content hash and cache.
- Updated `LogController` to support asynchronous validation and
  integrate `LogCacheService`.
- Added new configuration options: `MaxMinutesOldForExistingPods`,
  `BackfillGracePeriodMinutes`, and `EnableDuplicateDetection`.
- Enhanced error handling for real-time hub delivery.
- Added new methods in `LogRepo` for pod existence checks and
  fetching recent logs.
- Changed service lifetimes in `Program.cs` to `Singleton` for
  caching-related services.
- Improved logging format and added metadata versioning.
- Refactored code for better readability and maintainability.

Author: Timothy Dodd

src/LogMkApi/Controllers/LogController.cs

@@ -23,21 +23,41 @@ public class LogController : ControllerBase
     private readonly LogSummaryRepo _logSummaryRepo;
     private readonly WorkQueueRepo _workQueueRepo;
     private readonly LogApiMetrics _metrics;
-    private readonly int LogMaxDaysOld = 30; // Maximum age of logs to accept
-    public LogController(ILogger<LogController> logger, LogRepo logRepo, LogHubService logHubService, LogSummaryRepo logSummaryRepo, WorkQueueRepo workQueueRepo, LogApiMetrics metrics, IConfiguration configuration)
+    private readonly LogCacheService _cacheService;
+
+    // Configuration: Maximum age limits
+    private readonly int LogMaxDaysOld = 30; // For new pods (backfill scenario)
+    private readonly int LogMaxMinutesOldForExistingPods = 5; // For existing pods (real-time scenario)
+    private readonly bool EnableDuplicateDetection = true;
+
+    public LogController(ILogger<LogController> logger, LogRepo logRepo, LogHubService logHubService,
+        LogSummaryRepo logSummaryRepo, WorkQueueRepo workQueueRepo, LogApiMetrics metrics,
+        LogCacheService cacheService, IConfiguration configuration)
     {
         _logger = logger;
         _logRepo = logRepo;
         _logHubService = logHubService;
         _logSummaryRepo = logSummaryRepo;
         _workQueueRepo = workQueueRepo;
         _metrics = metrics;
+        _cacheService = cacheService;
+
         if (configuration != null)
         {
             if (int.TryParse(configuration["LogSettings:MaxDaysOld"], out var maxDays))
             {
                 LogMaxDaysOld = maxDays;
             }
+
+            if (int.TryParse(configuration["LogSettings:MaxMinutesOldForExistingPods"], out var maxMinutes))
+            {
+                LogMaxMinutesOldForExistingPods = maxMinutes;
+            }
+
+            if (bool.TryParse(configuration["LogSettings:EnableDuplicateDetection"], out var enableDuplicates))
+            {
+                EnableDuplicateDetection = enableDuplicates;
+            }
         }
     }
     [AllowAnonymous]
@@ -78,13 +98,16 @@ public async Task<ActionResult<LogResponse>> Create(
         _logger.LogDebug("Processing batch {BatchId} with {Count} log lines", batchId, logLines.Count);
         _metrics.IncrementLogsReceived(logLines.Count);
 
+        // Track pods for cache invalidation (new pods that were just inserted)
+        var newPods = new HashSet<string>();
+
         // Process each log line individually - don't let errors stop the batch
         for (int i = 0; i < logLines.Count; i++)
         {
             try
             {
                 var logLine = logLines[i];
-                var validationResult = ValidateLogLine(logLine, i);
+                var validationResult = await ValidateLogLineAsync(logLine, i);
 
                 if (!validationResult.IsValid)
                 {
@@ -98,6 +121,13 @@ public async Task<ActionResult<LogResponse>> Create(
                     continue;
                 }
 
+                // Track if this is a new pod
+                var podExists = await _cacheService.PodExistsAsync(logLine.PodName);
+                if (!podExists)
+                {
+                    newPods.Add(logLine.PodName);
+                }
+
                 // Convert to database entity
                 var logEntity = new Log
                 {
@@ -141,6 +171,25 @@ public async Task<ActionResult<LogResponse>> Create(
 
                 _logger.LogDebug("Successfully inserted {InsertedCount}/{ValidCount} logs for batch {BatchId}",
                     insertedCount, validLogs.Count, batchId);
+
+                // Update cache for successfully inserted logs
+                if (insertedCount > 0)
+                {
+                    // Invalidate pod existence cache for new pods and start backfill tracking
+                    foreach (var newPod in newPods)
+                    {
+                        _cacheService.InvalidatePodExistence(newPod);
+                        _cacheService.StartBackfillTracking(newPod);
+                        _logger.LogInformation("Started backfill tracking for new pod: {PodName}", newPod);
+                    }
+
+                    // Update recent logs cache for all pods in this batch
+                    var podGroups = validLogs.GroupBy(log => log.Pod);
+                    foreach (var group in podGroups)
+                    {
+                        _cacheService.UpdateRecentLogsCache(group.Key, group);
+                    }
+                }
             }
             catch (Exception ex)
             {
@@ -150,26 +199,23 @@ public async Task<ActionResult<LogResponse>> Create(
             }
         }
 
-        // Send to real-time hub immediately - don't let this fail the API call
+        // Send to real-time hub
         if (insertedCount > 0)
         {
-            _ = Task.Run(async () =>
+            try
             {
-                try
-                {
-                    // Send all inserted logs for real-time delivery (limit to 500 for safety)
-                    var logsToSend = validLogs.Take(500).ToList();
+                // Send all inserted logs for real-time delivery (limit to 500 for safety)
+                var logsToSend = validLogs.Take(500).ToList();
 
-                    if (logsToSend.Any())
-                    {
-                        await _logHubService.SendLogs(logsToSend);
-                    }
-                }
-                catch (Exception ex)
+                if (logsToSend.Any())
                 {
-                    _logger.LogWarning(ex, "Failed to send logs to hub for batch {BatchId}", batchId);
+                    await _logHubService.SendLogs(logsToSend);
                 }
-            }, cancellationToken);
+            }
+            catch (Exception ex)
+            {
+                _logger.LogWarning(ex, "Failed to send logs to hub for batch {BatchId}", batchId);
+            }
         }
 
         // Log validation errors for monitoring
@@ -270,7 +316,7 @@ private async Task<int> InsertLogsResilientlyAsync(List<Log> logs, string batchI
         }
     }
 
-    private LogValidationResult ValidateLogLine(LogLine logLine, int index)
+    private async Task<LogValidationResult> ValidateLogLineAsync(LogLine logLine, int index)
     {
         var errors = new List<string>();
 
@@ -294,12 +340,47 @@ private LogValidationResult ValidateLogLine(LogLine logLine, int index)
             // Additional validations
             if (logLine != null)
             {
+                // Future timestamp check
                 if (logLine.TimeStamp > DateTimeOffset.UtcNow.AddMinutes(5))
                     errors.Add("TimeStamp is too far in the future");
 
-                if (logLine.TimeStamp < DateTimeOffset.UtcNow.AddDays(-LogMaxDaysOld))
-                    errors.Add("TimeStamp is too old (>30 days)");
+                // Conditional timestamp validation based on pod existence and backfill period
+                var podExists = await _cacheService.PodExistsAsync(logLine.PodName);
+                var isInBackfillPeriod = podExists && _cacheService.IsInBackfillPeriod(logLine.PodName);
 
+                if (!podExists || isInBackfillPeriod)
+                {
+                    // New pod OR in backfill grace period: allow backfill (30 days)
+                    if (logLine.TimeStamp < DateTimeOffset.UtcNow.AddDays(-LogMaxDaysOld))
+                    {
+                        errors.Add($"TimeStamp is too old (>{LogMaxDaysOld} days)");
+                    }
+                }
+                else
+                {
+                    // Existing pod outside backfill window: strict real-time validation (5 minutes)
+                    if (logLine.TimeStamp < DateTimeOffset.UtcNow.AddMinutes(-LogMaxMinutesOldForExistingPods))
+                    {
+                        errors.Add($"TimeStamp is too old (>{LogMaxMinutesOldForExistingPods} minutes) for existing pod");
+                    }
+                }
+
+                // Duplicate detection (if enabled)
+                if (EnableDuplicateDetection && podExists)
+                {
+                    var isDuplicate = await _cacheService.IsDuplicateLogAsync(
+                        logLine.PodName,
+                        logLine.TimeStamp,
+                        logLine.SequenceNumber,
+                        logLine.Line);
+
+                    if (isDuplicate)
+                    {
+                        errors.Add("Duplicate log entry detected");
+                    }
+                }
+
+                // Length validations
                 if (logLine.Line?.Length > 10000)
                     errors.Add("Line content too long (max 10,000 characters)");
 
@@ -405,6 +486,7 @@ public IActionResult GetValidationSettings()
         var settings = new LogMkCommon.ValidationSettings
         {
             MaxDaysOld = LogMaxDaysOld,
+            MaxMinutesOldForExistingPods = LogMaxMinutesOldForExistingPods,
             MaxFutureMinutes = 5,
             MaxLineLength = 10000,
             MaxDeploymentNameLength = 100,
@@ -413,7 +495,8 @@ public IActionResult GetValidationSettings()
             PodNamePattern = @"^[a-zA-Z0-9\-._]+$",
             AllowEmptyLogLevel = false,
             MaxBatchSize = 1000,
-            Version = "1.0",
+            EnableDuplicateDetection = EnableDuplicateDetection,
+            Version = "1.1",
             LastUpdated = DateTime.UtcNow
         };
 
@@ -527,17 +610,14 @@ public async Task<ActionResult<LogResponse>> CreateSingle(
             _logger.LogDebug("Successfully inserted single log for Pod: {Pod}", logEntry.Pod);
 
             // Send to real-time hub
-            _ = Task.Run(async () =>
+            try
             {
-                try
-                {
-                    await _logHubService.SendLogs(new List<Log> { logEntity });
-                }
-                catch (Exception ex)
-                {
-                    _logger.LogWarning(ex, "Failed to send single log to hub");
-                }
-            }, cancellationToken);
+                await _logHubService.SendLogs(new List<Log> { logEntity });
+            }
+            catch (Exception ex)
+            {
+                _logger.LogWarning(ex, "Failed to send single log to hub");
+            }
 
             var response = new LogResponse
             {

src/LogMkApi/Data/LogRepo.cs

@@ -208,5 +208,33 @@ public async Task<int> PurgeLogsByPod(string pod, DateTime? startDate)
         }
     }
 
+    /// <summary>
+    /// Checks if any logs exist for a specific pod
+    /// </summary>
+    public async Task<bool> PodExistsAsync(string podName)
+    {
+        using (var db = _dbFactory.CreateConnection())
+        {
+            var query = "SELECT COUNT(*) FROM Log WHERE Pod = @podName LIMIT 1";
+            var count = await db.ExecuteScalarAsync<int>(query, new { podName });
+            return count > 0;
+        }
+    }
+
+    /// <summary>
+    /// Gets the most recent logs for a specific pod
+    /// </summary>
+    public async Task<IEnumerable<Log>> GetRecentLogsForPodAsync(string podName, int limit = 1000)
+    {
+        using (var db = _dbFactory.CreateConnection())
+        {
+            var query = @"
+                SELECT * FROM Log
+                WHERE Pod = @podName
+                ORDER BY TimeStamp DESC, SequenceNumber DESC
+                LIMIT @limit";
+            return await db.QueryAsync<Log>(query, new { podName, limit });
+        }
+    }
 
 }

src/LogMkApi/Program.cs

@@ -22,7 +22,7 @@ public class Program
     public static void Main(string[] args)
     {
         var builder = WebApplication.CreateBuilder(args);
-        
+
         // Configure cleaner logging format
         builder.Logging.ClearProviders();
         builder.Logging.AddSimpleConsole(options =>
@@ -32,7 +32,7 @@ public static void Main(string[] args)
             options.UseUtcTimestamp = true;
             options.IncludeScopes = false;
         });
-        
+
         builder.Services.AddRequestDecompression();
         builder.Services.AddResponseCompression(options =>
         {
@@ -95,7 +95,8 @@ public static void Main(string[] args)
         builder.Services.AddControllers();
         builder.Services.AddSignalR();
         builder.Services.AddMemoryCache();
-        
+        builder.Services.AddSingleton<LogCacheService>();
+
         var connectionString = builder.Configuration.GetConnectionString("DefaultConnection");
 
         if (connectionString is null)
@@ -106,11 +107,11 @@ public static void Main(string[] args)
 
         builder.Services.AddSingleton<DbConnectionFactory>(dbFactory);
         builder.Services.AddTransient<IDbConnection>(sp => sp.GetRequiredService<DbConnectionFactory>().CreateConnection());
-        builder.Services.AddScoped<LogRepo>();
-        builder.Services.AddScoped<LogSummaryRepo>();
-        builder.Services.AddScoped<WorkQueueRepo>();
-        builder.Services.AddScoped<LogHubService>();
-        builder.Services.AddScoped<DatabaseInitializer>();
+        builder.Services.AddSingleton<LogRepo>();
+        builder.Services.AddSingleton<LogSummaryRepo>();
+        builder.Services.AddSingleton<WorkQueueRepo>();
+        builder.Services.AddSingleton<LogHubService>();
+        builder.Services.AddSingleton<DatabaseInitializer>();
         builder.Services.AddSingleton<LogApiMetrics>();
         builder.Services.AddSingleton<PasswordService>();
         builder.Services.AddSingleton<AuthService>();
@@ -124,7 +125,7 @@ public static void Main(string[] args)
         }).AddJwtBearer(options =>
         {
             var jwtSettings = builder.Configuration.GetSection("JwtSettings");
-            
+
             options.TokenValidationParameters = new TokenValidationParameters
             {
                 ValidateIssuer = true,