also allow retries of mutating APIs

State changing APIs are not retried by default. But added an option to switch that on when the caller desires.
Should be safe when retried only on 50x server errors or connect errors.

The option can be switched on for the KuberContext by calling the `set_retries` method.
Alternatively each API call also accepts a `max_retries` keywork argument to override the default set in the context.

Author: tanmaykm

README.md

@@ -77,7 +77,7 @@ A Kubernetes context can be manipulated with:
 
 - `set_server`: Set the API server location ("http://localhost:8001" if not set)
 - `set_ns`: Set the namespace to deal with (`default` namespace is not set)
-- `set_retries`: Set the number of times an API call should be retried on a retriable error (5 if not set)
+- `set_retries`: Set the number of times an API call should be retried on a retriable error (5 if not set) and whether all APIs should be retried (only non mutating APIs are retried by default)
 
 Other convenience methods:
 

src/helpers.jl

@@ -37,6 +37,7 @@ mutable struct KuberContext
     modelapi::Dict{Symbol,KApi}
     namespace::String
     default_retries::Int
+    retry_all_apis::Bool
     initialized::Bool
 
     function KuberContext()
@@ -51,6 +52,7 @@ mutable struct KuberContext
         kctx.modelapi = Dict{Symbol,KApi}()
         kctx.namespace = DEFAULT_NAMESPACE
         kctx.default_retries = 5
+        kctx.retry_all_apis = false
         kctx.initialized = false
         return kctx
     end
@@ -61,11 +63,23 @@ struct KuberWatchContext
     stream::KuberEventStream
 end
 
-function set_retries(ctx::KuberContext, n)
-    ctx.default_retries = n
+"""
+    set_retries(ctx; count=5, all_apis=false)
+
+Args:
+- ctx: the context to set the options for
+
+Keyword Args:
+- count: how many times to retry (default 5)
+- all_apis: whether to retry even mutating APIs e.g. `put!` (default false)
+"""
+function set_retries(ctx::KuberContext; count::Int=ctx.default_retries, all_apis::Bool=ctx.all_apis)
+    ctx.default_retries = count
+    ctx.retry_all_apis = all_apis
+    ctx
 end
-retries(ctx::KuberContext) = ctx.default_retries
-retries(watch::KuberWatchContext) = retries(watch.ctx)
+retries(ctx::KuberContext, mutating::Bool=true) = (mutating && !ctx.retry_all_apis) ? 1 : ctx.default_retries
+retries(watch::KuberWatchContext, mutating::Bool=true) = retries(watch.ctx, mutating)
 
 convert(::Type{Vector{UInt8}}, s::T) where {T<:AbstractString} = collect(codeunits(s))
 convert(::Type{T}, json::String) where {T<:SwaggerModel} = convert(T, JSON.parse(json))
@@ -118,14 +132,37 @@ show(io::IO, ctx::KuberContext) = print("Kubernetes namespace ", ctx.namespace,
 get_server(ctx::KuberContext) = ctx.client.root
 get_ns(ctx::KuberContext) = ctx.namespace
 
-function set_server(ctx::KuberContext, uri::String=DEFAULT_URI, reset_api_versions::Bool=false; max_tries=retries(ctx), kwargs...)
+"""
+    set_server(ctx, uri, reset_api_versions=false; max_tries=5, kwargs...)
+
+Set the Kubernetes API server endpoint for a context.
+
+Args:
+- ctx: the context for which to set the API server endpoint
+- uri: the API server endpoint uri
+- reset_api_versions: whether to probe the server again for API versions supported (false by default)
+
+Keyword Args:
+- max_tries: retries allowed while probing API versions from server
+- kwargs: other keyword args to pass on while constructing the client for API server (see Swagger.jl - https://github.com/JuliaComputing/Swagger.jl#readme)
+"""
+function set_server(ctx::KuberContext, uri::String=DEFAULT_URI, reset_api_versions::Bool=false; max_tries=retries(ctx, false), kwargs...)
     rtfn = (default,data)->kuber_type(ctx, default, data)
     ctx.client = Swagger.Client(uri; get_return_type=rtfn, kwargs...)
     ctx.client.headers["Connection"] = "close"
     reset_api_versions && set_api_versions!(ctx; max_tries=max_tries)
     ctx.client
 end
 
+"""
+    set_ns(ctx, namespace)
+
+Set the namespace this context should operate with.
+
+Args:
+- ctx: the context to set the namespace for
+- namespace: the namespace to set (String)
+"""
 set_ns(ctx::KuberContext, namespace::String) = (ctx.namespace = namespace)
 
 camel(a) = string(uppercase(a[1])) * (a[2:end])
@@ -170,7 +207,7 @@ function override_pref(name, server_pref, override)
     server_pref
 end
 
-function fetch_misc_apis_versions(ctx::KuberContext; override=nothing, verbose::Bool=false, max_tries=retries(ctx))
+function fetch_misc_apis_versions(ctx::KuberContext; override=nothing, verbose::Bool=false, max_tries=retries(ctx, false))
     apis = ctx.apis
     vers = k8s_retry(; max_tries=max_tries) do
         getAPIVersions(ApisApi(ctx.client))
@@ -209,7 +246,7 @@ function fetch_misc_apis_versions(ctx::KuberContext; override=nothing, verbose::
     apis
 end
 
-function fetch_core_version(ctx::KuberContext; override=nothing, verbose::Bool=false, max_tries=retries(ctx))
+function fetch_core_version(ctx::KuberContext; override=nothing, verbose::Bool=false, max_tries=retries(ctx, false))
     apis = ctx.apis
     api_vers = k8s_retry(; max_tries=max_tries) do
         getCoreAPIVersions(CoreApi(ctx.client))
@@ -257,7 +294,7 @@ function build_model_api_map(ctx::KuberContext)
     modelapi
 end
 
-function set_api_versions!(ctx::KuberContext; override=nothing, verbose::Bool=false, max_tries=retries(ctx))
+function set_api_versions!(ctx::KuberContext; override=nothing, verbose::Bool=false, max_tries=retries(ctx, false))
     ctx.initialized = false
     empty!(ctx.apis)
     empty!(ctx.modelapi)

src/simpleapi.jl

@@ -12,7 +12,7 @@ _delopts(; kwargs...) = Typedefs.MetaV1.DeleteOptions(; preconditions=Typedefs.M
 _kubectx(ctx::KuberContext) = ctx
 _kubectx(ctx::KuberWatchContext) = ctx.ctx
 
-function _get_apictx(ctx::Union{KuberContext,KuberWatchContext}, O::Symbol, apiversion::Union{String,Nothing}; max_tries::Int=retries(ctx))
+function _get_apictx(ctx::Union{KuberContext,KuberWatchContext}, O::Symbol, apiversion::Union{String,Nothing}; max_tries::Int=retries(ctx, false))
     kubectx = _kubectx(ctx)
     kubectx.initialized || set_api_versions!(kubectx; max_tries=max_tries)
 
@@ -50,7 +50,7 @@ end
 function list(ctx::Union{KuberContext,KuberWatchContext}, O::Symbol, name::String;
         apiversion::Union{String,Nothing}=nothing,
         namespace::Union{String,Nothing}=_kubectx(ctx).namespace,
-        max_tries::Int=retries(ctx),
+        max_tries::Int=retries(ctx, false),
         watch=isa(ctx, KuberWatchContext),
         resourceVersion=nothing,
         kwargs...)
@@ -76,7 +76,7 @@ function list(ctx::Union{KuberContext,KuberWatchContext}, O::Symbol, name::Strin
         end
     end
 
-    # if not watching, retuen the first result
+    # if not watching, return the first result
     watch || (return result)
     if result !== nothing
         resourceVersion = result.metadata.resourceVersion
@@ -93,7 +93,7 @@ end
 function list(ctx::Union{KuberContext,KuberWatchContext}, O::Symbol;
         apiversion::Union{String,Nothing}=nothing,
         namespace::Union{String,Nothing}=_kubectx(ctx).namespace,
-        max_tries::Int=retries(ctx),
+        max_tries::Int=retries(ctx, false),
         watch=isa(ctx, KuberWatchContext),
         resourceVersion=nothing,
         kwargs...)
@@ -177,7 +177,7 @@ function get(ctx::Union{KuberContext,KuberWatchContext}, O::Symbol;
         apiversion::Union{String,Nothing}=nothing,
         label_selector=nothing,
         namespace::Union{String,Nothing}=_kubectx(ctx).namespace,
-        max_tries::Integer=retries(ctx),
+        max_tries::Integer=retries(ctx, false),
         watch=isa(ctx, KuberWatchContext),
         resourceVersion=nothing,
         kwargs...)
@@ -219,7 +219,7 @@ end
 function watch(ctx::KuberContext, O::Symbol, outstream::Channel, name::String;
         apiversion::Union{String,Nothing}=nothing,
         namespace::Union{String,Nothing}=ctx.namespace,
-        max_tries::Int=retries(ctx),
+        max_tries::Int=retries(ctx, false),
         kwargs...)
     apictx = _get_apictx(ctx, O, apiversion; max_tries=max_tries)
     namespaced = (namespace !== nothing) && !isempty(namespace)
@@ -246,7 +246,7 @@ end
 function watch(ctx::KuberContext, O::Symbol, outstream::Channel;
         apiversion::Union{String,Nothing}=nothing,
         namespace::Union{String,Nothing}=ctx.namespace,
-        max_tries::Int=retries(ctx),
+        max_tries::Int=retries(ctx, false),
         kwargs...)
     apictx = _get_apictx(ctx, O, apiversion; max_tries=max_tries)
     namespaced = (namespace !== nothing) && !isempty(namespace)
@@ -270,58 +270,70 @@ function watch(ctx::KuberContext, O::Symbol, outstream::Channel;
     end
 end
 
-function put!(ctx::KuberContext, v::T) where {T<:SwaggerModel}
+function put!(ctx::KuberContext, v::T; max_tries::Int=retries(ctx, true)) where {T<:SwaggerModel}
     vjson = convert(Dict{String,Any}, v)
-    put!(ctx, Symbol(vjson["kind"]), vjson)
+    put!(ctx, Symbol(vjson["kind"]), vjson; max_tries=max_tries)
 end
 
-function put!(ctx::KuberContext, O::Symbol, d::Dict{String,Any})
+function put!(ctx::KuberContext, O::Symbol, d::Dict{String,Any}; max_tries::Int=retries(ctx, true))
     apictx = _get_apictx(ctx, O, get(d, "apiVersion", nothing))
     if (apicall = _api_function("create$O")) !== nothing
-        return apicall(apictx, d)
+        return k8s_retry(; max_tries=max_tries) do
+            apicall(apictx, d)
+        end
     elseif (apicall = _api_function("createNamespaced$O")) !== nothing
-        return apicall(apictx, ctx.namespace, d)
+        return k8s_retry(; max_tries=max_tries) do
+            apicall(apictx, ctx.namespace, d)
+        end
     else
         throw(ArgumentError("No API functions could be located using :$O"))
     end
 end
 
-function delete!(ctx::KuberContext, v::T; kwargs...) where {T<:SwaggerModel}
+function delete!(ctx::KuberContext, v::T; max_tries::Int=retries(ctx, true), kwargs...) where {T<:SwaggerModel}
     vjson = convert(Dict{String,Any}, v)
     kind = vjson["kind"]
     name = vjson["metadata"]["name"]
-    delete!(ctx, Symbol(kind), name; apiversion=get(vjson, "apiVersion", nothing), kwargs...)
+    delete!(ctx, Symbol(kind), name; apiversion=get(vjson, "apiVersion", nothing), max_tries=max_tries, kwargs...)
 end
 
-function delete!(ctx::KuberContext, O::Symbol, name::String; apiversion::Union{String,Nothing}=nothing, kwargs...)
+function delete!(ctx::KuberContext, O::Symbol, name::String; apiversion::Union{String,Nothing}=nothing, max_tries::Int=retries(ctx, true), kwargs...)
     apictx = _get_apictx(ctx, O, apiversion)
 
     params = [apictx, name]
 
     if (apicall = _api_function("delete$O")) !== nothing
-        return apicall(params...; kwargs...)
+        return k8s_retry(; max_tries=max_tries) do
+            apicall(params...; kwargs...)
+        end
     elseif (apicall = _api_function("deleteNamespaced$O")) !== nothing
         push!(params, ctx.namespace)
-        return apicall(params...; kwargs...)
+        return k8s_retry(; max_tries=max_tries) do
+            apicall(params...; kwargs...)
+        end
     else
         throw(ArgumentError("No API functions could be located using :$O"))
     end
 end
 
-function update!(ctx::KuberContext, v::T, patch, patch_type) where {T<:SwaggerModel}
+function update!(ctx::KuberContext, v::T, patch, patch_type; max_tries::Int=retries(ctx, true)) where {T<:SwaggerModel}
     vjson = convert(Dict{String,Any}, v)
     kind = vjson["kind"]
     name = vjson["metadata"]["name"]
-    update!(ctx, Symbol(kind), name, patch, patch_type; apiversion=get(vjson, "apiVersion", nothing))
+    update!(ctx, Symbol(kind), name, patch, patch_type; apiversion=get(vjson, "apiVersion", nothing), max_tries=max_tries)
 end
 
-function update!(ctx::KuberContext, O::Symbol, name::String, patch, patch_type; apiversion::Union{String,Nothing}=nothing)
+function update!(ctx::KuberContext, O::Symbol, name::String, patch, patch_type; apiversion::Union{String,Nothing}=nothing, max_tries::Int=retries(ctx, true))
     apictx = _get_apictx(ctx, O, apiversion)
 
     if (apicall = _api_function("patch$O")) !== nothing
-        return apicall(apictx, name, patch; _mediaType=patch_type)
+        return k8s_retry(; max_tries=max_tries) do
+            apicall(apictx, name, patch; _mediaType=patch_type)
+        end
     elseif (apicall = _api_function("patchNamespaced$O")) !== nothing
-        return apicall(apictx, name, ctx.namespace, patch; _mediaType=patch_type)
+        return k8s_retry(; max_tries=max_tries) do
+            apicall(apictx, name, ctx.namespace, patch; _mediaType=patch_type)
+        end
     else
         throw(ArgumentError("No API functions could be located using :$O"))
     end

test/runtests.jl

@@ -15,7 +15,7 @@ function init_context(override=nothing, verbose=true)
     ctx = KuberContext()
     set_server(ctx, "http://localhost:8001")
     set_ns(ctx, "default")
-    set_retries(ctx, 3)
+    set_retries(ctx; count=3, all_apis=false)
     Kuber.set_api_versions!(ctx; override=override, verbose=verbose)
     ctx
 end