Add ElasticsearchSinkOptions.BufferFileRollingInterval option (#416)

* Add `ElasticsearchSinkOptions.BufferFileRollingInterval` option

- Using this option we can customize buffer file rolling interval. The default is `RollingInterval.Day` (so no changes here). In some cases higher granularity may be needed.
- Changed regular expression for FileSet to get buffer files to support different rolling interval file name formats - from Infinite to Minute. All of them are different amount of digits representing date - 0(Infinite), 4(Year),..., 12 (Minute). So replaced expression part for day format `(?<date>\\d{8})` with the expression for all interval date `(?<date>\\d{0,12})`.

* Add tests for the desired functionality, which fail now.

- Return code to support only Daily rolling interval
- Add RollingIntervalExtensions.cs (origin: Serilog.Sinks.File) with InternalVisible attribute to be able to test
- Add InternalVisible to FileSet.cs to be able to test it
- Sign tests assembly the same way as Serilog.Sinks.Elasticsearch for InternalVisible to work

* Support different rolling intervals for DurableElasticsearchSink rolling files.

- Make support only for intervals like Day, Hour, Minute. As for less frequent intervals we cannot get specific date (specific day) for passing to _getIndexForEvent in `ElasticsearchPayloadReader`.
- Support handling rolling files for different intervals in `ElasticsearchPayloadReader` and `FileSet` by using corresponding formats and search patterns.
- Add tests of changed code - for `ElasticsearchPayloadReader` and `FileSet`

* Remove redundant spaces

Author: andrey-kozlov-skuvault

src/Serilog.Sinks.Elasticsearch/Serilog.Sinks.Elasticsearch.csproj

@@ -55,4 +55,10 @@
         <PackageReference Include="Microsoft.CSharp" Version="4.6.0" />
     </ItemGroup>
 
+  <ItemGroup>
+    <AssemblyAttribute Include="System.Runtime.CompilerServices.InternalsVisibleTo">
+      <_Parameter1>Serilog.Sinks.Elasticsearch.Tests, PublicKey=0024000004800000940000000602000000240000525341310004000001000100fb8d13fd344a1c6fe0fe83ef33c1080bf30690765bc6eb0df26ebfdf8f21670c64265b30db09f73a0dea5b3db4c9d18dbf6d5a25af5ce9016f281014d79dc3b4201ac646c451830fc7e61a2dfd633d34c39f87b81894191652df5ac63cc40c77f3542f702bda692e6e8a9158353df189007a49da0f3cfd55eb250066b19485ec</_Parameter1>
+    </AssemblyAttribute>
+  </ItemGroup>
+
 </Project>

src/Serilog.Sinks.Elasticsearch/Sinks/ElasticSearch/Durable/Elasticsearch/DurableElasticsearchSink.cs

@@ -39,19 +39,17 @@ public DurableElasticsearchSink(ElasticsearchSinkOptions options)
                 throw new ArgumentException("Cannot create the durable ElasticSearch sink without a buffer base file name!");
             }
 
-            
             _sink = new LoggerConfiguration()
                 .MinimumLevel.Verbose()
                 .WriteTo.File(_state.DurableFormatter,
                     options.BufferBaseFilename + FileNameSuffix,
-                    rollingInterval: RollingInterval.Day,
+                    rollingInterval: options.BufferFileRollingInterval,
                     fileSizeLimitBytes: options.BufferFileSizeLimitBytes,
                     rollOnFileSizeLimit: true,
                     retainedFileCountLimit: options.BufferFileCountLimit,
                     levelSwitch: _state.Options.LevelSwitch,
                     encoding: Encoding.UTF8)
                 .CreateLogger();
-
 
             var elasticSearchLogClient = new ElasticsearchLogClient(
                 elasticLowLevelClient: _state.Client, 
@@ -63,7 +61,8 @@ public DurableElasticsearchSink(ElasticsearchSinkOptions options)
                  typeName:_state.Options.TypeName, 
                  serialize:_state.Serialize,  
                  getIndexForEvent: _state.GetBufferedIndexForEvent,
-                 elasticOpType: _state.Options.BatchAction);
+                 elasticOpType: _state.Options.BatchAction,
+                 rollingInterval: options.BufferFileRollingInterval);
 
             _shipper = new ElasticsearchLogShipper(
                 bufferBaseFilename: _state.Options.BufferBaseFilename,
@@ -75,7 +74,8 @@ public DurableElasticsearchSink(ElasticsearchSinkOptions options)
                 payloadReader: payloadReader,
                 retainedInvalidPayloadsLimitBytes: _state.Options.BufferRetainedInvalidPayloadsLimitBytes,
                 bufferSizeLimitBytes: _state.Options.BufferFileSizeLimitBytes,
-                registerTemplateIfNeeded: _state.RegisterTemplateIfNeeded);
+                registerTemplateIfNeeded: _state.RegisterTemplateIfNeeded,
+                rollingInterval: options.BufferFileRollingInterval);
 
         }
 

src/Serilog.Sinks.Elasticsearch/Sinks/ElasticSearch/Durable/Elasticsearch/ElasticsearchLogShipper.cs

@@ -30,12 +30,13 @@ public class ElasticsearchLogShipper : LogShipper<List<string>>
         /// <param name="retainedInvalidPayloadsLimitBytes"></param>
         /// <param name="bufferSizeLimitBytes"></param>
         /// <param name="registerTemplateIfNeeded"></param>
+        /// <param name="rollingInterval"></param>
         public ElasticsearchLogShipper(string bufferBaseFilename, int batchPostingLimit, TimeSpan period,
             long? eventBodyLimitBytes, LoggingLevelSwitch levelControlSwitch, ILogClient<List<string>> logClient,
             IPayloadReader<List<string>> payloadReader, long? retainedInvalidPayloadsLimitBytes,
-            long? bufferSizeLimitBytes, Action registerTemplateIfNeeded)
-            : base(bufferBaseFilename, batchPostingLimit, period, eventBodyLimitBytes,
-            levelControlSwitch, logClient, payloadReader, retainedInvalidPayloadsLimitBytes, bufferSizeLimitBytes)
+            long? bufferSizeLimitBytes, Action registerTemplateIfNeeded, RollingInterval rollingInterval)
+            : base(bufferBaseFilename, batchPostingLimit, period, eventBodyLimitBytes, levelControlSwitch, logClient, 
+                payloadReader, retainedInvalidPayloadsLimitBytes, bufferSizeLimitBytes, rollingInterval)
         {
             _registerTemplateIfNeeded = registerTemplateIfNeeded;                        
         }

src/Serilog.Sinks.Elasticsearch/Sinks/ElasticSearch/Durable/Elasticsearch/ElasticsearchPayloadReader.cs

@@ -17,6 +17,7 @@ public class ElasticsearchPayloadReader: APayloadReader<List<string>>
         private readonly Func<object, string> _serialize;
         private readonly Func<string, DateTime,string> _getIndexForEvent;
         private readonly ElasticOpType _elasticOpType;
+        private readonly RollingInterval _rollingInterval;
         private List<string> _payload;
         private int _count;
         private DateTime _date;
@@ -29,14 +30,21 @@ public class ElasticsearchPayloadReader: APayloadReader<List<string>>
         /// <param name="serialize"></param>
         /// <param name="getIndexForEvent"></param>
         /// <param name="elasticOpType"></param>
+        /// <param name="rollingInterval"></param>
         public ElasticsearchPayloadReader(string pipelineName, string typeName, Func<object, string> serialize,
-            Func<string, DateTime, string> getIndexForEvent, ElasticOpType elasticOpType)
+            Func<string, DateTime, string> getIndexForEvent, ElasticOpType elasticOpType, RollingInterval rollingInterval)
         {
+            if ((int)rollingInterval < (int)RollingInterval.Day)
+            {
+                throw new ArgumentException("Rolling intervals less frequent than RollingInterval.Day are not supported");
+            }
+            
             _pipelineName = pipelineName;
             _typeName = typeName;
             _serialize = serialize;
             _getIndexForEvent = getIndexForEvent;
             _elasticOpType = elasticOpType;
+            _rollingInterval = rollingInterval;
         }
 
         /// <summary>
@@ -64,8 +72,9 @@ protected override void InitPayLoad(string filename)
                 throw new FormatException(string.Format("The file name '{0}' does not seem to follow the right file pattern - it must be named [whatever]-{{Date}}[_n].json", Path.GetFileName(filename)));
             }
 
-            var dateString = lastToken.Substring(0, 8);
-            _date = DateTime.ParseExact(dateString, "yyyyMMdd", CultureInfo.InvariantCulture);            
+            var dateFormat = _rollingInterval.GetFormat();
+            var dateString = lastToken.Substring(0, dateFormat.Length);
+            _date = DateTime.ParseExact(dateString, dateFormat, CultureInfo.InvariantCulture);
         }
        /// <summary>
        /// 

src/Serilog.Sinks.Elasticsearch/Sinks/ElasticSearch/Durable/Elasticsearch/RollingIntervalExtensions.cs

@@ -0,0 +1,55 @@
+using System;
+using System.Runtime.CompilerServices;
+
+[assembly:
+    InternalsVisibleTo(
+        "Serilog.Sinks.Elasticsearch.Tests, PublicKey=0024000004800000940000000602000000240000525341310004000001000100fb8d13fd344a1c6fe0fe83ef33c1080bf30690765bc6eb0df26ebfdf8f21670c64265b30db09f73a0dea5b3db4c9d18dbf6d5a25af5ce9016f281014d79dc3b4201ac646c451830fc7e61a2dfd633d34c39f87b81894191652df5ac63cc40c77f3542f702bda692e6e8a9158353df189007a49da0f3cfd55eb250066b19485ec")]
+
+namespace Serilog.Sinks.Elasticsearch.Durable
+{
+    internal static class RollingIntervalExtensions
+    {
+        // From https://github.com/serilog/serilog-sinks-file/blob/dev/src/Serilog.Sinks.File/Sinks/File/RollingIntervalExtensions.cs#L19
+        public static string GetFormat(this RollingInterval interval)
+        {
+            switch (interval)
+            {
+                case RollingInterval.Infinite:
+                    return "";
+                case RollingInterval.Year:
+                    return "yyyy";
+                case RollingInterval.Month:
+                    return "yyyyMM";
+                case RollingInterval.Day:
+                    return "yyyyMMdd";
+                case RollingInterval.Hour:
+                    return "yyyyMMddHH";
+                case RollingInterval.Minute:
+                    return "yyyyMMddHHmm";
+                default:
+                    throw new ArgumentException("Invalid rolling interval");
+            }
+        }
+
+        public static string GetMatchingDateRegularExpressionPart(this RollingInterval interval)
+        {
+            switch (interval)
+            {
+                case RollingInterval.Infinite:
+                    return "";
+                case RollingInterval.Year:
+                    return "\\d{4}";
+                case RollingInterval.Month:
+                    return "\\d{6}";
+                case RollingInterval.Day:
+                    return "\\d{8}";
+                case RollingInterval.Hour:
+                    return "\\d{10}";
+                case RollingInterval.Minute:
+                    return "\\d{12}";
+                default:
+                    throw new ArgumentException("Invalid rolling interval");
+            }
+        }
+    }
+}

src/Serilog.Sinks.Elasticsearch/Sinks/ElasticSearch/Durable/FileSet.cs

@@ -19,9 +19,12 @@
 using System.IO;
 using System.Linq;
 using System.Net;
+using System.Runtime.CompilerServices;
 using System.Text.RegularExpressions;
 using Serilog.Debugging;
-
+[assembly:
+    InternalsVisibleTo(
+        "Serilog.Sinks.Elasticsearch.Tests, PublicKey=0024000004800000940000000602000000240000525341310004000001000100fb8d13fd344a1c6fe0fe83ef33c1080bf30690765bc6eb0df26ebfdf8f21670c64265b30db09f73a0dea5b3db4c9d18dbf6d5a25af5ce9016f281014d79dc3b4201ac646c451830fc7e61a2dfd633d34c39f87b81894191652df5ac63cc40c77f3542f702bda692e6e8a9158353df189007a49da0f3cfd55eb250066b19485ec")]
 namespace Serilog.Sinks.Elasticsearch.Durable
 {
     /// <summary>
@@ -36,14 +39,16 @@ class FileSet
 
         const string InvalidPayloadFilePrefix = "invalid-";
 
-        public FileSet(string bufferBaseFilename)
+        public FileSet(string bufferBaseFilename, RollingInterval rollingInterval)
         {
             if (bufferBaseFilename == null) throw new ArgumentNullException(nameof(bufferBaseFilename));
 
             _bookmarkFilename = Path.GetFullPath(bufferBaseFilename + ".bookmark");
             _logFolder = Path.GetDirectoryName(_bookmarkFilename);
             _candidateSearchPath = Path.GetFileName(bufferBaseFilename) + "-*.json";
-            _filenameMatcher = new Regex("^" + Regex.Escape(Path.GetFileName(bufferBaseFilename)) + "-(?<date>\\d{8})(?<sequence>_[0-9]{3,}){0,1}\\.json$");
+            var dateRegularExpressionPart = rollingInterval.GetMatchingDateRegularExpressionPart();
+            _filenameMatcher = new Regex("^" + Regex.Escape(Path.GetFileName(bufferBaseFilename)) + "-(?<date>" 
+                                         + dateRegularExpressionPart + ")(?<sequence>_[0-9]{3,}){0,1}\\.json$");
         }
 
         public BookmarkFile OpenBookmarkFile()

src/Serilog.Sinks.Elasticsearch/Sinks/ElasticSearch/Durable/LogShipper.cs

@@ -76,6 +76,7 @@ public class LogShipper<TPayload> : IDisposable
         /// <param name="payloadReader"></param>
         /// <param name="retainedInvalidPayloadsLimitBytes"></param>
         /// <param name="bufferSizeLimitBytes"></param>
+        /// <param name="rollingInterval"></param>
         public LogShipper(            
             string bufferBaseFilename,            
             int batchPostingLimit,
@@ -85,7 +86,8 @@ public LogShipper(
             ILogClient<TPayload> logClient,
             IPayloadReader<TPayload> payloadReader,
             long? retainedInvalidPayloadsLimitBytes,
-            long? bufferSizeLimitBytes)
+            long? bufferSizeLimitBytes,
+            RollingInterval rollingInterval = RollingInterval.Day)
         {
             _batchPostingLimit = batchPostingLimit;
             _eventBodyLimitBytes = eventBodyLimitBytes;
@@ -95,7 +97,7 @@ public LogShipper(
             _connectionSchedule = new ExponentialBackoffConnectionSchedule(period);
             _retainedInvalidPayloadsLimitBytes = retainedInvalidPayloadsLimitBytes;
             _bufferSizeLimitBytes = bufferSizeLimitBytes;
-            _fileSet = new FileSet(bufferBaseFilename);
+            _fileSet = new FileSet(bufferBaseFilename, rollingInterval);
             _timer = new PortableTimer(c => OnTick());
             SetTimer();
 

src/Serilog.Sinks.Elasticsearch/Sinks/ElasticSearch/ElasticsearchSinkOptions.cs

@@ -273,6 +273,13 @@ public int QueueSizeLimit
         /// When set to true splits the StackTrace by new line and writes it as a an array of strings.
         /// </summary>
         public bool FormatStackTraceAsArray { get; set; }
+        
+        /// <summary>
+        /// The interval at which buffer log files will roll over to a new file. The default is <see cref="RollingInterval.Day"/>.
+        /// Less frequent intervals like <see cref="RollingInterval.Infinite"/>, <see cref="RollingInterval.Year"/>,
+        /// <see cref="RollingInterval.Month"/> are not supported.
+        /// </summary>
+        public RollingInterval BufferFileRollingInterval { get; set; }
 
         /// <summary>
         /// Configures the elasticsearch sink defaults
@@ -294,6 +301,7 @@ public ElasticsearchSinkOptions()
             this.BufferFileSizeLimitBytes = 100L * 1024 * 1024;
             this.FormatStackTraceAsArray = false;
             this.ConnectionPool = new SingleNodeConnectionPool(_defaultNode);
+            this.BufferFileRollingInterval = RollingInterval.Day;
         }
 
         /// <summary>

test/Serilog.Sinks.Elasticsearch.Tests/ElasticsearchPayloadReaderTests.cs

@@ -0,0 +1,86 @@
+using System;
+using System.IO;
+using System.Text;
+using FluentAssertions;
+using Serilog.Sinks.Elasticsearch.Durable;
+using Xunit;
+
+namespace Serilog.Sinks.Elasticsearch.Tests;
+
+public class ElasticsearchPayloadReaderTests : IDisposable
+{
+    private readonly string _tempFileFullPathTemplate;
+    private string _bufferFileName;
+
+    public ElasticsearchPayloadReaderTests()
+    {
+        _tempFileFullPathTemplate = Path.Combine(Path.GetTempPath(), Guid.NewGuid().ToString("N")) + "-{0}.json";
+    }
+
+    public void Dispose()
+    {
+        if (!string.IsNullOrEmpty(_bufferFileName))
+        {
+            System.IO.File.Delete(_bufferFileName);
+        }
+    }
+
+    [Theory]
+    [InlineData(RollingInterval.Day)]
+    [InlineData(RollingInterval.Hour)]
+    [InlineData(RollingInterval.Minute)]
+    public void ReadPayload_ShouldReadSpecifiedTypesOfRollingFile(RollingInterval rollingInterval)
+    {
+        // Arrange
+        var format = rollingInterval.GetFormat();
+        var payloadReader = new ElasticsearchPayloadReader("testPipelineName",
+            "TestTypeName",
+            null,
+            (_, _) => "TestIndex",
+            ElasticOpType.Index,
+            rollingInterval);
+        var lines = new[]
+        {
+            rollingInterval.ToString()
+        };
+        _bufferFileName = string.Format(_tempFileFullPathTemplate,
+            string.IsNullOrEmpty(format) ? string.Empty : new DateTime(2000, 1, 1).ToString(format));
+        // Important to use UTF8 with BOM if we are starting from 0 position 
+        System.IO.File.WriteAllLines(_bufferFileName, lines, new UTF8Encoding(true));
+
+        // Act
+        var fileSetPosition = new FileSetPosition(0, _bufferFileName);
+        var count = 0;
+        var payload = payloadReader.ReadPayload(int.MaxValue,
+            null,
+            ref fileSetPosition,
+            ref count,
+            _bufferFileName);
+
+        // Assert
+        // Thus we ensure that file was properly handled by PayloadReader  
+        payload.Count.Should().Be(lines.Length * 2);
+        payload[1].Should().Be(lines[0]);
+    }
+
+    [Theory]
+    [InlineData(RollingInterval.Infinite)]
+    [InlineData(RollingInterval.Year)]
+    [InlineData(RollingInterval.Month)]
+    public void ElasticsearchPayloadReader_CannotUseRollingIntervalLessFrequentThanDay(RollingInterval rollingInterval)
+    {
+        // Arrange
+
+        // Act
+        Action act = () => new ElasticsearchPayloadReader("testPipelineName",
+            "TestTypeName",
+            null,
+            (_, _) => "TestIndex",
+            ElasticOpType.Index,
+            rollingInterval);
+
+        // Assert
+        act.ShouldThrow<ArgumentException>()
+            .WithMessage("Rolling intervals less frequent than RollingInterval.Day are not supported");
+    }
+}