Retry on request-interrupted watch failures

Project.toml

@@ -4,7 +4,7 @@ authors = ["JuliaHub Inc."]
 keywords = ["kubernetes", "client"]
 license = "MIT"
 desc = "Julia Kubernetes Client"
-version = "0.7.10"
+version = "0.7.11"
 
 [deps]
 Dates = "ade2ca70-3891-5945-98fb-dc099432e06a"

src/helpers.jl

@@ -25,18 +25,33 @@ OpenAPI status codes that can be retried.
 """
 const k8s_retryable_codes = [0, 500, 501, 502, 503, 504]
 
-function k8s_retry_cond(s, e, retryable_codes=k8s_retryable_codes)
+"""
+`stream`, if given, is the event channel a watch call streams into. Closing it is
+the documented way for a consumer to stop an in-progress watch (see `watch` in
+simpleapi.jl), and doing so also surfaces as `is_request_interrupted`. Only treat
+`is_request_interrupted` as retryable while that channel is still open, so an
+intentional stop isn't mistaken for a transient interruption and retried instead
+of being allowed to terminate.
+"""
+function k8s_retry_cond(s, e, retryable_codes=k8s_retryable_codes; stream::Union{Channel,Nothing}=nothing)
     if (e isa OpenAPI.Clients.ApiException) && (e.status in retryable_codes)
         return (s, true)
     end
+    if OpenAPI.Clients.is_request_interrupted(e) && (stream === nothing || isopen(stream))
+        return (s, true)
+    end
     (s, false)
 end
 
 """
 Retry api call automatically (if `max_tries > 1`) on certain retryable failures.
 Backoff to use when retrying k8s APIs. The default minimum is 2 TPS.
+
+`stream`, if given, is passed to `k8s_retry_cond` to distinguish an intentional
+watch-stop (channel closed by the consumer) from a transient interruption.
 """
-k8s_retry(f; max_tries=1, tps=2) = retry(f, delays=k8s_delay(tps,max_tries), check=k8s_retry_cond)()
+k8s_retry(f; max_tries=1, tps=2, stream::Union{Channel,Nothing}=nothing) =
+    retry(f, delays=k8s_delay(tps,max_tries), check=(s,e)->k8s_retry_cond(s,e; stream=stream))()
 
 """
 Build keyword arguments for OpenAPI.Clients.Client constructor.

src/simpleapi.jl

@@ -97,7 +97,7 @@ function list(ctx::Union{KuberContext,KuberWatchContext}, O::Symbol, name::Strin
     end
 
     # start watch and return the HTTP response object on completion
-    result = k8s_retry(; max_tries=max_tries) do
+    result = k8s_retry(; max_tries=max_tries, stream=eventstream) do
         check_api_response(apicall(apictx, eventstream, args...; watch=watch, resource_version=resource_version, kwargs...)...)
     end
 
@@ -148,7 +148,7 @@ function list(ctx::Union{KuberContext,KuberWatchContext}, O::Symbol;
     end
 
     # start watch and return the HTTP response object on completion
-    result = k8s_retry(; max_tries=max_tries) do
+    result = k8s_retry(; max_tries=max_tries, stream=eventstream) do
         check_api_response(apicall(apictx, eventstream, args...; watch=watch, resource_version=resource_version, kwargs...)...)
     end
 
@@ -197,7 +197,7 @@ function get(ctx::Union{KuberContext,KuberWatchContext}, O::Symbol, name::String
     end
 
     # start watch and return the HTTP response object on completion
-    result = k8s_retry(; max_tries=max_tries) do
+    result = k8s_retry(; max_tries=max_tries, stream=eventstream) do
         check_api_response(apicall(apictx, eventstream, args...; watch=watch, resource_version=resource_version, kwargs...)...)
     end
 
@@ -252,7 +252,7 @@ function get(ctx::Union{KuberContext,KuberWatchContext}, O::Symbol;
     end
 
     # start watch and return the HTTP response object on completion
-    result = k8s_retry(; max_tries=max_tries) do
+    result = k8s_retry(; max_tries=max_tries, stream=eventstream) do
         check_api_response(apicall(apictx, eventstream, args...; watch=watch, resource_version=resource_version, label_selector, kwargs...)...)
     end
 
@@ -272,17 +272,17 @@ function watch(ctx::KuberContext, O::Symbol, outstream::Channel, name::String;
 
     if allnamespaces
         apicall = apimodule(ctx).eval(Symbol("watch_$(_O_)_for_all_namespaces"))
-        result = k8s_retry(; max_tries=max_tries) do
+        result = k8s_retry(; max_tries=max_tries, stream=outstream) do
             check_api_response(apicall(apictx, outstream, name; kwargs...)...)
         end
     elseif namespaced
         apicall = apimodule(ctx).eval(Symbol("watch_namespaced_$(_O_)"))
-        result = k8s_retry(; max_tries=max_tries) do
+        result = k8s_retry(; max_tries=max_tries, stream=outstream) do
             check_api_response(apicall(apictx, outstream, name, namespace; kwargs...)...)
         end
     else
         apicall = apimodule(ctx).eval(Symbol("watch_$(_O_)"))
-        result = k8s_retry(; max_tries=max_tries) do
+        result = k8s_retry(; max_tries=max_tries, stream=outstream) do
             check_api_response(apicall(apictx, outstream, name; kwargs...)...)
         end
     end
@@ -303,17 +303,17 @@ function watch(ctx::KuberContext, O::Symbol, outstream::Channel;
 
     if allnamespaces
         apicall = apimodule(ctx).eval(Symbol("watch_$(_O_)_for_all_namespaces"))
-        result = k8s_retry(; max_tries=max_tries) do
+        result = k8s_retry(; max_tries=max_tries, stream=outstream) do
             check_api_response(apicall(apictx, outstream; kwargs...)...)
         end
     elseif namespaced
         apicall = apimodule(ctx).eval(Symbol("watch_namespaced_$(_O_)"))
-        result = k8s_retry(; max_tries=max_tries) do
+        result = k8s_retry(; max_tries=max_tries, stream=outstream) do
             check_api_response(apicall(apictx, outstream, namespace; kwargs...)...)
         end
     else
         apicall = apimodule(ctx).eval(Symbol("watch_$(_O_)"))
-        result = k8s_retry(; max_tries=max_tries) do
+        result = k8s_retry(; max_tries=max_tries, stream=outstream) do
             check_api_response(apicall(apictx, outstream; kwargs...)...)
         end
     end