From 892bbaaa5c1f62631070cc74820f349c4c80f55d Mon Sep 17 00:00:00 2001 From: vnugent Date: Fri, 27 Jan 2023 21:13:16 -0500 Subject: Object cache overhaul and logger updates --- .../src/Endpoints/BrokerHeartBeat.cs | 4 +- .../src/Endpoints/CacheConfiguration.cs | 54 +++++ .../src/Endpoints/ConnectEndpoint.cs | 220 ++++++++++++++++----- .../ObjectCacheServer/src/Endpoints/ICacheStore.cs | 56 ++++++ 4 files changed, 282 insertions(+), 52 deletions(-) create mode 100644 plugins/ObjectCacheServer/src/Endpoints/CacheConfiguration.cs create mode 100644 plugins/ObjectCacheServer/src/Endpoints/ICacheStore.cs (limited to 'plugins/ObjectCacheServer/src/Endpoints') diff --git a/plugins/ObjectCacheServer/src/Endpoints/BrokerHeartBeat.cs b/plugins/ObjectCacheServer/src/Endpoints/BrokerHeartBeat.cs index bd1233e..3930f90 100644 --- a/plugins/ObjectCacheServer/src/Endpoints/BrokerHeartBeat.cs +++ b/plugins/ObjectCacheServer/src/Endpoints/BrokerHeartBeat.cs @@ -30,12 +30,14 @@ using System.Threading; using System.Threading.Tasks; using System.Collections.Generic; +using VNLib.Plugins; +using VNLib.Plugins.Essentials; using VNLib.Hashing.IdentityUtility; using VNLib.Plugins.Essentials.Endpoints; using VNLib.Plugins.Essentials.Extensions; using VNLib.Plugins.Extensions.Loading; -namespace VNLib.Plugins.Essentials.Sessions.Server.Endpoints +namespace VNLib.Data.Caching.ObjectCache.Server { internal sealed class BrokerHeartBeat : ResourceEndpointBase { diff --git a/plugins/ObjectCacheServer/src/Endpoints/CacheConfiguration.cs b/plugins/ObjectCacheServer/src/Endpoints/CacheConfiguration.cs new file mode 100644 index 0000000..e9584b6 --- /dev/null +++ b/plugins/ObjectCacheServer/src/Endpoints/CacheConfiguration.cs @@ -0,0 +1,54 @@ +/* +* Copyright (c) 2022 Vaughn Nugent +* +* Library: VNLib +* Package: ObjectCacheServer +* File: ConnectEndpoint.cs +* +* ConnectEndpoint.cs is part of ObjectCacheServer which is part of the larger +* VNLib collection of libraries and utilities. +* +* ObjectCacheServer is free software: you can redistribute it and/or modify +* it under the terms of the GNU Affero General Public License as +* published by the Free Software Foundation, either version 3 of the +* License, or (at your option) any later version. +* +* ObjectCacheServer is distributed in the hope that it will be useful, +* but WITHOUT ANY WARRANTY; without even the implied warranty of +* MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the +* GNU Affero General Public License for more details. +* +* You should have received a copy of the GNU Affero General Public License +* along with this program. If not, see https://www.gnu.org/licenses/. +*/ + +using System.Text.Json.Serialization; + +namespace VNLib.Data.Caching.ObjectCache.Server +{ + internal sealed class CacheConfiguration + { + [JsonPropertyName("buffer_recv_max")] + public int MaxRecvBufferSize { get; set; } = 1000 * 1024; + [JsonPropertyName("buffer_recv_min")] + public int MinRecvBufferSize { get; set; } = 8 * 1024; + + + [JsonPropertyName("buffer_header_max")] + public int MaxHeaderBufferSize { get; set; } = 2 * 1024; + [JsonPropertyName("buffer_header_min")] + public int MinHeaderBufferSize { get; set; } = 128; + + + [JsonPropertyName("max_message_size")] + public int MaxMessageSize { get; set; } = 1000 * 1024; + + + [JsonPropertyName("change_queue_max_depth")] + public int MaxEventQueueDepth { get; set; } = 10 * 1000; + + + [JsonPropertyName("max_cache")] + public int MaxCacheEntries { get; set; } = 10000; + } +} diff --git a/plugins/ObjectCacheServer/src/Endpoints/ConnectEndpoint.cs b/plugins/ObjectCacheServer/src/Endpoints/ConnectEndpoint.cs index 15cc086..9a1ece0 100644 --- a/plugins/ObjectCacheServer/src/Endpoints/ConnectEndpoint.cs +++ b/plugins/ObjectCacheServer/src/Endpoints/ConnectEndpoint.cs @@ -31,34 +31,29 @@ using System.Threading.Channels; using System.Collections.Generic; using System.Collections.Concurrent; -using VNLib.Net.Http; +using VNLib.Plugins; using VNLib.Hashing; +using VNLib.Net.Http; using VNLib.Utils.Async; +using VNLib.Utils.Memory; using VNLib.Utils.Logging; using VNLib.Hashing.IdentityUtility; using VNLib.Net.Messaging.FBM; using VNLib.Net.Messaging.FBM.Client; using VNLib.Net.Messaging.FBM.Server; -using VNLib.Data.Caching.ObjectCache; +using VNLib.Plugins.Essentials; using VNLib.Plugins.Extensions.Loading; using VNLib.Plugins.Essentials.Endpoints; using VNLib.Plugins.Essentials.Extensions; +using System.Text.Json.Serialization; - -namespace VNLib.Plugins.Essentials.Sessions.Server.Endpoints +namespace VNLib.Data.Caching.ObjectCache.Server { - internal sealed class ConnectEndpoint : ResourceEndpointBase - { - const int MAX_RECV_BUF_SIZE = 1000 * 1024; - const int MIN_RECV_BUF_SIZE = 8 * 1024; - const int MAX_HEAD_BUF_SIZE = 2048; - const int MIN_MESSAGE_SIZE = 10 * 1024; - const int MAX_MESSAGE_SIZE = 1000 * 1024; - const int MIN_HEAD_BUF_SIZE = 128; - const int MAX_EVENT_QUEUE_SIZE = 10000; - const int MAX_RESPONSE_BUFFER_SIZE = 10 * 1024; - private static readonly TimeSpan AuthTokenExpiration = TimeSpan.FromSeconds(30); + [ConfigurationName("store")] + internal sealed class ConnectEndpoint : ResourceEndpointBase, IDisposable, IAsyncBackgroundWork + { + private static readonly TimeSpan AuthTokenExpiration = TimeSpan.FromSeconds(30); private readonly string AudienceLocalServerId; private readonly ObjectCacheStore Store; @@ -68,8 +63,16 @@ namespace VNLib.Plugins.Essentials.Sessions.Server.Endpoints private uint _connectedClients; + /// + /// Gets the number of active connections + /// public uint ConnectedClients => _connectedClients; + /// + /// The cache store configuration + /// + public CacheConfiguration CacheConfig { get; } + //Loosen up protection settings protected override ProtectionSettings EndpointProtectionSettings { get; } = new() { @@ -78,20 +81,89 @@ namespace VNLib.Plugins.Essentials.Sessions.Server.Endpoints DisableCrossSiteDenied = true }; - public ConnectEndpoint(string path, ObjectCacheStore store, PluginBase pbase) + public ConnectEndpoint(PluginBase plugin, IReadOnlyDictionary config) { - InitPathAndLog(path, pbase.Log); - Store = store;//Load client public key to verify signed messages - Pbase = pbase; + string? path = config["path"].GetString(); + InitPathAndLog(path, plugin.Log); + + Pbase = plugin; + + //Parse cache config or use default + if(config.TryGetValue("cache", out JsonElement confEl)) + { + CacheConfig = confEl.Deserialize()!; + } + else + { + //Init default config if not fount + CacheConfig = new(); + + Log.Verbose("Loading default cache buffer configuration"); + } + + //Create event queue client lookup table StatefulEventQueue = new(StringComparer.OrdinalIgnoreCase); - //Start the queue worker - _ = pbase.DeferTask(() => ChangeWorkerAsync(pbase.UnloadToken), 10); + //Init the cache store + Store = InitializeCache((ObjectCacheServerEntry)plugin, CacheConfig.MaxCacheEntries); + /* + * Generate a random guid for the current server when created so we + * know client tokens belong to us when singed by the same key + */ AudienceLocalServerId = Guid.NewGuid().ToString("N"); + + //Schedule the queue worker to be run + _ = plugin.ObserveWork(this, 100); + } + + private static ObjectCacheStore InitializeCache(ObjectCacheServerEntry plugin, int maxCache) + { + if(maxCache < 2) + { + throw new ArgumentException("You must configure a 'max_cache' size larger than 1 item"); + } + + //Suggestion + if(maxCache < 200) + { + plugin.Log.Information("Suggestion: You may want a larger cache size, you have less than 200 items in cache"); + } + + //Endpoint only allows for a single reader + return new (maxCache, plugin.Log, plugin.CacheHeap, true); } + /// + /// Gets the configured cache store + /// + /// + public ICacheStore GetCacheStore() => new CacheStore(Store); + + + //Dispose will be called by the host plugin on unload + void IDisposable.Dispose() + { + //Dispose the store on cleanup + Store.Dispose(); + } + + + private async Task GetClientPubAsync() + { + return await Pbase.TryGetSecretAsync("client_public_key").ToJsonWebKey() ?? throw new KeyNotFoundException("Missing required secret : client_public_key"); + } + private async Task GetCachePubAsync() + { + return await Pbase.TryGetSecretAsync("cache_public_key").ToJsonWebKey() ?? throw new KeyNotFoundException("Missing required secret : client_public_key"); + } + private async Task GetCachePrivateKeyAsync() + { + return await Pbase.TryGetSecretAsync("cache_private_key").ToJsonWebKey() ?? throw new KeyNotFoundException("Missing required secret : client_public_key"); + } + + /* * Used as a client negotiation and verification request * @@ -132,7 +204,7 @@ namespace VNLib.Plugins.Essentials.Sessions.Server.Endpoints { verified = true; } - //May be signed by a cahce server + //May be signed by a cache server else { using ReadOnlyJsonWebKey cacheCert = await GetCachePubAsync(); @@ -163,8 +235,10 @@ namespace VNLib.Plugins.Essentials.Sessions.Server.Endpoints } Log.Debug("Received negotiation request from node {node}", nodeId); + //Verified, now we can create an auth message with a short expiration using JsonWebToken auth = new(); + //Sign the auth message from the cache certificate's private key using (ReadOnlyJsonWebKey cert = await GetCachePrivateKeyAsync()) { @@ -179,9 +253,9 @@ namespace VNLib.Plugins.Essentials.Sessions.Server.Endpoints //Specify the server's node id if set .AddClaim("sub", nodeId!) //Add negotiaion args - .AddClaim(FBMClient.REQ_HEAD_BUF_QUERY_ARG, MAX_HEAD_BUF_SIZE) - .AddClaim(FBMClient.REQ_RECV_BUF_QUERY_ARG, MAX_RECV_BUF_SIZE) - .AddClaim(FBMClient.REQ_MAX_MESS_QUERY_ARG, MAX_MESSAGE_SIZE) + .AddClaim(FBMClient.REQ_HEAD_BUF_QUERY_ARG, CacheConfig.MaxHeaderBufferSize) + .AddClaim(FBMClient.REQ_RECV_BUF_QUERY_ARG, CacheConfig.MaxRecvBufferSize) + .AddClaim(FBMClient.REQ_MAX_MESS_QUERY_ARG, CacheConfig.MaxMessageSize) .CommitClaims(); auth.SignFromJwk(cert); @@ -192,27 +266,17 @@ namespace VNLib.Plugins.Essentials.Sessions.Server.Endpoints return VfReturnType.VirtualSkip; } - private async Task GetClientPubAsync() - { - return await Pbase.TryGetSecretAsync("client_public_key").ToJsonWebKey() ?? throw new KeyNotFoundException("Missing required secret : client_public_key"); - } - private async Task GetCachePubAsync() - { - return await Pbase.TryGetSecretAsync("cache_public_key").ToJsonWebKey() ?? throw new KeyNotFoundException("Missing required secret : client_public_key"); - } - private async Task GetCachePrivateKeyAsync() - { - return await Pbase.TryGetSecretAsync("cache_private_key").ToJsonWebKey() ?? throw new KeyNotFoundException("Missing required secret : client_public_key"); - } - private async Task ChangeWorkerAsync(CancellationToken cancellation) + //Background worker to process event queue items + async Task IAsyncBackgroundWork.DoWorkAsync(ILogProvider pluginLog, CancellationToken exitToken) { try { //Listen for changes while (true) { - ChangeEvent ev = await Store.EventQueue.DequeueAsync(cancellation); + ChangeEvent ev = await Store.EventQueue.DequeueAsync(exitToken); + //Add event to queues foreach (AsyncQueue queue in StatefulEventQueue.Values) { @@ -224,10 +288,8 @@ namespace VNLib.Plugins.Essentials.Sessions.Server.Endpoints } } catch (OperationCanceledException) - { } - catch (Exception ex) { - Log.Error(ex); + //Normal exit } } @@ -238,6 +300,12 @@ namespace VNLib.Plugins.Essentials.Sessions.Server.Endpoints public int MaxMessageSize { get; init; } public int MaxResponseBufferSize { get; init; } public AsyncQueue? SyncQueue { get; init; } + + public override string ToString() + { + return + $"{nameof(RecvBufferSize)}:{RecvBufferSize}, {nameof(MaxHeaderBufferSize)}: {MaxHeaderBufferSize}, {nameof(MaxMessageSize)}:{MaxMessageSize}, {nameof(MaxResponseBufferSize)}:{MaxResponseBufferSize}"; + } } protected override async ValueTask WebsocketRequestedAsync(HttpEntity entity) @@ -246,6 +314,7 @@ namespace VNLib.Plugins.Essentials.Sessions.Server.Endpoints { //Parse jwt from authorization string? jwtAuth = entity.Server.Headers[HttpRequestHeader.Authorization]; + if (string.IsNullOrWhiteSpace(jwtAuth)) { entity.CloseResponse(HttpStatusCode.Unauthorized); @@ -253,6 +322,7 @@ namespace VNLib.Plugins.Essentials.Sessions.Server.Endpoints } string? nodeId = null; + //Parse jwt using (JsonWebToken jwt = JsonWebToken.Parse(jwtAuth)) { @@ -301,11 +371,12 @@ namespace VNLib.Plugins.Essentials.Sessions.Server.Endpoints string maxMessageSizeCmd = entity.QueryArgs[FBMClient.REQ_MAX_MESS_QUERY_ARG]; //Parse recv buffer size - int recvBufSize = int.TryParse(recvBufCmd, out int rbs) ? rbs : MIN_RECV_BUF_SIZE; - int maxHeadBufSize = int.TryParse(maxHeaderCharCmd, out int hbs) ? hbs : MIN_HEAD_BUF_SIZE; - int maxMessageSize = int.TryParse(maxMessageSizeCmd, out int mxs) ? mxs : MIN_MESSAGE_SIZE; + int recvBufSize = int.TryParse(recvBufCmd, out int rbs) ? rbs : CacheConfig.MinRecvBufferSize; + int maxHeadBufSize = int.TryParse(maxHeaderCharCmd, out int hbs) ? hbs : CacheConfig.MinHeaderBufferSize; + int maxMessageSize = int.TryParse(maxMessageSizeCmd, out int mxs) ? mxs : CacheConfig.MaxMessageSize; AsyncQueue? nodeQueue = null; + //The connection may be a caching server node, so get its node-id if (!string.IsNullOrWhiteSpace(nodeId)) { @@ -317,7 +388,7 @@ namespace VNLib.Plugins.Essentials.Sessions.Server.Endpoints * and change events may be processed on mutliple threads. */ - BoundedChannelOptions queueOptions = new(MAX_EVENT_QUEUE_SIZE) + BoundedChannelOptions queueOptions = new(CacheConfig.MaxEventQueueDepth) { AllowSynchronousContinuations = true, SingleReader = false, @@ -327,21 +398,41 @@ namespace VNLib.Plugins.Essentials.Sessions.Server.Endpoints }; _ = StatefulEventQueue.TryAdd(nodeId, new(queueOptions)); + //Get the queue nodeQueue = StatefulEventQueue[nodeId]; } - + + /* + * Buffer sizing can get messy as the response/resquest sizes can vary + * and will include headers, this is a drawback of the FBM protocol + * so we need to properly calculate efficient buffer sizes as + * negotiated with the client. + */ + + int maxMessageSizeClamp = Math.Clamp(maxMessageSize, CacheConfig.MinRecvBufferSize, CacheConfig.MaxRecvBufferSize); + //Init new ws state object and clamp the suggested buffer sizes WsUserState state = new() { - RecvBufferSize = Math.Clamp(recvBufSize, MIN_RECV_BUF_SIZE, MAX_RECV_BUF_SIZE), - MaxHeaderBufferSize = Math.Clamp(maxHeadBufSize, MIN_HEAD_BUF_SIZE, MAX_HEAD_BUF_SIZE), - MaxMessageSize = Math.Clamp(maxMessageSize, MIN_MESSAGE_SIZE, MAX_MESSAGE_SIZE), - MaxResponseBufferSize = Math.Min(maxMessageSize, MAX_RESPONSE_BUFFER_SIZE), + RecvBufferSize = Math.Clamp(recvBufSize, CacheConfig.MinRecvBufferSize, CacheConfig.MaxRecvBufferSize), + MaxHeaderBufferSize = Math.Clamp(maxHeadBufSize, CacheConfig.MinHeaderBufferSize, CacheConfig.MaxHeaderBufferSize), + + MaxMessageSize = maxMessageSizeClamp, + + /* + * Response buffer needs to be large enough to store a max message + * as a response along with all response headers + */ + MaxResponseBufferSize = (int)MemoryUtil.NearestPage(maxMessageSizeClamp), + SyncQueue = nodeQueue }; Log.Debug("Client recv buffer suggestion {recv}, header buffer size {head}, response buffer size {r}", recvBufCmd, maxHeaderCharCmd, state.MaxResponseBufferSize); + + //Print state message to console + Log.Verbose("Client buffer state {state}", state); //Accept socket and pass state object entity.AcceptWebSocket(WebsocketAcceptedAsync, state); @@ -370,6 +461,7 @@ namespace VNLib.Plugins.Essentials.Sessions.Server.Endpoints RecvBufferSize = state.RecvBufferSize, ResponseBufferSize = state.MaxResponseBufferSize, MaxHeaderBufferSize = state.MaxHeaderBufferSize, + HeaderEncoding = Helpers.DefaultEncoding, }; @@ -395,5 +487,31 @@ namespace VNLib.Plugins.Essentials.Sessions.Server.Endpoints } Log.Debug("Server websocket exited"); } + + + private sealed class CacheStore : ICacheStore + { + private readonly ObjectCacheStore _cache; + + public CacheStore(ObjectCacheStore cache) + { + _cache = cache; + } + + ValueTask ICacheStore.AddOrUpdateBlobAsync(string objectId, string? alternateId, GetBodyDataCallback bodyData, T state, CancellationToken token) + { + return _cache.AddOrUpdateBlobAsync(objectId, alternateId, bodyData, state, token); + } + + void ICacheStore.Clear() + { + throw new NotImplementedException(); + } + + ValueTask ICacheStore.DeleteItemAsync(string id, CancellationToken token) + { + return _cache.DeleteItemAsync(id, token); + } + } } } diff --git a/plugins/ObjectCacheServer/src/Endpoints/ICacheStore.cs b/plugins/ObjectCacheServer/src/Endpoints/ICacheStore.cs new file mode 100644 index 0000000..3776269 --- /dev/null +++ b/plugins/ObjectCacheServer/src/Endpoints/ICacheStore.cs @@ -0,0 +1,56 @@ +/* +* Copyright (c) 2022 Vaughn Nugent +* +* Library: VNLib +* Package: ObjectCacheServer +* File: ConnectEndpoint.cs +* +* ConnectEndpoint.cs is part of ObjectCacheServer which is part of the larger +* VNLib collection of libraries and utilities. +* +* ObjectCacheServer is free software: you can redistribute it and/or modify +* it under the terms of the GNU Affero General Public License as +* published by the Free Software Foundation, either version 3 of the +* License, or (at your option) any later version. +* +* ObjectCacheServer is distributed in the hope that it will be useful, +* but WITHOUT ANY WARRANTY; without even the implied warranty of +* MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the +* GNU Affero General Public License for more details. +* +* You should have received a copy of the GNU Affero General Public License +* along with this program. If not, see https://www.gnu.org/licenses/. +*/ + +using System.Threading; +using System.Threading.Tasks; + +namespace VNLib.Data.Caching.ObjectCache.Server +{ + internal interface ICacheStore + { + /// + /// Asynchronously adds or updates an object in the store and optionally update's its id + /// + /// The current (or old) id of the object + /// An optional id to update the blob to + /// A callback that returns the data for the blob + /// The state parameter to pass to the data callback + /// A token to cancel the async operation + /// A value task that represents the async operation + ValueTask AddOrUpdateBlobAsync(string objectId, string? alternateId, GetBodyDataCallback bodyData, T state, CancellationToken token = default); + + /// + /// Clears all items from the store + /// + void Clear(); + + /// + /// Asynchronously deletes a previously stored item + /// + /// The id of the object to delete + /// A token to cancel the async lock await + /// A task that completes when the item has been deleted + ValueTask DeleteItemAsync(string id, CancellationToken token = default); + } +} -- cgit