Merge pull request #1037 from mysql-net/activity-source

Report telemetry with ActivitySource.

Author: bgrainger

docs/content/tutorials/tracing.md

@@ -0,0 +1,26 @@
+---
+date: 2021-10-16
+menu:
+  main:
+    parent: tutorials
+title: Tracing
+weight: 12
+---
+
+# Tracing
+
+MySqlConnector implements `ActivitySource` for tracing its operations. The `ActivitySource` name is `MySqlConnector`.
+
+The available activity names and tags are documented in [issue 1036](https://github.com/mysql-net/MySqlConnector/issues/1036).
+
+## OpenTelemetry
+
+To export traces using OpenTelemetry, install the [OpenTelemetry NuGet package](https://www.nuget.org/packages/OpenTelemetry/) and add code similar to the following:
+
+```csharp
+using var openTelemetry = Sdk.CreateTracerProviderBuilder()
+	.AddSource("MySqlConnector")
+	// add a destination, for example:
+	// .AddZipkinExporter(o => { o.Endpoint = new Uri(...); })
+	// .AddJaegerExporter(o => { o.AgentHost = "..."; o.AgentPort = 6831; })
+```

src/MySqlConnector/Core/CommandExecutor.cs

@@ -1,3 +1,4 @@
+using System.Diagnostics;
 using System.Net.Sockets;
 using MySqlConnector.Logging;
 using MySqlConnector.Protocol.Serialization;
@@ -7,62 +8,71 @@ namespace MySqlConnector.Core;
 
 internal static class CommandExecutor
 {
-	public static async Task<MySqlDataReader> ExecuteReaderAsync(IReadOnlyList<IMySqlCommand> commands, ICommandPayloadCreator payloadCreator, CommandBehavior behavior, IOBehavior ioBehavior, CancellationToken cancellationToken)
+	public static async Task<MySqlDataReader> ExecuteReaderAsync(IReadOnlyList<IMySqlCommand> commands, ICommandPayloadCreator payloadCreator, CommandBehavior behavior, Activity? activity, IOBehavior ioBehavior, CancellationToken cancellationToken)
 	{
-		cancellationToken.ThrowIfCancellationRequested();
-		var commandListPosition = new CommandListPosition(commands);
-		var command = commands[0];
+		try
+		{
+			cancellationToken.ThrowIfCancellationRequested();
+			var commandListPosition = new CommandListPosition(commands);
+			var command = commands[0];
 
-		// pre-requisite: Connection is non-null must be checked before calling this method
-		var connection = command.Connection!;
+			// pre-requisite: Connection is non-null must be checked before calling this method
+			var connection = command.Connection!;
 
-		if (Log.IsTraceEnabled())
-			Log.Trace("Session{0} ExecuteReader {1} CommandCount: {2}", connection.Session.Id, ioBehavior, commands.Count);
+			if (Log.IsTraceEnabled())
+				Log.Trace("Session{0} ExecuteReader {1} CommandCount: {2}", connection.Session.Id, ioBehavior, commands.Count);
 
-		Dictionary<string, CachedProcedure?>? cachedProcedures = null;
-		foreach (var command2 in commands)
-		{
-			if (command2.CommandType == CommandType.StoredProcedure)
+			Dictionary<string, CachedProcedure?>? cachedProcedures = null;
+			foreach (var command2 in commands)
 			{
-				cachedProcedures ??= new();
-				var commandText = command2.CommandText!;
-				if (!cachedProcedures.ContainsKey(commandText))
+				if (command2.CommandType == CommandType.StoredProcedure)
 				{
-					cachedProcedures.Add(commandText, await connection.GetCachedProcedure(commandText, revalidateMissing: false, ioBehavior, cancellationToken).ConfigureAwait(false));
+					cachedProcedures ??= new();
+					var commandText = command2.CommandText!;
+					if (!cachedProcedures.ContainsKey(commandText))
+					{
+						cachedProcedures.Add(commandText, await connection.GetCachedProcedure(commandText, revalidateMissing: false, ioBehavior, cancellationToken).ConfigureAwait(false));
 
-					// because the connection was used to execute a MySqlDataReader with the connection's DefaultCommandTimeout,
-					// we need to reapply the command's CommandTimeout (even if some of the time has elapsed)
-					command.CancellableCommand.ResetCommandTimeout();
+						// because the connection was used to execute a MySqlDataReader with the connection's DefaultCommandTimeout,
+						// we need to reapply the command's CommandTimeout (even if some of the time has elapsed)
+						command.CancellableCommand.ResetCommandTimeout();
+					}
 				}
 			}
-		}
 
-		var writer = new ByteBufferWriter();
-		// cachedProcedures will be non-null if there is a stored procedure, which is also the only time it will be read
-		if (!payloadCreator.WriteQueryCommand(ref commandListPosition, cachedProcedures!, writer))
-			throw new InvalidOperationException("ICommandPayloadCreator failed to write query payload");
+			var writer = new ByteBufferWriter();
+			// cachedProcedures will be non-null if there is a stored procedure, which is also the only time it will be read
+			if (!payloadCreator.WriteQueryCommand(ref commandListPosition, cachedProcedures!, writer))
+				throw new InvalidOperationException("ICommandPayloadCreator failed to write query payload");
 
-		cancellationToken.ThrowIfCancellationRequested();
+			cancellationToken.ThrowIfCancellationRequested();
 
-		using var payload = writer.ToPayloadData();
-		connection.Session.StartQuerying(command.CancellableCommand);
-		command.SetLastInsertedId(-1);
-		try
-		{
-			await connection.Session.SendAsync(payload, ioBehavior, CancellationToken.None).ConfigureAwait(false);
-			return await MySqlDataReader.CreateAsync(commandListPosition, payloadCreator, cachedProcedures, command, behavior, ioBehavior, cancellationToken).ConfigureAwait(false);
-		}
-		catch (MySqlException ex) when (ex.ErrorCode == MySqlErrorCode.QueryInterrupted && cancellationToken.IsCancellationRequested)
-		{
-			Log.Info("Session{0} query was interrupted", connection.Session.Id);
-			throw new OperationCanceledException(ex. Message, ex, cancellationToken);
+			using var payload = writer.ToPayloadData();
+			connection.Session.StartQuerying(command.CancellableCommand);
+			command.SetLastInsertedId(-1);
+			try
+			{
+				await connection.Session.SendAsync(payload, ioBehavior, CancellationToken.None).ConfigureAwait(false);
+				return await MySqlDataReader.CreateAsync(commandListPosition, payloadCreator, cachedProcedures, command, behavior, activity, ioBehavior, cancellationToken).ConfigureAwait(false);
+			}
+			catch (MySqlException ex) when (ex.ErrorCode == MySqlErrorCode.QueryInterrupted && cancellationToken.IsCancellationRequested)
+			{
+				Log.Info("Session{0} query was interrupted", connection.Session.Id);
+				throw new OperationCanceledException(ex.Message, ex, cancellationToken);
+			}
+			catch (Exception ex) when (payload.Span.Length > 4_194_304 && (ex is SocketException or IOException or MySqlProtocolException))
+			{
+				// the default MySQL Server value for max_allowed_packet (in MySQL 5.7) is 4MiB: https://dev.mysql.com/doc/refman/5.7/en/server-system-variables.html#sysvar_max_allowed_packet
+				// use "decimal megabytes" (to round up) when creating the exception message
+				int megabytes = payload.Span.Length / 1_000_000;
+				throw new MySqlException("Error submitting {0}MB packet; ensure 'max_allowed_packet' is greater than {0}MB.".FormatInvariant(megabytes), ex);
+			}
 		}
-		catch (Exception ex) when (payload.Span.Length > 4_194_304 && (ex is SocketException or IOException or MySqlProtocolException))
+		catch (Exception ex) when (activity is { IsAllDataRequested: true })
 		{
-			// the default MySQL Server value for max_allowed_packet (in MySQL 5.7) is 4MiB: https://dev.mysql.com/doc/refman/5.7/en/server-system-variables.html#sysvar_max_allowed_packet
-			// use "decimal megabytes" (to round up) when creating the exception message
-			int megabytes = payload.Span.Length / 1_000_000;
-			throw new MySqlException("Error submitting {0}MB packet; ensure 'max_allowed_packet' is greater than {0}MB.".FormatInvariant(megabytes), ex);
+			activity.SetException(ex);
+			activity.Stop();
+			throw;
 		}
 	}
 

src/MySqlConnector/Core/ResultSet.cs

@@ -1,4 +1,5 @@
 using System.Buffers;
+using System.Diagnostics;
 using System.Globalization;
 using System.Runtime.ExceptionServices;
 using MySqlConnector.Protocol;
@@ -41,7 +42,6 @@ public async Task ReadResultSetHeaderAsync(IOBehavior ioBehavior)
 			while (true)
 			{
 				var payload = await Session.ReceiveReplyAsync(ioBehavior, CancellationToken.None).ConfigureAwait(false);
-
 				var firstByte = payload.HeaderByte;
 				if (firstByte == OkPayload.Signature)
 				{
@@ -112,7 +112,7 @@ public async Task ReadResultSetHeaderAsync(IOBehavior ioBehavior)
 				}
 				else
 				{
-					int ReadColumnCount(ReadOnlySpan<byte> span)
+					static int ReadColumnCount(ReadOnlySpan<byte> span)
 					{
 						var reader = new ByteArrayReader(span);
 						var columnCount_ = (int) reader.ReadLengthEncodedInteger();
@@ -154,6 +154,8 @@ int ReadColumnCount(ReadOnlySpan<byte> span)
 						ContainsCommandParameters = true;
 					WarningCount = 0;
 					State = ResultSetState.ReadResultSetHeader;
+					if (DataReader.Activity is { IsAllDataRequested: true })
+						DataReader.Activity.AddEvent(new ActivityEvent("read-result-set-header"));
 					break;
 				}
 			}

src/MySqlConnector/Core/ServerSession.cs

@@ -38,6 +38,10 @@ public ServerSession(ConnectionPool? pool, int poolGeneration, int id)
 		PoolGeneration = poolGeneration;
 		HostName = "";
 		m_logArguments = new object?[] { "{0}".FormatInvariant(Id), null };
+		m_activityTags = new ActivityTagsCollection
+		{
+			{ ActivitySourceHelper.DatabaseSystemTagName, ActivitySourceHelper.DatabaseSystemValue },
+		};
 		Log.Trace("Session{0} created new session", m_logArguments);
 	}
 
@@ -59,6 +63,7 @@ public ServerSession(ConnectionPool? pool, int poolGeneration, int id)
 	public bool SupportsDeprecateEof => m_supportsDeprecateEof;
 	public bool SupportsSessionTrack => m_supportsSessionTrack;
 	public bool ProcAccessDenied { get; set; }
+	public ICollection<KeyValuePair<string, object?>> ActivityTags => m_activityTags;
 
 #if NETCOREAPP2_1_OR_GREATER || NETSTANDARD2_1_OR_GREATER
 	public ValueTask ReturnToPoolAsync(IOBehavior ioBehavior, MySqlConnection? owningConnection)
@@ -330,6 +335,19 @@ public void FinishQuerying()
 
 	public void SetTimeout(int timeoutMilliseconds) => m_payloadHandler!.ByteHandler.RemainingTimeout = timeoutMilliseconds;
 
+	public Activity? StartActivity(string name, string? tagName1 = null, object? tagValue1 = null)
+	{
+		var activity = ActivitySourceHelper.StartActivity(name, m_activityTags);
+		if (activity is { IsAllDataRequested: true })
+		{
+			if (DatabaseOverride is not null)
+				activity.SetTag(ActivitySourceHelper.DatabaseNameTagName, DatabaseOverride);
+			if (tagName1 is not null)
+				activity.SetTag(tagName1, tagValue1);
+		}
+		return activity;
+	}
+
 	public async Task DisposeAsync(IOBehavior ioBehavior, CancellationToken cancellationToken)
 	{
 		if (m_payloadHandler is not null)
@@ -515,6 +533,12 @@ public async Task DisposeAsync(IOBehavior ioBehavior, CancellationToken cancella
 				await GetRealServerDetailsAsync(ioBehavior, CancellationToken.None).ConfigureAwait(false);
 
 			m_payloadHandler.ByteHandler.RemainingTimeout = Constants.InfiniteTimeout;
+
+			m_activityTags.Add(ActivitySourceHelper.DatabaseConnectionIdTagName, ConnectionId.ToString(CultureInfo.InvariantCulture));
+			m_activityTags.Add(ActivitySourceHelper.DatabaseConnectionStringTagName, cs.ConnectionStringBuilder.GetConnectionString(cs.ConnectionStringBuilder.PersistSecurityInfo));
+			m_activityTags.Add(ActivitySourceHelper.DatabaseUserTagName, cs.UserID);
+			if (cs.Database.Length != 0)
+				m_activityTags.Add(ActivitySourceHelper.DatabaseNameTagName, cs.Database);
 		}
 		catch (ArgumentException ex)
 		{
@@ -1032,6 +1056,14 @@ private async Task<bool> OpenTcpSocketAsync(ConnectionSettings cs, ILoadBalancer
 					m_socket.NoDelay = true;
 					m_stream = m_tcpClient.GetStream();
 					m_socket.SetKeepAlive(cs.Keepalive);
+
+					m_activityTags.Add(ActivitySourceHelper.NetTransportTagName, ActivitySourceHelper.NetTransportTcpIpValue);
+					var ipAddressString = ipAddress.ToString();
+					m_activityTags.Add(ActivitySourceHelper.NetPeerIpTagName, ipAddressString);
+					if (ipAddressString != hostName)
+						m_activityTags.Add(ActivitySourceHelper.NetPeerNameTagName, hostName);
+					if (cs.Port != 3306)
+						m_activityTags.Add(ActivitySourceHelper.NetPeerPortTagName, cs.Port.ToString(CultureInfo.InvariantCulture));
 				}
 				catch (ObjectDisposedException) when (cancellationToken.IsCancellationRequested)
 				{
@@ -1089,6 +1121,9 @@ private async Task<bool> OpenUnixSocketAsync(ConnectionSettings cs, IOBehavior i
 			m_socket = socket;
 			m_stream = new NetworkStream(socket);
 
+			m_activityTags.Add(ActivitySourceHelper.NetTransportTagName, ActivitySourceHelper.NetTransportUnixValue);
+			m_activityTags.Add(ActivitySourceHelper.NetPeerNameTagName, cs.UnixSocket);
+
 			lock (m_lock)
 				m_state = State.Connected;
 			return true;
@@ -1139,6 +1174,10 @@ private async Task<bool> OpenNamedPipeAsync(ConnectionSettings cs, int startTick
 		{
 			m_stream = namedPipeStream;
 
+			// see https://docs.microsoft.com/en-us/windows/win32/ipc/pipe-names for pipe name format
+			m_activityTags.Add(ActivitySourceHelper.NetTransportTagName, ActivitySourceHelper.NetTransportNamedPipeValue);
+			m_activityTags.Add(ActivitySourceHelper.NetPeerNameTagName, @"\\" + cs.HostNames![0] + @"\pipe\" + cs.PipeName);
+
 			lock (m_lock)
 				m_state = State.Connected;
 			return true;
@@ -1790,5 +1829,6 @@ protected override void OnStatementBegin(int index)
 	bool m_supportsSessionTrack;
 	CharacterSet m_characterSet;
 	PayloadData m_setNamesPayload;
+	ActivityTagsCollection m_activityTags;
 	Dictionary<string, PreparedStatements>? m_preparedStatements;
 }

src/MySqlConnector/MySqlBatch.cs

@@ -157,7 +157,7 @@ private Task<MySqlDataReader> ExecuteReaderAsync(CommandBehavior behavior, IOBeh
 		var payloadCreator = Connection!.Session.SupportsComMulti ? BatchedCommandPayloadCreator.Instance :
 			IsPrepared ? SingleCommandPayloadCreator.Instance :
 			ConcatenatedCommandPayloadCreator.Instance;
-		return CommandExecutor.ExecuteReaderAsync(BatchCommands!.Commands, payloadCreator, behavior, ioBehavior, cancellationToken);
+		return CommandExecutor.ExecuteReaderAsync(BatchCommands!.Commands, payloadCreator, behavior, default, ioBehavior, cancellationToken);
 	}
 
 #if NET6_0_OR_GREATER

src/MySqlConnector/MySqlCommand.cs

@@ -1,3 +1,4 @@
+using System.Diagnostics;
 using System.Diagnostics.CodeAnalysis;
 using MySqlConnector.Core;
 using MySqlConnector.Logging;
@@ -130,6 +131,7 @@ public override void Prepare()
 	bool IMySqlCommand.AllowUserVariables => AllowUserVariables;
 
 	internal bool AllowUserVariables { get; set; }
+	internal bool NoActivity { get; set; }
 
 	private Task PrepareAsync(IOBehavior ioBehavior, CancellationToken cancellationToken)
 	{
@@ -315,8 +317,10 @@ internal Task<MySqlDataReader> ExecuteReaderNoResetTimeoutAsync(CommandBehavior
 		if (!IsValid(out var exception))
 			return Utility.TaskFromException<MySqlDataReader>(exception);
 
+		var activity = NoActivity ? null : Connection!.Session.StartActivity(ActivitySourceHelper.ExecuteActivityName,
+			ActivitySourceHelper.DatabaseStatementTagName, CommandText);
 		m_commandBehavior = behavior;
-		return CommandExecutor.ExecuteReaderAsync(new IMySqlCommand[] { this }, SingleCommandPayloadCreator.Instance, behavior, ioBehavior, cancellationToken);
+		return CommandExecutor.ExecuteReaderAsync(new IMySqlCommand[] { this }, SingleCommandPayloadCreator.Instance, behavior, activity, ioBehavior, cancellationToken);
 	}
 
 	public MySqlCommand Clone() => new(this);