chore: remove code for Kafka logging (#128)

Author: mortenpi

src/JuliaHub.jl

@@ -33,7 +33,6 @@ include("PackageBundler/PackageBundler.jl")
 include("jobs/jobs.jl")
 include("jobs/request.jl")
 include("jobs/logging.jl")
-include("jobs/logging-kafka.jl")
 include("jobs/logging-legacy.jl")
 include("packages.jl")
 include("projects.jl")

src/jobs/logging-kafka.jl

@@ -24,7 +24,7 @@ function JobLogMessage(::_LegacyLogging, json::AbstractDict, offset::Integer)
     eventId = _get_json_or(json, "eventId", String, nothing)
     JobLogMessage(;
         _offset=offset, timestamp, message, _metadata=metadata, _keywords=keywords,
-        _legacy_eventId=eventId, _kafka_stream=nothing, _json=json,
+        _legacy_eventId=eventId, _json=json,
     )
 end
 

src/jobs/logging-legacy.jl

@@ -1,36 +1,5 @@
 abstract type _JobLoggingAPIVersion end
 
-module _LoggingMode
-@enum T NOKAFKA AUTOMATIC FORCEKAFKA
-end
-const _OPTION_LoggingMode = Ref{_LoggingMode.T}(_LoggingMode.NOKAFKA)
-
-function _job_logging_api_version(
-    auth::Authentication, jobname::AbstractString
-)::_JobLoggingAPIVersion
-    # For debugging, development, and testing purposes, we allow the user to force the
-    # logging backend to use a particular endpoint. This is not a documented behaviour
-    # and should not be relied on.
-    if _OPTION_LoggingMode[] == _LoggingMode.NOKAFKA
-        return _LegacyLogging()
-    elseif _OPTION_LoggingMode[] == _LoggingMode.FORCEKAFKA
-        return _KafkaLogging()
-    elseif _OPTION_LoggingMode[] == _LoggingMode.AUTOMATIC
-        query = Dict("jobname" => jobname)
-        r = _restcall(auth, :HEAD, "juliaruncloud", "get_logs_v2"; query)
-        # If HEAD /juliaruncloud/get_logs_v2 returns a 200, then we know we can try to fetch
-        # the newer Kafka logs. If it returns anything else, we will try use the old endpoint.
-        # However, it _should_ return a 404 in the latter case, and we'll warn if it returns a
-        # different code.
-        r.status == 200 && return _KafkaLogging()
-        if r.status != 404
-            @warn "Unexpected response from HEAD /juliaruncloud/get_logs_v2" r.status r.body
-        end
-        return _LegacyLogging()
-    end
-    error("Invalid _OPTION_LoggingMode: $(_OPTION_LoggingMode[])")
-end
-
 """
     struct JobLogMessage
 
@@ -53,7 +22,6 @@ Base.@kwdef struct JobLogMessage
     _metadata::Dict{String, Any} # `metadata :: Dict{String, Any}`: additional metadata with not guaranteed fields; may also be empty (TODO)
     _keywords::Dict{String, Any} # `keywords :: Dict{String, Any}`: additional metadata with not guaranteed fields; may also be empty (TODO)
     _legacy_eventId::Union{String, Nothing}
-    _kafka_stream::Union{String, Nothing}
     _json::Dict
 end
 
@@ -254,11 +222,7 @@ function job_logs_buffered(
     stream::Bool=false,
     auth::Authentication=__auth__(),
 )
-    if _job_logging_api_version(auth, jobname) == _KafkaLogging()
-        return KafkaLogsBuffer(f, auth; jobname, offset, stream)
-    else
-        return _LegacyLogsBuffer(f, auth; jobname, offset, stream)
-    end
+    return _LegacyLogsBuffer(f, auth; jobname, offset, stream)
 end
 job_logs_buffered(job::Union{AbstractString, Job}; kwargs...) =
     job_logs_buffered(_noop, job; kwargs...)

src/jobs/logging.jl

@@ -15,9 +15,6 @@ function JET.configured_reports(::JuliaHubReportFilter, reports::Vector{JET.Infe
         if isa(report, JET.MethodErrorReport) && report.union_split > 1
             return false
         end
-        # The Kafka code (which is currently not enabled) has a few bugs that JET actually
-        # reveals. But we ignore them for now.
-        endswith(string(report.vst[end].file), "logging-kafka.jl") && return false
         # We also ignore the _restput_mockable() error in restapi.jl, since JET seems to
         # assume that kwargs... must be non-empty
         contains(string(report.vst[end].linfo.def.name), "_restput_mockable") && return false

test/jet.jl

@@ -656,9 +656,9 @@ end
     end
 end
 
-function logging_mocking_wrapper(f::Base.Callable, testset_name::AbstractString; legacy=false)
+function logging_mocking_wrapper(f::Base.Callable, testset_name::AbstractString)
     global MOCK_JULIAHUB_STATE
-    logengine = LogEngine(; kafkalogging=!legacy)
+    logengine = LogEngine()
     MOCK_JULIAHUB_STATE[:logengine] = logengine
     try
         Mocking.apply(mocking_patch) do
@@ -671,40 +671,24 @@ function logging_mocking_wrapper(f::Base.Callable, testset_name::AbstractString;
     end
 end
 
-JuliaHub._OPTION_LoggingMode[] = JuliaHub._LoggingMode.AUTOMATIC
-@testset "Job logs: legacy = $legacy" for legacy in [true, false]
-    logging_mocking_wrapper("Basic logging"; legacy=legacy) do logengine
+@testset "Job logs" begin
+    logging_mocking_wrapper("Basic logging") do logengine
         @testset "Invalid requests" begin
             # Negative offsets are not allowed
             @test_throws ArgumentError JuliaHub.job_logs_buffered("jr-test1"; offset=-1)
             # First, just double check that a missing job 403s on the backend
             @test_throws JuliaHub.PermissionError JuliaHub.job_logs_buffered("jr-test1"; offset=0)
         end
-        # Let's check that we are using the correct backend
-        @testset "Dispatching on backend" begin
-            auth = JuliaHub.current_authentication()
-            # If the job is not present, then the Kafka backend will always be disabled
-            @test JuliaHub._job_logging_api_version(auth, "jr-test1") == JuliaHub._LegacyLogging()
-            # Let's add a finished job without any logs. Since these jobs are marked as finished,
-            # we should not add logs to the "backend" after the buffer has been constructed.
-            logengine.jobs["jr-test1"] = LogEngineJob([])
-            # After the job is added, the expected backend depends on whether we're testing for legacy
-            # of the Kafka backend.
-            expected_backend = legacy ? JuliaHub._LegacyLogging() : JuliaHub._KafkaLogging()
-            @test JuliaHub._job_logging_api_version(auth, "jr-test1") == expected_backend
-            # Let's also test the override variable
-            JuliaHub._OPTION_LoggingMode[] = JuliaHub._LoggingMode.FORCEKAFKA
-            @test JuliaHub._job_logging_api_version(auth, "jr-test1") == JuliaHub._KafkaLogging()
-            JuliaHub._OPTION_LoggingMode[] = JuliaHub._LoggingMode.NOKAFKA
-            @test JuliaHub._job_logging_api_version(auth, "jr-test1") == JuliaHub._LegacyLogging()
-            JuliaHub._OPTION_LoggingMode[] = JuliaHub._LoggingMode.AUTOMATIC
-        end
         # Test the fetching of logs
+        auth = JuliaHub.current_authentication()
+        # Let's add a finished job without any logs. Since these jobs are marked as finished,
+        # we should not add logs to the "backend" after the buffer has been constructed.
+        logengine.jobs["jr-test1"] = LogEngineJob([])
         @testset "Zero logs" begin
             let lb = JuliaHub.job_logs_buffered("jr-test1"; offset=0)
                 # Just one check to make sure that the returned buffer actually matches the backend
                 # that we're trying to test.
-                @test lb isa (legacy ? JuliaHub._LegacyLogsBuffer : JuliaHub.KafkaLogsBuffer)
+                @test lb isa JuliaHub._LegacyLogsBuffer
                 @test length(lb.logs) == 0
                 @test JuliaHub.hasfirst(lb)
                 @test JuliaHub.haslast(lb)
@@ -937,7 +921,6 @@ JuliaHub._OPTION_LoggingMode[] = JuliaHub._LoggingMode.AUTOMATIC
         end
     end
 end
-JuliaHub._OPTION_LoggingMode[] = JuliaHub._LoggingMode.NOKAFKA
 
 @testset "JuliaHub.submit_job: expose=" begin
     empty!(MOCK_JULIAHUB_STATE)

test/jobs.jl

@@ -475,15 +475,6 @@ function _restcall_mocked(method, url, headers, payload; query)
         else
             serve_legacy(logengine, Dict(query))
         end
-    elseif in(method, (:HEAD, :GET)) && endswith(url, "/juliaruncloud/get_logs_v2")
-        logengine = get(MOCK_JULIAHUB_STATE, :logengine, nothing)
-        if isnothing(logengine)
-            JuliaHub._RESTResponse(500, "MOCK_JULIAHUB_STATE[:logengine] not set up")
-        elseif !logengine.kafkalogging
-            JuliaHub._RESTResponse(404, "MOCK_JULIAHUB_STATE[:logengine]: Kafka disabled")
-        else
-            serve_kafka(logengine, method, Dict(query))
-        end
     elseif (method == :GET) && occursin("/juliaruncloud/product_image_groups", url)
         Dict(
             "image_groups" => Dict(
@@ -532,91 +523,10 @@ end
 
 Base.@kwdef mutable struct LogEngine
     jobs::Dict{String, LogEngineJob} = Dict()
-    kafkalogging::Bool = false
-    # The real legacy endpoint has a limit of 10k. Similarly, for the real
-    # kafka endpoint this is determined by the max_bytes of the response,
-    # so it can actually vary.
+    # The real legacy endpoint has a limit of 10k
     max_response_size::Int = 10
 end
 
-function serve_kafka(logengine::LogEngine, method::Symbol, query::AbstractDict)
-    jobname = get(query, "jobname", nothing)
-    job = get(logengine.jobs, jobname, nothing)
-    # If the client is doing a HEAD request, then we immediately return the
-    # approriate empty response.
-    if method == :HEAD
-        if isnothing(jobname)
-            return JuliaHub._RESTResponse(400, "")
-        elseif isnothing(job)
-            return JuliaHub._RESTResponse(404, "")
-        else
-            return JuliaHub._RESTResponse(200, "")
-        end
-    end
-    # Error handing for the GET requests (like for HEAD, but with a body)
-    if isnothing(jobname)
-        return JuliaHub._RESTResponse(400, "jobname is missing")
-    elseif isnothing(job)
-        return JuliaHub._RESTResponse(404, "No such job $jobname")
-    end
-    # Return the normal response
-    offset = get(query, "offset", nothing)
-    # We'll construct the full list of logs for the job, including the meta message
-    # at the end if necessary.
-    logs::Vector{Any} = map(enumerate(job.logs)) do (i, log)
-        # Make the indexing start from 0, to match the Kafka offset logic
-        (i - 1, log)
-    end
-    # We'll add the meta=bottom message, if needed.
-    if !job.isrunning
-        push!(logs, (length(logs), :bottom))
-    end
-    logs = if isnothing(offset)
-        start = max(1, length(logs) - logengine.max_response_size + 1)
-        logs[start:end]
-    elseif offset + 1 <= length(logs)
-        start = offset + 1
-        stop = min(start + logengine.max_response_size - 1, length(logs))
-        logs[start:stop]
-    else
-        [] # For out of range offsets we just return an empty list of logs
-    end
-    start_timestamp = Dates.now()
-    logs = map(logs) do (i, log)
-        value = if isa(log, Symbol)
-            Dict("meta" => string(log))
-        else
-            Dict(
-                "timestamp" =>
-                    JuliaHub._log_legacy_datetime_to_ms(start_timestamp + Dates.Second(i)),
-                "log" => Dict(
-                    "message" => String(log),
-                    "keywords" => Dict(
-                        "typeof(logger)" => "LoggingExtras...",
-                        "jrun_hostname" => "jr-ecogm4cccn-x5z4g",
-                        "jrun_worker_id" => 1,
-                        "jrun_thread_id" => 1,
-                        "jrun_time" => start_timestamp,
-                        "jrun_process_id" => 27,
-                    ),
-                    "metadata" => Dict(
-                        "line" => 141,
-                        "id" => "Main_JuliaRunJob_53527a33",
-                        "_module" => "Main.JuliaRunJob",
-                        "filepath" => "/opt/juliahub/master_startup.jl",
-                        "group" => "master_startup",
-                        "level" => "Info",
-                        "steam" => "stdout",
-                    ),
-                ),
-            )
-        end
-        Dict("offset" => i, "value" => value)
-    end
-    logs_json = JSON.json(Dict("consumer_id" => 1234, "logs" => logs))
-    return JuliaHub._RESTResponse(200, logs_json)
-end
-
 function serve_legacy(logengine::LogEngine, query::AbstractDict)
     jobname = get(query, "jobname", nothing)
     job = get(logengine.jobs, jobname, nothing)

test/mocking.jl

@@ -209,7 +209,7 @@ end
             :extend_job,
             :interrupt!, :isdone, :job, :job_file, :job_files,
             :job_logs, :job_logs_buffered, :job_logs_newer!, :job_logs_older!,
-            :AbstractJobLogsBuffer, :KafkaLogsBuffer,
+            :AbstractJobLogsBuffer,
             :hasfirst, :haslast, :jobs, :kill_job,
             :nodespec, :nodespecs, :reauthenticate!, :submit_job,
             :update_dataset, :upload_dataset, :wait_job,