diff --git a/src/AgileConfig.Server.EventHandler/SyncEventHandlers.cs b/src/AgileConfig.Server.EventHandler/SyncEventHandlers.cs index d610196..8c26bdc 100644 --- a/src/AgileConfig.Server.EventHandler/SyncEventHandlers.cs +++ b/src/AgileConfig.Server.EventHandler/SyncEventHandlers.cs @@ -54,6 +54,9 @@ public class ConfigSyncEventHandler : IEventHandler return; } + // Clear existing failed records for this app+env before new sync + _retryService.ClearFailedRecord(timeline.AppId, timeline.Env); + // Convert to sync contexts var contexts = configs.Select(c => new SyncContext { diff --git a/src/AgileConfig.Server.SyncPlugin.Plugins.Etcd/EtcdSyncPlugin.cs b/src/AgileConfig.Server.SyncPlugin.Plugins.Etcd/EtcdSyncPlugin.cs index 0b27621..303d1b1 100644 --- a/src/AgileConfig.Server.SyncPlugin.Plugins.Etcd/EtcdSyncPlugin.cs +++ b/src/AgileConfig.Server.SyncPlugin.Plugins.Etcd/EtcdSyncPlugin.cs @@ -15,6 +15,9 @@ public class EtcdSyncPlugin : ISyncPlugin private SyncPluginConfig? _config; private HttpClient? _httpClient; private string _keyPrefix = "/agileconfig"; + private bool _allowOverwriteOtherData = false; + private int _maxTxnOperations = 500; + private string _syncStrategy = "FullReplace"; public string Name => "etcd"; public string DisplayName => "Etcd"; @@ -32,9 +35,21 @@ public class EtcdSyncPlugin : ISyncPlugin _config = config; var endpoints = config.Settings.GetValueOrDefault("endpoints", "http://localhost:2379"); - _keyPrefix = config.Settings.GetValueOrDefault("keyPrefix", "/agileconfig"); + _keyPrefix = config.Settings.GetValueOrDefault("keyPrefix", "/agileconfig").TrimEnd('/'); + _allowOverwriteOtherData = bool.TryParse(config.Settings.GetValueOrDefault("allowOverwriteOtherData", "false"), out var a) && a; + _maxTxnOperations = int.TryParse(config.Settings.GetValueOrDefault("maxTxnOperations", "500"), out var m) && m > 0 ? m : 500; + _syncStrategy = config.Settings.GetValueOrDefault("syncStrategy", "FullReplace"); - _logger.LogInformation("Initializing Etcd plugin with endpoints: {Endpoints}", endpoints); + // Validate key prefix + if (_keyPrefix == "/" || _keyPrefix.Length < 5 || !_keyPrefix.StartsWith('/')) + { + var error = $"Invalid key prefix '{_keyPrefix}'. Prefix must start with '/' and be at least 5 characters long, cannot be root path '/'."; + _logger.LogError(error); + return Task.FromResult(new SyncPluginResult { Success = false, Message = error }); + } + + _logger.LogInformation("Initializing Etcd plugin with endpoints: {Endpoints}, keyPrefix: {KeyPrefix}, allowOverwriteOtherData: {AllowOverwrite}, maxTxnOperations: {MaxTxn}, syncStrategy: {SyncStrategy}", + endpoints, _keyPrefix, _allowOverwriteOtherData, _maxTxnOperations, _syncStrategy); _httpClient = new HttpClient { @@ -51,7 +66,7 @@ public class EtcdSyncPlugin : ISyncPlugin } /// - /// Full sync: delete all + insert all + /// Full sync: use etcd transaction to atomically replace all configs /// public async Task SyncAllAsync(SyncContext[] contexts) { @@ -67,54 +82,198 @@ public class EtcdSyncPlugin : ISyncPlugin var env = contexts[0].Env; var prefix = $"{_keyPrefix}/{appId}/{env}/"; - // Step 1: Delete all existing keys with prefix - await DeleteRangeAsync(prefix); - _logger.LogInformation("Deleted all keys with prefix {Prefix}", prefix); - - // Step 2: Insert all new configs - foreach (var context in contexts) + List? existingKeys = null; + // Safety check: verify existing keys are valid AgileConfig keys if overwrite is not allowed + if (!_allowOverwriteOtherData) { - var key = BuildKey(context); - await PutAsync(key, context.Value); + existingKeys = await GetRangeKeysAsync(prefix); + foreach (var base64Key in existingKeys) + { + var key = System.Text.Encoding.UTF8.GetString(Convert.FromBase64String(base64Key)); + // Valid key format: {prefix}/{appId}/{env}/{group}/{key} + var relativePath = key.Substring(prefix.Length); + if (string.IsNullOrEmpty(relativePath) || !relativePath.Contains('/') || relativePath.StartsWith('/') || relativePath.EndsWith('/')) + { + var error = $"Found invalid key '{key}' under prefix '{prefix}' that does not match AgileConfig format. To overwrite anyway, set 'allowOverwriteOtherData' to true."; + _logger.LogError(error); + return new SyncPluginResult { Success = false, Message = error }; + } + } } - _logger.LogInformation("Synced {Count} configs to etcd for app {AppId} env {Env}", - contexts.Length, appId, env); + List operations; + int deletedCount = 0; + int addedCount = 0; + int updatedCount = 0; + + if (_syncStrategy.Equals("Incremental", StringComparison.OrdinalIgnoreCase)) + { + // Incremental sync: compare existing configs with new ones + var existingKvs = await GetRangeKvsAsync(prefix); + var existingDict = existingKvs.ToDictionary( + kv => System.Text.Encoding.UTF8.GetString(Convert.FromBase64String(kv.key)), + kv => System.Text.Encoding.UTF8.GetString(Convert.FromBase64String(kv.value)) + ); + + var newDict = contexts.ToDictionary( + c => BuildKey(c), + c => c.Value + ); + + operations = new List(); + + // Find keys to delete (exist in etcd but not in new configs) + foreach (var kv in existingDict) + { + if (!newDict.ContainsKey(kv.Key)) + { + operations.Add(new + { + request_delete_range = new + { + key = Convert.ToBase64String(System.Text.Encoding.UTF8.GetBytes(kv.Key)) + } + }); + deletedCount++; + } + } + + // Find keys to add/update + foreach (var kv in newDict) + { + if (!existingDict.TryGetValue(kv.Key, out var existingValue) || existingValue != kv.Value) + { + operations.Add(new + { + request_put = new + { + key = Convert.ToBase64String(System.Text.Encoding.UTF8.GetBytes(kv.Key)), + value = Convert.ToBase64String(System.Text.Encoding.UTF8.GetBytes(kv.Value)) + } + }); + if (!existingDict.ContainsKey(kv.Key)) + addedCount++; + else + updatedCount++; + } + } + + _logger.LogInformation("Incremental sync: {Deleted} to delete, {Added} to add, {Updated} to update", + deletedCount, addedCount, updatedCount); + } + else + { + // Full replace sync: delete all then insert all + operations = new List + { + // Add single delete_range operation to delete all existing keys with prefix + new + { + request_delete_range = new + { + key = Convert.ToBase64String(System.Text.Encoding.UTF8.GetBytes(prefix)), + range_end = Convert.ToBase64String(System.Text.Encoding.UTF8.GetBytes(GetRangeEnd(prefix))) + } + } + }; + existingKeys ??= await GetRangeKeysAsync(prefix); + deletedCount = existingKeys.Count; + + // Add put operations for all new configs + foreach (var context in contexts) + { + var key = BuildKey(context); + operations.Add(new + { + request_put = new + { + key = Convert.ToBase64String(System.Text.Encoding.UTF8.GetBytes(key)), + value = Convert.ToBase64String(System.Text.Encoding.UTF8.GetBytes(context.Value)) + } + }); + addedCount++; + } + + _logger.LogInformation("Full replace sync: {Deleted} old keys to delete, {Added} new keys to add", + deletedCount, addedCount); + } + + // Split operations into batches and execute + var batchSize = _maxTxnOperations; + var totalBatches = (int)Math.Ceiling((double)operations.Count / batchSize); + var processed = 0; + + for (var i = 0; i < totalBatches; i++) + { + var batch = operations.Skip(i * batchSize).Take(batchSize).ToList(); + try + { + var batchTxn = new + { + success = batch + }; + + var batchResponse = await _httpClient.PostAsJsonAsync("/v3/kv/txn", batchTxn); + batchResponse.EnsureSuccessStatusCode(); + processed += batch.Count; + _logger.LogDebug("Processed batch {Current}/{Total}, {Processed} operations completed", i + 1, totalBatches, processed); + } + catch (HttpRequestException ex) when (ex.Message.Contains("request too large") && batchSize > 10) + { + // Auto reduce batch size on request too large error + batchSize = Math.Max(10, batchSize / 2); + _logger.LogWarning("Request too large, reducing batch size to {BatchSize} and retrying batch", batchSize); + i--; // Retry current batch + } + } + + _logger.LogInformation("Successfully synced configs to etcd for app {AppId} env {Env}: {Deleted} deleted, {Added} added, {Updated} updated in {Batches} batches", + appId, env, deletedCount, addedCount, updatedCount, totalBatches); return new SyncPluginResult { Success = true, - Message = $"Synced {contexts.Length} configs" + Message = $"Synced: {deletedCount} deleted, {addedCount} added, {updatedCount} updated in {totalBatches} batches" }; } catch (Exception ex) { - _logger.LogError(ex, "Failed to sync configs to etcd"); + _logger.LogError(ex, "Failed to sync configs to etcd (transaction rolled back, no changes applied)"); return new SyncPluginResult { Success = false, Message = ex.Message, Exception = ex }; } } /// - /// Put a key-value pair + /// Get all keys with given prefix /// - private async Task PutAsync(string key, string value) + private async Task> GetRangeKeysAsync(string prefix) { - var request = new + var keys = new List(); + var rangeRequest = new { - key = Convert.ToBase64String(System.Text.Encoding.UTF8.GetBytes(key)), - value = Convert.ToBase64String(System.Text.Encoding.UTF8.GetBytes(value)) + key = Convert.ToBase64String(System.Text.Encoding.UTF8.GetBytes(prefix)), + range_end = Convert.ToBase64String(System.Text.Encoding.UTF8.GetBytes(GetRangeEnd(prefix))), + keys_only = true }; - var response = await _httpClient.PostAsJsonAsync("/v3/kv/put", request); - response.EnsureSuccessStatusCode(); + var rangeResponse = await _httpClient.PostAsJsonAsync("/v3/kv/range", rangeRequest); + rangeResponse.EnsureSuccessStatusCode(); + + var rangeResult = await rangeResponse.Content.ReadFromJsonAsync(); + + if (rangeResult?.kvs != null) + { + keys.AddRange(rangeResult.kvs.Select(kv => kv.key)); + } + + return keys; } /// - /// Delete all keys with given prefix + /// Get all key-value pairs with given prefix /// - private async Task DeleteRangeAsync(string prefix) + private async Task> GetRangeKvsAsync(string prefix) { - // Range request to get all keys with prefix var rangeRequest = new { key = Convert.ToBase64String(System.Text.Encoding.UTF8.GetBytes(prefix)), @@ -125,21 +284,8 @@ public class EtcdSyncPlugin : ISyncPlugin rangeResponse.EnsureSuccessStatusCode(); var rangeResult = await rangeResponse.Content.ReadFromJsonAsync(); - - if (rangeResult?.kvs != null && rangeResult.kvs.Count > 0) - { - // Delete each key - foreach (var kvp in rangeResult.kvs) - { - var deleteRequest = new - { - key = kvp.key - }; - - var deleteResponse = await _httpClient.PostAsJsonAsync("/v3/kv/deleterange", deleteRequest); - deleteResponse.EnsureSuccessStatusCode(); - } - } + + return rangeResult?.kvs ?? new List(); } /// diff --git a/src/AgileConfig.Server.SyncPlugin/Retry/FailedSyncRecord.cs b/src/AgileConfig.Server.SyncPlugin/Retry/FailedSyncRecord.cs index dd05f49..e0961b8 100644 --- a/src/AgileConfig.Server.SyncPlugin/Retry/FailedSyncRecord.cs +++ b/src/AgileConfig.Server.SyncPlugin/Retry/FailedSyncRecord.cs @@ -41,4 +41,14 @@ public class FailedSyncRecord /// Last error message /// public string? LastError { get; set; } + + /// + /// Next retry time + /// + public DateTimeOffset? NextRetryTime { get; set; } + + /// + /// Whether the record is in circuit breaker state + /// + public bool IsCircuitBroken { get; set; } } diff --git a/src/AgileConfig.Server.SyncPlugin/Retry/SyncRetryService.cs b/src/AgileConfig.Server.SyncPlugin/Retry/SyncRetryService.cs index 9de83e7..774d9d4 100644 --- a/src/AgileConfig.Server.SyncPlugin/Retry/SyncRetryService.cs +++ b/src/AgileConfig.Server.SyncPlugin/Retry/SyncRetryService.cs @@ -20,7 +20,9 @@ public class SyncRetryService // Configuration private const int MaxRetryCount = 10; - private const int RetryIntervalSeconds = 30; + private const int CircuitBreakDurationMinutes = 60; + private const int MinRetryDelaySeconds = 1; + private const int MaxRetryDelaySeconds = 60; public SyncRetryService( SyncEngine syncEngine, @@ -47,20 +49,38 @@ public class SyncRetryService existing.RetryCount++; existing.LastRetryTime = DateTimeOffset.UtcNow; existing.LastError = errorMessage; - _logger.LogWarning("Sync failed for app {AppId} env {Env}, retry count: {Count}", - appId, env, existing.RetryCount); + + if (existing.RetryCount >= MaxRetryCount) + { + // Circuit break: stop retrying for CircuitBreakDurationMinutes + existing.IsCircuitBroken = true; + existing.NextRetryTime = DateTimeOffset.UtcNow.AddMinutes(CircuitBreakDurationMinutes); + _logger.LogError("Sync failed for app {AppId} env {Env} for {Count} times, circuit breaker activated, next retry after {NextRetry}", + appId, env, existing.RetryCount, existing.NextRetryTime); + } + else + { + // Exponential backoff: 2^retryCount seconds, capped at MaxRetryDelaySeconds + var delaySeconds = Math.Min(MaxRetryDelaySeconds, Math.Pow(2, existing.RetryCount)); + existing.NextRetryTime = DateTimeOffset.UtcNow.AddSeconds(delaySeconds); + _logger.LogWarning("Sync failed for app {AppId} env {Env}, retry count: {Count}, next retry after {NextRetry} ({Delay}s)", + appId, env, existing.RetryCount, existing.NextRetryTime, delaySeconds); + } } else { - _failedRecords.Add(new FailedSyncRecord + var newRecord = new FailedSyncRecord { AppId = appId, Env = env, FailedTime = DateTimeOffset.UtcNow, RetryCount = 1, - LastError = errorMessage - }); - _logger.LogWarning("Recorded failed sync for app {AppId} env {Env}", appId, env); + LastError = errorMessage, + NextRetryTime = DateTimeOffset.UtcNow.AddSeconds(MinRetryDelaySeconds) + }; + _failedRecords.Add(newRecord); + _logger.LogWarning("Recorded failed sync for app {AppId} env {Env}, first retry after {NextRetry} ({Delay}s)", + appId, env, newRecord.NextRetryTime, MinRetryDelaySeconds); } } } @@ -75,10 +95,19 @@ public class SyncRetryService lock (_lock) { + var now = DateTimeOffset.UtcNow; // Get records that are ready to retry recordsToProcess = _failedRecords - .Where(x => x.RetryCount < MaxRetryCount) + .Where(x => x.NextRetryTime <= now) .ToList(); + + // Reset circuit breaker for records where next retry time has passed + foreach (var record in recordsToProcess.Where(x => x.IsCircuitBroken)) + { + record.IsCircuitBroken = false; + record.RetryCount = 0; // Reset retry count after circuit break + _logger.LogInformation("Circuit breaker reset for app {AppId} env {Env}, resuming retries", record.AppId, record.Env); + } } if (!recordsToProcess.Any()) @@ -192,4 +221,19 @@ public class SyncRetryService _failedRecords.Clear(); } } + + /// + /// Clear failed records for specific appId and env + /// + public void ClearFailedRecord(string appId, string env) + { + lock (_lock) + { + var removed = _failedRecords.RemoveAll(x => x.AppId == appId && x.Env == env); + if (removed > 0) + { + _logger.LogDebug("Cleared {Count} failed records for app {AppId} env {Env}", removed, appId, env); + } + } + } }