builds upload command (#278)

- Feature: **EXPERIMENTAL** New `zen builds` command to list, upload and download folders to Cloud Build API
  - `builds list` list available builds (**INCOMPLETE - FILTERING MISSING**)
  - `builds upload` upload a folder to Cloud Build API
	  - `--local-path` source folder to upload
	  - `--create-build` creates a new parent build object (using the object id), if omitted a parent build must exist and `--build-id` must be given
	  - `--build-id` an Oid in hex form for the Build identifier to use - omit to have the id auto generated
	  - `--build-part-id` and Oid in hex form for the Build Part identifier for the folder - omit to have the id auto generated
	  - `--build-part-name` name of the build part - if omitted the name of the leaf folder name give in `--local-path`
	  - `--metadata-path` path to a json formatted file with meta data information about the build. Meta-data must be provided if `--create-build` is set
	  - `--metadata` key-value pairs separated by ';' with build meta data for the build. (key1=value1;key2=value2). Meta-data must be provided if `--create-build` is set
	  - `--clean` ignore any existing blocks of chunk data and upload a fresh set of blocks
	  - `--allow-multipart` enable usage of multi-part http upload requests
	  - `--manifest-path` path to text file listing files to include in upload. Exclude to upload everything in `--local-path`
  - `builds download` download a folder from Cloud Build API (**INCOMPLETE - WILL WIPE UNTRACKED DATA FROM TARGET FOLDER**)
	  - `--local-path` target folder to download to
	  - `--build-id` an Oid in hex form for the Build identifier to use
	  - `--build-part-id` a comma separated list of Oid in hex for the build part identifier(s) to download - mutually exclusive to `--build-part-name`
	  - `--build-part-name` a comma separated list of names for the build part(s) to download - if omitted the name of the leaf folder name give in `--local-path`
	  - `--clean` deletes all data in target folder before downloading (NON-CLEAN IS NOT IMPLEMENTED YET)
	  - `--allow-multipart` enable usage of multi-part http download reqeusts
  - `builds diff` download a folder from Cloud Build API
	- `--local-path` target folder to download to
	- `--compare-path` folder to compare target with
	- `--only-chunked` compare only files that would be chunked
  - `builds fetch-blob` fetch and validate a blob from remote store
	  - `--build-id` an Oid in hex form for the Build identifier to use
    - `--blob-hash` an IoHash in hex form identifying the blob to download
  - `builds validate part` fetch a build part and validate all referenced attachments
	  - `--build-id` an Oid in hex form for the Build identifier to use
	  - `--build-part-id` an Oid in hex for the build part identifier to validate - mutually exclusive to `--build-part-name`
	  - `--build-part-name` a name for the build part to validate - mutually exclusive to `--build-part-id`
  - `builds test` a series of operation that uploads, downloads and test various aspects of incremental operations
	  - `--local-path` source folder to upload
  - Options for Cloud Build API remote store (`list`, `upload`, `download`, `fetch-blob`, `validate-part`)
	  - `--url` Cloud Builds URL
    - `--assume-http2` assume that the builds endpoint is a HTTP/2 endpoint skipping HTTP/1.1 upgrade handshake
    - `--namespace` Builds Storage namespace
    - `--bucket` Builds Storage bucket
    - Authentication options for Cloud Build API
	    - Auth token
	      - `--access-token` http auth Cloud Storage access token
	      - `--access-token-env` name of environment variable that holds the Http auth Cloud Storage access token
	      - `--access-token-path` path to json file that holds the Http auth Cloud Storage access token
	    - OpenId authentication
	      - `--openid-provider-name` Open ID provider name
	      - `--openid-provider-url` Open ID provider url
	      - `--openid-client-id`Open ID client id
	      - `--openid-refresh-token` Open ID refresh token
	      - `--encryption-aes-key` 256 bit AES encryption key for storing OpenID credentials
	      - `--encryption-aes-iv` 128 bit AES encryption initialization vector for storing OpenID credentials
	    - OAuth authentication
	      - `--oauth-url` OAuth provier url
	      - `--oauth-clientid` OAuth client id
	      - `--oauth-clientsecret` OAuth client secret
  - Options for file based remote store used for for testing purposes (`list`, `upload`, `download`, `fetch-blob`, `validate-part`, `test`)
	  - `--storage-path` path to folder to store builds data
    - `--json-metadata` enable json output in store for all compact binary objects (off by default)
  - Output options for all builds commands
    - `--plain-progress` use plain line-by-line progress output
    - `--verbose`

1 files changed, 371 insertions, 0 deletions

diff --git a/src/zenutil/jupiter/jupiterbuildstorage.cpp b/src/zenutil/jupiter/jupiterbuildstorage.cpp
new file mode 100644
index 000000000..481e9146f
--- /dev/null
+++ b/src/zenutil/jupiter/jupiterbuildstorage.cpp
@@ -0,0 +1,371 @@
+// Copyright Epic Games, Inc. All Rights Reserved.
+
+#include <zenutil/jupiter/jupiterbuildstorage.h>
+
+#include <zencore/compactbinarybuilder.h>
+#include <zencore/fmtutils.h>
+#include <zencore/scopeguard.h>
+#include <zencore/timer.h>
+#include <zenutil/jupiter/jupitersession.h>
+
+ZEN_THIRD_PARTY_INCLUDES_START
+#include <tsl/robin_map.h>
+ZEN_THIRD_PARTY_INCLUDES_END
+
+namespace zen {
+
+using namespace std::literals;
+
+class JupiterBuildStorage : public BuildStorage
+{
+public:
+	JupiterBuildStorage(LoggerRef					 InLog,
+						HttpClient&					 InHttpClient,
+						Statistics&					 Stats,
+						std::string_view			 Namespace,
+						std::string_view			 Bucket,
+						const std::filesystem::path& TempFolderPath)
+	: m_Session(InLog, InHttpClient)
+	, m_Stats(Stats)
+	, m_Namespace(Namespace)
+	, m_Bucket(Bucket)
+	, m_TempFolderPath(TempFolderPath)
+	{
+	}
+	virtual ~JupiterBuildStorage() {}
+
+	virtual CbObject ListBuilds(CbObject Query) override
+	{
+		Stopwatch ExecutionTimer;
+		auto	  _		  = MakeGuard([&]() { m_Stats.TotalExecutionTimeUs += ExecutionTimer.GetElapsedTimeUs(); });
+		IoBuffer  Payload = Query.GetBuffer().AsIoBuffer();
+		Payload.SetContentType(ZenContentType::kCbObject);
+		JupiterResult ListResult = m_Session.ListBuilds(m_Namespace, m_Bucket, Payload);
+		AddStatistic(ListResult);
+		if (!ListResult.Success)
+		{
+			throw std::runtime_error(fmt::format("Failed listing builds: {} ({})", ListResult.Reason, ListResult.ErrorCode));
+		}
+		return PayloadToJson("Failed listing builds"sv, ListResult.Response);
+	}
+
+	virtual CbObject PutBuild(const Oid& BuildId, const CbObject& MetaData) override
+	{
+		Stopwatch ExecutionTimer;
+		auto	  _		  = MakeGuard([&]() { m_Stats.TotalExecutionTimeUs += ExecutionTimer.GetElapsedTimeUs(); });
+		IoBuffer  Payload = MetaData.GetBuffer().AsIoBuffer();
+		Payload.SetContentType(ZenContentType::kCbObject);
+		JupiterResult PutResult = m_Session.PutBuild(m_Namespace, m_Bucket, BuildId, Payload);
+		AddStatistic(PutResult);
+		if (!PutResult.Success)
+		{
+			throw std::runtime_error(fmt::format("Failed creating build: {} ({})", PutResult.Reason, PutResult.ErrorCode));
+		}
+		return PayloadToJson(fmt::format("Failed creating build: {}", BuildId), PutResult.Response);
+	}
+
+	virtual CbObject GetBuild(const Oid& BuildId) override
+	{
+		Stopwatch	  ExecutionTimer;
+		auto		  _				 = MakeGuard([&]() { m_Stats.TotalExecutionTimeUs += ExecutionTimer.GetElapsedTimeUs(); });
+		JupiterResult GetBuildResult = m_Session.GetBuild(m_Namespace, m_Bucket, BuildId);
+		AddStatistic(GetBuildResult);
+		if (!GetBuildResult.Success)
+		{
+			throw std::runtime_error(fmt::format("Failed fetching build: {} ({})", GetBuildResult.Reason, GetBuildResult.ErrorCode));
+		}
+		return PayloadToJson(fmt::format("Failed fetching build {}:", BuildId), GetBuildResult.Response);
+	}
+
+	virtual void FinalizeBuild(const Oid& BuildId) override
+	{
+		Stopwatch	  ExecutionTimer;
+		auto		  _					  = MakeGuard([&]() { m_Stats.TotalExecutionTimeUs += ExecutionTimer.GetElapsedTimeUs(); });
+		JupiterResult FinalizeBuildResult = m_Session.FinalizeBuild(m_Namespace, m_Bucket, BuildId);
+		AddStatistic(FinalizeBuildResult);
+		if (!FinalizeBuildResult.Success)
+		{
+			throw std::runtime_error(
+				fmt::format("Failed finalizing build part: {} ({})", FinalizeBuildResult.Reason, FinalizeBuildResult.ErrorCode));
+		}
+	}
+
+	virtual std::pair<IoHash, std::vector<IoHash>> PutBuildPart(const Oid&		 BuildId,
+																const Oid&		 BuildPartId,
+																std::string_view PartName,
+																const CbObject&	 MetaData) override
+	{
+		Stopwatch ExecutionTimer;
+		auto	  _		  = MakeGuard([&]() { m_Stats.TotalExecutionTimeUs += ExecutionTimer.GetElapsedTimeUs(); });
+		IoBuffer  Payload = MetaData.GetBuffer().AsIoBuffer();
+		Payload.SetContentType(ZenContentType::kCbObject);
+		PutBuildPartResult PutPartResult = m_Session.PutBuildPart(m_Namespace, m_Bucket, BuildId, BuildPartId, PartName, Payload);
+		AddStatistic(PutPartResult);
+		if (!PutPartResult.Success)
+		{
+			throw std::runtime_error(fmt::format("Failed creating build part: {} ({})", PutPartResult.Reason, PutPartResult.ErrorCode));
+		}
+		return std::make_pair(PutPartResult.RawHash, std::move(PutPartResult.Needs));
+	}
+
+	virtual CbObject GetBuildPart(const Oid& BuildId, const Oid& BuildPartId) override
+	{
+		Stopwatch	  ExecutionTimer;
+		auto		  _					 = MakeGuard([&]() { m_Stats.TotalExecutionTimeUs += ExecutionTimer.GetElapsedTimeUs(); });
+		JupiterResult GetBuildPartResult = m_Session.GetBuildPart(m_Namespace, m_Bucket, BuildId, BuildPartId);
+		AddStatistic(GetBuildPartResult);
+		if (!GetBuildPartResult.Success)
+		{
+			throw std::runtime_error(fmt::format("Failed fetching build part {}: {} ({})",
+												 BuildPartId,
+												 GetBuildPartResult.Reason,
+												 GetBuildPartResult.ErrorCode));
+		}
+		return PayloadToJson(fmt::format("Failed fetching build part {}:", BuildPartId), GetBuildPartResult.Response);
+	}
+
+	virtual std::vector<IoHash> FinalizeBuildPart(const Oid& BuildId, const Oid& BuildPartId, const IoHash& PartHash) override
+	{
+		Stopwatch				ExecutionTimer;
+		auto					_ = MakeGuard([&]() { m_Stats.TotalExecutionTimeUs += ExecutionTimer.GetElapsedTimeUs(); });
+		FinalizeBuildPartResult FinalizePartResult = m_Session.FinalizeBuildPart(m_Namespace, m_Bucket, BuildId, BuildPartId, PartHash);
+		AddStatistic(FinalizePartResult);
+		if (!FinalizePartResult.Success)
+		{
+			throw std::runtime_error(
+				fmt::format("Failed finalizing build part: {} ({})", FinalizePartResult.Reason, FinalizePartResult.ErrorCode));
+		}
+		return std::move(FinalizePartResult.Needs);
+	}
+
+	virtual void PutBuildBlob(const Oid&			 BuildId,
+							  const IoHash&			 RawHash,
+							  ZenContentType		 ContentType,
+							  const CompositeBuffer& Payload) override
+	{
+		Stopwatch	  ExecutionTimer;
+		auto		  _				= MakeGuard([&]() { m_Stats.TotalExecutionTimeUs += ExecutionTimer.GetElapsedTimeUs(); });
+		JupiterResult PutBlobResult = m_Session.PutBuildBlob(m_Namespace, m_Bucket, BuildId, RawHash, ContentType, Payload);
+		AddStatistic(PutBlobResult);
+		if (!PutBlobResult.Success)
+		{
+			throw std::runtime_error(fmt::format("Failed putting build part: {} ({})", PutBlobResult.Reason, PutBlobResult.ErrorCode));
+		}
+	}
+
+	virtual std::vector<std::function<void()>> PutLargeBuildBlob(const Oid&												   BuildId,
+																 const IoHash&											   RawHash,
+																 ZenContentType											   ContentType,
+																 uint64_t												   PayloadSize,
+																 std::function<IoBuffer(uint64_t Offset, uint64_t Size)>&& Transmitter,
+																 std::function<void(uint64_t, bool)>&& OnSentBytes) override
+	{
+		Stopwatch ExecutionTimer;
+		auto	  _ = MakeGuard([&]() { m_Stats.TotalExecutionTimeUs += ExecutionTimer.GetElapsedTimeUs(); });
+		std::vector<std::function<JupiterResult(bool&)>> WorkItems;
+		JupiterResult									 PutMultipartBlobResult = m_Session.PutMultipartBuildBlob(m_Namespace,
+																				  m_Bucket,
+																				  BuildId,
+																				  RawHash,
+																				  ContentType,
+																				  PayloadSize,
+																				  std::move(Transmitter),
+																				  WorkItems);
+		AddStatistic(PutMultipartBlobResult);
+		if (!PutMultipartBlobResult.Success)
+		{
+			throw std::runtime_error(
+				fmt::format("Failed putting build part: {} ({})", PutMultipartBlobResult.Reason, PutMultipartBlobResult.ErrorCode));
+		}
+		OnSentBytes(PutMultipartBlobResult.SentBytes, WorkItems.empty());
+
+		std::vector<std::function<void()>> WorkList;
+		for (auto& WorkItem : WorkItems)
+		{
+			WorkList.emplace_back([this, WorkItem = std::move(WorkItem), OnSentBytes]() {
+				Stopwatch	  ExecutionTimer;
+				auto		  _			 = MakeGuard([&]() { m_Stats.TotalExecutionTimeUs += ExecutionTimer.GetElapsedTimeUs(); });
+				bool		  IsComplete = false;
+				JupiterResult PartResult = WorkItem(IsComplete);
+				AddStatistic(PartResult);
+				if (!PartResult.Success)
+				{
+					throw std::runtime_error(fmt::format("Failed putting build part: {} ({})", PartResult.Reason, PartResult.ErrorCode));
+				}
+				OnSentBytes(PartResult.SentBytes, IsComplete);
+			});
+		}
+		return WorkList;
+	}
+
+	virtual IoBuffer GetBuildBlob(const Oid& BuildId, const IoHash& RawHash) override
+	{
+		Stopwatch	  ExecutionTimer;
+		auto		  _					 = MakeGuard([&]() { m_Stats.TotalExecutionTimeUs += ExecutionTimer.GetElapsedTimeUs(); });
+		JupiterResult GetBuildBlobResult = m_Session.GetBuildBlob(m_Namespace, m_Bucket, BuildId, RawHash, m_TempFolderPath);
+		AddStatistic(GetBuildBlobResult);
+		if (!GetBuildBlobResult.Success)
+		{
+			throw std::runtime_error(
+				fmt::format("Failed fetching build blob {}: {} ({})", RawHash, GetBuildBlobResult.Reason, GetBuildBlobResult.ErrorCode));
+		}
+		return std::move(GetBuildBlobResult.Response);
+	}
+
+	virtual std::vector<std::function<void()>> GetLargeBuildBlob(
+		const Oid&																			   BuildId,
+		const IoHash&																		   RawHash,
+		uint64_t																			   ChunkSize,
+		std::function<void(uint64_t Offset, const IoBuffer& Chunk, uint64_t BytesRemaining)>&& Receiver) override
+	{
+		Stopwatch ExecutionTimer;
+		auto	  _ = MakeGuard([&]() { m_Stats.TotalExecutionTimeUs += ExecutionTimer.GetElapsedTimeUs(); });
+		std::vector<std::function<JupiterResult()>> WorkItems;
+		JupiterResult								GetMultipartBlobResult =
+			m_Session.GetMultipartBuildBlob(m_Namespace, m_Bucket, BuildId, RawHash, ChunkSize, std::move(Receiver), WorkItems);
+
+		AddStatistic(GetMultipartBlobResult);
+		if (!GetMultipartBlobResult.Success)
+		{
+			throw std::runtime_error(
+				fmt::format("Failed getting build part: {} ({})", GetMultipartBlobResult.Reason, GetMultipartBlobResult.ErrorCode));
+		}
+		std::vector<std::function<void()>> WorkList;
+		for (auto& WorkItem : WorkItems)
+		{
+			WorkList.emplace_back([this, WorkItem = std::move(WorkItem)]() {
+				Stopwatch	  ExecutionTimer;
+				auto		  _			 = MakeGuard([&]() { m_Stats.TotalExecutionTimeUs += ExecutionTimer.GetElapsedTimeUs(); });
+				JupiterResult PartResult = WorkItem();
+				AddStatistic(PartResult);
+				if (!PartResult.Success)
+				{
+					throw std::runtime_error(fmt::format("Failed getting build part: {} ({})", PartResult.Reason, PartResult.ErrorCode));
+				}
+			});
+		}
+		return WorkList;
+	}
+
+	virtual void PutBlockMetadata(const Oid& BuildId, const IoHash& BlockRawHash, const CbObject& MetaData) override
+	{
+		Stopwatch ExecutionTimer;
+		auto	  _		  = MakeGuard([&]() { m_Stats.TotalExecutionTimeUs += ExecutionTimer.GetElapsedTimeUs(); });
+		IoBuffer  Payload = MetaData.GetBuffer().AsIoBuffer();
+		Payload.SetContentType(ZenContentType::kCbObject);
+		JupiterResult PutMetaResult = m_Session.PutBlockMetadata(m_Namespace, m_Bucket, BuildId, BlockRawHash, Payload);
+		AddStatistic(PutMetaResult);
+		if (!PutMetaResult.Success)
+		{
+			throw std::runtime_error(
+				fmt::format("Failed putting build block metadata: {} ({})", PutMetaResult.Reason, PutMetaResult.ErrorCode));
+		}
+	}
+
+	virtual std::vector<ChunkBlockDescription> FindBlocks(const Oid& BuildId) override
+	{
+		Stopwatch	  ExecutionTimer;
+		auto		  _			 = MakeGuard([&]() { m_Stats.TotalExecutionTimeUs += ExecutionTimer.GetElapsedTimeUs(); });
+		JupiterResult FindResult = m_Session.FindBlocks(m_Namespace, m_Bucket, BuildId);
+		AddStatistic(FindResult);
+		if (!FindResult.Success)
+		{
+			throw std::runtime_error(fmt::format("Failed fetching known blocks: {} ({})", FindResult.Reason, FindResult.ErrorCode));
+		}
+		return ParseChunkBlockDescriptionList(PayloadToJson("Failed fetching known blocks"sv, FindResult.Response));
+	}
+
+	virtual std::vector<ChunkBlockDescription> GetBlockMetadata(const Oid& BuildId, std::span<const IoHash> BlockHashes) override
+	{
+		Stopwatch	   ExecutionTimer;
+		auto		   _ = MakeGuard([&]() { m_Stats.TotalExecutionTimeUs += ExecutionTimer.GetElapsedTimeUs(); });
+		CbObjectWriter Request;
+
+		Request.BeginArray("blocks"sv);
+		for (const IoHash& BlockHash : BlockHashes)
+		{
+			Request.AddHash(BlockHash);
+		}
+		Request.EndArray();
+
+		IoBuffer Payload = Request.Save().GetBuffer().AsIoBuffer();
+		Payload.SetContentType(ZenContentType::kCbObject);
+		JupiterResult GetBlockMetadataResult = m_Session.GetBlockMetadata(m_Namespace, m_Bucket, BuildId, Payload);
+		AddStatistic(GetBlockMetadataResult);
+		if (!GetBlockMetadataResult.Success)
+		{
+			throw std::runtime_error(
+				fmt::format("Failed fetching block metadatas: {} ({})", GetBlockMetadataResult.Reason, GetBlockMetadataResult.ErrorCode));
+		}
+		std::vector<ChunkBlockDescription> UnorderedList =
+			ParseChunkBlockDescriptionList(PayloadToJson("Failed fetching block metadatas", GetBlockMetadataResult.Response));
+		tsl::robin_map<IoHash, size_t, IoHash::Hasher> BlockDescriptionLookup;
+		for (size_t DescriptionIndex = 0; DescriptionIndex < UnorderedList.size(); DescriptionIndex++)
+		{
+			const ChunkBlockDescription& Description = UnorderedList[DescriptionIndex];
+			BlockDescriptionLookup.insert_or_assign(Description.BlockHash, DescriptionIndex);
+		}
+		std::vector<ChunkBlockDescription> SortedBlockDescriptions;
+		SortedBlockDescriptions.reserve(BlockDescriptionLookup.size());
+		for (const IoHash& BlockHash : BlockHashes)
+		{
+			if (auto It = BlockDescriptionLookup.find(BlockHash); It != BlockDescriptionLookup.end())
+			{
+				SortedBlockDescriptions.push_back(std::move(UnorderedList[It->second]));
+			}
+		}
+		return SortedBlockDescriptions;
+	}
+
+private:
+	static CbObject PayloadToJson(std::string_view Context, const IoBuffer& Payload)
+	{
+		if (Payload.GetContentType() == ZenContentType::kJSON)
+		{
+			std::string_view Json(reinterpret_cast<const char*>(Payload.GetData()), Payload.GetSize());
+			return LoadCompactBinaryFromJson(Json).AsObject();
+		}
+		else if (Payload.GetContentType() == ZenContentType::kCbObject)
+		{
+			return LoadCompactBinaryObject(Payload);
+		}
+		else if (Payload.GetContentType() == ZenContentType::kCompressedBinary)
+		{
+			IoHash	 RawHash;
+			uint64_t RawSize;
+			return LoadCompactBinaryObject(CompressedBuffer::FromCompressed(SharedBuffer(Payload), RawHash, RawSize));
+		}
+		else
+		{
+			throw std::runtime_error(
+				fmt::format("{}: {} ({})", "Unsupported response format", Context, ToString(Payload.GetContentType())));
+		}
+	}
+
+	void AddStatistic(const JupiterResult& Result)
+	{
+		m_Stats.TotalBytesWritten += Result.SentBytes;
+		m_Stats.TotalBytesRead += Result.ReceivedBytes;
+		m_Stats.TotalRequestTimeUs += uint64_t(Result.ElapsedSeconds * 1000000.0);
+		m_Stats.TotalRequestCount++;
+	}
+
+	JupiterSession				m_Session;
+	Statistics&					m_Stats;
+	const std::string			m_Namespace;
+	const std::string			m_Bucket;
+	const std::filesystem::path m_TempFolderPath;
+};
+
+std::unique_ptr<BuildStorage>
+CreateJupiterBuildStorage(LoggerRef					   InLog,
+						  HttpClient&				   InHttpClient,
+						  BuildStorage::Statistics&	   Stats,
+						  std::string_view			   Namespace,
+						  std::string_view			   Bucket,
+						  const std::filesystem::path& TempFolderPath)
+{
+	return std::make_unique<JupiterBuildStorage>(InLog, InHttpClient, Stats, Namespace, Bucket, TempFolderPath);
+}
+
+}  // namespace zen