From b2cef5900b6e251bed4bc0a02161fd90646d37f0 Mon Sep 17 00:00:00 2001 From: Dan Engelbrecht Date: Wed, 13 Sep 2023 16:13:30 -0400 Subject: job queue and async oplog-import/export (#395) - Feature: New http endpoint for background jobs `/admin/jobs/status` which will return a response listing the currently active background jobs and their status - Feature: New http endpoint for background jobs information `/admin/jobs/status/{jobid}` which will return a response detailing status, pending messages and progress status - GET will return a response detailing status, pending messages and progress status - DELETE will mark the job for cancelling and return without waiting for completion - If status returned is "Complete" or "Aborted" the jobid will be removed from the server and can not be queried again - Feature: New zen command `jobs` to list, get info about and cancel background jobs - If no options are given it will display a list of active background jobs - `--jobid` accepts an id (returned from for example `oplog-export` with `--async`) and will return a response detailing status, pending messages and progress status for that job - `--cancel` can be added when `--jobid` is given which will request zenserver to cancel the background job - Feature: oplog import and export http rpc requests are now async operations that will run in the background - Feature: `oplog-export` and `oplog-import` now reports progress to the console as work progress by default - Feature: `oplog-export` and `oplog-import` can now be cancelled using Ctrl+C - Feature: `oplog-export` and `oplog-import` has a new option `--async` which will only trigger the work and report a background job id back --- src/zenserver/admin/admin.cpp | 140 +++++++++++++++++++++++++++++++++++++++++- 1 file changed, 139 insertions(+), 1 deletion(-) (limited to 'src/zenserver/admin/admin.cpp') diff --git a/src/zenserver/admin/admin.cpp b/src/zenserver/admin/admin.cpp index 575a10d83..74131e624 100644 --- a/src/zenserver/admin/admin.cpp +++ b/src/zenserver/admin/admin.cpp @@ -3,6 +3,7 @@ #include "admin.h" #include +#include #include #include @@ -10,7 +11,9 @@ namespace zen { -HttpAdminService::HttpAdminService(GcScheduler& Scheduler) : m_GcScheduler(Scheduler) +HttpAdminService::HttpAdminService(GcScheduler& Scheduler, JobQueue& BackgroundJobQueue) +: m_GcScheduler(Scheduler) +, m_BackgroundJobQueue(BackgroundJobQueue) { using namespace std::literals; @@ -23,6 +26,141 @@ HttpAdminService::HttpAdminService(GcScheduler& Scheduler) : m_GcScheduler(Sched }, HttpVerb::kGet); + m_Router.AddPattern("jobid", "([[:digit:]]+?)"); + + m_Router.RegisterRoute( + "jobs", + [&](HttpRouterRequest& Req) { + std::vector Jobs = m_BackgroundJobQueue.GetJobs(); + CbObjectWriter Obj; + Obj.BeginArray("jobs"); + for (const auto& Job : Jobs) + { + Obj.BeginObject(); + Obj.AddInteger("Id", Job.Id.Id); + Obj.AddString("Status", JobQueue::ToString(Job.Status)); + Obj.EndObject(); + } + Obj.EndArray(); + Req.ServerRequest().WriteResponse(HttpResponseCode::OK, Obj.Save()); + }, + HttpVerb::kGet); + + m_Router.RegisterRoute( + "jobs/{jobid}", + [&](HttpRouterRequest& Req) { + const auto& JobIdString = Req.GetCapture(1); + std::optional JobIdArg = ParseInt(JobIdString); + if (!JobIdArg) + { + Req.ServerRequest().WriteResponse(HttpResponseCode::BadRequest); + } + JobId Id{.Id = JobIdArg.value_or(0)}; + if (Id.Id == 0) + { + return Req.ServerRequest().WriteResponse(HttpResponseCode::BadRequest, + ZenContentType::kText, + fmt::format("Invalid Job Id: {}", Id.Id)); + } + + std::optional CurrentState = m_BackgroundJobQueue.Get(Id); + if (!CurrentState) + { + return Req.ServerRequest().WriteResponse(HttpResponseCode::NotFound); + } + + auto WriteState = [](CbObjectWriter& Obj, const JobQueue::State& State) { + if (!State.CurrentOp.empty()) + { + Obj.AddString("CurrentOp"sv, State.CurrentOp); + Obj.AddInteger("CurrentOpPercentComplete"sv, State.CurrentOpPercentComplete); + } + if (!State.Messages.empty()) + { + Obj.BeginArray("Messages"); + for (const std::string& Message : State.Messages) + { + Obj.AddString(Message); + } + Obj.EndArray(); + } + }; + + auto GetAgeAsSeconds = [](std::chrono::system_clock::time_point Start, std::chrono::system_clock::time_point End) { + auto Age = End - Start; + auto Milliseconds = std::chrono::duration_cast(Age); + return Milliseconds.count() / 1000.0; + }; + + const std::chrono::system_clock::time_point Now = std::chrono::system_clock::now(); + + switch (CurrentState->Status) + { + case JobQueue::Status::Queued: + { + CbObjectWriter Obj; + Obj.AddString("Status"sv, "Queued"sv); + Obj.AddFloat("QueueTimeS", GetAgeAsSeconds(CurrentState->CreateTime, Now)); + Req.ServerRequest().WriteResponse(HttpResponseCode::OK, Obj.Save()); + } + break; + case JobQueue::Status::Running: + { + CbObjectWriter Obj; + Obj.AddString("Status"sv, "Running"sv); + WriteState(Obj, CurrentState->State); + Obj.AddFloat("QueueTimeS", GetAgeAsSeconds(CurrentState->CreateTime, CurrentState->StartTime)); + Obj.AddFloat("RunTimeS", GetAgeAsSeconds(CurrentState->StartTime, Now)); + Req.ServerRequest().WriteResponse(HttpResponseCode::OK, Obj.Save()); + } + break; + case JobQueue::Status::Aborted: + { + CbObjectWriter Obj; + Obj.AddString("Status"sv, "Aborted"sv); + WriteState(Obj, CurrentState->State); + Obj.AddFloat("QueueTimeS", GetAgeAsSeconds(CurrentState->CreateTime, CurrentState->StartTime)); + Obj.AddFloat("RunTimeS", GetAgeAsSeconds(CurrentState->StartTime, CurrentState->EndTime)); + Obj.AddFloat("CompleteTimeS", GetAgeAsSeconds(CurrentState->EndTime, Now)); + Req.ServerRequest().WriteResponse(HttpResponseCode::OK, Obj.Save()); + } + break; + case JobQueue::Status::Completed: + { + CbObjectWriter Obj; + Obj.AddString("Status"sv, "Complete"sv); + WriteState(Obj, CurrentState->State); + Obj.AddFloat("QueueTimeS", GetAgeAsSeconds(CurrentState->CreateTime, CurrentState->StartTime)); + Obj.AddFloat("RunTimeS", GetAgeAsSeconds(CurrentState->StartTime, CurrentState->EndTime)); + Obj.AddFloat("CompleteTimeS", GetAgeAsSeconds(CurrentState->EndTime, Now)); + Req.ServerRequest().WriteResponse(HttpResponseCode::OK, Obj.Save()); + } + break; + } + }, + HttpVerb::kGet); + + m_Router.RegisterRoute( + "jobs/{jobid}", + [&](HttpRouterRequest& Req) { + const auto& JobIdString = Req.GetCapture(1); + std::optional JobIdArg = ParseInt(JobIdString); + if (!JobIdArg) + { + Req.ServerRequest().WriteResponse(HttpResponseCode::BadRequest); + } + JobId Id{.Id = JobIdArg.value_or(0)}; + if (m_BackgroundJobQueue.CancelJob(Id)) + { + Req.ServerRequest().WriteResponse(HttpResponseCode::OK); + } + else + { + Req.ServerRequest().WriteResponse(HttpResponseCode::NotFound); + } + }, + HttpVerb::kDelete); + m_Router.RegisterRoute( "gc", [this](HttpRouterRequest& Req) { -- cgit v1.2.3