De/more upstream details (#168)v0.1.6-pre9v0.1.6-pre10

* Pass along endpoint info for each upstream request
* Add more timing details in log
* more log details for single item upstream fetch
* DISABLEDQUERY over SKIP

1 files changed, 69 insertions, 63 deletions

diff --git a/zenserver/upstream/upstreamcache.cpp b/zenserver/upstream/upstreamcache.cpp
index 1aa195469..c1c0395e7 100644
--- a/zenserver/upstream/upstreamcache.cpp
+++ b/zenserver/upstream/upstreamcache.cpp
@@ -180,7 +180,9 @@ namespace detail {
 
 		virtual UpstreamEndpointStatus GetStatus() override { return m_Status.EndpointStatus(); }
 
-		virtual GetUpstreamCacheResult GetCacheRecord(std::string_view Namespace, const CacheKey& CacheKey, ZenContentType Type) override
+		virtual GetUpstreamCacheSingleResult GetCacheRecord(std::string_view Namespace,
+															const CacheKey&	 CacheKey,
+															ZenContentType	 Type) override
 		{
 			ZEN_TRACE_CPU("Upstream::Horde::GetSingleCacheRecord");
 
@@ -274,21 +276,20 @@ namespace detail {
 
 				if (Result.ErrorCode == 0)
 				{
-					return {.Value			= Result.Response,
-							.Bytes			= Result.Bytes,
-							.ElapsedSeconds = Result.ElapsedSeconds,
-							.Success		= Result.Success};
+					return {.Status = {.Bytes = Result.Bytes, .ElapsedSeconds = Result.ElapsedSeconds, .Success = Result.Success},
+							.Value	= Result.Response,
+							.Source = &m_Info};
 				}
 				else
 				{
-					return {.Error{.ErrorCode = Result.ErrorCode, .Reason = std::move(Result.Reason)}};
+					return {.Status = {.Error{.ErrorCode = Result.ErrorCode, .Reason = std::move(Result.Reason)}}};
 				}
 			}
 			catch (std::exception& Err)
 			{
 				m_Status.Set(UpstreamEndpointState::kError, Err.what());
 
-				return {.Error{.ErrorCode = -1, .Reason = Err.what()}};
+				return {.Status = {.Error{.ErrorCode = -1, .Reason = Err.what()}}};
 			}
 		}
 
@@ -342,13 +343,16 @@ namespace detail {
 					}
 				}
 
-				OnComplete({.Request = *Request, .Record = Record, .Package = Package, .ElapsedSeconds = ElapsedSeconds});
+				OnComplete(
+					{.Request = *Request, .Record = Record, .Package = Package, .ElapsedSeconds = ElapsedSeconds, .Source = &m_Info});
 			}
 
 			return Result;
 		}
 
-		virtual GetUpstreamCacheResult GetCacheChunk(std::string_view Namespace, const CacheKey&, const IoHash& ValueContentId) override
+		virtual GetUpstreamCacheSingleResult GetCacheChunk(std::string_view Namespace,
+														   const CacheKey&,
+														   const IoHash& ValueContentId) override
 		{
 			ZEN_TRACE_CPU("Upstream::Horde::GetSingleCacheChunk");
 
@@ -362,21 +366,20 @@ namespace detail {
 
 				if (Result.ErrorCode == 0)
 				{
-					return {.Value			= Result.Response,
-							.Bytes			= Result.Bytes,
-							.ElapsedSeconds = Result.ElapsedSeconds,
-							.Success		= Result.Success};
+					return {.Status = {.Bytes = Result.Bytes, .ElapsedSeconds = Result.ElapsedSeconds, .Success = Result.Success},
+							.Value	= Result.Response,
+							.Source = &m_Info};
 				}
 				else
 				{
-					return {.Error{.ErrorCode = Result.ErrorCode, .Reason = std::move(Result.Reason)}};
+					return {.Status = {.Error{.ErrorCode = Result.ErrorCode, .Reason = std::move(Result.Reason)}}};
 				}
 			}
 			catch (std::exception& Err)
 			{
 				m_Status.Set(UpstreamEndpointState::kError, Err.what());
 
-				return {.Error{.ErrorCode = -1, .Reason = Err.what()}};
+				return {.Status = {.Error{.ErrorCode = -1, .Reason = Err.what()}}};
 			}
 		}
 
@@ -418,7 +421,8 @@ namespace detail {
 								.RawHash		= IoHash::FromBLAKE3(Compressed.GetRawHash()),
 								.RawSize		= Compressed.GetRawSize(),
 								.Value			= Payload,
-								.ElapsedSeconds = ElapsedSeconds});
+								.ElapsedSeconds = ElapsedSeconds,
+								.Source			= &m_Info});
 				}
 				else
 				{
@@ -487,7 +491,8 @@ namespace detail {
 								.RawHash		= IoHash::FromBLAKE3(Compressed.GetRawHash()),
 								.RawSize		= Compressed.GetRawSize(),
 								.Value			= Payload,
-								.ElapsedSeconds = ElapsedSeconds});
+								.ElapsedSeconds = ElapsedSeconds,
+								.Source			= &m_Info});
 				}
 				else
 				{
@@ -808,7 +813,9 @@ namespace detail {
 
 		virtual UpstreamEndpointStatus GetStatus() override { return m_Status.EndpointStatus(); }
 
-		virtual GetUpstreamCacheResult GetCacheRecord(std::string_view Namespace, const CacheKey& CacheKey, ZenContentType Type) override
+		virtual GetUpstreamCacheSingleResult GetCacheRecord(std::string_view Namespace,
+															const CacheKey&	 CacheKey,
+															ZenContentType	 Type) override
 		{
 			ZEN_TRACE_CPU("Upstream::Zen::GetSingleCacheRecord");
 
@@ -821,21 +828,20 @@ namespace detail {
 
 				if (Result.ErrorCode == 0)
 				{
-					return {.Value			= Result.Response,
-							.Bytes			= Result.Bytes,
-							.ElapsedSeconds = Result.ElapsedSeconds,
-							.Success		= Result.Success};
+					return {.Status = {.Bytes = Result.Bytes, .ElapsedSeconds = Result.ElapsedSeconds, .Success = Result.Success},
+							.Value	= Result.Response,
+							.Source = &m_Info};
 				}
 				else
 				{
-					return {.Error{.ErrorCode = Result.ErrorCode, .Reason = std::move(Result.Reason)}};
+					return {.Status = {.Error{.ErrorCode = Result.ErrorCode, .Reason = std::move(Result.Reason)}}};
 				}
 			}
 			catch (std::exception& Err)
 			{
 				m_Status.Set(UpstreamEndpointState::kError, Err.what());
 
-				return {.Error{.ErrorCode = -1, .Reason = Err.what()}};
+				return {.Status = {.Error{.ErrorCode = -1, .Reason = Err.what()}}};
 			}
 		}
 
@@ -908,7 +914,8 @@ namespace detail {
 							OnComplete({.Request		= *Request,
 										.Record			= Record.AsObjectView(),
 										.Package		= BatchResponse,
-										.ElapsedSeconds = Result.ElapsedSeconds});
+										.ElapsedSeconds = Result.ElapsedSeconds,
+										.Source			= &m_Info});
 						}
 
 						return {.Bytes = Result.Bytes, .ElapsedSeconds = Result.ElapsedSeconds, .Success = true};
@@ -924,9 +931,9 @@ namespace detail {
 			return {.Error{.ErrorCode = Result.ErrorCode, .Reason = std::move(Result.Reason)}};
 		}
 
-		virtual GetUpstreamCacheResult GetCacheChunk(std::string_view Namespace,
-													 const CacheKey&  CacheKey,
-													 const IoHash&	  ValueContentId) override
+		virtual GetUpstreamCacheSingleResult GetCacheChunk(std::string_view Namespace,
+														   const CacheKey&	CacheKey,
+														   const IoHash&	ValueContentId) override
 		{
 			ZEN_TRACE_CPU("Upstream::Zen::GetCacheChunk");
 
@@ -939,21 +946,20 @@ namespace detail {
 
 				if (Result.ErrorCode == 0)
 				{
-					return {.Value			= Result.Response,
-							.Bytes			= Result.Bytes,
-							.ElapsedSeconds = Result.ElapsedSeconds,
-							.Success		= Result.Success};
+					return {.Status = {.Bytes = Result.Bytes, .ElapsedSeconds = Result.ElapsedSeconds, .Success = Result.Success},
+							.Value	= Result.Response,
+							.Source = &m_Info};
 				}
 				else
 				{
-					return {.Error{.ErrorCode = Result.ErrorCode, .Reason = std::move(Result.Reason)}};
+					return {.Status = {.Error{.ErrorCode = Result.ErrorCode, .Reason = std::move(Result.Reason)}}};
 				}
 			}
 			catch (std::exception& Err)
 			{
 				m_Status.Set(UpstreamEndpointState::kError, Err.what());
 
-				return {.Error{.ErrorCode = -1, .Reason = Err.what()}};
+				return {.Status = {.Error{.ErrorCode = -1, .Reason = Err.what()}}};
 			}
 		}
 
@@ -1055,7 +1061,8 @@ namespace detail {
 										.RawHash		= RawHash,
 										.RawSize		= RawSize,
 										.Value			= std::move(Payload),
-										.ElapsedSeconds = Result.ElapsedSeconds});
+										.ElapsedSeconds = Result.ElapsedSeconds,
+										.Source			= &m_Info});
 						}
 
 						return {.Bytes = Result.Bytes, .ElapsedSeconds = Result.ElapsedSeconds, .Success = true};
@@ -1185,7 +1192,8 @@ namespace detail {
 										.RawHash		= RawHash,
 										.RawSize		= RawSize,
 										.Value			= std::move(Payload),
-										.ElapsedSeconds = Result.ElapsedSeconds});
+										.ElapsedSeconds = Result.ElapsedSeconds,
+										.Source			= &m_Info});
 						}
 
 						return {.Bytes = Result.Bytes, .ElapsedSeconds = Result.ElapsedSeconds, .Success = true};
@@ -1480,7 +1488,7 @@ public:
 		}
 	}
 
-	virtual GetUpstreamCacheResult GetCacheRecord(std::string_view Namespace, const CacheKey& CacheKey, ZenContentType Type) override
+	virtual GetUpstreamCacheSingleResult GetCacheRecord(std::string_view Namespace, const CacheKey& CacheKey, ZenContentType Type) override
 	{
 		ZEN_TRACE_CPU("Upstream::GetCacheRecord");
 
@@ -1495,31 +1503,29 @@ public:
 					continue;
 				}
 
-				UpstreamEndpointStats& Stats = Endpoint->Stats();
-				GetUpstreamCacheResult Result;
-				{
-					metrics::OperationTiming::Scope Scope(Stats.CacheGetRequestTiming);
-					Result = Endpoint->GetCacheRecord(Namespace, CacheKey, Type);
-				}
+				UpstreamEndpointStats&			Stats = Endpoint->Stats();
+				metrics::OperationTiming::Scope Scope(Stats.CacheGetRequestTiming);
+				GetUpstreamCacheSingleResult	Result = Endpoint->GetCacheRecord(Namespace, CacheKey, Type);
+				Scope.Stop();
 
 				Stats.CacheGetCount.Increment(1);
-				Stats.CacheGetTotalBytes.Increment(Result.Bytes);
+				Stats.CacheGetTotalBytes.Increment(Result.Status.Bytes);
 
-				if (Result.Success)
+				if (Result.Status.Success)
 				{
 					Stats.CacheHitCount.Increment(1);
 
 					return Result;
 				}
 
-				if (Result.Error)
+				if (Result.Status.Error)
 				{
 					Stats.CacheErrorCount.Increment(1);
 
 					ZEN_WARN("get cache record FAILED, endpoint '{}', reason '{}', error code '{}'",
 							 Endpoint->GetEndpointInfo().Url,
-							 Result.Error.Reason,
-							 Result.Error.ErrorCode);
+							 Result.Status.Error.Reason,
+							 Result.Status.Error.ErrorCode);
 				}
 			}
 		}
@@ -1588,6 +1594,7 @@ public:
 			}
 		}
 
+		const UpstreamEndpointInfo Info;
 		for (CacheKeyRequest* Request : RemainingKeys)
 		{
 			OnComplete({.Request = *Request, .Record = CbObjectView(), .Package = CbPackage()});
@@ -1655,15 +1662,16 @@ public:
 			}
 		}
 
+		const UpstreamEndpointInfo Info;
 		for (CacheChunkRequest* RequestPtr : RemainingKeys)
 		{
 			OnComplete({.Request = *RequestPtr, .RawHash = IoHash::Zero, .RawSize = 0, .Value = IoBuffer()});
 		}
 	}
 
-	virtual GetUpstreamCacheResult GetCacheChunk(std::string_view Namespace,
-												 const CacheKey&  CacheKey,
-												 const IoHash&	  ValueContentId) override
+	virtual GetUpstreamCacheSingleResult GetCacheChunk(std::string_view Namespace,
+													   const CacheKey&	CacheKey,
+													   const IoHash&	ValueContentId) override
 	{
 		ZEN_TRACE_CPU("Upstream::GetCacheChunk");
 
@@ -1676,32 +1684,29 @@ public:
 					continue;
 				}
 
-				UpstreamEndpointStats& Stats = Endpoint->Stats();
-				GetUpstreamCacheResult Result;
-
-				{
-					metrics::OperationTiming::Scope Scope(Stats.CacheGetRequestTiming);
-					Result = Endpoint->GetCacheChunk(Namespace, CacheKey, ValueContentId);
-				}
+				UpstreamEndpointStats&			Stats = Endpoint->Stats();
+				metrics::OperationTiming::Scope Scope(Stats.CacheGetRequestTiming);
+				GetUpstreamCacheSingleResult	Result = Endpoint->GetCacheChunk(Namespace, CacheKey, ValueContentId);
+				Scope.Stop();
 
 				Stats.CacheGetCount.Increment(1);
-				Stats.CacheGetTotalBytes.Increment(Result.Bytes);
+				Stats.CacheGetTotalBytes.Increment(Result.Status.Bytes);
 
-				if (Result.Success)
+				if (Result.Status.Success)
 				{
 					Stats.CacheHitCount.Increment(1);
 
 					return Result;
 				}
 
-				if (Result.Error)
+				if (Result.Status.Error)
 				{
 					Stats.CacheErrorCount.Increment(1);
 
 					ZEN_WARN("get cache chunk FAILED, endpoint '{}', reason '{}', error code '{}'",
 							 Endpoint->GetEndpointInfo().Url,
-							 Result.Error.Reason,
-							 Result.Error.ErrorCode);
+							 Result.Status.Error.Reason,
+							 Result.Status.Error.ErrorCode);
 				}
 			}
 		}
@@ -1770,6 +1775,7 @@ public:
 			}
 		}
 
+		const UpstreamEndpointInfo Info;
 		for (CacheValueRequest* RequestPtr : RemainingKeys)
 		{
 			OnComplete({.Request = *RequestPtr, .RawHash = IoHash::Zero, .RawSize = 0, .Value = IoBuffer()});