Merge pull request #510 from BrettJaner/feat/insert-statement-batch-writer

Add SqlInsertStatementWriter which uses INSERT statements instead of SqlBulkCopy

Author: ckadluba

README.md

@@ -246,6 +246,7 @@ Basic settings of the sink are configured using the properties in a `MSSqlServer
 * `BatchPeriod`
 * `EagerlyEmitFirstEvent`
 * `LevelSwitch`
+* `UseSqlBulkCopy`
 
 ### TableName
 
@@ -291,6 +292,11 @@ This setting is not used by the audit sink as it writes each event immediately a
 
 A switch allowing the pass-through minimum level to be changed at runtime. If this is set, the parameter `restrictedToMinimumLevel` in the [sink configuration method](#sink-configuration) is ignored.
 
+### UseSqlBulkCopy
+
+A flag to use `SqlBulkCopy` instead of individual INSERT statements when writing log events.  The default is `true`.
+This setting is not used by the audit sink as it always uses INSERT statements to write events.
+
 ## ColumnOptions Object
 
 Features of the log table are defined by changing properties on a `ColumnOptions` object:
@@ -319,7 +325,7 @@ Setting this to `true` changes the table to the clustered columnstore index (CCI
 
 ### DisableTriggers
 
-Disabling triggers can significantly improve batch-write performance.
+Disabling triggers can significantly improve batch-write performance. Only applies when `SqlBulkCopy` is used.
 
 ### AdditionalColumns
 

src/Serilog.Sinks.MSSqlServer/Configuration/Implementations/Microsoft.Extensions.Configuration/MicrosoftExtensionsSinkOptionsProvider.cs

@@ -33,6 +33,7 @@ private static void ReadBatchSettings(IConfigurationSection config, MSSqlServerS
             SetProperty.IfNotNull<int>(config["batchPostingLimit"], val => sinkOptions.BatchPostingLimit = val);
             SetProperty.IfNotNull<string>(config["batchPeriod"], val => sinkOptions.BatchPeriod = TimeSpan.Parse(val, CultureInfo.InvariantCulture));
             SetProperty.IfNotNull<bool>(config["eagerlyEmitFirstEvent"], val => sinkOptions.EagerlyEmitFirstEvent = val);
+            SetProperty.IfNotNull<bool>(config["useSqlBulkCopy"], val => sinkOptions.UseSqlBulkCopy = val);
         }
     }
 }

src/Serilog.Sinks.MSSqlServer/Configuration/Implementations/System.Configuration/MSSqlServerConfigurationSection.cs

@@ -198,6 +198,12 @@ public ValueConfigElement EagerlyEmitFirstEvent
         {
             get => (ValueConfigElement)base[nameof(EagerlyEmitFirstEvent)];
         }
+
+        [ConfigurationProperty(nameof(UseSqlBulkCopy))]
+        public ValueConfigElement UseSqlBulkCopy
+        {
+            get => (ValueConfigElement)base[nameof(UseSqlBulkCopy)];
+        }
     }
 }
 

src/Serilog.Sinks.MSSqlServer/Configuration/Implementations/System.Configuration/SystemConfigurationSinkOptionsProvider.cs

@@ -32,6 +32,8 @@ private static void ReadBatchSettings(MSSqlServerConfigurationSection config, MS
             SetProperty.IfProvided<string>(config.BatchPeriod, nameof(config.BatchPeriod.Value), value => sinkOptions.BatchPeriod = TimeSpan.Parse(value, CultureInfo.InvariantCulture));
             SetProperty.IfProvided<bool>(config.EagerlyEmitFirstEvent, nameof(config.EagerlyEmitFirstEvent.Value),
                 value => sinkOptions.EagerlyEmitFirstEvent = value);
+            SetProperty.IfProvided<bool>(config.UseSqlBulkCopy, nameof(config.UseSqlBulkCopy.Value),
+                value => sinkOptions.UseSqlBulkCopy = value);
         }
     }
 }

src/Serilog.Sinks.MSSqlServer/Sinks/MSSqlServer/ColumnOptions/ColumnOptions.cs

@@ -90,6 +90,7 @@ public ICollection<StandardColumn> Store
 
         /// <summary>
         /// Indicates if triggers should be disabled when inserting log entries.
+        /// Only applies when SqlBulkCopy is used.
         /// </summary>
         public bool DisableTriggers { get; set; }
 

src/Serilog.Sinks.MSSqlServer/Sinks/MSSqlServer/Dependencies/SinkDependenciesFactory.cs

@@ -52,10 +52,14 @@ internal static SinkDependencies Create(
                     sqlCreateDatabaseWriter, sqlConnectionFactoryNoDb),
                 SqlTableCreator = new SqlTableCreator(
                     sqlCreateTableWriter, sqlConnectionFactory),
-                SqlBulkBatchWriter = new SqlBulkBatchWriter(
-                    sinkOptions.TableName, sinkOptions.SchemaName, columnOptions.DisableTriggers,
-                    sqlConnectionFactory, logEventDataGenerator),
-                SqlLogEventWriter = new SqlLogEventWriter(
+                SqlBulkBatchWriter = sinkOptions.UseSqlBulkCopy
+                    ? (ISqlBulkBatchWriter)new SqlBulkBatchWriter(
+                        sinkOptions.TableName, sinkOptions.SchemaName, columnOptions.DisableTriggers,
+                        sqlConnectionFactory, logEventDataGenerator)
+                    : (ISqlBulkBatchWriter)new SqlInsertStatementWriter(
+                        sinkOptions.TableName, sinkOptions.SchemaName,
+                        sqlConnectionFactory, logEventDataGenerator),
+                SqlLogEventWriter = new SqlInsertStatementWriter(
                     sinkOptions.TableName, sinkOptions.SchemaName,
                     sqlConnectionFactory, logEventDataGenerator)
             };

src/Serilog.Sinks.MSSqlServer/Sinks/MSSqlServer/MSSqlServerSinkOptions.cs

@@ -17,6 +17,7 @@ public MSSqlServerSinkOptions()
             BatchPostingLimit = MSSqlServerSink.DefaultBatchPostingLimit;
             BatchPeriod = MSSqlServerSink.DefaultPeriod;
             EagerlyEmitFirstEvent = true;
+            UseSqlBulkCopy = true;
         }
 
         internal MSSqlServerSinkOptions(
@@ -77,5 +78,10 @@ internal MSSqlServerSinkOptions(
         /// A switch allowing the pass-through minimum level to be changed at runtime
         /// </summary>
         public LoggingLevelSwitch LevelSwitch { get; set; }
+
+        /// <summary>
+        /// Flag to use <see cref="Microsoft.Data.SqlClient.SqlBulkCopy"/> instead of individual INSERT statements (default: true)
+        /// </summary>
+        public bool UseSqlBulkCopy { get; set; }
     }
 }

src/Serilog.Sinks.MSSqlServer/Sinks/MSSqlServer/Platform/SqlClient/ISqlCommandWrapper.cs

@@ -1,5 +1,6 @@
 using System;
 using System.Data;
+using System.Threading.Tasks;
 
 namespace Serilog.Sinks.MSSqlServer.Platform.SqlClient
 {
@@ -10,5 +11,6 @@ internal interface ISqlCommandWrapper : IDisposable
 
         void AddParameter(string parameterName, object value);
         int ExecuteNonQuery();
+        Task<int> ExecuteNonQueryAsync();
     }
 }

src/Serilog.Sinks.MSSqlServer/Sinks/MSSqlServer/Platform/SqlClient/SqlCommandWrapper.cs

@@ -1,5 +1,6 @@
 using System;
 using System.Data;
+using System.Threading.Tasks;
 using Microsoft.Data.SqlClient;
 
 namespace Serilog.Sinks.MSSqlServer.Platform.SqlClient
@@ -43,6 +44,9 @@ public void AddParameter(string parameterName, object value)
         public int ExecuteNonQuery() =>
             _sqlCommand.ExecuteNonQuery();
 
+        public Task<int> ExecuteNonQueryAsync() =>
+            _sqlCommand.ExecuteNonQueryAsync();
+
         protected virtual void Dispose(bool disposing)
         {
             if (!_disposedValue)

src/Serilog.Sinks.MSSqlServer/Sinks/MSSqlServer/Platform/SqlInsertStatementWriter.cs

@@ -0,0 +1,88 @@
+using System;
+using System.Collections.Generic;
+using System.Data;
+using System.Text;
+using System.Threading.Tasks;
+using Serilog.Debugging;
+using Serilog.Events;
+using Serilog.Sinks.MSSqlServer.Output;
+using static System.FormattableString;
+
+namespace Serilog.Sinks.MSSqlServer.Platform
+{
+    internal class SqlInsertStatementWriter : ISqlBulkBatchWriter, ISqlLogEventWriter
+    {
+        private readonly string _tableName;
+        private readonly string _schemaName;
+        private readonly ISqlConnectionFactory _sqlConnectionFactory;
+        private readonly ILogEventDataGenerator _logEventDataGenerator;
+
+        public SqlInsertStatementWriter(
+            string tableName,
+            string schemaName,
+            ISqlConnectionFactory sqlConnectionFactory,
+            ILogEventDataGenerator logEventDataGenerator)
+        {
+            _tableName = tableName ?? throw new ArgumentNullException(nameof(tableName));
+            _schemaName = schemaName ?? throw new ArgumentNullException(nameof(schemaName));
+            _sqlConnectionFactory = sqlConnectionFactory ?? throw new ArgumentNullException(nameof(sqlConnectionFactory));
+            _logEventDataGenerator = logEventDataGenerator ?? throw new ArgumentNullException(nameof(logEventDataGenerator));
+        }
+
+        public Task WriteBatch(IEnumerable<LogEvent> events, DataTable dataTable) => WriteBatch(events);
+
+        public void WriteEvent(LogEvent logEvent) => WriteBatch(new[] { logEvent }).GetAwaiter().GetResult();
+
+        public async Task WriteBatch(IEnumerable<LogEvent> events)
+        {
+            try
+            {
+                using (var cn = _sqlConnectionFactory.Create())
+                {
+                    await cn.OpenAsync().ConfigureAwait(false);
+
+                    foreach (var logEvent in events)
+                    {
+                        using (var command = cn.CreateCommand())
+                        {
+                            command.CommandType = CommandType.Text;
+
+                            var fieldList = new StringBuilder(Invariant($"INSERT INTO [{_schemaName}].[{_tableName}] ("));
+                            var parameterList = new StringBuilder(") VALUES (");
+
+                            var index = 0;
+                            foreach (var field in _logEventDataGenerator.GetColumnsAndValues(logEvent))
+                            {
+                                if (index != 0)
+                                {
+                                    fieldList.Append(',');
+                                    parameterList.Append(',');
+                                }
+
+                                fieldList.Append(Invariant($"[{field.Key}]"));
+                                parameterList.Append("@P");
+                                parameterList.Append(index);
+
+                                command.AddParameter(Invariant($"@P{index}"), field.Value);
+
+                                index++;
+                            }
+
+                            parameterList.Append(')');
+                            fieldList.Append(parameterList);
+
+                            command.CommandText = fieldList.ToString();
+
+                            await command.ExecuteNonQueryAsync().ConfigureAwait(false);
+                        }
+                    }
+                }
+            }
+            catch (Exception ex)
+            {
+                SelfLog.WriteLine("Unable to write log event to the database due to following error: {0}", ex);
+                throw;
+            }
+        }
+    }
+}