// Licensed to the .NET Foundation under one or more agreements.
// The .NET Foundation licenses this file to you under the MIT license.
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;
using Aspire.Hosting.Utils;
using k8s;
using k8s.Autorest;
using k8s.Exceptions;
using k8s.Models;
using Microsoft.Extensions.Configuration;
using Microsoft.Extensions.Logging;
using Microsoft.Extensions.Options;
using Polly;
using Polly.Retry;
using Polly.Timeout;
using YamlDotNet.Core;
namespace Aspire.Hosting.Dcp;
internal enum DcpApiOperationType
{
Create = 1,
List = 2,
Delete = 3,
Watch = 4,
GetLogSubresource = 5,
Get = 6,
Patch = 7,
ServerStop = 8,
ResourceCleanup = 9
}
internal interface IKubernetesService
{
Task<T> GetAsync<T>(string name, string? namespaceParameter = null, CancellationToken cancellationToken = default)
where T : CustomResource, IKubernetesStaticMetadata;
Task<T> CreateAsync<T>(T obj, CancellationToken cancellationToken = default)
where T : CustomResource, IKubernetesStaticMetadata;
Task<T> PatchAsync<T>(T obj, V1Patch patch, CancellationToken cancellationToken = default)
where T : CustomResource, IKubernetesStaticMetadata;
Task<List<T>> ListAsync<T>(string? namespaceParameter = null, CancellationToken cancellationToken = default)
where T : CustomResource, IKubernetesStaticMetadata;
Task<T> DeleteAsync<T>(string name, string? namespaceParameter = null, CancellationToken cancellationToken = default)
where T : CustomResource, IKubernetesStaticMetadata;
IAsyncEnumerable<(WatchEventType, T)> WatchAsync<T>(
string? namespaceParameter = null,
CancellationToken cancellationToken = default)
where T : CustomResource, IKubernetesStaticMetadata;
/// <summary>
/// Returns a log stream for the specified resource.
/// </summary>
/// <param name="obj">The resource to get the log stream for.</param>
/// <param name="logStreamType">The type of log stream to retrieve ("stdout", "stderr", "startup_stdout", or "startup_stderr", see <see cref="Aspire.Hosting.Dcp.Model.Logs"/>).</param>
/// <param name="cancellationToken">The cancellation token for the stream retrieval operation (does not affect the returned stream).</param>
/// <param name="follow">If true, the log stream will be followed until the resource is deleted or the stream is disposed of.</param>
/// <param name="timestamps">If true, timestamps (RFC3339) will be included in the log stream.</param>
/// <param name="lineNumbers">If true, line numbers will be included in the log stream.</param>
/// <param name="limit">If specified, limits the number of log linets returned. Cannot be used with "follow".</param>
/// <param name="tail">If specified, limits the response to at most N existing, NEWEST log lines. If "follow" is true, new log lines that appear after the log stream was created do not count against the limit, and will be streamed until the client closes the stream.</param>
/// <param name="skip">If specified, skips the first N log lines in the result set. Cannot be used together with "tail".</param>
Task<Stream> GetLogStreamAsync<T>(
T obj,
string logStreamType,
CancellationToken cancellationToken = default,
bool? follow = true,
bool? timestamps = false,
bool? lineNumbers = false,
long? limit = null,
long? tail = null,
long? skip = null
) where T : CustomResource, IKubernetesStaticMetadata;
Task StopServerAsync(string resourceCleanup = ResourceCleanup.Full, CancellationToken cancellation = default);
Task CleanupResourcesAsync(CancellationToken cancellationToken = default);
}
internal sealed class KubernetesService(ILogger<KubernetesService> logger, IOptions<DcpOptions> dcpOptions, Locations locations, IConfiguration configuration, TimeProvider timeProvider) : IKubernetesService, IDisposable
{
// A pseudo-resource type used for log operations on the DCP execution document.
private const string DcpExecutionResourceType = "DcpExecution";
private static readonly TimeSpan s_initialRetryDelay = TimeSpan.FromMilliseconds(100);
private static GroupVersion GroupVersion => Model.Dcp.GroupVersion;
private readonly SemaphoreSlim _kubeconfigReadSemaphore = new(1);
private DcpKubernetesClient? _kubernetes;
private bool _kubernetesApiReady;
private ResiliencePipeline? _resiliencePipeline;
private bool _disposed;
internal TimeSpan WatchRestartInterval { get; set; } = TimeSpan.FromMinutes(5);
public Task<T> GetAsync<T>(string name, string? namespaceParameter = null, CancellationToken cancellationToken = default)
where T : CustomResource, IKubernetesStaticMetadata
{
var resourceType = GetResourceFor<T>();
return ExecuteWithRetry(
DcpApiOperationType.Get,
T.ObjectKind,
async (kubernetes, operationCancellationToken) =>
{
var response = string.IsNullOrEmpty(namespaceParameter)
? await kubernetes.CustomObjects.GetClusterCustomObjectWithHttpMessagesAsync(
GroupVersion.Group,
GroupVersion.Version,
resourceType,
name,
cancellationToken: operationCancellationToken).ConfigureAwait(false)
: await kubernetes.CustomObjects.GetNamespacedCustomObjectWithHttpMessagesAsync(
GroupVersion.Group,
GroupVersion.Version,
namespaceParameter,
resourceType,
name,
cancellationToken: operationCancellationToken).ConfigureAwait(false);
return KubernetesJson.Deserialize<T>(response.Body.ToString());
},
RetryOnConnectivityAndConflictErrors,
cancellationToken);
}
public async Task<T> CreateAsync<T>(T obj, CancellationToken cancellationToken = default)
where T : CustomResource, IKubernetesStaticMetadata
{
ObjectDisposedException.ThrowIf(_disposed, this);
var resourceType = GetResourceFor<T>();
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 { TimeProvider = timeProvider }
.AddTimeout(new TimeoutStrategyOptions
{
Timeout = recoveryTimeout,
})
.Build()
.ExecuteAsync(async recoveryCancellationToken =>
{
var hasAmbiguousCreate = false;
var retryDelay = s_initialRetryDelay;
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);
}
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<T>(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<T> PatchAsync<T>(T obj, V1Patch patch, CancellationToken cancellationToken = default)
where T : CustomResource, IKubernetesStaticMetadata
{
ObjectDisposedException.ThrowIf(_disposed, this);
var resourceType = GetResourceFor<T>();
var namespaceParameter = obj.Namespace();
return ExecuteWithRetry(
DcpApiOperationType.Patch,
T.ObjectKind,
async (kubernetes, operationCancellationToken) =>
{
var response = string.IsNullOrEmpty(namespaceParameter)
? await kubernetes.CustomObjects.PatchClusterCustomObjectWithHttpMessagesAsync(
patch,
GroupVersion.Group,
GroupVersion.Version,
resourceType,
obj.Metadata.Name,
cancellationToken: operationCancellationToken).ConfigureAwait(false)
: await kubernetes.CustomObjects.PatchNamespacedCustomObjectWithHttpMessagesAsync(
patch,
GroupVersion.Group,
GroupVersion.Version,
namespaceParameter,
resourceType,
obj.Metadata.Name,
cancellationToken: operationCancellationToken).ConfigureAwait(false);
return KubernetesJson.Deserialize<T>(response.Body.ToString());
},
RetryOnConnectivityErrors,
cancellationToken);
}
public Task<List<T>> ListAsync<T>(string? namespaceParameter = null, CancellationToken cancellationToken = default)
where T : CustomResource, IKubernetesStaticMetadata
{
ObjectDisposedException.ThrowIf(_disposed, this);
var resourceType = GetResourceFor<T>();
return ExecuteWithRetry(
DcpApiOperationType.List,
T.ObjectKind,
async (kubernetes, operationCancellationToken) =>
{
var response = string.IsNullOrEmpty(namespaceParameter)
? await kubernetes.CustomObjects.ListClusterCustomObjectWithHttpMessagesAsync(
GroupVersion.Group,
GroupVersion.Version,
resourceType,
cancellationToken: operationCancellationToken).ConfigureAwait(false)
: await kubernetes.CustomObjects.ListNamespacedCustomObjectWithHttpMessagesAsync(
GroupVersion.Group,
GroupVersion.Version,
namespaceParameter,
resourceType,
cancellationToken: operationCancellationToken).ConfigureAwait(false);
return KubernetesJson.Deserialize<CustomResourceList<T>>(response.Body.ToString()).Items;
},
RetryOnConnectivityAndConflictErrors,
cancellationToken);
}
public Task<T> DeleteAsync<T>(string name, string? namespaceParameter = null, CancellationToken cancellationToken = default)
where T : CustomResource, IKubernetesStaticMetadata
{
ObjectDisposedException.ThrowIf(_disposed, this);
var resourceType = GetResourceFor<T>();
return ExecuteWithRetry(
DcpApiOperationType.Delete,
T.ObjectKind,
async (kubernetes, operationCancellationToken) =>
{
var response = string.IsNullOrEmpty(namespaceParameter)
? await kubernetes.CustomObjects.DeleteClusterCustomObjectWithHttpMessagesAsync(
GroupVersion.Group,
GroupVersion.Version,
resourceType,
name,
cancellationToken: operationCancellationToken).ConfigureAwait(false)
: await kubernetes.CustomObjects.DeleteNamespacedCustomObjectWithHttpMessagesAsync(
GroupVersion.Group,
GroupVersion.Version,
namespaceParameter,
resourceType,
name,
cancellationToken: operationCancellationToken).ConfigureAwait(false);
return KubernetesJson.Deserialize<T>(response.Body.ToString());
},
RetryOnConnectivityAndConflictErrors,
cancellationToken);
}
public async IAsyncEnumerable<(WatchEventType, T)> WatchAsync<T>(
string? namespaceParameter = null,
[EnumeratorCancellation] CancellationToken cancellationToken = default)
where T : CustomResource, IKubernetesStaticMetadata
{
ObjectDisposedException.ThrowIf(_disposed, this);
var resourceType = GetResourceFor<T>();
using var watchCts = CancellationTokenSource.CreateLinkedTokenSource(cancellationToken);
var channel = Channel.CreateUnbounded<(WatchEventType, T)>(new UnboundedChannelOptions
{
SingleReader = true,
SingleWriter = true,
AllowSynchronousContinuations = false
});
var innerWatchFactory = ((WatchEventType, T)? lastValue, CancellationToken restartCancellationToken) =>
{
return ExecuteWithRetry(
DcpApiOperationType.Watch,
T.ObjectKind,
async (kubernetes, operationCancellationToken) =>
{
// The per-attempt budget covers response headers, not the lifetime of the
// stream. Its owner uses the restart token after the attempt has completed.
return await kubernetes.CreateWatchAsync<T>(
GroupVersion.Group, GroupVersion.Version, resourceType,
namespaceParameter, restartCancellationToken,
operationCancellationToken).ConfigureAwait(false);
},
RetryOnConnectivityAndConflictErrors,
restartCancellationToken,
timeoutEachAttempt: true);
};
// Drain DCP independently of resource handlers. Waiting on the resource watcher's output
// semaphore, lifecycle callbacks, or synchronous log subscribers must not leave HTTP/2
// flow control blocking DCP mutations. Keep every event, including intermediate states.
var watchTask = Task.Run(async () =>
{
Exception? error = null;
try
{
await foreach (var item in PeriodicRestartAsyncEnumerable.CreateAsync(
innerWatchFactory, WatchRestartInterval, watchCts.Token).ConfigureAwait(false))
{
channel.Writer.TryWrite(item);
}
}
catch (OperationCanceledException) when (watchCts.IsCancellationRequested)
{
// The consumer canceled or disposed its enumeration.
}
catch (Exception ex)
{
error = ex;
}
finally
{
channel.Writer.TryComplete(error);
}
}, CancellationToken.None);
try
{
await foreach (var item in channel.Reader.ReadAllAsync(cancellationToken).ConfigureAwait(false))
{
cancellationToken.ThrowIfCancellationRequested();
yield return item;
}
}
finally
{
await watchCts.CancelAsync().ConfigureAwait(false);
await watchTask.ConfigureAwait(false);
}
}
public Task<Stream> GetLogStreamAsync<T>(
T obj,
string logStreamType,
CancellationToken cancellationToken = default,
bool? follow = true,
bool? timestamps = false,
bool? lineNumbers = false,
long? limit = null,
long? tail = null,
long? skip = null) where T : CustomResource, IKubernetesStaticMetadata
{
ObjectDisposedException.ThrowIf(_disposed, this);
var resourceType = GetResourceFor<T>();
List<(string name, string value)> queryParams = [
(name: "follow", value: follow == true ? "true": "false"),
(name: "timestamps", value: timestamps == true ? "true" : "false"),
(name: "source", value: logStreamType),
(name: "line_numbers", value: lineNumbers == true ? "true" : "false"),
];
if (limit.HasValue)
{
ArgumentOutOfRangeException.ThrowIfLessThan(limit.Value, 1, nameof(limit));
queryParams.Add((name: "limit", value: limit.Value.ToString(CultureInfo.InvariantCulture)));
}
if (tail.HasValue)
{
ArgumentOutOfRangeException.ThrowIfLessThan(tail.Value, 1, nameof(tail));
queryParams.Add((name: "tail", value: tail.Value.ToString(CultureInfo.InvariantCulture)));
}
if (skip.HasValue)
{
ArgumentOutOfRangeException.ThrowIfLessThan(skip.Value, 1, nameof(skip));
queryParams.Add((name: "skip", value: skip.Value.ToString(CultureInfo.InvariantCulture)));
}
return ExecuteWithRetry(
DcpApiOperationType.GetLogSubresource,
T.ObjectKind,
async (kubernetes, operationCancellationToken) =>
{
var response = await kubernetes.ReadSubResourceAsStreamAsync(
GroupVersion.Group,
GroupVersion.Version,
resourceType,
obj.Metadata.Name,
Logs.SubResourceName,
obj.Metadata.Namespace(),
queryParams,
operationCancellationToken
).ConfigureAwait(false);
return response.Body;
},
RetryOnConnectivityAndConflictErrors,
cancellationToken
);
}
public Task StopServerAsync(string resourceCleanup = ResourceCleanup.Full, CancellationToken cancellationToken = default)
{
ObjectDisposedException.ThrowIf(_disposed, this);
return ExecuteWithRetry(
DcpApiOperationType.ServerStop,
DcpExecutionResourceType,
async (kubernetes, operationCancellationToken) =>
{
await kubernetes.PatchExecutionDocumentAsync(
new ApiServerExecution
{
ApiServerStatus = ApiServerStatus.Stopping,
ShutdownResourceCleanup = ResourceCleanup.Full
},
operationCancellationToken
).ConfigureAwait(false);
return (object?)null;
},
RetryOnConnectivityErrors,
cancellationToken
);
}
public async Task CleanupResourcesAsync(CancellationToken cancellationToken = default)
{
ObjectDisposedException.ThrowIf(_disposed, this);
var executionDoc = await ExecuteWithRetry(
DcpApiOperationType.ResourceCleanup,
DcpExecutionResourceType,
async (kubernetes, operationCancellationToken) =>
{
return await kubernetes.PatchExecutionDocumentAsync(
new ApiServerExecution
{
ApiServerStatus = ApiServerStatus.CleaningResources,
ShutdownResourceCleanup = ResourceCleanup.Full
},
operationCancellationToken
).ConfigureAwait(false);
},
RetryOnConnectivityErrors,
cancellationToken
).ConfigureAwait(false);
if (executionDoc.ResourcesCleanedUp)
{
return;
}
var retryPipeline = new ResiliencePipelineBuilder<ApiServerExecution> { TimeProvider = timeProvider }
.AddRetry(new RetryStrategyOptions<ApiServerExecution>()
{
ShouldHandle = new PredicateBuilder<ApiServerExecution>()
.Handle<Exception>(RetryOnConnectivityErrors)
.HandleResult(executionDoc => !executionDoc.ResourcesCleanedUp),
BackoffType = DelayBackoffType.Constant,
MaxRetryAttempts = int.MaxValue,
Delay = s_initialRetryDelay,
MaxDelay = TimeSpan.FromSeconds(1),
})
.Build();
await retryPipeline.ExecuteAsync(async cancellationContext =>
{
return await ExecuteWithRetry(
DcpApiOperationType.ResourceCleanup,
DcpExecutionResourceType,
(kubernetes, operationCancellationToken) => kubernetes.GetExecutionDocumentAsync(operationCancellationToken),
RetryOnConnectivityErrors,
cancellationContext).ConfigureAwait(false);
}, cancellationToken).ConfigureAwait(false);
}
public void Dispose()
{
_disposed = true;
_kubeconfigReadSemaphore?.Dispose();
_kubernetes?.Dispose();
}
private static string GetResourceFor<T>() where T : CustomResource
{
if (!Model.Dcp.Schema.TryGet<T>(out var kindWithResource))
{
throw new InvalidOperationException($"Unknown custom resource type: {typeof(T).Name}");
}
return kindWithResource.Resource;
}
private async Task<T> CreateOnceAsync<T>(
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<T>(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<TResult> ExecuteWithRetry<TResult>(
DcpApiOperationType operationType,
string resourceType,
Func<DcpKubernetesClient, CancellationToken, Task<TResult>> operation,
Func<Exception, bool> isRetryable,
CancellationToken cancellationToken,
bool timeoutEachAttempt = false)
{
using var activity = ProfilingTelemetry.StartDcpKubernetesApi(configuration, operationType, resourceType);
var retryCount = 0;
try
{
var resiliencePipeline = CreateKubernetesCallResiliencePipeline(
isRetryable,
activity,
() => retryCount++,
timeoutEachAttempt);
return await resiliencePipeline.ExecuteAsync(async (cancellationToken) =>
{
// 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);
return result;
}, cancellationToken).ConfigureAwait(false);
}
finally
{
activity.SetDcpApiRetryCount(retryCount);
}
}
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 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 timeProvider.GetUtcNow() + (delay > recoveryTimeout ? recoveryTimeout : delay);
}
return null;
}
private async Task<TimeSpan> 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 - timeProvider.GetUtcNow()).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, timeProvider, 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 ResiliencePipeline CreateKubernetesCallResiliencePipeline(
Func<Exception, bool> isRetryable,
ProfilingTelemetry.ActivityScope activity,
Action recordRetry,
bool timeoutEachAttempt = false)
{
var timeoutOptions = new TimeoutStrategyOptions
{
// 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<TimeSpan>(timeout);
},
OnTimeout = (_) =>
{
activity.AddKubernetesApiTimeout();
return ValueTask.CompletedTask;
}
};
var retryOptions = new RetryStrategyOptions
{
ShouldHandle = timeoutEachAttempt
? new PredicateBuilder().Handle(isRetryable).Handle<TimeoutRejectedException>()
: new PredicateBuilder().Handle(isRetryable),
BackoffType = DelayBackoffType.Exponential,
MaxRetryAttempts = int.MaxValue,
Delay = s_initialRetryDelay,
MaxDelay = TimeSpan.FromSeconds(5),
OnRetry = (retry) =>
{
recordRetry();
activity.AddKubernetesApiRetry(retry.AttemptNumber, retry.RetryDelay, retry.Outcome.Exception);
return ValueTask.CompletedTask;
}
};
// Regular API calls have one total retry budget. Watches instead retry a stalled connection
// after each timeout until their outer periodic-restart token is canceled.
return timeoutEachAttempt
? new ResiliencePipelineBuilder { TimeProvider = timeProvider }.AddRetry(retryOptions).AddTimeout(timeoutOptions).Build()
: new ResiliencePipelineBuilder { TimeProvider = timeProvider }.AddTimeout(timeoutOptions).AddRetry(retryOptions).Build();
}
private ResiliencePipeline CreateReadKubeconfigResiliencePipeline()
{
if (_resiliencePipeline == null)
{
var configurationReadRetry = new RetryStrategyOptions()
{
// Handle exceptions caused by races between writing and reading the configuration file.
// If the file is loaded while it is still being written, this can result in a YamlException being thrown.
ShouldHandle = new PredicateBuilder().Handle<KubeConfigException>().Handle<YamlException>().Handle<IOException>(),
BackoffType = DelayBackoffType.Constant,
MaxRetryAttempts = dcpOptions.Value.KubernetesConfigReadRetryCount,
MaxDelay = TimeSpan.FromMilliseconds(dcpOptions.Value.KubernetesConfigReadRetryIntervalMilliseconds),
OnRetry = (retry) =>
{
logger.LogDebug(
"Waiting for Kubernetes configuration file at '{DcpKubeconfigPath}' (attempt {Iteration}).",
locations.DcpKubeconfigPath,
retry.AttemptNumber
);
return ValueTask.CompletedTask;
}
};
_resiliencePipeline = new ResiliencePipelineBuilder { TimeProvider = timeProvider }.AddRetry(configurationReadRetry).Build();
}
return _resiliencePipeline;
}
private async Task<KubernetesClientReady> EnsureKubernetesAsync(CancellationToken cancellationToken = default)
{
ObjectDisposedException.ThrowIf(_disposed, this);
// Return early before waiting for the semaphore if we can.
if (_kubernetes != null)
{
return new KubernetesClientReady(WaitMilliseconds: 0, Initialized: false);
}
var lockWaitStopwatch = Stopwatch.StartNew();
await _kubeconfigReadSemaphore.WaitAsync(-1, cancellationToken).ConfigureAwait(false);
lockWaitStopwatch.Stop();
var lockWaitMilliseconds = lockWaitStopwatch.ElapsedMilliseconds;
try
{
// Second chance shortcut if multiple threads got caught.
if (_kubernetes != null)
{
return new KubernetesClientReady(lockWaitMilliseconds, Initialized: false);
}
using var activity = ProfilingTelemetry.StartDcpEnsureKubernetesClient(configuration, File.Exists(locations.DcpKubeconfigPath));
activity.SetDcpKubeconfigLockWait(lockWaitMilliseconds);
activity.AddKubeconfigLockAcquired();
// We retry reading the kubeconfig file because DCP takes a few moments to write
// it to disk. This retry pipeline will only be invoked by a single thread the
// rest will be held at the semaphore.
var readStopwatch = new Stopwatch();
readStopwatch.Start();
var readPipeline = CreateReadKubeconfigResiliencePipeline();
try
{
_kubernetes = await readPipeline.ExecuteAsync(async (cancellationToken) =>
{
var fileWaitStopwatch = Stopwatch.StartNew();
var fileInfo = new FileInfo(locations.DcpKubeconfigPath);
while (!fileInfo.Exists)
{
await Task.Delay(TimeSpan.FromMilliseconds(dcpOptions.Value.KubernetesConfigReadRetryIntervalMilliseconds), timeProvider, 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)
{
activity.SetError(ex);
throw;
}
activity.AddKubernetesClientCreated();
return new KubernetesClientReady(lockWaitMilliseconds, Initialized: true);
}
finally
{
_kubeconfigReadSemaphore.Release();
}
}
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);
}