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..78ddce7 100644 --- a/src/helpers.jl +++ b/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. 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