CurtHagenlocher commented on code in PR #2949: URL: https://github.com/apache/arrow-adbc/pull/2949#discussion_r2193102926
########## csharp/src/Telemetry/Traces/Exporters/Apache.Arrow.Adbc.Telemetry.Traces.Exporters.csproj: ########## @@ -0,0 +1,25 @@ +<Project Sdk="Microsoft.NET.Sdk"> + + <PropertyGroup> + <TargetFrameworks>netstandard2.0;net6.0</TargetFrameworks> Review Comment: Consider targeting net8.0 instead of net6.0 as we're about to update the other projects. ########## csharp/src/Telemetry/Traces/Exporters/FileExporter/TracingFile.cs: ########## @@ -0,0 +1,233 @@ +/* + * 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.Linq; +using System.Threading; +using System.Threading.Tasks; + +namespace Apache.Arrow.Adbc.Telemetry.Traces.Exporters.FileExporter +{ + /// <summary> + /// Provides access to writing trace files, limiting the + /// individual files size and ensuring unique file names. + /// </summary> + internal class TracingFile : IDisposable + { + private static readonly string s_defaultTracePath = FileExporter.TracingLocationDefault; + private readonly string _fileBaseName; + private readonly DirectoryInfo _tracingDirectory; + private FileInfo? _currentTraceFileInfo; + private bool _disposedValue; + private readonly long _maxFileSizeKb = FileExporter.MaxFileSizeKbDefault; + private readonly int _maxTraceFiles = FileExporter.MaxTraceFilesDefault; + + internal TracingFile(string fileBaseName, string? traceDirectoryPath = default, long maxFileSizeKb = FileExporter.MaxFileSizeKbDefault, int maxTraceFiles = FileExporter.MaxTraceFilesDefault) : + this(fileBaseName, traceDirectoryPath == null ? new DirectoryInfo(s_defaultTracePath) : new DirectoryInfo(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(); + } + + /// <summary> + /// Writes lines of trace where each stream is a line in the trace file. + /// </summary> + /// <param name="streams">The enumerable of trace lines.</param> + /// <param name="cancellationToken">The cancellation token.</param> + /// <returns></returns> + internal async Task WriteLinesAsync(IAsyncEnumerable<Stream> streams, CancellationToken cancellationToken = default) + { + if (cancellationToken.IsCancellationRequested) return; + + string searchPattern = _fileBaseName + "-trace-*.log"; + if (_currentTraceFileInfo == null) + { + IOrderedEnumerable<FileInfo>? traceFileInfos = GetTracingFiles(_tracingDirectory, searchPattern); + 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<IOException>(async () => await WriteLinesAsync(streams), cancellationToken: cancellationToken); + + // Check if we need to remove old files + if (_tracingDirectory.Exists) + { + FileInfo[] tracingFiles = [.. GetTracingFiles(_tracingDirectory, searchPattern)]; + 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 tokenm, as we want this to ALWAYS run at the end. + await ActionWithRetryAsync<IOException>(() => file?.Delete()); + } + } + } + } + + private async Task WriteLinesAsync(IAsyncEnumerable<Stream> 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); + } + } + if (newFileRequired) + { + // If tracing file is maxxed-out, start a new tracing file. + _currentTraceFileInfo = new FileInfo(NewFileName()); + } + } while (hasMoreData); + } + + private static IOrderedEnumerable<FileInfo> GetTracingFiles(DirectoryInfo tracingDirectory, string searchPattern) + { + return tracingDirectory + .EnumerateFiles(searchPattern, SearchOption.TopDirectoryOnly) + .OrderByDescending(f => f.LastWriteTimeUtc); + } + + private static async Task ActionWithRetryAsync<T>(Action action, int maxRetries = 5, CancellationToken cancellationToken = default) where T : Exception + { + int retryCount = 0; + TimeSpan pauseTime = TimeSpan.FromMilliseconds(10); + bool completed = false; + + while (!cancellationToken.IsCancellationRequested && !completed && retryCount < maxRetries) + { + try + { + action.Invoke(); + completed = true; + } + catch (T) + { + retryCount++; + if (retryCount >= maxRetries) + { + throw; + } + try + { + await Task.Delay(pauseTime, cancellationToken); + } + catch (OperationCanceledException) + { + // Need to catch this exception or it will be propagated. + break; + } + } + } + } + + private static async Task ActionWithRetryAsync<T>( + Func<Task> action, + int maxRetries = 5, + CancellationToken cancellationToken = default) where T : Exception + { + int retryCount = 0; + TimeSpan pauseTime = TimeSpan.FromMilliseconds(10); + bool completed = false; + + while (!cancellationToken.IsCancellationRequested && !completed && retryCount < maxRetries) + { + try + { + await action.Invoke(); + completed = true; + } + catch (T) + { + retryCount++; + if (retryCount >= maxRetries) + { + throw; + } + 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}.log"); + } + + private void EnsureTraceDirectory() + { + if (!Directory.Exists(_tracingDirectory.FullName)) + { + Directory.CreateDirectory(_tracingDirectory.FullName); + } + } + + protected virtual void Dispose(bool disposing) + { + if (!_disposedValue && disposing) + { + _disposedValue = true; Review Comment: This is an internal class with no derived classes and `Dispose` does nothing. Does it need to be `IDisposable`? ########## csharp/src/Telemetry/Traces/Exporters/FileExporter/TracingFile.cs: ########## @@ -0,0 +1,233 @@ +/* + * 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.Linq; +using System.Threading; +using System.Threading.Tasks; + +namespace Apache.Arrow.Adbc.Telemetry.Traces.Exporters.FileExporter +{ + /// <summary> + /// Provides access to writing trace files, limiting the + /// individual files size and ensuring unique file names. + /// </summary> + internal class TracingFile : IDisposable + { + private static readonly string s_defaultTracePath = FileExporter.TracingLocationDefault; + private readonly string _fileBaseName; + private readonly DirectoryInfo _tracingDirectory; + private FileInfo? _currentTraceFileInfo; + private bool _disposedValue; + private readonly long _maxFileSizeKb = FileExporter.MaxFileSizeKbDefault; + private readonly int _maxTraceFiles = FileExporter.MaxTraceFilesDefault; + + internal TracingFile(string fileBaseName, string? traceDirectoryPath = default, long maxFileSizeKb = FileExporter.MaxFileSizeKbDefault, int maxTraceFiles = FileExporter.MaxTraceFilesDefault) : + this(fileBaseName, traceDirectoryPath == null ? new DirectoryInfo(s_defaultTracePath) : new DirectoryInfo(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(); + } + + /// <summary> + /// Writes lines of trace where each stream is a line in the trace file. + /// </summary> + /// <param name="streams">The enumerable of trace lines.</param> + /// <param name="cancellationToken">The cancellation token.</param> + /// <returns></returns> + internal async Task WriteLinesAsync(IAsyncEnumerable<Stream> streams, CancellationToken cancellationToken = default) + { + if (cancellationToken.IsCancellationRequested) return; + + string searchPattern = _fileBaseName + "-trace-*.log"; + if (_currentTraceFileInfo == null) + { + IOrderedEnumerable<FileInfo>? traceFileInfos = GetTracingFiles(_tracingDirectory, searchPattern); + 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<IOException>(async () => await WriteLinesAsync(streams), cancellationToken: cancellationToken); + + // Check if we need to remove old files + if (_tracingDirectory.Exists) + { + FileInfo[] tracingFiles = [.. GetTracingFiles(_tracingDirectory, searchPattern)]; + 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 tokenm, as we want this to ALWAYS run at the end. + await ActionWithRetryAsync<IOException>(() => file?.Delete()); + } + } + } + } + + private async Task WriteLinesAsync(IAsyncEnumerable<Stream> 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); + } + } + if (newFileRequired) + { + // If tracing file is maxxed-out, start a new tracing file. + _currentTraceFileInfo = new FileInfo(NewFileName()); + } + } while (hasMoreData); + } + + private static IOrderedEnumerable<FileInfo> GetTracingFiles(DirectoryInfo tracingDirectory, string searchPattern) + { + return tracingDirectory + .EnumerateFiles(searchPattern, SearchOption.TopDirectoryOnly) Review Comment: Is there no way to enumerate the files in an async fashion? ########## csharp/src/Telemetry/Traces/Exporters/FileExporter/FileExporter.cs: ########## @@ -0,0 +1,263 @@ +/* + * 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<Activity> + { + 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 string s_tracingLocationDefault = TracingLocationDefault; + private static readonly ConcurrentDictionary<string, Lazy<FileExporterInstance>> 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<Activity> _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 ?? s_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<FileExporterInstance>(() => + { + 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<Activity> 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) + { + 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); + } + } + + 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<Stream> GetActivitiesAsync(ConcurrentQueue<Activity> activityQueue) + { + MemoryStream stream = new MemoryStream(); + while (activityQueue.TryDequeue(out Activity? activity)) + { + stream.SetLength(0); + SerializableActivity serializableActivity = new(activity); + await JsonSerializer.SerializeAsync(stream, 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 => + 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 flushTimeoutTS = new(); Review Comment: Was this supposed to be `flushTimeoutCTS`? Can it be just `flushTimeout`? ########## csharp/src/Telemetry/Traces/Exporters/FileExporter/TracingFile.cs: ########## @@ -0,0 +1,233 @@ +/* + * 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.Linq; +using System.Threading; +using System.Threading.Tasks; + +namespace Apache.Arrow.Adbc.Telemetry.Traces.Exporters.FileExporter +{ + /// <summary> + /// Provides access to writing trace files, limiting the + /// individual files size and ensuring unique file names. + /// </summary> + internal class TracingFile : IDisposable + { + private static readonly string s_defaultTracePath = FileExporter.TracingLocationDefault; + private readonly string _fileBaseName; + private readonly DirectoryInfo _tracingDirectory; + private FileInfo? _currentTraceFileInfo; + private bool _disposedValue; + private readonly long _maxFileSizeKb = FileExporter.MaxFileSizeKbDefault; + private readonly int _maxTraceFiles = FileExporter.MaxTraceFilesDefault; + + internal TracingFile(string fileBaseName, string? traceDirectoryPath = default, long maxFileSizeKb = FileExporter.MaxFileSizeKbDefault, int maxTraceFiles = FileExporter.MaxTraceFilesDefault) : + this(fileBaseName, traceDirectoryPath == null ? new DirectoryInfo(s_defaultTracePath) : new DirectoryInfo(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(); + } + + /// <summary> + /// Writes lines of trace where each stream is a line in the trace file. + /// </summary> + /// <param name="streams">The enumerable of trace lines.</param> + /// <param name="cancellationToken">The cancellation token.</param> + /// <returns></returns> + internal async Task WriteLinesAsync(IAsyncEnumerable<Stream> streams, CancellationToken cancellationToken = default) + { + if (cancellationToken.IsCancellationRequested) return; + + string searchPattern = _fileBaseName + "-trace-*.log"; + if (_currentTraceFileInfo == null) + { + IOrderedEnumerable<FileInfo>? traceFileInfos = GetTracingFiles(_tracingDirectory, searchPattern); + 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<IOException>(async () => await WriteLinesAsync(streams), cancellationToken: cancellationToken); + + // Check if we need to remove old files + if (_tracingDirectory.Exists) + { + FileInfo[] tracingFiles = [.. GetTracingFiles(_tracingDirectory, searchPattern)]; + 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 tokenm, as we want this to ALWAYS run at the end. + await ActionWithRetryAsync<IOException>(() => file?.Delete()); + } + } + } + } + + private async Task WriteLinesAsync(IAsyncEnumerable<Stream> 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); + } + } + if (newFileRequired) + { + // If tracing file is maxxed-out, start a new tracing file. + _currentTraceFileInfo = new FileInfo(NewFileName()); + } + } while (hasMoreData); + } + + private static IOrderedEnumerable<FileInfo> GetTracingFiles(DirectoryInfo tracingDirectory, string searchPattern) + { + return tracingDirectory + .EnumerateFiles(searchPattern, SearchOption.TopDirectoryOnly) + .OrderByDescending(f => f.LastWriteTimeUtc); + } + + private static async Task ActionWithRetryAsync<T>(Action action, int maxRetries = 5, CancellationToken cancellationToken = default) where T : Exception + { + int retryCount = 0; + TimeSpan pauseTime = TimeSpan.FromMilliseconds(10); + bool completed = false; + + while (!cancellationToken.IsCancellationRequested && !completed && retryCount < maxRetries) + { + try + { + action.Invoke(); + completed = true; + } + catch (T) Review Comment: ```suggestion catch (T) if (retryCount < maxRetries) ``` ########## csharp/src/Telemetry/Traces/Exporters/FileExporter/TracingFile.cs: ########## @@ -0,0 +1,233 @@ +/* + * 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.Linq; +using System.Threading; +using System.Threading.Tasks; + +namespace Apache.Arrow.Adbc.Telemetry.Traces.Exporters.FileExporter +{ + /// <summary> + /// Provides access to writing trace files, limiting the + /// individual files size and ensuring unique file names. Review Comment: What would happen if two instances of this were pointed at the same directory? ########## csharp/src/Telemetry/Traces/Exporters/FileExporter/TracingFile.cs: ########## @@ -0,0 +1,233 @@ +/* + * 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.Linq; +using System.Threading; +using System.Threading.Tasks; + +namespace Apache.Arrow.Adbc.Telemetry.Traces.Exporters.FileExporter +{ + /// <summary> + /// Provides access to writing trace files, limiting the + /// individual files size and ensuring unique file names. + /// </summary> + internal class TracingFile : IDisposable + { + private static readonly string s_defaultTracePath = FileExporter.TracingLocationDefault; + private readonly string _fileBaseName; + private readonly DirectoryInfo _tracingDirectory; + private FileInfo? _currentTraceFileInfo; + private bool _disposedValue; + private readonly long _maxFileSizeKb = FileExporter.MaxFileSizeKbDefault; + private readonly int _maxTraceFiles = FileExporter.MaxTraceFilesDefault; + + internal TracingFile(string fileBaseName, string? traceDirectoryPath = default, long maxFileSizeKb = FileExporter.MaxFileSizeKbDefault, int maxTraceFiles = FileExporter.MaxTraceFilesDefault) : + this(fileBaseName, traceDirectoryPath == null ? new DirectoryInfo(s_defaultTracePath) : new DirectoryInfo(traceDirectoryPath), maxFileSizeKb, maxTraceFiles) Review Comment: Consider lifting the creation of the `DirectoryInfo` into a static helper function for readability. ########## csharp/test/Telemetry/Traces/Exporters/FileExporter/FileExporterTests.cs: ########## @@ -0,0 +1,277 @@ +/* +* 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()); + } + + 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; + Review Comment: nit: extra blank line ########## csharp/src/Telemetry/Traces/Exporters/ExportersBuilder.cs: ########## @@ -0,0 +1,214 @@ +/* + * 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<string, Func<string, string?, TracerProvider?>> s_tracerProviderFactoriesDefault; + + private readonly string _sourceName; + private readonly string? _sourceVersion; + private readonly IReadOnlyDictionary<string, Func<string, string?, TracerProvider?>> _tracerProviderFactories; + + static ExportersBuilder() + { + var defaultProviers = new Dictionary<string, Func<string, string?, TracerProvider?>> + { + [ExportersOptions.Exporters.None] = NewNoopTracerProvider, + [ExportersOptions.Exporters.Otlp] = NewOtlpTracerProvider, + [ExportersOptions.Exporters.Console] = NewConsoleTracerProvider, + [ExportersOptions.Exporters.AdbcFile] = NewAdbcFileTracerProvider, + }; + s_tracerProviderFactoriesDefault = defaultProviers; + } + + private ExportersBuilder(IBuilder builder) + { + _sourceName = builder.SourceName; + _sourceVersion = builder.SourceVersion; + _tracerProviderFactories = builder.TracerProviderFactories; + } + + /// <summary> + /// Build an <see cref="ExportersBuilder"/> from different possible Exporters. Use the Add* functions to add + /// one or more <see cref="TracerProvider"/> factories. + /// </summary> + /// <param name="sourceName">The name of the source that the exporter will filter on.</param> + /// <param name="sourceVersion">The (optional) version of the source that the exporter will filter on.</param> + /// <returns>A <see cref="Builder"/> object.</returns> + /// <exception cref="ArgumentNullException"></exception> + public static Builder Build(string sourceName, string? sourceVersion = default) + { + if (string.IsNullOrWhiteSpace(sourceName)) + { + throw new ArgumentNullException(nameof(sourceName)); + } + return new Builder(sourceName, sourceVersion); + } + + /// <summary> + /// <para> + /// Attempts to activate an exporter based on the dictionary of <see cref="TracerProvider"/> factories + /// added to the <see cref="Builder"/>. 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 <see cref="ExportersOptions.Environment.Exporter"/>). + /// 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. + /// </para> + /// <para> + /// 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. + /// </para> + /// </summary> + /// <param name="exporterOption">The value (name) of the exporter option, typically passed as option <see cref="ExportersOptions.Exporter"/>.</param> + /// <param name="exporterName">The actual exporter name when successfully activated.</param> + /// <param name="environmentName">The (optional) name of the environment variable to test for the exporter name. Default: <see cref="ExportersOptions.Environment.Exporter"/></param> + /// <exception cref="AdbcException" /> + /// <returns>The a non-null <see cref="TracerProvider"/> when successfully activated. Note: this object must be explicitly disposed when no longer necessary.</returns> + 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<string, string?, TracerProvider?>? 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; + + private interface IBuilder Review Comment: What's the advantage in defining a private interface here? ########## csharp/src/Telemetry/Traces/Exporters/ExportersBuilder.cs: ########## @@ -0,0 +1,214 @@ +/* + * 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<string, Func<string, string?, TracerProvider?>> s_tracerProviderFactoriesDefault; + + private readonly string _sourceName; + private readonly string? _sourceVersion; + private readonly IReadOnlyDictionary<string, Func<string, string?, TracerProvider?>> _tracerProviderFactories; + + static ExportersBuilder() + { + var defaultProviers = new Dictionary<string, Func<string, string?, TracerProvider?>> Review Comment: ```suggestion var defaultProviders = new Dictionary<string, Func<string, string?, TracerProvider?>> ``` ########## csharp/test/Telemetry/Traces/Exporters/FileExporter/FileExporterTests.cs: ########## @@ -0,0 +1,277 @@ +/* +* Licensed to the Apache Software Foundation (ASF) under one or more Review Comment: nit: fix indentation -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: github-unsubscr...@arrow.apache.org For queries about this service, please contact Infrastructure at: us...@infra.apache.org
