WIP RpcAnalyzeCommand

3 files changed, 296 insertions, 7 deletions

diff --git a/src/zen/cmds/rpcreplay_cmd.cpp b/src/zen/cmds/rpcreplay_cmd.cpp
index 53f45358e..a4502614e 100644
--- a/src/zen/cmds/rpcreplay_cmd.cpp
+++ b/src/zen/cmds/rpcreplay_cmd.cpp
@@ -3,6 +3,7 @@
 #include "rpcreplay_cmd.h"
 
 #include <zencore/compactbinarybuilder.h>
+#include <zencore/compactbinaryvalidation.h>
 #include <zencore/filesystem.h>
 #include <zencore/fmtutils.h>
 #include <zencore/logging.h>
@@ -295,17 +296,14 @@ RpcReplayCommand::Run(const ZenCliOptions& GlobalOptions, int argc, char** argv)
 							switch (RequestInfo.ContentType)
 							{
 								case ZenContentType::kCbPackage:
+									if (ParsePackageMessageWithLegacyFallback(Payload, RequestPackage))
 									{
-										if (ParsePackageMessageWithLegacyFallback(Payload, RequestPackage))
-										{
-											Request = RequestPackage.GetObject();
-										}
+										Request = RequestPackage.GetObject();
 									}
 									break;
+
 								case ZenContentType::kCbObject:
-									{
-										Request = LoadCompactBinaryObject(Payload);
-									}
+									Request = LoadCompactBinaryObject(Payload);
 									break;
 							}
 
@@ -482,4 +480,277 @@ RpcReplayCommand::Run(const ZenCliOptions& GlobalOptions, int argc, char** argv)
 	return 0;
 }
 
+//////////////////////////////////////////////////////////////////////////
+
+RpcAnalyzeCommand::RpcAnalyzeCommand()
+{
+	m_Options.add_options()("h,help", "Print help");
+	m_Options.add_option("", "p", "path", "Recording file path", cxxopts::value(m_RecordingPath), "<path>");
+	m_Options.add_option("", "o", "output", "Report directory path", cxxopts::value(m_ReportPath), "<path>");
+
+	m_Options.parse_positional("path");
+	m_Options.parse_positional("output");
+}
+
+RpcAnalyzeCommand::~RpcAnalyzeCommand() = default;
+
+int
+RpcAnalyzeCommand::Run(const ZenCliOptions& GlobalOptions, int argc, char** argv)
+{
+	ZEN_UNUSED(GlobalOptions, argc, argv);
+
+	if (!ParseOptions(argc, argv))
+	{
+		return 0;
+	}
+
+	if (m_RecordingPath.empty())
+	{
+		throw zen::OptionParseException("Rpc replay command requires a path");
+	}
+
+	if (!std::filesystem::exists(m_RecordingPath) || !std::filesystem::is_directory(m_RecordingPath))
+	{
+		throw std::runtime_error(fmt::format("could not find recording at '{}'", m_RecordingPath));
+	}
+
+	std::unique_ptr<cache::IRpcRequestReplayer> Replayer   = cache::MakeDiskRequestReplayer(m_RecordingPath, true);
+	const uint64_t								EntryCount = Replayer->GetRequestCount();
+
+	Stopwatch Timer;
+
+	std::set<uint32_t> RequestMethods;
+	int				   MaxBatchSeen = 0;
+
+	for (uint64_t EntryIndex = 0; EntryIndex < EntryCount; ++EntryIndex)
+	{
+		IoBuffer							  Payload;
+		const zen::cache::RecordedRequestInfo RequestInfo = Replayer->GetRequest(EntryIndex, /* out */ Payload);
+
+		if (RequestInfo == zen::cache::RecordedRequestInfo::NullRequest)
+		{
+			continue;
+		}
+
+		CbPackage RequestPackage;
+		CbObject  Request;
+
+		switch (RequestInfo.ContentType)
+		{
+			case ZenContentType::kCbPackage:
+				if (ParsePackageMessageWithLegacyFallback(Payload, RequestPackage))
+				{
+					Request = RequestPackage.GetObject();
+				}
+				break;
+
+			case ZenContentType::kCbObject:
+				if (ValidateCompactBinary(Payload, CbValidateMode::All) == CbValidateError::None)
+				{
+					Request = LoadCompactBinaryObject(Payload);
+				}
+				break;
+		}
+
+		if (!Request)
+		{
+			continue;
+		}
+
+		static constinit uint32_t kGetCacheValues  = HashStringAsLowerDjb2("GetCacheValues"sv);
+		static constinit uint32_t kGetCacheRecords = HashStringAsLowerDjb2("GetCacheRecords"sv);
+		static constinit uint32_t kPutCacheValues  = HashStringAsLowerDjb2("PutCacheValues"sv);
+		static constinit uint32_t kPutCacheRecords = HashStringAsLowerDjb2("PutCacheRecords"sv);
+		static constinit uint32_t kGetCacheChunks  = HashStringAsLowerDjb2("GetCacheChunks"sv);
+
+		std::string_view MethodString = Request["Method"sv].AsString();
+		const uint32_t	 MethodHash	  = HashStringAsLowerDjb2(MethodString);
+
+		if (auto It = RequestMethods.find(MethodHash); It == RequestMethods.end())
+		{
+			RequestMethods.insert(MethodHash);
+
+			ExtendableStringBuilder<1024> ObjStr;
+			CompactBinaryToJson(Request, ObjStr);
+			ZEN_CONSOLE("{}", ObjStr);
+		}
+
+		CbObjectView	 Params	   = Request["Params"sv].AsObjectView();
+		std::string_view Namespace = Request["Namespace"sv].AsString();
+
+		int BatchSize = 0;
+
+		for (auto RequestEntry : Params["Requests"sv])
+		{
+			auto Key = RequestEntry.AsObjectView()["Key"sv].AsObjectView();
+
+			std::string_view Bucket = Key["Bucket"sv].AsString();
+			const IoHash	 Hash	= Key["Hash"sv].AsHash();
+
+			if (Bucket.empty())
+			{
+				continue;
+			}
+
+			++BatchSize;
+
+			if (MethodHash == kGetCacheValues)
+			{
+				/*
+
+				{
+					"Method": "GetCacheValues",
+					"Accept": 2859969228,
+					"Params": {
+							"DefaultPolicy": "Default,PartialRecord",
+							"Namespace": "ue.ddc",
+							"Requests": [
+									{
+											"Key": {
+													"Bucket": "AnimationSequence",
+													"Hash": "272757e7b8d4638acb6c725d8538b87dfebb3cf9"
+											}
+									}
+							]
+					}
+				}
+
+				*/
+
+				// Record GET
+			}
+			else if (MethodHash == kGetCacheRecords)
+			{
+				/*
+
+				{
+					"Method": "GetCacheRecords",
+					"Accept": 2859969228,
+					"Params": {
+							"DefaultPolicy": "Default,PartialRecord",
+							"Namespace": "fortnite.sddc",
+							"Requests": [
+									{
+											"Key": {
+													"Bucket": "BulkDataList",
+													"Hash": "6a9dd7da26cca613821bf5aa434ce64bcef6712e"
+											}
+									}
+							]
+					}
+				}
+
+				*/
+
+				// Record GET
+			}
+			else if (MethodHash == kPutCacheValues)
+			{
+				/*
+
+				{
+					"Method": "PutCacheValues",
+					"Accept": 2859969228,
+					"Params": {
+							"DefaultPolicy": "Default",
+							"Namespace": "ue.ddc",
+							"Requests": [
+									{
+											"Key": {
+													"Bucket": "MaterialShaderMap",
+													"Hash": "bbdb638a34449e8cc3c89efc63431fc7193124ec"
+											},
+											"RawHash": "2a9d7bab0c776ff299b927d70d3e400851cdddd6"
+									}
+							]
+					}
+				}
+
+				*/
+			}
+			else if (MethodHash == kPutCacheRecords)
+			{
+				/*
+
+				{
+					"Method": "PutCacheRecords",
+					"Accept": 2859969228,
+					"Params": {
+							"DefaultPolicy": "Default",
+							"Namespace": "ue.ddc",
+							"Requests": [
+									{
+											"Record": {
+													"Key": {
+															"Bucket": "MaterialTranslation",
+															"Hash": "4eaea52861838d09645b228e97bd246cbdb76bac"
+													},
+													"Values": [
+															{
+																	"Id": "0d586d5a600af5163b955814",
+																	"RawHash": "bfef9d6bfabecce61880b1d2f06e6e25838d5b8e",
+																	"RawSize": 3609
+															},
+															{
+																	"Id": "52d979e80b2ccf0dbc303f10",
+																	"RawHash": "17cbea1698d9904fd1272261f0a08c221e5e6ba7",
+																	"RawSize": 12665
+															},
+															{
+																	"Id": "a2b11ed5fab981f31e6504df",
+																	"RawHash": "0542d53449a8f0e9b8f0c7ac8506bc2f6e060d75",
+																	"RawSize": 326
+															}
+													]
+											}
+									}
+							]
+					}
+				}
+
+				*/
+			}
+			else if (MethodHash == kGetCacheChunks)
+			{
+				/*
+
+				{
+					"Method": "GetCacheChunks",
+					"Accept": 2859969228,
+					"Params": {
+							"DefaultPolicy": "Default",
+							"Namespace": "fortnite.sddc",
+							"ChunkRequests": [
+									{
+											"Key": {
+													"Bucket": "StaticMesh",
+													"Hash": "267ab8179533a57eaffec8de3bf032df17c23fc4"
+											},
+											"ValueId": "7fdeebb90ed07ba4a490149b",
+											"ChunkId": "54897be0cf047e5b14476cdea8390e80deab0a29"
+									}
+							]
+					}
+				}
+
+				*/
+			}
+			// ZEN_CONSOLE("key: {}/{}", Bucket, Hash);
+		}
+
+		if (BatchSize > MaxBatchSeen)
+		{
+			ZEN_CONSOLE("wow {}", BatchSize);
+
+			ExtendableStringBuilder<1024> ObjStr;
+			CompactBinaryToJson(Request, ObjStr);
+			ZEN_CONSOLE("{}", ObjStr);
+
+			MaxBatchSeen = BatchSize;
+		}
+	}
+
+	return 0;
+}
+
 }  // namespace zen
diff --git a/src/zen/cmds/rpcreplay_cmd.h b/src/zen/cmds/rpcreplay_cmd.h
index 42cdd4ac1..5f2c8cf0b 100644
--- a/src/zen/cmds/rpcreplay_cmd.h
+++ b/src/zen/cmds/rpcreplay_cmd.h
@@ -63,4 +63,20 @@ private:
 	bool			 m_DryRun					  = false;
 };
 
+class RpcAnalyzeCommand : public CacheStoreCommand
+{
+public:
+	RpcAnalyzeCommand();
+	~RpcAnalyzeCommand();
+
+	virtual int				  Run(const ZenCliOptions& GlobalOptions, int argc, char** argv) override;
+	virtual cxxopts::Options& Options() override { return m_Options; }
+
+private:
+	cxxopts::Options m_Options{"rpc-record-analyze", "Analyses an RPC recording session"};
+	std::string		 m_HostName;
+	std::string		 m_RecordingPath;
+	std::string		 m_ReportPath;
+};
+
 }  // namespace zen
diff --git a/src/zen/zen.cpp b/src/zen/zen.cpp
index 10d2f5593..cb9e04e9a 100644
--- a/src/zen/zen.cpp
+++ b/src/zen/zen.cpp
@@ -283,6 +283,7 @@ main(int argc, char** argv)
 	ProjectInfoCommand		 ProjectInfoCmd;
 	ProjectStatsCommand		 ProjectStatsCmd;
 	PsCommand				 PsCmd;
+	RpcAnalyzeCommand		 RpcAnalyzeCmd;
 	RpcReplayCommand		 RpcReplayCmd;
 	RpcStartRecordingCommand RpcStartRecordingCmd;
 	RpcStopRecordingCommand	 RpcStopRecordingCmd;
@@ -336,6 +337,7 @@ main(int argc, char** argv)
 		{"project-info",		&ProjectInfoCmd,		"Info on project or project oplog"},
 		{"project-stats",		&ProjectStatsCmd,		"Stats on project store"},
 		{"ps",					&PsCmd,					"Enumerate running zen server instances"},
+		{"rpc-record-analyze",	&RpcAnalyzeCmd,			"Analyzes a previously recorded session of rpc requests"},
 		{"rpc-record-replay",	&RpcReplayCmd,			"Replays a previously recorded session of rpc requests"},
 		{"rpc-record-start",	&RpcStartRecordingCmd,	"Starts recording of cache rpc requests on a host"},
 		{"rpc-record-stop",		&RpcStopRecordingCmd,	"Stops recording of cache rpc requests on a host"},