diff --git a/.github/dependabot.yml b/.github/dependabot.yml index 700707c..69e4e87 100644 --- a/.github/dependabot.yml +++ b/.github/dependabot.yml @@ -5,3 +5,7 @@ updates: directory: "/" # Location of package manifests schedule: interval: "weekly" + - package-ecosystem: "julia" + directory: "/" + schedule: + interval: "weekly" diff --git a/.github/workflows/CI.yml b/.github/workflows/CI.yml index 1f8ccdb..b6bb37c 100644 --- a/.github/workflows/CI.yml +++ b/.github/workflows/CI.yml @@ -22,21 +22,21 @@ jobs: matrix: version: - "1.10" + - "1" os: - ubuntu-latest arch: - x64 steps: - - uses: actions/checkout@v4 - - uses: julia-actions/setup-julia@v2 + - uses: actions/checkout@v7 + - uses: julia-actions/setup-julia@v3 with: version: ${{ matrix.version }} arch: ${{ matrix.arch }} - - uses: julia-actions/cache@v2 - - uses: julia-actions/julia-buildpkg@v1 + - uses: julia-actions/cache@v3 - uses: julia-actions/julia-runtest@v1 - uses: julia-actions/julia-processcoverage@v1 - - uses: codecov/codecov-action@v5 + - uses: codecov/codecov-action@v7 with: token: ${{ secrets.CODECOV_TOKEN }} files: lcov.info diff --git a/Project.toml b/Project.toml index dd949b2..7c14d78 100644 --- a/Project.toml +++ b/Project.toml @@ -15,7 +15,7 @@ UUIDs = "cf7118a7-6976-5b1a-9a39-7adc72f591a4" [compat] CodecZlib = "0.7.8" Dates = "1.10" -HTTP = "1.10" +HTTP = "2" JSON = "1.1" Logging = "1.10" PkgVersion = "0.3.3" diff --git a/src/Sentry.jl b/src/Sentry.jl index b07e54b..0f32d68 100644 --- a/src/Sentry.jl +++ b/src/Sentry.jl @@ -27,8 +27,12 @@ export capture_message, Error, init -function init(dsn=nothing ; traces_sample_rate=nothing, traces_sampler=nothing, debug=false, release=nothing) - main_hub.initialised && @warn "Sentry already initialised." +function init(dsn=nothing ; traces_sample_rate=nothing, traces_sampler=nothing, debug=false, release=nothing, shutdown_timeout=DEFAULT_SHUTDOWN_TIMEOUT) + if main_hub.initialised + # Returning early, otherwise we would leak another send_worker task. + @warn "Sentry already initialised." + return nothing + end if dsn === nothing dsn = get(ENV, "SENTRY_DSN", nothing) if dsn === nothing @@ -38,12 +42,8 @@ function init(dsn=nothing ; traces_sample_rate=nothing, traces_sampler=nothing, end end - if !main_hub.initialised - atexit(clear_queue) - end - - main_hub.debug = debug + main_hub.shutdown_timeout = shutdown_timeout main_hub.dsn = dsn upstream, project_id, public_key = parse_dsn(dsn) @@ -62,7 +62,8 @@ function init(dsn=nothing ; traces_sample_rate=nothing, traces_sampler=nothing, main_hub.traces_sampler = NoSamples() end - main_hub.sender_task = @async send_worker() + main_hub.sender_task = Threads.@spawn send_worker() + atexit(clear_queue) bind(main_hub.queued_tasks, main_hub.sender_task) main_hub.initialised = true @@ -158,10 +159,12 @@ function PrepareBody(event::Event, buf) println(buf, JSON.json(item_header)) println(buf, item_str) - for attachment in event.attachments + for (i, attachment) in enumerate(event.attachments) attachment_str = JSON.json((;data=attachment)) + # `filename` is required by sentry for attachment items. attachment_header = (; type="attachment", length=sizeof(attachment_str), + filename="attachment-$i.json", content_type="application/json") println(buf, JSON.json(attachment_header)) @@ -214,6 +217,7 @@ function PrepareBody(transaction::Transaction, buf) # root_span..., root_span.start_timestamp, root_span.timestamp, + main_hub.release, tags = MergeTags(global_tags, root_span.tags), contexts = (; trace), @@ -223,7 +227,7 @@ function PrepareBody(transaction::Transaction, buf) item_header = (; type="transaction", content_type="application/json", - length=sizeof(item_str)+1) # +1 for the newline to come + length=sizeof(item_str)) println(buf, JSON.json(envelope_header)) @@ -254,7 +258,7 @@ function send_envelope(task::TaskPayload) if main_hub.dsn === "fake" body = String(transcode(CodecZlib.GzipDecompressor, body)) lines = map(eachline(IOBuffer(body))) do line - line = JSON.Parser.parse(line) + line = JSON.parse(line) line = JSON.json(line, 4) end @info "Would have sent this body" @@ -263,18 +267,19 @@ function send_envelope(task::TaskPayload) end r = HTTP.request("POST", target, headers, body) if r.status == 200 - return r.body + return Vector{UInt8}(r.body) else - throw(HTTP.Exceptions.StatusError(r.status, "POST", target, r)) + throw(HTTP.Exceptions.StatusError(r.status, r)) end return nothing end function send_worker() - while true + # Iterating the channel drains whatever is still buffered once it has been + # closed, and then finishes, which is what lets clear_queue wait for the + # sends themselves to complete rather than just for the queue to empty. + for event in main_hub.queued_tasks try - event = take!(main_hub.queued_tasks) - yield() send_envelope(event) catch exc if main_hub.debug @@ -286,10 +291,9 @@ function send_worker() end function clear_queue() - while isready(main_hub.queued_tasks) - @info "Waiting for queue to finish before closing" - # send_envelope(take!(main_hub.queued_tasks)) - sleep(1) + close(main_hub.queued_tasks) + if timedwait(() -> istaskdone(main_hub.sender_task), main_hub.shutdown_timeout) === :timed_out + @warn "Timed out sending queued events to sentry" end end @@ -300,7 +304,19 @@ end function capture_event(task::TaskPayload) main_hub.initialised || return - push!(main_hub.queued_tasks, task) + try + push!(main_hub.queued_tasks, task) + catch exc + if !(exc isa InvalidStateException) + rethrow() + end + + # The queue is closed while shutting down, so there is nothing left to + # send it to. Never let that propagate into the calling program. + if main_hub.debug + @error "Could not queue an event for sentry" exc + end + end end function capture_message(message, level::LogLevel=Info ; kwds...) @@ -323,7 +339,7 @@ end # This assumes that we are calling from within a catch capture_exception(exc::Exception) = capture_exception([(exc, catch_backtrace())]) -function capture_exception(exceptions=catch_stack()) +function capture_exception(exceptions=Base.current_exceptions()) main_hub.initialised || return formatted_excs = map(exceptions) do (exc,strace) diff --git a/src/structs.jl b/src/structs.jl index 6e7693f..76e52ee 100644 --- a/src/structs.jl +++ b/src/structs.jl @@ -53,6 +53,10 @@ sample(sampler::RatioSampler) = rand() < sampler.ratio sample(sampler::Function) = sampler() const TaskPayload = Union{Event,Transaction} + +# Seconds to wait for queued events to be sent while the program is exiting +const DEFAULT_SHUTDOWN_TIMEOUT = 10.0 + # This is to supposedly support the "unified api" of the sentry sdk. I'm not a # fan, so it will only go partway to this goal. # Note: a proper implementation here would make Hub a module. @@ -68,6 +72,7 @@ Base.@kwdef mutable struct Hub release::Union{Nothing,String} = nothing debug::Bool = false + shutdown_timeout::Float64 = DEFAULT_SHUTDOWN_TIMEOUT last_send_time = nothing queued_tasks = Channel{TaskPayload}(100) diff --git a/src/transactions.jl b/src/transactions.jl index 3efd1b3..0acc976 100644 --- a/src/transactions.jl +++ b/src/transactions.jl @@ -3,7 +3,13 @@ # * Transactions #---------------------------- -struct InhibitTransaction end +# Mimics a transaction, so that it is cleared from the task local storage once +# the outermost inhibited transaction finishes. Otherwise a single declined +# sample would lock out tracing for the remaining life of the task. +mutable struct InhibitTransaction + num_open_spans::Int +end +InhibitTransaction() = InhibitTransaction(0) function start_transaction(func ; kwds...) previous = get(task_local_storage(), :sentry_transaction, nothing) @@ -20,7 +26,7 @@ function start_transaction(; name="", force_new=(name!=""), trace_id=:auto, pare # trace_id === nothing && return nothing # Need to pass through nothings so that we can hit an InhibitTransaction t = get_transaction(; name, trace_id, force_new) - if t === nothing || t === InhibitTransaction() + if t === nothing || t isa InhibitTransaction return t end @@ -45,7 +51,13 @@ function finish_transaction(current, previous) task_local_storage(:sentry_transaction, previous) end finish_transaction(::Nothing) = nothing -finish_transaction(::InhibitTransaction) = nothing +function finish_transaction(inhibit::InhibitTransaction) + inhibit.num_open_spans -= 1 + if inhibit.num_open_spans <= 0 + task_local_storage(:sentry_transaction, nothing) + end + nothing +end function finish_transaction((transaction, parent_span, span)) complete(span) if transaction.root_span !== span @@ -70,11 +82,12 @@ function get_transaction(; force_new=false, trace_id=:auto, kwds...) transaction = get(task_local_storage(), :sentry_transaction, nothing) end - if transaction === InhibitTransaction() + if transaction isa InhibitTransaction + transaction.num_open_spans += 1 return transaction elseif transaction === nothing if trace_id === nothing - transaction = InhibitTransaction() + transaction = InhibitTransaction(1) task_local_storage(:sentry_transaction, transaction) return transaction elseif sample(main_hub.traces_sampler) @@ -83,15 +96,10 @@ function get_transaction(; force_new=false, trace_id=:auto, kwds...) end transaction = Transaction(; trace_id = trace_id, kwds...) else - transaction = InhibitTransaction() + transaction = InhibitTransaction(1) task_local_storage(:sentry_transaction, transaction) return transaction end - # TODO: Note that the cases which store an InhibitTransaction in here - # are bad. They will lock out the start_transaction context manager from - # taking effect in any future cases. Instead, we should track this and - # later undo its effects once the outermost transaction is completed - # (meaning the InhibitTransaction itself should mimic a transaction). task_local_storage(:sentry_transaction, transaction) return (; transaction, parent_span=nothing) else @@ -106,7 +114,9 @@ end set_task_transaction(::Nothing) = nothing function set_task_transaction(::InhibitTransaction) - task_local_storage(:sentry_transaction, InhibitTransaction()) + # Starts at 1 with no matching finish, so that the parent's sampling + # decision holds for the whole life of this task. + task_local_storage(:sentry_transaction, InhibitTransaction(1)) end function set_task_transaction((transaction, ignored, parent_span)) task_local_storage(:sentry_transaction, transaction) diff --git a/test/runtests.jl b/test/runtests.jl index 71292c3..b25015b 100644 --- a/test/runtests.jl +++ b/test/runtests.jl @@ -1,7 +1,43 @@ using Sentry using Test -Sentry.init() +using CodecZlib +using HTTP +using JSON + +# Collects the envelopes that Sentry.jl would have sent to a real sentry server. +const received = Channel{String}(16) + +# Delays the response, to emulate a sentry server that is slower than the local +# one. Nothing is recorded if the client gives up in the meantime. +const response_delay = Ref(0.0) + +const server = HTTP.serve!("127.0.0.1", 0; listenany=true) do request + sleep(response_delay[]) + put!(received, String(transcode(GzipDecompressor, Vector{UInt8}(request.body)))) + HTTP.Response(200, "ok") +end + +Sentry.init("http://abcdef1234567890@127.0.0.1:$(HTTP.port(server))/42"; + traces_sample_rate=1.0, + release="v1.2.3", + shutdown_timeout=20.0) + +""" +Wait for the next envelope and split it into its headers and payloads. +""" +function next_envelope(timeout=15) + deadline = time() + timeout + while !isready(received) && time() < deadline + sleep(0.05) + end + if !isready(received) + error("Timed out waiting for an envelope") + end + + items = filter(!isempty, split(take!(received), '\n')) + return map(JSON.parse, items), items +end @testset "Sentry.jl" begin @test Sentry.parse_dsn("fake") == (upstream = "", project_id = "", public_key = "") @@ -15,7 +51,144 @@ Sentry.init() @test length(Sentry.generate_uuid4()) == 32 + @test Sentry.main_hub.shutdown_timeout == 20.0 + @test Sentry.Hub().shutdown_timeout == Sentry.DEFAULT_SHUTDOWN_TIMEOUT + d = Sentry.FilterNothings([1, nothing, 2]) @test d[3] == 2 @test d[1] == 1 + + @testset "capture_message" begin + capture_message("hello", Warn; attachments=[(; command="ls")]) + parsed, items = next_envelope() + _, event_header, event, attachment_header, attachment = parsed + + @test event_header["type"] == "event" + @test event["level"] == "warning" + @test event["message"]["formatted"] == "hello" + @test event["release"] == "v1.2.3" + @test event["tags"]["test"] == "message" + + # A declared length must not include the newline separating the items. + @test event_header["length"] == sizeof(items[3]) + @test attachment_header["length"] == sizeof(items[5]) + + @test attachment_header["type"] == "attachment" + @test attachment_header["filename"] == "attachment-1.json" + @test attachment["data"]["command"] == "ls" + end + + @testset "capture_exception" begin + try + error("boom") + catch exc + capture_exception(exc) + end + parsed, _ = next_envelope() + exception = parsed[3]["exception"]["values"][1] + @test exception["type"] == "ErrorException" + @test exception["value"] == "boom" + + # The zero argument method reads the current exception stack. + try + error("implicit boom") + catch + capture_exception() + end + parsed, _ = next_envelope() + @test parsed[3]["exception"]["values"][1]["value"] == "implicit boom" + end + + @testset "transactions" begin + start_transaction(name="job", op="task") do _ + start_transaction(op="child", description="inner") do _ end + end + + parsed, items = next_envelope() + _, header, transaction = parsed + + @test header["type"] == "transaction" + @test header["length"] == sizeof(items[3]) + @test transaction["transaction"] == "job" + @test transaction["release"] == "v1.2.3" + + trace = transaction["contexts"]["trace"] + @test trace["op"] == "task" + @test length(transaction["spans"]) == 1 + @test transaction["spans"][1]["parent_span_id"] == trace["span_id"] + @test transaction["spans"][1]["trace_id"] == trace["trace_id"] + end + + @testset "inhibited transactions" begin + # An inhibited transaction must not lock out tracing for the rest of the + # task, so run this in its own task to keep the check honest. The + # results are collected because the task has its own testset state. + results = fetch(@async begin + inhibited = start_transaction(trace_id=nothing) + nested = start_transaction(op="nested") + + finish_transaction(inhibited) + after_nested = task_local_storage(:sentry_transaction) + finish_transaction(inhibited) + after_outer = task_local_storage(:sentry_transaction) + + # Tracing works again now that the inhibition has been undone. + t = start_transaction(name="after") + finish_transaction(t) + + (; inhibited, nested, after_nested, after_outer, t) + end) + + @test results.inhibited isa Sentry.InhibitTransaction + @test results.nested === results.inhibited + @test results.after_nested === results.inhibited + @test results.after_outer === nothing + @test results.t.transaction isa Sentry.Transaction + + parsed, _ = next_envelope() + @test parsed[3]["transaction"] == "after" + end + + @testset "flushing on exit" begin + # Needs a real process exit to run the atexit handler, and deliberately + # does not wait for the send itself. The response is delayed so that + # just emptying the queue is not enough to get the event through. + code = """ + using Sentry + Sentry.init("http://abcdef1234567890@127.0.0.1:$(HTTP.port(server))/42") + capture_message("sent while exiting") + """ + + response_delay[] = 3.0 + try + elapsed = @elapsed run(`$(Base.julia_cmd()) --project=$(Base.active_project()) --eval $code`) + + # Exiting has to block until the send itself came back + @test elapsed > response_delay[] + + parsed, _ = next_envelope() + @test parsed[3]["message"]["formatted"] == "sent while exiting" + finally + response_delay[] = 0.0 + end + end + + @testset "fake dsn" begin + # The fake dsn pretty prints the envelope instead of sending it. + dsn = Sentry.main_hub.dsn + Sentry.main_hub.dsn = "fake" + try + @test_logs (:info, "Would have sent this body") match_mode=:any Sentry.send_envelope(Sentry.Event(message=(; formatted="dry run"))) + finally + Sentry.main_hub.dsn = dsn + end + end + + # Kept last because re-initialising must not disturb the running hub. + @testset "re-initialising" begin + @test_warn "Sentry already initialised." Sentry.init("http://cafe@127.0.0.1:1/99") + @test Sentry.main_hub.dsn == "http://abcdef1234567890@127.0.0.1:$(HTTP.port(server))/42" + end end + +close(server)