From 44bb0cd1d56fb7369d72eccae8fd671d8930cc2d Mon Sep 17 00:00:00 2001 From: Corentin Altepe Date: Sat, 12 Mar 2022 11:04:47 +0100 Subject: [PATCH 1/9] Implemented IDisposable for LokiHttpClient, HttpLokiTransport and LokiTarget. Initialized LokiTarget in its constructor to allow for a smooth Dispose(). Updated unit tests to properly dispose target and dependencies. Changed style of internal fields to a camel case with _ prefix. Applied file scoped namespaces in some of the files. --- .editorconfig | 5 +- src/Benchmark/Transport.cs | 14 +-- src/NLog.Loki.Tests/LokiTargetTests.cs | 161 +++++++++++-------------- src/NLog.Loki/HttpLokiTransport.cs | 101 ++++++++-------- src/NLog.Loki/ILokiHttpClient.cs | 3 +- src/NLog.Loki/ILokiTransport.cs | 12 +- src/NLog.Loki/LokiHttpClient.cs | 26 ++-- src/NLog.Loki/LokiTarget.cs | 41 ++++--- src/NLog.Loki/NullLokiTransport.cs | 4 + 9 files changed, 183 insertions(+), 184 deletions(-) diff --git a/.editorconfig b/.editorconfig index 82911d0..f25e2f1 100644 --- a/.editorconfig +++ b/.editorconfig @@ -258,6 +258,9 @@ dotnet_naming_style.prefix_interface_with_i_style.required_prefix = I # prefix_type_parameters_with_t_style - Generic Type Parameters must be PascalCase and the first character must be a 'T' dotnet_naming_style.prefix_type_parameters_with_t_style.capitalization = pascal_case dotnet_naming_style.prefix_type_parameters_with_t_style.required_prefix = T +# prefix_underscore_style - Variable name starts with _ +dotnet_naming_style.prefix_underscore_style.capitalization = camel_case +dotnet_naming_style.prefix_underscore_style.required_prefix = _ # disallowed_style - Anything that has this style applied is marked as disallowed dotnet_naming_style.disallowed_style.capitalization = pascal_case dotnet_naming_style.disallowed_style.required_prefix = ____RULE_VIOLATION____ @@ -337,7 +340,7 @@ dotnet_naming_rule.stylecop_instance_fields_must_be_private_rule.severity dotnet_naming_symbols.stylecop_private_fields_group.applicable_accessibilities = private dotnet_naming_symbols.stylecop_private_fields_group.applicable_kinds = field dotnet_naming_rule.stylecop_private_fields_must_be_camel_case_rule.symbols = stylecop_private_fields_group -dotnet_naming_rule.stylecop_private_fields_must_be_camel_case_rule.style = camel_case_style +dotnet_naming_rule.stylecop_private_fields_must_be_camel_case_rule.style = prefix_underscore_style dotnet_naming_rule.stylecop_private_fields_must_be_camel_case_rule.severity = warning # Local variables must be camelCase diff --git a/src/Benchmark/Transport.cs b/src/Benchmark/Transport.cs index 459bf18..2296f88 100644 --- a/src/Benchmark/Transport.cs +++ b/src/Benchmark/Transport.cs @@ -11,31 +11,31 @@ namespace Benchmark; [MemoryDiagnoser] public class Transport { - private readonly IList manyLokiEvents; - private readonly IList lokiEvents = new List { + private readonly IList _manyLokiEvents; + private readonly IList _lokiEvents = new List { new( new LokiLabels(new LokiLabel("env", "benchmark"), new LokiLabel("job", "WriteLogEventsAsync")), DateTime.Now, "Info|Receive message from \"A\" with destination \"B\".")}; - private readonly HttpLokiTransport transport = new(new LokiHttpClient( + private readonly HttpLokiTransport _transport = new(new LokiHttpClient( new HttpClient { BaseAddress = new Uri("http://localhost:3100") }), false); public Transport() { - manyLokiEvents = new List(100); + _manyLokiEvents = new List(100); for(var i = 0; i < 100; i++) - manyLokiEvents.Add(new LokiEvent(lokiEvents[0].Labels, DateTime.Now, lokiEvents[0].Line)); + _manyLokiEvents.Add(new LokiEvent(_lokiEvents[0].Labels, DateTime.Now, _lokiEvents[0].Line)); } [Benchmark] public async Task WriteLogEventsAsync() { - await transport.WriteLogEventsAsync(lokiEvents).ConfigureAwait(false); + await _transport.WriteLogEventsAsync(_lokiEvents).ConfigureAwait(false); } [Benchmark] public async Task ManyWriteLogEventsAsync() { - await transport.WriteLogEventsAsync(manyLokiEvents).ConfigureAwait(false); + await _transport.WriteLogEventsAsync(_manyLokiEvents).ConfigureAwait(false); } } diff --git a/src/NLog.Loki.Tests/LokiTargetTests.cs b/src/NLog.Loki.Tests/LokiTargetTests.cs index decff0a..4c5ce5d 100644 --- a/src/NLog.Loki.Tests/LokiTargetTests.cs +++ b/src/NLog.Loki.Tests/LokiTargetTests.cs @@ -1,121 +1,98 @@ using System; -using System.Net; -using System.Net.Http; -using System.Text; -using System.Threading.Tasks; using NLog.Config; using NLog.Layouts; using NLog.Targets.Wrappers; using NUnit.Framework; -namespace NLog.Loki.Tests +namespace NLog.Loki.Tests; + +[TestFixture] +public class LokiTargetTests { - [TestFixture] - public class LokiTargetTests + [Test] + public void Write() { - public class NullLokiHttpClient : ILokiHttpClient - { - private readonly StringBuilder stringBuilder; - - public NullLokiHttpClient(StringBuilder stringBuilder) - { - this.stringBuilder = stringBuilder; - } - - public async Task PostAsync(string requestUri, HttpContent httpContent) - { - var result = await httpContent.ReadAsStringAsync(); - stringBuilder.Append(result); - stringBuilder.AppendLine(); - - return new HttpResponseMessage(HttpStatusCode.OK); - } - } + var configuration = new LoggingConfiguration(); - [Test] - public void Write() + using var lokiTarget = new LokiTarget { - var configuration = new LoggingConfiguration(); - - var lokiTarget = new LokiTarget - { - Endpoint = "http://grafana.lvh.me:3100", - IncludeMdlc = true, - Labels = { - new LokiTargetLabel { - Name = "env", - Layout = Layout.FromString("${basedir}") - }, - new LokiTargetLabel { - Name = "server", - Layout = Layout.FromString("${machinename:lowercase=true}") - }, - new LokiTargetLabel { - Name = "level", - Layout = Layout.FromString("${level:lowercase=true}") - } + Endpoint = "http://grafana.lvh.me:3100", + IncludeMdlc = true, + Labels = { + new LokiTargetLabel { + Name = "env", + Layout = Layout.FromString("${basedir}") + }, + new LokiTargetLabel { + Name = "server", + Layout = Layout.FromString("${machinename:lowercase=true}") + }, + new LokiTargetLabel { + Name = "level", + Layout = Layout.FromString("${level:lowercase=true}") } - }; + } + }; - var target = new BufferingTargetWrapper(lokiTarget) - { - BufferSize = 500 - }; + var target = new BufferingTargetWrapper(lokiTarget) + { + BufferSize = 500 + }; - configuration.AddTarget("loki", target); + configuration.AddTarget("loki", target); - var rule = new LoggingRule("*", LogLevel.Debug, target); - configuration.LoggingRules.Add(rule); + var rule = new LoggingRule("*", LogLevel.Debug, target); + configuration.LoggingRules.Add(rule); - LogManager.Configuration = configuration; + LogManager.Configuration = configuration; - var log = LogManager.GetLogger(typeof(LokiTargetTests).FullName); + var log = LogManager.GetLogger(typeof(LokiTargetTests).FullName); - for(var n = 0; n < 100; ++n) + for(var n = 0; n < 100; ++n) + { + using(MappedDiagnosticsLogicalContext.SetScoped("env", "dev")) { - using(MappedDiagnosticsLogicalContext.SetScoped("env", "dev")) - { - log.Fatal("Hello world"); - } + log.Fatal("Hello world"); + } - using(MappedDiagnosticsLogicalContext.SetScoped("server", Environment.MachineName)) - { - log.Info($"hello again {n}"); + using(MappedDiagnosticsLogicalContext.SetScoped("server", Environment.MachineName)) + { + log.Info($"hello again {n}"); - log.Info($"hello again {n * 2}"); - log.Warn($"hello again {n * 3}"); - } + log.Info($"hello again {n * 2}"); + log.Warn($"hello again {n * 3}"); + } - using(MappedDiagnosticsLogicalContext.SetScoped("cfg", "v1")) - log.Error($"hello again {n * 4}"); + using(MappedDiagnosticsLogicalContext.SetScoped("cfg", "v1")) + log.Error($"hello again {n * 4}"); - try - { - throw new InvalidOperationException(); - } - catch(Exception e) - { - log.Error(e); - } + try + { + throw new InvalidOperationException(); + } + catch(Exception e) + { + log.Error(e); } - - LogManager.Shutdown(); } - [Test] - [TestCase("${environment:SCHEME}://${environment:HOST}:3100/", ExpectedResult = typeof(HttpLokiTransport))] - [TestCase("udp://${environment:HOST}:3100/", ExpectedResult = typeof(NullLokiTransport))] - [TestCase("", ExpectedResult = typeof(NullLokiTransport))] - [TestCase(null, ExpectedResult = typeof(NullLokiTransport))] - public Type GetLokiTransport(string endpointLayout) - { - Environment.SetEnvironmentVariable("SCHEME", "https"); - Environment.SetEnvironmentVariable("HOST", "loki.lvh.me"); + LogManager.Shutdown(); + } - var endpoint = Layout.FromString(endpointLayout); - var lokiTargetTransport = new LokiTarget().GetLokiTransport(endpoint, null, null, false); + [Test] + [TestCase("${environment:SCHEME}://${environment:HOST}:3100/", ExpectedResult = typeof(HttpLokiTransport))] + [TestCase("udp://${environment:HOST}:3100/", ExpectedResult = typeof(NullLokiTransport))] + [TestCase("", ExpectedResult = typeof(NullLokiTransport))] + [TestCase(null, ExpectedResult = typeof(NullLokiTransport))] + public Type GetLokiTransport(string endpointLayout) + { + Environment.SetEnvironmentVariable("SCHEME", "https"); + Environment.SetEnvironmentVariable("HOST", "loki.lvh.me"); - return lokiTargetTransport.GetType(); - } + var endpoint = Layout.FromString(endpointLayout); + using var target = new LokiTarget(); + using var lokiTargetTransport = target.GetLokiTransport(endpoint, null, null, false); + return lokiTargetTransport.GetType(); } } + diff --git a/src/NLog.Loki/HttpLokiTransport.cs b/src/NLog.Loki/HttpLokiTransport.cs index 48f629c..8e0908b 100644 --- a/src/NLog.Loki/HttpLokiTransport.cs +++ b/src/NLog.Loki/HttpLokiTransport.cs @@ -6,56 +6,61 @@ using NLog.Common; using NLog.Loki.Model; -namespace NLog.Loki +namespace NLog.Loki; + +/// +/// See https://grafana.com/docs/loki/latest/api/#examples-4 +/// +internal sealed class HttpLokiTransport : ILokiTransport { - /// - /// See https://grafana.com/docs/loki/latest/api/#examples-4 - /// - internal class HttpLokiTransport : ILokiTransport + private readonly JsonSerializerOptions _jsonOptions; + private readonly ILokiHttpClient _lokiHttpClient; + + public HttpLokiTransport(ILokiHttpClient lokiHttpClient, bool orderWrites) + { + _lokiHttpClient = lokiHttpClient; + + _jsonOptions = new JsonSerializerOptions(); + _jsonOptions.Converters.Add(new LokiEventsSerializer(orderWrites)); + _jsonOptions.Converters.Add(new LokiEventSerializer()); + } + + public async Task WriteLogEventsAsync(IEnumerable lokiEvents) { - private readonly JsonSerializerOptions jsonOptions; - private readonly ILokiHttpClient lokiHttpClient; - - public HttpLokiTransport(ILokiHttpClient lokiHttpClient, bool orderWrites) - { - this.lokiHttpClient = lokiHttpClient; - - jsonOptions = new JsonSerializerOptions(); - jsonOptions.Converters.Add(new LokiEventsSerializer(orderWrites)); - jsonOptions.Converters.Add(new LokiEventSerializer()); - } - - public async Task WriteLogEventsAsync(IEnumerable lokiEvents) - { - using var jsonStreamContent = JsonContent.Create(lokiEvents, options: jsonOptions); - using var response = await lokiHttpClient.PostAsync("loki/api/v1/push", jsonStreamContent).ConfigureAwait(false); - await ValidateHttpResponse(response); - } - - public async Task WriteLogEventsAsync(LokiEvent lokiEvent) - { - using var jsonStreamContent = JsonContent.Create(lokiEvent, options: jsonOptions); - using var response = await lokiHttpClient.PostAsync("loki/api/v1/push", jsonStreamContent).ConfigureAwait(false); - await ValidateHttpResponse(response); - } - - private static async ValueTask ValidateHttpResponse(HttpResponseMessage response) - { - if(response.IsSuccessStatusCode) - return; - - // Read the response's content - string content = response.Content == null ? null : - await response.Content.ReadAsStringAsync().ConfigureAwait(false); - - InternalLogger.Error("Failed pushing logs to Loki. Code: {Code}. Reason: {Reason}. Message: {Message}.", - response.StatusCode, response.ReasonPhrase, content); - - #if NET6_0_OR_GREATER + using var jsonStreamContent = JsonContent.Create(lokiEvents, options: _jsonOptions); + using var response = await _lokiHttpClient.PostAsync("loki/api/v1/push", jsonStreamContent).ConfigureAwait(false); + await ValidateHttpResponse(response); + } + + public async Task WriteLogEventsAsync(LokiEvent lokiEvent) + { + using var jsonStreamContent = JsonContent.Create(lokiEvent, options: _jsonOptions); + using var response = await _lokiHttpClient.PostAsync("loki/api/v1/push", jsonStreamContent).ConfigureAwait(false); + await ValidateHttpResponse(response); + } + + private static async ValueTask ValidateHttpResponse(HttpResponseMessage response) + { + if(response.IsSuccessStatusCode) + return; + + // Read the response's content + var content = response.Content == null ? null : + await response.Content.ReadAsStringAsync().ConfigureAwait(false); + + InternalLogger.Error("Failed pushing logs to Loki. Code: {Code}. Reason: {Reason}. Message: {Message}.", + response.StatusCode, response.ReasonPhrase, content); + +#if NET6_0_OR_GREATER throw new HttpRequestException("Failed pushing logs to Loki.", inner: null, response.StatusCode); - #else - throw new HttpRequestException("Failed pushing logs to Loki."); - #endif - } +#else + throw new HttpRequestException("Failed pushing logs to Loki."); +#endif + } + + public void Dispose() + { + _lokiHttpClient.Dispose(); } } + diff --git a/src/NLog.Loki/ILokiHttpClient.cs b/src/NLog.Loki/ILokiHttpClient.cs index 880334e..9e9f5b1 100644 --- a/src/NLog.Loki/ILokiHttpClient.cs +++ b/src/NLog.Loki/ILokiHttpClient.cs @@ -1,9 +1,10 @@ +using System; using System.Net.Http; using System.Threading.Tasks; namespace NLog.Loki { - internal interface ILokiHttpClient + internal interface ILokiHttpClient : IDisposable { Task PostAsync(string requestUri, HttpContent httpContent); } diff --git a/src/NLog.Loki/ILokiTransport.cs b/src/NLog.Loki/ILokiTransport.cs index 7c24eb3..9575f4a 100644 --- a/src/NLog.Loki/ILokiTransport.cs +++ b/src/NLog.Loki/ILokiTransport.cs @@ -1,12 +1,12 @@ +using System; using System.Collections.Generic; using System.Threading.Tasks; using NLog.Loki.Model; -namespace NLog.Loki +namespace NLog.Loki; + +public interface ILokiTransport : IDisposable { - public interface ILokiTransport - { - Task WriteLogEventsAsync(IEnumerable lokiEvents); - Task WriteLogEventsAsync(LokiEvent lokiEvent); - } + Task WriteLogEventsAsync(IEnumerable lokiEvents); + Task WriteLogEventsAsync(LokiEvent lokiEvent); } diff --git a/src/NLog.Loki/LokiHttpClient.cs b/src/NLog.Loki/LokiHttpClient.cs index 7d3763b..9e935b0 100644 --- a/src/NLog.Loki/LokiHttpClient.cs +++ b/src/NLog.Loki/LokiHttpClient.cs @@ -1,20 +1,24 @@ using System.Net.Http; using System.Threading.Tasks; -namespace NLog.Loki +namespace NLog.Loki; + +internal sealed class LokiHttpClient : ILokiHttpClient { - internal class LokiHttpClient : ILokiHttpClient + private readonly HttpClient _httpClient; + + public LokiHttpClient(HttpClient httpClient) { - private readonly HttpClient httpClient; + _httpClient = httpClient; + } - public LokiHttpClient(HttpClient httpClient) - { - this.httpClient = httpClient; - } + public Task PostAsync(string requestUri, HttpContent httpContent) + { + return _httpClient.PostAsync(requestUri, httpContent); + } - public Task PostAsync(string requestUri, HttpContent httpContent) - { - return httpClient.PostAsync(requestUri, httpContent); - } + public void Dispose() + { + _httpClient.Dispose(); } } diff --git a/src/NLog.Loki/LokiTarget.cs b/src/NLog.Loki/LokiTarget.cs index cf01fb8..3cbd8b4 100644 --- a/src/NLog.Loki/LokiTarget.cs +++ b/src/NLog.Loki/LokiTarget.cs @@ -16,7 +16,7 @@ namespace NLog.Loki [Target("loki")] public class LokiTarget : AsyncTaskTarget { - private readonly Lazy lazyLokiTransport; + private readonly Lazy _lazyLokiTransport; [RequiredParameter] public Layout Endpoint { get; set; } @@ -41,27 +41,28 @@ public LokiTarget() { Labels = new List(); - lazyLokiTransport = new Lazy( + _lazyLokiTransport = new Lazy( () => GetLokiTransport(Endpoint, Username, Password, OrderWrites), LazyThreadSafetyMode.ExecutionAndPublication); + InitializeTarget(); } protected override void Write(IList logEvents) { var events = GetLokiEvents(logEvents.Select(alei => alei.LogEvent)); - lazyLokiTransport.Value.WriteLogEventsAsync(events).ConfigureAwait(false).GetAwaiter().GetResult(); + _lazyLokiTransport.Value.WriteLogEventsAsync(events).ConfigureAwait(false).GetAwaiter().GetResult(); } protected override Task WriteAsyncTask(LogEventInfo logEvent, CancellationToken cancellationToken) { var @event = GetLokiEvent(logEvent); - return lazyLokiTransport.Value.WriteLogEventsAsync(@event); + return _lazyLokiTransport.Value.WriteLogEventsAsync(@event); } protected override Task WriteAsyncTask(IList logEvents, CancellationToken cancellationToken) { var events = GetLokiEvents(logEvents); - return lazyLokiTransport.Value.WriteLogEventsAsync(events); + return _lazyLokiTransport.Value.WriteLogEventsAsync(events); } private IEnumerable GetLokiEvents(IEnumerable logEvents) @@ -91,19 +92,11 @@ internal ILokiTransport GetLokiTransport( if(Uri.TryCreate(endpointUri, UriKind.Absolute, out var uri)) { if(uri.Scheme == Uri.UriSchemeHttp || uri.Scheme == Uri.UriSchemeHttps) - { - var lokiHttpClient = LokiHttpClientFactory(uri, usr, pwd); - var httpLokiTransport = new HttpLokiTransport(lokiHttpClient, orderWrites); - - return httpLokiTransport; - } + return new HttpLokiTransport(LokiHttpClientFactory(uri, usr, pwd), orderWrites); } InternalLogger.Warn("Unable to create a valid Loki Endpoint URI from '{0}'", endpoint); - - var nullLokiTransport = new NullLokiTransport(); - - return nullLokiTransport; + return new NullLokiTransport(); } internal static ILokiHttpClient GetLokiHttpClient(Uri uri, string username, string password) @@ -114,10 +107,22 @@ internal static ILokiHttpClient GetLokiHttpClient(Uri uri, string username, stri var byteArray = Encoding.ASCII.GetBytes($"{username}:{password}"); httpClient.DefaultRequestHeaders.Authorization = new System.Net.Http.Headers.AuthenticationHeaderValue("Basic", Convert.ToBase64String(byteArray)); } + return new LokiHttpClient(httpClient); + } - var lokiHttpClient = new LokiHttpClient(httpClient); - - return lokiHttpClient; + private bool _isDisposed; + protected override void Dispose(bool isDisposing) + { + if(!_isDisposed) + { + if(isDisposing) + { + if(_lazyLokiTransport.IsValueCreated) + _lazyLokiTransport.Value.Dispose(); + } + _isDisposed = true; + } + base.Dispose(isDisposing); } } } diff --git a/src/NLog.Loki/NullLokiTransport.cs b/src/NLog.Loki/NullLokiTransport.cs index 18a49c2..fb1b7be 100644 --- a/src/NLog.Loki/NullLokiTransport.cs +++ b/src/NLog.Loki/NullLokiTransport.cs @@ -8,4 +8,8 @@ internal class NullLokiTransport : ILokiTransport { public Task WriteLogEventsAsync(IEnumerable lokiEvents) => Task.CompletedTask; public Task WriteLogEventsAsync(LokiEvent lokiEvent) => Task.CompletedTask; + public void Dispose() + { + // Nothing to dispose in this null implementation. + } } From ae3a415d408a8032d13556fa62f6c5d516cc364d Mon Sep 17 00:00:00 2001 From: Corentin Altepe Date: Sat, 12 Mar 2022 11:37:05 +0100 Subject: [PATCH 2/9] Implemented HTTP Gzip compression. --- src/Benchmark/Transport.cs | 3 +- src/NLog.Loki.Tests/HttpLokiTransportTests.cs | 19 +++--- src/NLog.Loki/CompressedContent.cs | 58 +++++++++++++++++++ src/NLog.Loki/HttpLokiTransport.cs | 28 +++++++-- src/NLog.Loki/LokiTarget.cs | 9 ++- 5 files changed, 102 insertions(+), 15 deletions(-) create mode 100644 src/NLog.Loki/CompressedContent.cs diff --git a/src/Benchmark/Transport.cs b/src/Benchmark/Transport.cs index 2296f88..bc031f3 100644 --- a/src/Benchmark/Transport.cs +++ b/src/Benchmark/Transport.cs @@ -1,5 +1,6 @@ using System; using System.Collections.Generic; +using System.IO.Compression; using System.Net.Http; using System.Threading.Tasks; using BenchmarkDotNet.Attributes; @@ -18,7 +19,7 @@ public class Transport DateTime.Now, "Info|Receive message from \"A\" with destination \"B\".")}; private readonly HttpLokiTransport _transport = new(new LokiHttpClient( - new HttpClient { BaseAddress = new Uri("http://localhost:3100") }), false); + new HttpClient { BaseAddress = new Uri("http://localhost:3100") }), false, CompressionLevel.NoCompression); public Transport() { diff --git a/src/NLog.Loki.Tests/HttpLokiTransportTests.cs b/src/NLog.Loki.Tests/HttpLokiTransportTests.cs index 3a9f7d1..5cf8c2f 100644 --- a/src/NLog.Loki.Tests/HttpLokiTransportTests.cs +++ b/src/NLog.Loki.Tests/HttpLokiTransportTests.cs @@ -1,5 +1,6 @@ using System; using System.Collections.Generic; +using System.IO.Compression; using System.Net; using System.Net.Http; using System.Net.Http.Json; @@ -43,7 +44,7 @@ public async Task SerializeMessageToHttpLokiWithoutOrdering() }); // Send the logging request - var transport = new HttpLokiTransport(httpClient.Object, orderWrites: false); + var transport = new HttpLokiTransport(httpClient.Object, orderWrites: false, CompressionLevel.NoCompression); await transport.WriteLogEventsAsync(events).ConfigureAwait(false); // Verify the json message format @@ -70,7 +71,7 @@ public async Task SerializeMessageToHttpLokiWithOrdering() }); // Send the logging request - var transport = new HttpLokiTransport(httpClient.Object, orderWrites: true); + var transport = new HttpLokiTransport(httpClient.Object, orderWrites: true, CompressionLevel.NoCompression); await transport.WriteLogEventsAsync(events).ConfigureAwait(false); // Verify the json message format @@ -97,7 +98,7 @@ public async Task SerializeMessageToHttpLokiSingleEvent() }); // Send the logging request - var transport = new HttpLokiTransport(httpClient.Object, false); + var transport = new HttpLokiTransport(httpClient.Object, false, CompressionLevel.NoCompression); await transport.WriteLogEventsAsync(lokiEvent).ConfigureAwait(false); // Verify the json message format @@ -115,7 +116,7 @@ public void ThrowOnHttpClientException() .ThrowsAsync(new Exception("Something went wrong whem sending HTTP message.")); // Send the logging request - var transport = new HttpLokiTransport(httpClient.Object, false); + var transport = new HttpLokiTransport(httpClient.Object, false, CompressionLevel.NoCompression); var exception = Assert.ThrowsAsync(() => transport.WriteLogEventsAsync(CreateLokiEvents())); Assert.AreEqual("Something went wrong whem sending HTTP message.", exception.Message); } @@ -133,12 +134,12 @@ public void ThrowOnNonSuccessResponseCode() .Returns(Task.FromResult(response)); // Send the logging request - var transport = new HttpLokiTransport(httpClient.Object, false); + var transport = new HttpLokiTransport(httpClient.Object, false, CompressionLevel.NoCompression); var exception = Assert.ThrowsAsync(() => transport.WriteLogEventsAsync(CreateLokiEvents())); Assert.AreEqual("Failed pushing logs to Loki.", exception.Message); - - #if NET6_0_OR_GREATER - Assert.AreEqual(HttpStatusCode.Conflict, exception.StatusCode); - #endif + +#if NET6_0_OR_GREATER + Assert.AreEqual(HttpStatusCode.Conflict, exception.StatusCode); +#endif } } diff --git a/src/NLog.Loki/CompressedContent.cs b/src/NLog.Loki/CompressedContent.cs new file mode 100644 index 0000000..0721c0d --- /dev/null +++ b/src/NLog.Loki/CompressedContent.cs @@ -0,0 +1,58 @@ +using System; +using System.IO; +using System.IO.Compression; +using System.Net; +using System.Net.Http; +using System.Threading.Tasks; + +namespace NLog.Loki; + +/// +/// GZipped HTTP Content. +/// Inspired from https://programmer.help/blogs/httpclient-and-aps.net-web-api-compression-and-decompression-of-request-content.html +/// by arunmj82 (Tue, 18 Dec 2018). +/// +internal sealed class CompressedContent : HttpContent +{ + private readonly HttpContent _originalContent; + private readonly CompressionLevel _level; + + public CompressedContent(HttpContent content, CompressionLevel level) + { + _originalContent = content ?? throw new ArgumentNullException("content"); + _level = level; + + // Copy the underlying content's headers + foreach(var header in _originalContent.Headers) + _ = Headers.TryAddWithoutValidation(header.Key, header.Value); + + // Add Content-Encoding header + Headers.ContentEncoding.Add("gzip"); + } + + protected override bool TryComputeLength(out long length) + { + length = -1; + return false; + } + + protected async override Task SerializeToStreamAsync(Stream stream, TransportContext context) + { + using var gzipStream = new GZipStream(stream, _level, leaveOpen: true); + await _originalContent.CopyToAsync(gzipStream); + } + + private bool _isDisposed; + protected override void Dispose(bool isDisposing) + { + if(!_isDisposed) + { + if(isDisposing) + { + _originalContent?.Dispose(); + } + _isDisposed = true; + } + base.Dispose(isDisposing); + } +} diff --git a/src/NLog.Loki/HttpLokiTransport.cs b/src/NLog.Loki/HttpLokiTransport.cs index 8e0908b..3a32b29 100644 --- a/src/NLog.Loki/HttpLokiTransport.cs +++ b/src/NLog.Loki/HttpLokiTransport.cs @@ -1,4 +1,5 @@ using System.Collections.Generic; +using System.IO.Compression; using System.Net.Http; using System.Net.Http.Json; using System.Text.Json; @@ -15,11 +16,15 @@ internal sealed class HttpLokiTransport : ILokiTransport { private readonly JsonSerializerOptions _jsonOptions; private readonly ILokiHttpClient _lokiHttpClient; + private readonly CompressionLevel _gzipLevel; - public HttpLokiTransport(ILokiHttpClient lokiHttpClient, bool orderWrites) + public HttpLokiTransport( + ILokiHttpClient lokiHttpClient, + bool orderWrites, + CompressionLevel gzipLevel) { _lokiHttpClient = lokiHttpClient; - + _gzipLevel = gzipLevel; _jsonOptions = new JsonSerializerOptions(); _jsonOptions.Converters.Add(new LokiEventsSerializer(orderWrites)); _jsonOptions.Converters.Add(new LokiEventSerializer()); @@ -27,18 +32,33 @@ public HttpLokiTransport(ILokiHttpClient lokiHttpClient, bool orderWrites) public async Task WriteLogEventsAsync(IEnumerable lokiEvents) { - using var jsonStreamContent = JsonContent.Create(lokiEvents, options: _jsonOptions); + using var jsonStreamContent = CreateContent(lokiEvents); using var response = await _lokiHttpClient.PostAsync("loki/api/v1/push", jsonStreamContent).ConfigureAwait(false); await ValidateHttpResponse(response); } public async Task WriteLogEventsAsync(LokiEvent lokiEvent) { - using var jsonStreamContent = JsonContent.Create(lokiEvent, options: _jsonOptions); + using var jsonStreamContent = CreateContent(lokiEvent); using var response = await _lokiHttpClient.PostAsync("loki/api/v1/push", jsonStreamContent).ConfigureAwait(false); await ValidateHttpResponse(response); } + /// + /// Prepares the HttpContent for the loki event(s). + /// If gzip compression is enabled, prepares a gzip stream with the appropriate headers. + /// + private HttpContent CreateContent(T lokiEvent) + { + var jsonContent = JsonContent.Create(lokiEvent, options: _jsonOptions); + + // If no compression required + if(_gzipLevel == CompressionLevel.NoCompression) + return jsonContent; + + return new CompressedContent(jsonContent, _gzipLevel); + } + private static async ValueTask ValidateHttpResponse(HttpResponseMessage response) { if(response.IsSuccessStatusCode) diff --git a/src/NLog.Loki/LokiTarget.cs b/src/NLog.Loki/LokiTarget.cs index 3cbd8b4..c231237 100644 --- a/src/NLog.Loki/LokiTarget.cs +++ b/src/NLog.Loki/LokiTarget.cs @@ -1,5 +1,6 @@ using System; using System.Collections.Generic; +using System.IO.Compression; using System.Linq; using System.Net.Http; using System.Text; @@ -32,6 +33,12 @@ public class LokiTarget : AsyncTaskTarget /// public bool OrderWrites { get; set; } = true; + /// + /// Defines if the HTTP messages sent to Loki must be gzip compressed, and with which compression level. + /// Possible values: NoCompression (default), Optimal, Fastest and SmallestSize (.NET 6 support only). + /// + public CompressionLevel CompressionLevel { get; set; } = CompressionLevel.NoCompression; + [ArrayParameter(typeof(LokiTargetLabel), "label")] public IList Labels { get; } @@ -92,7 +99,7 @@ internal ILokiTransport GetLokiTransport( if(Uri.TryCreate(endpointUri, UriKind.Absolute, out var uri)) { if(uri.Scheme == Uri.UriSchemeHttp || uri.Scheme == Uri.UriSchemeHttps) - return new HttpLokiTransport(LokiHttpClientFactory(uri, usr, pwd), orderWrites); + return new HttpLokiTransport(LokiHttpClientFactory(uri, usr, pwd), orderWrites, CompressionLevel); } InternalLogger.Warn("Unable to create a valid Loki Endpoint URI from '{0}'", endpoint); From b47b5432eabe4489684f30c8066041ca10825fc3 Mon Sep 17 00:00:00 2001 From: Corentin Altepe Date: Sat, 12 Mar 2022 12:07:02 +0100 Subject: [PATCH 3/9] Implemented test on the CompressedContent for all 4 levels. --- src/NLog.Loki.Tests/HttpLokiTransportTests.cs | 60 +++++++++++++++---- 1 file changed, 50 insertions(+), 10 deletions(-) diff --git a/src/NLog.Loki.Tests/HttpLokiTransportTests.cs b/src/NLog.Loki.Tests/HttpLokiTransportTests.cs index 5cf8c2f..3bd0b22 100644 --- a/src/NLog.Loki.Tests/HttpLokiTransportTests.cs +++ b/src/NLog.Loki.Tests/HttpLokiTransportTests.cs @@ -1,9 +1,11 @@ using System; using System.Collections.Generic; using System.IO.Compression; +using System.Linq; using System.Net; using System.Net.Http; using System.Net.Http.Json; +using System.Text; using System.Threading.Tasks; using Moq; using NLog.Loki.Model; @@ -13,17 +15,17 @@ namespace NLog.Loki.Tests; public class HttpLokiTransportTests { - private static List CreateLokiEvents() + private static IEnumerable CreateLokiEvents(int numberEvents = 3) { var date = new DateTime(2021, 12, 27, 9, 48, 26, DateTimeKind.Utc); - return new List { - new(new LokiLabels(new LokiLabel("env", "unittest"), new LokiLabel("job", "Job1")), - date, "Info|Receive message from A with destination B."), - new(new LokiLabels(new LokiLabel("env", "unittest"), new LokiLabel("job", "Job1")), - date + TimeSpan.FromSeconds(2.2), "Info|Another event has occured here."), - new(new LokiLabels(new LokiLabel("env", "unittest"), new LokiLabel("job", "Job1")), - date - TimeSpan.FromSeconds(0.9), "Info|Event from another stream."), - }; + for(var i = 0; i < numberEvents; i++) + { + yield return new(new LokiLabels(new LokiLabel("env", "unittest"), new LokiLabel("job", "Job1")), date, "Info|Receive message from A with destination B."); + i++; + yield return new(new LokiLabels(new LokiLabel("env", "unittest"), new LokiLabel("job", "Job1")), date + TimeSpan.FromSeconds(2.2), "Info|Another event has occured here."); + i++; + yield return new(new LokiLabels(new LokiLabel("env", "unittest"), new LokiLabel("job", "Job1")), date - TimeSpan.FromSeconds(0.9), "Info|Event from another stream."); + } } [Test] @@ -84,7 +86,7 @@ public async Task SerializeMessageToHttpLokiWithOrdering() public async Task SerializeMessageToHttpLokiSingleEvent() { // Prepare the event to be sent to loki - var lokiEvent = CreateLokiEvents()[2]; + var lokiEvent = CreateLokiEvents().ToList()[2]; // Configure the ILokiHttpClient such that we intercept the JSON content and simulate an OK response from Loki. string serializedJsonMessage = null; @@ -142,4 +144,42 @@ public void ThrowOnNonSuccessResponseCode() Assert.AreEqual(HttpStatusCode.Conflict, exception.StatusCode); #endif } + + [Test] + [TestCase(CompressionLevel.NoCompression)] + [TestCase(CompressionLevel.Fastest)] + [TestCase(CompressionLevel.Optimal)] +#if NET6_0_OR_GREATER + [TestCase(CompressionLevel.SmallestSize)] +#endif + public async Task CompressMessage(CompressionLevel level) + { + // Prepare the events to be sent to loki + var events = CreateLokiEvents(3); + + // Configure the ILokiHttpClient such that we intercept the JSON content and simulate an OK response from Loki. + string serializedJsonMessage = null; + var httpClient = new Mock(); + _ = httpClient.Setup(c => c.PostAsync("loki/api/v1/push", It.IsAny())) + .Returns(async (s, content) => + { + // Intercept the gzipped json content so that we can verify it. + var stream = await content.ReadAsStreamAsync(); + if(level != CompressionLevel.NoCompression) + stream = new GZipStream(stream, CompressionMode.Decompress); + var buffer = new byte[128000]; + var length = stream.Read(buffer, 0, buffer.Length); + serializedJsonMessage = Encoding.UTF8.GetString(buffer, 0, length); + return new HttpResponseMessage(HttpStatusCode.OK); + }); + + // Send the logging request + var transport = new HttpLokiTransport(httpClient.Object, orderWrites: false, level); + await transport.WriteLogEventsAsync(events).ConfigureAwait(false); + + // Verify the json message format + Assert.AreEqual( + "{\"streams\":[{\"stream\":{\"env\":\"unittest\",\"job\":\"Job1\"},\"values\":[[\"1640598506000000000\",\"Info|Receive message from A with destination B.\"],[\"1640598508200000000\",\"Info|Another event has occured here.\"],[\"1640598505100000000\",\"Info|Event from another stream.\"]]}]}", + serializedJsonMessage); + } } From ce66f0121248d2cf9d07e373a4a5769a95585ae6 Mon Sep 17 00:00:00 2001 From: Corentin Altepe Date: Sat, 12 Mar 2022 12:13:53 +0100 Subject: [PATCH 4/9] Added test assertions on the Http message headers ensuring we have application/json and gzip used when compressed. --- src/NLog.Loki.Tests/HttpLokiTransportTests.cs | 24 ++++++++++++------- 1 file changed, 15 insertions(+), 9 deletions(-) diff --git a/src/NLog.Loki.Tests/HttpLokiTransportTests.cs b/src/NLog.Loki.Tests/HttpLokiTransportTests.cs index 3bd0b22..43377ba 100644 --- a/src/NLog.Loki.Tests/HttpLokiTransportTests.cs +++ b/src/NLog.Loki.Tests/HttpLokiTransportTests.cs @@ -38,10 +38,11 @@ public async Task SerializeMessageToHttpLokiWithoutOrdering() string serializedJsonMessage = null; var httpClient = new Mock(); _ = httpClient.Setup(c => c.PostAsync("loki/api/v1/push", It.IsAny())) - .Returns(async (s, json) => + .Returns(async (s, content) => { // Intercept the json content so that we can verify it. - serializedJsonMessage = await json.ReadAsStringAsync().ConfigureAwait(false); + serializedJsonMessage = await content.ReadAsStringAsync().ConfigureAwait(false); + Assert.AreEqual("application/json", content.Headers.ContentType.MediaType); return new HttpResponseMessage(HttpStatusCode.OK); }); @@ -65,10 +66,11 @@ public async Task SerializeMessageToHttpLokiWithOrdering() string serializedJsonMessage = null; var httpClient = new Mock(); _ = httpClient.Setup(c => c.PostAsync("loki/api/v1/push", It.IsAny())) - .Returns(async (s, json) => + .Returns(async (s, content) => { // Intercept the json content so that we can verify it. - serializedJsonMessage = await json.ReadAsStringAsync().ConfigureAwait(false); + serializedJsonMessage = await content.ReadAsStringAsync().ConfigureAwait(false); + Assert.AreEqual("application/json", content.Headers.ContentType.MediaType); return new HttpResponseMessage(HttpStatusCode.OK); }); @@ -92,10 +94,11 @@ public async Task SerializeMessageToHttpLokiSingleEvent() string serializedJsonMessage = null; var httpClient = new Mock(); _ = httpClient.Setup(c => c.PostAsync("loki/api/v1/push", It.IsAny())) - .Returns(async (s, json) => + .Returns(async (s, content) => { // Intercept the json content so that we can verify it. - serializedJsonMessage = await json.ReadAsStringAsync().ConfigureAwait(false); + serializedJsonMessage = await content.ReadAsStringAsync().ConfigureAwait(false); + Assert.AreEqual("application/json", content.Headers.ContentType.MediaType); return new HttpResponseMessage(HttpStatusCode.OK); }); @@ -146,7 +149,6 @@ public void ThrowOnNonSuccessResponseCode() } [Test] - [TestCase(CompressionLevel.NoCompression)] [TestCase(CompressionLevel.Fastest)] [TestCase(CompressionLevel.Optimal)] #if NET6_0_OR_GREATER @@ -165,11 +167,15 @@ public async Task CompressMessage(CompressionLevel level) { // Intercept the gzipped json content so that we can verify it. var stream = await content.ReadAsStreamAsync(); - if(level != CompressionLevel.NoCompression) - stream = new GZipStream(stream, CompressionMode.Decompress); + Assert.True(content.Headers.ContentEncoding.Any(s => s == "gzip")); + stream = new GZipStream(stream, CompressionMode.Decompress); var buffer = new byte[128000]; var length = stream.Read(buffer, 0, buffer.Length); serializedJsonMessage = Encoding.UTF8.GetString(buffer, 0, length); + + Assert.True(content.Headers.ContentEncoding.Any(s => s == "gzip")); + Assert.AreEqual("application/json", content.Headers.ContentType.MediaType); + return new HttpResponseMessage(HttpStatusCode.OK); }); From 5467026c76663991c82a35124169941e92d33ce7 Mon Sep 17 00:00:00 2001 From: Corentin Altepe Date: Sun, 13 Mar 2022 09:17:58 +0100 Subject: [PATCH 5/9] Added .ConfigureAwait(false) to all await calls. --- src/NLog.Loki.Tests/HttpLokiTransportTests.cs | 2 +- src/NLog.Loki/CompressedContent.cs | 2 +- src/NLog.Loki/HttpLokiTransport.cs | 4 ++-- 3 files changed, 4 insertions(+), 4 deletions(-) diff --git a/src/NLog.Loki.Tests/HttpLokiTransportTests.cs b/src/NLog.Loki.Tests/HttpLokiTransportTests.cs index 43377ba..5bede32 100644 --- a/src/NLog.Loki.Tests/HttpLokiTransportTests.cs +++ b/src/NLog.Loki.Tests/HttpLokiTransportTests.cs @@ -166,7 +166,7 @@ public async Task CompressMessage(CompressionLevel level) .Returns(async (s, content) => { // Intercept the gzipped json content so that we can verify it. - var stream = await content.ReadAsStreamAsync(); + var stream = await content.ReadAsStreamAsync().ConfigureAwait(false); Assert.True(content.Headers.ContentEncoding.Any(s => s == "gzip")); stream = new GZipStream(stream, CompressionMode.Decompress); var buffer = new byte[128000]; diff --git a/src/NLog.Loki/CompressedContent.cs b/src/NLog.Loki/CompressedContent.cs index 0721c0d..b671778 100644 --- a/src/NLog.Loki/CompressedContent.cs +++ b/src/NLog.Loki/CompressedContent.cs @@ -39,7 +39,7 @@ protected override bool TryComputeLength(out long length) protected async override Task SerializeToStreamAsync(Stream stream, TransportContext context) { using var gzipStream = new GZipStream(stream, _level, leaveOpen: true); - await _originalContent.CopyToAsync(gzipStream); + await _originalContent.CopyToAsync(gzipStream).ConfigureAwait(false); } private bool _isDisposed; diff --git a/src/NLog.Loki/HttpLokiTransport.cs b/src/NLog.Loki/HttpLokiTransport.cs index 3a32b29..6745f88 100644 --- a/src/NLog.Loki/HttpLokiTransport.cs +++ b/src/NLog.Loki/HttpLokiTransport.cs @@ -34,14 +34,14 @@ public async Task WriteLogEventsAsync(IEnumerable lokiEvents) { using var jsonStreamContent = CreateContent(lokiEvents); using var response = await _lokiHttpClient.PostAsync("loki/api/v1/push", jsonStreamContent).ConfigureAwait(false); - await ValidateHttpResponse(response); + await ValidateHttpResponse(response).ConfigureAwait(false); } public async Task WriteLogEventsAsync(LokiEvent lokiEvent) { using var jsonStreamContent = CreateContent(lokiEvent); using var response = await _lokiHttpClient.PostAsync("loki/api/v1/push", jsonStreamContent).ConfigureAwait(false); - await ValidateHttpResponse(response); + await ValidateHttpResponse(response).ConfigureAwait(false); } /// From 862122de58879876abec31ee99c3b4ccd9992212 Mon Sep 17 00:00:00 2001 From: Corentin Altepe Date: Sun, 13 Mar 2022 09:25:50 +0100 Subject: [PATCH 6/9] Added compressionLevel option in template project. --- src/Template/nlog.config | 1 + 1 file changed, 1 insertion(+) diff --git a/src/Template/nlog.config b/src/Template/nlog.config index c0a3593..fc53421 100644 --- a/src/Template/nlog.config +++ b/src/Template/nlog.config @@ -17,6 +17,7 @@ endpoint="http://localhost:3100" retryCount="3" orderWrites="true" + compressionLevel="fastest" layout="${level}|${message}${onexception:|${exception:format=type,message,method:maxInnerExceptionLevel=5:innerFormat=shortType,message,method}}|source=${logger}">