kubernetes-client/csharp
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat: enhance Kubernetes client with watch functionality (#1667)

Browse Source 
* feat: enhance Kubernetes client with watch functionality

* refactor: simplify watch event handling in Kubernetes client example

* refactor: update Kubernetes watch functionality to use new event handling methods and add async enumerable support

* fix

* fix

* fix: correct usage of Pod list items in client example and update Obsolete attribute formatting

* fix: update client example to use correct Pod list method and improve Obsolete attribute formatting

* refactor: enhance type resolution for list items in TypeHelper by adding TryGetItemTypeFromSchema method

* feat: mark Watch methods as obsolete to prepare for future deprecation

* fix

* refactor: update WatcherExt class to internal and remove obsolete attributes; improve example method signature in Program.cs

* refactor: change WatcherExt class from internal to public and mark methods as obsolete for future deprecation
This commit is contained in:
Qing Long
2025-10-12 06:10:53 +08:00
committed by GitHub
parent ca5d9f4793
commit 6d27bd900b
9 changed files with 647 additions and 90 deletions
Download Patch File Download Diff File Expand all files Collapse all files

13
examples/clientset/Program.cs
View File

@@ -1,7 +1,6 @@
// See https://aka.ms/new-console-template for more information
using k8s;
using k8s.ClientSets;
using k8s;
using k8s.Models;
using k8s.ClientSets;
using System.Threading.Tasks;
namespace clientset
@@ -13,7 +12,7 @@ namespace clientset
var config = KubernetesClientConfiguration.BuildConfigFromConfigFile();
var client = new Kubernetes(config);
ClientSet clientSet = new ClientSet(client);
var clientSet = new ClientSet(client);
var list = await clientSet.CoreV1.Pod.ListAsync("default").ConfigureAwait(false);
foreach (var item in list)
{
@@ -22,6 +21,12 @@ namespace clientset
var pod = await clientSet.CoreV1.Pod.GetAsync("test", "default").ConfigureAwait(false);
System.Console.WriteLine(pod?.Metadata?.Name);
var watch = clientSet.CoreV1.Pod.WatchListAsync("default");
await foreach (var (_, item) in watch.ConfigureAwait(false))
{
System.Console.WriteLine(item.Metadata.Name);
}
}
}
}

9
examples/watch/Program.cs
View File

@@ -1,5 +1,4 @@
using k8s;
using k8s.Models;
using System;
using System.Threading;
using System.Threading.Tasks;
@@ -8,9 +7,10 @@ var config = KubernetesClientConfiguration.BuildConfigFromConfigFile();
IKubernetes client = new Kubernetes(config);
var podlistResp = client.CoreV1.ListNamespacedPodWithHttpMessagesAsync("default", watch: true);
var podlistResp = client.CoreV1.WatchListNamespacedPodAsync("default");
// C# 8 required https://docs.microsoft.com/en-us/archive/msdn-magazine/2019/november/csharp-iterating-with-async-enumerables-in-csharp-8
await foreach (var (type, item) in podlistResp.WatchAsync<V1Pod, V1PodList>().ConfigureAwait(false))
await foreach (var (type, item) in podlistResp.ConfigureAwait(false))
{
Console.WriteLine("==on watch event==");
Console.WriteLine(type);
@@ -22,8 +22,7 @@ await foreach (var (type, item) in podlistResp.WatchAsync<V1Pod, V1PodList>().Co
void WatchUsingCallback(IKubernetes client)
#pragma warning restore CS8321 // Remove unused private members
{
var podlistResp = client.CoreV1.ListNamespacedPodWithHttpMessagesAsync("default", watch: true);
using (podlistResp.Watch<V1Pod, V1PodList>((type, item) =>
using (var podlistResp = client.CoreV1.WatchListNamespacedPod("default", onEvent: (type, item) =>
{
Console.WriteLine("==on watch event==");
Console.WriteLine(type);

3
src/KubernetesClient/WatcherExt.cs
View File

@@ -16,6 +16,7 @@ namespace k8s
/// The action to invoke when the server closes the connection.
/// </param>
/// <returns>a watch object</returns>
[Obsolete("This method will be deprecated in future versions. Please use the Watch method instead.")]
public static Watcher<T> Watch<T, L>(
this Task<HttpOperationResponse<L>> responseTask,
Action<WatchEventType, T> onEvent,
@@ -52,6 +53,7 @@ namespace k8s
/// The action to invoke when the server closes the connection.
/// </param>
/// <returns>a watch object</returns>
[Obsolete("This method will be deprecated in future versions. Please use the Watch method instead.")]
public static Watcher<T> Watch<T, L>(
this HttpOperationResponse<L> response,
Action<WatchEventType, T> onEvent,
@@ -71,6 +73,7 @@ namespace k8s
/// <param name="onError">a callback when any exception was caught during watching</param>
/// <param name="cancellationToken">cancellation token</param>
/// <returns>IAsyncEnumerable of watch events</returns>
[Obsolete("This method will be deprecated in future versions. Please use the WatchAsync method instead.")]
public static IAsyncEnumerable<(WatchEventType, T)> WatchAsync<T, L>(
this Task<HttpOperationResponse<L>> responseTask,
Action<Exception> onError = null,

20
src/LibKubernetesGenerator/ParamHelper.cs
View File

@@ -3,6 +3,7 @@ using NSwag;
using Scriban.Runtime;
using System;
using System.Linq;
using System.Collections.Generic;
namespace LibKubernetesGenerator
{
@@ -21,6 +22,8 @@ namespace LibKubernetesGenerator
{
scriptObject.Import(nameof(GetModelCtorParam), new Func<JsonSchema, string>(GetModelCtorParam));
scriptObject.Import(nameof(IfParamContains), IfParamContains);
scriptObject.Import(nameof(FilterParameters), FilterParameters);
scriptObject.Import(nameof(GetParameterValueForWatch), new Func<OpenApiParameter, bool, string, string>(GetParameterValueForWatch));
}
public static bool IfParamContains(OpenApiOperation operation, string name)
@@ -39,6 +42,23 @@ namespace LibKubernetesGenerator
return found;
}
public static IEnumerable<OpenApiParameter> FilterParameters(OpenApiOperation operation, string excludeParam)
{
return operation.Parameters.Where(p => p.Name != excludeParam);
}
public string GetParameterValueForWatch(OpenApiParameter parameter, bool watch, string init = "false")
{
if (parameter.Name == "watch")
{
return watch ? "true" : "false";
}
else
{
return generalNameHelper.GetDotNetNameOpenApiParameter(parameter, init);
}
}
public string GetModelCtorParam(JsonSchema schema)
{
return string.Join(", ", schema.Properties.Values

32
src/LibKubernetesGenerator/TypeHelper.cs
View File

@@ -122,7 +122,6 @@ namespace LibKubernetesGenerator
return $"IDictionary<string, {GetDotNetType(schema.AdditionalPropertiesSchema, parent)}>";
}
if (schema?.Reference != null)
{
return classNameHelper.GetClassNameForSchemaDefinition(schema.Reference);
@@ -245,6 +244,16 @@ namespace LibKubernetesGenerator
}
break;
case "T":
var itemType = TryGetItemTypeFromSchema(response);
if (itemType != null)
{
return itemType;
}
break;
case "TList":
return t;
}
return t;
@@ -283,5 +292,26 @@ namespace LibKubernetesGenerator
return false;
}
private string TryGetItemTypeFromSchema(OpenApiResponse response)
{
var listSchema = response?.Schema?.Reference;
if (listSchema?.Properties?.TryGetValue("items", out var itemsProperty) != true)
{
return null;
}
if (itemsProperty.Reference != null)
{
return classNameHelper.GetClassNameForSchemaDefinition(itemsProperty.Reference);
}
if (itemsProperty.Item?.Reference != null)
{
return classNameHelper.GetClassNameForSchemaDefinition(itemsProperty.Item.Reference);
}
return null;
}
}
}

81
src/LibKubernetesGenerator/templates/Client.cs.template
View File

@@ -17,10 +17,11 @@ public partial class {{name}}Client : ResourceClient
}
{{for api in apis }}
{{~ $filteredParams = FilterParameters api.operation "watch" ~}}
/// <summary>
/// {{ToXmlDoc api.operation.description}}
/// </summary>
{{ for parameter in api.operation.parameters}}
{{ for parameter in $filteredParams}}
/// <param name="{{GetDotNetNameOpenApiParameter parameter "false"}}">
/// {{ToXmlDoc parameter.description}}
/// </param>
@@ -29,7 +30,7 @@ public partial class {{name}}Client : ResourceClient
/// A <see cref="CancellationToken"/> which can be used to cancel the asynchronous operation.
/// </param>
public async Task{{GetReturnType api.operation "<>"}} {{GetActionName api.operation name "Async"}}(
{{ for parameter in api.operation.parameters}}
{{ for parameter in $filteredParams}}
{{GetDotNetTypeOpenApiParameter parameter}} {{GetDotNetNameOpenApiParameter parameter "true"}},
{{ end }}
CancellationToken cancellationToken = default(CancellationToken))
@@ -37,7 +38,7 @@ public partial class {{name}}Client : ResourceClient
{{if IfReturnType api.operation "stream"}}
var _result = await Client.{{group}}.{{GetOperationId api.operation "WithHttpMessagesAsync"}}(
{{ for parameter in api.operation.parameters}}
{{GetDotNetNameOpenApiParameter parameter "false"}},
{{GetParameterValueForWatch parameter false}},
{{end}}
null,
cancellationToken);
@@ -47,7 +48,7 @@ public partial class {{name}}Client : ResourceClient
{{if IfReturnType api.operation "obj"}}
using (var _result = await Client.{{group}}.{{GetOperationId api.operation "WithHttpMessagesAsync"}}(
{{ for parameter in api.operation.parameters}}
{{GetDotNetNameOpenApiParameter parameter "false"}},
{{GetParameterValueForWatch parameter false}},
{{end}}
null,
cancellationToken).ConfigureAwait(false))
@@ -58,7 +59,7 @@ public partial class {{name}}Client : ResourceClient
{{if IfReturnType api.operation "void"}}
using (var _result = await Client.{{group}}.{{GetOperationId api.operation "WithHttpMessagesAsync"}}(
{{ for parameter in api.operation.parameters}}
{{GetDotNetNameOpenApiParameter parameter "false"}},
{{GetParameterValueForWatch parameter false}},
{{end}}
null,
cancellationToken).ConfigureAwait(false))
@@ -71,7 +72,7 @@ public partial class {{name}}Client : ResourceClient
/// <summary>
/// {{ToXmlDoc api.operation.description}}
/// </summary>
{{ for parameter in api.operation.parameters}}
{{ for parameter in $filteredParams}}
/// <param name="{{GetDotNetNameOpenApiParameter parameter "false"}}">
/// {{ToXmlDoc parameter.description}}
/// </param>
@@ -80,14 +81,14 @@ public partial class {{name}}Client : ResourceClient
/// A <see cref="CancellationToken"/> which can be used to cancel the asynchronous operation.
/// </param>
public async Task<T> {{GetActionName api.operation name "Async"}}<T>(
{{ for parameter in api.operation.parameters}}
{{ for parameter in $filteredParams}}
{{GetDotNetTypeOpenApiParameter parameter}} {{GetDotNetNameOpenApiParameter parameter "false"}},
{{ end }}
CancellationToken cancellationToken = default(CancellationToken))
{
using (var _result = await Client.{{group}}.{{GetOperationId api.operation "WithHttpMessagesAsync"}}<T>(
{{ for parameter in api.operation.parameters}}
{{GetDotNetNameOpenApiParameter parameter "false"}},
{{GetParameterValueForWatch parameter false}},
{{end}}
null,
cancellationToken).ConfigureAwait(false))
@@ -96,5 +97,69 @@ public partial class {{name}}Client : ResourceClient
}
}
{{end}}
#if !K8S_AOT
{{if IfParamContains api.operation "watch"}}
/// <summary>
/// Watch {{ToXmlDoc api.operation.description}}
/// </summary>
{{ for parameter in $filteredParams}}
/// <param name="{{GetDotNetNameOpenApiParameter parameter "false"}}">
/// {{ToXmlDoc parameter.description}}
/// </param>
{{ end }}
/// <param name="onEvent">Callback when any event raised from api server</param>
/// <param name="onError">Callback when any exception was caught during watching</param>
/// <param name="onClosed">Callback when the server closes the connection</param>
public Watcher<{{GetReturnType api.operation "T"}}> Watch{{GetActionName api.operation name ""}}(
{{ for parameter in $filteredParams}}
{{GetDotNetTypeOpenApiParameter parameter}} {{GetDotNetNameOpenApiParameter parameter "true"}},
{{ end }}
Action<WatchEventType, {{GetReturnType api.operation "T"}}> onEvent = null,
Action<Exception> onError = null,
Action onClosed = null)
{
if (onEvent == null) throw new ArgumentNullException(nameof(onEvent));
var responseTask = Client.{{group}}.{{GetOperationId api.operation "WithHttpMessagesAsync"}}(
{{ for parameter in api.operation.parameters}}
{{GetParameterValueForWatch parameter true}},
{{ end }}
null,
CancellationToken.None);
return responseTask.Watch<{{GetReturnType api.operation "T"}}, {{GetReturnType api.operation "TList"}}>(
onEvent, onError, onClosed);
}
/// <summary>
/// Watch {{ToXmlDoc api.operation.description}} as async enumerable
/// </summary>
{{ for parameter in $filteredParams}}
/// <param name="{{GetDotNetNameOpenApiParameter parameter "false"}}">
/// {{ToXmlDoc parameter.description}}
/// </param>
{{ end }}
/// <param name="onError">Callback when any exception was caught during watching</param>
/// <param name="cancellationToken">Cancellation token</param>
public IAsyncEnumerable<(WatchEventType, {{GetReturnType api.operation "T"}})> Watch{{GetActionName api.operation name "Async"}}(
{{ for parameter in $filteredParams}}
{{GetDotNetTypeOpenApiParameter parameter}} {{GetDotNetNameOpenApiParameter parameter "true"}},
{{ end }}
Action<Exception> onError = null,
CancellationToken cancellationToken = default)
{
var responseTask = Client.{{group}}.{{GetOperationId api.operation "WithHttpMessagesAsync"}}(
{{ for parameter in api.operation.parameters}}
{{GetParameterValueForWatch parameter true}},
{{ end }}
null,
cancellationToken);
return responseTask.WatchAsync<{{GetReturnType api.operation "T"}}, {{GetReturnType api.operation "TList"}}>(
onError, cancellationToken);
}
{{end}}
#endif
{{end}}
}

103
src/LibKubernetesGenerator/templates/OperationsExtensions.cs.template
View File

@@ -12,26 +12,27 @@ namespace k8s;
public static partial class {{name}}OperationsExtensions
{
{{for api in apis }}
{{~ $filteredParams = FilterParameters api.operation "watch" ~}}
/// <summary>
/// {{ToXmlDoc api.operation.description}}
/// </summary>
/// <param name='operations'>
/// The operations group for this extension method.
/// </param>
{{ for parameter in api.operation.parameters}}
{{ for parameter in $filteredParams}}
/// <param name="{{GetDotNetNameOpenApiParameter parameter "false"}}">
/// {{ToXmlDoc api.description}}
/// {{ToXmlDoc parameter.description}}
/// </param>
{{ end }}
public static {{GetReturnType api.operation "void"}} {{GetOperationId api.operation ""}}(
this I{{name}}Operations operations
{{ for parameter in api.operation.parameters}}
{{ for parameter in $filteredParams}}
,{{GetDotNetTypeOpenApiParameter parameter}} {{GetDotNetNameOpenApiParameter parameter "true"}}
{{end}}
)
{
{{GetReturnType api.operation "return"}} operations.{{GetOperationId api.operation "Async"}}(
{{ for parameter in api.operation.parameters}}
{{ for parameter in $filteredParams}}
{{GetDotNetNameOpenApiParameter parameter "false"}},
{{end}}
CancellationToken.None
@@ -45,20 +46,20 @@ public static partial class {{name}}OperationsExtensions
/// <param name='operations'>
/// The operations group for this extension method.
/// </param>
{{ for parameter in api.operation.parameters}}
{{ for parameter in $filteredParams}}
/// <param name="{{GetDotNetNameOpenApiParameter parameter "false"}}">
/// {{ToXmlDoc parameter.description}}
/// </param>
{{end}}
public static T {{GetOperationId api.operation ""}}<T>(
this I{{name}}Operations operations
{{ for parameter in api.operation.parameters}}
{{ for parameter in $filteredParams}}
,{{GetDotNetTypeOpenApiParameter parameter}} {{GetDotNetNameOpenApiParameter parameter "true"}}
{{end}}
)
{
return operations.{{GetOperationId api.operation "Async"}}<T>(
{{ for parameter in api.operation.parameters}}
{{ for parameter in $filteredParams}}
{{GetDotNetNameOpenApiParameter parameter "false"}},
{{end}}
CancellationToken.None
@@ -72,7 +73,7 @@ public static partial class {{name}}OperationsExtensions
/// <param name='operations'>
/// The operations group for this extension method.
/// </param>
{{ for parameter in api.operation.parameters}}
{{ for parameter in $filteredParams}}
/// <param name="{{GetDotNetNameOpenApiParameter parameter "false"}}">
/// {{ToXmlDoc parameter.description}}
/// </param>
@@ -82,7 +83,7 @@ public static partial class {{name}}OperationsExtensions
/// </param>
public static async Task{{GetReturnType api.operation "<>"}} {{GetOperationId api.operation "Async"}}(
this I{{name}}Operations operations,
{{ for parameter in api.operation.parameters}}
{{ for parameter in $filteredParams}}
{{GetDotNetTypeOpenApiParameter parameter}} {{GetDotNetNameOpenApiParameter parameter "true"}},
{{ end }}
CancellationToken cancellationToken = default(CancellationToken))
@@ -90,7 +91,7 @@ public static partial class {{name}}OperationsExtensions
{{if IfReturnType api.operation "stream"}}
var _result = await operations.{{GetOperationId api.operation "WithHttpMessagesAsync"}}(
{{ for parameter in api.operation.parameters}}
{{GetDotNetNameOpenApiParameter parameter "false"}},
{{GetParameterValueForWatch parameter false}},
{{end}}
null,
cancellationToken);
@@ -100,7 +101,7 @@ public static partial class {{name}}OperationsExtensions
{{if IfReturnType api.operation "obj"}}
using (var _result = await operations.{{GetOperationId api.operation "WithHttpMessagesAsync"}}(
{{ for parameter in api.operation.parameters}}
{{GetDotNetNameOpenApiParameter parameter "false"}},
{{GetParameterValueForWatch parameter false}},
{{end}}
null,
cancellationToken).ConfigureAwait(false))
@@ -111,7 +112,7 @@ public static partial class {{name}}OperationsExtensions
{{if IfReturnType api.operation "void"}}
using (var _result = await operations.{{GetOperationId api.operation "WithHttpMessagesAsync"}}(
{{ for parameter in api.operation.parameters}}
{{GetDotNetNameOpenApiParameter parameter "false"}},
{{GetParameterValueForWatch parameter false}},
{{end}}
null,
cancellationToken).ConfigureAwait(false))
@@ -127,7 +128,7 @@ public static partial class {{name}}OperationsExtensions
/// <param name='operations'>
/// The operations group for this extension method.
/// </param>
{{ for parameter in api.operation.parameters}}
{{ for parameter in $filteredParams}}
/// <param name="{{GetDotNetNameOpenApiParameter parameter "false"}}">
/// {{ToXmlDoc parameter.description}}
/// </param>
@@ -137,14 +138,14 @@ public static partial class {{name}}OperationsExtensions
/// </param>
public static async Task<T> {{GetOperationId api.operation "Async"}}<T>(
this I{{name}}Operations operations,
{{ for parameter in api.operation.parameters}}
{{ for parameter in $filteredParams}}
{{GetDotNetTypeOpenApiParameter parameter}} {{GetDotNetNameOpenApiParameter parameter "true"}},
{{ end }}
CancellationToken cancellationToken = default(CancellationToken))
{
using (var _result = await operations.{{GetOperationId api.operation "WithHttpMessagesAsync"}}<T>(
{{ for parameter in api.operation.parameters}}
{{GetDotNetNameOpenApiParameter parameter "false"}},
{{GetParameterValueForWatch parameter false}},
{{end}}
null,
cancellationToken).ConfigureAwait(false))
@@ -154,5 +155,77 @@ public static partial class {{name}}OperationsExtensions
}
{{end}}
#if !K8S_AOT
{{if IfParamContains api.operation "watch"}}
{{~ $filteredParams = FilterParameters api.operation "watch" ~}}
/// <summary>
/// Watch {{ToXmlDoc api.operation.description}}
/// </summary>
/// <param name='operations'>
/// The operations group for this extension method.
/// </param>
{{ for parameter in $filteredParams}}
/// <param name="{{GetDotNetNameOpenApiParameter parameter "false"}}">
/// {{ToXmlDoc parameter.description}}
/// </param>
{{ end }}
/// <param name="onEvent">Callback when any event raised from api server</param>
/// <param name="onError">Callback when any exception was caught during watching</param>
/// <param name="onClosed">Callback when the server closes the connection</param>
public static Watcher<{{GetReturnType api.operation "T"}}> Watch{{GetOperationId api.operation ""}}(
this I{{name}}Operations operations,
{{ for parameter in $filteredParams}}
{{GetDotNetTypeOpenApiParameter parameter}} {{GetDotNetNameOpenApiParameter parameter "true"}},
{{end}}
Action<WatchEventType, {{GetReturnType api.operation "T"}}> onEvent = null,
Action<Exception> onError = null,
Action onClosed = null)
{
if (onEvent == null) throw new ArgumentNullException(nameof(onEvent));
var responseTask = operations.{{GetOperationId api.operation "WithHttpMessagesAsync"}}(
{{ for parameter in api.operation.parameters}}
{{GetParameterValueForWatch parameter true}},
{{end}}
null,
CancellationToken.None);
return responseTask.Watch<{{GetReturnType api.operation "T"}}, {{GetReturnType api.operation "TList"}}>(
onEvent, onError, onClosed);
}
/// <summary>
/// Watch {{ToXmlDoc api.operation.description}} as async enumerable
/// </summary>
/// <param name='operations'>
/// The operations group for this extension method.
/// </param>
{{ for parameter in $filteredParams}}
/// <param name="{{GetDotNetNameOpenApiParameter parameter "false"}}">
/// {{ToXmlDoc parameter.description}}
/// </param>
{{ end }}
/// <param name="onError">Callback when any exception was caught during watching</param>
/// <param name="cancellationToken">Cancellation token</param>
public static IAsyncEnumerable<(WatchEventType, {{GetReturnType api.operation "T"}})> Watch{{GetOperationId api.operation "Async"}}(
this I{{name}}Operations operations,
{{ for parameter in $filteredParams}}
{{GetDotNetTypeOpenApiParameter parameter}} {{GetDotNetNameOpenApiParameter parameter "true"}},
{{end}}
Action<Exception> onError = null,
CancellationToken cancellationToken = default)
{
var responseTask = operations.{{GetOperationId api.operation "WithHttpMessagesAsync"}}(
{{ for parameter in api.operation.parameters}}
{{GetParameterValueForWatch parameter true}},
{{end}}
null,
cancellationToken);
return responseTask.WatchAsync<{{GetReturnType api.operation "T"}}, {{GetReturnType api.operation "TList"}}>(
onError, cancellationToken);
}
{{end}}
#endif
{{end}}
}

85
tests/E2E.Tests/MinikubeTests.cs
View File

@@ -224,13 +224,12 @@ namespace k8s.E2E
var started = new AsyncManualResetEvent();
var connectionClosed = new AsyncManualResetEvent();
var watcher = kubernetes.BatchV1.ListNamespacedJobWithHttpMessagesAsync(
var watcher = kubernetes.BatchV1.WatchListNamespacedJob(
job.Metadata.NamespaceProperty,
fieldSelector: $"metadata.name={job.Metadata.Name}",
resourceVersion: job.Metadata.ResourceVersion,
timeoutSeconds: 30,
watch: true).Watch<V1Job, V1JobList>(
(type, source) =>
onEvent: (type, source) =>
{
Debug.WriteLine($"Watcher 1: {type}, {source}");
events.Add(new Tuple<WatchEventType, V1Job>(type, source));
@@ -250,6 +249,86 @@ namespace k8s.E2E
new V1DeleteOptions() { PropagationPolicy = "Foreground" }).ConfigureAwait(false);
}
[MinikubeFact]
public async Task WatcherIntegrationTestAsyncEnumerable()
{
using var kubernetes = CreateClient();
var job = await kubernetes.BatchV1.CreateNamespacedJobAsync(
new V1Job()
{
ApiVersion = "batch/v1",
Kind = V1Job.KubeKind,
Metadata = new V1ObjectMeta() { Name = nameof(WatcherIntegrationTestAsyncEnumerable).ToLowerInvariant() },
Spec = new V1JobSpec()
{
Template = new V1PodTemplateSpec()
{
Spec = new V1PodSpec()
{
Containers = new List<V1Container>()
{
new V1Container()
{
Image = "ubuntu",
Name = "runner",
Command = new List<string>() { "/bin/bash", "-c", "--" },
Args = new List<string>()
{
"trap : TERM INT; sleep infinity & wait",
},
},
},
RestartPolicy = "Never",
},
},
},
},
"default").ConfigureAwait(false);
var events = new Collection<Tuple<WatchEventType, V1Job>>();
var started = new AsyncManualResetEvent();
var watchCompleted = new AsyncManualResetEvent();
// Start async enumerable watch in background task to mimic callback behavior
var watchTask = Task.Run(async () =>
{
try
{
await foreach (var (type, source) in kubernetes.BatchV1.WatchListNamespacedJobAsync(
job.Metadata.NamespaceProperty,
fieldSelector: $"metadata.name={job.Metadata.Name}",
resourceVersion: job.Metadata.ResourceVersion,
timeoutSeconds: 30).ConfigureAwait(false))
{
Debug.WriteLine($"AsyncEnumerable Watcher: {type}, {source}");
events.Add(new Tuple<WatchEventType, V1Job>(type, source));
job = source;
started.Set();
}
}
catch (Exception ex)
{
Debug.WriteLine($"Watch exception: {ex.GetType().FullName}: {ex.Message}");
}
finally
{
watchCompleted.Set();
}
});
await started.WaitAsync().ConfigureAwait(false);
await Task.WhenAny(watchCompleted.WaitAsync(), Task.Delay(TimeSpan.FromMinutes(3))).ConfigureAwait(false);
Assert.True(watchCompleted.IsSet);
var st = await kubernetes.BatchV1.DeleteNamespacedJobAsync(
job.Metadata.Name,
job.Metadata.NamespaceProperty,
new V1DeleteOptions() { PropagationPolicy = "Foreground" }).ConfigureAwait(false);
}
[MinikubeFact]
public void LeaderIntegrationTest()
{

379
tests/KubernetesClient.Tests/WatchTests.cs
View File

@@ -54,24 +54,29 @@ namespace k8s.Tests
var client = new Kubernetes(new KubernetesClientConfiguration { Host = server.Uri.ToString() });
// did not pass watch param
var listTask = client.CoreV1.ListNamespacedPodWithHttpMessagesAsync("default", watch: true);
var onErrorCalled = false;
using (listTask.Watch<V1Pod, V1PodList>((type, item) => { }, e => { onErrorCalled = true; }))
using (var watcher = client.CoreV1.WatchListNamespacedPod(
"default",
onEvent: (type, item) => { },
onError: e => { onErrorCalled = true; }))
{
await Task.Delay(TimeSpan.FromSeconds(1)).ConfigureAwait(true); // delay for onerror to be called
}
Assert.True(onErrorCalled);
// server did not response line by line
await Assert.ThrowsAnyAsync<Exception>(() =>
{
return client.CoreV1.ListNamespacedPodWithHttpMessagesAsync("default", watch: true);
using (var testWatcher = client.CoreV1.WatchListNamespacedPod(
"default"))
{
return Task.CompletedTask;
}
// this line did not throw
// listTask.Watch<Corev1Pod>((type, item) => { });
// using (var testWatcher = client.CoreV1.WatchListNamespacedPod("default", onEvent: (type, item) => { }))
}).ConfigureAwait(true);
}
}
@@ -93,8 +98,7 @@ namespace k8s.Tests
var client = new Kubernetes(new KubernetesClientConfiguration { Host = server.Uri.ToString() });
var listTask = client.CoreV1.ListNamespacedPodWithHttpMessagesAsync("default", watch: true);
using (listTask.Watch<V1Pod, V1PodList>((type, item) => { eventsReceived.Set(); }))
using (var watcher = client.CoreV1.WatchListNamespacedPod("default", onEvent: (type, item) => { eventsReceived.Set(); }))
{
// here watcher is ready to use, but http server has not responsed yet.
created.Set();
@@ -134,27 +138,26 @@ namespace k8s.Tests
{
var client = new Kubernetes(new KubernetesClientConfiguration { Host = server.Uri.ToString() });
var listTask = await client.CoreV1.ListNamespacedPodWithHttpMessagesAsync("default", watch: true).ConfigureAwait(true);
var events = new HashSet<WatchEventType>();
var errors = 0;
var watcher = listTask.Watch<V1Pod, V1PodList>(
(type, item) =>
var watcher = client.CoreV1.WatchListNamespacedPod(
"default",
onEvent: (type, item) =>
{
testOutput.WriteLine($"Watcher received '{type}' event.");
events.Add(type);
eventsReceived.Signal();
},
error =>
onError: error =>
{
testOutput.WriteLine($"Watcher received '{error.GetType().FullName}' error.");
errors += 1;
eventsReceived.Signal();
},
connectionClosed.Set);
onClosed: connectionClosed.Set);
// wait server yields all events
await Task.WhenAny(eventsReceived.WaitAsync(), Task.Delay(TestTimeout)).ConfigureAwait(true);
@@ -195,17 +198,16 @@ namespace k8s.Tests
{
var client = new Kubernetes(new KubernetesClientConfiguration { Host = server.Uri.ToString() });
var listTask = await client.CoreV1.ListNamespacedPodWithHttpMessagesAsync("default", watch: true).ConfigureAwait(true);
var events = new HashSet<WatchEventType>();
var watcher = listTask.Watch<V1Pod, V1PodList>(
(type, item) =>
var watcher = client.CoreV1.WatchListNamespacedPod(
"default",
onEvent: (type, item) =>
{
events.Add(type);
eventsReceived.Signal();
},
error =>
onError: error =>
{
testOutput.WriteLine($"Watcher received '{error.GetType().FullName}' error.");
},
@@ -255,27 +257,26 @@ namespace k8s.Tests
{
var client = new Kubernetes(new KubernetesClientConfiguration { Host = server.Uri.ToString() });
var listTask = await client.CoreV1.ListNamespacedPodWithHttpMessagesAsync("default", watch: true).ConfigureAwait(true);
var events = new HashSet<WatchEventType>();
var errors = 0;
var watcher = listTask.Watch<V1Pod, V1PodList>(
(type, item) =>
var watcher = client.CoreV1.WatchListNamespacedPod(
"default",
onEvent: (type, item) =>
{
testOutput.WriteLine($"Watcher received '{type}' event.");
events.Add(type);
eventsReceived.Signal();
},
error =>
onError: error =>
{
testOutput.WriteLine($"Watcher received '{error.GetType().FullName}' error.");
errors += 1;
eventsReceived.Signal();
},
waitForClosed.Set);
onClosed: waitForClosed.Set);
// wait server yields all events
await Task.WhenAny(eventsReceived.WaitAsync(), Task.Delay(TestTimeout)).ConfigureAwait(true);
@@ -324,27 +325,26 @@ namespace k8s.Tests
{
var client = new Kubernetes(new KubernetesClientConfiguration { Host = server.Uri.ToString() });
var listTask = await client.CoreV1.ListNamespacedPodWithHttpMessagesAsync("default", watch: true).ConfigureAwait(true);
var events = new HashSet<WatchEventType>();
var errors = 0;
var watcher = listTask.Watch<V1Pod, V1PodList>(
(type, item) =>
var watcher = client.CoreV1.WatchListNamespacedPod(
"default",
onEvent: (type, item) =>
{
testOutput.WriteLine($"Watcher received '{type}' event.");
events.Add(type);
eventsReceived.Signal();
},
error =>
onError: error =>
{
testOutput.WriteLine($"Watcher received '{error.GetType().FullName}' error.");
errors += 1;
eventsReceived.Signal();
},
connectionClosed.Set);
onClosed: connectionClosed.Set);
// wait server yields all events
await Task.WhenAny(eventsReceived.WaitAsync(), Task.Delay(TestTimeout)).ConfigureAwait(true);
@@ -386,18 +386,17 @@ namespace k8s.Tests
{
var client = new Kubernetes(new KubernetesClientConfiguration { Host = server.Uri.ToString() });
var listTask = await client.CoreV1.ListNamespacedPodWithHttpMessagesAsync("default", watch: true).ConfigureAwait(true);
waitForException.Set();
Watcher<V1Pod> watcher;
watcher = listTask.Watch<V1Pod, V1PodList>(
(type, item) => { },
e =>
watcher = client.CoreV1.WatchListNamespacedPod(
"default",
onEvent: (type, item) => { },
onError: e =>
{
exceptionCatched = e;
exceptionReceived.Set();
},
waitForClosed.Set);
onClosed: waitForClosed.Set);
// wait server down
await Task.WhenAny(exceptionReceived.WaitAsync(), Task.Delay(TestTimeout)).ConfigureAwait(true);
@@ -456,12 +455,11 @@ namespace k8s.Tests
Assert.False(handler1.Called);
Assert.False(handler2.Called);
var listTask = await client.CoreV1.ListNamespacedPodWithHttpMessagesAsync("default", watch: true).ConfigureAwait(true);
var events = new HashSet<WatchEventType>();
var watcher = listTask.Watch<V1Pod, V1PodList>(
(type, item) =>
var watcher = client.CoreV1.WatchListNamespacedPod(
"default",
onEvent: (type, item) =>
{
events.Add(type);
eventsReceived.Signal();
@@ -507,7 +505,9 @@ namespace k8s.Tests
var events = new HashSet<WatchEventType>();
var errors = 0;
var watcher = client.CoreV1.ListNamespacedPodWithHttpMessagesAsync("default", fieldSelector: $"metadata.name=${"myPod"}", watch: true).Watch<V1Pod, V1PodList>(
var watcher = client.CoreV1.WatchListNamespacedPod(
"default",
fieldSelector: $"metadata.name=${"myPod"}",
onEvent:
(type, item) =>
{
@@ -568,7 +568,7 @@ namespace k8s.Tests
Host = server.Uri.ToString(),
HttpClientTimeout = TimeSpan.FromSeconds(5),
});
await client.CoreV1.ListNamespacedPodWithHttpMessagesAsync("default").ConfigureAwait(true);
await client.CoreV1.ListNamespacedPodAsync("default").ConfigureAwait(true);
}).ConfigureAwait(true);
// cts
@@ -580,7 +580,7 @@ namespace k8s.Tests
{
Host = server.Uri.ToString(),
});
await client.CoreV1.ListNamespacedPodWithHttpMessagesAsync("default", cancellationToken: cts.Token).ConfigureAwait(true);
await client.CoreV1.ListNamespacedPodAsync("default", cancellationToken: cts.Token).ConfigureAwait(true);
}).ConfigureAwait(true);
}
@@ -608,7 +608,9 @@ namespace k8s.Tests
var events = new HashSet<WatchEventType>();
var errors = 0;
var watcher = client.CoreV1.ListNamespacedPodWithHttpMessagesAsync("default", fieldSelector: $"metadata.name=${"myPod"}", watch: true).Watch<V1Pod, V1PodList>(
var watcher = client.CoreV1.WatchListNamespacedPod(
"default",
fieldSelector: $"metadata.name=${"myPod"}",
onEvent:
(type, item) =>
{
@@ -656,7 +658,6 @@ namespace k8s.Tests
httpContext.Response.StatusCode = 200;
await httpContext.Response.Body.FlushAsync().ConfigureAwait(true);
await Task.Delay(TimeSpan.FromSeconds(5)).ConfigureAwait(true); // The default timeout is 100 seconds
return true;
}, resp: ""))
{
@@ -667,8 +668,10 @@ namespace k8s.Tests
await Assert.ThrowsAnyAsync<OperationCanceledException>(async () =>
{
await client.CoreV1.ListNamespacedPodWithHttpMessagesAsync("default", watch: true,
cancellationToken: cts.Token).ConfigureAwait(true);
using var watcher = client.CoreV1.WatchListNamespacedPod(
"default",
onEvent: (type, item) => { });
await Task.Delay(TimeSpan.FromSeconds(5), cts.Token).ConfigureAwait(true);
}).ConfigureAwait(true);
}
}
@@ -742,8 +745,288 @@ namespace k8s.Tests
new KubernetesClientConfiguration { Host = server.Uri.ToString() }, handler);
Assert.Null(handler.Version);
await client.CoreV1.ListNamespacedPodWithHttpMessagesAsync("default", watch: true).ConfigureAwait(true);
using var watcher = client.CoreV1.WatchListNamespacedPod("default", onEvent: (type, item) => { });
Assert.Equal(HttpVersion.Version20, handler.Version);
await Task.CompletedTask.ConfigureAwait(true);
}
[Fact]
public async Task AsyncEnumerableWatchAllEvents()
{
var eventsReceived = new AsyncCountdownEvent(4);
var serverShutdown = new AsyncManualResetEvent();
var watchCompleted = new AsyncManualResetEvent();
using (var server = new MockKubeApiServer(testOutput, async httpContext =>
{
await WriteStreamLine(httpContext, MockAddedEventStreamLine).ConfigureAwait(true);
await WriteStreamLine(httpContext, MockDeletedStreamLine).ConfigureAwait(true);
await WriteStreamLine(httpContext, MockModifiedStreamLine).ConfigureAwait(true);
await WriteStreamLine(httpContext, MockErrorStreamLine).ConfigureAwait(true);
// make server alive, cannot set to int.max as of it would block response
await serverShutdown.WaitAsync().ConfigureAwait(true);
return false;
}))
{
var client = new Kubernetes(new KubernetesClientConfiguration { Host = server.Uri.ToString() });
var events = new HashSet<WatchEventType>();
var errors = 0;
// Start async enumerable watch in background task
var watchTask = Task.Run(async () =>
{
try
{
await foreach (var (type, item) in client.CoreV1.WatchListNamespacedPodAsync("default").ConfigureAwait(false))
{
testOutput.WriteLine($"AsyncEnumerable Watcher received '{type}' event.");
events.Add(type);
eventsReceived.Signal();
// Break when we have all expected events
if (events.Count >= 4)
{
break;
}
}
}
catch (Exception ex)
{
testOutput.WriteLine($"AsyncEnumerable Watcher received exception: {ex.GetType().FullName}");
errors++;
eventsReceived.Signal();
}
finally
{
watchCompleted.Set();
}
});
// wait server yields all events
await Task.WhenAny(eventsReceived.WaitAsync(), Task.Delay(TestTimeout)).ConfigureAwait(true);
Assert.True(
eventsReceived.CurrentCount == 0,
"Timed out waiting for all events / errors to be received.");
Assert.Contains(WatchEventType.Added, events);
Assert.Contains(WatchEventType.Deleted, events);
Assert.Contains(WatchEventType.Modified, events);
Assert.Contains(WatchEventType.Error, events);
Assert.Equal(0, errors);
serverShutdown.Set();
await Task.WhenAny(watchCompleted.WaitAsync(), Task.Delay(TestTimeout)).ConfigureAwait(true);
Assert.True(watchCompleted.IsSet);
}
}
[Fact]
public async Task AsyncEnumerableWatchWithCancellation()
{
var eventsReceived = new AsyncCountdownEvent(2);
var serverShutdown = new AsyncManualResetEvent();
using (var server = new MockKubeApiServer(testOutput, async httpContext =>
{
await WriteStreamLine(httpContext, MockAddedEventStreamLine).ConfigureAwait(true);
await WriteStreamLine(httpContext, MockModifiedStreamLine).ConfigureAwait(true);
// Keep server alive
await serverShutdown.WaitAsync().ConfigureAwait(true);
return false;
}))
{
var client = new Kubernetes(new KubernetesClientConfiguration { Host = server.Uri.ToString() });
var events = new HashSet<WatchEventType>();
var cts = new CancellationTokenSource();
var watchTask = Task.Run(async () =>
{
try
{
await foreach (var (type, item) in client.CoreV1.WatchListNamespacedPodAsync("default", cancellationToken: cts.Token).ConfigureAwait(false))
{
testOutput.WriteLine($"AsyncEnumerable Watcher received '{type}' event.");
events.Add(type);
eventsReceived.Signal();
}
}
catch (OperationCanceledException)
{
testOutput.WriteLine("AsyncEnumerable Watcher was cancelled as expected.");
}
});
// Wait for some events to be received
await Task.WhenAny(eventsReceived.WaitAsync(), Task.Delay(TestTimeout)).ConfigureAwait(true);
Assert.True(
eventsReceived.CurrentCount == 0,
"Timed out waiting for events to be received.");
Assert.Contains(WatchEventType.Added, events);
Assert.Contains(WatchEventType.Modified, events);
// Cancel the watch
cts.Cancel();
// Wait for watch task to complete
await Task.WhenAny(watchTask, Task.Delay(TimeSpan.FromSeconds(5))).ConfigureAwait(true);
Assert.True(watchTask.IsCompletedSuccessfully || watchTask.IsCanceled);
serverShutdown.Set();
}
}
[Fact]
public async Task AsyncEnumerableWatchWithFieldSelector()
{
var eventsReceived = new AsyncCountdownEvent(3);
var serverShutdown = new AsyncManualResetEvent();
var watchCompleted = new AsyncManualResetEvent();
using (var server = new MockKubeApiServer(testOutput, async httpContext =>
{
await WriteStreamLine(httpContext, MockAddedEventStreamLine).ConfigureAwait(true);
await WriteStreamLine(httpContext, MockDeletedStreamLine).ConfigureAwait(true);
await WriteStreamLine(httpContext, MockModifiedStreamLine).ConfigureAwait(true);
await serverShutdown.WaitAsync().ConfigureAwait(true);
return false;
}))
{
var client = new Kubernetes(new KubernetesClientConfiguration { Host = server.Uri.ToString() });
var events = new List<(WatchEventType, V1Pod)>();
var watchTask = Task.Run(async () =>
{
try
{
await foreach (var (type, item) in client.CoreV1.WatchListNamespacedPodAsync(
"default",
fieldSelector: $"metadata.name={"testPod"}").ConfigureAwait(false))
{
testOutput.WriteLine($"AsyncEnumerable Watcher received '{type}' event for pod '{item?.Metadata?.Name}'.");
events.Add((type, item));
eventsReceived.Signal();
if (events.Count >= 3)
{
break;
}
}
}
catch (Exception ex)
{
testOutput.WriteLine($"AsyncEnumerable Watcher received exception: {ex.GetType().FullName}");
}
finally
{
watchCompleted.Set();
}
});
// Wait for events
await Task.WhenAny(eventsReceived.WaitAsync(), Task.Delay(TestTimeout)).ConfigureAwait(true);
Assert.True(
eventsReceived.CurrentCount == 0,
"Timed out waiting for all events to be received.");
Assert.Equal(3, events.Count);
Assert.Contains(events, e => e.Item1 == WatchEventType.Added);
Assert.Contains(events, e => e.Item1 == WatchEventType.Deleted);
Assert.Contains(events, e => e.Item1 == WatchEventType.Modified);
serverShutdown.Set();
await Task.WhenAny(watchCompleted.WaitAsync(), Task.Delay(TestTimeout)).ConfigureAwait(true);
Assert.True(watchCompleted.IsSet);
}
}
[Fact]
public async Task AsyncEnumerableWatchErrorHandling()
{
var eventsReceived = new AsyncCountdownEvent(3);
var serverShutdown = new AsyncManualResetEvent();
var watchCompleted = new AsyncManualResetEvent();
var errorReceived = new AsyncManualResetEvent();
using (var server = new MockKubeApiServer(testOutput, async httpContext =>
{
await WriteStreamLine(httpContext, MockKubeApiServer.MockPodResponse).ConfigureAwait(true);
await WriteStreamLine(httpContext, MockBadStreamLine).ConfigureAwait(true);
await WriteStreamLine(httpContext, MockAddedEventStreamLine).ConfigureAwait(true);
await serverShutdown.WaitAsync().ConfigureAwait(true);
return false;
}))
{
var client = new Kubernetes(new KubernetesClientConfiguration { Host = server.Uri.ToString() });
var events = new List<(WatchEventType, V1Pod)>();
var errorCaught = false;
var watchTask = Task.Run(async () =>
{
try
{
await foreach (var (type, item) in client.CoreV1.WatchListNamespacedPodAsync(
"default",
onError: ex =>
{
testOutput.WriteLine($"AsyncEnumerable Watcher onError called: {ex.GetType().FullName}");
errorCaught = true;
errorReceived.Set();
eventsReceived.Signal();
}).ConfigureAwait(false))
{
testOutput.WriteLine($"AsyncEnumerable Watcher received '{type}' event.");
events.Add((type, item));
eventsReceived.Signal();
// Expect some valid events plus error handling
if (events.Count >= 2)
{
break;
}
}
}
catch (Exception ex)
{
testOutput.WriteLine($"AsyncEnumerable Watcher caught exception: {ex.GetType().FullName}");
}
finally
{
watchCompleted.Set();
}
});
// Wait for events and errors
await Task.WhenAny(eventsReceived.WaitAsync(), Task.Delay(TestTimeout)).ConfigureAwait(true);
Assert.True(
eventsReceived.CurrentCount == 0,
"Timed out waiting for events and errors to be received.");
// Should have received at least one valid event and one error
Assert.True(events.Count >= 1, "Should have received at least one valid event");
Assert.True(errorCaught, "Should have caught parsing error");
Assert.True(errorReceived.IsSet, "Error callback should have been called");
serverShutdown.Set();
await Task.WhenAny(watchCompleted.WaitAsync(), Task.Delay(TestTimeout)).ConfigureAwait(true);
Assert.True(watchCompleted.IsSet);
}
}
}
}