[SelfDiagnosticsModule] Write event to log file in a circular way in …

src/OpenTelemetry/Internal/SelfDiagnosticsConfigRefresher.cs

@@ -15,6 +15,7 @@
 // </copyright>
 
 using System;
+using System.IO;
 
 namespace OpenTelemetry.Internal
 {
@@ -31,13 +32,20 @@ internal class SelfDiagnosticsConfigRefresher : IDisposable
  // Commented out for now to avoid the "field was never used" compiler error.
  // private SelfDiagnosticsEventListener eventListener;
 
- // TODO
- // public bool TryGetLogStream(int byteCount, out Stream stream, out int availableByteCount)
- // {
- // stream = null;
- // availableByteCount = 0;
- // return false;
- // }
+ /// <summary>
+ /// Try to get the log stream which is seeked to the position where the next line of log should be written.
+ /// </summary>
+ /// <param name="byteCount">The number of bytes that need to be written.</param>
+ /// <param name="stream">When this method returns, contains the Stream object where `byteCount` of bytes can be written.</param>
+ /// <param name="availableByteCount">The number of bytes that is remaining until the end of the stream.</param>
+ /// <returns>Whether the logger should log in the stream.</returns>
+ public virtual bool TryGetLogStream(int byteCount, out Stream stream, out int availableByteCount)
+ {
+ // TODO in next PR
+ stream = null;
+ availableByteCount = 0;
+ return false;
+ }
 
  /// <inheritdoc/>
  public void Dispose()

src/OpenTelemetry/Internal/SelfDiagnosticsEventListener.cs

@@ -15,24 +15,138 @@
 // </copyright>
 
 using System;
+using System.Collections.ObjectModel;
 using System.Diagnostics.Tracing;
+using System.Globalization;
+using System.IO;
+using System.Linq;
+using System.Text;
+using System.Threading;
 
 namespace OpenTelemetry.Internal
 {
  /// <summary>
  /// SelfDiagnosticsEventListener class enables the events from OpenTelemetry event sources
- /// and record the events to a local file via the SelfDiagnosticsRecorder class.
+ /// and write the events to a local file in a circular way.
  /// </summary>
  internal class SelfDiagnosticsEventListener : EventListener
  {
+ // Buffer size of the log line. A UTF-16 encoded character in C# can take up to 4 bytes if encoded in UTF-8.
+ private const int BUFFERSIZE = 4 * 5120;
  private const string EventSourceNamePrefix = "OpenTelemetry-";
  private readonly EventLevel logLevel;
  private readonly SelfDiagnosticsConfigRefresher configRefresher;
+ private readonly ThreadLocal<byte[]> writeBuffer = new ThreadLocal<byte[]>(() => null);
 
  public SelfDiagnosticsEventListener(EventLevel logLevel, SelfDiagnosticsConfigRefresher configRefresher)
  {
  this.logLevel = logLevel;
- this.configRefresher = configRefresher;
+ this.configRefresher = configRefresher ?? throw new ArgumentNullException(nameof(configRefresher));
+ }
+
+ /// <summary>
+ /// Encode a string into the designated position in a buffer of bytes, which will be written as log.
+ /// If isParameter is true, wrap "{}" around the string.
+ /// The buffer should not be filled to full, leaving at least one byte empty space to fill a '\n' later.
+ /// If the buffer cannot hold all characters, truncate the string and replace extra content with "...".
+ /// The buffer is not guaranteed to be filled until the last byte due to variable encoding length of UTF-8,
+ /// in order to prioritize speed over space.
+ /// </summary>
+ /// <param name="str">The string to be encoded.</param>
+ /// <param name="isParameter">Whether the string is a parameter. If true, "{}" will be wrapped around the string.</param>
+ /// <param name="buffer">The byte array to contain the resulting sequence of bytes.</param>
+ /// <param name="position">The position at which to start writing the resulting sequence of bytes.</param>
+ /// <returns>The position of the buffer after the last byte of the resulting sequence.</returns>
+ internal static int EncodeInBuffer(string str, bool isParameter, byte[] buffer, int position)
+ {
+ int charCount = str.Length;
+ int ellipses = isParameter ? "{...}\n".Length : "...\n".Length;
+
+ // Ensure there is space for "{...}\n" or "...\n".
+ if (buffer.Length - position - ellipses < 0)
+ {
+ return position;
+ }
+
+ int estimateOfCharacters = (buffer.Length - position - ellipses) / 2;
+
+ // Ensure the UTF-16 encoded string can fit in buffer UTF-8 encoding.
+ // And leave space for "{...}\n" or "...\n".
+ if (charCount > estimateOfCharacters)
+ {
+ charCount = estimateOfCharacters;
+ }
+
+ if (isParameter)
+ {
+ buffer[position++] = (byte)'{';
+ }
+
+ position += Encoding.UTF8.GetBytes(str, 0, charCount, buffer, position);
+ if (charCount != str.Length)
+ {
+ buffer[position++] = (byte)'.';
+ buffer[position++] = (byte)'.';
+ buffer[position++] = (byte)'.';
+ }
+
+ if (isParameter)
+ {
+ buffer[position++] = (byte)'}';
+ }
+
+ return position;
+ }
+
+ internal void WriteEvent(string eventMessage, ReadOnlyCollection<object> payload)
+ {
+ try
+ {
+ var buffer = this.writeBuffer.Value;
+ if (buffer == null)
+ {
+ buffer = new byte[BUFFERSIZE]; // TODO: handle OOM
+ this.writeBuffer.Value = buffer;
+ }
+
+ var timestamp = DateTime.UtcNow.ToString("yyyy-MM-ddTHH:mm:ss.fffffffZ", CultureInfo.InvariantCulture);
+ var pos = Encoding.UTF8.GetBytes(timestamp, 0, timestamp.Length, buffer, 0);
+ buffer[pos++] = (byte)':';
+ pos = EncodeInBuffer(eventMessage, false, buffer, pos);
+ if (payload != null)
+ {
+ // Not using foreach because it can cause allocations
+ for (int i = 0; i < payload.Count; ++i)
+ {
+ object obj = payload.ElementAt(i);
+ if (obj != null)
+ {
+ pos = EncodeInBuffer(obj.ToString(), true, buffer, pos);
+ }
+ }
+ }
+
+ buffer[pos++] = (byte)'\n';
+ int byteCount = pos - 0;
+ if (this.configRefresher.TryGetLogStream(byteCount, out Stream stream, out int availableByteCount))
+ {
+ if (availableByteCount >= byteCount)
+ {
+ stream.Write(buffer, 0, byteCount);
+ }
+ else
+ {
+ stream.Write(buffer, 0, availableByteCount);
+ stream.Seek(0, SeekOrigin.Begin);
+ stream.Write(buffer, availableByteCount, byteCount - availableByteCount);
+ }
+ }
+ }
+ catch (Exception)
+ {
+ // One concurrent condition: memory mapped file is disposed in other thread after TryGetLogStream() finishes.
+ // In this case, silently fail.
+ }
  }
 
  protected override void OnEventSourceCreated(EventSource eventSource)
@@ -58,6 +172,7 @@ protected override void OnEventSourceCreated(EventSource eventSource)
  protected override void OnEventWritten(EventWrittenEventArgs eventData)
  {
  // TODO: retrieve the file stream object from configRefresher and write to it
+ this.WriteEvent(eventData.Message, eventData.Payload);
  }
  }
 }

test/OpenTelemetry.Tests/Internal/SelfDiagnosticsEventListenerTest.cs

@@ -0,0 +1,189 @@
+// <copyright file="SelfDiagnosticsEventListenerTest.cs" company="OpenTelemetry Authors">
+// Copyright The OpenTelemetry Authors
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+// </copyright>
+
+using System;
+using System.Diagnostics.Tracing;
+using System.IO;
+using System.IO.MemoryMappedFiles;
+using System.Text;
+using Moq;
+using Xunit;
+
+namespace OpenTelemetry.Internal
+{
+ public class SelfDiagnosticsEventListenerTest
+ {
+ private const string LOGFILEPATH = "Diagnostics.log";
+ private const string Ellipses = "...\n";
+ private const string EllipsesWithBrackets = "{...}\n";
+
+ [Fact]
+ [Trait("Platform", "Any")]
+ public void SelfDiagnosticsEventListener_constructor_Invalid_Input()
+ {
+ // no configRefresher object
+ Assert.Throws<ArgumentNullException>(() =>
+ {
+ _ = new SelfDiagnosticsEventListener(EventLevel.Error, null);
+ });
+ }
+
+ [Fact]
+ [Trait("Platform", "Any")]
+ public void SelfDiagnosticsEventListener_WriteEvent()
+ {
+ var configRefresherMock = new Mock<SelfDiagnosticsConfigRefresher>();
+ var memoryMappedFile = MemoryMappedFile.CreateFromFile(LOGFILEPATH, FileMode.Create, null, 1024);
+ Stream stream = memoryMappedFile.CreateViewStream();
+ string eventMessage = "Event Message";
+ int timestampPrefixLength = "2020-08-14T20:33:24.4788109Z:".Length;
+ byte[] bytes = Encoding.UTF8.GetBytes(eventMessage);
+ int availableByteCount = 100;
+ configRefresherMock.Setup(configRefresher => configRefresher.TryGetLogStream(timestampPrefixLength + bytes.Length + 1, out stream, out availableByteCount))
+ .Returns(true);
+ var listener = new SelfDiagnosticsEventListener(EventLevel.Error, configRefresherMock.Object);
+ listener.WriteEvent(eventMessage, null);
+ configRefresherMock.Verify(refresher => refresher.TryGetLogStream(timestampPrefixLength + bytes.Length + 1, out stream, out availableByteCount));
+ stream.Dispose();
+ memoryMappedFile.Dispose();
+ AssertFileOutput(LOGFILEPATH, eventMessage);
+ }
+
+ [Fact]
+ [Trait("Platform", "Any")]
+ public void SelfDiagnosticsEventListener_EncodeInBuffer_Empty()
+ {
+ byte[] buffer = new byte[20];
+ int startPos = 0;
+ int endPos = SelfDiagnosticsEventListener.EncodeInBuffer(string.Empty, false, buffer, startPos);
+ byte[] expected = Encoding.UTF8.GetBytes(string.Empty);
+ AssertBufferOutput(expected, buffer, startPos, endPos);
+ }
+
+ [Fact]
+ [Trait("Platform", "Any")]
+ public void SelfDiagnosticsEventListener_EncodeInBuffer_EnoughSpace()
+ {
+ byte[] buffer = new byte[20];
+ int startPos = buffer.Length - Ellipses.Length - 6; // Just enough space for "abc" even if "...\n" needs to be added.
+ int endPos = SelfDiagnosticsEventListener.EncodeInBuffer("abc", false, buffer, startPos);
+
+ // '\n' will be appended to the original string "abc" after EncodeInBuffer is called.
+ // The byte where '\n' will be placed should not be touched within EncodeInBuffer, so it stays as '\0'.
+ byte[] expected = Encoding.UTF8.GetBytes("abc\0");
+ AssertBufferOutput(expected, buffer, startPos, endPos + 1);
+ }
+
+ [Fact]
+ [Trait("Platform", "Any")]
+ public void SelfDiagnosticsEventListener_EncodeInBuffer_NotEnoughSpaceForFullString()
+ {
+ byte[] buffer = new byte[20];
+ int startPos = buffer.Length - Ellipses.Length - 5; // Just not space for "abc" if "...\n" needs to be added.
+
+ // It's a quick estimate by assumption that most Unicode characters takes up to 2 16-bit UTF-16 chars,
+ // which can be up to 4 bytes when encoded in UTF-8.
+ int endPos = SelfDiagnosticsEventListener.EncodeInBuffer("abc", false, buffer, startPos);
+ byte[] expected = Encoding.UTF8.GetBytes("ab...\0");
+ AssertBufferOutput(expected, buffer, startPos, endPos + 1);
+ }
+
+ [Fact]
+ [Trait("Platform", "Any")]
+ public void SelfDiagnosticsEventListener_EncodeInBuffer_NotEvenSpaceForTruncatedString()
+ {
+ byte[] buffer = new byte[20];
+ int startPos = buffer.Length - Ellipses.Length; // Just enough space for "...\n".
+ int endPos = SelfDiagnosticsEventListener.EncodeInBuffer("abc", false, buffer, startPos);
+ byte[] expected = Encoding.UTF8.GetBytes("...\0");
+ AssertBufferOutput(expected, buffer, startPos, endPos + 1);
+ }
+
+ [Fact]
+ [Trait("Platform", "Any")]
+ public void SelfDiagnosticsEventListener_EncodeInBuffer_NotEvenSpaceForTruncationEllipses()
+ {
+ byte[] buffer = new byte[20];
+ int startPos = buffer.Length - Ellipses.Length + 1; // Not enough space for "...\n".
+ int endPos = SelfDiagnosticsEventListener.EncodeInBuffer("abc", false, buffer, startPos);
+ Assert.Equal(startPos, endPos);
+ }
+
+ [Fact]
+ [Trait("Platform", "Any")]
+ public void SelfDiagnosticsEventListener_EncodeInBuffer_IsParameter_EnoughSpace()
+ {
+ byte[] buffer = new byte[20];
+ int startPos = buffer.Length - EllipsesWithBrackets.Length - 6; // Just enough space for "abc" even if "...\n" need to be added.
+ int endPos = SelfDiagnosticsEventListener.EncodeInBuffer("abc", true, buffer, startPos);
+ byte[] expected = Encoding.UTF8.GetBytes("{abc}\0");
+ AssertBufferOutput(expected, buffer, startPos, endPos + 1);
+ }
+
+ [Fact]
+ [Trait("Platform", "Any")]
+ public void SelfDiagnosticsEventListener_EncodeInBuffer_IsParameter_NotEnoughSpaceForFullString()
+ {
+ byte[] buffer = new byte[20];
+ int startPos = buffer.Length - EllipsesWithBrackets.Length - 5; // Just not space for "...\n".
+ int endPos = SelfDiagnosticsEventListener.EncodeInBuffer("abc", true, buffer, startPos);
+ byte[] expected = Encoding.UTF8.GetBytes("{ab...}\0");
+ AssertBufferOutput(expected, buffer, startPos, endPos + 1);
+ }
+
+ [Fact]
+ [Trait("Platform", "Any")]
+ public void SelfDiagnosticsEventListener_EncodeInBuffer_IsParameter_NotEvenSpaceForTruncatedString()
+ {
+ byte[] buffer = new byte[20];
+ int startPos = buffer.Length - EllipsesWithBrackets.Length; // Just enough space for "{...}\n".
+ int endPos = SelfDiagnosticsEventListener.EncodeInBuffer("abc", true, buffer, startPos);
+ byte[] expected = Encoding.UTF8.GetBytes("{...}\0");
+ AssertBufferOutput(expected, buffer, startPos, endPos + 1);
+ }
+
+ [Fact]
+ [Trait("Platform", "Any")]
+ public void SelfDiagnosticsEventListener_EncodeInBuffer_IsParameter_NotEvenSpaceForTruncationEllipses()
+ {
+ byte[] buffer = new byte[20];
+ int startPos = buffer.Length - EllipsesWithBrackets.Length + 1; // Not enough space for "{...}\n".
+ int endPos = SelfDiagnosticsEventListener.EncodeInBuffer("abc", true, buffer, startPos);
+ Assert.Equal(startPos, endPos);
+ }
+
+ private static void AssertFileOutput(string filePath, string eventMessage)
+ {
+ using FileStream file = File.Open(filePath, FileMode.Open, FileAccess.Read, FileShare.ReadWrite | FileShare.Delete);
+ var buffer = new byte[256];
+ file.Read(buffer, 0, buffer.Length);
+ string logMessage = Encoding.UTF8.GetString(buffer);
+ int timestampPrefixLength = "2020-08-14T20:33:24.4788109Z:".Length;
+ string parsedEventMessage = logMessage.Substring(timestampPrefixLength);
+ Assert.StartsWith(eventMessage, parsedEventMessage);
+ Assert.Matches(@"\d{4}-\d{2}-\d{2}T\d{2}:\d{2}:\d{2}\.\d{7}Z:", logMessage.Substring(0, timestampPrefixLength));
+ }
+
+ private static void AssertBufferOutput(byte[] expected, byte[] buffer, int startPos, int endPos)
+ {
+ Assert.Equal(expected.Length, endPos - startPos);
+ for (int i = 0, j = startPos; j < endPos; ++i, ++j)
+ {
+ Assert.Equal(expected[i], buffer[j]);
+ }
+ }
+ }
+}