Merge pull request #41 from JuliaComputing/tan/retry

fix and improve retries

Author: tanmaykm

Project.toml

@@ -4,21 +4,19 @@ authors = ["Tanmay Mohapatra <[email protected]>"]
 keywords = ["kubernetes", "client"]
 license = "MIT"
 desc = "Julia Kubernetes Client"
-version = "0.5.0"
+version = "0.5.1"
 
 [deps]
 Dates = "ade2ca70-3891-5945-98fb-dc099432e06a"
 Downloads = "f43a241f-c20a-4ad4-852c-f6b1247861c6"
 JSON = "682c06a0-de6a-54ab-a142-c8b1cf79cde6"
 Random = "9a3f8284-a2c9-5f02-9a11-845980a1fd5c"
 Swagger = "2d69052b-6a58-5cd9-a030-a48559c324ac"
-Retry = "20febd7b-183b-5ae2-ac4a-720e7ce64774"
 
 [compat]
 Downloads = "1"
 Swagger = "0.3"
 JSON = "0.21"
-Retry = "0.4"
 julia = "1"
 
 [extras]

README.md

@@ -75,8 +75,9 @@ end
 
 A Kubernetes context can be manipulated with:
 
-- `set_server`: Set the API server location
-- `set_ns`: Set the namespace to deal 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)
 
 Other convenience methods:
 

src/Kuber.jl

@@ -3,7 +3,6 @@ module Kuber
 using JSON
 using Swagger
 using Downloads
-using Retry
 
 include("api/Kubernetes.jl")
 using .Kubernetes
@@ -17,7 +16,7 @@ include("apialiases.jl")
 include("helpers.jl")
 include("simpleapi.jl")
 
-export KuberContext, set_server, set_ns, get_server, get_ns, kuber_type, kuber_obj, @K_str
+export KuberContext, set_server, set_ns, set_retries, get_server, get_ns, kuber_type, kuber_obj, @K_str
 export get, list, watch, put!, update!, delete!, sel, get_logs, list_namespaced_custom_metrics, list_custom_metrics
 
 end # module

src/helpers.jl

@@ -1,6 +1,29 @@
 const DEFAULT_NAMESPACE = "default"
 const DEFAULT_URI = "http://localhost:8001"
 
+"""delay customized by TPS requirement"""
+k8s_delay(tps, max_tries=1) = ExponentialBackOff(n=max_tries, first_delay=(1/tps), factor=1.75, jitter=0.1)
+
+"""
+Swagger status codes that can be retried.
+0: network error (HTTP was not even attempted)
+500-504: unexpected server error
+"""
+const k8s_retryable_codes = [0, 500, 501, 502, 503, 504]
+
+function k8s_retry_cond(s, e, retryable_codes=k8s_retryable_codes)
+    if (e isa Swagger.ApiException) && (e.status in retryable_codes)
+        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.
+"""
+k8s_retry(f; max_tries=1, tps=2) = retry(f, delays=k8s_delay(tps,max_tries), check=k8s_retry_cond)()
+
 const KuberEventStream = Channel{Any}
 
 struct KApi
@@ -13,6 +36,7 @@ mutable struct KuberContext
     apis::Dict{Symbol,Vector{KApi}}
     modelapi::Dict{Symbol,KApi}
     namespace::String
+    default_retries::Int
     initialized::Bool
 
     function KuberContext()
@@ -26,6 +50,7 @@ mutable struct KuberContext
         kctx.apis = Dict{Symbol,Vector}()
         kctx.modelapi = Dict{Symbol,KApi}()
         kctx.namespace = DEFAULT_NAMESPACE
+        kctx.default_retries = 5
         kctx.initialized = false
         return kctx
     end
@@ -36,6 +61,12 @@ struct KuberWatchContext
     stream::KuberEventStream
 end
 
+function set_retries(ctx::KuberContext, n)
+    ctx.default_retries = n
+end
+retries(ctx::KuberContext) = ctx.default_retries
+retries(watch::KuberWatchContext) = retries(watch.ctx)
+
 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))
 convert(::Type{Dict{String,Any}}, model::T) where {T<:SwaggerModel} = JSON.parse(JSON.json(model))
@@ -87,7 +118,7 @@ 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=1, kwargs...)
+function set_server(ctx::KuberContext, uri::String=DEFAULT_URI, reset_api_versions::Bool=false; max_tries=retries(ctx), kwargs...)
     rtfn = (default,data)->kuber_type(ctx, default, data)
     ctx.client = Swagger.Client(uri; get_return_type=rtfn, kwargs...)
     ctx.client.headers["Connection"] = "close"
@@ -139,15 +170,10 @@ function override_pref(name, server_pref, override)
     server_pref
 end
 
-function fetch_misc_apis_versions(ctx::KuberContext; override=nothing, verbose::Bool=false, max_tries=1)
+function fetch_misc_apis_versions(ctx::KuberContext; override=nothing, verbose::Bool=false, max_tries=retries(ctx))
     apis = ctx.apis
-    vers = @repeat max_tries try
+    vers = k8s_retry(; max_tries=max_tries) do
         getAPIVersions(ApisApi(ctx.client))
-    catch e
-        @retry if isa(e, Base.IOError)
-            @debug("Retrying getAPIVersions")
-            sleep(2)
-        end
     end
     api_groups = vers.groups
     for apigrp in api_groups
@@ -183,15 +209,10 @@ 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=1)
+function fetch_core_version(ctx::KuberContext; override=nothing, verbose::Bool=false, max_tries=retries(ctx))
     apis = ctx.apis
-    api_vers = @repeat max_tries try
+    api_vers = k8s_retry(; max_tries=max_tries) do
         getCoreAPIVersions(CoreApi(ctx.client))
-    catch e
-        @retry if isa(e, Base.IOError)
-            @debug("Retrying getCoreAPIVersions")
-            sleep(2)
-        end
     end
     name = "Core"
     pref_vers = override_pref(name, api_vers.versions[1], override)
@@ -236,7 +257,7 @@ function build_model_api_map(ctx::KuberContext)
     modelapi
 end
 
-function set_api_versions!(ctx::KuberContext; override=nothing, verbose::Bool=false, max_tries=1)
+function set_api_versions!(ctx::KuberContext; override=nothing, verbose::Bool=false, max_tries=retries(ctx))
     ctx.initialized = false
     empty!(ctx.apis)
     empty!(ctx.modelapi)
@@ -255,28 +276,3 @@ function set_api_versions!(ctx::KuberContext; override=nothing, verbose::Bool=fa
     ctx.initialized = true
     nothing
 end
-
-"""
-Retry function `f` for `max_tries` number of times if function fails with `IOError`
-"""
-function retry_on_error(f::Function; max_tries=1)
-    @repeat max_tries try
-        return f()
-    catch e
-        @retry if isa(e, Base.IOError)
-            @debug("Retrying Kubernetes API call ...")
-            sleep(2)
-        end
-    end
-end
-
-"""
-Macro to retry an expression on `IOError`. Note that the variable `max_tries` needs to be inscope for this to work.
-"""
-macro retry_on_error(e)
-   esc(quote
-      retry_on_error(;max_tries=(@isdefined max_tries) ? max_tries : 1) do
-          $(e)
-      end
-   end)
-end

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=1)
+function _get_apictx(ctx::Union{KuberContext,KuberWatchContext}, O::Symbol, apiversion::Union{String,Nothing}; max_tries::Int=retries(ctx))
     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=1,
+        max_tries::Int=retries(ctx),
         watch=isa(ctx, KuberWatchContext),
         resourceVersion=nothing,
         kwargs...)
@@ -71,7 +71,9 @@ function list(ctx::Union{KuberContext,KuberWatchContext}, O::Symbol, name::Strin
     end
 
     if !watch || resourceVersion === nothing
-        result = @retry_on_error apicall(apictx, args...; kwargs...)
+        result = k8s_retry(; max_tries=max_tries) do
+            apicall(apictx, args...; kwargs...)
+        end
     end
 
     # if not watching, retuen the first result
@@ -83,13 +85,15 @@ function list(ctx::Union{KuberContext,KuberWatchContext}, O::Symbol, name::Strin
     end
 
     # start watch and return the HTTP response object on completion
-    return @retry_on_error apicall(apictx, eventstream, args...; watch=watch, resourceVersion=resourceVersion, kwargs...)
+    return k8s_retry(; max_tries=max_tries) do
+        apicall(apictx, eventstream, args...; watch=watch, resourceVersion=resourceVersion, kwargs...)
+    end
 end
 
 function list(ctx::Union{KuberContext,KuberWatchContext}, O::Symbol;
         apiversion::Union{String,Nothing}=nothing,
         namespace::Union{String,Nothing}=_kubectx(ctx).namespace,
-        max_tries::Int=1,
+        max_tries::Int=retries(ctx),
         watch=isa(ctx, KuberWatchContext),
         resourceVersion=nothing,
         kwargs...)
@@ -111,7 +115,9 @@ function list(ctx::Union{KuberContext,KuberWatchContext}, O::Symbol;
     end
 
     if !watch || resourceVersion === nothing
-        result = @retry_on_error apicall(apictx, args...; kwargs...)
+        result = k8s_retry(; max_tries=max_tries) do
+            apicall(apictx, args...; kwargs...)
+        end
     end
 
     # if not watching, retuen the first result
@@ -123,12 +129,14 @@ function list(ctx::Union{KuberContext,KuberWatchContext}, O::Symbol;
     end
 
     # start watch and return the HTTP response object on completion
-    return @retry_on_error apicall(apictx, eventstream, args...; watch=watch, resourceVersion=resourceVersion, kwargs...)
+    return k8s_retry(; max_tries=max_tries) do
+        apicall(apictx, eventstream, args...; watch=watch, resourceVersion=resourceVersion, kwargs...)
+    end
 end
 
 function get(ctx::Union{KuberContext,KuberWatchContext}, O::Symbol, name::String;
         apiversion::Union{String,Nothing}=nothing,
-        max_tries::Integer=1,
+        max_tries::Integer=retries(ctx),
         watch=isa(ctx, KuberWatchContext),
         resourceVersion=nothing,
         kwargs...)
@@ -146,7 +154,9 @@ function get(ctx::Union{KuberContext,KuberWatchContext}, O::Symbol, name::String
     end
 
     if !watch || resourceVersion === nothing
-        result = @retry_on_error apicall(apictx, args...; kwargs...)
+        result = k8s_retry(; max_tries=max_tries) do
+            apicall(apictx, args...; kwargs...)
+        end
     end
 
     # if not watching, retuen the first result
@@ -158,14 +168,16 @@ function get(ctx::Union{KuberContext,KuberWatchContext}, O::Symbol, name::String
     end
 
     # start watch and return the HTTP response object on completion
-    return @retry_on_error apicall(apictx, eventstream, args...; watch=watch, resourceVersion=resourceVersion, kwargs...)
+    return k8s_retry(; max_tries=max_tries) do
+        apicall(apictx, eventstream, args...; watch=watch, resourceVersion=resourceVersion, kwargs...)
+    end
 end
 
 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=1,
+        max_tries::Integer=retries(ctx),
         watch=isa(ctx, KuberWatchContext),
         resourceVersion=nothing,
         kwargs...)
@@ -185,7 +197,9 @@ function get(ctx::Union{KuberContext,KuberWatchContext}, O::Symbol;
     end
 
     if !watch || resourceVersion === nothing
-        result = @retry_on_error apicall(apictx, args...; labelSelector=label_selector, kwargs...)
+        result = k8s_retry(; max_tries=max_tries) do
+            apicall(apictx, args...; labelSelector=label_selector, kwargs...)
+        end
     end
 
     # if not watching, retuen the first result
@@ -197,50 +211,62 @@ function get(ctx::Union{KuberContext,KuberWatchContext}, O::Symbol;
     end
 
     # start watch and return the HTTP response object on completion
-    return @retry_on_error apicall(apictx, eventstream, args...; watch=watch, resourceVersion=resourceVersion, labelSelector=label_selector, kwargs...)
+    return k8s_retry(; max_tries=max_tries) do
+        apicall(apictx, eventstream, args...; watch=watch, resourceVersion=resourceVersion, labelSelector=label_selector, kwargs...)
+    end
 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=1,
-        resourceVersion=nothing,
+        max_tries::Int=retries(ctx),
         kwargs...)
     apictx = _get_apictx(ctx, O, apiversion; max_tries=max_tries)
     namespaced = (namespace !== nothing) && !isempty(namespace)
     allnamespaces = namespaced && (namespace == "*")
 
     if allnamespaces
         apicall = eval(Symbol("watch$(O)ForAllNamespaces"))
-        return @retry_on_error apicall(apictx, outstream, name; kwargs...)
+        return k8s_retry(; max_tries=max_tries) do
+            apicall(apictx, outstream, name; kwargs...)
+        end
     elseif namespaced
         apicall = eval(Symbol("watchNamespaced$O"))
-        return @retry_on_error apicall(apictx, outstream, name, namespace; kwargs...)
+        return k8s_retry(; max_tries=max_tries) do
+            apicall(apictx, outstream, name, namespace; kwargs...)
+        end
     else
         apicall = eval(Symbol("watch$O"))
-        return @retry_on_error apicall(apictx, outstream, name; kwargs...)
+        return k8s_retry(; max_tries=max_tries) do
+            apicall(apictx, outstream, name; kwargs...)
+        end
     end
 end
 
 function watch(ctx::KuberContext, O::Symbol, outstream::Channel;
         apiversion::Union{String,Nothing}=nothing,
         namespace::Union{String,Nothing}=ctx.namespace,
-        max_tries::Int=1,
-        resourceVersion=nothing,
+        max_tries::Int=retries(ctx),
         kwargs...)
     apictx = _get_apictx(ctx, O, apiversion; max_tries=max_tries)
     namespaced = (namespace !== nothing) && !isempty(namespace)
     allnamespaces = namespaced && (namespace == "*")
 
     if allnamespaces
         apicall = eval(Symbol("watch$(O)ForAllNamespaces"))
-        return @retry_on_error apicall(apictx, outstream; kwargs...)
+        return k8s_retry(; max_tries=max_tries) do
+            apicall(apictx, outstream; kwargs...)
+        end
     elseif namespaced
         apicall = eval(Symbol("watchNamespaced$O"))
-        return @retry_on_error apicall(apictx, outstream, namespace; kwargs...)
+        return k8s_retry(; max_tries=max_tries) do
+            apicall(apictx, outstream, namespace; kwargs...)
+        end
     else
         apicall = eval(Symbol("watch$O"))
-        return @retry_on_error apicall(apictx, outstream; kwargs...)
+        return k8s_retry(; max_tries=max_tries) do
+            apicall(apictx, outstream; kwargs...)
+        end
     end
 end
 

test/runtests.jl

@@ -15,6 +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)
     Kuber.set_api_versions!(ctx; override=override, verbose=verbose)
     ctx
 end
@@ -308,6 +309,4 @@ function test_all()
     end
 end
 
-#if !parse(Bool, get(ENV, "CI", "false"))
-    test_all()
-#end
+test_all()