use new builds api for oplogs (#464)

- Improvement: Refactored jupiter oplog export code to reuse builds jupiter wrapper classes
- Improvement: If `zen builds`, `zen oplog-import` or `zen oplog-import` command fails due to a http error, the return code for the program will be set to the error/status code

1 files changed, 41 insertions, 35 deletions

diff --git a/src/zenutil/jupiter/jupiterbuildstorage.cpp b/src/zenutil/jupiter/jupiterbuildstorage.cpp
index 9974725ff..c9278acb4 100644
--- a/src/zenutil/jupiter/jupiterbuildstorage.cpp
+++ b/src/zenutil/jupiter/jupiterbuildstorage.cpp
@@ -17,6 +17,16 @@ namespace zen {
 
 using namespace std::literals;
 
+namespace {
+	void ThrowFromJupiterResult(const JupiterResult& Result, std::string_view Prefix)
+	{
+		int				 Error = Result.ErrorCode < (int)HttpResponseCode::Continue ? Result.ErrorCode : 0;
+		HttpResponseCode Status =
+			Result.ErrorCode >= int(HttpResponseCode::Continue) ? HttpResponseCode(Result.ErrorCode) : HttpResponseCode::ImATeapot;
+		throw HttpClientError(fmt::format("{}: {} ({})", Prefix, Result.Reason, Result.ErrorCode), Error, Status);
+	}
+}  // namespace
+
 class JupiterBuildStorage : public BuildStorage
 {
 public:
@@ -46,7 +56,7 @@ public:
 		AddStatistic(ListResult);
 		if (!ListResult.Success)
 		{
-			throw std::runtime_error(fmt::format("Failed listing namespaces: {} ({})", ListResult.Reason, ListResult.ErrorCode));
+			ThrowFromJupiterResult(ListResult, "Failed listing namespaces");
 		}
 		CbObject NamespaceResponse = PayloadToCbObject("Failed listing namespaces"sv, ListResult.Response);
 
@@ -66,10 +76,10 @@ public:
 					AddStatistic(BucketsResult);
 					if (!BucketsResult.Success)
 					{
-						throw std::runtime_error(
-							fmt::format("Failed listing namespaces: {} ({})", BucketsResult.Reason, BucketsResult.ErrorCode));
+						ThrowFromJupiterResult(BucketsResult, fmt::format("Failed listing buckets in namespace {}", Namespace));
 					}
-					CbObject BucketResponse = PayloadToCbObject("Failed listing namespaces"sv, BucketsResult.Response);
+					CbObject BucketResponse =
+						PayloadToCbObject(fmt::format("Failed listing buckets in namespace {}", Namespace), BucketsResult.Response);
 
 					Response.BeginArray("items");
 					for (CbFieldView BucketField : BucketResponse["buckets"])
@@ -103,7 +113,7 @@ public:
 		AddStatistic(ListResult);
 		if (!ListResult.Success)
 		{
-			throw std::runtime_error(fmt::format("Failed listing builds: {} ({})", ListResult.Reason, ListResult.ErrorCode));
+			ThrowFromJupiterResult(ListResult, "Failed listing builds"sv);
 		}
 		return PayloadToCbObject("Failed listing builds"sv, ListResult.Response);
 	}
@@ -120,7 +130,7 @@ public:
 		AddStatistic(PutResult);
 		if (!PutResult.Success)
 		{
-			throw std::runtime_error(fmt::format("Failed creating build: {} ({})", PutResult.Reason, PutResult.ErrorCode));
+			ThrowFromJupiterResult(PutResult, "Failed creating build"sv);
 		}
 		return PayloadToCbObject(fmt::format("Failed creating build: {}", BuildId), PutResult.Response);
 	}
@@ -135,7 +145,7 @@ public:
 		AddStatistic(GetBuildResult);
 		if (!GetBuildResult.Success)
 		{
-			throw std::runtime_error(fmt::format("Failed fetching build: {} ({})", GetBuildResult.Reason, GetBuildResult.ErrorCode));
+			ThrowFromJupiterResult(GetBuildResult, "Failed fetching build"sv);
 		}
 		return PayloadToCbObject(fmt::format("Failed fetching build {}:", BuildId), GetBuildResult.Response);
 	}
@@ -150,8 +160,7 @@ public:
 		AddStatistic(FinalizeBuildResult);
 		if (!FinalizeBuildResult.Success)
 		{
-			throw std::runtime_error(
-				fmt::format("Failed finalizing build part: {} ({})", FinalizeBuildResult.Reason, FinalizeBuildResult.ErrorCode));
+			ThrowFromJupiterResult(FinalizeBuildResult, "Failed finalizing build"sv);
 		}
 	}
 
@@ -170,7 +179,7 @@ public:
 		AddStatistic(PutPartResult);
 		if (!PutPartResult.Success)
 		{
-			throw std::runtime_error(fmt::format("Failed creating build part: {} ({})", PutPartResult.Reason, PutPartResult.ErrorCode));
+			ThrowFromJupiterResult(PutPartResult, "Failed creating build part"sv);
 		}
 		return std::make_pair(PutPartResult.RawHash, std::move(PutPartResult.Needs));
 	}
@@ -185,10 +194,7 @@ public:
 		AddStatistic(GetBuildPartResult);
 		if (!GetBuildPartResult.Success)
 		{
-			throw std::runtime_error(fmt::format("Failed fetching build part {}: {} ({})",
-												 BuildPartId,
-												 GetBuildPartResult.Reason,
-												 GetBuildPartResult.ErrorCode));
+			ThrowFromJupiterResult(GetBuildPartResult, "Failed fetching build part"sv);
 		}
 		return PayloadToCbObject(fmt::format("Failed fetching build part {}:", BuildPartId), GetBuildPartResult.Response);
 	}
@@ -203,8 +209,7 @@ public:
 		AddStatistic(FinalizePartResult);
 		if (!FinalizePartResult.Success)
 		{
-			throw std::runtime_error(
-				fmt::format("Failed finalizing build part: {} ({})", FinalizePartResult.Reason, FinalizePartResult.ErrorCode));
+			ThrowFromJupiterResult(FinalizePartResult, "Failed finalizing build part"sv);
 		}
 		return std::move(FinalizePartResult.Needs);
 	}
@@ -222,7 +227,7 @@ public:
 		AddStatistic(PutBlobResult);
 		if (!PutBlobResult.Success)
 		{
-			throw std::runtime_error(fmt::format("Failed putting build part: {} ({})", PutBlobResult.Reason, PutBlobResult.ErrorCode));
+			ThrowFromJupiterResult(PutBlobResult, "Failed putting build part"sv);
 		}
 	}
 
@@ -249,8 +254,7 @@ public:
 		AddStatistic(PutMultipartBlobResult);
 		if (!PutMultipartBlobResult.Success)
 		{
-			throw std::runtime_error(
-				fmt::format("Failed putting build part: {} ({})", PutMultipartBlobResult.Reason, PutMultipartBlobResult.ErrorCode));
+			ThrowFromJupiterResult(PutMultipartBlobResult, "Failed putting large build blob"sv);
 		}
 		OnSentBytes(PutMultipartBlobResult.SentBytes, WorkItems.empty());
 
@@ -265,7 +269,7 @@ public:
 				AddStatistic(PartResult);
 				if (!PartResult.Success)
 				{
-					throw std::runtime_error(fmt::format("Failed putting build part: {} ({})", PartResult.Reason, PartResult.ErrorCode));
+					ThrowFromJupiterResult(PartResult, "Failed putting large build blob"sv);
 				}
 				OnSentBytes(PartResult.SentBytes, IsComplete);
 			});
@@ -285,8 +289,7 @@ public:
 		AddStatistic(GetBuildBlobResult);
 		if (!GetBuildBlobResult.Success)
 		{
-			throw std::runtime_error(
-				fmt::format("Failed fetching build blob {}: {} ({})", RawHash, GetBuildBlobResult.Reason, GetBuildBlobResult.ErrorCode));
+			ThrowFromJupiterResult(GetBuildBlobResult, "Failed fetching build blob"sv);
 		}
 		return std::move(GetBuildBlobResult.Response);
 	}
@@ -314,8 +317,7 @@ public:
 		AddStatistic(GetMultipartBlobResult);
 		if (!GetMultipartBlobResult.Success)
 		{
-			throw std::runtime_error(
-				fmt::format("Failed getting build part: {} ({})", GetMultipartBlobResult.Reason, GetMultipartBlobResult.ErrorCode));
+			ThrowFromJupiterResult(GetMultipartBlobResult, "Failed getting large build part"sv);
 		}
 		std::vector<std::function<void()>> WorkList;
 		for (auto& WorkItem : WorkItems)
@@ -327,7 +329,7 @@ public:
 				AddStatistic(PartResult);
 				if (!PartResult.Success)
 				{
-					throw std::runtime_error(fmt::format("Failed getting build part: {} ({})", PartResult.Reason, PartResult.ErrorCode));
+					ThrowFromJupiterResult(PartResult, "Failed getting large build part"sv);
 				}
 			});
 		}
@@ -350,8 +352,7 @@ public:
 			{
 				return false;
 			}
-			throw std::runtime_error(
-				fmt::format("Failed putting build block metadata: {} ({})", PutMetaResult.Reason, PutMetaResult.ErrorCode));
+			ThrowFromJupiterResult(PutMetaResult, "Failed putting build block metadata"sv);
 		}
 		return true;
 	}
@@ -366,7 +367,7 @@ public:
 		AddStatistic(FindResult);
 		if (!FindResult.Success)
 		{
-			throw std::runtime_error(fmt::format("Failed fetching known blocks: {} ({})", FindResult.Reason, FindResult.ErrorCode));
+			ThrowFromJupiterResult(FindResult, "Failed fetching known blocks"sv);
 		}
 		return PayloadToCbObject("Failed fetching known blocks"sv, FindResult.Response);
 	}
@@ -392,8 +393,7 @@ public:
 		AddStatistic(GetBlockMetadataResult);
 		if (!GetBlockMetadataResult.Success)
 		{
-			throw std::runtime_error(
-				fmt::format("Failed fetching block metadatas: {} ({})", GetBlockMetadataResult.Reason, GetBlockMetadataResult.ErrorCode));
+			ThrowFromJupiterResult(GetBlockMetadataResult, "Failed fetching block metadatas"sv);
 		}
 		return PayloadToCbObject("Failed fetching block metadatas", GetBlockMetadataResult.Response);
 	}
@@ -416,14 +416,12 @@ public:
 		AddStatistic(PutBuildPartStatsResult);
 		if (!PutBuildPartStatsResult.Success)
 		{
-			throw std::runtime_error(fmt::format("Failed posting build part statistics: {} ({})",
-												 PutBuildPartStatsResult.Reason,
-												 PutBuildPartStatsResult.ErrorCode));
+			ThrowFromJupiterResult(PutBuildPartStatsResult, "Failed posting build part statistics"sv);
 		}
 	}
 
 private:
-	static CbObject PayloadToCbObject(std::string_view Context, const IoBuffer& Payload)
+	static CbObject PayloadToCbObject(std::string_view ErrorContext, const IoBuffer& Payload)
 	{
 		if (Payload.GetContentType() == ZenContentType::kJSON)
 		{
@@ -443,7 +441,7 @@ private:
 		else
 		{
 			throw std::runtime_error(
-				fmt::format("{}: {} ({})", "Unsupported response format", Context, ToString(Payload.GetContentType())));
+				fmt::format("{}: {} ({})", "Unsupported response format", ErrorContext, ToString(Payload.GetContentType())));
 		}
 	}
 
@@ -453,6 +451,14 @@ private:
 		m_Stats.TotalBytesRead += Result.ReceivedBytes;
 		m_Stats.TotalRequestTimeUs += uint64_t(Result.ElapsedSeconds * 1000000.0);
 		m_Stats.TotalRequestCount++;
+
+		SetAtomicMax(m_Stats.PeakSentBytes, Result.SentBytes);
+		SetAtomicMax(m_Stats.PeakReceivedBytes, Result.ReceivedBytes);
+		if (Result.ElapsedSeconds > 0.0)
+		{
+			uint64_t BytesPerSec = uint64_t((Result.SentBytes + Result.ReceivedBytes) / Result.ElapsedSeconds);
+			SetAtomicMax(m_Stats.PeakBytesPerSec, BytesPerSec);
+		}
 	}
 
 	JupiterSession				m_Session;