diff --git a/src/Aspire.Hosting/Dcp/DcpKubernetesClient.cs b/src/Aspire.Hosting/Dcp/DcpKubernetesClient.cs index 909f92afd83..9cebb583f7a 100644 --- a/src/Aspire.Hosting/Dcp/DcpKubernetesClient.cs +++ b/src/Aspire.Hosting/Dcp/DcpKubernetesClient.cs @@ -23,6 +23,10 @@ internal sealed class DcpKubernetesClient : k8s.Kubernetes public DcpKubernetesClient(KubernetesClientConfiguration config, params DelegatingHandler[] handlers) : base(config, handlers) { + // KubernetesService supplies the operation deadlines. The KubernetesClient NuGet package's + // separate HttpClientTimeout timer defaults to 100 seconds and would otherwise truncate + // configured budgets and surface as an unrelated cancellation. + HttpClientTimeout = Timeout.InfiniteTimeSpan; } /// @@ -115,10 +119,6 @@ public async Task> ReadSubResourceAsStreamAsync( ArgumentException.ThrowIfNullOrWhiteSpace(name, nameof(name)); ArgumentException.ThrowIfNullOrWhiteSpace(subResource, nameof(subResource)); - using var cts = CancellationTokenSource.CreateLinkedTokenSource(cancellationToken); - cts.CancelAfter(HttpClientTimeout); - cancellationToken = cts.Token; - string url; if (string.IsNullOrEmpty(namespaceParameter)) { @@ -155,10 +155,6 @@ public async Task> ReadSubResourceAsStreamAsync( /// public async Task GetExecutionDocumentAsync(CancellationToken cancellationToken = default) { - using var cts = CancellationTokenSource.CreateLinkedTokenSource(cancellationToken); - cts.CancelAfter(HttpClientTimeout); - cancellationToken = cts.Token; - var httpRequest = new HttpRequestMessage { Method = HttpMethod.Get, @@ -181,10 +177,6 @@ public async Task PatchExecutionDocumentAsync( ApiServerExecution apiServerExecution, CancellationToken cancellationToken = default) { - using var cts = CancellationTokenSource.CreateLinkedTokenSource(cancellationToken); - cts.CancelAfter(HttpClientTimeout); - cancellationToken = cts.Token; - var httpRequest = new HttpRequestMessage { Method = HttpMethod.Patch, diff --git a/src/Aspire.Hosting/Dcp/DcpOptions.cs b/src/Aspire.Hosting/Dcp/DcpOptions.cs index ab3d6b28629..6c5eae6c782 100644 --- a/src/Aspire.Hosting/Dcp/DcpOptions.cs +++ b/src/Aspire.Hosting/Dcp/DcpOptions.cs @@ -94,6 +94,29 @@ internal sealed class DcpOptions public int KubernetesConfigReadRetryIntervalMilliseconds { get; set; } = 100; + /// + /// Additional time added to before the first successful API operation. + /// The combined budget covers client setup and the API operation. + /// Defaults to 20 seconds and is configured through DcpPublisher:KubernetesInitializationAdditionalTimeout. + /// Must be non-negative, and the combined budget must not exceed ten minutes. + /// + public TimeSpan KubernetesInitializationAdditionalTimeout { get; set; } = TimeSpan.FromSeconds(20); + + /// + /// The base API operation budget, extended by until the first successful API operation. + /// Applies to connection establishment rather than stream lifetime for watches and logs. + /// Defaults to 40 seconds and is configured through DcpPublisher:KubernetesApiTimeout. + /// Valid values range from one second to ten minutes. + /// + public TimeSpan KubernetesApiTimeout { get; set; } = TimeSpan.FromSeconds(40); + + /// + /// The total budget for creating a DCP object, including client setup, retries, and reconciliation after ambiguous failures. + /// Defaults to two minutes and is configured through DcpPublisher:KubernetesCreateRecoveryTimeout. + /// Valid values range from one second to ten minutes. + /// + public TimeSpan KubernetesCreateRecoveryTimeout { get; set; } = TimeSpan.FromMinutes(2); + /// /// The duration to wait for the container runtime to become healthy before aborting startup. /// @@ -175,8 +198,33 @@ public ValidateOptionsResult Validate(string? name, DcpOptions options) builder.AddError("The proxyless endpoint port range start must be less than or equal to the range end.", nameof(options.ProxylessEndpointPortRangeStart)); } + var apiTimeoutValid = ValidateKubernetesTimeout(builder, options.KubernetesApiTimeout, nameof(options.KubernetesApiTimeout), "API"); + ValidateKubernetesTimeout(builder, options.KubernetesCreateRecoveryTimeout, nameof(options.KubernetesCreateRecoveryTimeout), "create recovery"); + + if (options.KubernetesInitializationAdditionalTimeout < TimeSpan.Zero) + { + builder.AddError("The Kubernetes additional initialization timeout must be non-negative.", nameof(options.KubernetesInitializationAdditionalTimeout)); + } + // Compare the remaining headroom rather than adding potentially overflowing configuration values. + else if (apiTimeoutValid && + options.KubernetesInitializationAdditionalTimeout > TimeSpan.FromMinutes(10) - options.KubernetesApiTimeout) + { + builder.AddError("The combined Kubernetes API and additional initialization timeouts must not exceed ten minutes.", nameof(options.KubernetesInitializationAdditionalTimeout)); + } + return builder.Build(); } + + private static bool ValidateKubernetesTimeout(ValidateOptionsResultBuilder builder, TimeSpan timeout, string propertyName, string description) + { + if (timeout < TimeSpan.FromSeconds(1) || timeout > TimeSpan.FromMinutes(10)) + { + builder.AddError($"The Kubernetes {description} timeout must be between one second and ten minutes.", propertyName); + return false; + } + + return true; + } } internal class ConfigureDefaultDcpOptions( @@ -330,6 +378,9 @@ public void Configure(DcpOptions options) options.KubernetesConfigReadRetryCount = dcpPublisherConfiguration.GetValue(nameof(options.KubernetesConfigReadRetryCount), options.KubernetesConfigReadRetryCount); options.KubernetesConfigReadRetryIntervalMilliseconds = dcpPublisherConfiguration.GetValue(nameof(options.KubernetesConfigReadRetryIntervalMilliseconds), options.KubernetesConfigReadRetryIntervalMilliseconds); + options.KubernetesInitializationAdditionalTimeout = dcpPublisherConfiguration.GetValue(nameof(options.KubernetesInitializationAdditionalTimeout), options.KubernetesInitializationAdditionalTimeout); + options.KubernetesApiTimeout = dcpPublisherConfiguration.GetValue(nameof(options.KubernetesApiTimeout), options.KubernetesApiTimeout); + options.KubernetesCreateRecoveryTimeout = dcpPublisherConfiguration.GetValue(nameof(options.KubernetesCreateRecoveryTimeout), options.KubernetesCreateRecoveryTimeout); if (!string.IsNullOrEmpty(dcpPublisherConfiguration[nameof(options.ResourceNameSuffix)])) { diff --git a/src/Aspire.Hosting/Dcp/KubernetesService.cs b/src/Aspire.Hosting/Dcp/KubernetesService.cs index dfe66665ecc..cce02e12874 100644 --- a/src/Aspire.Hosting/Dcp/KubernetesService.cs +++ b/src/Aspire.Hosting/Dcp/KubernetesService.cs @@ -3,7 +3,10 @@ using System.Diagnostics; using System.Globalization; +using System.Net; +using System.Net.Http.Headers; using System.Runtime.CompilerServices; +using System.Text.Json; using System.Threading.Channels; using Aspire.Hosting.Diagnostics; using Aspire.Hosting.Dcp.Model; @@ -93,8 +96,6 @@ internal sealed class KubernetesService(ILogger logger, IOpti private ResiliencePipeline? _resiliencePipeline; private bool _disposed; - public TimeSpan MaxRetryDuration { get; set; } = TimeSpan.FromSeconds(20); - public TimeSpan KubernetesInitializationTimeout { get; set; } = TimeSpan.FromSeconds(60); internal TimeSpan WatchRestartInterval { get; set; } = TimeSpan.FromMinutes(5); public Task GetAsync(string name, string? namespaceParameter = null, CancellationToken cancellationToken = default) @@ -128,37 +129,75 @@ public Task GetAsync(string name, string? namespaceParameter = null, Cance cancellationToken); } - public Task CreateAsync(T obj, CancellationToken cancellationToken = default) + public async Task CreateAsync(T obj, CancellationToken cancellationToken = default) where T : CustomResource, IKubernetesStaticMetadata { ObjectDisposedException.ThrowIf(_disposed, this); var resourceType = GetResourceFor(); var namespaceParameter = obj.Namespace(); + var name = obj.Metadata.Name; + var recoveryTimeout = dcpOptions.Value.KubernetesCreateRecoveryTimeout; + + // A timed-out write can still commit. Confirm ambiguous creates before retrying the same object. + // Kubernetes explicitly allows operations to complete after a timeout: + // https://github.com/kubernetes/apimachinery/blob/master/pkg/api/errors/errors.go + return await new ResiliencePipelineBuilder() + .AddTimeout(new TimeoutStrategyOptions + { + Timeout = recoveryTimeout, + }) + .Build() + .ExecuteAsync(async recoveryCancellationToken => + { + var hasAmbiguousCreate = false; + var retryDelay = s_initialRetryDelay; - return ExecuteWithRetry( - DcpApiOperationType.Create, - T.ObjectKind, - async (kubernetes, operationCancellationToken) => - { - var response = string.IsNullOrEmpty(namespaceParameter) - ? await kubernetes.CustomObjects.CreateClusterCustomObjectWithHttpMessagesAsync( - obj, - GroupVersion.Group, - GroupVersion.Version, - resourceType, - cancellationToken: operationCancellationToken).ConfigureAwait(false) - : await kubernetes.CustomObjects.CreateNamespacedCustomObjectWithHttpMessagesAsync( - obj, - GroupVersion.Group, - GroupVersion.Version, - namespaceParameter, - resourceType, - cancellationToken: operationCancellationToken).ConfigureAwait(false); + while (true) + { + DateTimeOffset? retryAfter = null; + try + { + return await CreateOnceAsync(obj, resourceType, namespaceParameter, recoveryCancellationToken).ConfigureAwait(false); + } + catch (KubernetesClientInitializationException) + { + retryDelay = await DelayCreateRetryAsync(null, retryDelay, recoveryTimeout, recoveryCancellationToken).ConfigureAwait(false); + continue; + } + catch (HttpOperationException ex) when (hasAmbiguousCreate && IsAlreadyExists(ex)) + { + // A previous ambiguous create may have committed after its confirmation returned 404. + retryAfter = GetCreateRetryAfter(ex, recoveryTimeout); + } + catch (Exception ex) when (RetryOnTransientCreateErrors(ex)) + { + hasAmbiguousCreate = true; + retryAfter = GetCreateRetryAfter(ex, recoveryTimeout); + } - return KubernetesJson.Deserialize(response.Body.ToString()); - }, - RetryOnConnectivityErrors, - cancellationToken); + try + { + // GET bypasses Tilt's publication lock. A retried POST can block behind the original + // create's watcher notification before it can check whether the object already exists. + // https://github.com/tilt-dev/tilt-apiserver/blob/v0.19.1/pkg/storage/filepath/jsonfile_rest.go + return await GetAsync(name, namespaceParameter, recoveryCancellationToken).ConfigureAwait(false); + } + catch (HttpOperationException ex) when (ex.Response.StatusCode == HttpStatusCode.NotFound) + { + // Creation is unconfirmed; retry the same named object. + } + catch (Exception ex) when (RetryOnTransientCreateErrors(ex)) + { + if (GetCreateRetryAfter(ex, recoveryTimeout) is { } confirmationRetryAfter && + (retryAfter is null || confirmationRetryAfter > retryAfter.Value)) + { + retryAfter = confirmationRetryAfter; + } + } + + retryDelay = await DelayCreateRetryAsync(retryAfter, retryDelay, recoveryTimeout, recoveryCancellationToken).ConfigureAwait(false); + } + }, cancellationToken).ConfigureAwait(false); } public Task PatchAsync(T obj, V1Patch patch, CancellationToken cancellationToken = default) @@ -461,7 +500,12 @@ public async Task CleanupResourcesAsync(CancellationToken cancellationToken = de await retryPipeline.ExecuteAsync(async cancellationContext => { - return await _kubernetes!.GetExecutionDocumentAsync(cancellationContext).ConfigureAwait(false); + return await ExecuteWithRetry( + DcpApiOperationType.ResourceCleanup, + DcpExecutionResourceType, + (kubernetes, operationCancellationToken) => kubernetes.GetExecutionDocumentAsync(operationCancellationToken), + RetryOnConnectivityErrors, + cancellationContext).ConfigureAwait(false); }, cancellationToken).ConfigureAwait(false); } @@ -482,6 +526,50 @@ private static string GetResourceFor() where T : CustomResource return kindWithResource.Resource; } + private async Task CreateOnceAsync( + T obj, + string resourceType, + string? namespaceParameter, + CancellationToken cancellationToken) + where T : CustomResource, IKubernetesStaticMetadata + { + var requestStarted = false; + try + { + return await ExecuteWithRetry( + DcpApiOperationType.Create, + T.ObjectKind, + async (kubernetes, operationCancellationToken) => + { + requestStarted = true; + var response = string.IsNullOrEmpty(namespaceParameter) + ? await kubernetes.CustomObjects.CreateClusterCustomObjectWithHttpMessagesAsync( + obj, + GroupVersion.Group, + GroupVersion.Version, + resourceType, + cancellationToken: operationCancellationToken).ConfigureAwait(false) + : await kubernetes.CustomObjects.CreateNamespacedCustomObjectWithHttpMessagesAsync( + obj, + GroupVersion.Group, + GroupVersion.Version, + namespaceParameter, + resourceType, + cancellationToken: operationCancellationToken).ConfigureAwait(false); + + return KubernetesJson.Deserialize(response.Body.ToString()); + }, + // The caller must confirm an ambiguous write before another POST. + _ => false, + cancellationToken).ConfigureAwait(false); + } + catch (Exception ex) when (!requestStarted && RetryOnTransientCreateErrors(ex)) + { + // Client setup can time out before sending a POST; there is no write to confirm in that case. + throw new KubernetesClientInitializationException(ex); + } + } + private async Task ExecuteWithRetry( DcpApiOperationType operationType, string resourceType, @@ -495,43 +583,21 @@ private async Task ExecuteWithRetry( try { - async ValueTask EnsureKubernetesClientAsync(CancellationToken cancellationToken) - { - var clientReady = await EnsureKubernetesAsync(cancellationToken).ConfigureAwait(false); - if (clientReady.Initialized) - { - activity.AddKubernetesClientReady(clientReady.WaitMilliseconds, clientReady.Initialized); - } - } - - if (_kubernetes is null) - { - // The first DCP request must also wait for DCP to create its kubeconfig. Give that initialization - // its own budget so slower hosts do not consume the shorter already-running API retry budget. - var initializationPipeline = CreateKubernetesCallResiliencePipeline( - KubernetesInitializationTimeout, - RetryOnConnectivityErrors, - activity, - () => retryCount++); - await initializationPipeline.ExecuteAsync(EnsureKubernetesClientAsync, cancellationToken).ConfigureAwait(false); - } - - // A parsed kubeconfig does not guarantee that DCP can process requests yet. Keep using the startup - // budget until an API operation succeeds, then fail steady-state API calls on the normal budget. - var retryDuration = Volatile.Read(ref _kubernetesApiReady) - ? MaxRetryDuration - : KubernetesInitializationTimeout; - var resiliencePipeline = CreateKubernetesCallResiliencePipeline( - retryDuration, isRetryable, activity, () => retryCount++, timeoutEachAttempt); return await resiliencePipeline.ExecuteAsync(async (cancellationToken) => { - // Keep connection establishment inside the retry loop so kubeconfig read failures remain retryable. - await EnsureKubernetesClientAsync(cancellationToken).ConfigureAwait(false); + // Client setup and the API operation share one deadline. A parsed kubeconfig alone + // does not confirm that DCP can process requests. + var clientReady = await EnsureKubernetesAsync(cancellationToken).ConfigureAwait(false); + if (clientReady.Initialized) + { + activity.AddKubernetesClientReady(clientReady.WaitMilliseconds, clientReady.Initialized); + } + var result = await operation(_kubernetes!, cancellationToken).ConfigureAwait(false); Volatile.Write(ref _kubernetesApiReady, true); @@ -545,13 +611,84 @@ async ValueTask EnsureKubernetesClientAsync(CancellationToken cancellationToken) } private static bool RetryOnConnectivityErrors(Exception ex) => ex is HttpRequestException || ex is KubeConfigException; + private static bool RetryOnTransientCreateErrors(Exception ex) => + RetryOnConnectivityErrors(ex) || + ex is TimeoutRejectedException || + ex is HttpOperationException { Response.StatusCode: HttpStatusCode.BadGateway or HttpStatusCode.ServiceUnavailable or HttpStatusCode.GatewayTimeout or HttpStatusCode.TooManyRequests }; + + private static bool IsAlreadyExists(HttpOperationException exception) + { + if (exception.Response.StatusCode != HttpStatusCode.Conflict) + { + return false; + } + + // Kubernetes distinguishes {"kind":"Status","reason":"AlreadyExists","code":409} + // from other conflicts, which must retain their normal failure behavior. + try + { + using var status = JsonDocument.Parse(exception.Response.Content); + + return status.RootElement.ValueKind is JsonValueKind.Object && + status.RootElement.TryGetProperty("reason", out var reason) && + reason.ValueKind is JsonValueKind.String && + reason.GetString() is "AlreadyExists"; + } + catch (JsonException) + { + // An unrecognized error body is not proof of an existing object; propagate the HTTP failure. + return false; + } + } + + private static DateTimeOffset? GetCreateRetryAfter(Exception exception, TimeSpan recoveryTimeout) + { + // Retry-After is either a delay in seconds ("5") or an HTTP date + // ("Mon, 05 Oct 2026 18:00:00 GMT"). Keep a deadline so confirmation time counts toward the wait. + // Malformed hints fall back to exponential backoff. + // https://www.rfc-editor.org/rfc/rfc9110.html#section-10.2.3 + if (exception is HttpOperationException operationException && + operationException.Response.Headers.TryGetValue("Retry-After", out var values) && + RetryConditionHeaderValue.TryParse(values.FirstOrDefault(), out var retryAfter)) + { + if (retryAfter.Date is { } date) + { + return date; + } + + // A delay beyond the recovery budget cannot lead to another attempt. Capping it also prevents + // very large delta-seconds values from overflowing DateTimeOffset. + var delay = retryAfter.Delta!.Value; + + return DateTimeOffset.UtcNow + (delay > recoveryTimeout ? recoveryTimeout : delay); + } + + return null; + } + + private static async Task DelayCreateRetryAsync( + DateTimeOffset? retryAfter, + TimeSpan retryDelay, + TimeSpan recoveryTimeout, + CancellationToken cancellationToken) + { + // Full jitter avoids synchronizing concurrent creates; Retry-After is a server-imposed minimum. + // https://aws.amazon.com/blogs/architecture/exponential-backoff-and-jitter/ + var delay = retryAfter is { } deadline + ? TimeSpan.FromTicks(Math.Max((deadline - DateTimeOffset.UtcNow).Ticks, 0)) + : TimeSpan.FromMilliseconds(Random.Shared.NextDouble() * retryDelay.TotalMilliseconds); + // A longer delay cannot lead to another attempt, and may exceed Task.Delay's supported range. + await Task.Delay(delay > recoveryTimeout ? recoveryTimeout : delay, cancellationToken).ConfigureAwait(false); + + return TimeSpan.FromMilliseconds(Math.Min(retryDelay.TotalMilliseconds * 2, 5000)); + } + private static bool RetryOnConnectivityAndConflictErrors(Exception ex) => ex is HttpRequestException || ex is KubeConfigException || (ex is HttpOperationException hoe && hoe.Response.StatusCode == System.Net.HttpStatusCode.Conflict); - private static ResiliencePipeline CreateKubernetesCallResiliencePipeline( - TimeSpan retryDuration, + private ResiliencePipeline CreateKubernetesCallResiliencePipeline( Func isRetryable, ProfilingTelemetry.ActivityScope activity, Action recordRetry, @@ -559,7 +696,17 @@ private static ResiliencePipeline CreateKubernetesCallResiliencePipeline( { var timeoutOptions = new TimeoutStrategyOptions { - Timeout = retryDuration, + // Select once per operation or watch connection attempt. Confirming readiness elsewhere + // must not shorten an in-flight deadline, but the next connection attempt uses the new budget. + TimeoutGenerator = _ => + { + var options = dcpOptions.Value; + var timeout = Volatile.Read(ref _kubernetesApiReady) + ? options.KubernetesApiTimeout + : options.KubernetesApiTimeout + options.KubernetesInitializationAdditionalTimeout; + + return new ValueTask(timeout); + }, OnTimeout = (_) => { activity.AddKubernetesApiTimeout(); @@ -708,4 +855,7 @@ private async Task EnsureKubernetesAsync(CancellationToke } private readonly record struct KubernetesClientReady(long WaitMilliseconds, bool Initialized); + + private sealed class KubernetesClientInitializationException(Exception innerException) + : Exception("The DCP Kubernetes client could not be initialized before sending the create request.", innerException); } diff --git a/tests/Aspire.Hosting.Tests/Dcp/ConfigureDefaultDcpOptionsTests.cs b/tests/Aspire.Hosting.Tests/Dcp/ConfigureDefaultDcpOptionsTests.cs index 989cb44475f..9b387f3342b 100644 --- a/tests/Aspire.Hosting.Tests/Dcp/ConfigureDefaultDcpOptionsTests.cs +++ b/tests/Aspire.Hosting.Tests/Dcp/ConfigureDefaultDcpOptionsTests.cs @@ -8,6 +8,214 @@ namespace Aspire.Hosting.Tests.Dcp; public class ConfigureDefaultDcpOptionsTests { + [Fact] + public void KubernetesApiTimeoutDefaultsToFortySecondsWithTwentyAdditionalInitializationSeconds() + { + var options = ConfigureWithDcpPublisher([]); + + Assert.Equal(TimeSpan.FromSeconds(20), options.KubernetesInitializationAdditionalTimeout); + Assert.Equal(TimeSpan.FromSeconds(40), options.KubernetesApiTimeout); + Assert.Equal(TimeSpan.FromSeconds(60), options.KubernetesApiTimeout + options.KubernetesInitializationAdditionalTimeout); + } + + [Fact] + public void KubernetesApiTimeoutsCanBeConfigured() + { + var options = ConfigureWithDcpPublisher(new() + { + ["DcpPublisher:KubernetesInitializationAdditionalTimeout"] = "00:00:30", + ["DcpPublisher:KubernetesApiTimeout"] = "00:00:50", + }); + + Assert.Equal(TimeSpan.FromSeconds(30), options.KubernetesInitializationAdditionalTimeout); + Assert.Equal(TimeSpan.FromSeconds(50), options.KubernetesApiTimeout); + Assert.Equal(TimeSpan.FromSeconds(80), options.KubernetesApiTimeout + options.KubernetesInitializationAdditionalTimeout); + } + + [Fact] + public void KubernetesApiTimeoutOverrideAlsoIncreasesInitializationBudget() + { + var options = ConfigureWithDcpPublisher(new() + { + ["DcpPublisher:KubernetesApiTimeout"] = "00:00:50", + }); + + Assert.Equal(TimeSpan.FromSeconds(20), options.KubernetesInitializationAdditionalTimeout); + Assert.Equal(TimeSpan.FromSeconds(70), options.KubernetesApiTimeout + options.KubernetesInitializationAdditionalTimeout); + } + + [Theory] + [InlineData("00:00:00")] + [InlineData("-00:00:01")] + [InlineData("00:00:00.999")] + [InlineData("00:10:00.001")] + [InlineData("00:00:00.010")] + [InlineData("1.00:00:00")] + public void KubernetesApiTimeoutMustBeWithinSupportedRange(string configuredTimeout) + { + var options = ConfigureWithDcpPublisher(new() + { + ["DcpPublisher:KubernetesApiTimeout"] = configuredTimeout, + }); + var validator = new ValidateDcpOptions(new DistributedApplicationExecutionContext(DistributedApplicationOperation.Publish)); + + var result = validator.Validate(null, options); + + Assert.True(result.Failed); + Assert.Equal("Property KubernetesApiTimeout: The Kubernetes API timeout must be between one second and ten minutes.", Assert.Single(result.Failures!)); + } + + [Theory] + [InlineData("00:00:01", 1000)] + [InlineData("00:10:00", 600000)] + public void KubernetesApiTimeoutAcceptsSupportedBoundaries(string configuredTimeout, int expectedMilliseconds) + { + var options = ConfigureWithDcpPublisher(new() + { + ["DcpPublisher:KubernetesInitializationAdditionalTimeout"] = "00:00:00", + ["DcpPublisher:KubernetesApiTimeout"] = configuredTimeout, + }); + var validator = new ValidateDcpOptions(new DistributedApplicationExecutionContext(DistributedApplicationOperation.Publish)); + + var result = validator.Validate(null, options); + + Assert.True(result.Succeeded); + Assert.Equal(TimeSpan.Zero, options.KubernetesInitializationAdditionalTimeout); + Assert.Equal(TimeSpan.FromMilliseconds(expectedMilliseconds), options.KubernetesApiTimeout); + } + + [Theory] + [InlineData("00:00:40", "00:00:00", 40000)] + [InlineData("00:00:40", "00:09:20", 600000)] + [InlineData("00:00:01", "00:09:59", 600000)] + public void KubernetesInitializationAdditionalTimeoutAcceptsSupportedBoundaries(string apiTimeout, string additionalTimeout, int expectedMilliseconds) + { + var options = ConfigureWithDcpPublisher(new() + { + ["DcpPublisher:KubernetesApiTimeout"] = apiTimeout, + ["DcpPublisher:KubernetesInitializationAdditionalTimeout"] = additionalTimeout, + }); + var validator = new ValidateDcpOptions(new DistributedApplicationExecutionContext(DistributedApplicationOperation.Publish)); + + var result = validator.Validate(null, options); + + Assert.True(result.Succeeded); + Assert.Equal(TimeSpan.FromMilliseconds(expectedMilliseconds), options.KubernetesApiTimeout + options.KubernetesInitializationAdditionalTimeout); + } + + [Theory] + [InlineData("-00:00:00.001")] + [InlineData("-00:00:01")] + public void KubernetesInitializationAdditionalTimeoutMustBeNonNegative(string configuredTimeout) + { + var options = ConfigureWithDcpPublisher(new() + { + ["DcpPublisher:KubernetesInitializationAdditionalTimeout"] = configuredTimeout, + }); + var validator = new ValidateDcpOptions(new DistributedApplicationExecutionContext(DistributedApplicationOperation.Publish)); + + var result = validator.Validate(null, options); + + Assert.True(result.Failed); + Assert.Equal("Property KubernetesInitializationAdditionalTimeout: The Kubernetes additional initialization timeout must be non-negative.", Assert.Single(result.Failures!)); + } + + [Theory] + [InlineData("00:00:40", "00:09:20.001")] + [InlineData("00:10:00", "00:00:00.001")] + [InlineData("00:00:01", "00:10:00")] + public void KubernetesInitializationBudgetMustNotExceedTenMinutes(string apiTimeout, string additionalTimeout) + { + var options = ConfigureWithDcpPublisher(new() + { + ["DcpPublisher:KubernetesApiTimeout"] = apiTimeout, + ["DcpPublisher:KubernetesInitializationAdditionalTimeout"] = additionalTimeout, + }); + var validator = new ValidateDcpOptions(new DistributedApplicationExecutionContext(DistributedApplicationOperation.Publish)); + + var result = validator.Validate(null, options); + + Assert.True(result.Failed); + Assert.Equal("Property KubernetesInitializationAdditionalTimeout: The combined Kubernetes API and additional initialization timeouts must not exceed ten minutes.", Assert.Single(result.Failures!)); + } + + [Theory] + [InlineData(false)] + [InlineData(true)] + public void KubernetesInitializationBudgetValidationDoesNotOverflow(bool apiTimeoutAlsoOverflows) + { + var options = ConfigureWithDcpPublisher([]); + options.KubernetesInitializationAdditionalTimeout = TimeSpan.MaxValue; + if (apiTimeoutAlsoOverflows) + { + options.KubernetesApiTimeout = TimeSpan.MaxValue; + } + var validator = new ValidateDcpOptions(new DistributedApplicationExecutionContext(DistributedApplicationOperation.Publish)); + + var result = validator.Validate(null, options); + + Assert.True(result.Failed); + Assert.Equal(apiTimeoutAlsoOverflows + ? "Property KubernetesApiTimeout: The Kubernetes API timeout must be between one second and ten minutes." + : "Property KubernetesInitializationAdditionalTimeout: The combined Kubernetes API and additional initialization timeouts must not exceed ten minutes.", + Assert.Single(result.Failures!)); + } + + [Fact] + public void KubernetesCreateRecoveryTimeoutDefaultsToTwoMinutes() + { + Assert.Equal(TimeSpan.FromMinutes(2), ConfigureWithDcpPublisher([]).KubernetesCreateRecoveryTimeout); + } + + [Fact] + public void KubernetesCreateRecoveryTimeoutCanBeConfigured() + { + var options = ConfigureWithDcpPublisher(new() + { + ["DcpPublisher:KubernetesCreateRecoveryTimeout"] = "00:03:00", + }); + + Assert.Equal(TimeSpan.FromMinutes(3), options.KubernetesCreateRecoveryTimeout); + } + + [Theory] + [InlineData("00:00:00")] + [InlineData("-00:00:01")] + [InlineData("00:00:00.999")] + [InlineData("00:10:00.001")] + [InlineData("00:00:00.010")] + [InlineData("1.00:00:00")] + public void KubernetesCreateRecoveryTimeoutMustBeWithinSupportedRange(string configuredTimeout) + { + var options = ConfigureWithDcpPublisher(new() + { + ["DcpPublisher:KubernetesCreateRecoveryTimeout"] = configuredTimeout, + }); + var validator = new ValidateDcpOptions(new DistributedApplicationExecutionContext(DistributedApplicationOperation.Publish)); + + var result = validator.Validate(null, options); + + Assert.True(result.Failed); + Assert.Equal("Property KubernetesCreateRecoveryTimeout: The Kubernetes create recovery timeout must be between one second and ten minutes.", Assert.Single(result.Failures!)); + } + + [Theory] + [InlineData("00:00:01", 1000)] + [InlineData("00:10:00", 600000)] + public void KubernetesCreateRecoveryTimeoutAcceptsSupportedBoundaries(string configuredTimeout, int expectedMilliseconds) + { + var options = ConfigureWithDcpPublisher(new() + { + ["DcpPublisher:KubernetesCreateRecoveryTimeout"] = configuredTimeout, + }); + var validator = new ValidateDcpOptions(new DistributedApplicationExecutionContext(DistributedApplicationOperation.Publish)); + + var result = validator.Validate(null, options); + + Assert.True(result.Succeeded); + Assert.Equal(TimeSpan.FromMilliseconds(expectedMilliseconds), options.KubernetesCreateRecoveryTimeout); + } + [Fact] public void TerminalHostFallsBackToAspireManagedDashboardPath() { diff --git a/tests/Aspire.Hosting.Tests/Dcp/KubernetesServiceTests.cs b/tests/Aspire.Hosting.Tests/Dcp/KubernetesServiceTests.cs index d44c683b58b..8f38063077f 100644 --- a/tests/Aspire.Hosting.Tests/Dcp/KubernetesServiceTests.cs +++ b/tests/Aspire.Hosting.Tests/Dcp/KubernetesServiceTests.cs @@ -30,6 +30,841 @@ namespace Aspire.Hosting.Tests.Dcp; public class KubernetesServiceTests(DcpApiServerCertificateFixture certificateFixture) : IClassFixture { + [Fact] + public async Task CreateAsync_DoesNotReadOrRetrySuccessfulCreate() + { + using var cts = new CancellationTokenSource(TimeSpan.FromSeconds(20)); + var (service, kubeconfigPath, fileSystem) = CreateService(); + using var disposableFileSystem = fileSystem; + using var disposableService = service; + var requests = new ConcurrentQueue(); + await using var server = await StartServerAsync(cts.Token, requestHandler: async context => + { + requests.Enqueue(context.Request.Method); + var created = await ReadContainerAsync(context); + created.Metadata.Uid = "created-object"; + await WriteContainerAsync(context, created, StatusCodes.Status201Created); + }); + WriteKubeconfig(kubeconfigPath, server.Port); + + var result = await service.CreateAsync(Container.Create("frontend", "test-image"), cts.Token); + + Assert.Equal("created-object", result.Metadata.Uid); + Assert.Equal(["POST"], requests.ToArray()); + } + + [Fact] + public async Task DcpKubernetesClient_UsesCallerDeadlineInsteadOfSdkTimeout() + { + using var cts = new CancellationTokenSource(TimeSpan.FromSeconds(20)); + var requests = new ConcurrentQueue(); + await using var server = await StartServerAsync(cts.Token, requestHandler: async context => + { + requests.Enqueue(context.Request.Method); + var created = await ReadContainerAsync(context); + created.Metadata.Uid = "created-object"; + await Task.Delay(TimeSpan.FromMilliseconds(500), context.RequestAborted); + await WriteContainerAsync(context, created, StatusCodes.Status201Created); + }); + using var client = new DcpKubernetesClient(new KubernetesClientConfiguration + { + Host = $"https://127.0.0.1:{server.Port}", + SkipTlsVerify = true, + HttpClientTimeout = TimeSpan.FromMilliseconds(100), + }); + + using var response = await client.CustomObjects.CreateClusterCustomObjectWithHttpMessagesAsync( + Container.Create("frontend", "test-image"), + "usvc-dev.developer.microsoft.com", + "v1", + "containers", + cancellationToken: cts.Token); + var result = KubernetesJson.Deserialize(response.Body.ToString()); + + Assert.Equal("created-object", result.Metadata.Uid); + Assert.Equal(["POST"], requests.ToArray()); + } + + [Theory] + [InlineData(null)] + [InlineData("test-namespace")] + public async Task CreateAsync_ReconcilesGatewayTimeoutAfterCommit(string? namespaceParameter) + { + using var cts = new CancellationTokenSource(TimeSpan.FromSeconds(20)); + var (service, kubeconfigPath, fileSystem) = CreateService(); + using var disposableFileSystem = fileSystem; + using var disposableService = service; + var requests = new ConcurrentQueue(); + Container? stored = null; + await using var server = await StartServerAsync(cts.Token, requestHandler: async context => + { + requests.Enqueue($"{context.Request.Method} {context.Request.Path}"); + if (HttpMethods.IsPost(context.Request.Method)) + { + var created = await ReadContainerAsync(context); + created.Metadata.Uid = "created-object"; + Volatile.Write(ref stored, created); + await WriteStatusAsync(context, StatusCodes.Status504GatewayTimeout, "Timeout"); + return; + } + + await WriteContainerAsync(context, Volatile.Read(ref stored)!); + }); + WriteKubeconfig(kubeconfigPath, server.Port); + var container = Container.Create("frontend", "test-image"); + container.Metadata.NamespaceProperty = namespaceParameter; + + var result = await service.CreateAsync(container, cts.Token); + + Assert.Equal("created-object", result.Metadata.Uid); + Assert.Equal("frontend", result.Metadata.Name); + Assert.Equal(namespaceParameter, result.Metadata.NamespaceProperty); + var collectionPath = namespaceParameter is null + ? "/apis/usvc-dev.developer.microsoft.com/v1/containers" + : $"/apis/usvc-dev.developer.microsoft.com/v1/namespaces/{namespaceParameter}/containers"; + Assert.Equal([$"POST {collectionPath}", $"GET {collectionPath}/frontend"], requests.ToArray()); + } + + [Theory] + [InlineData(429)] + [InlineData(502)] + [InlineData(503)] + [InlineData(504)] + public async Task CreateAsync_RetriesTransientFailureAfterConfirmingObjectAbsent(int statusCode) + { + using var cts = new CancellationTokenSource(TimeSpan.FromSeconds(20)); + var (service, kubeconfigPath, fileSystem) = CreateService(); + using var disposableFileSystem = fileSystem; + using var disposableService = service; + var requests = new ConcurrentQueue(); + var postedObjects = new ConcurrentQueue(); + var postCount = 0; + await using var server = await StartServerAsync(cts.Token, requestHandler: async context => + { + requests.Enqueue(context.Request.Method); + if (HttpMethods.IsGet(context.Request.Method)) + { + await WriteStatusAsync(context, StatusCodes.Status404NotFound, "NotFound"); + return; + } + + var created = await ReadContainerAsync(context); + postedObjects.Enqueue(created); + if (Interlocked.Increment(ref postCount) == 1) + { + await WriteStatusAsync(context, statusCode, "TransientFailure"); + return; + } + + created.Metadata.Uid = "retried-object"; + await WriteContainerAsync(context, created, StatusCodes.Status201Created); + }); + WriteKubeconfig(kubeconfigPath, server.Port); + var container = Container.Create("frontend", "test-image"); + container.Annotate("existing-annotation", "preserved"); + + var result = await service.CreateAsync(container, cts.Token); + + Assert.Equal("retried-object", result.Metadata.Uid); + Assert.Equal(["POST", "GET", "POST"], requests.ToArray()); + Assert.Collection(postedObjects, + first => + { + Assert.Equal("frontend", first.Metadata.Name); + Assert.Equal("test-image", first.Spec.Image); + Assert.Equal(container.Metadata.Annotations, first.Metadata.Annotations); + }, + second => + { + Assert.Equal("frontend", second.Metadata.Name); + Assert.Equal("test-image", second.Spec.Image); + Assert.Equal(container.Metadata.Annotations, second.Metadata.Annotations); + }); + } + + [Fact] + public async Task CreateAsync_ReconcilesAlreadyExistsWhenOriginalCreateCommitsAfterNotFound() + { + using var cts = new CancellationTokenSource(TimeSpan.FromSeconds(20)); + var (service, kubeconfigPath, fileSystem) = CreateService(); + using var disposableFileSystem = fileSystem; + using var disposableService = service; + var requests = new ConcurrentQueue(); + Container? original = null; + var postCount = 0; + await using var server = await StartServerAsync(cts.Token, requestHandler: async context => + { + requests.Enqueue(context.Request.Method); + if (HttpMethods.IsPost(context.Request.Method)) + { + var created = await ReadContainerAsync(context); + if (Interlocked.Increment(ref postCount) == 1) + { + created.Metadata.Uid = "late-original"; + Volatile.Write(ref original, created); + await WriteStatusAsync(context, StatusCodes.Status504GatewayTimeout, "Timeout"); + } + else + { + Assert.Equal(Volatile.Read(ref original)!.Metadata.Name, created.Metadata.Name); + Assert.Equal(Volatile.Read(ref original)!.Spec.Image, created.Spec.Image); + await WriteStatusAsync(context, StatusCodes.Status409Conflict, "AlreadyExists"); + } + + return; + } + + if (Volatile.Read(ref postCount) == 1) + { + await WriteStatusAsync(context, StatusCodes.Status404NotFound, "NotFound"); + return; + } + + await WriteContainerAsync(context, Volatile.Read(ref original)!); + }); + WriteKubeconfig(kubeconfigPath, server.Port); + + var result = await service.CreateAsync(Container.Create("frontend", "test-image"), cts.Token); + + Assert.Equal("late-original", result.Metadata.Uid); + Assert.Equal(["POST", "GET", "POST", "GET"], requests.ToArray()); + } + + [Theory] + [InlineData(429)] + [InlineData(502)] + [InlineData(503)] + [InlineData(504)] + public async Task CreateAsync_RetriesCreateAfterTransientConfirmationFailure(int confirmationStatusCode) + { + using var cts = new CancellationTokenSource(TimeSpan.FromSeconds(20)); + var (service, kubeconfigPath, fileSystem) = CreateService(); + using var disposableFileSystem = fileSystem; + using var disposableService = service; + var requests = new ConcurrentQueue(); + Container? stored = null; + var postCount = 0; + var getCount = 0; + await using var server = await StartServerAsync(cts.Token, requestHandler: async context => + { + requests.Enqueue(context.Request.Method); + if (HttpMethods.IsPost(context.Request.Method)) + { + var created = await ReadContainerAsync(context); + if (Interlocked.Increment(ref postCount) == 1) + { + created.Metadata.Uid = "original-object"; + Volatile.Write(ref stored, created); + await WriteStatusAsync(context, StatusCodes.Status504GatewayTimeout, "Timeout"); + } + else + { + await WriteStatusAsync(context, StatusCodes.Status409Conflict, "AlreadyExists"); + } + } + else if (Interlocked.Increment(ref getCount) == 1) + { + await WriteStatusAsync(context, confirmationStatusCode, "TransientFailure"); + } + else + { + await WriteContainerAsync(context, Volatile.Read(ref stored)!); + } + }); + WriteKubeconfig(kubeconfigPath, server.Port); + + var result = await service.CreateAsync(Container.Create("frontend", "test-image"), cts.Token); + + Assert.Equal("original-object", result.Metadata.Uid); + Assert.Equal(["POST", "GET", "POST", "GET"], requests.ToArray()); + } + + [Theory] + [InlineData("""{"kind":"Status","reason":"Conflict","code":409}""")] + [InlineData("""{"kind":"Status","reason":409,"code":409}""")] + [InlineData("[]")] + [InlineData("not-json")] + public async Task CreateAsync_DoesNotConfirmOtherConflictsAfterAmbiguousCreate(string errorBody) + { + using var cts = new CancellationTokenSource(TimeSpan.FromSeconds(20)); + var (service, kubeconfigPath, fileSystem) = CreateService(); + using var disposableFileSystem = fileSystem; + using var disposableService = service; + var requests = new ConcurrentQueue(); + var postCount = 0; + await using var server = await StartServerAsync(cts.Token, requestHandler: async context => + { + requests.Enqueue(context.Request.Method); + if (HttpMethods.IsGet(context.Request.Method)) + { + await WriteStatusAsync(context, StatusCodes.Status404NotFound, "NotFound"); + } + else if (Interlocked.Increment(ref postCount) == 1) + { + await WriteStatusAsync(context, StatusCodes.Status504GatewayTimeout, "Timeout"); + } + else + { + context.Response.StatusCode = StatusCodes.Status409Conflict; + context.Response.ContentType = "application/json"; + await context.Response.WriteAsync(errorBody, context.RequestAborted); + } + }); + WriteKubeconfig(kubeconfigPath, server.Port); + + var exception = await Assert.ThrowsAsync( + () => service.CreateAsync(Container.Create("frontend", "test-image"), cts.Token)); + + Assert.Equal(HttpStatusCode.Conflict, exception.Response.StatusCode); + Assert.Equal(["POST", "GET", "POST"], requests.ToArray()); + } + + [Theory] + [InlineData(400)] + [InlineData(401)] + [InlineData(403)] + [InlineData(404)] + [InlineData(409)] + [InlineData(422)] + [InlineData(500)] + public async Task CreateAsync_DoesNotRetryPermanentFailureOrInitialConflict(int statusCode) + { + using var cts = new CancellationTokenSource(TimeSpan.FromSeconds(20)); + var (service, kubeconfigPath, fileSystem) = CreateService(); + using var disposableFileSystem = fileSystem; + using var disposableService = service; + var requests = new ConcurrentQueue(); + await using var server = await StartServerAsync(cts.Token, requestHandler: async context => + { + requests.Enqueue(context.Request.Method); + await WriteStatusAsync(context, statusCode, statusCode == 409 ? "AlreadyExists" : "PermanentFailure"); + }); + WriteKubeconfig(kubeconfigPath, server.Port); + + var exception = await Assert.ThrowsAsync( + () => service.CreateAsync(Container.Create("frontend", "test-image"), cts.Token)); + + Assert.Equal((HttpStatusCode)statusCode, exception.Response.StatusCode); + Assert.Equal(["POST"], requests.ToArray()); + } + + [Theory] + [InlineData(false)] + [InlineData(true)] + public async Task CreateAsync_ReconcilesExistingObjectByName(bool hasAnnotations) + { + using var cts = new CancellationTokenSource(TimeSpan.FromSeconds(20)); + var (service, kubeconfigPath, fileSystem) = CreateService(); + using var disposableFileSystem = fileSystem; + using var disposableService = service; + var requests = new ConcurrentQueue(); + await using var server = await StartServerAsync(cts.Token, requestHandler: async context => + { + requests.Enqueue(context.Request.Method); + if (HttpMethods.IsPost(context.Request.Method)) + { + await WriteStatusAsync(context, StatusCodes.Status504GatewayTimeout, "Timeout"); + return; + } + + var existing = Container.Create("frontend", "test-image"); + existing.Metadata.Uid = "existing-object"; + if (hasAnnotations) + { + existing.Annotate("existing-annotation", "preserved"); + } + + await WriteContainerAsync(context, existing); + }); + WriteKubeconfig(kubeconfigPath, server.Port); + + var result = await service.CreateAsync(Container.Create("frontend", "test-image"), cts.Token); + + Assert.Equal("existing-object", result.Metadata.Uid); + if (hasAnnotations) + { + Assert.Equal("preserved", result.Metadata.Annotations["existing-annotation"]); + } + + Assert.Equal(["POST", "GET"], requests.ToArray()); + } + + [Theory] + [InlineData(false)] + [InlineData(true)] + public async Task CreateAsync_RecoversAfterAttemptTimeout(bool committedBeforeTimeout) + { + using var cts = new CancellationTokenSource(TimeSpan.FromSeconds(20)); + var (service, kubeconfigPath, fileSystem) = CreateService( + kubernetesApiTimeout: TimeSpan.FromMilliseconds(500), + kubernetesInitializationAdditionalTimeout: TimeSpan.FromSeconds(4.5), + createRecoveryTimeout: TimeSpan.FromSeconds(10)); + using var disposableFileSystem = fileSystem; + using var disposableService = service; + var requests = new ConcurrentQueue(); + var canceledRequest = new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously); + Container? stored = null; + var postCount = 0; + await using var server = await StartServerAsync(cts.Token, requestHandler: async context => + { + if (context.Request.Path.Value!.EndsWith("/containers", StringComparison.Ordinal) && + HttpMethods.IsGet(context.Request.Method)) + { + context.Response.ContentType = "application/json"; + await context.Response.WriteAsync("""{"apiVersion":"usvc-dev.developer.microsoft.com/v1","kind":"ContainerList","items":[]}"""); + return; + } + + requests.Enqueue(context.Request.Method); + if (HttpMethods.IsGet(context.Request.Method)) + { + if (Volatile.Read(ref stored) is { } existing) + { + await WriteContainerAsync(context, existing); + } + else + { + await WriteStatusAsync(context, StatusCodes.Status404NotFound, "NotFound"); + } + + return; + } + + var created = await ReadContainerAsync(context); + if (Interlocked.Increment(ref postCount) == 1) + { + if (committedBeforeTimeout) + { + Volatile.Write(ref stored, created); + } + + try + { + await Task.Delay(Timeout.InfiniteTimeSpan, context.RequestAborted); + } + catch (OperationCanceledException) when (context.RequestAborted.IsCancellationRequested) + { + canceledRequest.TrySetResult(); + throw; + } + } + else + { + await WriteContainerAsync(context, created, StatusCodes.Status201Created); + } + }); + WriteKubeconfig(kubeconfigPath, server.Port); + Assert.Empty(await service.ListAsync(cancellationToken: cts.Token)); + + var result = await service.CreateAsync(Container.Create("frontend", "test-image"), cts.Token); + + await canceledRequest.Task.WaitAsync(cts.Token); + Assert.Equal("frontend", result.Metadata.Name); + Assert.Equal(committedBeforeTimeout ? ["POST", "GET"] : ["POST", "GET", "POST"], requests.ToArray()); + } + + [Fact] + public async Task CreateAsync_ReconcilesLostResponseAfterCommit() + { + using var cts = new CancellationTokenSource(TimeSpan.FromSeconds(20)); + var (service, kubeconfigPath, fileSystem) = CreateService(); + using var disposableFileSystem = fileSystem; + using var disposableService = service; + var requests = new ConcurrentQueue(); + Container? stored = null; + await using var server = await StartServerAsync(cts.Token, requestHandler: async context => + { + requests.Enqueue(context.Request.Method); + if (HttpMethods.IsPost(context.Request.Method)) + { + Volatile.Write(ref stored, await ReadContainerAsync(context)); + context.Abort(); + return; + } + + await WriteContainerAsync(context, Volatile.Read(ref stored)!); + }); + WriteKubeconfig(kubeconfigPath, server.Port); + + var result = await service.CreateAsync(Container.Create("frontend", "test-image"), cts.Token); + + Assert.Equal("frontend", result.Metadata.Name); + Assert.Equal(["POST", "GET"], requests.ToArray()); + } + + [Fact] + public async Task CreateAsync_RetriesCreateAfterConfirmationTimeout() + { + using var cts = new CancellationTokenSource(TimeSpan.FromSeconds(20)); + var (service, kubeconfigPath, fileSystem) = CreateService( + kubernetesApiTimeout: TimeSpan.FromMilliseconds(500), + createRecoveryTimeout: TimeSpan.FromSeconds(10)); + using var disposableFileSystem = fileSystem; + using var disposableService = service; + var requests = new ConcurrentQueue(); + var requestCanceled = new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously); + Container? stored = null; + var postCount = 0; + var getCount = 0; + await using var server = await StartServerAsync(cts.Token, requestHandler: async context => + { + if (HttpMethods.IsGet(context.Request.Method) && + context.Request.Path.Value!.EndsWith("/containers", StringComparison.Ordinal)) + { + context.Response.ContentType = "application/json"; + await context.Response.WriteAsync("""{"apiVersion":"usvc-dev.developer.microsoft.com/v1","kind":"ContainerList","items":[]}"""); + return; + } + + requests.Enqueue(context.Request.Method); + if (HttpMethods.IsPost(context.Request.Method)) + { + var created = await ReadContainerAsync(context); + if (Interlocked.Increment(ref postCount) == 1) + { + created.Metadata.Uid = "original-object"; + Volatile.Write(ref stored, created); + await WriteStatusAsync(context, StatusCodes.Status504GatewayTimeout, "Timeout"); + } + else + { + await WriteStatusAsync(context, StatusCodes.Status409Conflict, "AlreadyExists"); + } + + return; + } + + if (Interlocked.Increment(ref getCount) == 1) + { + try + { + await Task.Delay(Timeout.InfiniteTimeSpan, context.RequestAborted); + } + catch (OperationCanceledException) when (context.RequestAborted.IsCancellationRequested) + { + requestCanceled.TrySetResult(); + throw; + } + + return; + } + + await WriteContainerAsync(context, Volatile.Read(ref stored)!); + }); + WriteKubeconfig(kubeconfigPath, server.Port); + Assert.Empty(await service.ListAsync(cancellationToken: cts.Token)); + + var result = await service.CreateAsync(Container.Create("frontend", "test-image"), cts.Token); + + await requestCanceled.Task.WaitAsync(cts.Token); + Assert.Equal("original-object", result.Metadata.Uid); + Assert.Equal(["POST", "GET", "POST", "GET"], requests.ToArray()); + } + + [Fact] + public async Task CreateAsync_RetriesWhenAlreadyExistsCannotBeConfirmed() + { + using var cts = new CancellationTokenSource(TimeSpan.FromSeconds(20)); + var (service, kubeconfigPath, fileSystem) = CreateService(); + using var disposableFileSystem = fileSystem; + using var disposableService = service; + var requests = new ConcurrentQueue(); + var postCount = 0; + await using var server = await StartServerAsync(cts.Token, requestHandler: async context => + { + requests.Enqueue(context.Request.Method); + if (HttpMethods.IsGet(context.Request.Method)) + { + await WriteStatusAsync(context, StatusCodes.Status404NotFound, "NotFound"); + return; + } + + var created = await ReadContainerAsync(context); + switch (Interlocked.Increment(ref postCount)) + { + case 1: + await WriteStatusAsync(context, StatusCodes.Status504GatewayTimeout, "Timeout"); + break; + case 2: + await WriteStatusAsync(context, StatusCodes.Status409Conflict, "AlreadyExists"); + break; + default: + created.Metadata.Uid = "retried-object"; + await WriteContainerAsync(context, created, StatusCodes.Status201Created); + break; + } + }); + WriteKubeconfig(kubeconfigPath, server.Port); + + var result = await service.CreateAsync(Container.Create("frontend", "test-image"), cts.Token); + + Assert.Equal("retried-object", result.Metadata.Uid); + Assert.Equal(["POST", "GET", "POST", "GET", "POST"], requests.ToArray()); + } + + [Theory] + [InlineData(false, 3600)] + [InlineData(false, int.MaxValue)] + [InlineData(true, 0)] + public async Task CreateAsync_RecoveryBudgetIncludesServerRequestedRetryDelay(bool useHttpDate, int retryAfterSeconds) + { + using var cts = new CancellationTokenSource(TimeSpan.FromSeconds(20)); + var recoveryTimeout = TimeSpan.FromMilliseconds(500); + var (service, kubeconfigPath, fileSystem) = CreateService(createRecoveryTimeout: recoveryTimeout); + using var disposableFileSystem = fileSystem; + using var disposableService = service; + var requests = new ConcurrentQueue(); + await using var server = await StartServerAsync(cts.Token, requestHandler: async context => + { + requests.Enqueue(context.Request.Method); + if (HttpMethods.IsGet(context.Request.Method)) + { + await WriteStatusAsync(context, StatusCodes.Status404NotFound, "NotFound"); + return; + } + + context.Response.Headers.RetryAfter = useHttpDate + ? DateTimeOffset.UtcNow.AddHours(1).ToString("r", CultureInfo.InvariantCulture) + : retryAfterSeconds.ToString(CultureInfo.InvariantCulture); + await WriteStatusAsync(context, StatusCodes.Status504GatewayTimeout, "Timeout"); + }); + WriteKubeconfig(kubeconfigPath, server.Port); + + var exception = await Assert.ThrowsAsync( + () => service.CreateAsync(Container.Create("frontend", "test-image"), cts.Token)); + + Assert.Equal(recoveryTimeout, exception.Timeout); + Assert.Equal(["POST", "GET"], requests.ToArray()); + } + + [Theory] + [InlineData(false, 3600)] + [InlineData(false, int.MaxValue)] + [InlineData(true, 0)] + public async Task CreateAsync_RecoveryBudgetIncludesConfirmationRetryDelay(bool useHttpDate, int retryAfterSeconds) + { + using var cts = new CancellationTokenSource(TimeSpan.FromSeconds(20)); + var recoveryTimeout = TimeSpan.FromMilliseconds(500); + var (service, kubeconfigPath, fileSystem) = CreateService(createRecoveryTimeout: recoveryTimeout); + using var disposableFileSystem = fileSystem; + using var disposableService = service; + var requests = new ConcurrentQueue(); + await using var server = await StartServerAsync(cts.Token, requestHandler: async context => + { + requests.Enqueue(context.Request.Method); + if (HttpMethods.IsPost(context.Request.Method)) + { + await WriteStatusAsync(context, StatusCodes.Status504GatewayTimeout, "Timeout"); + return; + } + + context.Response.Headers.RetryAfter = useHttpDate + ? DateTimeOffset.UtcNow.AddHours(1).ToString("r", CultureInfo.InvariantCulture) + : retryAfterSeconds.ToString(CultureInfo.InvariantCulture); + await WriteStatusAsync(context, StatusCodes.Status429TooManyRequests, "TooManyRequests"); + }); + WriteKubeconfig(kubeconfigPath, server.Port); + + var exception = await Assert.ThrowsAsync( + () => service.CreateAsync(Container.Create("frontend", "test-image"), cts.Token)); + + Assert.Equal(recoveryTimeout, exception.Timeout); + Assert.Equal(["POST", "GET"], requests.ToArray()); + } + + [Theory] + [InlineData(false)] + [InlineData(true)] + public async Task CreateAsync_ConfirmationTimeCountsTowardRetryAfter(bool useHttpDate) + { + using var cts = new CancellationTokenSource(TimeSpan.FromSeconds(20)); + var (service, kubeconfigPath, fileSystem) = CreateService(createRecoveryTimeout: TimeSpan.FromSeconds(4.5)); + using var disposableFileSystem = fileSystem; + using var disposableService = service; + var requests = new ConcurrentQueue(); + var postCount = 0; + await using var server = await StartServerAsync(cts.Token, requestHandler: async context => + { + if (HttpMethods.IsGet(context.Request.Method) && + context.Request.Path.Value!.EndsWith("/containers", StringComparison.Ordinal)) + { + context.Response.ContentType = "application/json"; + await context.Response.WriteAsync("""{"apiVersion":"usvc-dev.developer.microsoft.com/v1","kind":"ContainerList","items":[]}"""); + return; + } + + requests.Enqueue(context.Request.Method); + if (HttpMethods.IsGet(context.Request.Method)) + { + // Both retry hints expire before confirmation finishes. Restarting the original delay + // after this read would exhaust the recovery budget before the second POST. + await Task.Delay(TimeSpan.FromSeconds(3.2), context.RequestAborted); + await WriteStatusAsync(context, StatusCodes.Status404NotFound, "NotFound"); + return; + } + + var created = await ReadContainerAsync(context); + if (Interlocked.Increment(ref postCount) == 1) + { + context.Response.Headers.RetryAfter = useHttpDate + ? DateTimeOffset.FromUnixTimeSeconds(DateTimeOffset.UtcNow.ToUnixTimeSeconds() + 3).ToString("r", CultureInfo.InvariantCulture) + : "2"; + await WriteStatusAsync(context, StatusCodes.Status504GatewayTimeout, "Timeout"); + return; + } + + created.Metadata.Uid = "retried-object"; + await WriteContainerAsync(context, created, StatusCodes.Status201Created); + }); + WriteKubeconfig(kubeconfigPath, server.Port); + Assert.Empty(await service.ListAsync(cancellationToken: cts.Token)); + + var result = await service.CreateAsync(Container.Create("frontend", "test-image"), cts.Token); + + Assert.Equal("retried-object", result.Metadata.Uid); + Assert.Equal(["POST", "GET", "POST"], requests.ToArray()); + } + + [Fact] + public async Task CreateAsync_InitializationTimeoutDoesNotConfirmOrAdoptExistingObject() + { + using var cts = new CancellationTokenSource(TimeSpan.FromSeconds(20)); + var (service, kubeconfigPath, fileSystem) = CreateService( + kubernetesApiTimeout: TimeSpan.FromMilliseconds(500), + kubernetesInitializationAdditionalTimeout: TimeSpan.FromMilliseconds(500), + createRecoveryTimeout: TimeSpan.FromSeconds(5)); + using var disposableFileSystem = fileSystem; + using var disposableService = service; + var requests = new ConcurrentQueue(); + await using var server = await StartServerAsync(cts.Token, requestHandler: async context => + { + requests.Enqueue(context.Request.Method); + if (HttpMethods.IsGet(context.Request.Method)) + { + var existing = Container.Create("frontend", "test-image"); + existing.Metadata.Uid = "pre-existing-object"; + await WriteContainerAsync(context, existing); + } + else + { + await WriteStatusAsync(context, StatusCodes.Status409Conflict, "AlreadyExists"); + } + }); + + var createTask = service.CreateAsync(Container.Create("frontend", "test-image"), cts.Token); + + // Let the first initialization budget expire before making kubeconfig available. + await Task.Delay(TimeSpan.FromSeconds(1.2), cts.Token); + Assert.Empty(requests); + WriteKubeconfig(kubeconfigPath, server.Port); + var exception = await Assert.ThrowsAsync(() => createTask); + + Assert.Equal(HttpStatusCode.Conflict, exception.Response.StatusCode); + Assert.Equal(["POST"], requests.ToArray()); + } + + [Fact] + public async Task CreateAsync_RecoveryBudgetCancelsStalledReconciliation() + { + using var cts = new CancellationTokenSource(TimeSpan.FromSeconds(20)); + var recoveryTimeout = TimeSpan.FromSeconds(2); + var (service, kubeconfigPath, fileSystem) = CreateService( + kubernetesApiTimeout: TimeSpan.FromSeconds(1), + kubernetesInitializationAdditionalTimeout: TimeSpan.FromSeconds(4), + createRecoveryTimeout: recoveryTimeout); + using var disposableFileSystem = fileSystem; + using var disposableService = service; + var requests = new ConcurrentQueue(); + var requestCanceled = new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously); + await using var server = await StartServerAsync(cts.Token, requestHandler: async context => + { + requests.Enqueue(context.Request.Method); + if (HttpMethods.IsPost(context.Request.Method)) + { + await WriteStatusAsync(context, StatusCodes.Status504GatewayTimeout, "Timeout"); + return; + } + + try + { + await Task.Delay(Timeout.InfiniteTimeSpan, context.RequestAborted); + } + catch (OperationCanceledException) when (context.RequestAborted.IsCancellationRequested) + { + requestCanceled.TrySetResult(); + throw; + } + }); + WriteKubeconfig(kubeconfigPath, server.Port); + + var exception = await Assert.ThrowsAsync( + () => service.CreateAsync(Container.Create("frontend", "test-image"), cts.Token)); + + await requestCanceled.Task.WaitAsync(cts.Token); + Assert.Equal(recoveryTimeout, exception.Timeout); + Assert.Equal(["POST", "GET"], requests.ToArray()); + } + + [Fact] + public async Task CreateAsync_DoesNotRetryPermanentReconciliationFailure() + { + using var cts = new CancellationTokenSource(TimeSpan.FromSeconds(20)); + var (service, kubeconfigPath, fileSystem) = CreateService(); + using var disposableFileSystem = fileSystem; + using var disposableService = service; + var requests = new ConcurrentQueue(); + await using var server = await StartServerAsync(cts.Token, requestHandler: async context => + { + requests.Enqueue(context.Request.Method); + await WriteStatusAsync(context, + HttpMethods.IsPost(context.Request.Method) ? StatusCodes.Status504GatewayTimeout : StatusCodes.Status403Forbidden, + "Failure"); + }); + WriteKubeconfig(kubeconfigPath, server.Port); + + var exception = await Assert.ThrowsAsync( + () => service.CreateAsync(Container.Create("frontend", "test-image"), cts.Token)); + + Assert.Equal(HttpStatusCode.Forbidden, exception.Response.StatusCode); + Assert.Equal(["POST", "GET"], requests.ToArray()); + } + + [Fact] + public async Task CreateAsync_CallerCancellationDoesNotReconcileOrRetry() + { + using var cts = new CancellationTokenSource(TimeSpan.FromSeconds(20)); + using var createCts = CancellationTokenSource.CreateLinkedTokenSource(cts.Token); + var (service, kubeconfigPath, fileSystem) = CreateService(); + using var disposableFileSystem = fileSystem; + using var disposableService = service; + var requests = new ConcurrentQueue(); + var requestArrived = new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously); + var requestCanceled = new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously); + await using var server = await StartServerAsync(cts.Token, requestHandler: async context => + { + requests.Enqueue(context.Request.Method); + requestArrived.TrySetResult(); + try + { + await Task.Delay(Timeout.InfiniteTimeSpan, context.RequestAborted); + } + catch (OperationCanceledException) when (context.RequestAborted.IsCancellationRequested) + { + requestCanceled.TrySetResult(); + throw; + } + }); + WriteKubeconfig(kubeconfigPath, server.Port); + + var createTask = service.CreateAsync(Container.Create("frontend", "test-image"), createCts.Token); + await requestArrived.Task.WaitAsync(cts.Token); + createCts.Cancel(); + + await Assert.ThrowsAnyAsync(() => createTask); + await requestCanceled.Task.WaitAsync(cts.Token); + Assert.Equal(["POST"], requests.ToArray()); + } + [Theory] [InlineData(null)] [InlineData("test-namespace")] @@ -200,8 +1035,8 @@ public async Task WatchAsync_IdleHttp2StreamOutlivesConnectionDeadlineAndOtherAp using var cts = CancellationTokenSource.CreateLinkedTokenSource(timeout.Token, TestContext.Current.CancellationToken); using var watchCts = CancellationTokenSource.CreateLinkedTokenSource(cts.Token); var (service, kubeconfigPath, fileSystem) = CreateService( - maxRetryDuration: TimeSpan.FromMilliseconds(500), - kubernetesInitializationTimeout: TimeSpan.FromSeconds(5)); + kubernetesApiTimeout: TimeSpan.FromMilliseconds(500), + kubernetesInitializationAdditionalTimeout: TimeSpan.FromSeconds(5)); using var disposableFileSystem = fileSystem; using var disposableService = service; var responseStarted = new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously); @@ -905,14 +1740,14 @@ public async Task ExecuteWithRetry_UsesInitializationTimeout_WhenKubeconfigAppea using var cts = new CancellationTokenSource(TimeSpan.FromSeconds(20)); var (service, kubeconfigPath, fileSystem) = CreateService( - maxRetryDuration: TimeSpan.FromSeconds(1)); + kubernetesApiTimeout: TimeSpan.FromSeconds(1)); using var disposableFileSystem = fileSystem; using var disposableService = service; var listTask = service.ListAsync(cancellationToken: cts.Token); - // The kubeconfig appears after the normal API retry budget. Initial DCP connection establishment - // needs its own startup budget because DCP has not exposed an API endpoint yet. + // Kubeconfig availability is not proof of API readiness, so the operation starts with + // the initialization budget rather than the shorter steady-state budget. await Task.Delay(TimeSpan.FromSeconds(2), cts.Token); await using var server = await StartServerAsync(cts.Token); @@ -922,14 +1757,120 @@ public async Task ExecuteWithRetry_UsesInitializationTimeout_WhenKubeconfigAppea Assert.Empty(result); } + [Fact] + public async Task ExecuteWithRetry_InitializationBudgetIncludesApiTimeoutPlusAdditionalTime() + { + using var cts = new CancellationTokenSource(TimeSpan.FromSeconds(20)); + var (service, kubeconfigPath, fileSystem) = CreateService( + kubernetesApiTimeout: TimeSpan.FromSeconds(1), + kubernetesInitializationAdditionalTimeout: TimeSpan.FromSeconds(1)); + using var disposableFileSystem = fileSystem; + using var disposableService = service; + await using var server = await StartServerAsync(cts.Token); + server.DelayResponses(TimeSpan.FromSeconds(1.5)); + WriteKubeconfig(kubeconfigPath, server.Port); + + Assert.Empty(await service.ListAsync(cancellationToken: cts.Token)); + await Assert.ThrowsAsync( + () => service.ListAsync(cancellationToken: cts.Token)); + await server.WaitForRequestCancellationAsync(cts.Token); + } + + [Fact] + public async Task ExecuteWithRetry_ZeroInitializationAllowanceUsesApiBudget() + { + using var cts = new CancellationTokenSource(TimeSpan.FromSeconds(20)); + var (service, kubeconfigPath, fileSystem) = CreateService( + kubernetesApiTimeout: TimeSpan.FromSeconds(1), + kubernetesInitializationAdditionalTimeout: TimeSpan.Zero); + using var disposableFileSystem = fileSystem; + using var disposableService = service; + await using var server = await StartServerAsync(cts.Token); + server.DelayResponses(TimeSpan.FromSeconds(2)); + WriteKubeconfig(kubeconfigPath, server.Port); + + await Assert.ThrowsAsync( + () => service.ListAsync(cancellationToken: cts.Token)); + await server.WaitForRequestCancellationAsync(cts.Token); + } + + [Fact] + public async Task ExecuteWithRetry_ClientSetupAndApiRequestShareInitializationBudget() + { + using var cts = new CancellationTokenSource(TimeSpan.FromSeconds(20)); + var (service, kubeconfigPath, fileSystem) = CreateService( + kubernetesApiTimeout: TimeSpan.FromSeconds(1), + kubernetesInitializationAdditionalTimeout: TimeSpan.FromSeconds(3)); + using var disposableFileSystem = fileSystem; + using var disposableService = service; + await using var server = await StartServerAsync(cts.Token); + server.DelayResponses(TimeSpan.FromSeconds(3)); + + var listTask = service.ListAsync(cancellationToken: cts.Token); + await Task.Delay(TimeSpan.FromSeconds(2), cts.Token); + WriteKubeconfig(kubeconfigPath, server.Port); + + // Setup and the HTTP response each fit individually, but their sum exceeds the shared budget. + await Assert.ThrowsAsync(() => listTask); + await server.WaitForRequestCancellationAsync(cts.Token); + } + + [Fact] + public async Task CreateAsync_ConfirmsTimedOutWrite_WhenSetupConsumesInitializationBudget() + { + using var cts = new CancellationTokenSource(TimeSpan.FromSeconds(20)); + var (service, kubeconfigPath, fileSystem) = CreateService( + kubernetesApiTimeout: TimeSpan.FromSeconds(1), + kubernetesInitializationAdditionalTimeout: TimeSpan.FromSeconds(3), + createRecoveryTimeout: TimeSpan.FromSeconds(10)); + using var disposableFileSystem = fileSystem; + using var disposableService = service; + var requests = new ConcurrentQueue(); + var requestCanceled = new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously); + Container? stored = null; + await using var server = await StartServerAsync(cts.Token, requestHandler: async context => + { + requests.Enqueue(context.Request.Method); + if (HttpMethods.IsGet(context.Request.Method)) + { + await WriteContainerAsync(context, Volatile.Read(ref stored)!); + return; + } + + var created = await ReadContainerAsync(context); + created.Metadata.Uid = "committed-object"; + Volatile.Write(ref stored, created); + try + { + await Task.Delay(TimeSpan.FromSeconds(3), context.RequestAborted); + await WriteContainerAsync(context, created, StatusCodes.Status201Created); + } + catch (OperationCanceledException) when (context.RequestAborted.IsCancellationRequested) + { + requestCanceled.TrySetResult(); + throw; + } + }); + + var createTask = service.CreateAsync(Container.Create("frontend", "test-image"), cts.Token); + await Task.Delay(TimeSpan.FromSeconds(2), cts.Token); + WriteKubeconfig(kubeconfigPath, server.Port); + + var result = await createTask; + + await requestCanceled.Task.WaitAsync(cts.Token); + Assert.Equal("committed-object", result.Metadata.Uid); + Assert.Equal(["POST", "GET"], requests.ToArray()); + } + [Fact] public async Task ExecuteWithRetry_UsesInitializationTimeout_UntilFirstApiOperationSucceeds() { using var cts = new CancellationTokenSource(TimeSpan.FromSeconds(20)); var (service, kubeconfigPath, fileSystem) = CreateService( - maxRetryDuration: TimeSpan.FromMilliseconds(500), - kubernetesInitializationTimeout: TimeSpan.FromSeconds(5)); + kubernetesApiTimeout: TimeSpan.FromMilliseconds(500), + kubernetesInitializationAdditionalTimeout: TimeSpan.FromSeconds(4.5)); using var disposableFileSystem = fileSystem; using var disposableService = service; @@ -951,8 +1892,8 @@ public async Task ExecuteWithRetry_UsesApiRetryDuration_AfterFirstApiOperationSu using var cts = new CancellationTokenSource(TimeSpan.FromSeconds(20)); var (service, kubeconfigPath, fileSystem) = CreateService( - maxRetryDuration: TimeSpan.FromMilliseconds(500), - kubernetesInitializationTimeout: TimeSpan.FromSeconds(5)); + kubernetesApiTimeout: TimeSpan.FromMilliseconds(500), + kubernetesInitializationAdditionalTimeout: TimeSpan.FromSeconds(4.5)); using var disposableFileSystem = fileSystem; using var disposableService = service; @@ -967,14 +1908,18 @@ await Assert.ThrowsAsync( () => service.ListAsync(cancellationToken: cts.Token)); } - [Fact] - public async Task ExecuteWithRetry_CancelsApiRequest_WhenRetryDurationExpires() + [Theory] + [InlineData(nameof(KubernetesService.ListAsync))] + [InlineData(nameof(KubernetesService.GetAsync))] + [InlineData(nameof(KubernetesService.PatchAsync))] + [InlineData(nameof(KubernetesService.DeleteAsync))] + public async Task ExecuteWithRetry_CancelsApiRequest_WhenRetryDurationExpires(string operationName) { using var cts = new CancellationTokenSource(TimeSpan.FromSeconds(20)); var (service, kubeconfigPath, fileSystem) = CreateService( - maxRetryDuration: TimeSpan.FromMilliseconds(500), - kubernetesInitializationTimeout: TimeSpan.FromSeconds(5)); + kubernetesApiTimeout: TimeSpan.FromMilliseconds(500), + kubernetesInitializationAdditionalTimeout: TimeSpan.FromSeconds(4.5)); using var disposableFileSystem = fileSystem; using var disposableService = service; @@ -983,9 +1928,62 @@ public async Task ExecuteWithRetry_CancelsApiRequest_WhenRetryDurationExpires() Assert.Empty(await service.ListAsync(cancellationToken: cts.Token)); server.DelayResponses(TimeSpan.FromSeconds(2)); + Func operation = operationName switch + { + nameof(KubernetesService.ListAsync) => () => service.ListAsync(cancellationToken: cts.Token), + nameof(KubernetesService.GetAsync) => () => service.GetAsync("frontend", cancellationToken: cts.Token), + nameof(KubernetesService.PatchAsync) => () => service.PatchAsync(Container.Create("frontend", "test-image"), new V1Patch("[]", V1Patch.PatchType.JsonPatch), cts.Token), + nameof(KubernetesService.DeleteAsync) => () => service.DeleteAsync("frontend", cancellationToken: cts.Token), + _ => throw new InvalidOperationException($"Unknown API operation {operationName}."), + }; + + await Assert.ThrowsAsync(operation); + await server.WaitForRequestCancellationAsync(cts.Token); + } + + [Fact] + public async Task ExecuteWithRetry_ReadinessDoesNotShortenInFlightOperation() + { + using var cts = new CancellationTokenSource(TimeSpan.FromSeconds(20)); + var (service, kubeconfigPath, fileSystem) = CreateService( + kubernetesApiTimeout: TimeSpan.FromMilliseconds(500), + kubernetesInitializationAdditionalTimeout: TimeSpan.FromSeconds(4.5)); + using var disposableFileSystem = fileSystem; + using var disposableService = service; + var requestArrived = new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously); + var requestCanceled = new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously); + await using var server = await StartServerAsync(cts.Token, requestHandler: async context => + { + if (context.Request.Path.Value!.EndsWith("/readiness-probe", StringComparison.Ordinal)) + { + await WriteContainerAsync(context, Container.Create("readiness-probe", "test-image")); + return; + } + + requestArrived.TrySetResult(); + try + { + await Task.Delay(TimeSpan.FromSeconds(1), context.RequestAborted); + context.Response.ContentType = "application/json"; + await context.Response.WriteAsync("""{"apiVersion":"usvc-dev.developer.microsoft.com/v1","kind":"ContainerList","items":[]}""", context.RequestAborted); + } + catch (OperationCanceledException) when (context.RequestAborted.IsCancellationRequested) + { + requestCanceled.TrySetResult(); + throw; + } + }); + WriteKubeconfig(kubeconfigPath, server.Port); + + var listTask = service.ListAsync(cancellationToken: cts.Token); + await requestArrived.Task.WaitAsync(cts.Token); + var probe = await service.GetAsync("readiness-probe", cancellationToken: cts.Token); + + Assert.Equal("readiness-probe", probe.Metadata.Name); + Assert.Empty(await listTask); await Assert.ThrowsAsync( () => service.ListAsync(cancellationToken: cts.Token)); - await server.WaitForRequestCancellationAsync(cts.Token); + await requestCanceled.Task.WaitAsync(cts.Token); } [Fact] @@ -995,8 +1993,8 @@ public async Task WatchAsync_DoesNotMarkApiReady_WhileHttpResponseIsPending() using var watchCts = CancellationTokenSource.CreateLinkedTokenSource(cts.Token); var (service, kubeconfigPath, fileSystem) = CreateService( - maxRetryDuration: TimeSpan.FromMilliseconds(500), - kubernetesInitializationTimeout: TimeSpan.FromSeconds(5)); + kubernetesApiTimeout: TimeSpan.FromMilliseconds(500), + kubernetesInitializationAdditionalTimeout: TimeSpan.FromSeconds(4.5)); using var disposableFileSystem = fileSystem; using var disposableService = service; @@ -1022,6 +2020,126 @@ public async Task WatchAsync_DoesNotMarkApiReady_WhileHttpResponseIsPending() } } + [Fact] + public async Task WatchAsync_ReconnectUsesApiTimeout_AfterAnotherOperationConfirmsReadiness() + { + using var cts = new CancellationTokenSource(TimeSpan.FromSeconds(20)); + using var watchCts = CancellationTokenSource.CreateLinkedTokenSource(cts.Token); + var (service, kubeconfigPath, fileSystem) = CreateService( + kubernetesApiTimeout: TimeSpan.FromMilliseconds(250), + kubernetesInitializationAdditionalTimeout: TimeSpan.FromSeconds(2.75)); + using var disposableFileSystem = fileSystem; + using var disposableService = service; + await using var server = await StartServerAsync(cts.Token); + server.BlockWatchResponses(); + WriteKubeconfig(kubeconfigPath, server.Port); + + await using var watchEnumerator = service.WatchAsync(cancellationToken: watchCts.Token).GetAsyncEnumerator(); + var watchTask = watchEnumerator.MoveNextAsync().AsTask(); + try + { + await server.WaitForWatchRequestAsync(cts.Token); + Assert.Empty(await service.ListAsync(cancellationToken: cts.Token)); + await server.WaitForWatchRequestCountAsync(2, cts.Token); + + // The next stalled connection must use the shorter budget rather than another startup budget. + using var reconnectCts = CancellationTokenSource.CreateLinkedTokenSource(cts.Token); + reconnectCts.CancelAfter(TimeSpan.FromSeconds(1.5)); + await server.WaitForWatchRequestCountAsync(3, reconnectCts.Token); + Assert.False(watchTask.IsCompleted); + } + finally + { + watchCts.Cancel(); + await Assert.ThrowsAnyAsync(() => watchTask); + } + } + + [Fact] + public async Task WatchAsync_ConnectedStreamOutlivesApiTimeout() + { + using var cts = new CancellationTokenSource(TimeSpan.FromSeconds(20)); + using var watchCts = CancellationTokenSource.CreateLinkedTokenSource(cts.Token); + var (service, kubeconfigPath, fileSystem) = CreateService( + kubernetesApiTimeout: TimeSpan.FromMilliseconds(500), + kubernetesInitializationAdditionalTimeout: TimeSpan.FromSeconds(1.5)); + using var disposableFileSystem = fileSystem; + using var disposableService = service; + var requestCanceled = new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously); + await using var server = await StartServerAsync(cts.Token, requestHandler: async context => + { + var container = Container.Create("frontend", "test-image"); + context.Response.ContentType = "application/json"; + await context.Response.WriteAsync( + $$"""{"type":"ADDED","object":{{KubernetesJson.Serialize(container)}}}""" + "\n", + context.RequestAborted); + await context.Response.Body.FlushAsync(context.RequestAborted); + try + { + await Task.Delay(Timeout.InfiniteTimeSpan, context.RequestAborted); + } + catch (OperationCanceledException) when (context.RequestAborted.IsCancellationRequested) + { + requestCanceled.TrySetResult(); + throw; + } + }); + WriteKubeconfig(kubeconfigPath, server.Port); + + await using var watchEnumerator = service.WatchAsync(cancellationToken: watchCts.Token).GetAsyncEnumerator(); + Assert.True(await watchEnumerator.MoveNextAsync()); + Assert.Equal(WatchEventType.Added, watchEnumerator.Current.Item1); + Assert.Equal("frontend", watchEnumerator.Current.Item2.Metadata.Name); + var watchTask = watchEnumerator.MoveNextAsync().AsTask(); + try + { + await Task.Delay(TimeSpan.FromSeconds(3), cts.Token); + Assert.False(watchTask.IsCompleted); + } + finally + { + watchCts.Cancel(); + await Assert.ThrowsAnyAsync(() => watchTask); + } + + await requestCanceled.Task.WaitAsync(cts.Token); + } + + [Fact] + public async Task GetLogStreamAsync_ConnectedStreamOutlivesApiTimeout() + { + using var cts = new CancellationTokenSource(TimeSpan.FromSeconds(20)); + var (service, kubeconfigPath, fileSystem) = CreateService( + kubernetesApiTimeout: TimeSpan.FromMilliseconds(500), + kubernetesInitializationAdditionalTimeout: TimeSpan.FromSeconds(1.5)); + using var disposableFileSystem = fileSystem; + using var disposableService = service; + var continueStream = new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously); + await using var server = await StartServerAsync(cts.Token, requestHandler: async context => + { + context.Response.ContentType = "text/plain"; + await context.Response.WriteAsync("first\n", context.RequestAborted); + await context.Response.Body.FlushAsync(context.RequestAborted); + await continueStream.Task.WaitAsync(context.RequestAborted); + await context.Response.WriteAsync("second\n", context.RequestAborted); + }); + WriteKubeconfig(kubeconfigPath, server.Port); + + try + { + await using var stream = await service.GetLogStreamAsync(Container.Create("frontend", "test-image"), "stdout", cts.Token); + using var reader = new StreamReader(stream, leaveOpen: true); + Assert.Equal("first", await reader.ReadLineAsync(cts.Token)); + await Task.Delay(TimeSpan.FromSeconds(3), cts.Token); + continueStream.TrySetResult(); + Assert.Equal("second", await reader.ReadLineAsync(cts.Token)); + } + finally + { + continueStream.TrySetResult(); + } + } + [Fact] public async Task WatchAsync_RemainsActiveBeyondApiRetryTimeout() { @@ -1029,8 +2147,8 @@ public async Task WatchAsync_RemainsActiveBeyondApiRetryTimeout() using var watchCts = CancellationTokenSource.CreateLinkedTokenSource(cts.Token); var (service, kubeconfigPath, fileSystem) = CreateService( - maxRetryDuration: TimeSpan.FromMilliseconds(100), - kubernetesInitializationTimeout: TimeSpan.FromMilliseconds(250)); + kubernetesApiTimeout: TimeSpan.FromMilliseconds(100), + kubernetesInitializationAdditionalTimeout: TimeSpan.FromMilliseconds(150)); using var disposableFileSystem = fileSystem; using var disposableService = service; @@ -1080,8 +2198,9 @@ public async Task ExecuteWithRetry_EstablishesConnection_WhenKubeconfigInitially } private static (KubernetesService Service, string KubeconfigPath, IDisposable FileSystem) CreateService( - TimeSpan? maxRetryDuration = null, - TimeSpan? kubernetesInitializationTimeout = null) + TimeSpan? kubernetesApiTimeout = null, + TimeSpan? kubernetesInitializationAdditionalTimeout = null, + TimeSpan? createRecoveryTimeout = null) { var configuration = new ConfigurationBuilder().Build(); @@ -1096,14 +2215,12 @@ private static (KubernetesService Service, string KubeconfigPath, IDisposable Fi // Poll quickly so the kubeconfig file-wait/read retries react promptly in tests. KubernetesConfigReadRetryIntervalMilliseconds = 50, KubernetesConfigReadRetryCount = 300, + KubernetesApiTimeout = kubernetesApiTimeout ?? TimeSpan.FromSeconds(40), + KubernetesInitializationAdditionalTimeout = kubernetesInitializationAdditionalTimeout ?? TimeSpan.FromSeconds(20), + KubernetesCreateRecoveryTimeout = createRecoveryTimeout ?? TimeSpan.FromMinutes(2), }); - var service = new KubernetesService(NullLogger.Instance, dcpOptions, locations, configuration) - { - // Generous enough that the test can flip the kubeconfig before the retry budget is exhausted. - MaxRetryDuration = maxRetryDuration ?? TimeSpan.FromSeconds(30), - KubernetesInitializationTimeout = kubernetesInitializationTimeout ?? TimeSpan.FromSeconds(60), - }; + var service = new KubernetesService(NullLogger.Instance, dcpOptions, locations, configuration); return (service, locations.DcpKubeconfigPath, fileSystem); } @@ -1115,6 +2232,19 @@ private static (KubernetesService Service, string KubeconfigPath, IDisposable Fi } } + private static async Task ReadContainerAsync(HttpContext context) + { + using var reader = new StreamReader(context.Request.Body); + return KubernetesJson.Deserialize(await reader.ReadToEndAsync(context.RequestAborted)); + } + + private static Task WriteContainerAsync(HttpContext context, Container container, int statusCode = StatusCodes.Status200OK) + { + context.Response.StatusCode = statusCode; + context.Response.ContentType = "application/json"; + return context.Response.WriteAsync(KubernetesJson.Serialize(container), context.RequestAborted); + } + private static Task WriteStatusAsync(HttpContext context, int statusCode, string reason) { // Match Kubernetes API errors, for example: