Merged main

1 files changed, 163 insertions, 112 deletions

diff --git a/zenserver/cache/structuredcache.cpp b/zenserver/cache/structuredcache.cpp
index dc96aecae..4a2a3748a 100644
--- a/zenserver/cache/structuredcache.cpp
+++ b/zenserver/cache/structuredcache.cpp
@@ -12,6 +12,7 @@
 #include <zenhttp/httpserver.h>
 #include <zenstore/CAS.h>
 
+#include "monitoring/httpstats.h"
 #include "structuredcache.h"
 #include "structuredcachestore.h"
 #include "upstream/jupiter.h"
@@ -149,13 +150,19 @@ ParseCachePolicy(const HttpServerRequest::QueryParams& QueryParams)
 HttpStructuredCacheService::HttpStructuredCacheService(ZenCacheStore&				  InCacheStore,
 													   CasStore&					  InStore,
 													   CidStore&					  InCidStore,
+													   HttpStatsService&			  StatsService,
+													   HttpStatusService&			  StatusService,
 													   std::unique_ptr<UpstreamCache> UpstreamCache)
 : m_Log(logging::Get("cache"))
 , m_CacheStore(InCacheStore)
+, m_StatsService(StatsService)
+, m_StatusService(StatusService)
 , m_CasStore(InStore)
 , m_CidStore(InCidStore)
 , m_UpstreamCache(std::move(UpstreamCache))
 {
+	StatsService.RegisterHandler("z$", *this);
+	StatusService.RegisterHandler("z$", *this);
 }
 
 HttpStructuredCacheService::~HttpStructuredCacheService()
@@ -200,11 +207,6 @@ HttpStructuredCacheService::HandleRequest(HttpServerRequest& Request)
 	{
 		std::string_view Key = Request.RelativeUri();
 
-		if (Key.empty())
-		{
-			return HandleStatusRequest(Request);
-		}
-
 		if (std::all_of(begin(Key), end(Key), [](const char c) { return std::isalnum(c); }))
 		{
 			// Bucket reference
@@ -270,10 +272,6 @@ HttpStructuredCacheService::HandleCacheRecordRequest(HttpServerRequest& Request,
 		case kHead:
 		case kGet:
 			{
-				if (Verb == kHead)
-				{
-					Request.SetSuppressResponseBody();
-				}
 				HandleGetCacheRecord(Request, Ref, Policy);
 			}
 			break;
@@ -288,28 +286,104 @@ HttpStructuredCacheService::HandleCacheRecordRequest(HttpServerRequest& Request,
 void
 HttpStructuredCacheService::HandleGetCacheRecord(zen::HttpServerRequest& Request, const CacheRef& Ref, CachePolicy Policy)
 {
-	const ZenContentType AcceptType = Request.AcceptContentType();
+	const ZenContentType AcceptType		 = Request.AcceptContentType();
+	const bool			 SkipData		 = (Policy & CachePolicy::SkipData) == CachePolicy::SkipData;
+	const bool			 SkipAttachments = (Policy & CachePolicy::SkipAttachments) == CachePolicy::SkipAttachments;
+	const bool			 QueryUpstream	 = m_UpstreamCache && ((Policy & CachePolicy::QueryRemote) == CachePolicy::QueryRemote);
 
-	ZenCacheValue Value;
-	bool		  Success		  = m_CacheStore.Get(Ref.BucketSegment, Ref.HashKey, /* out */ Value);
-	bool		  InUpstreamCache = false;
+	bool		  Success = false;
+	ZenCacheValue LocalCacheValue;
 
-	const bool QueryUpstream = !Success && m_UpstreamCache && (CachePolicy::QueryRemote == (Policy & CachePolicy::QueryRemote));
+	if (m_CacheStore.Get(Ref.BucketSegment, Ref.HashKey, LocalCacheValue))
+	{
+		Success = true;
 
-	if (QueryUpstream)
+		if (!SkipData && AcceptType == ZenContentType::kCbPackage)
+		{
+			CbPackage	 Package;
+			CbObjectView CacheRecord(LocalCacheValue.Value.Data());
+			uint32_t	 AttachmentCount = 0;
+			uint32_t	 ValidCount		 = 0;
+
+			if (!SkipAttachments)
+			{
+				CacheRecord.IterateAttachments([this, &Ref, &Package, &AttachmentCount, &ValidCount](CbFieldView AttachmentHash) {
+					if (IoBuffer Chunk = m_CidStore.FindChunkByCid(AttachmentHash.AsHash()))
+					{
+						Package.AddAttachment(CbAttachment(CompressedBuffer::FromCompressed(SharedBuffer(Chunk))));
+						ValidCount++;
+					}
+					AttachmentCount++;
+				});
+
+				if (ValidCount != AttachmentCount)
+				{
+					Success = false;
+					ZEN_WARN("GET - '{}/{}' '{}' FAILED, found '{}' of '{}' attachments",
+							 Ref.BucketSegment,
+							 Ref.HashKey,
+							 ToString(AcceptType),
+							 ValidCount,
+							 AttachmentCount);
+				}
+			}
+
+			Package.SetObject(LoadCompactBinaryObject(LocalCacheValue.Value));
+
+			BinaryWriter MemStream;
+			Package.Save(MemStream);
+
+			LocalCacheValue.Value = IoBuffer(IoBuffer::Clone, MemStream.Data(), MemStream.Size());
+			LocalCacheValue.Value.SetContentType(HttpContentType::kCbPackage);
+		}
+	}
+
+	if (Success)
 	{
-		const ZenContentType CacheRecordType = AcceptType;
+		ZEN_DEBUG("HIT - '{}/{}' {} '{}' (LOCAL)",
+				  Ref.BucketSegment,
+				  Ref.HashKey,
+				  NiceBytes(LocalCacheValue.Value.Size()),
+				  ToString(LocalCacheValue.Value.GetContentType()));
+
+		m_CacheStats.HitCount++;
+
+		if (SkipData)
+		{
+			return Request.WriteResponse(HttpResponseCode::OK);
+		}
+		else
+		{
+			return Request.WriteResponse(HttpResponseCode::OK, LocalCacheValue.Value.GetContentType(), LocalCacheValue.Value);
+		}
+	}
+	else if (!QueryUpstream)
+	{
+		ZEN_DEBUG("MISS - '{}/{}' '{}'", Ref.BucketSegment, Ref.HashKey, ToString(AcceptType));
+		m_CacheStats.MissCount++;
+		return Request.WriteResponse(HttpResponseCode::NotFound);
+	}
+
+	// Issue upstream query asynchronously in order to keep requests flowing without
+	// hogging I/O servicing threads with blocking work
+
+	Request.WriteResponseAsync([this, AcceptType, SkipData, SkipAttachments, Ref](HttpServerRequest& AsyncRequest) {
+		bool		  Success = false;
+		ZenCacheValue UpstreamCacheValue;
 
-		if (auto UpstreamResult = m_UpstreamCache->GetCacheRecord({Ref.BucketSegment, Ref.HashKey}, CacheRecordType);
+		metrics::OperationTiming::Scope $(m_UpstreamGetRequestTiming);
+
+		if (GetUpstreamCacheResult UpstreamResult = m_UpstreamCache->GetCacheRecord({Ref.BucketSegment, Ref.HashKey}, AcceptType);
 			UpstreamResult.Success)
 		{
-			Value.Value		= UpstreamResult.Value;
-			Success			= true;
-			InUpstreamCache = true;
+			Success					 = true;
+			UpstreamCacheValue.Value = UpstreamResult.Value;
+
+			UpstreamCacheValue.Value.SetContentType(AcceptType);
 
-			if (CacheRecordType == ZenContentType::kBinary || CacheRecordType == ZenContentType::kCbObject)
+			if (AcceptType == ZenContentType::kBinary || AcceptType == ZenContentType::kCbObject)
 			{
-				if (CacheRecordType == ZenContentType::kCbObject)
+				if (AcceptType == ZenContentType::kCbObject)
 				{
 					const CbValidateError ValidationResult = ValidateCompactBinary(UpstreamResult.Value, CbValidateMode::All);
 
@@ -322,7 +396,7 @@ HttpStructuredCacheService::HandleGetCacheRecord(zen::HttpServerRequest& Request
 						CacheRecord.IterateAttachments([&](CbFieldView Attachment) { IndexData.AddHash(Attachment.AsHash()); });
 						IndexData.EndArray();
 
-						Value.IndexData = IndexData.Save();
+						UpstreamCacheValue.IndexData = IndexData.Save();
 					}
 					else
 					{
@@ -336,19 +410,17 @@ HttpStructuredCacheService::HandleGetCacheRecord(zen::HttpServerRequest& Request
 
 				if (Success)
 				{
-					m_CacheStore.Put(Ref.BucketSegment, Ref.HashKey, Value);
+					m_CacheStore.Put(Ref.BucketSegment, Ref.HashKey, UpstreamCacheValue);
 				}
 			}
-			else
+			else if (AcceptType == ZenContentType::kCbPackage)
 			{
-				ZEN_ASSERT(CacheRecordType == ZenContentType::kCbPackage);
-
 				CbPackage Package;
-				if (Package.TryLoad(UpstreamResult.Value))
+				if (Package.TryLoad(UpstreamCacheValue.Value))
 				{
+					CbObject CacheRecord	 = Package.GetObject();
 					uint32_t AttachmentCount = 0;
 					uint32_t ValidCount		 = 0;
-					CbObject CacheRecord	 = Package.GetObject();
 
 					CacheRecord.IterateAttachments([this, &Package, &Ref, &AttachmentCount, &ValidCount](CbFieldView AttachmentHash) {
 						if (const CbAttachment* Attachment = Package.FindAttachment(AttachmentHash.AsHash()))
@@ -373,7 +445,7 @@ HttpStructuredCacheService::HandleGetCacheRecord(zen::HttpServerRequest& Request
 					{
 						m_CacheStore.Put(Ref.BucketSegment, Ref.HashKey, {.Value = CacheRecord.GetBuffer().AsIoBuffer()});
 
-						if (zen::CachePolicy::SkipAttachments == (Policy & zen::CachePolicy::SkipAttachments))
+						if (SkipAttachments)
 						{
 							CbPackage PackageWithoutAttachments;
 							PackageWithoutAttachments.SetObject(CacheRecord);
@@ -381,7 +453,7 @@ HttpStructuredCacheService::HandleGetCacheRecord(zen::HttpServerRequest& Request
 							BinaryWriter MemStream;
 							PackageWithoutAttachments.Save(MemStream);
 
-							Value.Value = IoBuffer(IoBuffer::Clone, MemStream.Data(), MemStream.Size());
+							UpstreamCacheValue.Value = IoBuffer(IoBuffer::Clone, MemStream.Data(), MemStream.Size());
 						}
 					}
 					else
@@ -400,86 +472,34 @@ HttpStructuredCacheService::HandleGetCacheRecord(zen::HttpServerRequest& Request
 				}
 			}
 		}
-	}
-
-	if (!Success)
-	{
-		ZEN_DEBUG("MISS - '{}/{}' '{}'", Ref.BucketSegment, Ref.HashKey, ToString(AcceptType));
-		return Request.WriteResponse(HttpResponseCode::NotFound);
-	}
-
-	if (AcceptType == ZenContentType::kCbPackage && !InUpstreamCache)
-	{
-		CbObjectView CacheRecord(Value.Value.Data());
-
-		const CbValidateError ValidationResult = ValidateCompactBinary(Value.Value, CbValidateMode::All);
 
-		if (ValidationResult != CbValidateError::None)
+		if (Success)
 		{
-			ZEN_WARN("GET - '{}/{}' '{}' FAILED, invalid compact binary object", Ref.BucketSegment, Ref.HashKey, ToString(AcceptType));
-			return Request.WriteResponse(HttpResponseCode::NotFound, HttpContentType::kText, "Invalid cache record"sv);
-		}
-
-		const bool SkipAttachments = zen::CachePolicy::SkipAttachments == (Policy & zen::CachePolicy::SkipAttachments);
-		uint32_t   AttachmentCount = 0;
-		uint32_t   ValidCount	   = 0;
-		uint64_t   AttachmentBytes = 0ull;
-
-		CbPackage Package;
+			ZEN_DEBUG("HIT - '{}/{}' {} '{}' (UPSTREAM)",
+					  Ref.BucketSegment,
+					  Ref.HashKey,
+					  NiceBytes(UpstreamCacheValue.Value.Size()),
+					  ToString(UpstreamCacheValue.Value.GetContentType()));
 
-		if (!SkipAttachments)
-		{
-			CacheRecord.IterateAttachments(
-				[this, &Ref, &Package, &AttachmentCount, &ValidCount, &AttachmentBytes](CbFieldView AttachmentHash) {
-					if (IoBuffer Chunk = m_CidStore.FindChunkByCid(AttachmentHash.AsHash()))
-					{
-						Package.AddAttachment(CbAttachment(CompressedBuffer::FromCompressed(SharedBuffer(Chunk))));
-						AttachmentBytes += Chunk.Size();
-						ValidCount++;
-					}
-					AttachmentCount++;
-				});
+			m_CacheStats.HitCount++;
+			m_CacheStats.UpstreamHitCount++;
 
-			if (ValidCount != AttachmentCount)
+			if (SkipData)
 			{
-				ZEN_WARN("GET - '{}/{}' '{}' FAILED, found '{}' of '{}' attachments",
-						 Ref.BucketSegment,
-						 Ref.HashKey,
-						 ToString(AcceptType),
-						 ValidCount,
-						 AttachmentCount);
-
-				return Request.WriteResponse(HttpResponseCode::NotFound, HttpContentType::kText, "Missing attachments"sv);
+				AsyncRequest.WriteResponse(HttpResponseCode::OK);
+			}
+			else
+			{
+				AsyncRequest.WriteResponse(HttpResponseCode::OK, UpstreamCacheValue.Value.GetContentType(), UpstreamCacheValue.Value);
 			}
 		}
-
-		Package.SetObject(LoadCompactBinaryObject(Value.Value));
-
-		ZEN_DEBUG("HIT - '{}/{}' {} '{}', {} attachments (LOCAL)",
-				  Ref.BucketSegment,
-				  Ref.HashKey,
-				  NiceBytes(AttachmentBytes + Value.Value.Size()),
-				  ToString(HttpContentType::kCbPackage),
-				  AttachmentCount);
-
-		BinaryWriter MemStream;
-		Package.Save(MemStream);
-
-		IoBuffer Response(IoBuffer::Clone, MemStream.Data(), MemStream.Size());
-
-		Request.WriteResponse(HttpResponseCode::OK, HttpContentType::kCbPackage, Response);
-	}
-	else
-	{
-		ZEN_DEBUG("HIT - '{}/{}' {} '{}' ({})",
-				  Ref.BucketSegment,
-				  Ref.HashKey,
-				  NiceBytes(Value.Value.Size()),
-				  ToString(Value.Value.GetContentType()),
-				  InUpstreamCache ? "UPSTREAM" : "LOCAL");
-
-		Request.WriteResponse(HttpResponseCode::OK, Value.Value.GetContentType(), Value.Value);
-	}
+		else
+		{
+			ZEN_DEBUG("MISS - '{}/{}' '{}'", Ref.BucketSegment, Ref.HashKey, ToString(AcceptType));
+			m_CacheStats.MissCount++;
+			AsyncRequest.WriteResponse(HttpResponseCode::NotFound);
+		}
+	});
 }
 
 void
@@ -668,10 +688,6 @@ HttpStructuredCacheService::HandleCachePayloadRequest(HttpServerRequest& Request
 		case kGet:
 			{
 				HandleGetCachePayload(Request, Ref, Policy);
-				if (Verb == kHead)
-				{
-					Request.SetSuppressResponseBody();
-				}
 			}
 			break;
 		case kPut:
@@ -712,7 +728,8 @@ HttpStructuredCacheService::HandleGetCachePayload(zen::HttpServerRequest& Reques
 
 	if (!Payload)
 	{
-		ZEN_DEBUG("MISS - '{}/{}/{}'", Ref.BucketSegment, Ref.HashKey, Ref.PayloadId);
+		ZEN_DEBUG("MISS - '{}/{}/{}' '{}'", Ref.BucketSegment, Ref.HashKey, Ref.PayloadId, ToString(Request.AcceptContentType()));
+		m_CacheStats.MissCount++;
 		return Request.WriteResponse(HttpResponseCode::NotFound);
 	}
 
@@ -724,7 +741,20 @@ HttpStructuredCacheService::HandleGetCachePayload(zen::HttpServerRequest& Reques
 			  ToString(Payload.GetContentType()),
 			  InUpstreamCache ? "UPSTREAM" : "LOCAL");
 
-	Request.WriteResponse(HttpResponseCode::OK, HttpContentType::kBinary, Payload);
+	m_CacheStats.HitCount++;
+	if (InUpstreamCache)
+	{
+		m_CacheStats.UpstreamHitCount++;
+	}
+
+	if ((Policy & CachePolicy::SkipData) == CachePolicy::SkipData)
+	{
+		Request.WriteResponse(HttpResponseCode::OK);
+	}
+	else
+	{
+		Request.WriteResponse(HttpResponseCode::OK, HttpContentType::kBinary, Payload);
+	}
 }
 
 void
@@ -839,12 +869,25 @@ HttpStructuredCacheService::ValidateKeyUri(HttpServerRequest& Request, CacheRef&
 }
 
 void
-HttpStructuredCacheService::HandleStatusRequest(zen::HttpServerRequest& Request)
+HttpStructuredCacheService::HandleStatsRequest(zen::HttpServerRequest& Request)
 {
 	CbObjectWriter Cbo;
-	Cbo << "ok" << true;
 
 	EmitSnapshot("requests", m_HttpRequests, Cbo);
+	EmitSnapshot("upstream_gets", m_UpstreamGetRequestTiming, Cbo);
+
+	const uint64_t HitCount			= m_CacheStats.HitCount;
+	const uint64_t UpstreamHitCount = m_CacheStats.UpstreamHitCount;
+	const uint64_t MissCount		= m_CacheStats.MissCount;
+	const uint64_t TotalCount		= HitCount + MissCount;
+
+	Cbo.BeginObject("cache");
+	Cbo << "hits" << HitCount << "misses" << MissCount;
+	Cbo << "hit_ratio" << (TotalCount > 0 ? (double(HitCount) / double(TotalCount) * 100.0) : 0.0);
+	Cbo << "upstream_hits" << m_CacheStats.UpstreamHitCount;
+	Cbo << "upstream_ratio" << (HitCount > 0 ? (double(UpstreamHitCount) / double(HitCount)) * 100.0 : 0.0);
+	Cbo.EndObject();
+
 	if (m_UpstreamCache)
 	{
 		Cbo.BeginObject("upstream");
@@ -855,4 +898,12 @@ HttpStructuredCacheService::HandleStatusRequest(zen::HttpServerRequest& Request)
 	Request.WriteResponse(HttpResponseCode::OK, Cbo.Save());
 }
 
+void
+HttpStructuredCacheService::HandleStatusRequest(zen::HttpServerRequest& Request)
+{
+	CbObjectWriter Cbo;
+	Cbo << "ok" << true;
+	Request.WriteResponse(HttpResponseCode::OK, Cbo.Save());
+}
+
 }  // namespace zen