show download source data (#689)

* show source stats for jupiter/cache

12 files changed, 223 insertions, 11 deletions

diff --git a/src/zen/cmds/builds_cmd.cpp b/src/zen/cmds/builds_cmd.cpp
index b1052654d..3b08a9290 100644
--- a/src/zen/cmds/builds_cmd.cpp
+++ b/src/zen/cmds/builds_cmd.cpp
@@ -1667,14 +1667,15 @@ namespace {
 		bool					 AppendNewContent			= false;
 	};
 
-	void DownloadFolder(OperationLogOutput&			 Output,
-						ThreadWorkers&				 Workers,
-						StorageInstance&			 Storage,
-						const Oid&					 BuildId,
-						const std::vector<Oid>&		 BuildPartIds,
-						std::span<const std::string> BuildPartNames,
-						const std::filesystem::path& Path,
-						const DownloadOptions&		 Options)
+	void DownloadFolder(OperationLogOutput&					 Output,
+						ThreadWorkers&						 Workers,
+						StorageInstance&					 Storage,
+						const BuildStorageCache::Statistics& StorageCacheStats,
+						const Oid&							 BuildId,
+						const std::vector<Oid>&				 BuildPartIds,
+						std::span<const std::string>		 BuildPartNames,
+						const std::filesystem::path&		 Path,
+						const DownloadOptions&				 Options)
 	{
 		ZEN_TRACE_CPU("DownloadFolder");
 
@@ -1965,9 +1966,45 @@ namespace {
 					{
 						CloneInfo = fmt::format(" ({} cloned)", NiceBytes(Updater.m_DiskStats.CloneByteCount.load()));
 					}
+
+					std::string DownloadDetails;
+					{
+						ExtendableStringBuilder<128>		 SB;
+						BuildStorageBase::ExtendedStatistics ExtendedDownloadStats;
+						if (Storage.BuildStorage->GetExtendedStatistics(ExtendedDownloadStats))
+						{
+							if (!ExtendedDownloadStats.ReceivedBytesPerSource.empty())
+							{
+								for (auto& It : ExtendedDownloadStats.ReceivedBytesPerSource)
+								{
+									if (SB.Size() > 0)
+									{
+										SB.Append(", "sv);
+									}
+									SB.Append(It.first);
+									SB.Append(": "sv);
+									SB.Append(NiceBytes(It.second));
+								}
+							}
+						}
+						if (Storage.BuildCacheStorage)
+						{
+							if (SB.Size() > 0)
+							{
+								SB.Append(", "sv);
+							}
+							SB.Append("Cache: ");
+							SB.Append(NiceBytes(StorageCacheStats.TotalBytesRead.load()));
+						}
+						if (SB.Size() > 0)
+						{
+							DownloadDetails = fmt::format(" ({})", SB.ToView());
+						}
+					}
+
 					ZEN_CONSOLE(
 						"Downloaded build {}, parts:{} in {}\n"
-						"    Download:   {} ({}) {}bits/s\n"
+						"    Download:   {} ({}) {}bits/s{}\n"
 						"    Write:      {} ({}) {}B/s{}\n"
 						"    Clean:      {}\n"
 						"    Finalize:   {}\n"
@@ -1979,6 +2016,7 @@ namespace {
 						DownloadCount,
 						NiceBytes(DownloadByteCount),
 						NiceNum(GetBytesPerSecond(Updater.m_WriteChunkStats.DownloadTimeUs, DownloadByteCount * 8)),
+						DownloadDetails,
 
 						Updater.m_DiskStats.WriteCount.load(),
 						NiceBytes(Updater.m_WrittenChunkByteCount.load()),
@@ -3748,6 +3786,7 @@ BuildsCommand::Run(const ZenCliOptions& GlobalOptions, int argc, char** argv)
 		DownloadFolder(*Output,
 					   Workers,
 					   Storage,
+					   StorageCacheStats,
 					   BuildId,
 					   BuildPartIds,
 					   BuildPartNames,
@@ -4072,6 +4111,7 @@ BuildsCommand::Run(const ZenCliOptions& GlobalOptions, int argc, char** argv)
 			DownloadFolder(*Output,
 						   Workers,
 						   Storage,
+						   StorageCacheStats,
 						   BuildId,
 						   {},
 						   {},
@@ -4272,6 +4312,7 @@ BuildsCommand::Run(const ZenCliOptions& GlobalOptions, int argc, char** argv)
 			DownloadFolder(*Output,
 						   Workers,
 						   Storage,
+						   StorageCacheStats,
 						   BuildId,
 						   {BuildPartId},
 						   {},
@@ -4302,6 +4343,7 @@ BuildsCommand::Run(const ZenCliOptions& GlobalOptions, int argc, char** argv)
 			DownloadFolder(*Output,
 						   Workers,
 						   Storage,
+						   StorageCacheStats,
 						   BuildId,
 						   {BuildPartId},
 						   {},
@@ -4328,6 +4370,7 @@ BuildsCommand::Run(const ZenCliOptions& GlobalOptions, int argc, char** argv)
 			DownloadFolder(*Output,
 						   Workers,
 						   Storage,
+						   StorageCacheStats,
 						   BuildId,
 						   {BuildPartId},
 						   {},
@@ -4355,6 +4398,7 @@ BuildsCommand::Run(const ZenCliOptions& GlobalOptions, int argc, char** argv)
 		DownloadFolder(*Output,
 					   Workers,
 					   Storage,
+					   StorageCacheStats,
 					   BuildId,
 					   {BuildPartId},
 					   {},
@@ -4379,6 +4423,7 @@ BuildsCommand::Run(const ZenCliOptions& GlobalOptions, int argc, char** argv)
 		DownloadFolder(*Output,
 					   Workers,
 					   Storage,
+					   StorageCacheStats,
 					   BuildId,
 					   {BuildPartId},
 					   {},
@@ -4495,6 +4540,7 @@ BuildsCommand::Run(const ZenCliOptions& GlobalOptions, int argc, char** argv)
 		DownloadFolder(*Output,
 					   Workers,
 					   Storage,
+					   StorageCacheStats,
 					   BuildId,
 					   {BuildPartId},
 					   {},
@@ -4554,6 +4600,7 @@ BuildsCommand::Run(const ZenCliOptions& GlobalOptions, int argc, char** argv)
 		DownloadFolder(*Output,
 					   Workers,
 					   Storage,
+					   StorageCacheStats,
 					   BuildId,
 					   {BuildPartId},
 					   {},
@@ -4578,6 +4625,7 @@ BuildsCommand::Run(const ZenCliOptions& GlobalOptions, int argc, char** argv)
 		DownloadFolder(*Output,
 					   Workers,
 					   Storage,
+					   StorageCacheStats,
 					   BuildId2,
 					   {BuildPartId2},
 					   {},
@@ -4601,6 +4649,7 @@ BuildsCommand::Run(const ZenCliOptions& GlobalOptions, int argc, char** argv)
 		DownloadFolder(*Output,
 					   Workers,
 					   Storage,
+					   StorageCacheStats,
 					   BuildId2,
 					   {BuildPartId2},
 					   {},
@@ -4624,6 +4673,7 @@ BuildsCommand::Run(const ZenCliOptions& GlobalOptions, int argc, char** argv)
 		DownloadFolder(*Output,
 					   Workers,
 					   Storage,
+					   StorageCacheStats,
 					   BuildId,
 					   {BuildPartId},
 					   {},
@@ -4647,6 +4697,7 @@ BuildsCommand::Run(const ZenCliOptions& GlobalOptions, int argc, char** argv)
 		DownloadFolder(*Output,
 					   Workers,
 					   Storage,
+					   StorageCacheStats,
 					   BuildId,
 					   {BuildPartId},
 					   {},
diff --git a/src/zenremotestore/builds/filebuildstorage.cpp b/src/zenremotestore/builds/filebuildstorage.cpp
index 153deaa9f..1474fd819 100644
--- a/src/zenremotestore/builds/filebuildstorage.cpp
+++ b/src/zenremotestore/builds/filebuildstorage.cpp
@@ -639,6 +639,12 @@ public:
 		WriteAsJson(BuildPartStatsDataPath, Payload);
 	}
 
+	virtual bool GetExtendedStatistics(ExtendedStatistics& OutStats) override
+	{
+		ZEN_UNUSED(OutStats);
+		return false;
+	}
+
 protected:
 	std::filesystem::path GetBuildsFolder() const { return m_StoragePath / "builds"; }
 	std::filesystem::path GetBlobsFolder() const { return m_StoragePath / "blobs"; }
diff --git a/src/zenremotestore/builds/jupiterbuildstorage.cpp b/src/zenremotestore/builds/jupiterbuildstorage.cpp
index fe8067905..962ffaaed 100644
--- a/src/zenremotestore/builds/jupiterbuildstorage.cpp
+++ b/src/zenremotestore/builds/jupiterbuildstorage.cpp
@@ -423,6 +423,16 @@ public:
 		}
 	}
 
+	virtual bool GetExtendedStatistics(ExtendedStatistics& OutStats) override
+	{
+		OutStats.ReceivedBytesPerSource.reserve(m_ReceivedBytesPerSource.size());
+		for (auto& It : m_ReceivedBytesPerSource)
+		{
+			OutStats.ReceivedBytesPerSource.insert_or_assign(It.first, m_SourceBytes[It.second]);
+		}
+		return true;
+	}
+
 private:
 	static CbObject PayloadToCbObject(std::string_view ErrorContext, const IoBuffer& Payload)
 	{
@@ -490,6 +500,19 @@ private:
 			uint64_t BytesPerSec = uint64_t((Result.SentBytes + Result.ReceivedBytes) / Result.ElapsedSeconds);
 			SetAtomicMax(m_Stats.PeakBytesPerSec, BytesPerSec);
 		}
+		if (!Result.Source.empty())
+		{
+			if (tsl::robin_map<std::string, uint32_t>::const_iterator It = m_ReceivedBytesPerSource.find(Result.Source);
+				It != m_ReceivedBytesPerSource.end())
+			{
+				m_SourceBytes[It->second] += Result.ReceivedBytes;
+			}
+			else
+			{
+				m_ReceivedBytesPerSource.insert_or_assign(Result.Source, m_SourceBytes.size());
+				m_SourceBytes.push_back(Result.ReceivedBytes);
+			}
+		}
 	}
 
 	JupiterSession				m_Session;
@@ -497,6 +520,9 @@ private:
 	const std::string			m_Namespace;
 	const std::string			m_Bucket;
 	const std::filesystem::path m_TempFolderPath;
+
+	tsl::robin_map<std::string, uint32_t> m_ReceivedBytesPerSource;
+	std::vector<uint64_t>				  m_SourceBytes;
 };
 
 std::unique_ptr<BuildStorageBase>
diff --git a/src/zenremotestore/include/zenremotestore/builds/buildstorage.h b/src/zenremotestore/include/zenremotestore/builds/buildstorage.h
index d4df979e2..4b7e54d85 100644
--- a/src/zenremotestore/include/zenremotestore/builds/buildstorage.h
+++ b/src/zenremotestore/include/zenremotestore/builds/buildstorage.h
@@ -26,6 +26,11 @@ public:
 		std::atomic<uint64_t> PeakBytesPerSec	   = 0;
 	};
 
+	struct ExtendedStatistics
+	{
+		tsl::robin_map<std::string, uint64_t> ReceivedBytesPerSource;
+	};
+
 	virtual ~BuildStorageBase() {}
 
 	virtual CbObject ListNamespaces(bool bRecursive = false)				= 0;
@@ -63,6 +68,8 @@ public:
 	virtual CbObject		   GetBlockMetadatas(const Oid& BuildId, std::span<const IoHash> BlockHashes)				  = 0;
 
 	virtual void PutBuildPartStats(const Oid& BuildId, const Oid& BuildPartId, const tsl::robin_map<std::string, double>& FloatStats) = 0;
+
+	[[nodiscard]] virtual bool GetExtendedStatistics(ExtendedStatistics& OutStats) = 0;
 };
 
 }  // namespace zen
diff --git a/src/zenremotestore/include/zenremotestore/jupiter/jupitersession.h b/src/zenremotestore/include/zenremotestore/jupiter/jupitersession.h
index b79790f25..1945f6cff 100644
--- a/src/zenremotestore/include/zenremotestore/jupiter/jupitersession.h
+++ b/src/zenremotestore/include/zenremotestore/jupiter/jupitersession.h
@@ -20,6 +20,7 @@ struct JupiterResult
 	double		ElapsedSeconds{};
 	int32_t		ErrorCode{};
 	std::string Reason;
+	std::string Source;
 	bool		Success = false;
 };
 
diff --git a/src/zenremotestore/include/zenremotestore/projectstore/remoteprojectstore.h b/src/zenremotestore/include/zenremotestore/projectstore/remoteprojectstore.h
index de858f818..182b64609 100644
--- a/src/zenremotestore/include/zenremotestore/projectstore/remoteprojectstore.h
+++ b/src/zenremotestore/include/zenremotestore/projectstore/remoteprojectstore.h
@@ -9,6 +9,10 @@
 
 #include <unordered_set>
 
+ZEN_THIRD_PARTY_INCLUDES_START
+#include <tsl/robin_map.h>
+ZEN_THIRD_PARTY_INCLUDES_END
+
 namespace zen {
 
 class CidStore;
@@ -89,11 +93,17 @@ public:
 		std::uint64_t m_PeakBytesPerSec;
 	};
 
+	struct ExtendedStats
+	{
+		tsl::robin_map<std::string, uint64_t> m_ReceivedBytesPerSource;
+	};
+
 	RemoteProjectStore();
 	virtual ~RemoteProjectStore();
 
-	virtual RemoteStoreInfo GetInfo() const	 = 0;
-	virtual Stats			GetStats() const = 0;
+	virtual RemoteStoreInfo GetInfo() const									= 0;
+	virtual Stats			GetStats() const								= 0;
+	virtual bool			GetExtendedStats(ExtendedStats& OutStats) const = 0;
 
 	virtual CreateContainerResult CreateContainer()																					   = 0;
 	virtual SaveResult			  SaveContainer(const IoBuffer& Payload)															   = 0;
diff --git a/src/zenremotestore/jupiter/jupitersession.cpp b/src/zenremotestore/jupiter/jupitersession.cpp
index a0c961e37..3dac87d96 100644
--- a/src/zenremotestore/jupiter/jupitersession.cpp
+++ b/src/zenremotestore/jupiter/jupitersession.cpp
@@ -19,6 +19,48 @@ using namespace std::literals;
 namespace zen {
 
 namespace detail {
+	std::string_view FindTimingEntry(const std::string_view Pattern, const std::string_view TimingString)
+	{
+		ZEN_ASSERT_SLOW(Pattern.back() == '*');
+		std::string_view FindString = Pattern.substr(0, Pattern.length() - 1);
+		if (auto FindPos = TimingString.rfind(FindString); FindPos != std::string::npos)
+		{
+			std::string_view::size_type MatchStart = FindPos + FindString.length();
+			std::string_view::size_type EndPos	   = TimingString.find(';', MatchStart);
+			if (EndPos == std::string::npos)
+			{
+				EndPos = TimingString.length();
+			}
+			return TimingString.substr(MatchStart, EndPos - MatchStart);
+		}
+		else
+		{
+			return {};
+		}
+	};
+
+	std::string_view GetSource(const HttpClient::KeyValueMap& Headers)
+	{
+		if (auto HeaderIt = Headers.Entries.find("Server-Timing"); HeaderIt != Headers.Entries.end())
+		{
+			const std::string& ServerTimings = HeaderIt->second;
+			if (std::string_view ReplicatedSource = FindTimingEntry("blob.replicate.*"sv, ServerTimings); !ReplicatedSource.empty())
+			{
+				return ReplicatedSource;
+			}
+			else if (std::string_view::size_type ReplicatePos = ServerTimings.find("blob.replicate"sv);
+					 ReplicatePos != std::string_view::npos)
+			{
+				return "replicated"sv;
+			}
+			else if (std::string_view GetSource = FindTimingEntry("blob.get.*"sv, ServerTimings); !GetSource.empty())
+			{
+				return GetSource == "FileSystemStore"sv ? "Cloud Storage"sv : GetSource;
+			}
+		}
+		return {};
+	}
+
 	JupiterResult ConvertResponse(const HttpClient::Response& Response, const std::string_view ErrorPrefix = ""sv)
 	{
 		if (Response.Error)
@@ -39,11 +81,15 @@ namespace detail {
 					.Reason			= Response.ErrorMessage(ErrorPrefix),
 					.Success		= false};
 		}
+
+		std::string_view Source = GetSource(Response.Header);
+
 		return {.Response		= Response.ResponsePayload,
 				.SentBytes		= gsl::narrow<uint64_t>(Response.UploadedBytes),
 				.ReceivedBytes	= gsl::narrow<uint64_t>(Response.DownloadedBytes),
 				.ElapsedSeconds = Response.ElapsedSeconds,
 				.ErrorCode		= 0,
+				.Source			= std::string(Source),
 				.Success		= true};
 	}
 }  // namespace detail
diff --git a/src/zenremotestore/projectstore/buildsremoteprojectstore.cpp b/src/zenremotestore/projectstore/buildsremoteprojectstore.cpp
index 7a03ce50a..706f11e8c 100644
--- a/src/zenremotestore/projectstore/buildsremoteprojectstore.cpp
+++ b/src/zenremotestore/projectstore/buildsremoteprojectstore.cpp
@@ -97,6 +97,26 @@ public:
 		};
 	}
 
+	virtual bool GetExtendedStats(ExtendedStats& OutStats) const override
+	{
+		bool								 Result = false;
+		BuildStorageBase::ExtendedStatistics StorageStats;
+		if (m_BuildStorage->GetExtendedStatistics(StorageStats))
+		{
+			for (auto It : StorageStats.ReceivedBytesPerSource)
+			{
+				OutStats.m_ReceivedBytesPerSource.insert_or_assign(It.first, It.second);
+			}
+			Result = true;
+		}
+		if (m_BuildCacheStorage)
+		{
+			OutStats.m_ReceivedBytesPerSource.insert_or_assign("Cache", m_StorageCacheStats.TotalBytesRead);
+			Result = true;
+		}
+		return Result;
+	}
+
 	virtual CreateContainerResult CreateContainer() override
 	{
 		ZEN_ASSERT(m_OplogBuildPartId == Oid::Zero);
diff --git a/src/zenremotestore/projectstore/fileremoteprojectstore.cpp b/src/zenremotestore/projectstore/fileremoteprojectstore.cpp
index f49f0d72e..50be5d2d9 100644
--- a/src/zenremotestore/projectstore/fileremoteprojectstore.cpp
+++ b/src/zenremotestore/projectstore/fileremoteprojectstore.cpp
@@ -60,6 +60,12 @@ public:
 				.m_PeakBytesPerSec	 = m_PeakBytesPerSec.load()};
 	}
 
+	virtual bool GetExtendedStats(ExtendedStats& OutStats) const override
+	{
+		ZEN_UNUSED(OutStats);
+		return false;
+	}
+
 	virtual CreateContainerResult CreateContainer() override
 	{
 		// Nothing to do here
diff --git a/src/zenremotestore/projectstore/jupiterremoteprojectstore.cpp b/src/zenremotestore/projectstore/jupiterremoteprojectstore.cpp
index 1dd0a235c..e26a5e88d 100644
--- a/src/zenremotestore/projectstore/jupiterremoteprojectstore.cpp
+++ b/src/zenremotestore/projectstore/jupiterremoteprojectstore.cpp
@@ -65,6 +65,12 @@ public:
 				.m_PeakBytesPerSec	 = m_PeakBytesPerSec.load()};
 	}
 
+	virtual bool GetExtendedStats(ExtendedStats& OutStats) const override
+	{
+		ZEN_UNUSED(OutStats);
+		return false;
+	}
+
 	virtual CreateContainerResult CreateContainer() override
 	{
 		// Nothing to do here
diff --git a/src/zenremotestore/projectstore/remoteprojectstore.cpp b/src/zenremotestore/projectstore/remoteprojectstore.cpp
index 99c36b460..5652d5271 100644
--- a/src/zenremotestore/projectstore/remoteprojectstore.cpp
+++ b/src/zenremotestore/projectstore/remoteprojectstore.cpp
@@ -3282,6 +3282,33 @@ LoadOplog(CidStore&			   ChunkStore,
 
 	remotestore_impl::LogRemoteStoreStatsDetails(RemoteStore.GetStats());
 
+	{
+		std::string						  DownloadDetails;
+		RemoteProjectStore::ExtendedStats ExtendedStats;
+		if (RemoteStore.GetExtendedStats(ExtendedStats))
+		{
+			if (!ExtendedStats.m_ReceivedBytesPerSource.empty())
+			{
+				uint64_t					 Total = 0;
+				ExtendableStringBuilder<128> SB;
+
+				for (auto& It : ExtendedStats.m_ReceivedBytesPerSource)
+				{
+					if (SB.Size() > 0)
+					{
+						SB.Append(", "sv);
+					}
+					SB.Append(It.first);
+					SB.Append(": "sv);
+					SB.Append(NiceBytes(It.second));
+					Total += It.second;
+				}
+
+				remotestore_impl::ReportMessage(OptionalContext, fmt::format("Downloaded {} ({})", NiceBytes(Total), SB.ToView()));
+			}
+		}
+	}
+
 	remotestore_impl::ReportMessage(
 		OptionalContext,
 		fmt::format("Loaded oplog '{}' {} in {} ({}), Blocks: {} ({}), Attachments: {} ({}), Stored: {} ({}), Missing: {} {}",
diff --git a/src/zenremotestore/projectstore/zenremoteprojectstore.cpp b/src/zenremotestore/projectstore/zenremoteprojectstore.cpp
index e520466e0..ab82edbef 100644
--- a/src/zenremotestore/projectstore/zenremoteprojectstore.cpp
+++ b/src/zenremotestore/projectstore/zenremoteprojectstore.cpp
@@ -52,6 +52,12 @@ public:
 				.m_PeakBytesPerSec	 = m_PeakBytesPerSec.load()};
 	}
 
+	virtual bool GetExtendedStats(ExtendedStats& OutStats) const override
+	{
+		ZEN_UNUSED(OutStats);
+		return false;
+	}
+
 	virtual CreateContainerResult CreateContainer() override
 	{
 		// Nothing to do here