diff --git a/src/Aspire.Hosting/Dcp/KubernetesService.cs b/src/Aspire.Hosting/Dcp/KubernetesService.cs index 37e61b6247b..bf38c54e1fb 100644 --- a/src/Aspire.Hosting/Dcp/KubernetesService.cs +++ b/src/Aspire.Hosting/Dcp/KubernetesService.cs @@ -471,17 +471,23 @@ private async Task ExecuteWithRetry( Func isRetryable, CancellationToken cancellationToken) { - var clientReady = await EnsureKubernetesAsync(cancellationToken).ConfigureAwait(false); using var activity = ProfilingTelemetry.StartDcpKubernetesApi(configuration, operationType, resourceType); - activity.AddKubernetesClientReady(clientReady.WaitMilliseconds, clientReady.Initialized); var retryCount = 0; var resiliencePipeline = CreateKubernetesCallResiliencePipeline(isRetryable, activity, () => retryCount++); try { - return await resiliencePipeline.ExecuteAsync(async (cancellationToken) => + return await resiliencePipeline.ExecuteAsync(async (cancellationToken) => { + // Establish (or re-establish) the connection to DCP inside the retry loop, + // to guard against a failure from missing or partially-written kubeconfig. + var clientReady = await EnsureKubernetesAsync(cancellationToken).ConfigureAwait(false); + if (clientReady.Initialized) + { + activity.AddKubernetesClientReady(clientReady.WaitMilliseconds, clientReady.Initialized); + } + return await operation(_kubernetes!).ConfigureAwait(false); }, cancellationToken).ConfigureAwait(false); } @@ -593,43 +599,44 @@ private async Task EnsureKubernetesAsync(CancellationToke readStopwatch.Start(); var readPipeline = CreateReadKubeconfigResiliencePipeline(); - var timeoutPipeline = new ResiliencePipelineBuilder() - .AddTimeout(new TimeoutStrategyOptions { Timeout = MaxRetryDuration }) - .Build(); try { - _kubernetes = await timeoutPipeline.ExecuteAsync(async (cancellationToken) => + _kubernetes = await readPipeline.ExecuteAsync(async (cancellationToken) => { - return await readPipeline.ExecuteAsync(async (cancellationToken) => + var fileWaitStopwatch = Stopwatch.StartNew(); + var fileInfo = new FileInfo(locations.DcpKubeconfigPath); + while (!fileInfo.Exists) { - var fileWaitStopwatch = Stopwatch.StartNew(); - var fileInfo = new FileInfo(locations.DcpKubeconfigPath); - while (!fileInfo.Exists) - { - await Task.Delay(TimeSpan.FromMilliseconds(dcpOptions.Value.KubernetesConfigReadRetryIntervalMilliseconds), cancellationToken).ConfigureAwait(false); - fileInfo = new FileInfo(locations.DcpKubeconfigPath); - } - fileWaitStopwatch.Stop(); - activity.SetDcpKubeconfigFileWait(fileWaitStopwatch.ElapsedMilliseconds); - activity.AddKubeconfigFileDetected(); - - var buildConfigStopwatch = Stopwatch.StartNew(); - var config = await KubernetesClientConfiguration.BuildConfigFromConfigFileAsync(kubeconfig: fileInfo, useRelativePaths: false).ConfigureAwait(false); - buildConfigStopwatch.Stop(); - readStopwatch.Stop(); - - logger.LogDebug( - "Successfully read Kubernetes configuration from '{DcpKubeconfigPath}' after {DurationMs} milliseconds.", - locations.DcpKubeconfigPath, - readStopwatch.ElapsedMilliseconds - ); - activity.SetDcpKubeconfigBuildDuration(buildConfigStopwatch.ElapsedMilliseconds); - activity.SetDcpKubeconfigReadDuration(readStopwatch.ElapsedMilliseconds); - activity.AddKubeconfigReadComplete(); - - return new DcpKubernetesClient(config); - }, cancellationToken).ConfigureAwait(false); + await Task.Delay(TimeSpan.FromMilliseconds(dcpOptions.Value.KubernetesConfigReadRetryIntervalMilliseconds), cancellationToken).ConfigureAwait(false); + fileInfo = new FileInfo(locations.DcpKubeconfigPath); + } + fileWaitStopwatch.Stop(); + activity.SetDcpKubeconfigFileWait(fileWaitStopwatch.ElapsedMilliseconds); + activity.AddKubeconfigFileDetected(); + + var buildConfigStopwatch = Stopwatch.StartNew(); + // Open the file with FileShare.ReadWrite | FileShare.Delete so we do not interfere with DCP + // if we happen to open the file while DCP is still writing to it. + var kubeconfigStream = new FileStream(fileInfo.FullName, FileMode.Open, FileAccess.Read, FileShare.ReadWrite | FileShare.Delete); + KubernetesClientConfiguration config; + await using (kubeconfigStream.ConfigureAwait(false)) + { + config = await KubernetesClientConfiguration.BuildConfigFromConfigFileAsync(kubeconfigStream).ConfigureAwait(false); + } + buildConfigStopwatch.Stop(); + readStopwatch.Stop(); + + logger.LogDebug( + "Successfully read Kubernetes configuration from '{DcpKubeconfigPath}' after {DurationMs} milliseconds.", + locations.DcpKubeconfigPath, + readStopwatch.ElapsedMilliseconds + ); + activity.SetDcpKubeconfigBuildDuration(buildConfigStopwatch.ElapsedMilliseconds); + activity.SetDcpKubeconfigReadDuration(readStopwatch.ElapsedMilliseconds); + activity.AddKubeconfigReadComplete(); + + return new DcpKubernetesClient(config); }, cancellationToken).ConfigureAwait(false); } catch (Exception ex) diff --git a/tests/Aspire.Hosting.Tests/Dcp/KubernetesServiceTests.cs b/tests/Aspire.Hosting.Tests/Dcp/KubernetesServiceTests.cs new file mode 100644 index 00000000000..6752fbb7a75 --- /dev/null +++ b/tests/Aspire.Hosting.Tests/Dcp/KubernetesServiceTests.cs @@ -0,0 +1,287 @@ +// Licensed to the .NET Foundation under one or more agreements. +// The .NET Foundation licenses this file to you under the MIT license. + +#pragma warning disable ASPIREFILESYSTEM001 // IFileSystemService is for evaluation purposes only. + +using System.Globalization; +using Aspire.Hosting.Dcp; +using Aspire.Hosting.Dcp.Model; +using Microsoft.AspNetCore.Builder; +using Microsoft.AspNetCore.Hosting; +using Microsoft.AspNetCore.Http; +using Microsoft.Extensions.Configuration; +using Microsoft.Extensions.Logging; +using Microsoft.Extensions.Logging.Abstractions; +using Microsoft.Extensions.Options; + +namespace Aspire.Hosting.Tests.Dcp; + +public class KubernetesServiceTests +{ + // Verifies that establishing the connection happens inside the retry loop: when the kubeconfig does not + // exist yet (DCP has not finished writing it), the operation waits and succeeds once it appears. + [Fact] + public async Task ExecuteWithRetry_EstablishesConnection_WhenKubeconfigInitiallyMissing() + { + using var cts = new CancellationTokenSource(TimeSpan.FromSeconds(60)); + + var (service, kubeconfigPath, fileSystem) = CreateService(); + using var disposableFileSystem = fileSystem; + using var disposableService = service; + + // No kubeconfig on disk initially. + Assert.False(File.Exists(kubeconfigPath)); + + var listTask = service.ListAsync(cancellationToken: cts.Token); + + await Task.Delay(300, cts.Token); + + await using var server = await TestDcpApiServer.StartAsync(cts.Token); + WriteKubeconfig(kubeconfigPath, server.Port); + + var result = await listTask; + Assert.Empty(result); + } + + // Verifies that establishing the connection survives a partially-written kubeconfig: when the file exists + // but DCP has only flushed part of it (so it does not yet parse as a valid kubeconfig), the read is retried + // and the operation succeeds once the complete, valid kubeconfig is written. + [Fact] + public async Task ExecuteWithRetry_EstablishesConnection_WhenKubeconfigInitiallyPartiallyWritten() + { + using var cts = new CancellationTokenSource(TimeSpan.FromSeconds(60)); + + var (service, kubeconfigPath, fileSystem) = CreateService(); + using var disposableFileSystem = fileSystem; + using var disposableService = service; + + // Simulate DCP having flushed only the first part of the kubeconfig + WritePartialKubeconfig(kubeconfigPath); + + var listTask = service.ListAsync(cancellationToken: cts.Token); + + // Give the read pipeline time to observe and retry the partial file before we finish writing it. + await Task.Delay(300, cts.Token); + + await using var server = await TestDcpApiServer.StartAsync(cts.Token); + + // Finish the write by appending the remainder onto the same file. In-flight DCP calls will now succeed. + CompleteKubeconfig(kubeconfigPath, server.Port); + + var result = await listTask; + Assert.Empty(result); + } + + private static (KubernetesService Service, string KubeconfigPath, IDisposable FileSystem) CreateService() + { + var configuration = new ConfigurationBuilder().Build(); + + // Decouple the kubeconfig location from the production FileSystemService + var fileSystem = new TestFileSystemService(); + try + { + var locations = new Locations(fileSystem); + + var dcpOptions = Options.Create(new DcpOptions + { + // Poll quickly so the kubeconfig file-wait/read retries react promptly in tests. + KubernetesConfigReadRetryIntervalMilliseconds = 50, + KubernetesConfigReadRetryCount = 300, + }); + + 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 = TimeSpan.FromSeconds(30), + }; + + return (service, locations.DcpKubeconfigPath, fileSystem); + } + catch + { + // Don't orphan the temp directory if wiring up the service fails before the test takes ownership. + fileSystem.Dispose(); + throw; + } + } + + private static void WriteKubeconfig(string path, int port) + { + // Minimal kubeconfig pointing at a plain-HTTP loopback endpoint with no auth, which is all the + // DcpKubernetesClient needs to issue custom-object requests against the fake server. + var content = string.Format(CultureInfo.InvariantCulture, """ + apiVersion: v1 + kind: Config + clusters: + - name: dcp + cluster: + server: http://127.0.0.1:{0} + contexts: + - name: dcp + context: + cluster: dcp + user: dcp + current-context: dcp + users: + - name: dcp + user: + token: dcp-test-token + """, port); + + // Write atomically (temp file + move on the same volume) so a concurrent read by the service never + // observes a half-written file. + var tempPath = path + ".tmp"; + File.WriteAllText(tempPath, content); + File.Move(tempPath, path, overwrite: true); + } + + private static void WritePartialKubeconfig(string path) + { + // A genuine prefix of the final kubeconfig that stops in the middle of the double-quoted server value. + // The unterminated quote makes this deterministically fail YAML parsing, which models DCP having only + // flushed part of the file. There is intentionally no trailing newline, so appending the remainder in + // CompleteKubeconfig closes the quote and yields exactly-valid YAML. + File.WriteAllText(path, """ + apiVersion: v1 + kind: Config + clusters: + - name: dcp + cluster: + server: "http://127.0.0.1: + """); + } + + private static void CompleteKubeconfig(string path, int port) + { + // Append (do not rewrite) the remainder of the kubeconfig that WritePartialKubeconfig left unfinished. + // The remainder begins by closing the open server scalar ({port}") so the combined file parses as a + // valid kubeconfig pointing at the loopback fake server with no auth. + File.AppendAllText(path, string.Format(CultureInfo.InvariantCulture, """ + {0}" + contexts: + - name: dcp + context: + cluster: dcp + user: dcp + current-context: dcp + users: + - name: dcp + user: + token: dcp-test-token + """, port)); + } + + // A self-contained IFileSystemService for these tests. It hands Locations a single, uniquely-suffixed temp + // directory that the test owns, decoupling the kubeconfig location from the production FileSystemService so + // concurrent runs on the same machine never share a path. Every file a test writes lives under this root, so + // disposing the fake (always, via `using`) removes the kubeconfig and any partial/temp files regardless of the + // test outcome. + private sealed class TestFileSystemService : IFileSystemService, IDisposable + { + private readonly TestTempFileSystemService _tempDirectory = new(); + + public ITempFileSystemService TempDirectory => _tempDirectory; + + public void Dispose() => _tempDirectory.Dispose(); + + private sealed class TestTempFileSystemService : ITempFileSystemService, IDisposable + { + private string? _root; + + public TempDirectory CreateTempSubdirectory(string? prefix = null) + { + // Created lazily (Locations calls this exactly once) so the directory can't be orphaned if the + // test fails to take ownership, and with a random suffix so each test instance is isolated. + _root ??= Directory.CreateTempSubdirectory("test-kubeconfig-").FullName; + return new TestTempDirectory(_root); + } + + public TempFile CreateTempFile(string? fileName = null) + => throw new NotSupportedException("The kubeconfig tests only allocate a temp subdirectory."); + + public void Dispose() + { + if (_root is null) + { + return; + } + + try + { + if (Directory.Exists(_root)) + { + Directory.Delete(_root, recursive: true); + } + } + catch + { + // Best-effort cleanup; a teardown failure must never mask the test result. + } + } + } + + // The owning TestTempFileSystemService deletes the root recursively on Dispose, + // so this handle has nothing of its own to release. + private sealed class TestTempDirectory(string path) : TempDirectory + { + public override string Path => path; + + public override void Dispose() + { + } + } + } + + // A minimal stand-in for the DCP API server. It answers every request with an empty Kubernetes list, which + // is enough for ListAsync() to deserialize successfully. + // + // It runs a real Kestrel server bound to port 0 so the OS assigns a free port that Kestrel actually binds and + // holds for the lifetime of the server. The bound port is read back after startup. This avoids the classic + // "probe a free port then release it and hope nobody grabs it before we rebind" race. + private sealed class TestDcpApiServer : IAsyncDisposable + { + private readonly WebApplication _app; + + private TestDcpApiServer(WebApplication app, int port) + { + _app = app; + Port = port; + } + + public int Port { get; } + + public static async Task StartAsync(CancellationToken cancellationToken = default) + { + var builder = WebApplication.CreateSlimBuilder(); + // Keep the test output clean; the fake server's logs are noise. + builder.Logging.ClearProviders(); + // Port 0 lets the OS pick a free port that Kestrel binds and holds. After StartAsync the addresses + // feature (exposed via app.Urls) is rewritten with the resolved address, so we can read the real port. + builder.WebHost.UseUrls("http://127.0.0.1:0"); + + var app = builder.Build(); + + // Answer every request with an empty container list. + app.Run(async context => + { + context.Response.StatusCode = StatusCodes.Status200OK; + context.Response.ContentType = "application/json"; + await context.Response.WriteAsync("""{"apiVersion":"usvc-dev.developer.microsoft.com/v1","kind":"ContainerList","items":[]}"""); + }); + + await app.StartAsync(cancellationToken).ConfigureAwait(false); + + // e.g. "http://127.0.0.1:54321" -> 54321 + var address = app.Urls.First(); + var port = new Uri(address).Port; + + return new TestDcpApiServer(app, port); + } + + public async ValueTask DisposeAsync() + { + await _app.StopAsync().ConfigureAwait(false); + await _app.DisposeAsync().ConfigureAwait(false); + } + } +}