Cache request record/replay (#198)

This adds recording and playback of cache request with full data - both get and put operations can be replayed.

Invoke via web request.

`<host>/z$/exec$/start-recording?<disk-storage-path>`
`<host>/z$/exec$/stop-recording`
`<host>/z$/exec$/replay-recording?<thread-count>&<disk-storage-path>`

1 files changed, 47 insertions, 20 deletions

diff --git a/zenserver/cache/structuredcache.h b/zenserver/cache/structuredcache.h
index a74d6b7a6..a2a4a940c 100644
--- a/zenserver/cache/structuredcache.h
+++ b/zenserver/cache/structuredcache.h
@@ -26,10 +26,33 @@ class ScrubContext;
 class UpstreamCache;
 class ZenCacheStore;
 enum class CachePolicy : uint32_t;
+enum class RpcAcceptOptions : uint16_t;
 
 namespace cache::detail {
 	struct RecordBody;
 	struct ChunkRequest;
+	struct RecordedRequests;
+
+	class IRequestRecorder
+	{
+	public:
+		virtual ~IRequestRecorder() {}
+		virtual uint64_t IRequestRecorder_RecordRequest(const ZenContentType ContentType, const IoBuffer& RequestBuffer) = 0;
+		virtual void	 IRequestRecorder_RecordResponse(uint64_t			  RequestIndex,
+														 const ZenContentType ContentType,
+														 const IoBuffer&	  ResponseBuffer)								 = 0;
+		virtual void	 IRequestRecorder_RecordResponse(uint64_t				RequestIndex,
+														 const ZenContentType	ContentType,
+														 const CompositeBuffer& ResponseBuffer)							 = 0;
+	};
+	class IRequestReplayer
+	{
+	public:
+		virtual ~IRequestReplayer() {}
+		virtual uint64_t	   IRequestReplayer_GetRequestCount() const									= 0;
+		virtual ZenContentType IRequestReplayer_GetRequest(uint64_t RequestIndex, IoBuffer& OutBuffer)	= 0;
+		virtual ZenContentType IRequestRecorder_GetResponse(uint64_t RequestIndex, IoBuffer& OutBuffer) = 0;
+	};
 }  // namespace cache::detail
 
 /**
@@ -98,18 +121,24 @@ private:
 		Invalid,
 	};
 
-	void		 HandleCacheRecordRequest(zen::HttpServerRequest& Request, const CacheRef& Ref, CachePolicy PolicyFromUrl);
-	void		 HandleGetCacheRecord(zen::HttpServerRequest& Request, const CacheRef& Ref, CachePolicy PolicyFromUrl);
-	void		 HandlePutCacheRecord(zen::HttpServerRequest& Request, const CacheRef& Ref, CachePolicy PolicyFromUrl);
-	void		 HandleCacheChunkRequest(zen::HttpServerRequest& Request, const CacheRef& Ref, CachePolicy PolicyFromUrl);
-	void		 HandleGetCacheChunk(zen::HttpServerRequest& Request, const CacheRef& Ref, CachePolicy PolicyFromUrl);
-	void		 HandlePutCacheChunk(zen::HttpServerRequest& Request, const CacheRef& Ref, CachePolicy PolicyFromUrl);
-	void		 HandleRpcRequest(zen::HttpServerRequest& Request);
-	void		 HandleRpcPutCacheRecords(zen::HttpServerRequest& Request, const CbPackage& BatchRequest);
-	void		 HandleRpcGetCacheRecords(zen::HttpServerRequest& Request, CbObjectView BatchRequest);
-	void		 HandleRpcPutCacheValues(zen::HttpServerRequest& Request, const CbPackage& BatchRequest);
-	void		 HandleRpcGetCacheValues(zen::HttpServerRequest& Request, CbObjectView BatchRequest);
-	void		 HandleRpcGetCacheChunks(zen::HttpServerRequest& Request, CbObjectView BatchRequest);
+	void HandleCacheRecordRequest(zen::HttpServerRequest& Request, const CacheRef& Ref, CachePolicy PolicyFromUrl);
+	void HandleGetCacheRecord(zen::HttpServerRequest& Request, const CacheRef& Ref, CachePolicy PolicyFromUrl);
+	void HandlePutCacheRecord(zen::HttpServerRequest& Request, const CacheRef& Ref, CachePolicy PolicyFromUrl);
+	void HandleCacheChunkRequest(zen::HttpServerRequest& Request, const CacheRef& Ref, CachePolicy PolicyFromUrl);
+	void HandleGetCacheChunk(zen::HttpServerRequest& Request, const CacheRef& Ref, CachePolicy PolicyFromUrl);
+	void HandlePutCacheChunk(zen::HttpServerRequest& Request, const CacheRef& Ref, CachePolicy PolicyFromUrl);
+	void HandleRpcRequest(zen::HttpServerRequest& Request);
+
+	CbPackage HandleRpcPutCacheRecords(const CbPackage& BatchRequest);
+	CbPackage HandleRpcGetCacheRecords(CbObjectView BatchRequest);
+	CbPackage HandleRpcPutCacheValues(const CbPackage& BatchRequest);
+	CbPackage HandleRpcGetCacheValues(CbObjectView BatchRequest);
+	CbPackage HandleRpcGetCacheChunks(CbObjectView BatchRequest);
+	CbPackage HandleRpcRequest(const ZenContentType ContentType,
+							   IoBuffer&&			Body,
+							   uint32_t&			OutAcceptMagic,
+							   RpcAcceptOptions&	OutAcceptFlags);
+
 	void		 HandleCacheNamespaceRequest(zen::HttpServerRequest& Request, std::string_view Namespace);
 	void		 HandleCacheBucketRequest(zen::HttpServerRequest& Request, std::string_view Namespace, std::string_view Bucket);
 	virtual void HandleStatsRequest(zen::HttpServerRequest& Request) override;
@@ -117,9 +146,7 @@ private:
 	PutResult	 PutCacheRecord(PutRequestData& Request, const CbPackage* Package);
 
 	/** HandleRpcGetCacheChunks Helper: Parse the Body object into RecordValue Requests and Value Requests. */
-	bool ParseGetCacheChunksRequest(uint32_t&								   AcceptMagic,
-									uint16_t&								   AcceptFlags,
-									std::string&							   Namespace,
+	bool ParseGetCacheChunksRequest(std::string&							   Namespace,
 									std::vector<CacheKeyRequest>&			   RecordKeys,
 									std::vector<cache::detail::RecordBody>&	   Records,
 									std::vector<CacheChunkRequest>&			   RequestKeys,
@@ -143,11 +170,7 @@ private:
 								std::vector<CacheChunkRequest>&			  RequestKeys,
 								std::vector<cache::detail::ChunkRequest>& Requests);
 	/** HandleRpcGetCacheChunks Helper: Send response message containing all chunk results. */
-	void WriteGetCacheChunksResponse(uint32_t								   AcceptMagic,
-									 uint16_t								   AcceptFlags,
-									 std::string_view						   Namespace,
-									 std::vector<cache::detail::ChunkRequest>& Requests,
-									 zen::HttpServerRequest&				   HttpRequest);
+	CbPackage WriteGetCacheChunksResponse(std::string_view Namespace, std::vector<cache::detail::ChunkRequest>& Requests);
 
 	spdlog::logger&			 Log() { return m_Log; }
 	spdlog::logger&			 m_Log;
@@ -160,6 +183,10 @@ private:
 	metrics::OperationTiming m_HttpRequests;
 	metrics::OperationTiming m_UpstreamGetRequestTiming;
 	CacheStats				 m_CacheStats;
+
+	void ReplayRequestRecorder(cache::detail::IRequestReplayer& Replayer, uint32_t ThreadCount);
+
+	std::unique_ptr<cache::detail::IRequestRecorder> m_RequestRecorder;
 };
 
 /** Recognize both kBinary and kCompressedBinary as kCompressedBinary for structured cache value keys.