Refactor agent config and heartbeat handling

Enhanced agent configuration management by introducing global settings and leveraging the heartbeat mechanism to dynamically deliver configurations.

- Added `HeartbeatResponse` to streamline config updates via heartbeat.
- Increased heartbeat interval to 60 seconds.
- Introduced global configuration endpoints and UI for shared settings.
- Added `AgentId` to logs for better traceability.
- Updated database schema to include `AgentId` column and index.
- Improved log details modal with additional metadata.
- Refined UI for managing global and per-agent configurations.
- Removed periodic config polling in favor of heartbeat-based updates.

These changes improve performance, usability, and traceability.

Author: Timothy Dodd

src/LogMkAgent/Common/ApiJsonSerializerContext.cs

@@ -11,6 +11,7 @@ namespace LogMkAgent;
 [JsonSerializable(typeof(ValidationSettings))]
 [JsonSerializable(typeof(AgentHeartbeat))]
 [JsonSerializable(typeof(AgentConfig))]
+[JsonSerializable(typeof(HeartbeatResponse))]
 public partial class ApiJsonSerializerContext : JsonSerializerContext
 {
 }

src/LogMkAgent/Services/AgentRegistrationService.cs

@@ -1,6 +1,8 @@
+using System.Net.Http.Json;
 using System.Net.NetworkInformation;
 using System.Net.Sockets;
 using System.Reflection;
+using System.Text.Json;
 using LogMkCommon;
 
 namespace LogMkAgent.Services;
@@ -17,8 +19,13 @@ public class AgentRegistrationService : BackgroundService
     private readonly LogWatcher _logWatcher;
     private readonly ILogger<AgentRegistrationService> _logger;
 
-    private readonly TimeSpan _heartbeatInterval = TimeSpan.FromSeconds(30);
-    private readonly TimeSpan _configPollInterval = TimeSpan.FromSeconds(60);
+    private readonly TimeSpan _heartbeatInterval = TimeSpan.FromSeconds(60);
+
+    private static readonly JsonSerializerOptions _jsonOptions = new()
+    {
+        TypeInfoResolver = new ApiJsonSerializerContext(),
+        PropertyNameCaseInsensitive = true
+    };
 
     private readonly DateTime _startedAt = DateTime.UtcNow;
     private readonly string _agentId;
@@ -42,48 +49,45 @@ public AgentRegistrationService(
         _version = Assembly.GetExecutingAssembly().GetName().Version?.ToString();
         _ipAddress = TryGetLocalIp();
 
+        // Stamp every outgoing log line with this agent's id so the API/UI
+        // can show which node a log came from.
+        _batchingService.AgentId = _agentId;
+
         _logger.LogInformation("AgentRegistrationService starting: AgentId={AgentId}, Hostname={Hostname}, Version={Version}",
             _agentId, _hostname, _version);
     }
 
     protected override async Task ExecuteAsync(CancellationToken stoppingToken)
     {
-        // Pull config once on startup before the regular polling loop
-        await PullConfigAsync(stoppingToken).ConfigureAwait(false);
-
-        var lastConfigPull = DateTime.UtcNow;
+        // Send an initial heartbeat right away so the agent's config is applied
+        // before the first batch is processed.
+        try
+        {
+            await SendHeartbeatAsync(stoppingToken).ConfigureAwait(false);
+        }
+        catch (Exception ex)
+        {
+            _logger.LogWarning(ex, "Initial heartbeat failed");
+        }
 
         while (!stoppingToken.IsCancellationRequested)
         {
             try
             {
-                await SendHeartbeatAsync(stoppingToken).ConfigureAwait(false);
-            }
-            catch (Exception ex)
-            {
-                _logger.LogWarning(ex, "Heartbeat failed");
+                await Task.Delay(_heartbeatInterval, stoppingToken).ConfigureAwait(false);
             }
-
-            if (DateTime.UtcNow - lastConfigPull >= _configPollInterval)
+            catch (OperationCanceledException)
             {
-                try
-                {
-                    await PullConfigAsync(stoppingToken).ConfigureAwait(false);
-                    lastConfigPull = DateTime.UtcNow;
-                }
-                catch (Exception ex)
-                {
-                    _logger.LogWarning(ex, "Config pull failed");
-                }
+                break;
             }
 
             try
             {
-                await Task.Delay(_heartbeatInterval, stoppingToken).ConfigureAwait(false);
+                await SendHeartbeatAsync(stoppingToken).ConfigureAwait(false);
             }
-            catch (OperationCanceledException)
+            catch (Exception ex)
             {
-                break;
+                _logger.LogWarning(ex, "Heartbeat failed");
             }
         }
     }
@@ -113,28 +117,24 @@ private async Task SendHeartbeatAsync(CancellationToken cancellationToken)
         if (!response.IsSuccessStatusCode)
         {
             _logger.LogDebug("Heartbeat returned non-success status {StatusCode}", response.StatusCode);
+            return;
         }
-    }
 
-    private async Task PullConfigAsync(CancellationToken cancellationToken)
-    {
         try
         {
-            var config = await _apiClient.GetDataAsync<AgentConfig>($"api/agents/{Uri.EscapeDataString(_agentId)}/config")
-                                         .ConfigureAwait(false);
+            var payload = await response.Content
+                .ReadFromJsonAsync<HeartbeatResponse>(_jsonOptions, cancellationToken)
+                .ConfigureAwait(false);
 
-            if (config == null)
+            if (payload?.Config != null)
             {
-                _logger.LogDebug("No agent config returned for {AgentId}", _agentId);
-                return;
+                _logWatcher.ApplyConfig(payload.Config);
+                _batchingService.ApplyConfig(payload.Config);
             }
-
-            _logWatcher.ApplyConfig(config);
-            _batchingService.ApplyConfig(config);
         }
-        catch (HttpRequestException ex)
+        catch (Exception ex)
         {
-            _logger.LogDebug(ex, "Could not reach API to pull config");
+            _logger.LogDebug(ex, "Failed to parse heartbeat response");
         }
     }
 

src/LogMkAgent/Services/BatchService.cs

@@ -115,6 +115,8 @@ public void ApplyConfig(AgentConfig config)
             _options.MaxBatchSize, _maxBatchesPerMinute, _options.MaxQueueSize);
     }
 
+    public string? AgentId { get; set; }
+
     public void AddData(LogLine data)
     {
         if (_disposed)
@@ -129,6 +131,11 @@ public void AddData(LogLine data)
             return;
         }
 
+        if (string.IsNullOrEmpty(data.AgentId))
+        {
+            data.AgentId = AgentId;
+        }
+
         // Enforce queue size limit - drop 10% of queue to make room and avoid per-item churn
         if (_options.MaxQueueSize > 0 && _batchData.Count >= _options.MaxQueueSize)
         {

src/LogMkApi/Controllers/AgentController.cs

@@ -20,10 +20,12 @@ public AgentController(AgentRepo agentRepo, ILogger<AgentController> logger)
         _logger = logger;
     }
 
-    // Called by agents — anonymous like other agent endpoints
+    // Called by agents — anonymous like other agent endpoints. Returns the
+    // latest config so changes saved in the web UI are applied on the next
+    // heartbeat (no separate poll needed).
     [AllowAnonymous]
     [HttpPost("heartbeat")]
-    public async Task<IActionResult> Heartbeat([FromBody] AgentHeartbeat heartbeat)
+    public async Task<ActionResult<HeartbeatResponse>> Heartbeat([FromBody] AgentHeartbeat heartbeat)
     {
         if (heartbeat == null || string.IsNullOrWhiteSpace(heartbeat.AgentId))
         {
@@ -33,7 +35,12 @@ public async Task<IActionResult> Heartbeat([FromBody] AgentHeartbeat heartbeat)
         try
         {
             await _agentRepo.UpsertHeartbeatAsync(heartbeat);
-            return Ok(new { Status = "ok", ReceivedAt = DateTime.UtcNow });
+            var config = await _agentRepo.GetMergedConfigAsync(heartbeat.AgentId);
+            return Ok(new HeartbeatResponse
+            {
+                ReceivedAt = DateTime.UtcNow,
+                Config = config
+            });
         }
         catch (Exception ex)
         {
@@ -42,8 +49,32 @@ public async Task<IActionResult> Heartbeat([FromBody] AgentHeartbeat heartbeat)
         }
     }
 
-    // Called by agents — anonymous so the agent can fetch its own config
-    [AllowAnonymous]
+    [Authorize]
+    [HttpGet("global/config")]
+    public async Task<ActionResult<AgentConfig>> GetGlobalConfig()
+    {
+        var config = await _agentRepo.GetGlobalConfigAsync();
+        return Ok(config);
+    }
+
+    [Authorize]
+    [HttpPut("global/config")]
+    public async Task<ActionResult<AgentConfig>> UpdateGlobalConfig([FromBody] AgentConfig config)
+    {
+        if (config == null)
+        {
+            return BadRequest(new { Error = "Config is required" });
+        }
+
+        await _agentRepo.SaveGlobalConfigAsync(config, User.Identity?.Name);
+        var saved = await _agentRepo.GetGlobalConfigAsync();
+        _logger.LogInformation("Global agent config updated by {User}", User.Identity?.Name);
+        return Ok(saved);
+    }
+
+    // Called by the web UI to load raw per-agent config for editing.
+    // Agents themselves get config via the heartbeat response (merged).
+    [Authorize]
     [HttpGet("{agentId}/config")]
     public async Task<ActionResult<AgentConfig>> GetConfig(string agentId)
     {
@@ -52,6 +83,11 @@ public async Task<ActionResult<AgentConfig>> GetConfig(string agentId)
             return BadRequest(new { Error = "AgentId is required" });
         }
 
+        if (string.Equals(agentId, AgentRepo.GlobalConfigId, StringComparison.Ordinal))
+        {
+            return BadRequest(new { Error = "Use /api/agents/global/config for global settings" });
+        }
+
         var config = await _agentRepo.GetConfigAsync(agentId);
         return Ok(config);
     }
@@ -82,6 +118,11 @@ public async Task<ActionResult<AgentConfig>> UpdateConfig(string agentId, [FromB
             return BadRequest(new { Error = "Config is required" });
         }
 
+        if (string.Equals(agentId, AgentRepo.GlobalConfigId, StringComparison.Ordinal))
+        {
+            return BadRequest(new { Error = "Use /api/agents/global/config for global settings" });
+        }
+
         if (config.MaxBatchSize <= 0 || config.MaxBatchSize > 5000)
         {
             return BadRequest(new { Error = "MaxBatchSize must be between 1 and 5000" });

src/LogMkApi/Controllers/LogController.cs

@@ -150,7 +150,8 @@ public async Task<ActionResult<LogResponse>> Create(
                     SequenceNumber = logLine.SequenceNumber,
                     Fingerprint = logLine.Fingerprint,
                     BatchId = batchId,
-                    ReceivedAt = receivedAt.UtcDateTime
+                    ReceivedAt = receivedAt.UtcDateTime,
+                    AgentId = logLine.AgentId
                 };
 
                 validLogs.Add(logEntity);

src/LogMkApi/Data/AgentRepo.cs

@@ -7,6 +7,8 @@ namespace LogMkApi.Data;
 
 public class AgentRepo
 {
+    public const string GlobalConfigId = "__global__";
+
     private readonly DbConnectionFactory _dbFactory;
     private readonly ILogger<AgentRepo> _logger;
 
@@ -69,11 +71,63 @@ public async Task UpsertHeartbeatAsync(AgentHeartbeat heartbeat)
     public async Task<List<AgentInfo>> GetAllAsync()
     {
         using var db = _dbFactory.CreateConnection();
-        var query = db.From<AgentRecord>().OrderByDescending(x => x.LastSeenAt);
+        var query = db.From<AgentRecord>()
+            .Where(x => x.AgentId != GlobalConfigId)
+            .OrderByDescending(x => x.LastSeenAt);
         var records = await db.SelectAsync(query);
         return records.Select(MapToInfo).ToList();
     }
 
+    public Task<AgentConfig> GetGlobalConfigAsync() => GetConfigAsync(GlobalConfigId);
+
+    public Task SaveGlobalConfigAsync(AgentConfig config, string? updatedBy) =>
+        SaveConfigAsync(GlobalConfigId, config, updatedBy);
+
+    /// <summary>
+    /// Returns the per-agent config merged with the global config.
+    /// Scalar fields (MaxBatchSize, MaxDaysOld, DefaultLogLevel, ...) come from
+    /// the per-agent config. Collection fields (PodLogLevels, IgnoredPods) are
+    /// unioned across global + per-agent, with the per-agent value winning for
+    /// any key conflict.
+    /// </summary>
+    public async Task<AgentConfig> GetMergedConfigAsync(string agentId)
+    {
+        if (string.Equals(agentId, GlobalConfigId, StringComparison.Ordinal))
+        {
+            return await GetGlobalConfigAsync();
+        }
+
+        var globalConfig = await GetGlobalConfigAsync();
+        var agentConfig = await GetConfigAsync(agentId);
+
+        var mergedPodLevels = new Dictionary<string, string>(globalConfig.PodLogLevels, StringComparer.OrdinalIgnoreCase);
+        foreach (var kvp in agentConfig.PodLogLevels)
+        {
+            mergedPodLevels[kvp.Key] = kvp.Value;
+        }
+
+        var mergedIgnored = new HashSet<string>(globalConfig.IgnoredPods, StringComparer.OrdinalIgnoreCase);
+        foreach (var pod in agentConfig.IgnoredPods)
+        {
+            mergedIgnored.Add(pod);
+        }
+
+        return new AgentConfig
+        {
+            AgentId = agentId,
+            DefaultLogLevel = agentConfig.DefaultLogLevel,
+            PodLogLevels = mergedPodLevels,
+            IgnoredPods = mergedIgnored.ToList(),
+            MaxDaysOld = agentConfig.MaxDaysOld,
+            MaxLineLength = agentConfig.MaxLineLength,
+            MaxBatchSize = agentConfig.MaxBatchSize,
+            MaxBatchesPerMinute = agentConfig.MaxBatchesPerMinute,
+            MaxQueueSize = agentConfig.MaxQueueSize,
+            UpdatedAt = agentConfig.UpdatedAt > globalConfig.UpdatedAt ? agentConfig.UpdatedAt : globalConfig.UpdatedAt,
+            UpdatedBy = agentConfig.UpdatedBy ?? globalConfig.UpdatedBy
+        };
+    }
+
     public async Task<AgentInfo?> GetByIdAsync(string agentId)
     {
         using var db = _dbFactory.CreateConnection();

src/LogMkApi/Data/DatabaseInitializer.cs

@@ -43,10 +43,35 @@ public void CreateTable()
             }
             db.CreateTableIfNotExists<RefreshToken>(true);
             // Ensure critical indexes exist (migration for existing databases)
+            EnsureLogColumns(db);
             EnsureLogIndexes(db);
         }
     }
 
+    private void EnsureLogColumns(System.Data.IDbConnection db)
+    {
+        try
+        {
+            var checkSql = @"
+                SELECT COUNT(*)
+                FROM information_schema.columns
+                WHERE table_schema = DATABASE()
+                AND table_name = 'Log'
+                AND column_name = 'AgentId'";
+
+            var exists = db.ExecuteScalar<int>(checkSql) > 0;
+            if (!exists)
+            {
+                _logger.LogInformation("Adding AgentId column to Log table");
+                db.Execute("ALTER TABLE `Log` ADD COLUMN `AgentId` VARCHAR(100) NULL");
+            }
+        }
+        catch (Exception ex)
+        {
+            _logger.LogWarning(ex, "Failed to ensure Log.AgentId column. Add it manually: ALTER TABLE `Log` ADD COLUMN `AgentId` VARCHAR(100) NULL");
+        }
+    }
+
     private void EnsureLogIndexes(System.Data.IDbConnection db)
     {
         var missingIndexes = new List<(string Name, string Sql)>();
@@ -56,7 +81,8 @@ private void EnsureLogIndexes(System.Data.IDbConnection db)
             var indexesToCheck = new[]
             {
                 ("Deployment_Pod_TimeStamp_idx", "CREATE INDEX Deployment_Pod_TimeStamp_idx ON `Log` (Deployment, Pod, TimeStamp)"),
-                ("LogDate_idx", "CREATE INDEX LogDate_idx ON `Log` (LogDate)")
+                ("LogDate_idx", "CREATE INDEX LogDate_idx ON `Log` (LogDate)"),
+                ("AgentId_idx", "CREATE INDEX AgentId_idx ON `Log` (AgentId)")
             };
 
             foreach (var (name, createSql) in indexesToCheck)

src/LogMkApi/Data/Models/Log.cs

@@ -41,6 +41,9 @@ public class Log
     [MaxLength(50)]
     public string BatchId { get; set; } = string.Empty;
     public DateTime ReceivedAt { get; set; }
+    [Index]
+    [MaxLength(100)]
+    public string? AgentId { get; set; }
     // Computed property for pod key (useful for queries)
     [Ignore] // Don't store in database
     public string PodKey => $"{Deployment}:{Pod}";