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

lqlive · web-flow · commit 6d27bd900bcb · 2025-10-11T15:10:53.000-07:00
* 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
diff --git a/examples/clientset/Program.cs b/examples/clientset/Program.cs
@@ -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 @@ private static async Task Main(string[] args)
             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 @@ private static async Task Main(string[] args)
 
             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);
+            }
         }
     }
-}
+}
diff --git a/examples/watch/Program.cs b/examples/watch/Program.cs
@@ -1,5 +1,4 @@
 using k8s;
-using k8s.Models;
 using System;
 using System.Threading;
 using System.Threading.Tasks;
@@ -8,9 +7,10 @@
 
 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 @@
 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);
@@ -37,4 +36,4 @@ void WatchUsingCallback(IKubernetes client)
         Console.CancelKeyPress += (sender, eventArgs) => ctrlc.Set();
         ctrlc.Wait();
     }
-}
+}
diff --git a/src/KubernetesClient/WatcherExt.cs b/src/KubernetesClient/WatcherExt.cs
@@ -16,6 +16,7 @@ public static class WatcherExt
         /// 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 @@ private static Func<Task<TextReader>> MakeStreamReaderCreator<T, L>(Task<HttpOpe
         /// 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 @@ public static Watcher<T> Watch<T, L>(
         /// <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,
diff --git a/src/LibKubernetesGenerator/ParamHelper.cs b/src/LibKubernetesGenerator/ParamHelper.cs
@@ -3,6 +3,7 @@
 using Scriban.Runtime;
 using System;
 using System.Linq;
+using System.Collections.Generic;
 
 namespace LibKubernetesGenerator
 {
@@ -21,6 +22,8 @@ public void RegisterHelper(ScriptObject scriptObject)
         {
             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 @@ public static bool IfParamContains(OpenApiOperation operation, string name)
             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
@@ -57,4 +77,4 @@ public string GetModelCtorParam(JsonSchema schema)
                     }));
         }
     }
-}
+}
diff --git a/src/LibKubernetesGenerator/TypeHelper.cs b/src/LibKubernetesGenerator/TypeHelper.cs
@@ -122,7 +122,6 @@ private string GetDotNetType(JsonSchema schema, JsonSchemaProperty parent)
                     return $"IDictionary<string, {GetDotNetType(schema.AdditionalPropertiesSchema, parent)}>";
                 }
 
-
                 if (schema?.Reference != null)
                 {
                     return classNameHelper.GetClassNameForSchemaDefinition(schema.Reference);
@@ -245,6 +244,16 @@ string toType()
                     }
 
                     break;
+                case "T":
+                    var itemType = TryGetItemTypeFromSchema(response);
+                    if (itemType != null)
+                    {
+                        return itemType;
+                    }
+
+                    break;
+                case "TList":
+                    return t;
             }
 
             return t;
@@ -283,5 +292,26 @@ public static bool IfType(JsonSchemaProperty property, string type)
 
             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;
+        }
     }
-}
+}
diff --git a/src/LibKubernetesGenerator/templates/Client.cs.template b/src/LibKubernetesGenerator/templates/Client.cs.template
@@ -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,15 +30,15 @@ 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))
     {
         {{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}} 
+}
diff --git a/src/LibKubernetesGenerator/templates/OperationsExtensions.cs.template b/src/LibKubernetesGenerator/templates/OperationsExtensions.cs.template
diff --git a/tests/E2E.Tests/MinikubeTests.cs b/tests/E2E.Tests/MinikubeTests.cs
diff --git a/tests/KubernetesClient.Tests/WatchTests.cs b/tests/KubernetesClient.Tests/WatchTests.cs
