Skeleton of rpc recording analysis command

Also changes the RecordedRequestsReader segment recovery implementation so that it can traverse a folder hierarchy to find available recording segments

1 files changed, 71 insertions, 26 deletions

diff --git a/src/zenutil/cache/rpcrecording.cpp b/src/zenutil/cache/rpcrecording.cpp
index 46e80f6b7..2eb8b3258 100644
--- a/src/zenutil/cache/rpcrecording.cpp
+++ b/src/zenutil/cache/rpcrecording.cpp
@@ -4,6 +4,7 @@
 
 #include <zencore/basicfile.h>
 #include <zencore/compactbinarybuilder.h>
+#include <zencore/compactbinaryvalidation.h>
 #include <zencore/filesystem.h>
 #include <zencore/fmtutils.h>
 #include <zencore/logging.h>
@@ -931,29 +932,60 @@ RecordedRequestsReader::BeginRead(const std::filesystem::path& BasePath, bool In
 	uint64_t TotalRequestCount = 0;
 	uint64_t MaxSegmentIndex   = 0;
 
+	auto IterateSegments = [&](const std::filesystem::path& RootPath) {
+		std::vector<std::filesystem::path> Paths;
+
+		std::error_code								  Ec;
+		std::filesystem::recursive_directory_iterator DirIt(RootPath, Ec);
+
+		if (!Ec)
+		{
+			for (auto& Dir : DirIt)
+			{
+				if (Dir.is_regular_file() && Dir.path().filename().string().ends_with("rpc_segment_info.zcb"))
+				{
+					Paths.emplace_back(Dir.path());
+				}
+			}
+		}
+
+		std::sort(begin(Paths), end(Paths));
+
+		return Paths;
+	};
+
 	try
 	{
-		for (int SegmentIndex = 0;; ++SegmentIndex)
+		auto SegmentPaths = IterateSegments(BasePath);
+
+		for (const std::filesystem::path ZcbPath : SegmentPaths)
 		{
-			const std::filesystem::path ZcbPath = BasePath / MakeSegmentPath(SegmentIndex) / "rpc_segment_info.zcb";
-			FileContents				Fc		= ReadFile(ZcbPath);
+			FileContents Fc = ReadFile(ZcbPath);
 
 			if (Fc.ErrorCode)
-				break;
-
-			if (IoBuffer SegmentInfoBuffer = Fc.Flatten())
 			{
-				CbObject Segment = LoadCompactBinaryObject(SegmentInfoBuffer);
-
-				const SegmentInfo& Info = m_KnownSegments.emplace_back(SegmentInfo{.SegmentIndex = Segment["segment_index"sv].AsUInt64(),
-																				   .BaseRequestIndex = 0,
-																				   .RequestCount	 = Segment["entry_count"sv].AsUInt64(),
-																				   .RequestBytes	 = 0,
-																				   .StartTime		 = Segment["time_start"sv].AsDateTime(),
-																				   .EndTime			 = Segment["time_end"sv].AsDateTime()});
-
-				TotalRequestCount += Info.RequestCount;
-				MaxSegmentIndex = Max(MaxSegmentIndex, Info.SegmentIndex);
+				ZEN_WARN("Error opening '{}': {}", ZcbPath, Fc.ErrorCode.message());
+			}
+			else
+			{
+				if (IoBuffer SegmentInfoBuffer = Fc.Flatten())
+				{
+					if (ValidateCompactBinaryRange(SegmentInfoBuffer.GetView(), CbValidateMode::Default) == CbValidateError::None)
+					{
+						CbObject Segment = LoadCompactBinaryObject(SegmentInfoBuffer);
+
+						const SegmentInfo& Info =
+							m_KnownSegments.emplace_back(SegmentInfo{.SegmentIndex	   = Segment["segment_index"sv].AsUInt64(),
+																	 .BaseRequestIndex = 0,
+																	 .RequestCount	   = Segment["entry_count"sv].AsUInt64(),
+																	 .RequestBytes	   = 0,
+																	 .StartTime		   = Segment["time_start"sv].AsDateTime(),
+																	 .EndTime		   = Segment["time_end"sv].AsDateTime()});
+
+						TotalRequestCount += Info.RequestCount;
+						MaxSegmentIndex = Max(MaxSegmentIndex, Info.SegmentIndex);
+					}
+				}
 			}
 		}
 	}
@@ -1014,6 +1046,8 @@ RecordedRequestsReader::ReadRequest(uint64_t RequestIndex, IoBuffer& OutBuffer)
 		return *(SegmentReaderPtr.get());
 	};
 
+	// This is pretty slow for large replays, should have an index lookup for this instead
+
 	for (auto& Segment : m_KnownSegments)
 	{
 		if (Segment.RequestCount > RequestIndex)
@@ -1055,24 +1089,35 @@ public:
 	}
 	virtual ~DiskRequestReplayer() { m_RequestReader.EndRead(); }
 
-	static bool IsCompatible(const std::filesystem::path& BasePath)
+	static bool HasSegments(const std::filesystem::path& RootPath)
 	{
-		if (IsFile(BasePath / "rpc_recording_info.zcb"))
+		std::error_code								  Ec;
+		std::filesystem::recursive_directory_iterator DirIt(RootPath, Ec);
+
+		if (Ec)
 		{
-			return true;
+			return false;
 		}
 
-		const std::filesystem::path SegmentZero = BasePath / MakeSegmentPath(0);
-
-		if (IsFile(SegmentZero / "rpc_segment_info.zcb") && IsFile(SegmentZero / "index.bin"))
+		for (auto& Dir : DirIt)
 		{
-			// top-level metadata is missing, possibly because of premature exit
-			// on the recording side
+			if (Dir.is_regular_file() && Dir.path().filename().string().ends_with("rpc_segment_info.zcb"))
+			{
+				return true;
+			}
+		}
+
+		return false;
+	};
 
+	static bool IsCompatible(const std::filesystem::path& BasePath)
+	{
+		if (IsFile(BasePath / "rpc_recording_info.zcb"))
+		{
 			return true;
 		}
 
-		return false;
+		return HasSegments(BasePath);
 	}
 
 	virtual uint64_t GetRequestCount() const override { return m_RequestCount; }