diff --git a/Project.toml b/Project.toml index d0c229b..9919848 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.9" +version = "0.7.10" [deps] Dates = "ade2ca70-3891-5945-98fb-dc099432e06a" diff --git a/src/helpers.jl b/src/helpers.jl index d7eb2da..51dd737 100644 --- a/src/helpers.jl +++ b/src/helpers.jl @@ -243,9 +243,9 @@ end # OpenAPI conversions insist that JSONs objects are always `Dict{String,Any}`. # To ensure that for a user supplied Dict, we serialize that to string and parse it back as json. -kuber_obj(ctx::KuberContext, j::Dict{String,Any}) = kuber_obj(ctx, JSON.json(j)) +kuber_obj(ctx::KuberContext, j::AbstractDict) = kuber_obj(ctx, JSON.json(j)) kuber_obj(ctx::KuberContext, data::String) = _kuber_obj(ctx, _parse_json(data)) -_kuber_obj(ctx::KuberContext, j::Dict{String,Any}) = convert(kind_to_type(ctx, j["kind"], get(j, "apiVersion", nothing)), j) +_kuber_obj(ctx::KuberContext, j::AbstractDict) = convert(kind_to_type(ctx, j["kind"], get(j, "apiVersion", nothing)), j) show(io::IO, ctx::KuberContext) = print(io, "Kubernetes namespace ", ctx.namespace, " at ", ctx.client.root) diff --git a/src/simpleapi.jl b/src/simpleapi.jl index 04943fa..c20d15e 100644 --- a/src/simpleapi.jl +++ b/src/simpleapi.jl @@ -40,7 +40,16 @@ function watch(streamprocessor::Function, ctx::KuberContext, watched::Function, finally close(stream) end - @async streamprocessor(stream) + @async try + streamprocessor(stream) + finally + # Symmetric to the watcher task above: if the stream processor + # dies (e.g. an exception while converting an event), close the + # stream so the HTTP watch task aborts too. Otherwise `@sync` + # silently waits for the connection to end while events pile up + # unconsumed — a deaf watch with no error surfaced. + close(stream) + end end end end diff --git a/test/runtests.jl b/test/runtests.jl index 96851bb..adebc9c 100644 --- a/test/runtests.jl +++ b/test/runtests.jl @@ -314,6 +314,15 @@ function test_versioned(ctx, testid) @test !isempty(events) for event in events @test isa(event, Union{Typedefs.CoreV1.WatchEvent,Typedefs.CoreV1.PodList}) + # Watch event objects parse to `JSON.Object` under JSON.jl 1.x — + # an AbstractDict, not a Dict{String,Any}. `kuber_obj` must accept + # them: a MethodError here used to kill the stream processor and + # leave the watch silently deaf (see the watch-processor-failure + # test below for the propagation side). + if isa(event, Typedefs.CoreV1.WatchEvent) + obj = Kuber.kuber_obj(ctx, event.object) + @test isa(obj, OpenAPI.APIModel) + end end end end @@ -350,6 +359,36 @@ function test_all() end end +function test_watch_processor_failure() + # A `streamprocessor` that throws must abort the watch promptly and + # propagate the error. Before the fix, the processor task died silently + # while `@sync` kept waiting on the (long-running) `watched` task — a + # deaf watch: events kept buffering with no error surfaced until the + # server dropped the connection. + ctx = KuberContext() # only passed through to KuberWatchContext; no server needed + producer = (watchctx) -> begin + # mimic the HTTP watch task: keep streaming events; ends only when the + # stream is closed under it (put! on a closed channel throws) + i = 0 + while true + put!(watchctx.stream, (i += 1)) + sleep(0.05) + end + end + t0 = time() + @test_throws Exception watch(ctx, producer) do stream + take!(stream) + error("processor failure") + end + # must fail fast — processor death closes the stream, which kills the + # producer's next put! — not linger until the producer would have ended + @test (time() - t0) < 10.0 +end + +@testset "Watch processor failure aborts watch" begin + test_watch_processor_failure() +end + test_all() # Shutdown the kubectl proxy if we are running on github CI.