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

1 files changed, 15 insertions, 3 deletions

diff --git a/src/zenserver/zenserver.cpp b/src/zenserver/zenserver.cpp
index cfc4a228b..8056b6506 100644
--- a/src/zenserver/zenserver.cpp
+++ b/src/zenserver/zenserver.cpp
@@ -6,6 +6,7 @@
 #include <zencore/filesystem.h>
 #include <zencore/fmtutils.h>
 #include <zencore/iobuffer.h>
+#include <zencore/jobqueue.h>
 #include <zencore/logging.h>
 #include <zencore/refcount.h>
 #include <zencore/scopeguard.h>
@@ -271,6 +272,8 @@ public:
 
 		InitializeState(ServerOptions);
 
+		m_JobQueue = MakeJobQueue(8, "backgroundjobs");
+
 		m_HealthService.SetHealthInfo({.DataRoot		= m_DataRoot,
 									   .AbsLogPath		= ServerOptions.AbsLogFile,
 									   .HttpServerClass = std::string(ServerOptions.HttpServerClass),
@@ -341,7 +344,7 @@ public:
 
 		ZEN_INFO("instantiating project service");
 
-		m_ProjectStore = new zen::ProjectStore(*m_CidStore, m_DataRoot / "projects", m_GcManager);
+		m_ProjectStore = new zen::ProjectStore(*m_CidStore, m_DataRoot / "projects", m_GcManager, *m_JobQueue);
 		m_HttpProjectService.reset(new zen::HttpProjectService{*m_CidStore, m_ProjectStore, m_StatsService, *m_AuthMgr});
 
 #if ZEN_WITH_COMPUTE_SERVICES
@@ -365,7 +368,6 @@ public:
 		}
 
 		m_Http->RegisterService(m_TestService);	 // NOTE: this is intentionally not limited to test mode as it's useful for diagnostics
-		m_Http->RegisterService(m_AdminService);
 
 #if ZEN_WITH_TESTS
 		m_Http->RegisterService(m_TestingService);
@@ -431,6 +433,10 @@ public:
 			.MinimumFreeDiskSpaceToAllowWrites = ServerOptions.GcConfig.MinimumFreeDiskSpaceToAllowWrites};
 		m_GcScheduler.Initialize(GcConfig);
 
+		// Create and register admin interface last to make sure all is properly initialized
+		m_AdminService = std::make_unique<HttpAdminService>(m_GcScheduler, *m_JobQueue);
+		m_Http->RegisterService(*m_AdminService);
+
 		return EffectiveBasePort;
 	}
 
@@ -498,6 +504,11 @@ public:
 		}
 
 		Flush();
+
+		if (m_JobQueue)
+		{
+			m_JobQueue->Stop();
+		}
 	}
 
 	void RequestExit(int ExitCode)
@@ -733,13 +744,14 @@ private:
 	std::unique_ptr<zen::UpstreamCache>				 m_UpstreamCache;
 	std::unique_ptr<zen::HttpUpstreamService>		 m_UpstreamService;
 	std::unique_ptr<zen::HttpStructuredCacheService> m_StructuredCacheService;
-	zen::HttpAdminService							 m_AdminService{m_GcScheduler};
 	zen::HttpHealthService							 m_HealthService;
 #if ZEN_WITH_COMPUTE_SERVICES
 	std::unique_ptr<zen::HttpFunctionService> m_HttpFunctionService;
 #endif	// ZEN_WITH_COMPUTE_SERVICES
 	std::unique_ptr<zen::HttpFrontendService>	 m_FrontendService;
 	std::unique_ptr<zen::HttpObjectStoreService> m_ObjStoreService;
+	std::unique_ptr<JobQueue>					 m_JobQueue;
+	std::unique_ptr<zen::HttpAdminService>		 m_AdminService;
 
 	bool m_DebugOptionForcedCrash = false;
 	bool m_UseSentry			  = false;