diff --git a/csharp/Apache.Arrow.Adbc.sln b/csharp/Apache.Arrow.Adbc.sln index 7bb6260dc0..e79981db6b 100644 --- a/csharp/Apache.Arrow.Adbc.sln +++ b/csharp/Apache.Arrow.Adbc.sln @@ -42,6 +42,22 @@ Project("{FAE04EC0-301F-11D3-BF4B-00C04F79EFBC}") = "Apache.Arrow.Adbc.Drivers.D EndProject Project("{FAE04EC0-301F-11D3-BF4B-00C04F79EFBC}") = "Apache.Arrow.Adbc.Tests.Drivers.Databricks", "test\Drivers\Databricks\Apache.Arrow.Adbc.Tests.Drivers.Databricks.csproj", "{BA07EB2C-5246-EB72-153C-493C7E7412D2}" EndProject +Project("{2150E333-8FDC-42A3-9474-1A3956D46DE8}") = "Telemetry", "Telemetry", "{02EA681E-C7D8-13C7-8484-4AC65E1B71E8}" +EndProject +Project("{2150E333-8FDC-42A3-9474-1A3956D46DE8}") = "Traces", "Traces", "{22EF23A3-1566-446F-B696-9323F3B6F56C}" +EndProject +Project("{2150E333-8FDC-42A3-9474-1A3956D46DE8}") = "Exporters", "Exporters", "{4A0C233A-90A0-42ED-8D8A-F21E7D5BB7EB}" +EndProject +Project("{FAE04EC0-301F-11D3-BF4B-00C04F79EFBC}") = "Apache.Arrow.Adbc.Telemetry.Traces.Exporters", "src\Telemetry\Traces\Exporters\Apache.Arrow.Adbc.Telemetry.Traces.Exporters.csproj", "{85D2EC10-F461-02F4-83B6-E4FC320C244F}" +EndProject +Project("{2150E333-8FDC-42A3-9474-1A3956D46DE8}") = "Telemetry", "Telemetry", "{9FE39661-2A39-4E9F-A5F2-11FB0D54CB42}" +EndProject +Project("{2150E333-8FDC-42A3-9474-1A3956D46DE8}") = "Traces", "Traces", "{B74532A7-8A78-4AD9-9B2E-584765491E48}" +EndProject +Project("{2150E333-8FDC-42A3-9474-1A3956D46DE8}") = "Exporters", "Exporters", "{43910445-FEC4-4AEC-A698-6A48327600C3}" +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 Global GlobalSection(SolutionConfigurationPlatforms) = preSolution Debug|Any CPU = Debug|Any CPU @@ -112,6 +128,14 @@ Global {BA07EB2C-5246-EB72-153C-493C7E7412D2}.Debug|Any CPU.Build.0 = Debug|Any CPU {BA07EB2C-5246-EB72-153C-493C7E7412D2}.Release|Any CPU.ActiveCfg = Release|Any CPU {BA07EB2C-5246-EB72-153C-493C7E7412D2}.Release|Any CPU.Build.0 = Release|Any CPU + {85D2EC10-F461-02F4-83B6-E4FC320C244F}.Debug|Any CPU.ActiveCfg = Debug|Any CPU + {85D2EC10-F461-02F4-83B6-E4FC320C244F}.Debug|Any CPU.Build.0 = Debug|Any CPU + {85D2EC10-F461-02F4-83B6-E4FC320C244F}.Release|Any CPU.ActiveCfg = Release|Any CPU + {85D2EC10-F461-02F4-83B6-E4FC320C244F}.Release|Any CPU.Build.0 = Release|Any CPU + {1558BC4B-6E76-434B-8877-6C49B1460544}.Debug|Any CPU.ActiveCfg = Debug|Any CPU + {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 EndGlobalSection GlobalSection(SolutionProperties) = preSolution HideSolutionNode = FALSE @@ -133,6 +157,13 @@ Global {BAF2CF14-BA77-429E-AF54-A34B978E9F5C} = {5BD04C26-CE52-4893-8C1A-479705195CEF} {25042111-6B86-8B75-7EF6-5BFAA36F72B1} = {FEB257A0-4FD3-495E-9A47-9E1649755445} {BA07EB2C-5246-EB72-153C-493C7E7412D2} = {C7290227-E925-47E7-8B6B-A8B171645D58} + {22EF23A3-1566-446F-B696-9323F3B6F56C} = {02EA681E-C7D8-13C7-8484-4AC65E1B71E8} + {4A0C233A-90A0-42ED-8D8A-F21E7D5BB7EB} = {22EF23A3-1566-446F-B696-9323F3B6F56C} + {85D2EC10-F461-02F4-83B6-E4FC320C244F} = {4A0C233A-90A0-42ED-8D8A-F21E7D5BB7EB} + {9FE39661-2A39-4E9F-A5F2-11FB0D54CB42} = {5BD04C26-CE52-4893-8C1A-479705195CEF} + {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} EndGlobalSection GlobalSection(ExtensibilityGlobals) = postSolution SolutionGuid = {4795CF16-0FDB-4BE0-9768-5CF31564DC03} diff --git a/csharp/src/Telemetry/Traces/Exporters/Apache.Arrow.Adbc.Telemetry.Traces.Exporters.csproj b/csharp/src/Telemetry/Traces/Exporters/Apache.Arrow.Adbc.Telemetry.Traces.Exporters.csproj new file mode 100644 index 0000000000..b4109883f2 --- /dev/null +++ b/csharp/src/Telemetry/Traces/Exporters/Apache.Arrow.Adbc.Telemetry.Traces.Exporters.csproj @@ -0,0 +1,25 @@ + + + + netstandard2.0;net8.0 + enable + readme.md + + + + true + \ + PreserveNewest + + + + + + + + + + + + + diff --git a/csharp/src/Telemetry/Traces/Exporters/ExportersBuilder.cs b/csharp/src/Telemetry/Traces/Exporters/ExportersBuilder.cs new file mode 100644 index 0000000000..1c232448c2 --- /dev/null +++ b/csharp/src/Telemetry/Traces/Exporters/ExportersBuilder.cs @@ -0,0 +1,211 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You 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. + */ + +using System; +using System.Collections.Generic; +using Apache.Arrow.Adbc.Telemetry.Traces.Exporters.FileExporter; +using OpenTelemetry; +using OpenTelemetry.Resources; +using OpenTelemetry.Trace; + +namespace Apache.Arrow.Adbc.Telemetry.Traces.Exporters +{ + public class ExportersBuilder + { + private static readonly IReadOnlyDictionary> s_tracerProviderFactoriesDefault; + + private readonly string _sourceName; + private readonly string? _sourceVersion; + private readonly IReadOnlyDictionary> _tracerProviderFactories; + + static ExportersBuilder() + { + var defaultProviders = new Dictionary> + { + [ExportersOptions.Exporters.None] = NewNoopTracerProvider, + [ExportersOptions.Exporters.Otlp] = NewOtlpTracerProvider, + [ExportersOptions.Exporters.Console] = NewConsoleTracerProvider, + [ExportersOptions.Exporters.AdbcFile] = NewAdbcFileTracerProvider, + }; + s_tracerProviderFactoriesDefault = defaultProviders; + } + + private ExportersBuilder(Builder builder) + { + _sourceName = builder.SourceName; + _sourceVersion = builder.SourceVersion; + _tracerProviderFactories = builder.TracerProviderFactories; + } + + /// + /// Build an from different possible Exporters. Use the Add* functions to add + /// one or more factories. + /// + /// The name of the source that the exporter will filter on. + /// The (optional) version of the source that the exporter will filter on. + /// A object. + /// + public static Builder Build(string sourceName, string? sourceVersion = default, bool addDefaultExporters = false) + { + if (string.IsNullOrWhiteSpace(sourceName)) + { + throw new ArgumentNullException(nameof(sourceName)); + } + return new Builder(sourceName, sourceVersion, addDefaultExporters); + } + + /// + /// + /// Attempts to activate an exporter based on the dictionary of factories + /// added to the . If the value of exporterOption is not null and not empty, it will be + /// used as the key to the dictionary. If exporterOption is null or empty, then the exporter option will + /// check the environment variable identified by environmentName (default ). + /// If both the exporterOption and the environment variable value are null or empty, then this function will return null + /// and no exporeter will be activated. + /// + /// + /// If the exporterOption or the value of the environment variable are not null and not empty, then + /// if a matching factory delegate is found, it is called to activate the exporter. If no exception is thrown, + /// the result of the factory method returns the result which may be null. If a matching function is found, the expoertName is set. + /// If the factory delegate throws an exception it is not caught. + /// + /// + /// The value (name) of the exporter option, typically passed as option . + /// The actual exporter name when successfully activated. + /// The (optional) name of the environment variable to test for the exporter name. Default: + /// + /// The a non-null when successfully activated. Note: this object must be explicitly disposed when no longer necessary. + public TracerProvider? Activate( + string? exporterOption, + out string? exporterName, + string environmentName = ExportersOptions.Environment.Exporter) + { + TracerProvider? tracerProvider = null; + exporterName = null; + + if (string.IsNullOrWhiteSpace(exporterOption)) + { + // Fall back to check the environment variable + exporterOption = Environment.GetEnvironmentVariable(environmentName); + } + if (string.IsNullOrWhiteSpace(exporterOption)) + { + // Neither option or environment variable is set - no tracer provider will be activated. + return null; + } + + if (!_tracerProviderFactories.TryGetValue(exporterOption!, out Func? factory)) + { + // Requested option has not been added via the builder + throw AdbcException.NotImplemented($"Exporter option '{exporterOption}' is not implemented."); + } + + tracerProvider = factory.Invoke(_sourceName, _sourceVersion); + if (tracerProvider != null) + { + exporterName = exporterOption; + } + return tracerProvider; + } + + public static TracerProvider NewAdbcFileTracerProvider(string sourceName, string? sourceVersion) => + Sdk.CreateTracerProviderBuilder() + .AddSource(sourceName) + .ConfigureResource(resource => + resource.AddService( + serviceName: sourceName, + serviceVersion: sourceVersion)) + .AddAdbcFileExporter(sourceName) + .Build(); + + public static TracerProvider NewConsoleTracerProvider(string sourceName, string? sourceVersion) => + Sdk.CreateTracerProviderBuilder() + .AddSource(sourceName) + .ConfigureResource(resource => + resource.AddService( + serviceName: sourceName, + serviceVersion: sourceVersion)) + .AddConsoleExporter() + .Build(); + + public static TracerProvider NewOtlpTracerProvider(string sourceName, string? sourceVersion) => + Sdk.CreateTracerProviderBuilder() + .AddSource(sourceName) + .ConfigureResource(resource => + resource.AddService( + serviceName: sourceName, + serviceVersion: sourceVersion)) + .AddOtlpExporter() + .Build(); + + public static TracerProvider? NewNoopTracerProvider(string sourceName, string? sourceVersion) => + null; + + public class Builder + { + private readonly string _sourceName; + private readonly string? _sourceVersion; + private readonly Dictionary> _tracerProviderFactories; + + internal string SourceName => _sourceName; + + internal string? SourceVersion => _sourceVersion; + + internal IReadOnlyDictionary> TracerProviderFactories => _tracerProviderFactories; + + public Builder(string sourceName, string? sourceVersion, bool addDefaultExporters = false) + { + _sourceName = sourceName; + _sourceVersion = sourceVersion; + _tracerProviderFactories = []; + if (addDefaultExporters) + { + AddDefaultExporters(); + } + } + + public ExportersBuilder Build() + { + if (_tracerProviderFactories.Count == 0) + { + throw new InvalidOperationException("No options provided. Please add one or more exporter."); + } + return new ExportersBuilder(this); + } + + public Builder AddExporter(string exporterName, Func tracerProviderFactory) + { + if (string.IsNullOrWhiteSpace(exporterName)) + { + throw new ArgumentNullException(nameof(exporterName)); + } + + _tracerProviderFactories.Add(exporterName, tracerProviderFactory); + return this; + } + + private Builder AddDefaultExporters() + { + foreach (KeyValuePair> item in s_tracerProviderFactoriesDefault) + { + _tracerProviderFactories[item.Key] = item.Value; + } + return this; + } + } + } +} diff --git a/csharp/src/Telemetry/Traces/Exporters/ExportersOptions.cs b/csharp/src/Telemetry/Traces/Exporters/ExportersOptions.cs new file mode 100644 index 0000000000..8cd631ac24 --- /dev/null +++ b/csharp/src/Telemetry/Traces/Exporters/ExportersOptions.cs @@ -0,0 +1,37 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You 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. + */ + +namespace Apache.Arrow.Adbc.Telemetry.Traces.Exporters +{ + public class ExportersOptions + { + public const string Exporter = "adbc.traces.exporter"; + + public static class Environment + { + public const string Exporter = "OTEL_TRACES_EXPORTER"; + } + + public static class Exporters + { + public const string None = "none"; + public const string Otlp = "otlp"; + public const string Console = "console"; + public const string AdbcFile = "adbcfile"; + } + } +} diff --git a/csharp/src/Telemetry/Traces/Exporters/FileExporter/FileExporter.cs b/csharp/src/Telemetry/Traces/Exporters/FileExporter/FileExporter.cs new file mode 100644 index 0000000000..a034a938a3 --- /dev/null +++ b/csharp/src/Telemetry/Traces/Exporters/FileExporter/FileExporter.cs @@ -0,0 +1,272 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You 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. + */ + +using System; +using System.Collections.Concurrent; +using System.Collections.Generic; +using System.Diagnostics; +using System.IO; +using System.Text; +using System.Text.Json; +using System.Threading; +using System.Threading.Tasks; +using OpenTelemetry; + +namespace Apache.Arrow.Adbc.Telemetry.Traces.Exporters.FileExporter +{ + internal class FileExporter : BaseExporter + { + internal const long MaxFileSizeKbDefault = 1024; + internal const int MaxTraceFilesDefault = 999; + internal const string ApacheArrowAdbcNamespace = "Apache.Arrow.Adbc"; + private const string TracesFolderName = "Traces"; + + private static readonly ConcurrentDictionary> s_fileExporters = new(); + private static readonly byte[] s_newLine = Encoding.UTF8.GetBytes(Environment.NewLine); + + private readonly TracingFile _tracingFile; + private readonly string _fileBaseName; + private readonly string _tracesDirectoryFullName; + private readonly ConcurrentQueue _activityQueue = new(); + private readonly CancellationTokenSource _cancellationTokenSource; + + private bool _disposed = false; + + internal static bool TryCreate(FileExporterOptions options, out FileExporter? fileExporter) + { + return TryCreate( + options.FileBaseName ?? ApacheArrowAdbcNamespace, + options.TraceLocation ?? TracingLocationDefault, + options.MaxTraceFileSizeKb, + options.MaxTraceFiles, + out fileExporter); + } + + internal static bool TryCreate( + string fileBaseName, + string traceLocation, + long maxTraceFileSizeKb, + int maxTraceFiles, + out FileExporter? fileExporter) + { + ValidateParameters(fileBaseName, traceLocation, maxTraceFileSizeKb, maxTraceFiles); + + DirectoryInfo tracesDirectory = new(traceLocation ?? TracingLocationDefault); + string tracesDirectoryFullName = tracesDirectory.FullName; + + // In case we don't need to create this object, we'll lazy load the object only if added to the collection. + var exporterInstance = new Lazy(() => + { + CancellationTokenSource cancellationTokenSource = new(); + FileExporter fileExporter = new(fileBaseName, tracesDirectory, maxTraceFileSizeKb, maxTraceFiles); + return new FileExporterInstance( + fileExporter, + // This listens/polls for activity in the queue and writes them to file + Task.Run(async () => await ProcessActivitiesAsync(fileExporter, cancellationTokenSource.Token)), + cancellationTokenSource); + }); + + // We only want one exporter listening on a source in a particular folder. + // If two or more exporters are running, it'll create duplicate trace entries. + // On Dispose, ensure to stop and remove the only instance, in case we need a new one later. + string listenerId = GetListenerId(fileBaseName, tracesDirectoryFullName); + bool isAdded = s_fileExporters.TryAdd(listenerId, exporterInstance); + if (isAdded) + { + // This instance was added so load the object now. + fileExporter = exporterInstance.Value.FileExporter; + return true; + } + + // There is already an exporter listening on the source/location + fileExporter = null; + return false; + } + + internal static void ValidateParameters(string fileBaseName, string traceLocation, long maxTraceFileSizeKb, int maxTraceFiles) + { + if (string.IsNullOrWhiteSpace(fileBaseName)) + throw new ArgumentNullException(nameof(fileBaseName)); + if (fileBaseName.IndexOfAny(Path.GetInvalidFileNameChars()) >= 0) + throw new ArgumentException("Invalid or unsupported file name", nameof(fileBaseName)); + if (string.IsNullOrWhiteSpace(traceLocation) || traceLocation.IndexOfAny(Path.GetInvalidPathChars()) >= 0) + throw new ArgumentException("Invalid or unsupported folder name", nameof(traceLocation)); + if (maxTraceFileSizeKb < 1) + throw new ArgumentException("maxTraceFileSizeKb must be greater than zero", nameof(maxTraceFileSizeKb)); + if (maxTraceFiles < 1) + throw new ArgumentException("maxTraceFiles must be greater than zero.", nameof(maxTraceFiles)); + + IsDirectoryWritable(traceLocation, throwIfFails: true); + } + + private static string GetListenerId(string sourceName, string traceFolderLocation) => $"{sourceName}{traceFolderLocation}"; + + public override ExportResult Export(in Batch batch) + { + foreach (Activity activity in batch) + { + if (activity == null) continue; + _activityQueue.Enqueue(activity); + } + return ExportResult.Success; + } + + private static async Task ProcessActivitiesAsync(FileExporter fileExporter, CancellationToken cancellationToken) + { + try + { + TimeSpan delay = TimeSpan.FromMilliseconds(100); + // Polls for and then writes any activities in the queue + while (!cancellationToken.IsCancellationRequested) + { + await Task.Delay(delay, cancellationToken); + await fileExporter._tracingFile.WriteLinesAsync(GetActivitiesAsync(fileExporter._activityQueue), cancellationToken); + } + } + catch (Exception ex) + { + // Since this will be called on an independent thread, we need to avoid uncaught exceptions. + Debug.WriteLine(ex); + } + } + + private static bool IsDirectoryWritable(string traceLocation, bool throwIfFails = false) + { + try + { + if (!Directory.Exists(traceLocation)) + { + Directory.CreateDirectory(traceLocation); + } + string tempFilePath = Path.Combine(traceLocation, Path.GetRandomFileName()); + using FileStream fs = File.Create(tempFilePath, 1, FileOptions.DeleteOnClose); + return true; + } + catch when (!throwIfFails) + { + return false; + } + } + + private static async IAsyncEnumerable GetActivitiesAsync(ConcurrentQueue activityQueue) + { + using MemoryStream stream = new MemoryStream(); + while (activityQueue.TryDequeue(out Activity? activity)) + { + stream.SetLength(0); + SerializableActivity serializableActivity = new(activity); + await JsonSerializer.SerializeAsync( + stream, + serializableActivity, + SerializableActivityJsonContext.Default.SerializableActivity); + stream.Write(s_newLine, 0, s_newLine.Length); + stream.Position = 0; + + yield return stream; + } + } + + private FileExporter(string fileBaseName, DirectoryInfo tracesDirectory, long maxTraceFileSizeKb, int maxTraceFiles) + { + string fullName = tracesDirectory.FullName; + _fileBaseName = fileBaseName; + _tracesDirectoryFullName = fullName; + _tracingFile = new(fileBaseName, fullName, maxTraceFileSizeKb, maxTraceFiles); + _cancellationTokenSource = new CancellationTokenSource(); + } + + internal static string TracingLocationDefault { get; } = + new DirectoryInfo( + Path.Combine( + Environment.GetFolderPath(Environment.SpecialFolder.LocalApplicationData), + ApacheArrowAdbcNamespace, + TracesFolderName)).FullName; + + private async Task FlushAsync(CancellationToken cancellationToken = default) + { + // Ensure existing writes are completed. + while (!cancellationToken.IsCancellationRequested && !_activityQueue.IsEmpty) + { + await Task.Delay(100); + } + } + + protected override void Dispose(bool disposing) + { + if (!_disposed && disposing) + { + // Allow flush of any existing events. + using CancellationTokenSource flushTimeout = new(); + flushTimeout.CancelAfter(TimeSpan.FromSeconds(5)); + FlushAsync(flushTimeout.Token).Wait(); + + // Remove and dispose of single instance of exporter + string listenerId = GetListenerId(_fileBaseName, _tracesDirectoryFullName); + bool isRemoved = s_fileExporters.TryRemove(listenerId, out Lazy? listener); + if (isRemoved && listener != null && listener.IsValueCreated) + { + listener?.Value.Dispose(); + } + _cancellationTokenSource.Cancel(); + _cancellationTokenSource.Dispose(); + _disposed = true; + } + base.Dispose(disposing); + } + + private class FileExporterInstance( + FileExporter fileExporter, + Task writeActivitiesTask, + CancellationTokenSource cancellationTokenSource) + : IDisposable + { + private bool _disposedValue; + + internal FileExporter FileExporter { get; } = fileExporter; + + internal CancellationTokenSource CancellationTokenSource { get; } = cancellationTokenSource; + + internal Task WriteActivitiesTask { get; } = writeActivitiesTask; + + protected virtual void Dispose(bool disposing) + { + if (!_disposedValue && disposing) + { + CancellationTokenSource.Cancel(); + try + { + WriteActivitiesTask.Wait(); + } + catch + { + // Ignore + } + WriteActivitiesTask.Dispose(); + CancellationTokenSource.Dispose(); + _disposedValue = true; + } + } + + public void Dispose() + { + // Do not change this code. Put cleanup code in 'Dispose(bool disposing)' method + Dispose(disposing: true); + GC.SuppressFinalize(this); + } + } + } +} diff --git a/csharp/src/Telemetry/Traces/Exporters/FileExporter/FileExporterExtensions.cs b/csharp/src/Telemetry/Traces/Exporters/FileExporter/FileExporterExtensions.cs new file mode 100644 index 0000000000..b35ca43f33 --- /dev/null +++ b/csharp/src/Telemetry/Traces/Exporters/FileExporter/FileExporterExtensions.cs @@ -0,0 +1,137 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You 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. + */ + +using System; +using Microsoft.Extensions.DependencyInjection; +using Microsoft.Extensions.Options; +using OpenTelemetry; +using OpenTelemetry.Trace; + +namespace Apache.Arrow.Adbc.Telemetry.Traces.Exporters.FileExporter +{ + public static class FileExporterExtensions + { + /// + /// Adds an ADBC file exporter to listen for and write trace entries into files. + /// + /// + /// The to add to. + /// Ensure to call before calling this method. + /// + /// The previous with the exporter added. + /// + /// Note that only one global instance of the exporter is added. + /// If there is already an existing exporter for the same source and location, no new one is added. + /// + public static OpenTelemetry.Trace.TracerProviderBuilder AddAdbcFileExporter(this OpenTelemetry.Trace.TracerProviderBuilder builder) + => builder.AddAdbcFileExporter(null, null); + + /// + /// Adds an ADBC file exporter to listen for and write trace entries into files. + /// + /// + /// The to add to. + /// Ensure to call before calling this method. + /// + /// The configuratio action to set the with. + /// The previous with the exporter added. + /// + /// Note that only one global instance of the exporter is added. + /// If there is already an existing exporter for the same source and location, no new one is added. + /// + public static OpenTelemetry.Trace.TracerProviderBuilder AddAdbcFileExporter(this OpenTelemetry.Trace.TracerProviderBuilder builder, Action? configure) + => builder.AddAdbcFileExporter(null, configure); + + /// + /// Adds an ADBC file exporter to listen for and write trace entries into files. + /// + /// + /// The to add to. + /// Ensure to call before calling this method. + /// + /// The name of this configuration. + /// The configuratio action to set the with. + /// The previous with the exporter added. + /// + /// Note that only one global instance of the exporter is added. + /// If there is already an existing exporter for the same source and location, no new one is added. + /// + public static OpenTelemetry.Trace.TracerProviderBuilder AddAdbcFileExporter( + this OpenTelemetry.Trace.TracerProviderBuilder builder, + string? name, + Action? configure) + { + name ??= Options.DefaultName; + + if (configure != null) + { + builder.ConfigureServices(services => services.Configure(name, configure)); + } + + FileExporterOptions options = new(); + configure?.Invoke(options); + if (FileExporter.TryCreate(options, out FileExporter? fileExporter)) + { + // Only add a new processor if there isn't already one listening for the source/location. + return builder.AddProcessor(_ => new BatchActivityExportProcessor(fileExporter!)); + } + return builder; + } + + /// + /// Adds an ADBC file exporter to listen for and write trace entries into files. + /// + /// + /// The to add to. + /// Ensure to call before calling this method. + /// + /// + /// The base file name (typically the tracing source name). + /// Trace files will be created with the following name template: {fileBaseName}-trace-{dateTime}.log + /// + /// + /// The full or partial path to a folder where the trace files will be written. + /// If the folder doesn not exist, it will be created. + /// + /// The maximum size of each trace file (in KB). If a trace file exceeds this limit, a new trace file is created. + /// The maximum number of trace files in the tracing folder. If the number of files exceeds this maximum, older files will be removed. + /// The previous with the exporter added. + /// + /// Note that only one global instance of the exporter is added. + /// If there is already an existing exporter for the same source and location, no new one is added. + /// + public static OpenTelemetry.Trace.TracerProviderBuilder AddAdbcFileExporter( + this OpenTelemetry.Trace.TracerProviderBuilder builder, + string fileBaseName, + string? traceLocation = default, + long? maxTraceFileSizeKb = default, + int? maxTraceFiles = default) + { + maxTraceFileSizeKb ??= FileExporter.MaxFileSizeKbDefault; + maxTraceFiles ??= FileExporter.MaxTraceFilesDefault; + traceLocation ??= FileExporter.TracingLocationDefault; + FileExporter.ValidateParameters(fileBaseName, traceLocation, maxTraceFileSizeKb.Value, maxTraceFiles.Value); + + if (FileExporter.TryCreate(fileBaseName, traceLocation, maxTraceFileSizeKb.Value, maxTraceFiles.Value, out FileExporter? fileExporter)) + { + // Only add a new processor if there isn't already one listening for the source/location. + return builder.AddProcessor(_ => new BatchActivityExportProcessor(fileExporter!)); + } + return builder; + } + } +} diff --git a/csharp/src/Telemetry/Traces/Exporters/FileExporter/FileExporterOptions.cs b/csharp/src/Telemetry/Traces/Exporters/FileExporter/FileExporterOptions.cs new file mode 100644 index 0000000000..691ade0369 --- /dev/null +++ b/csharp/src/Telemetry/Traces/Exporters/FileExporter/FileExporterOptions.cs @@ -0,0 +1,47 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You 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. + */ + +namespace Apache.Arrow.Adbc.Telemetry.Traces.Exporters.FileExporter +{ + /// + /// The options a uses to trace activity to files. + /// + public class FileExporterOptions + { + /// + /// Gets or sets the base file name (typically the tracing source name). + /// Trace files will be created with the following name template: {fileBaseName}-trace-{dateTime}.log + /// + public string FileBaseName { get; set; } = FileExporter.ApacheArrowAdbcNamespace; + + /// + /// The full or partial path to a folder where the trace files will be written. + /// If the folder does not exist, it will be created. + /// + public string? TraceLocation { get; set; } = FileExporter.TracingLocationDefault; + + /// + /// The maximum size of each trace file (in KB). If a trace file exceeds this limit, a new trace file is created. + /// + public long MaxTraceFileSizeKb { get; set; } = FileExporter.MaxFileSizeKbDefault; + + /// + /// The maximum number of trace files in the tracing folder. If the number of files exceeds this maximum, older files will be removed. + /// + public int MaxTraceFiles { get; set; } = FileExporter.MaxTraceFilesDefault; + } +} diff --git a/csharp/src/Telemetry/Traces/Exporters/FileExporter/SerializableActivity.cs b/csharp/src/Telemetry/Traces/Exporters/FileExporter/SerializableActivity.cs new file mode 100644 index 0000000000..aff03a868e --- /dev/null +++ b/csharp/src/Telemetry/Traces/Exporters/FileExporter/SerializableActivity.cs @@ -0,0 +1,192 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You 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. + */ + +using System; +using System.Collections.Generic; +using System.Diagnostics; +using System.Linq; +using System.Text.Json.Serialization; + +namespace Apache.Arrow.Adbc.Telemetry.Traces.Exporters.FileExporter +{ + /// + /// Simplified version of that excludes some properties, etc. + /// + internal class SerializableActivity + { + [JsonConstructor] + public SerializableActivity() { } + + internal SerializableActivity( + ActivityStatusCode status, + string? statusDescription, + bool hasRemoteParent, + ActivityKind kind, + string operationName, + TimeSpan duration, + DateTime startTimeUtc, + string? id, + string? parentId, + string? rootId, + string? traceStateString, + ActivitySpanId spanId, + ActivityTraceId traceId, + bool recorded, + bool isAllDataRequested, + ActivityTraceFlags activityTraceFlags, + ActivitySpanId parentSpanId, + ActivityIdFormat idFormat, + IReadOnlyDictionary tagObjects, + IReadOnlyList events, + IReadOnlyList links, + IReadOnlyDictionary baggage) + { + Status = statusDescription ?? status.ToString(); + HasRemoteParent = hasRemoteParent; + Kind = kind.ToString(); + OperationName = operationName; + Duration = duration; + StartTimeUtc = startTimeUtc; + Id = id; + ParentId = parentId; + RootId = rootId; + TraceStateString = traceStateString; + SpanId = spanId.ToHexString(); + TraceId = traceId.ToHexString(); + Recorded = recorded; + IsAllDataRequested = isAllDataRequested; + ActivityTraceFlags = activityTraceFlags.ToString(); + ParentSpanId = parentSpanId.ToHexString(); + IdFormat = idFormat.ToString(); + TagObjects = tagObjects; + Events = events; + Links = links; + Baggage = baggage; + } + + internal SerializableActivity(Activity activity) : this( + activity.Status, + activity.StatusDescription, + activity.HasRemoteParent, + activity.Kind, + activity.OperationName, + activity.Duration, + activity.StartTimeUtc, + activity.Id, + activity.ParentId, + activity.RootId, + activity.TraceStateString, + activity.SpanId, + activity.TraceId, + activity.Recorded, + activity.IsAllDataRequested, + activity.ActivityTraceFlags, + activity.ParentSpanId, + activity.IdFormat, + activity.TagObjects.ToDictionary(kv => kv.Key, kv => kv.Value), + activity.Events.Select(e => (SerializableActivityEvent)e).ToArray(), + activity.Links.Select(l => (SerializableActivityLink)l).ToArray(), + activity.Baggage.ToDictionary(kv => kv.Key, kv => kv.Value)) + { } + + public string? Status { get; set; } + public bool HasRemoteParent { get; set; } + public string? Kind { get; set; } + public string OperationName { get; set; } = ""; + public TimeSpan Duration { get; set; } + public DateTime StartTimeUtc { get; set; } + public string? Id { get; set; } + public string? ParentId { get; set; } + public string? RootId { get; set; } + + public string? TraceStateString { get; set; } + public string? SpanId { get; set; } + public string? TraceId { get; set; } + public bool Recorded { get; set; } + public bool IsAllDataRequested { get; set; } + public string? ActivityTraceFlags { get; set; } + public string? ParentSpanId { get; set; } + public string? IdFormat { get; set; } + + public IReadOnlyDictionary TagObjects { get; set; } = new Dictionary(); + public IReadOnlyList Events { get; set; } = []; + public IReadOnlyList Links { get; set; } = []; + public IReadOnlyDictionary Baggage { get; set; } = new Dictionary(); + } + + internal class SerializableActivityEvent + { + /// + /// Gets the name. + /// + public string? Name { get; set; } + + /// + /// Gets the timestamp. + /// + public DateTimeOffset Timestamp { get; set; } + + public IReadOnlyList> Tags { get; set; } = []; + + public static implicit operator SerializableActivityEvent(ActivityEvent source) + { + return new SerializableActivityEvent() + { + Name = source.Name, + Timestamp = source.Timestamp, + Tags = source.Tags.ToArray(), + }; + } + } + + internal class SerializableActivityLink + { + public SerializableActivityContext? Context { get; set; } + + public IReadOnlyList>? Tags { get; set; } = []; + + public static implicit operator SerializableActivityLink(ActivityLink source) + { + return new SerializableActivityLink() + { + Context = source.Context, + Tags = source.Tags?.ToArray(), + }; + } + } + + internal class SerializableActivityContext + { + public string? SpanId { get; set; } + public string? TraceId { get; set; } + public string? TraceState { get; set; } + public ActivityTraceFlags? TraceFlags { get; set; } + public bool IsRemote { get; set; } + + public static implicit operator SerializableActivityContext(ActivityContext source) + { + return new SerializableActivityContext() + { + SpanId = source.SpanId.ToHexString(), + TraceId = source.TraceId.ToHexString(), + TraceState = source.TraceState, + TraceFlags = source.TraceFlags, + IsRemote = source.IsRemote, + }; + } + } +} diff --git a/csharp/src/Telemetry/Traces/Exporters/FileExporter/SerializableActivityJsonContext.cs b/csharp/src/Telemetry/Traces/Exporters/FileExporter/SerializableActivityJsonContext.cs new file mode 100644 index 0000000000..b5cec25f6f --- /dev/null +++ b/csharp/src/Telemetry/Traces/Exporters/FileExporter/SerializableActivityJsonContext.cs @@ -0,0 +1,32 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You 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. + */ + +using System.Text.Json.Serialization; + +namespace Apache.Arrow.Adbc.Telemetry.Traces.Exporters.FileExporter +{ + /// + /// Provides a source-generated JSON serialization context for the type. + /// + /// This context is used to optimize JSON serialization and deserialization of objects by leveraging source generation. It is intended for internal use within + /// the application. + [JsonSerializable(typeof(SerializableActivity))] + internal partial class SerializableActivityJsonContext : JsonSerializerContext + { + } +} diff --git a/csharp/src/Telemetry/Traces/Exporters/FileExporter/TracingFile.cs b/csharp/src/Telemetry/Traces/Exporters/FileExporter/TracingFile.cs new file mode 100644 index 0000000000..8595056df8 --- /dev/null +++ b/csharp/src/Telemetry/Traces/Exporters/FileExporter/TracingFile.cs @@ -0,0 +1,211 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You 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. + */ + +using System; +using System.Collections.Generic; +using System.Diagnostics; +using System.IO; +using System.Linq; +using System.Threading; +using System.Threading.Tasks; + +namespace Apache.Arrow.Adbc.Telemetry.Traces.Exporters.FileExporter +{ + /// + /// Provides access to writing trace files, limiting the + /// individual files size and ensuring unique file names. + /// + internal class TracingFile + { + private static readonly string s_defaultTracePath = FileExporter.TracingLocationDefault; + private static readonly Random s_globalRandom = new(); + private static readonly ThreadLocal s_threadLocalRandom = new(NewRandom); + private static readonly Lazy s_processId = new(() => Process.GetCurrentProcess().Id.ToString(), isThreadSafe: true); + private readonly string _fileBaseName; + private readonly DirectoryInfo _tracingDirectory; + private FileInfo? _currentTraceFileInfo; + private readonly long _maxFileSizeKb; + private readonly int _maxTraceFiles; + + internal TracingFile(string fileBaseName, string? traceDirectoryPath = default, long maxFileSizeKb = FileExporter.MaxFileSizeKbDefault, int maxTraceFiles = FileExporter.MaxTraceFilesDefault) : + this(fileBaseName, ResolveTraceDirectory(traceDirectoryPath), maxFileSizeKb, maxTraceFiles) + { } + + internal TracingFile(string fileBaseName, DirectoryInfo traceDirectory, long maxFileSizeKb, int maxTraceFiles) + { + if (string.IsNullOrWhiteSpace(fileBaseName)) throw new ArgumentNullException(nameof(fileBaseName)); + _fileBaseName = fileBaseName; + _tracingDirectory = traceDirectory; + _maxFileSizeKb = maxFileSizeKb; + _maxTraceFiles = maxTraceFiles; + EnsureTraceDirectory(); + } + + /// + /// Writes lines of trace where each stream is a line in the trace file. + /// + /// The enumerable of trace lines. + /// The cancellation token. + /// + internal async Task WriteLinesAsync(IAsyncEnumerable streams, CancellationToken cancellationToken = default) + { + if (cancellationToken.IsCancellationRequested) return; + + string findSearchPattern = _fileBaseName + $"-trace-*-{ProcessId}.log"; + if (_currentTraceFileInfo == null) + { + IOrderedEnumerable? traceFileInfos = await GetTracingFilesAsync(_tracingDirectory, findSearchPattern); + FileInfo? mostRecentFile = traceFileInfos?.FirstOrDefault(); + mostRecentFile?.Refresh(); + + // Use the latest file, if it is not maxxed-out, or start a new tracing file. + _currentTraceFileInfo = mostRecentFile != null && mostRecentFile.Length < _maxFileSizeKb * 1024 + ? mostRecentFile + : new FileInfo(NewFileName()); + } + + // Write out to the file and retry if IO errors occur. + await ActionWithRetryAsync(async () => await WriteLinesAsync(streams), cancellationToken: cancellationToken); + + // Check if we need to remove old files + if (_tracingDirectory.Exists) + { + // This will clean-up files for all processes in the same directory. + string deleteSearchPattern = _fileBaseName + $"-trace-*.log"; + FileInfo[] tracingFiles = [.. await GetTracingFilesAsync(_tracingDirectory, deleteSearchPattern)]; + if (tracingFiles != null && tracingFiles.Length > _maxTraceFiles) + { + for (int i = tracingFiles.Length - 1; i >= _maxTraceFiles; i--) + { + FileInfo? file = tracingFiles.ElementAtOrDefault(i); + // Note: don't pass the cancellation token, as we want this to ALWAYS run at the end. + await ActionWithRetryAsync(() => + { + file?.Delete(); + return Task.CompletedTask; + }); + } + } + } + } + + private async Task WriteLinesAsync(IAsyncEnumerable streams) + { + bool hasMoreData; + do + { + bool newFileRequired = false; + _currentTraceFileInfo!.Refresh(); + using (FileStream fileStream = _currentTraceFileInfo!.OpenWrite()) + { + fileStream.Position = fileStream.Length; + hasMoreData = false; + await foreach (Stream stream in streams) + { + if (fileStream.Length >= _maxFileSizeKb * 1024) + { + hasMoreData = true; + newFileRequired = true; + break; + } + + await stream.CopyToAsync(fileStream); + } + } + await Task.Yield(); // Yield to allow other tasks to run. + if (newFileRequired) + { + // If tracing file is maxxed-out, start a new tracing file. + _currentTraceFileInfo = new FileInfo(NewFileName()); + } + } while (hasMoreData); + } + + private static async Task> GetTracingFilesAsync(DirectoryInfo tracingDirectory, string searchPattern) + { + return await Task.Run(() => tracingDirectory + .EnumerateFiles(searchPattern, SearchOption.TopDirectoryOnly) + .OrderByDescending(f => f.LastWriteTimeUtc)); + } + + private static async Task ActionWithRetryAsync( + Func action, + int maxRetries = 100, + CancellationToken cancellationToken = default) where T : Exception + { + int retryCount = 0; + int delayTime = ThreadLocalRandom.Next(50, 500); // Introduce a small random delay to avoid contention. + TimeSpan pauseTime = TimeSpan.FromMilliseconds(delayTime); + bool completed = false; + + while (!cancellationToken.IsCancellationRequested && !completed && retryCount < maxRetries) + { + try + { + await action.Invoke(); + completed = true; + } + catch (T) when (retryCount < (maxRetries - 1)) + { + retryCount++; + try + { + await Task.Delay(pauseTime, cancellationToken); + } + catch (OperationCanceledException) + { + // Need to catch this exception or it will be propagated. + break; + } + } + } + } + + private string NewFileName() + { + string dateTimeSortable = DateTimeOffset.UtcNow.ToString("yyyy-MM-dd-HH-mm-ss-fff"); + return Path.Combine(_tracingDirectory.FullName, $"{_fileBaseName}-trace-{dateTimeSortable}-{ProcessId}.log"); + } + + private void EnsureTraceDirectory() + { + if (!Directory.Exists(_tracingDirectory.FullName)) + { + Directory.CreateDirectory(_tracingDirectory.FullName); + } + } + + private static DirectoryInfo ResolveTraceDirectory(string? traceDirectoryPath) => + traceDirectoryPath == null ? new DirectoryInfo(s_defaultTracePath) : new DirectoryInfo(traceDirectoryPath); + + private static string ProcessId => s_processId.Value; + + private static Random ThreadLocalRandom => s_threadLocalRandom.Value!; + + private static Random NewRandom() + { + int seed; + lock (s_globalRandom) + { + seed = s_globalRandom.Next(); + } + // Create a new random instance based on the global random seed. + // This ensures that each thread gets a different seed. + return new Random(seed); + } + } +} diff --git a/csharp/src/Telemetry/Traces/Exporters/Properties/AssemblyInfo.cs b/csharp/src/Telemetry/Traces/Exporters/Properties/AssemblyInfo.cs new file mode 100644 index 0000000000..2e9095f680 --- /dev/null +++ b/csharp/src/Telemetry/Traces/Exporters/Properties/AssemblyInfo.cs @@ -0,0 +1,20 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You 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. + */ + +using System.Runtime.CompilerServices; + +[assembly: InternalsVisibleTo("Apache.Arrow.Adbc.Tests.Telemetry.Traces.Exporters, PublicKey=0024000004800000940000000602000000240000525341310004000001000100e504183f6d470d6b67b6d19212be3e1f598f70c246a120194bc38130101d0c1853e4a0f2232cb12e37a7a90e707aabd38511dac4f25fcb0d691b2aa265900bf42de7f70468fc997551a40e1e0679b605aa2088a4a69e07c117e988f5b1738c570ee66997fba02485e7856a49eca5fd0706d09899b8312577cbb9034599fc92d4")] diff --git a/csharp/src/Telemetry/Traces/Exporters/readme.md b/csharp/src/Telemetry/Traces/Exporters/readme.md new file mode 100644 index 0000000000..8b743c7e71 --- /dev/null +++ b/csharp/src/Telemetry/Traces/Exporters/readme.md @@ -0,0 +1,54 @@ + + +# Traces Exporters + +## FileExporter + +Provides an OpenTelemetry (OTel) exporter to write telemetry traces to +rotating files in folder. File names are created with the following pattern: +`--.log`. +For example: `apache.arrow.adbc.drivers.databricks-2025-08-15-10-35-56-012345-99999.log`. +The default folder used is: + +| 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 100 files of maximum size 1024 KB are written to +the trace folder. + +## ExportersBuilder + +Helps activate one of a dictionary of supported exporters. + +The environment variable `OTEL_TRACES_EXPORTER` can be used to select one of the +available exporters. Or the database parameter `adbc.traces.exporter` can be used, +which has precedence over the environment variable. + +The following exporters are supported: + +| Exporter | Description | +| --- | --- | +| `otlp` | Exports traces to an OpenTelemetry Collector or directly to an Open Telemetry Line Protocol (OTLP) endpoint. | +| `file` | Exports traces to rotating files in a folder. | +| `console` | Exports traces to the console output. | +| `none` | Disables trace exporting. | diff --git a/csharp/test/Telemetry/Traces/Exporters/Apache.Arrow.Adbc.Tests.Telemetry.Traces.Exporters.csproj b/csharp/test/Telemetry/Traces/Exporters/Apache.Arrow.Adbc.Tests.Telemetry.Traces.Exporters.csproj new file mode 100644 index 0000000000..e0965451ee --- /dev/null +++ b/csharp/test/Telemetry/Traces/Exporters/Apache.Arrow.Adbc.Tests.Telemetry.Traces.Exporters.csproj @@ -0,0 +1,27 @@ + + + + net8.0;net472 + net8.0 + enable + + false + true + + + + + + + + + + + + + + + + + + diff --git a/csharp/test/Telemetry/Traces/Exporters/ExportersBuilderTests.cs b/csharp/test/Telemetry/Traces/Exporters/ExportersBuilderTests.cs new file mode 100644 index 0000000000..132b189393 --- /dev/null +++ b/csharp/test/Telemetry/Traces/Exporters/ExportersBuilderTests.cs @@ -0,0 +1,140 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You 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. + */ + +using System; +using Apache.Arrow.Adbc.Telemetry.Traces.Exporters; +using OpenTelemetry.Trace; + +namespace Apache.Arrow.Adbc.Tests.Telemetry.Traces.Exporters +{ + public class ExportersBuilderTests + { + [Theory] + [ClassData(typeof(ActivatorOptionData))] + public void CanActivateUsingOption(string sourceName, string? exporterOption, bool willActivate, Type? exceptionType = default) + { + string environmentName = NewName(); + Environment.SetEnvironmentVariable(environmentName, null); + if (exceptionType == null) + { + using TracerProvider? tracerProvider = ExportersBuilder.Build(sourceName, addDefaultExporters: true) + .Build() + .Activate(exporterOption, out string? exporterName, environmentName); + if (willActivate) + { + Assert.NotNull(exporterName); + Assert.NotNull(tracerProvider); + } + else + { + Assert.Null(tracerProvider); + } + } + else + { + Assert.Throws(exceptionType, () => ExportersBuilder.Build(sourceName, addDefaultExporters: true) + .Build() + .Activate(exporterOption, out string? exporterName, environmentName)); + } + } + + [Fact] + public void CanDetectMissingExporters() + { + Assert.Throws(() => ExportersBuilder.Build("sourceName") + .Build() + .Activate(ExportersOptions.Exporters.Console, out string? exporterName)); + } + + [Theory] + [ClassData(typeof(ActivatorOptionData))] + public void CanActivateUsingEnvironment(string sourceName, string? exporterOption, bool activateExpected, Type? exceptionType = default) + { + string environmentName = NewName(); + Environment.SetEnvironmentVariable(environmentName, exporterOption); + if (exceptionType == null) + { + using TracerProvider? tracerProvider = ExportersBuilder.Build(sourceName, addDefaultExporters: true) + .Build() + .Activate(null, out string? exporterName, environmentName); + if (activateExpected) + { + Assert.NotNull(exporterName); + Assert.NotNull(tracerProvider); + } + else + { + Assert.Null(tracerProvider); + } + } + else + { + Assert.Throws(exceptionType, () => ExportersBuilder.Build(sourceName, addDefaultExporters: true) + .Build() + .Activate(null, out string? exporterName, environmentName)); + } + } + + [Fact] + public void CanAddExporters() + { + // Test single + string exporterOption = NewName(); + using (TracerProvider? tracerProvider = ExportersBuilder.Build("sourceName") + .AddExporter(exporterOption, ExportersBuilder.NewConsoleTracerProvider) + .Build() + .Activate(exporterOption, out string? exporterName)) + { + Assert.NotNull(tracerProvider); + Assert.NotNull(exporterName); + Assert.Equal(exporterOption, exporterName); + } + + // Test with other default exporters + exporterOption = NewName(); + using (TracerProvider? tracerProvider = ExportersBuilder.Build("sourceName", addDefaultExporters: true) + .AddExporter(exporterOption, ExportersBuilder.NewConsoleTracerProvider) + .Build() + .Activate(exporterOption, out string? exporterName)) + { + Assert.NotNull(tracerProvider); + Assert.NotNull(exporterName); + Assert.Equal(exporterOption, exporterName); + } + } + + public class ActivatorOptionData : TheoryData + { + public ActivatorOptionData() + { + Add("a", null, false, null); + Add("a", "", false, null); + Add("a", " ", false, null); + Add("a", ExportersOptions.Exporters.None, false, null); + Add("a", ExportersOptions.Exporters.Otlp, true, null); + Add("a", ExportersOptions.Exporters.Console, true, null); + Add("a", ExportersOptions.Exporters.AdbcFile, true, null); + Add("a", "unknown", false, typeof(AdbcException)); + Add(null, ExportersOptions.Exporters.Console, false, typeof(ArgumentNullException)); + Add("", ExportersOptions.Exporters.Console, false, typeof(ArgumentNullException)); + Add(" ", ExportersOptions.Exporters.Console, false, typeof(ArgumentNullException)); + } + } + + internal static string NewName() => Guid.NewGuid().ToString("N"); + } +} diff --git a/csharp/test/Telemetry/Traces/Exporters/FileExporter/FileExporterTests.cs b/csharp/test/Telemetry/Traces/Exporters/FileExporter/FileExporterTests.cs new file mode 100644 index 0000000000..16f0c370d9 --- /dev/null +++ b/csharp/test/Telemetry/Traces/Exporters/FileExporter/FileExporterTests.cs @@ -0,0 +1,337 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You 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. + */ + +using System; +using System.Diagnostics; +using System.IO; +using System.Linq; +using System.Text; +using System.Threading.Tasks; +using OpenTelemetry; +using OpenTelemetry.Trace; +using Xunit; +using Xunit.Abstractions; +using Apache.Arrow.Adbc.Telemetry.Traces.Exporters.FileExporter; + +namespace Apache.Arrow.Adbc.Tests.Telemetry.Traces.Exporters.FileExporter +{ + public class TracingFileExporterTests : IDisposable + { + private readonly ITestOutputHelper? _outputHelper; + private bool _disposed; + private readonly string _activitySourceName; + private readonly ActivitySource _activitySource; + private static readonly string s_localApplicationDataFolderPath = Environment.GetFolderPath(Environment.SpecialFolder.LocalApplicationData); + + public TracingFileExporterTests(ITestOutputHelper? outputHelper) + { + _outputHelper = outputHelper; + _activitySourceName = ExportersBuilderTests.NewName(); + _activitySource = new ActivitySource(_activitySourceName); + } + + [Fact] + internal async Task CanSetCustomTraceFolder() + { + string customFolderName = ExportersBuilderTests.NewName(); + string traceFolder = Path.Combine(s_localApplicationDataFolderPath, customFolderName); + + if (Directory.Exists(traceFolder)) Directory.Delete(traceFolder, true); + try + { + using (TracerProvider provider = Sdk.CreateTracerProviderBuilder() + .AddSource(_activitySourceName) + .AddAdbcFileExporter(_activitySourceName, traceFolder) + .Build()) + { + await AddEvent("test"); + } + Assert.True(Directory.Exists(traceFolder)); + DirectoryInfo traceDirectory = new(traceFolder); + FileInfo[] files = traceDirectory.GetFiles(); + Assert.Single(files); + } + finally + { + if (Directory.Exists(traceFolder)) Directory.Delete(traceFolder, true); + } + } + + [Fact] + internal async Task CanSetCustomFileBaseName() + { + const string customFileBaseName = "custom-base-name"; + string customFolderName = ExportersBuilderTests.NewName(); + string traceFolder = Path.Combine(s_localApplicationDataFolderPath, customFolderName); + + if (Directory.Exists(traceFolder)) Directory.Delete(traceFolder, true); + try + { + using (TracerProvider provider = Sdk.CreateTracerProviderBuilder() + .AddSource(_activitySourceName) + .AddAdbcFileExporter(customFileBaseName, traceFolder) + .Build()) + { + await AddEvent("test"); + } + + Assert.True(Directory.Exists(traceFolder)); + DirectoryInfo traceDirectory = new(traceFolder); + FileInfo[] files = traceDirectory.GetFiles(); + Assert.Single(files); + Assert.StartsWith(customFileBaseName, files[0].Name); + } + finally + { + if (Directory.Exists(traceFolder)) Directory.Delete(traceFolder, true); + } + } + + [Fact] + internal async Task CanSetCustomMaxFileSize() + { + const long maxTraceFileSizeKb = 30; + const long kilobyte = 1024; + string customFolderName = ExportersBuilderTests.NewName(); + string traceFolder = Path.Combine(s_localApplicationDataFolderPath, customFolderName); + + if (Directory.Exists(traceFolder)) Directory.Delete(traceFolder, true); + try + { + TracerProviderBuilder x = Sdk.CreateTracerProviderBuilder(); + using (TracerProvider provider = Sdk.CreateTracerProviderBuilder() + .AddSource(_activitySourceName) + .AddAdbcFileExporter(_activitySourceName, traceFolder, maxTraceFileSizeKb) + .Build()) + { + for (int i = 0; i < 1000; i++) + { + await AddEvent("test"); + await Task.Delay(TimeSpan.FromMilliseconds(0.1)); + } + } + + Assert.True(Directory.Exists(traceFolder)); + DirectoryInfo traceDirectory = new(traceFolder); + string searchPattern = _activitySourceName + "-trace-*.log"; + FileInfo[] files = [.. traceDirectory + .EnumerateFiles(searchPattern, SearchOption.TopDirectoryOnly) + .OrderByDescending(f => f.LastWriteTimeUtc)]; + //FileInfo[] files = traceDirectory.GetFiles(); + Assert.True(files.Length > 2, $"actual # of trace files: {files.Length}"); + Assert.True(files.All(f => f.Name.StartsWith(_activitySourceName))); + StringBuilder summary = new(); + for (int i = 0; i < files.Length; i++) + { + summary.AppendLine($"{i}: {files[i].Name}: {files[i].Length}: {files[i].LastWriteTimeUtc}"); + } + for (int i = 0; i < files.Length; i++) + { + long expectedUpperSizeLimit = (maxTraceFileSizeKb + (long)(0.2 * maxTraceFileSizeKb)) * kilobyte; + Assert.True(files[i].Length < expectedUpperSizeLimit, summary.ToString()); + } + _outputHelper?.WriteLine($"number of files: {files.Length}"); + Console.WriteLine($"number of files: {files.Length}"); + } + finally + { + if (Directory.Exists(traceFolder)) Directory.Delete(traceFolder, true); + } + } + + [Fact] + internal async Task CanSetCustomMaxFiles() + { + const long maxTraceFileSizeKb = 5; + const int maxTraceFiles = 3; + string customFolderName = ExportersBuilderTests.NewName(); + string traceFolder = Path.Combine(s_localApplicationDataFolderPath, customFolderName); + + if (Directory.Exists(traceFolder)) Directory.Delete(traceFolder, true); + try + { + using (TracerProvider provider = Sdk.CreateTracerProviderBuilder() + .AddSource(_activitySourceName) + .AddAdbcFileExporter(_activitySourceName, traceFolder, maxTraceFileSizeKb, maxTraceFiles) + .Build()) + { + for (int i = 0; i < 1000; i++) + { + await AddEvent("test"); + await Task.Delay(TimeSpan.FromMilliseconds(0.1)); + } + } + + Assert.True(Directory.Exists(traceFolder)); + DirectoryInfo traceDirectory = new(traceFolder); + FileInfo[] files = traceDirectory.GetFiles(); + Assert.True(files.Length > 2, $"actual # of trace files: {files.Length}"); + Assert.True(files.Length <= maxTraceFiles, $"Expecting {maxTraceFiles} files. Actual # of files: {files.Length}"); + } + finally + { + if (Directory.Exists(traceFolder)) Directory.Delete(traceFolder, true); + } + } + + [Fact] + internal async Task CanSetSingleMaxFiles() + { + const long maxTraceFileSizeKb = 5; + const int maxTraceFiles = 1; + var delay = TimeSpan.FromSeconds(8); + string customFolderName = ExportersBuilderTests.NewName(); + string traceFolder = Path.Combine(s_localApplicationDataFolderPath, customFolderName); + + if (Directory.Exists(traceFolder)) Directory.Delete(traceFolder, true); + try + { + using (TracerProvider provider = Sdk.CreateTracerProviderBuilder() + .AddSource(_activitySourceName) + .AddAdbcFileExporter(_activitySourceName, traceFolder, maxTraceFileSizeKb, maxTraceFiles) + .Build()) + { + for (int i = 0; i < 100; i++) + { + await AddEvent("test"); + await Task.Delay(TimeSpan.FromMilliseconds(0.1)); + } + } + + Assert.True(Directory.Exists(traceFolder)); + DirectoryInfo traceDirectory = new(traceFolder); + FileInfo[] files = traceDirectory.GetFiles(); + Assert.Single(files); + } + finally + { + if (Directory.Exists(traceFolder)) Directory.Delete(traceFolder, true); + } + } + + [Theory] + [InlineData("", "abc", 1, 1, typeof(ArgumentNullException))] + [InlineData(" ", "abc", 1, 1, typeof(ArgumentNullException))] + [InlineData("abc", "abc", 0, 1, typeof(ArgumentException))] + [InlineData("abc", "abc", -1, 1, typeof(ArgumentException))] + [InlineData("abc", "abc", 1, 0, typeof(ArgumentException))] + [InlineData("abc", "abc", 1, -1, typeof(ArgumentException))] + internal void CanDetectInvalidOptions(string fileBaseName, string? traceLocation, long maxTraceFileSizeKb, int maxTraceFiles, Type expectedException) + { + string customFolderName = ExportersBuilderTests.NewName(); + string? traceFolder = traceLocation != null ? Path.Combine(s_localApplicationDataFolderPath, traceLocation) : null; + _ = Assert.Throws(expectedException, () => + Sdk.CreateTracerProviderBuilder() + .AddSource(_activitySourceName) + .AddAdbcFileExporter(fileBaseName, traceFolder, maxTraceFileSizeKb, maxTraceFiles) + .Build()); + } + + [Fact] + internal async Task CanTraceMultipleConcurrentWriters() + { + const int writeCount = 1000; + var delay = TimeSpan.FromSeconds(8); + string customFolderName = ExportersBuilderTests.NewName(); + string traceFolder = Path.Combine(s_localApplicationDataFolderPath, customFolderName); + + if (Directory.Exists(traceFolder)) Directory.Delete(traceFolder, true); + try + { + var tasks = new Task[] + { + Task.Run(async () => await TraceActivities(traceFolder, "activity1", writeCount)), + Task.Run(async () => await TraceActivities(traceFolder, "activity2", writeCount)), + }; + await Task.WhenAll(tasks); + + int activity1Count = 0; + int activity2Count = 0; + foreach (string file in Directory.GetFiles(traceFolder)) + { + foreach (string line in File.ReadLines(file, Encoding.UTF8)) + { + if (line.Contains("activity1")) + { + activity1Count++; + } + else if (line.Contains("activity2")) + { + activity2Count++; + } + } + } + // Note, because we don't reference count, one of the listeners will likely + // close the shared instance before the other is finished. + // That can result in some events not being written. + Assert.InRange(activity1Count, writeCount * 0.9, writeCount); + Assert.InRange(activity2Count, writeCount * 0.9, writeCount); + } + finally + { + if (Directory.Exists(traceFolder)) Directory.Delete(traceFolder, true); + } + } + + private async Task TraceActivities(string traceFolder, string activityName, int writeCount) + { + using (TracerProvider provider = Sdk.CreateTracerProviderBuilder() + .AddSource(_activitySourceName) + .AddAdbcFileExporter(_activitySourceName, traceFolder) + .Build()) + { + for (int i = 0; i < writeCount; i++) + { + await StartActivity(activityName); + await Task.Delay(TimeSpan.FromMilliseconds(0.1)); + } + } + } + + private Task AddEvent(string eventName, string activityName = nameof(AddEvent)) + { + using Activity? activity = _activitySource.StartActivity(activityName); + activity?.AddEvent(new ActivityEvent(eventName)); + return Task.CompletedTask; + } + + private Task StartActivity(string activityName = nameof(StartActivity)) + { + using Activity? activity = _activitySource.StartActivity(activityName); + return Task.CompletedTask; + } + + protected virtual void Dispose(bool disposing) + { + if (!_disposed) + { + if (disposing) + { + _activitySource.Dispose(); + } + _disposed = true; + } + } + + public void Dispose() + { + // Do not change this code. Put cleanup code in 'Dispose(bool disposing)' method + Dispose(disposing: true); + GC.SuppressFinalize(this); + } + } +} diff --git a/csharp/test/Telemetry/Traces/Exporters/FileExporter/TracingFileTests.cs b/csharp/test/Telemetry/Traces/Exporters/FileExporter/TracingFileTests.cs new file mode 100644 index 0000000000..99767756ea --- /dev/null +++ b/csharp/test/Telemetry/Traces/Exporters/FileExporter/TracingFileTests.cs @@ -0,0 +1,97 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You 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. + */ + +using System; +using System.Collections.Generic; +using System.IO; +using System.Runtime.CompilerServices; +using System.Threading; +using System.Threading.Tasks; +using Apache.Arrow.Adbc.Telemetry.Traces.Exporters.FileExporter; + +namespace Apache.Arrow.Adbc.Tests.Telemetry.Traces.Exporters.FileExporter +{ + public class TracingFileTests + { + private static readonly string s_localApplicationDataFolderPath = Environment.GetFolderPath(Environment.SpecialFolder.LocalApplicationData); + private int _testInstance = 0; + + /// + /// This test performs concurrent writes to the same trace file(s). + /// This isn't the likely intended use case, as the FileExporter will + /// instance by the combined key of folder and file base name. So there is + /// little to no change of concurrent writes to the same file. + /// + /// + [Fact] + internal async Task TestMultipleConcurrentTracingFiles() + { + CancellationTokenSource tokenSource = new CancellationTokenSource(); + int concurrentCount = 5; + Task[] tasks = new Task[concurrentCount]; + int[] lineCounts = new int[concurrentCount]; + string sourceName = ExportersBuilderTests.NewName(); + string customFolderName = ExportersBuilderTests.NewName(); + string traceFolder = Path.Combine(s_localApplicationDataFolderPath, customFolderName); + if (Directory.Exists(traceFolder)) Directory.Delete(traceFolder, true); + try + { + for (int i = 0; i < concurrentCount; i++) + { + tasks[i] = Task.Run(async () => await Run(sourceName, traceFolder, tokenSource.Token)); + } + await Task.WhenAll(tasks); + + foreach (var file in Directory.GetFiles(traceFolder)) + { + foreach (string line in File.ReadLines(file)) + { + Assert.StartsWith("line", line); + Assert.True(int.TryParse(line.Substring(4), out int writerNumber)); + Assert.InRange(writerNumber, 0, concurrentCount - 1); + lineCounts[writerNumber]++; + } + } + for (int i = 0; i < concurrentCount; i++) + { + Assert.True(100 == lineCounts[i], $"index {i} != 100. Actual {lineCounts[i]}"); + } + } + finally + { + if (Directory.Exists(traceFolder)) Directory.Delete(traceFolder, true); + } + } + + private async Task Run(string sourceName, string traceFolder, CancellationToken cancellationToken) + { + int instanceNumber = Interlocked.Increment(ref _testInstance) - 1; + TracingFile tracingFile = new TracingFile(sourceName, traceFolder); + await tracingFile.WriteLinesAsync(GetLinesAsync(instanceNumber, 100, cancellationToken), cancellationToken); + } + + private static async IAsyncEnumerable GetLinesAsync(int instanceNumber, int lineCount, [EnumeratorCancellation] CancellationToken cancellationToken = default) + { + for (int i = 0; i < lineCount; i++) + { + if (cancellationToken.IsCancellationRequested) yield break; + yield return new MemoryStream(System.Text.Encoding.UTF8.GetBytes($"line{instanceNumber}" + Environment.NewLine)); + await Task.Delay(10); // Simulate some delay + } + } + } +}