Initial cut at streaming

Author: merbla

sample/Serilog.Sinks.Splunk.Sample/Program.cs

@@ -1,4 +1,5 @@
 using System;
+using System.Linq;
 using System.Net;
 using Splunk.Client;
 
@@ -17,16 +18,21 @@ static void Main(string[] args)
             var eco = new ViaEventCollectorWithExtendedOptions();
 
             eco.Configure();
-
             //ec.Configure();
-       
             //udp.Configure();
             //tcp.Configure();
 
             Log.Information("Simulation running, press any key to exit.");
 
             stub.Run();
 
+            var range = Enumerable.Range(0, 10000);
+
+            foreach (var i in range)
+            {
+                Log.Information("Say hello to {0}", i);
+            }
+
             Console.ReadLine();
         }
     }
@@ -39,7 +45,10 @@ public void Configure()
 
             Log.Logger = new LoggerConfiguration()
                  .WriteTo.LiterateConsole()
-                 .WriteTo.SplunkViaEventCollector("https://mysplunk:8088/services/collector", "685546AE-0278-4786-97C4-5971676D5D70",renderTemplate:false)
+                 .WriteTo.SplunkViaEventCollector("https://mysplunk:8088/services/collector", "685546AE-0278-4786-97C4-5971676D5D70",
+                    renderTemplate:false,
+                    batchSizeLimit:150,
+                    batchIntervalInSeconds:5)
                  .Enrich.WithThreadId()
                  .Enrich.WithProperty("Serilog.Sinks.Splunk.Sample", "ViaEventCollector")
                  .MinimumLevel.Debug()

src/Serilog.Sinks.Splunk/Sinks/Splunk/EventCollectorRequest.cs

@@ -4,36 +4,24 @@
 
 namespace Serilog.Sinks.Splunk
 {
-    internal class EventCollectorRequest : HttpRequestMessage
+    internal class SplunkEvent
     {
-        internal EventCollectorRequest(string splunkHost, string jsonPayLoad) 
-        {
+        private string _payload;
 
-            var stringContent = new StringContent(jsonPayLoad, Encoding.UTF8, "application/json");
-            RequestUri = new Uri(splunkHost);
-            Content = stringContent;
-            Method = HttpMethod.Post;
-        }
-
-        internal EventCollectorRequest(
-            string splunkHost, 
-            string logEvent, 
-            string source,
-            string sourceType, 
-            string host, 
-            string index)
+        internal SplunkEvent(string logEvent, string source, string sourceType, string host, string index)
         {
+            _payload = string.Empty;
 
-            var jsonPayLoad = @"{""event"":" + logEvent 
-              .Replace("\r\n", string.Empty);
+            var jsonPayLoad = @"{""event"":" + logEvent
+            .Replace("\r\n", string.Empty);
 
             if (!string.IsNullOrWhiteSpace(source))
             {
                 jsonPayLoad = jsonPayLoad + @",""source"":""" + source + @"""";
             }
             if (!string.IsNullOrWhiteSpace(sourceType))
             {
-                  jsonPayLoad = jsonPayLoad + @",""sourceType"":""" + sourceType + @"""";
+                jsonPayLoad = jsonPayLoad + @",""sourceType"":""" + sourceType + @"""";
             }
             if (!string.IsNullOrWhiteSpace(host))
             {
@@ -44,7 +32,20 @@ internal EventCollectorRequest(
                 jsonPayLoad = jsonPayLoad + @",""index"":""" + index + @"""";
             }
 
-            jsonPayLoad = jsonPayLoad + "}"; 
+            jsonPayLoad = jsonPayLoad + "}";
+            _payload = jsonPayLoad;
+        }
+
+        public string Payload
+        {
+            get { return _payload; }
+        }
+    }
+
+    internal class EventCollectorRequest : HttpRequestMessage
+    {
+        internal EventCollectorRequest(string splunkHost, string jsonPayLoad)
+        {
 
             var stringContent = new StringContent(jsonPayLoad, Encoding.UTF8, "application/json");
             RequestUri = new Uri(splunkHost);

src/Serilog.Sinks.Splunk/Sinks/Splunk/EventCollectorSink.cs

@@ -1,3 +1,4 @@
+
 // Copyright 2014 Serilog Contributors
 // 
 // Licensed under the Apache License, Version 2.0 (the "License");
@@ -29,7 +30,7 @@ namespace Serilog.Sinks.Splunk
     /// <summary>
     /// A sink to log to the Event Collector available in Splunk 6.3
     /// </summary>
-    public class EventCollectorSink : ILogEventSink
+    public class EventCollectorSink : ILogEventSink, IDisposable
     {
         private readonly string _splunkHost;
         private readonly string _eventCollectorToken;
@@ -40,9 +41,8 @@ public class EventCollectorSink : ILogEventSink
         private readonly int _batchSizeLimitLimit;
         private readonly SplunkJsonFormatter _jsonFormatter;
         private readonly ConcurrentQueue<LogEvent> _queue;
-        private readonly TimeSpan _batchInterval;
-        private readonly EventCollectorClient _httpClient;
-        
+        private readonly EventCollectorClient _httpClient; 
+
         /// <summary>
         /// Taken from Splunk.Logging.Common
         /// </summary>
@@ -76,13 +76,13 @@ public EventCollectorSink(
             _queue = new ConcurrentQueue<LogEvent>();
             _jsonFormatter = new SplunkJsonFormatter(renderMessage: true, formatProvider: formatProvider, renderTemplate: renderTemplate);
             _batchSizeLimitLimit = batchSizeLimit;
-            _batchInterval = TimeSpan.FromSeconds(batchIntervalInSeconds);
+            var batchInterval = TimeSpan.FromSeconds(batchIntervalInSeconds);
 
             _httpClient = new EventCollectorClient(_eventCollectorToken);
 
             //TODO: Implement handling similar to the Seq HTTP sink, including dispose flush
 
-            RepeatAction.OnInterval(_batchInterval, () => ProcessQueue().Wait(), new CancellationToken());
+            RepeatAction.OnInterval(batchInterval, () => ProcessQueue().Wait(), new CancellationToken());
 
         }
 
@@ -153,39 +153,44 @@ private async Task ProcessQueue()
                     if (events.Count == 0)
                         return;
 
-                    //TODO: Add streaming capability for performance.  
-                    // - Stream writer needs to move to outer scope.
-                    // - Change Event Collector Request to only take string (or stream) of events and host
-                    // - New object to stream many events as collection
+                    string allEvents = string.Empty;
 
                     foreach (var logEvent in events)
                     {
                         var sw = new StringWriter();
-
                         _jsonFormatter.Format(logEvent, sw);
-                        var evt = sw.ToString();
 
-                        var request = new EventCollectorRequest(_splunkHost, evt, _source, _sourceType, _host, _index);
-                        var response = await _httpClient.SendAsync(request);
+                        var serialisedEvent = sw.ToString();
+                        
+                        var splunkEvent = new SplunkEvent(serialisedEvent, _source, _sourceType, _host, _index);
+
+                        allEvents = $"{allEvents}{splunkEvent.Payload}";
+
+                    }
+                    var request = new EventCollectorRequest(_splunkHost, allEvents);
+                    
+                    var response = await _httpClient.SendAsync(request);
 
-                        if (response.IsSuccessStatusCode) {  //Do Nothing?
+                    if (response.IsSuccessStatusCode)
+                    {  //Do Nothing?
+                    }
+                    else
+                    {
+                        //Application Errors sent via HTTP Event Collector
+                        if (HttpEventCollectorApplicationErrors.Any(x => x == response.StatusCode))
+                        {
+                            SelfLog.WriteLine("A status code of {0} was received when attempting to send to {1}.  The event has been discarded and will not be placed back in the queue.", response.StatusCode.ToString(), _splunkHost);
                         }
                         else
                         {
-                            //Application Errors sent via HTTP Event Collector
-                            if (HttpEventCollectorApplicationErrors.Any(x => x == response.StatusCode))
-                            {
-                                SelfLog.WriteLine("A status code of {0} was received when attempting to send to {1}.  The event has been discarded and will not be placed back in the queue.", response.StatusCode.ToString(), _splunkHost);
-                            }
-                            else
-                            {
-                                //Put the item back in the queue & retry on next go
-                                SelfLog.WriteLine("A status code of {0} was received when attempting to send to {1}.  The event has been placed back in the queue", response.StatusCode.ToString(), _splunkHost);
+                            //Put the item back in the queue & retry on next go
+                            SelfLog.WriteLine("A status code of {0} was received when attempting to send to {1}.  The event has been placed back in the queue", response.StatusCode.ToString(), _splunkHost);
 
+                            foreach (var logEvent in events)
+                            {
                                 _queue.Enqueue(logEvent);
                             }
                         }
-                        
                     }
                 } while (true);
             }