remote project store stats (#44)

* add remote oplog store statistics
* block chunking when uploading oplog to zenserver (mirroring)
* make sure we can move temporary dechunked file into cas store

1 files changed, 41 insertions, 0 deletions

diff --git a/src/zenserver/projectstore/fileremoteprojectstore.cpp b/src/zenserver/projectstore/fileremoteprojectstore.cpp
index 764bea355..5c7a5536a 100644
--- a/src/zenserver/projectstore/fileremoteprojectstore.cpp
+++ b/src/zenserver/projectstore/fileremoteprojectstore.cpp
@@ -39,12 +39,24 @@ public:
 		return {
 			.CreateBlocks	   = m_EnableBlocks,
 			.UseTempBlockFiles = m_UseTempBlocks,
+			.AllowChunking	   = true,
 			.ContainerName	   = m_Name,
 			.BaseContainerName = m_OptionalBaseName,
 			.Description =
 				fmt::format("[file] {}/{}{}{}"sv, m_OutputPath, m_Name, m_OptionalBaseName.empty() ? "" : " Base: ", m_OptionalBaseName)};
 	}
 
+	virtual Stats GetStats() const override
+	{
+		return {.m_SentBytes		 = m_SentBytes.load(),
+				.m_ReceivedBytes	 = m_ReceivedBytes.load(),
+				.m_RequestTimeNS	 = m_RequestTimeNS.load(),
+				.m_RequestCount		 = m_RequestCount.load(),
+				.m_PeakSentBytes	 = m_PeakSentBytes.load(),
+				.m_PeakReceivedBytes = m_PeakReceivedBytes.load(),
+				.m_PeakBytesPerSec	 = m_PeakBytesPerSec.load()};
+	}
+
 	virtual SaveResult SaveContainer(const IoBuffer& Payload) override
 	{
 		Stopwatch  Timer;
@@ -84,6 +96,7 @@ public:
 			Result.ErrorCode = gsl::narrow<int32_t>(HttpResponseCode::InternalServerError);
 			Result.Reason	 = fmt::format("Failed saving oplog container to '{}'. Reason: {}", ContainerPath, Ex.what());
 		}
+		AddStats(Payload.GetSize(), 0, Timer.GetElapsedTimeUs() * 1000);
 		Result.ElapsedSeconds = Timer.GetElapsedTimeMs() / 1000.0;
 		return Result;
 	}
@@ -114,6 +127,7 @@ public:
 				Result.Reason	 = fmt::format("Failed saving oplog attachment to '{}'. Reason: {}", ChunkPath, Ex.what());
 			}
 		}
+		AddStats(Payload.GetSize(), 0, Timer.GetElapsedTimeUs() * 1000);
 		Result.ElapsedSeconds = Timer.GetElapsedTimeMs() / 1000.0;
 		return Result;
 	}
@@ -161,6 +175,7 @@ public:
 				Result.Needs.insert(RawHash);
 			}
 		}
+		AddStats(0, 0, Timer.GetElapsedTimeUs() * 1000);
 		Result.ElapsedSeconds = Timer.GetElapsedTimeMs() / 1000.0;
 		return Result;
 	}
@@ -182,6 +197,7 @@ public:
 			ChunkFile.Open(ChunkPath, BasicFile::Mode::kRead);
 			Result.Bytes = ChunkFile.ReadAll();
 		}
+		AddStats(0, Result.Bytes.GetSize(), Timer.GetElapsedTimeUs() * 1000);
 		Result.ElapsedSeconds = Timer.GetElapsedTimeMs() / 1000.0;
 		return Result;
 	}
@@ -225,6 +241,7 @@ private:
 			ContainerFile.Open(SourcePath, BasicFile::Mode::kRead);
 			ContainerPayload = ContainerFile.ReadAll();
 		}
+		AddStats(0, ContainerPayload.GetSize(), Timer.GetElapsedTimeUs() * 1000);
 		Result.ContainerObject = LoadCompactBinaryObject(ContainerPayload);
 		if (!Result.ContainerObject)
 		{
@@ -256,11 +273,35 @@ private:
 		return ShardedPath.ToPath();
 	}
 
+	void AddStats(uint64_t UploadedBytes, uint64_t DownloadedBytes, uint64_t ElapsedNS)
+	{
+		m_SentBytes.fetch_add(UploadedBytes);
+		m_ReceivedBytes.fetch_add(DownloadedBytes);
+		m_RequestTimeNS.fetch_add(ElapsedNS);
+		SetAtomicMax(m_PeakSentBytes, UploadedBytes);
+		SetAtomicMax(m_PeakReceivedBytes, DownloadedBytes);
+		if (ElapsedNS > 0)
+		{
+			uint64_t BytesPerSec = (gsl::narrow<uint64_t>(UploadedBytes + DownloadedBytes) * 1000000) / ElapsedNS;
+			SetAtomicMax(m_PeakBytesPerSec, BytesPerSec);
+		}
+
+		m_RequestCount.fetch_add(1);
+	}
+
 	const std::string			m_Name;
 	const std::string			m_OptionalBaseName;
 	const std::filesystem::path m_OutputPath;
 	bool						m_EnableBlocks	= true;
 	bool						m_UseTempBlocks = false;
+
+	std::atomic_uint64_t m_SentBytes;
+	std::atomic_uint64_t m_ReceivedBytes;
+	std::atomic_uint64_t m_RequestTimeNS;
+	std::atomic_uint64_t m_RequestCount;
+	std::atomic_uint64_t m_PeakSentBytes;
+	std::atomic_uint64_t m_PeakReceivedBytes;
+	std::atomic_uint64_t m_PeakBytesPerSec;
 };
 
 std::shared_ptr<RemoteProjectStore>