feat(csharp/src/Drivers): instrument tracing exporters for BigQuery/Apache drivers

csharp/Apache.Arrow.Adbc.sln

@@ -58,6 +58,14 @@ Project("{2150E333-8FDC-42A3-9474-1A3956D46DE8}") = "Exporters", "Exporters", "{
 EndProject
 Project("{FAE04EC0-301F-11D3-BF4B-00C04F79EFBC}") = "Apache.Arrow.Adbc.Tests.Telemetry.Traces.Exporters", "test\Telemetry\Traces\Exporters\Apache.Arrow.Adbc.Tests.Telemetry.Traces.Exporters.csproj", "{1558BC4B-6E76-434B-8877-6C49B1460544}"
 EndProject
+Project("{2150E333-8FDC-42A3-9474-1A3956D46DE8}") = "Listeners", "Listeners", "{53C45FD3-7277-49FA-AEEB-DF8F2386ECAA}"
+EndProject
+Project("{2150E333-8FDC-42A3-9474-1A3956D46DE8}") = "Listeners", "Listeners", "{1C530561-1008-4F39-B437-15B2FD59EAC9}"
+EndProject
+Project("{FAE04EC0-301F-11D3-BF4B-00C04F79EFBC}") = "Apache.Arrow.Adbc.Tests.Telemetry.Traces.Listeners", "test\Telemetry\Traces\Listeners\Apache.Arrow.Adbc.Tests.Telemetry.Traces.Listeners.csproj", "{9CE9106B-ACBB-54C1-DE57-370E5CF09363}"
+EndProject
+Project("{FAE04EC0-301F-11D3-BF4B-00C04F79EFBC}") = "Apache.Arrow.Adbc.Telemetry.Traces.Listeners", "src\Telemetry\Traces\Listeners\Apache.Arrow.Adbc.Telemetry.Traces.Listeners.csproj", "{4D5ADA1A-2DEE-5860-2351-221090CF4442}"
+EndProject
 Global
 	GlobalSection(SolutionConfigurationPlatforms) = preSolution
 		Debug|Any CPU = Debug|Any CPU
@@ -136,6 +144,14 @@ Global
 		{1558BC4B-6E76-434B-8877-6C49B1460544}.Debug|Any CPU.Build.0 = Debug|Any CPU
 		{1558BC4B-6E76-434B-8877-6C49B1460544}.Release|Any CPU.ActiveCfg = Release|Any CPU
 		{1558BC4B-6E76-434B-8877-6C49B1460544}.Release|Any CPU.Build.0 = Release|Any CPU
+		{9CE9106B-ACBB-54C1-DE57-370E5CF09363}.Debug|Any CPU.ActiveCfg = Debug|Any CPU
+		{9CE9106B-ACBB-54C1-DE57-370E5CF09363}.Debug|Any CPU.Build.0 = Debug|Any CPU
+		{9CE9106B-ACBB-54C1-DE57-370E5CF09363}.Release|Any CPU.ActiveCfg = Release|Any CPU
+		{9CE9106B-ACBB-54C1-DE57-370E5CF09363}.Release|Any CPU.Build.0 = Release|Any CPU
+		{4D5ADA1A-2DEE-5860-2351-221090CF4442}.Debug|Any CPU.ActiveCfg = Debug|Any CPU
+		{4D5ADA1A-2DEE-5860-2351-221090CF4442}.Debug|Any CPU.Build.0 = Debug|Any CPU
+		{4D5ADA1A-2DEE-5860-2351-221090CF4442}.Release|Any CPU.ActiveCfg = Release|Any CPU
+		{4D5ADA1A-2DEE-5860-2351-221090CF4442}.Release|Any CPU.Build.0 = Release|Any CPU
 	EndGlobalSection
 	GlobalSection(SolutionProperties) = preSolution
 		HideSolutionNode = FALSE
@@ -164,6 +180,10 @@ Global
 		{B74532A7-8A78-4AD9-9B2E-584765491E48} = {9FE39661-2A39-4E9F-A5F2-11FB0D54CB42}
 		{43910445-FEC4-4AEC-A698-6A48327600C3} = {B74532A7-8A78-4AD9-9B2E-584765491E48}
 		{1558BC4B-6E76-434B-8877-6C49B1460544} = {43910445-FEC4-4AEC-A698-6A48327600C3}
+		{53C45FD3-7277-49FA-AEEB-DF8F2386ECAA} = {B74532A7-8A78-4AD9-9B2E-584765491E48}
+		{1C530561-1008-4F39-B437-15B2FD59EAC9} = {22EF23A3-1566-446F-B696-9323F3B6F56C}
+		{9CE9106B-ACBB-54C1-DE57-370E5CF09363} = {53C45FD3-7277-49FA-AEEB-DF8F2386ECAA}
+		{4D5ADA1A-2DEE-5860-2351-221090CF4442} = {1C530561-1008-4F39-B437-15B2FD59EAC9}
 	EndGlobalSection
 	GlobalSection(ExtensibilityGlobals) = postSolution
 		SolutionGuid = {4795CF16-0FDB-4BE0-9768-5CF31564DC03}

csharp/src/Drivers/Apache/Apache.Arrow.Adbc.Drivers.Apache.csproj

@@ -14,6 +14,7 @@
 
   <ItemGroup>
     <ProjectReference Include="..\..\Apache.Arrow.Adbc\Apache.Arrow.Adbc.csproj" />
+    <ProjectReference Include="..\..\Telemetry\Traces\Listeners\Apache.Arrow.Adbc.Telemetry.Traces.Listeners.csproj" />
   </ItemGroup>
 
 </Project>

csharp/src/Drivers/Apache/Hive2/HiveServer2Connection.cs

@@ -27,6 +27,8 @@
 using System.Threading.Tasks;
 using Apache.Arrow.Adbc.Drivers.Apache.Thrift;
 using Apache.Arrow.Adbc.Extensions;
+using Apache.Arrow.Adbc.Telemetry.Traces.Listeners;
+using Apache.Arrow.Adbc.Telemetry.Traces.Listeners.FileListener;
 using Apache.Arrow.Adbc.Tracing;
 using Apache.Arrow.Ipc;
 using Apache.Arrow.Types;
@@ -49,6 +51,9 @@ internal abstract class HiveServer2Connection : TracingConnection
         private readonly Lazy<string> _vendorVersion;
         private readonly Lazy<string> _vendorName;
         private bool _isDisposed;
+        // Note: this needs to be set before the constructor runs
+        private readonly string _traceInstanceId = Guid.NewGuid().ToString("N");
+        private readonly FileActivityListener? _fileActivityListener;
 
         readonly AdbcInfoCode[] infoSupportedCodes = [
             AdbcInfoCode.DriverName,
@@ -278,6 +283,8 @@ internal HiveServer2Connection(IReadOnlyDictionary<string, string> properties)
         {
             Properties = properties;
 
+            TryInitTracerProvider(out _fileActivityListener);
+
             // Note: "LazyThreadSafetyMode.PublicationOnly" is thread-safe initialization where
             // the first successful thread sets the value. If an exception is thrown, initialization
             // will retry until it successfully returns a value without an exception.
@@ -294,6 +301,31 @@ internal HiveServer2Connection(IReadOnlyDictionary<string, string> properties)
             }
         }
 
+        private bool TryInitTracerProvider(out FileActivityListener? fileActivityListener)
+        {
+            Properties.TryGetValue(ListenersOptions.Exporter, out string? exporterOption);
+            // This listener will only listen for activity from this specific connection instance.
+            bool shouldListenTo(ActivitySource source) => source.Tags?.Any(t => ReferenceEquals(t.Key, _traceInstanceId)) == true;
+            return FileActivityListener.TryActivateFileListener(AssemblyName, exporterOption, out fileActivityListener, shouldListenTo: shouldListenTo);
+        }
+
+        public override IEnumerable<KeyValuePair<string, object?>>? GetActivitySourceTags(IReadOnlyDictionary<string, string> properties)
+        {
+            IEnumerable<KeyValuePair<string, object?>>? tags = base.GetActivitySourceTags(properties);
+            tags ??= [];
+            tags = tags.Concat([new(_traceInstanceId, null)]);
+            return tags;
+        }
+
+        /// <summary>
+        /// Conditional used to determines if it is safe to trace
+        /// </summary>
+        /// <remarks>
+        /// It is safe to write to some output types (ie, files) but not others (ie, a shared resource).
+        /// </remarks>
+        /// <returns></returns>
+        internal bool IsSafeToTrace => _fileActivityListener != null;
+
         internal TCLIService.IAsync Client
         {
             get { return _client ?? throw new InvalidOperationException("connection not open"); }
@@ -732,6 +764,7 @@ protected override void Dispose(bool disposing)
             if (!_isDisposed && disposing)
             {
                 DisposeClient();
+                _fileActivityListener?.Dispose();
                 _isDisposed = true;
             }
             base.Dispose(disposing);
@@ -1540,7 +1573,7 @@ public override IArrowArrayStream GetInfo(IReadOnlyList<AdbcInfoCode> codes)
                             nullCount++;
                             break;
                     }
-                    ActivityExtensions.AddTag(activity, tagKey, tagValue);
+                    Tracing.ActivityExtensions.AddTag(activity, tagKey, tagValue);
                 }
 
                 StructType entryType = new StructType(

csharp/src/Drivers/Apache/Hive2/README.md

@@ -149,3 +149,34 @@ The Collector can be configure to receive trace messages from the driver and exp
 Ensure to set the [environment variable](https://opentelemetry.io/docs/specs/otel/protocol/exporter/) `OTEL_EXPORTER_OTLP_INSECURE` to `true`, in this scenario.
 
 Ensure to follow [Collector configuration best practices](https://opentelemetry.io/docs/security/config-best-practices/).
+
+## Tracing
+
+### Tracing Exporters
+
+To enable tracing messages to be observed, a tracing exporter needs to be activated.
+Use either the environment variable `OTEL_TRACES_EXPORTER` or the parameter `adbc.traces.exporter` to select one of the
+supported exporters. The parameter has precedence over the environment variable. The parameter must be set before
+the connection is initialized.
+
+The following exporters are supported:
+
+| Exporter | Description |
+| --- | --- |
+| `adbcfile` | Exports traces to rotating files in a folder. |
+
+#### File Exporter (adbcfile)
+
+Rotating trace files are written to a folder. The file names are created with the following pattern:
+`apache.arrow.adbc.drivers.bigquery-<YYYY-MM-DD-HH-mm-ss-fff>-<process-id>.log`.
+
+The folder used depends on the platform.
+
+| Platform | Folder |
+| --- | --- |
+| Windows | `%LOCALAPPDATA%/Apache.Arrow.Adbc/Traces` |
+| macOS   | `$HOME/Library/Application Support/Apache.Arrow.Adbc/Traces` |
+| Linux   | `$HOME/.local/share/Apache.Arrow.Adbc/Traces` |
+
+By default, up to 999 files of maximum size 1024 KB are written to
+the trace folder.

csharp/src/Drivers/Apache/Impala/README.md

@@ -139,3 +139,34 @@ The Collector can be configure to receive trace messages from the driver and exp
 Ensure to set the [environment variable](https://opentelemetry.io/docs/specs/otel/protocol/exporter/) `OTEL_EXPORTER_OTLP_INSECURE` to `true`, in this scenario.
 
 Ensure to follow [Collector configuration best practices](https://opentelemetry.io/docs/security/config-best-practices/).
+
+## Tracing
+
+### Tracing Exporters
+
+To enable tracing messages to be observed, a tracing exporter needs to be activated.
+Use either the environment variable `OTEL_TRACES_EXPORTER` or the parameter `adbc.traces.exporter` to select one of the
+supported exporters. The parameter has precedence over the environment variable. The parameter must be set before
+the connection is initialized.
+
+The following exporters are supported:
+
+| Exporter | Description |
+| --- | --- |
+| `adbcfile` | Exports traces to rotating files in a folder. |
+
+#### File Exporter (adbcfile)
+
+Rotating trace files are written to a folder. The file names are created with the following pattern:
+`apache.arrow.adbc.drivers.bigquery-<YYYY-MM-DD-HH-mm-ss-fff>-<process-id>.log`.
+
+The folder used depends on the platform.
+
+| Platform | Folder |
+| --- | --- |
+| Windows | `%LOCALAPPDATA%/Apache.Arrow.Adbc/Traces` |
+| macOS   | `$HOME/Library/Application Support/Apache.Arrow.Adbc/Traces` |
+| Linux   | `$HOME/.local/share/Apache.Arrow.Adbc/Traces` |
+
+By default, up to 999 files of maximum size 1024 KB are written to
+the trace folder.

csharp/src/Drivers/Apache/Spark/README.md

@@ -149,3 +149,34 @@ The Collector can be configure to receive trace messages from the driver and exp
 Ensure to set the [environment variable](https://opentelemetry.io/docs/specs/otel/protocol/exporter/) `OTEL_EXPORTER_OTLP_INSECURE` to `true`, in this scenario.
 
 Ensure to follow [Collector configuration best practices](https://opentelemetry.io/docs/security/config-best-practices/).
+
+## Tracing
+
+### Tracing Exporters
+
+To enable tracing messages to be observed, a tracing exporter needs to be activated.
+Use either the environment variable `OTEL_TRACES_EXPORTER` or the parameter `adbc.traces.exporter` to select one of the
+supported exporters. The parameter has precedence over the environment variable. The parameter must be set before
+the connection is initialized.
+
+The following exporters are supported:
+
+| Exporter | Description |
+| --- | --- |
+| `adbcfile` | Exports traces to rotating files in a folder. |
+
+#### File Exporter (adbcfile)
+
+Rotating trace files are written to a folder. The file names are created with the following pattern:
+`apache.arrow.adbc.drivers.bigquery-<YYYY-MM-DD-HH-mm-ss-fff>-<process-id>.log`.
+
+The folder used depends on the platform.
+
+| Platform | Folder |
+| --- | --- |
+| Windows | `%LOCALAPPDATA%/Apache.Arrow.Adbc/Traces` |
+| macOS   | `$HOME/Library/Application Support/Apache.Arrow.Adbc/Traces` |
+| Linux   | `$HOME/.local/share/Apache.Arrow.Adbc/Traces` |
+
+By default, up to 999 files of maximum size 1024 KB are written to
+the trace folder.

csharp/src/Drivers/BigQuery/Apache.Arrow.Adbc.Drivers.BigQuery.csproj

@@ -13,6 +13,7 @@
   </ItemGroup>
   <ItemGroup>
     <ProjectReference Include="..\..\Apache.Arrow.Adbc\Apache.Arrow.Adbc.csproj" />
+    <ProjectReference Include="..\..\Telemetry\Traces\Listeners\Apache.Arrow.Adbc.Telemetry.Traces.Listeners.csproj" />
   </ItemGroup>
   <ItemGroup>
     <Content Include="readme.md">

csharp/src/Drivers/BigQuery/BigQueryConnection.cs

@@ -25,6 +25,8 @@
 using System.Text.RegularExpressions;
 using System.Threading.Tasks;
 using Apache.Arrow.Adbc.Extensions;
+using Apache.Arrow.Adbc.Telemetry.Traces.Listeners;
+using Apache.Arrow.Adbc.Telemetry.Traces.Listeners.FileListener;
 using Apache.Arrow.Adbc.Tracing;
 using Apache.Arrow.Ipc;
 using Apache.Arrow.Types;
@@ -45,6 +47,9 @@ public class BigQueryConnection : TracingConnection, ITokenProtectedResource
         bool includePublicProjectIds = false;
         const string infoDriverName = "ADBC BigQuery Driver";
         const string infoVendorName = "BigQuery";
+        // Note: this needs to be set before the constructor runs
+        private readonly string _traceInstanceId = Guid.NewGuid().ToString("N");
+        private readonly FileActivityListener? _fileActivityListener;
 
         private readonly string infoDriverArrowVersion = BigQueryUtils.GetAssemblyVersion(typeof(IArrowArray));
 
@@ -66,6 +71,8 @@ public BigQueryConnection(IReadOnlyDictionary<string, string> properties) : base
                 this.properties = properties.ToDictionary(k => k.Key, v => v.Value);
             }
 
+            TryInitTracerProvider(out _fileActivityListener);
+
             // add the default value for now and set to true until C# has a BigDecimal
             this.properties[BigQueryParameters.LargeDecimalsAsString] = BigQueryConstants.TreatLargeDecimalAsString;
             this.httpClient = new HttpClient();
@@ -85,6 +92,31 @@ public BigQueryConnection(IReadOnlyDictionary<string, string> properties) : base
             }
         }
 
+        private bool TryInitTracerProvider(out FileActivityListener? fileActivityListener)
+        {
+            properties.TryGetValue(ListenersOptions.Exporter, out string? exporterOption);
+            // This listener will only listen for activity from this specific connection instance.
+            bool shouldListenTo(ActivitySource source) => source.Tags?.Any(t => ReferenceEquals(t.Key, _traceInstanceId)) == true;
+            return FileActivityListener.TryActivateFileListener(AssemblyName, exporterOption, out fileActivityListener, shouldListenTo: shouldListenTo);
+        }
+
+        public override IEnumerable<KeyValuePair<string, object?>>? GetActivitySourceTags(IReadOnlyDictionary<string, string> properties)
+        {
+            IEnumerable<KeyValuePair<string, object?>>? tags = base.GetActivitySourceTags(properties);
+            tags ??= [];
+            tags = tags.Concat([new(_traceInstanceId, null)]);
+            return tags;
+        }
+
+        /// <summary>
+        /// Conditional used to determines if it is safe to trace
+        /// </summary>
+        /// <remarks>
+        /// It is safe to write to some output types (ie, files) but not others (ie, a shared resource).
+        /// </remarks>
+        /// <returns></returns>
+        internal bool IsSafeToTrace => _fileActivityListener != null;
+
         /// <summary>
         /// The function to call when updating the token.
         /// </summary>
@@ -470,7 +502,7 @@ internal void UpdateClientToken()
 
             return this.TraceActivity(activity =>
             {
-                activity?.AddConditionalTag(SemanticConventions.Db.Query.Text, sql, BigQueryUtils.IsSafeToTrace());
+                activity?.AddConditionalTag(SemanticConventions.Db.Query.Text, sql, IsSafeToTrace);
 
                 Func<Task<BigQueryResults?>> func = () => Client.ExecuteQueryAsync(sql, parameters ?? Enumerable.Empty<BigQueryParameter>(), queryOptions, resultsOptions);
                 BigQueryResults? result = ExecuteWithRetriesAsync<BigQueryResults?>(func, activity).GetAwaiter().GetResult();
@@ -1273,6 +1305,7 @@ public override void Dispose()
             Client?.Dispose();
             Client = null;
             this.httpClient?.Dispose();
+            this._fileActivityListener?.Dispose();
         }
 
         private static Regex sanitizedInputRegex = new Regex("^[a-zA-Z0-9_-]+");

csharp/src/Drivers/BigQuery/BigQueryStatement.cs

@@ -91,7 +91,7 @@ private async Task<QueryResult> ExecuteQueryInternalAsync()
             {
                 QueryOptions queryOptions = ValidateOptions(activity);
 
-                activity?.AddConditionalTag(SemanticConventions.Db.Query.Text, SqlQuery, BigQueryUtils.IsSafeToTrace());
+                activity?.AddConditionalTag(SemanticConventions.Db.Query.Text, SqlQuery, this.bigQueryConnection.IsSafeToTrace);
 
                 BigQueryJob job = await Client.CreateQueryJobAsync(SqlQuery, null, queryOptions);
 
@@ -216,7 +216,7 @@ private async Task<IEnumerable<IArrowReader>> GetArrowReaders(
             ReadSession rrs = await bigQueryReadClient.CreateReadSessionAsync("projects/" + projectId, rs, maxStreamCount);
 
             var readers = rrs.Streams
-                             .Select(s => ReadChunk(bigQueryReadClient, s.Name, activity))
+                             .Select(s => ReadChunk(bigQueryReadClient, s.Name, activity, this.bigQueryConnection.IsSafeToTrace))
                              .Where(chunk => chunk != null)
                              .Cast<IArrowReader>();
 
@@ -242,7 +242,7 @@ private async Task<UpdateResult> ExecuteUpdateInternalAsync()
                     activity?.AddBigQueryParameterTag(BigQueryParameters.GetQueryResultsOptionsTimeout, seconds);
                 }
 
-                activity?.AddConditionalTag(SemanticConventions.Db.Query.Text, SqlQuery, BigQueryUtils.IsSafeToTrace());
+                activity?.AddConditionalTag(SemanticConventions.Db.Query.Text, SqlQuery, this.bigQueryConnection.IsSafeToTrace);
 
                 // Cannot set destination table in jobs with DDL statements, otherwise an error will be prompted
                 Func<Task<BigQueryResults?>> func = () => Client.ExecuteQueryAsync(SqlQuery, null, null, getQueryResultsOptions);
@@ -340,11 +340,11 @@ private IArrowType GetType(TableFieldSchema field, IArrowType type)
             return type;
         }
 
-        private static IArrowReader? ReadChunk(BigQueryReadClient client, string streamName, Activity? activity)
+        private static IArrowReader? ReadChunk(BigQueryReadClient client, string streamName, Activity? activity, bool isSafeToTrace)
         {
             // Ideally we wouldn't need to indirect through a stream, but the necessary APIs in Arrow
             // are internal. (TODO: consider changing Arrow).
-            activity?.AddConditionalBigQueryTag("read_stream", streamName, BigQueryUtils.IsSafeToTrace());
+            activity?.AddConditionalBigQueryTag("read_stream", streamName, isSafeToTrace);
             BigQueryReadClient.ReadRowsStream readRowsStream = client.ReadRows(new ReadRowsRequest { ReadStream = streamName });
             IAsyncEnumerator<ReadRowsResponse> enumerator = readRowsStream.GetResponseStream().GetAsyncEnumerator();
 

csharp/src/Drivers/BigQuery/BigQueryUtils.cs

@@ -42,18 +42,5 @@ public static bool TokenRequiresUpdate(Exception ex)
         internal static string GetAssemblyName(Type type) => type.Assembly.GetName().Name!;
 
         internal static string GetAssemblyVersion(Type type) => FileVersionInfo.GetVersionInfo(type.Assembly.Location).ProductVersion ?? string.Empty;
-
-        /// <summary>
-        /// Conditional used to determines if it is safe to trace
-        /// </summary>
-        /// <remarks>
-        /// It is safe to write to some output types (ie, files) but not others (ie, a shared resource).
-        /// </remarks>
-        /// <returns></returns>
-        internal static bool IsSafeToTrace()
-        {
-            // TODO: Add logic to determine if a file writer is listening
-            return false;
-        }
     }
 }