Added StatsD outputter and smart shutdown code.

.nuget/packages.config

@@ -2,6 +2,6 @@
 <packages>
   <package id="Newtonsoft.Json" version="6.0.4" targetFramework="net40" />
   <package id="NUnit.Runners" version="2.6.4" />
-  <package id="RapidRegex.Core" version="1.0.0.2" targetFramework="net40" />
+  <package id="RapidRegex.Core" version="1.0.0.4" targetFramework="net40" />
   <package id="System.Linq.Dynamic" version="1.0.4" targetFramework="net40" />
 </packages>

README.md

@@ -16,7 +16,7 @@ TimberWinR uses a configuration file to control how the logs are collected, filt
 These are broken down into:
  1. Inputs  (Collect data from different sources)
  2. Filters (Are applied to all Inputs)
- 3. Outputs (Redis, Elasticsearch or Stdout)
+ 3. Outputs (e.g. Redis, Elasticsearch, Stdout, StatsD)
 
 ### Support ###
 Please use the TimberWinR Google Group for discussion and support:
@@ -108,6 +108,7 @@ represented as a JSON Property or Array.
 2. [Elasticsearch](https://github.com/Cimpress-MCP/TimberWinR/blob/master/TimberWinR/mdocs/ElasticsearchOutput.md)
 3. [Stdout](https://github.com/Cimpress-MCP/TimberWinR/blob/master/TimberWinR/mdocs/StdoutOutput.md)
 4. [File](https://github.com/Cimpress-MCP/TimberWinR/blob/master/TimberWinR/mdocs/FileOutput.md)
+5. [StatsD](https://github.com/Cimpress-MCP/TimberWinR/blob/master/TimberWinR/mdocs/StatsD.md)
 
 ## Sample Configuration
 TimberWinR reads a JSON configuration file, an example file is shown here:

TimberWinR.ServiceHost/Program.cs

@@ -25,11 +25,11 @@ namespace TimberWinR.ServiceHost
     {
         const string KeyPath = @"HKEY_LOCAL_MACHINE\SYSTEM\CurrentControlSet\services\TimberWinR";
         const string KeyName = "ImagePath";
-        
+
         private static void Main(string[] args)
         {
             Arguments arguments = new Arguments();
-           
+
             HostFactory.Run(hostConfigurator =>
             {
                 string cmdLine = Environment.CommandLine;
@@ -45,7 +45,7 @@ namespace TimberWinR.ServiceHost
                 hostConfigurator.AddCommandLineDefinition("configFile", c => arguments.ConfigFile = c);
                 hostConfigurator.AddCommandLineDefinition("logLevel", c => arguments.LogLevel = c);
                 hostConfigurator.AddCommandLineDefinition("logDir", c => arguments.LogfileDir = c);
-                hostConfigurator.AddCommandLineDefinition("diagnosticPort", c => arguments.DiagnosticPort = int.Parse(c));    
+                hostConfigurator.AddCommandLineDefinition("diagnosticPort", c => arguments.DiagnosticPort = int.Parse(c));
 
                 hostConfigurator.ApplyCommandLine();
                 hostConfigurator.RunAsLocalSystem();
@@ -56,7 +56,7 @@ namespace TimberWinR.ServiceHost
                 hostConfigurator.SetServiceName("TimberWinR");
 
                 hostConfigurator.AfterInstall(() =>
-                { 
+                {
                     var currentValue = Registry.GetValue(KeyPath, KeyName, "").ToString();
                     if (!string.IsNullOrEmpty(currentValue))
                     {
@@ -72,18 +72,18 @@ namespace TimberWinR.ServiceHost
         }
 
         private static void AddServiceParameter(string paramName, string value)
-        {         
+        {
             string currentValue = Registry.GetValue(KeyPath, KeyName, "").ToString();
 
-            if (!string.IsNullOrEmpty(paramName) && !currentValue.Contains(string.Format("{0} ", paramName)))           
+            if (!string.IsNullOrEmpty(paramName) && !currentValue.Contains(string.Format("{0} ", paramName)))
             {
-                currentValue += string.Format(" {0} \"{1}\"", paramName, value.Replace("\\\\", "\\"));               
-                Registry.SetValue(KeyPath, KeyName, currentValue);               
+                currentValue += string.Format(" {0} \"{1}\"", paramName, value.Replace("\\\\", "\\"));
+                Registry.SetValue(KeyPath, KeyName, currentValue);
             }
         }
 
         private static void AddServiceParameter(string paramName, int value)
-        {    
+        {
             string currentValue = Registry.GetValue(KeyPath, KeyName, "").ToString();
 
             if (!string.IsNullOrEmpty(paramName) && !currentValue.Contains(string.Format("{0}:", paramName)))
@@ -131,13 +131,15 @@ namespace TimberWinR.ServiceHost
         private readonly Arguments _args;
         private TimberWinR.Diagnostics.Diagnostics _diags;
         private TimberWinR.Manager _manager;
+        public bool StartingUp { get; set; }
+        public bool Started { get; set; }
 
         public TimberWinRService(Arguments args)
         {
             _args = args;
             _cancellationTokenSource = new CancellationTokenSource();
             _cancellationToken = _cancellationTokenSource.Token;
-            _serviceTask = new Task(RunService, _cancellationToken);           
+            _serviceTask = new Task(RunService, _cancellationToken);
         }
 
         public void Start()
@@ -147,22 +149,41 @@ namespace TimberWinR.ServiceHost
 
         public void Stop()
         {
+            WaitForStartupToComplete();
+
             _cancellationTokenSource.Cancel();
             if (_diags != null)
-             _diags.Shutdown();
+                _diags.Shutdown();
 
             if (_manager != null)
                 _manager.Shutdown();
         }
 
+        // If you bounce the service too quickly, the shutdown can occur
+        // before the service has started, which results in a hang, this blocks until 
+        // all thread have properly started (waiting up to 10 seconds max) 
+        private void WaitForStartupToComplete()
+        {
+            int tries = 100; // 10 seconds max
+            if (StartingUp)
+            {
+                while (!Started && tries-- >= 0)
+                {                  
+                    Thread.Sleep(100);
+                }
+            }
+        }
+
         /// <summary>
         /// The Main body of the Service Worker Thread
         /// </summary>
         private void RunService()
         {
+            StartingUp = true;
             _manager = new TimberWinR.Manager(_args.ConfigFile, _args.LogLevel, _args.LogfileDir, _args.LiveMonitor, _cancellationToken);
             if (_args.DiagnosticPort > 0)
                 _diags = new Diagnostics.Diagnostics(_manager, _cancellationToken, _args.DiagnosticPort);
+            Started = true;
         }
     }
 }

TimberWinR.ServiceHost/TimberWinR.ServiceHost.csproj

@@ -38,9 +38,9 @@
     <ApplicationIcon>timberwinr.ico</ApplicationIcon>
   </PropertyGroup>
   <ItemGroup>
-    <Reference Include="RapidRegex.Core, Version=1.0.0.0, Culture=neutral, processorArchitecture=MSIL">
+    <Reference Include="RapidRegex.Core, Version=1.0.0.4, Culture=neutral, processorArchitecture=MSIL">
       <SpecificVersion>False</SpecificVersion>
-      <HintPath>..\packages\RapidRegex.Core.1.0.0.2\lib\net40\RapidRegex.Core.dll</HintPath>
+      <HintPath>..\packages\RapidRegex.Core.1.0.0.4\lib\net40\RapidRegex.Core.dll</HintPath>
     </Reference>
     <Reference Include="System" />
     <Reference Include="System.Configuration.Install" />

TimberWinR.ServiceHost/packages.config

@@ -1,5 +1,5 @@
 <?xml version="1.0" encoding="utf-8"?>
 <packages>
-  <package id="RapidRegex.Core" version="1.0.0.2" targetFramework="net40" />
+  <package id="RapidRegex.Core" version="1.0.0.4" targetFramework="net40" />
   <package id="Topshelf" version="3.1.4" targetFramework="net40" />
 </packages>

TimberWinR.TestGenerator/CommandLineOptions.cs

@@ -27,6 +27,9 @@ namespace TimberWinR.TestGenerator
         [Option("resultsFile", HelpText = "Expected results Results json file")]
         public string ExpectedResultsFile { get; set; }
 
+        [Option("totalMessages", DefaultValue = 0, HelpText = "The total number of messages to send to the output(s)")]
+        public int TotalMessages { get; set; }
+
         [Option('n', "numMessages", DefaultValue = 1000, HelpText = "The number of messages to send to the output(s)")]
         public int NumMessages { get; set; }
 

TimberWinR.TestGenerator/Program.cs

@@ -350,7 +350,7 @@ namespace TimberWinR.TestGenerator
                     var mbc = outputToken["queuedMessageCount"].Value<int>();
                     var smc = outputToken["sentMessageCount"].Value<int>();
 
-                    // LogManager.GetCurrentClassLogger().Info("Queued: {0}, Sent: {1}", mbc, smc);
+                    //LogManager.GetCurrentClassLogger().Info("Output: {2} Queued: {0}, Sent: {1}", mbc, smc, outputToken.ToString());
 
                     completed = mbc == 0 && smc >= _totalMessagesToSend;
                 }
@@ -504,6 +504,8 @@ namespace TimberWinR.TestGenerator
 
         static Task[] RunGenerators(CommandLineOptions options)
         {
+            _totalMessagesToSend = options.TotalMessages;
+
             _monitorTask = Task.Factory.StartNew(() =>
                 {
                     using (var syncHandle = new ManualResetEventSlim())

TimberWinR.TestGenerator/TimberWinR.TestGenerator.csproj

@@ -126,6 +126,18 @@
     <Content Include="results5.json">
       <CopyToOutputDirectory>PreserveNewest</CopyToOutputDirectory>
     </Content>
+    <Content Include="test7-tw.json">
+      <CopyToOutputDirectory>PreserveNewest</CopyToOutputDirectory>
+    </Content>
+    <Content Include="test7.json">
+      <CopyToOutputDirectory>PreserveNewest</CopyToOutputDirectory>
+    </Content>
+    <Content Include="results7.json">
+      <CopyToOutputDirectory>PreserveNewest</CopyToOutputDirectory>
+    </Content>
+    <None Include="sample-apache.log">
+      <CopyToOutputDirectory>PreserveNewest</CopyToOutputDirectory>
+    </None>
   </ItemGroup>
   <ItemGroup>
     <ProjectReference Include="..\TimberWinR\TimberWinR.csproj">

TimberWinR.TestGenerator/results7.json

@@ -0,0 +1,13 @@
+{
+    "Results": {
+        "Inputs": [           
+            {
+                "udp": {
+                    "test1: message sent count": "[messages] == 80000",
+                    "test2: average cpu": "[avgCpuUsage] <= 30",
+                    "test3: maximum memory": "[maxMemUsage] <= 30"              
+                }
+            }
+        ]
+    }
+}

TimberWinR.TestGenerator/test2-tw.json

@@ -36,7 +36,7 @@
                     "batch_count": 500,
                     "threads":  2,
                     "host": [
-                        "tstlexiceapp006.mycompany.svc"
+                        "tstlexiceapp006.vistaprint.svc"
                     ]
                 }
             ]

TimberWinR.TestGenerator/test2.json

@@ -1,11 +1,12 @@
 {
     "test": "Test 2",
     "arguments": {
+        "--start":  "",
         "--testFile": "test2.json",
         "--testDir": "test2",
         "--timberWinRConfig": "test2-tw.json",
         "--numMessages": 1234,
-        "--logLevel": "trace",
+        "--logLevel": "debug",
         "--udp": "5140",
         "--jroll": ["r1.jlog", "r2.jlog"],
         "--json": ["1.jlog", "2.jlog", "3.jlog", "4.jlog"],

TimberWinR.TestGenerator/test7-tw.json

@@ -0,0 +1,36 @@
+{
+    "TimberWinR": {
+        "Inputs": {
+            "TailFiles": [
+                {
+                    "interval": 5,
+                    "logSource": "apache log files",
+                    "location": "..\\sample-apache.log",
+                    "recurse": -1
+                }
+            ]
+        },
+        "Filters": [
+            {
+                "grok": {
+                    "type": "Win32-TailLog",
+                    "match": [
+                        "Text",
+                        "%{COMBINEDAPACHELOG}"
+                    ]
+                }
+            }
+        ],
+        "Outputs": {            
+            "StatsD": [
+                {
+                    "type": "Win32-TailLog",
+                    "port": 8125,
+                    "host": "devlexicesnu003.vistaprint.svc",
+                    "increment":  ["apache.response.%{response}"],
+                    "count":  ["apache.bytes", "%{bytes}"]
+                }
+            ]
+        }
+    }
+}

TimberWinR.TestGenerator/test7.json

@@ -0,0 +1,13 @@
+{
+    "test": "Test 7",
+    "arguments": {
+        "--totalMessages":  2223,
+        "--start":  "",
+        "--testFile": "test7.json",
+        "--testDir": "test7",
+        "--timberWinRConfig": "test7-tw.json",
+        "--numMessages": 1234,
+        "--logLevel": "debug",       
+        "--resultsFile": "results7.json"
+    }
+}

TimberWinR/Configuration.cs

@@ -39,6 +39,12 @@ namespace TimberWinR
             get { return _events; }
         }
 
+        private List<StatsDOutputParameters> _statsdOutputs = new List<StatsDOutputParameters>();
+        public IEnumerable<StatsDOutputParameters> StatsDOutputs
+        {
+            get { return _statsdOutputs; }
+        }
+
         private List<RedisOutputParameters> _redisOutputs = new List<RedisOutputParameters>();
         public IEnumerable<RedisOutputParameters> RedisOutputs
         {
@@ -265,6 +271,8 @@ namespace TimberWinR
 
             if (x.TimberWinR.Outputs != null)
             {
+                if (x.TimberWinR.Outputs.StatsD != null)
+                    c._statsdOutputs.AddRange(x.TimberWinR.Outputs.StatsD.ToList());
                 if (x.TimberWinR.Outputs.Redis != null)
                     c._redisOutputs.AddRange(x.TimberWinR.Outputs.Redis.ToList());
                 if (x.TimberWinR.Outputs.Elasticsearch != null)
@@ -307,6 +315,7 @@ namespace TimberWinR
             _events = new List<WindowsEvent>();
             _iisw3clogs = new List<IISW3CLogParameters>();
             _logs = new List<LogParameters>();
+            _statsdOutputs = new List<StatsDOutputParameters>();
             _redisOutputs = new List<RedisOutputParameters>();
             _elasticsearchOutputs = new List<ElasticsearchOutputParameters>();
             _stdoutOutputs = new List<StdoutOutputParameters>();

TimberWinR/Manager.cs

@@ -24,7 +24,7 @@ namespace TimberWinR
         public Configuration Config { get; set; }
         public List<OutputSender> Outputs { get; set; }       
         public List<InputListener> Listeners { get; set; }
-        public bool LiveMonitor { get; set; }
+        public bool LiveMonitor { get; set; }      
 
         public event Action<Configuration> OnConfigurationProcessed;
 
@@ -47,7 +47,7 @@ namespace TimberWinR
 
 
         public void Shutdown()
-        {
+        {          
             LogManager.GetCurrentClassLogger().Info("Shutting Down");
 
             foreach (InputListener listener in Listeners)
@@ -68,7 +68,7 @@ namespace TimberWinR
         }
 
         public Manager(string jsonConfigFile, string logLevel, string logfileDir, bool liveMonitor, CancellationToken cancelToken, bool processConfiguration = true)
-        {
+        {          
             LogsFileDatabase.Manager = this;
 
             StartedOn = DateTime.UtcNow;
@@ -149,7 +149,7 @@ namespace TimberWinR
             if (processConfiguration)
             {
                 ProcessConfiguration(cancelToken, Config);
-            }
+            }         
         }
 
         public void Start(CancellationToken cancelToken)
@@ -161,10 +161,19 @@ namespace TimberWinR
         {
             // Read the Configuration file
             if (config != null)
-            {
+            {                
                 if (OnConfigurationProcessed != null)
                     OnConfigurationProcessed(config);
 
+                if (config.StatsDOutputs != null)
+                {
+                    foreach (var ro in config.StatsDOutputs)
+                    {
+                        var output = new StatsDOutput(this, ro, cancelToken);
+                        Outputs.Add(output);
+                    }
+                }
+
                 if (config.RedisOutputs != null)
                 {
                     foreach (var ro in config.RedisOutputs)
@@ -294,7 +303,7 @@ namespace TimberWinR
                     json.Add(new JProperty("type", "Win32-TimberWinR"));
                     json.Add(new JProperty("host", computerName));
                     output.Startup(json);
-                }
+                }               
             }
         }
 

TimberWinR/NStatsD/Client.cs

@@ -0,0 +1,178 @@
+using System;
+using System.Collections.Generic;
+using System.Configuration;
+using System.Linq;
+using System.Net.Sockets;
+using System.Threading;
+
+namespace NStatsD
+{
+    public sealed class Client
+    {
+        private static UdpClient _client;
+
+        public static string Host { get; set; }
+        public static int Port { get; set; }
+
+        Client()
+        {
+            if (Config != null)
+            {
+                var host = Config.Server.Host;
+                var port = Config.Server.Port;
+                _client = new UdpClient(host, port);
+            }
+            else
+            {
+                Config = new StatsDConfigurationSection();
+                Config.Server.Host = Host;;
+                Config.Server.Port = Port;
+                _client = new UdpClient(Host, Port);
+            }
+        }
+
+        public static Client Current
+        {
+            get { return CurrentClient.Instance; }
+        }
+
+        class CurrentClient
+        {
+            static CurrentClient() { }
+
+            internal static readonly Client Instance = new Client();
+        }
+
+        private StatsDConfigurationSection _config;
+        public StatsDConfigurationSection Config
+        {
+            get
+            {
+                if (_config == null)
+                {
+                    _config = (StatsDConfigurationSection)ConfigurationManager.GetSection("statsD");
+                }
+
+                if(_config != null)
+                    _config.Prefix = ValidatePrefix(_config.Prefix);
+
+                return _config;
+            }
+            set
+            {
+                _config = value;
+                _config.Prefix = ValidatePrefix(_config.Prefix);
+            }
+        }
+
+        private string ValidatePrefix(string prefix)
+        {
+            if (string.IsNullOrWhiteSpace(prefix))
+                return prefix;
+
+            if (prefix.EndsWith("."))
+                return prefix;
+
+            return string.Format("{0}.", prefix);
+        }
+
+        /// <summary>
+        /// Sends timing statistics.
+        /// </summary>
+        /// <param name="stat">Name of statistic being updated.</param>
+        /// <param name="time">The timing it took to complete.</param>
+        /// <param name="sampleRate">Tells StatsD how often to sample this value. Defaults to 1 (send all values).</param>
+        /// <param name="callback">A callback for when the send is complete. Defaults to null.</param>
+        public void Timing(string stat, long time, double sampleRate = 1, AsyncCallback callback = null)
+        {
+            var data = new Dictionary<string, string> { { stat, string.Format("{0}|ms", time) } };
+
+            Send(data, sampleRate, callback);
+        }
+
+        /// <summary>
+        /// Increments a counter
+        /// </summary>
+        /// <param name="stat">Name of statistic being updated.</param>
+        /// <param name="sampleRate">Tells StatsD how often to sample this value. Defaults to 1 (send all values).</param>
+        /// <param name="callback">A callback for when the send is complete. Defaults to null.</param>
+        public void Increment(string stat, double sampleRate = 1, AsyncCallback callback = null)
+        {
+            UpdateStats(stat, 1, sampleRate, callback);
+        }
+
+        /// <summary>
+        /// Decrements a counter
+        /// </summary>
+        /// <param name="stat">Name of statistic being updated.</param>
+        /// <param name="sampleRate">Tells StatsD how often to sample this value. Defaults to 1 (send all values).</param>
+        /// <param name="callback">A callback for when the send is complete. Defaults to null.</param>
+        public void Decrement(string stat, double sampleRate = 1, AsyncCallback callback = null)
+        {
+            UpdateStats(stat, -1, sampleRate, callback);
+        }
+
+        /// <summary>
+        /// Updates a counter by an arbitrary amount
+        /// </summary>
+        /// <param name="stat">Name of statistic being updated.</param>
+        /// <param name="value">The value of the metric.</param>
+        /// <param name="sampleRate">Tells StatsD how often to sample this value. Defaults to 1 (send all values).</param>
+        /// <param name="callback">A callback for when the send is complete. Defaults to null.</param>
+        public void Gauge(string stat, int value, double sampleRate = 1, AsyncCallback callback = null)
+        {
+            var data = new Dictionary<string, string> { { stat, string.Format("{0}|g", value) } };
+            Send(data, sampleRate, callback);
+        }
+
+        /// <summary>
+        /// Updates a counter by an arbitrary amount
+        /// </summary>
+        /// <param name="stat">Name of statistic(s) being updated.</param>
+        /// <param name="delta">The amount to adjust the counter</param>
+        /// <param name="sampleRate">Tells StatsD how often to sample this value. Defaults to 1 (send all values).</param>
+        /// <param name="callback">A callback for when the send is complete. Defaults to null.</param>
+        public void UpdateStats(string stat, int delta = 1, double sampleRate = 1, AsyncCallback callback = null)
+        {
+            var dictionary = new Dictionary<string, string> { { stat, string.Format("{0}|c", delta) } };
+            Send(dictionary, sampleRate, callback);
+        }
+
+        private static int _seed = Environment.TickCount;
+        private static readonly ThreadLocal<Random> random = new ThreadLocal<Random>(() => new Random(Interlocked.Increment(ref _seed)));
+
+        private void Send(Dictionary<string, string> data, double sampleRate, AsyncCallback callback)
+        {
+            if (!Config.Enabled)
+                return;
+
+            if (sampleRate < 1)
+            {
+                var nextRand = random.Value.NextDouble();
+                if (nextRand <= sampleRate)
+                {
+                    var sampledData = data.Keys.ToDictionary(stat => stat,
+                        stat => string.Format("{0}|@{1}", data[stat], sampleRate));
+                    SendToStatsD(sampledData, callback);
+                }
+            }
+            else
+            {
+                SendToStatsD(data, callback);
+            }
+        }
+
+        private void SendToStatsD(Dictionary<string, string> sampledData, AsyncCallback callback)
+        {
+            var prefix = Config.Prefix;
+            var encoding = new System.Text.ASCIIEncoding();
+
+            foreach (var stat in sampledData.Keys)
+            {
+                var stringToSend = string.Format("{0}{1}:{2}", prefix, stat, sampledData[stat]);
+                var sendData = encoding.GetBytes(stringToSend);
+                _client.BeginSend(sendData, sendData.Length, callback, null);
+            }
+        }
+    }
+}

TimberWinR/NStatsD/StatsDConfigurationSection.cs

@@ -0,0 +1,51 @@
+using System.Configuration;
+
+namespace NStatsD
+{
+    public class StatsDConfigurationSection : ConfigurationSection
+    {
+        [ConfigurationProperty("enabled", DefaultValue = "true", IsRequired = false)]
+        public bool Enabled
+        {
+            get { return (bool)this["enabled"]; }
+            set { this["enabled"] = value; }
+        }
+
+        [ConfigurationProperty("server")]
+        public ServerElement Server
+        {
+            get { return (ServerElement)this["server"]; }
+            set { this["server"] = value; }
+        }
+
+        [ConfigurationProperty("prefix", DefaultValue = "", IsRequired = false)]
+        public string Prefix
+        {
+            get { return (string)this["prefix"]; }
+            set { this["prefix"] = value; }
+        }
+
+        public override bool IsReadOnly()
+        {
+            return false;
+        }
+    }
+
+
+    public class ServerElement : ConfigurationElement
+    {
+        [ConfigurationProperty("host", DefaultValue = "localhost", IsRequired = true)]
+        public string Host
+        {
+            get { return (string)this["host"]; }
+            set { this["host"] = value; }
+        }
+
+        [ConfigurationProperty("port", DefaultValue = "8125", IsRequired = false)]
+        public int Port
+        {
+            get { return (int)this["port"]; }
+            set { this["port"] = value; }
+        }
+    }
+}

TimberWinR/Outputs/StatsD.cs

@@ -0,0 +1,362 @@
+using System;
+using System.Collections.Generic;
+using System.Diagnostics.Eventing.Reader;
+using System.Linq;
+using System.Linq.Expressions;
+using System.Net.Sockets;
+using System.Text;
+using System.Threading;
+using CSRedis;
+using Nest;
+using Newtonsoft.Json;
+using Newtonsoft.Json.Linq;
+using NLog;
+using System.Threading.Tasks;
+using RapidRegex.Core;
+using System.Text.RegularExpressions;
+using System.Globalization;
+using TimberWinR.Parser;
+
+namespace TimberWinR.Outputs
+{   
+    public class StatsDOutput : OutputSender
+    {
+        public int QueueDepth
+        {
+            get { return _jsonQueue.Count; }
+        }
+
+        public long SentMessages
+        {
+            get { return _sentMessages; }
+        }
+
+       
+        private readonly int _port;
+        public string _host { get; set; }
+
+        private readonly int _interval;
+        private readonly object _locker = new object();
+        private readonly List<JObject> _jsonQueue;
+        private TimberWinR.Manager _manager;        
+        private long _sentMessages;
+        private long _errorCount;
+        private readonly int _maxQueueSize;
+        private readonly bool _queueOverflowDiscardOldest;
+        private readonly int _flushSize;
+        private readonly int _idleFlushTimeSeconds;
+        private readonly int _numThreads;
+        private Parser.StatsDOutputParameters _params;
+
+        public bool Stop { get; set; }
+       
+        public override JObject ToJson()
+        {
+            var json = new JObject(
+                new JProperty("statsd",
+                    new JObject(                       
+                        new JProperty("errors", _errorCount),                       
+                        new JProperty("sentMessageCount", _sentMessages),
+                        new JProperty("queuedMessageCount", _jsonQueue.Count),
+                        new JProperty("port", _port),
+                        new JProperty("threads", _numThreads),
+                        new JProperty("flushSize", _flushSize),
+                        new JProperty("idleFlushTime", _idleFlushTimeSeconds),  
+                        new JProperty("maxQueueSize", _maxQueueSize),
+                        new JProperty("overflowDiscardOldest", _queueOverflowDiscardOldest),
+                        new JProperty("interval", _interval),
+                        new JProperty("host", _host)
+                        )));
+                       
+            return json;
+        }
+
+        public StatsDOutput(TimberWinR.Manager manager, Parser.StatsDOutputParameters parameters, CancellationToken cancelToken)
+            : base(cancelToken, "StatsD")
+        {
+            _params = parameters;
+            _manager = manager;                     
+            _port = parameters.Port;
+            _host = parameters.Host;
+            _interval = parameters.Interval;
+            _flushSize = parameters.FlushSize;
+            _idleFlushTimeSeconds = parameters.IdleFlushTimeInSeconds;
+            _maxQueueSize = parameters.MaxQueueSize;
+            _queueOverflowDiscardOldest = parameters.QueueOverflowDiscardOldest;
+            _numThreads = parameters.NumThreads;
+            _jsonQueue = new List<JObject>();
+
+            NStatsD.Client.Host = _host;
+            NStatsD.Client.Port = _port;
+
+            for (int i = 0; i < _numThreads; i++)
+            {
+                Task.Factory.StartNew(StatsDSender, cancelToken, TaskCreationOptions.LongRunning, TaskScheduler.Current);
+            }
+
+        }
+
+        public override string ToString()
+        {
+            return string.Format("StatsD Host: {0} Port: {1}", _host, _port);
+        }
+
+        /// <summary>
+        /// Forward on Json message to Redis Logstash queue
+        /// </summary>
+        /// <param name="jsonMessage"></param>
+        protected override void MessageReceivedHandler(JObject jsonMessage)
+        {
+            if (_manager.Config.Filters != null)
+            {
+                if (ApplyFilters(jsonMessage))
+                    return;
+            }
+           
+
+            var message = jsonMessage.ToString();
+            LogManager.GetCurrentClassLogger().Trace(message);
+
+            lock (_locker)
+            {
+                if (_jsonQueue.Count >= _maxQueueSize)
+                {
+                    // If we've exceeded our queue size, and we're supposed to throw out the oldest objects first,
+                    // then remove as many as necessary to get us under our limit
+                    if (_queueOverflowDiscardOldest)
+                    {
+                        LogManager.GetCurrentClassLogger()
+                            .Warn("Overflow discarding oldest {0} messages", _jsonQueue.Count - _maxQueueSize + 1);
+
+                        _jsonQueue.RemoveRange(0, (_jsonQueue.Count - _maxQueueSize) + 1);
+                    }
+                    // Otherwise we're in a "discard newest" mode, and this is the newest message, so just ignore it
+                    else
+                    {
+                        LogManager.GetCurrentClassLogger()
+                            .Warn("Overflow discarding newest message: {0}", message);
+
+                        return;
+                    }
+                }
+
+                _jsonQueue.Add(jsonMessage);
+            }
+        }
+
+        private bool ApplyFilters(JObject json)
+        {
+            bool drop = false;
+            foreach (var filter in _manager.Config.Filters)
+            {
+                if (!filter.Apply(json))
+                {
+                    LogManager.GetCurrentClassLogger().Debug("{0}: Dropping: {1}", Thread.CurrentThread.ManagedThreadId, json.ToString());
+                    drop = true;
+                }
+            }
+            // Check for matching type (if defined).
+            if (!drop && !string.IsNullOrEmpty(_params.InputType) && json["type"] != null)
+            {
+                string msgType = json["type"].ToString();
+                if (!string.IsNullOrEmpty(msgType) && msgType != _params.InputType)
+                    return true;
+            }
+
+            return drop;
+        }
+
+        // Places messages back into the queue (for a future attempt)
+        private void interlockedInsert(List<JObject> messages)
+        {
+            lock (_locker)
+            {
+                Interlocked.Increment(ref _errorCount);
+                _jsonQueue.InsertRange(0, messages);
+                if (_jsonQueue.Count > _maxQueueSize)
+                {
+                    LogManager.GetCurrentClassLogger().Warn("Exceeded maximum queue depth");
+                }
+            }
+        }
+
+        // 
+        // Pull off messages from the Queue, batch them up and send them all across
+        // 
+        private void StatsDSender()
+        {
+            DateTime lastFlushTime = DateTime.MinValue;
+
+            using (var syncHandle = new ManualResetEventSlim())
+            {
+                // Execute the query
+                while (!Stop)
+                {
+                    if (!CancelToken.IsCancellationRequested)
+                    {
+                        try
+                        {
+                            int messageCount = 0;
+                            List<JObject> messages = new List<JObject>();
+
+                            // Lets get whats in the queue
+                            lock (_locker)
+                            {
+                                messageCount = _jsonQueue.Count;
+
+                                // Time to flush?                             
+                                if (messageCount >= _flushSize || (DateTime.UtcNow - lastFlushTime).Seconds >= _idleFlushTimeSeconds)
+                                {
+                                    messages = _jsonQueue.Take(messageCount).ToList();
+                                    _jsonQueue.RemoveRange(0, messageCount);
+                                    if (messages.Count > 0)
+                                        _manager.IncrementMessageCount(messages.Count);
+                                }
+                            }
+
+                            TransmitStats(messages);
+
+                            if (!Stop)
+                                syncHandle.Wait(TimeSpan.FromMilliseconds(_interval), CancelToken);
+                        }
+                        catch (OperationCanceledException)
+                        {
+                            break;
+                        }
+                        catch (ThreadAbortException)
+                        {
+                            break;
+                        }
+                        catch (Exception ex)
+                        {                           
+                            Interlocked.Increment(ref _errorCount);
+                            LogManager.GetCurrentClassLogger().Error(ex);
+                        }
+                    }
+                }
+            }
+        }
+
+        protected string ExpandField(string fieldName, JObject json)
+        {
+            foreach (var token in json.Children())
+            {
+                string replaceString = "%{" + token.Path + "}";
+                fieldName = fieldName.Replace(replaceString, json[token.Path].ToString());
+            }
+            return fieldName;
+        }
+
+        private string BuildMetricPath(string metric, JObject json)
+        {
+            return string.Format("{0}.{1}.{2}", ExpandField(_params.Namespace, json), ExpandField(_params.Sender, json), ExpandField(metric, json));
+        }
+       
+        private void TransmitStats(List<JObject> messages)
+        {
+            // We've got some to send.
+            if (messages.Count > 0)
+            {
+                do
+                {
+                    try
+                    {
+                        int numMessages = messages.Count;
+                        foreach (var m in messages)
+                        {
+                            SendMetrics(m);
+                        }
+                        messages.RemoveRange(0, numMessages);
+                        Interlocked.Add(ref _sentMessages, numMessages);  
+                    }
+                    catch (Exception ex)
+                    {
+                        LogManager.GetCurrentClassLogger().Error(ex);
+                       
+                        interlockedInsert(messages);  // Put the messages back into the queue
+                        break;
+                    }
+                } while (messages.Count > 0);
+            }
+        }
+
+        // Process all the metrics for this json
+        private void SendMetrics(JObject m)
+        {
+            if (_params.Gauges != null && _params.Gauges.Length > 0)
+                DoGauges(m);
+            if (_params.Counts != null && _params.Counts.Length > 0)
+                DoCounts(m);
+            if (_params.Timings != null && _params.Timings.Length > 0)
+                DoTimings(m);
+            if (_params.Increments != null && _params.Increments.Length > 0)
+                DoIncrements(m);
+            if (_params.Decrements != null && _params.Decrements.Length > 0)
+                DoDecrements(m);
+        }
+
+        // Process the Gauges
+        private void DoGauges(JObject json)
+        {
+            for (int i=0; i<_params.Gauges.Length; i += 2)
+            {
+                string metricPath = BuildMetricPath(_params.Gauges[i], json);
+                string gaugeName = ExpandField(_params.Gauges[i + 1], json);
+                int value;
+                if (int.TryParse(gaugeName, out value))
+                {
+                    NStatsD.Client.Current.Gauge(metricPath, value, _params.SampleRate);
+                }
+            }
+        }
+
+        // Process the Gauges
+        private void DoTimings(JObject json)
+        {
+            for (int i = 0; i < _params.Timings.Length; i += 2)
+            {
+                string metricPath = BuildMetricPath(_params.Timings[i], json);
+                string timingName = ExpandField(_params.Timings[i + 1], json);
+                long value;
+                if (long.TryParse(timingName, out value))
+                {
+                    NStatsD.Client.Current.Timing(metricPath, value, _params.SampleRate);
+                }
+            }
+        }
+
+        // Process the Counts
+        private void DoCounts(JObject json)
+        {
+            for (int i = 0; i < _params.Counts.Length; i += 2)
+            {
+                string metricPath = BuildMetricPath(_params.Counts[i], json);
+                string countName = ExpandField(_params.Counts[i + 1], json);
+                int value;
+                if (int.TryParse(countName, out value))
+                {
+                    NStatsD.Client.Current.UpdateStats(metricPath, value, _params.SampleRate);
+                }
+            }
+        }
+        // Process the Increments
+        private void DoIncrements(JObject json)
+        {
+            foreach (var metric in _params.Increments)
+            {
+                string metricPath = BuildMetricPath(metric, json);
+                NStatsD.Client.Current.Increment(metricPath, _params.SampleRate);
+            }
+        }
+
+        // Process the Increments
+        private void DoDecrements(JObject json)
+        {
+            foreach (var metric in _params.Increments)
+            {
+                string metricPath = BuildMetricPath(metric, json);
+                NStatsD.Client.Current.Decrement(metricPath, _params.SampleRate);
+            }
+        }
+    }
+}

TimberWinR/Outputs/Stdout.cs

@@ -34,6 +34,7 @@ namespace TimberWinR.Outputs
             JObject json = new JObject(
                 new JProperty("stdout",
                     new JObject(
+                        new JProperty("queuedMessageCount", _jsonQueue.Count),
                         new JProperty("sentMessageCount", _sentMessages))));
 
             return json;
@@ -67,7 +68,7 @@ namespace TimberWinR.Outputs
                                     foreach (JObject obj in messages)
                                     {
                                         Console.WriteLine(obj.ToString());
-                                        _sentMessages++;
+                                        Interlocked.Increment(ref _sentMessages);
                                     }
                                 }
                                 catch (Exception ex)

TimberWinR/Parser.cs

@@ -543,6 +543,88 @@ namespace TimberWinR.Parser
         }
     }
 
+    public class StatsDOutputParameters : IValidateSchema
+    {
+        public class StatsDGaugeHashException : Exception
+        {
+            public StatsDGaugeHashException()
+                : base("StatsD output 'gauge' must be an array of pairs.")
+            {
+            }
+        }
+
+        public class StatsDCountHashException : Exception
+        {
+            public StatsDCountHashException()
+                : base("StatsD output 'count' must be an array of pairs.")
+            {
+            }
+        }
+        [JsonProperty(PropertyName = "type")]
+        public string InputType { get; set; }
+        [JsonProperty(PropertyName = "sender")]
+        public string Sender { get; set; }
+        [JsonProperty(PropertyName = "namespace")]
+        public string Namespace { get; set; }    
+        [JsonProperty(PropertyName = "host")]
+        public string Host { get; set; }
+        [JsonProperty(PropertyName = "port")]
+        public int Port { get; set; }
+        [JsonProperty(PropertyName = "interval")]
+        public int Interval { get; set; }
+        [JsonProperty(PropertyName = "flush_size")]
+        public int FlushSize { get; set; }
+        [JsonProperty(PropertyName = "idle_flush_time")]
+        public int IdleFlushTimeInSeconds { get; set; }
+        [JsonProperty(PropertyName = "max_queue_size")]
+        public int MaxQueueSize { get; set; }
+        [JsonProperty(PropertyName = "queue_overflow_discard_oldest")]
+        public bool QueueOverflowDiscardOldest { get; set; }
+        [JsonProperty(PropertyName = "threads")]
+        public int NumThreads { get; set; }
+        [JsonProperty(PropertyName = "sample_rate")]
+        public double SampleRate { get; set; }
+        [JsonProperty(PropertyName = "increment")] // Array: metric names
+        public string[] Increments { get; set; }
+        [JsonProperty(PropertyName = "decrement")] // Array: metric names
+        public string[] Decrements { get; set; }
+        [JsonProperty(PropertyName = "gauge")] // Hash: metric_name => gauge
+        public string[] Gauges { get; set; }
+        [JsonProperty(PropertyName = "count")] // Hash: metric_name => count
+        public string[] Counts { get; set; }
+        [JsonProperty(PropertyName = "timing")] // Hash: metric_name => count
+        public string[] Timings { get; set; }
+
+        public StatsDOutputParameters()
+        {
+            SampleRate = 1;
+            Port = 8125;
+            Host = "localhost";
+            Interval = 5000;
+            FlushSize = 5000;
+            IdleFlushTimeInSeconds = 10;
+            QueueOverflowDiscardOldest = true;
+            MaxQueueSize = 50000;
+            NumThreads = 1;
+            Namespace = "timberwinr";
+            Sender = System.Environment.MachineName.ToLower() + "." +
+                     Microsoft.Win32.Registry.LocalMachine.OpenSubKey(
+                         @"SYSTEM\CurrentControlSet\services\Tcpip\Parameters")
+                         .GetValue("Domain", "")
+                         .ToString().ToLower();
+        }
+
+        public void Validate()
+        {
+            if (Gauges != null && Gauges.Length % 2 != 0)
+                throw new StatsDGaugeHashException();
+
+            if (Counts != null && Counts.Length % 2 != 0)
+                throw new StatsDCountHashException();
+        }
+    }
+
+
     public class ElasticsearchOutputParameters
     {
         const string IndexDatePattern = "(%\\{(?<format>[^\\}]+)\\})";
@@ -668,6 +750,8 @@ namespace TimberWinR.Parser
         }
     }
 
+
+
     public class StdoutOutputParameters
     {
         [JsonProperty(PropertyName = "interval")]
@@ -697,7 +781,7 @@ namespace TimberWinR.Parser
 
         public FileOutputParameters()
         {
-            Format = FormatKind.none;            
+            Format = FormatKind.none;
             Interval = 1000;
             FileName = "timberwinr.out";
         }
@@ -729,6 +813,9 @@ namespace TimberWinR.Parser
 
         [JsonProperty("File")]
         public FileOutputParameters[] File { get; set; }
+
+        [JsonProperty("StatsD")]
+        public StatsDOutputParameters[] StatsD { get; set; }
     }
 
     public class InputSources

TimberWinR/TimberWinR.csproj

@@ -60,13 +60,15 @@
     <Reference Include="NLog">
       <HintPath>..\packages\NLog.3.2.0.0\lib\net40\NLog.dll</HintPath>
     </Reference>
-    <Reference Include="RapidRegex.Core">
-      <HintPath>..\packages\RapidRegex.Core.1.0.0.2\lib\net40\RapidRegex.Core.dll</HintPath>
+    <Reference Include="RapidRegex.Core, Version=1.0.0.4, Culture=neutral, processorArchitecture=MSIL">
+      <SpecificVersion>False</SpecificVersion>
+      <HintPath>..\packages\RapidRegex.Core.1.0.0.4\lib\net40\RapidRegex.Core.dll</HintPath>
     </Reference>
     <Reference Include="RestSharp">
       <HintPath>..\packages\RestSharp.105.0.0\lib\net4\RestSharp.dll</HintPath>
     </Reference>
     <Reference Include="System" />
+    <Reference Include="System.configuration" />
     <Reference Include="System.Core" />
     <Reference Include="System.Linq.Dynamic">
       <HintPath>..\packages\System.Linq.Dynamic.1.0.4\lib\net40\System.Linq.Dynamic.dll</HintPath>
@@ -112,8 +114,11 @@
     <Compile Include="Inputs\WindowsEvtInputListener.cs" />
     <Compile Include="LogErrors.cs" />
     <Compile Include="Manager.cs" />
+    <Compile Include="NStatsD\Client.cs" />
+    <Compile Include="NStatsD\StatsDConfigurationSection.cs" />
     <Compile Include="Outputs\Elasticsearch.cs" />
     <Compile Include="Outputs\OutputSender.cs" />
+    <Compile Include="Outputs\StatsD.cs" />
     <Compile Include="Outputs\Redis.cs" />
     <Compile Include="Outputs\File.cs" />
     <Compile Include="Outputs\Stdout.cs" />
@@ -139,6 +144,7 @@
     </Content>
     <None Include="mdocs\Codec.md" />
     <None Include="mdocs\DateFilter.md" />
+    <None Include="mdocs\StatsD.md" />
     <None Include="mdocs\Filters.md" />
     <None Include="mdocs\GeoIPFilter.md" />
     <None Include="mdocs\Generator.md" />

TimberWinR/mdocs/ElasticsearchOutput.md

@@ -3,7 +3,7 @@
 The Elasticsearch output passes on data directly to Elasticsearch.
 
 ## Parameters
-The following parameters are allowed when configuring the Redis output.
+The following parameters are allowed when configuring the Elasticsearch output.
 
 | Parameter     |   Type   |  Description                                                | Details               |  Default |
 | :-------------|:---------|:------------------------------------------------------------| :---------------------------  | :-- |

TimberWinR/mdocs/GrokFilter.md

@@ -33,7 +33,7 @@ The following operations are allowed when mutating a field.
 | *remove_field*  | property:array  |If the filter is successful, remove arbitrary fields from this event.  Field names can be dynamic and include parts of the event using the %{field} syntax.                                
 | *remove_tag*    | property:array  |If the filter is successful, remove arbitrary tags from this event.  Field names can be dynamic and include parts of the event using the %{field} syntax.                          
 | *rename*        | property:array  |Rename one or more fields                                       
-| *type*          | property:string |Type to which this filter applyes, if empty, applies to all types.
+| *type*          | property:string |Type to which this filter applies, if empty, applies to all types.
 
 ## Operation Details
 ### match 

TimberWinR/mdocs/StatsD.md

@@ -0,0 +1,78 @@
+# Output: StatsD
+
+The StatsD output passes on data directly to StatsD. (https://github.com/etsy/statsd)
+
+## Parameters
+The following parameters are allowed when configuring the StatsD output.
+
+| Parameter     |   Type   |  Description                                                | Details               |  Default |
+| :-------------|:---------|:------------------------------------------------------------| :---------------------------  | :-- |
+| *count*                         | string | Array of (metric_name, gauge name) pairs counted | Must come in pairs  |  |
+| *decrement*                     | string | Array of metrics to be decremented |  |  |
+| *flush_size*                    | integer  | Maximum number of messages before flushing           |   | 50000  |
+| *gauge*                         | string | Array of (metric_name, gauge name) pairs gauged | Must come in pairs  |  |
+| *host*                          | string | Hostname or IP of StatsD server | localhost |  |
+| *idle_flush_time*               | integer  | Maximum number of seconds elapsed before triggering a flush           |   | 10  |
+| *increment*                     | string | Array of metrics to be incremented |  |  |
+| *interval*                      | integer  | Interval in milliseconds to sleep between sends     | Interval       | 5000 |
+| *max_queue_size*                | integer  | Maximum StatsD queue depth       |  | 50000 |
+| *namespace*                     | string | Namespace for stats | timberwinr |  |
+| *port*                          | integer  | StatsD port number                                   | This port must be open  | 8125  |
+| *queue_overflow_discard_oldest* | bool  | If true, discard oldest messages when max_queue_size reached otherwise discard newest |  | true |
+| *sample_rate*                   | integer  | StatsD sample rate           |   | 1  |
+| *sender*                        | string | Sender name | FQDN |  |
+| *threads*                       | string    | Number of Threads processing messages          | | 1 |
+| *timing*                        | string | Array of (metric_name, timing_name) pairs timed | Must come in pairs  |  |
+| *type*                          | string |Type to which this filter applies, if empty, applies to all types.
+
+### Example Usage
+Example Input: Tail an apache log file, and record counts for bytes and increments for response codes.
+
+sample-apache.log (snip)
+```
+180.76.5.25 - - [13/May/2015:17:02:26 -0700] "GET /frameset.htm HTTP/1.1" 404 89 "-" "Mozilla/5.0 (compatible; Baiduspider/2.0; +http://www.baidu.com/search/spider.html)" "www.redlug.com"
+208.115.113.94 - - [13/May/2015:17:03:55 -0700] "GET /robots.txt HTTP/1.1" 200 37 "-" "Mozilla/5.0 (compatible; DotBot/1.1; http://www.opensiteexplorer.org/dotbot, help@moz.com)" "redlug.com"
+208.115.113.94 - - [13/May/2015:17:03:55 -0700] "GET /robots.txt HTTP/1.1" 200 37 "-" "Mozilla/5.0 (compatible; DotBot/1.1; http://www.opensiteexplorer.org/dotbot, help@moz.com)" "www.redlug.com"
+```
+
+TimberWinR configuration
+
+```json
+{
+    "TimberWinR": {
+        "Inputs": {
+            "TailFiles": [
+                {
+                    "interval": 5,
+                    "logSource": "apache log files",
+                    "location": "..\\sample-apache.log",
+                    "recurse": -1
+                }
+            ]
+        },
+        "Filters": [
+            {
+                "grok": {
+                    "type": "Win32-TailLog",
+                    "match": [
+                        "Text",
+                        "%{COMBINEDAPACHELOG}"
+                    ]
+                }
+            }
+        ],
+        "Outputs": {            
+            "StatsD": [
+                {
+                    "type": "Win32-TailLog",
+                    "port": 8125,
+                    "host": "stats.mycompany.svc",
+                    "increment":  ["apache.response.%{response}"],
+                    "count":  ["apache.bytes", "%{bytes}"]
+                }
+            ]
+        }
+    }
+}
+
+```

TimberWinR/packages.config

@@ -7,7 +7,7 @@
   <package id="NEST" version="1.3.1" targetFramework="net40" />
   <package id="Newtonsoft.Json" version="6.0.5" targetFramework="net40" />
   <package id="NLog" version="3.2.0.0" targetFramework="net40" />
-  <package id="RapidRegex.Core" version="1.0.0.2" targetFramework="net40" />
+  <package id="RapidRegex.Core" version="1.0.0.4" targetFramework="net40" />
   <package id="RestSharp" version="105.0.0" targetFramework="net40" />
   <package id="System.Linq.Dynamic" version="1.0.4" targetFramework="net40" />
   <package id="Topshelf" version="3.1.4" targetFramework="net40" />

chocolateyUninstall.ps1.template

@@ -2,6 +2,6 @@ $packageName = 'TimberWinR-${version}' # arbitrary name for the package, used in
 $installerType = 'msi' #only one of these: exe, msi, msu
 $scriptPath =  $(Split-Path $MyInvocation.MyCommand.Path)
 $fileFullPath = Join-Path $scriptPath 'TimberWinR-${version}.0.msi'
-$silentArgs = '{CC4DF908-07C4-4BD8-A9FA-6E6AC315E30B} /quiet' 
+$silentArgs = '{54D22382-5BDE-4CD7-90D4-91AEC0D9B447} /quiet' 
 $validExitCodes = @(0) #please insert other valid exit codes here, exit codes for ms http://msdn.microsoft.com/en-us/library/aa368542(VS.85).aspx
 UnInstall-ChocolateyPackage "$packageName" "$installerType" "$silentArgs" "fileFullPath" -validExitCodes $validExitCodes