From b044177bad41c68397160595b50356357ffb6455 Mon Sep 17 00:00:00 2001 From: "Nishanth H. Kottary" Date: Wed, 15 Jul 2026 09:55:26 +0530 Subject: [PATCH 1/2] Retry on request-interrupted watch failures Long-lived Pod watches can be interrupted by transient connection resets (idle LB/proxy timeouts, apiserver restarts, network blips), which OpenAPI.jl surfaces as InvocationException("request was interrupted"). k8s_retry_cond only retried ApiException with retryable status codes, so this case fell straight through to callers as an unrecoverable error. Treat it as retryable, matching the existing is_longpoll_timeout handling. Co-Authored-By: Claude Sonnet 5 --- Project.toml | 2 +- src/helpers.jl | 3 +++ 2 files changed, 4 insertions(+), 1 deletion(-) diff --git a/Project.toml b/Project.toml index 9919848..86b94a1 100644 --- a/Project.toml +++ b/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" diff --git a/src/helpers.jl b/src/helpers.jl index 51dd737..e6d7cd7 100644 --- a/src/helpers.jl +++ b/src/helpers.jl @@ -29,6 +29,9 @@ function k8s_retry_cond(s, e, retryable_codes=k8s_retryable_codes) if (e isa OpenAPI.Clients.ApiException) && (e.status in retryable_codes) return (s, true) end + if OpenAPI.Clients.is_request_interrupted(e) + return (s, true) + end (s, false) end From cf1fb070b6caa6e06ba70115ebb299e0ebb6c7d7 Mon Sep 17 00:00:00 2001 From: "Nishanth H. Kottary" Date: Wed, 15 Jul 2026 10:14:30 +0530 Subject: [PATCH 2/2] Don't retry request-interrupted watches that were intentionally stopped @tanmaykm pointed out on #68 that closing the watch event channel is the documented way for a consumer to stop an in-progress watch (see the `watch(streamprocessor, ctx, watched, ...)` composition), and that this also surfaces as `is_request_interrupted` in the :downloads backend: the interrupt-watcher task force-interrupts the download on channel close, leaving `resp` nothing, which `exec()` reports identically to a genuine dropped connection. Thread the event/output stream through k8s_retry so k8s_retry_cond can tell the two cases apart: only treat is_request_interrupted as retryable while that channel is still open. A closed channel means the consumer asked to stop, so let it propagate instead of retrying. Co-Authored-By: Claude Sonnet 5 --- src/helpers.jl | 18 +++++++++++++++--- src/simpleapi.jl | 20 ++++++++++---------- 2 files changed, 25 insertions(+), 13 deletions(-) diff --git a/src/helpers.jl b/src/helpers.jl index e6d7cd7..78ddce7 100644 --- a/src/helpers.jl +++ b/src/helpers.jl @@ -25,11 +25,19 @@ 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) + if OpenAPI.Clients.is_request_interrupted(e) && (stream === nothing || isopen(stream)) return (s, true) end (s, false) @@ -38,8 +46,12 @@ 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. diff --git a/src/simpleapi.jl b/src/simpleapi.jl index c20d15e..37f9d14 100644 --- a/src/simpleapi.jl +++ b/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