<commit_before>﻿using System;
using NLog;
using NLog.Targets;
using NLog.Config;
using System.Text;
using System.Threading;
using System.ComponentModel;
using PaletteInsightAgent.Helpers;
using System.Net.Http;
using System.Net.Http.Headers;
using System.Collections.Concurrent;
using System.Threading.Tasks;

namespace SplunkNLog
{
    public class SplunkMessage
    {
        public string[] Event { get; set; }
    }

    [Target("SplunkNLog")]
    public sealed class SplunkNLog : TargetWithLayout
    {
        public SplunkNLog()
        {
            isStopping = false;

            messagesToSplunk = new ConcurrentQueue<string>();

            stopSign = new EventWaitHandle(false, EventResetMode.ManualReset);
            hasMessageToLog = new EventWaitHandle(false, EventResetMode.AutoReset);

            MaxPendingQueueSize = 65000;
            MaxBatchSize = 100;

            SplunkerThread = new Thread(ProcessLogMessages);
            // Make sure that the worker thread is configured as a background thread,
            // otherwise it will be a frontend thread, and it will make the agent
            // not being able to stop.
            SplunkerThread.IsBackground = true;
            SplunkerThread.Start();
        }

        ~SplunkNLog()
        {
            Dispose();
        }

        [RequiredParameter]
        public string Host { get; set; }

        [RequiredParameter]
        public int Port { get; set; }

        [RequiredParameter]
        public string Token { get; set; }

        [DefaultValue(65000), RequiredParameter]
        public int MaxPendingQueueSize { get; set; }

        [DefaultValue(100), RequiredParameter]
        public int MaxBatchSize { get; set; }


        private ConcurrentQueue<string> messagesToSplunk;
        private EventWaitHandle         stopSign;
        private EventWaitHandle         hasMessageToLog;
        private Thread                  SplunkerThread;
        private bool                    isStopping;

        private void ProcessLogMessages()
        {
            WaitHandle[] handles = { stopSign, hasMessageToLog };

            while (true)
            {
                int index = WaitHandle.WaitAny(handles);
                if (index == 0)
                {
                    // Stop signal received. Quit.
                    return;
                }

                if (messagesToSplunk.Count > MaxPendingQueueSize)
                {
                    // Discard the oldest messages, until we get back below the limit.
                    messagesToSplunk.DiscardItems(messagesToSplunk.Count - MaxPendingQueueSize);
                }

                int messageCount = messagesToSplunk.Count;
                if (messageCount <= 0)
                {
                    continue;
                }

                // Do not work with too large batches
                if (messageCount > MaxBatchSize) messageCount = MaxBatchSize;

                string[] messageBatch = new string[messageCount];
                for (int i = 0; i < messageCount; ++i)
                {
                    try
                    {
                        messageBatch[i] = messagesToSplunk.Dequeue();
                    }
                    catch (InvalidOperationException)
                    {
                        // Failed to dequeue anything. This shouldn't happen, because it means
                        // that the queue got empty somehow. But we are supposed to be the
                        // only thread consuming this queue... Anyway, put an empty string there.
                        messageBatch[i] = "";
                    }
                }

                if (isStopping)
                {
                    // Shutting down. Abort.
                    return;
                }

                try
                {
                    SplunkMessage spm = new SplunkMessage();
                    spm.Event = messageBatch;

                    fastJSON.JSONParameters param = new fastJSON.JSONParameters();
                    param.SerializeToLowerCaseNames = true;
                    param.UseExtensions = false;
                    string jsonContent = fastJSON.JSON.ToJSON(spm, param);
                    byte[] byteArray = Encoding.UTF8.GetBytes(jsonContent);

                    // Try to send messages to Splunk, until we have space in the buffer.
                    // This is the way we are handling network issues.
                    while (messagesToSplunk.Count < MaxPendingQueueSize)
                    {
                        var result = PostSplunkEvent(byteArray);
                        result.Wait();
                        if (result.Result)
                        {
                            /// Successfully posted message to Splunk. No need to re-try.
                            break;
                        }
                    }
                }
                catch (Exception)
                {
                    continue;
                }
            }
        }

        /// <summary>
        /// Posts the given byte array to the Splunk server.
        /// </summary>
        /// <param name="splunkEvent"></param>
        /// <returns>This function returns true, if the post operation was successful, or it is
        /// expected to end up with the same result always. If this function returns false, it 
        /// might be wise to re-try the post operation.</returns>
        private async Task<bool> PostSplunkEvent(byte[] splunkEvent)
        {
            using (var handler = APIClient.GetHttpClientHandler())
            using (var httpClient = new HttpClient(handler))
            {
                var authHeader = new AuthenticationHeaderValue("Splunk", Token);
                httpClient.DefaultRequestHeaders.Authorization = authHeader;

                try
                {
                    using (var response = await httpClient.PostAsync(String.Format("{0}:{1}/services/collector/event", this.Host, this.Port), new ByteArrayContent(splunkEvent)))
                    {
                        if (response.StatusCode == System.Net.HttpStatusCode.RequestTimeout)
                        {
                            return false;
                        }
                    }
                }
                catch (HttpRequestException)
                {
                    return false;
                }
                catch (Exception)
                {
                    // We believe that this exception would be permanent, so there is no point
                    // in re-trying in this case.
                }
            }

            return true;
        }

        protected override void Write(LogEventInfo logEvent)
        {
            string logMessage = this.Layout.Render(logEvent);
            messagesToSplunk.Enqueue(logMessage);
            hasMessageToLog.Set();
        }

        protected override void CloseTarget()
        {
            isStopping = true;
            stopSign.Set();

            base.CloseTarget();<commit_msg>Retry on TaskCanceledException too<commit_after>﻿using System;
using NLog;
using NLog.Targets;
using NLog.Config;
using System.Text;
using System.Threading;
using System.ComponentModel;
using PaletteInsightAgent.Helpers;
using System.Net.Http;
using System.Net.Http.Headers;
using System.Collections.Concurrent;
using System.Threading.Tasks;

namespace SplunkNLog
{
    public class SplunkMessage
    {
        public string[] Event { get; set; }
    }

    [Target("SplunkNLog")]
    public sealed class SplunkNLog : TargetWithLayout
    {
        public SplunkNLog()
        {
            isStopping = false;

            messagesToSplunk = new ConcurrentQueue<string>();

            stopSign = new EventWaitHandle(false, EventResetMode.ManualReset);
            hasMessageToLog = new EventWaitHandle(false, EventResetMode.AutoReset);

            MaxPendingQueueSize = 65000;
            MaxBatchSize = 100;

            SplunkerThread = new Thread(ProcessLogMessages);
            // Make sure that the worker thread is configured as a background thread,
            // otherwise it will be a frontend thread, and it will make the agent
            // not being able to stop.
            SplunkerThread.IsBackground = true;
            SplunkerThread.Start();
        }

        ~SplunkNLog()
        {
            Dispose();
        }

        [RequiredParameter]
        public string Host { get; set; }

        [RequiredParameter]
        public int Port { get; set; }

        [RequiredParameter]
        public string Token { get; set; }

        [DefaultValue(65000), RequiredParameter]
        public int MaxPendingQueueSize { get; set; }

        [DefaultValue(100), RequiredParameter]
        public int MaxBatchSize { get; set; }


        private ConcurrentQueue<string> messagesToSplunk;
        private EventWaitHandle         stopSign;
        private EventWaitHandle         hasMessageToLog;
        private Thread                  SplunkerThread;
        private bool                    isStopping;

        private void ProcessLogMessages()
        {
            WaitHandle[] handles = { stopSign, hasMessageToLog };

            while (true)
            {
                int index = WaitHandle.WaitAny(handles);
                if (index == 0)
                {
                    // Stop signal received. Quit.
                    return;
                }

                if (messagesToSplunk.Count > MaxPendingQueueSize)
                {
                    // Discard the oldest messages, until we get back below the limit.
                    messagesToSplunk.DiscardItems(messagesToSplunk.Count - MaxPendingQueueSize);
                }

                int messageCount = messagesToSplunk.Count;
                if (messageCount <= 0)
                {
                    continue;
                }

                // Do not work with too large batches
                if (messageCount > MaxBatchSize) messageCount = MaxBatchSize;

                string[] messageBatch = new string[messageCount];
                for (int i = 0; i < messageCount; ++i)
                {
                    try
                    {
                        messageBatch[i] = messagesToSplunk.Dequeue();
                    }
                    catch (InvalidOperationException)
                    {
                        // Failed to dequeue anything. This shouldn't happen, because it means
                        // that the queue got empty somehow. But we are supposed to be the
                        // only thread consuming this queue... Anyway, put an empty string there.
                        messageBatch[i] = "";
                    }
                }

                if (isStopping)
                {
                    // Shutting down. Abort.
                    return;
                }

                try
                {
                    SplunkMessage spm = new SplunkMessage();
                    spm.Event = messageBatch;

                    fastJSON.JSONParameters param = new fastJSON.JSONParameters();
                    param.SerializeToLowerCaseNames = true;
                    param.UseExtensions = false;
                    string jsonContent = fastJSON.JSON.ToJSON(spm, param);
                    byte[] byteArray = Encoding.UTF8.GetBytes(jsonContent);

                    // Try to send messages to Splunk, until we have space in the buffer.
                    // This is the way we are handling network issues.
                    while (messagesToSplunk.Count < MaxPendingQueueSize)
                    {
                        var result = PostSplunkEvent(byteArray);
                        result.Wait();
                        if (result.Result)
                        {
                            /// Successfully posted message to Splunk. No need to re-try.
                            break;
                        }
                    }
                }
                catch (Exception)
                {
                    continue;
                }
            }
        }

        /// <summary>
        /// Posts the given byte array to the Splunk server.
        /// </summary>
        /// <param name="splunkEvent"></param>
        /// <returns>This function returns true, if the post operation was successful, or it is
        /// expected to end up with the same result always. If this function returns false, it 
        /// might be wise to re-try the post operation.</returns>
        private async Task<bool> PostSplunkEvent(byte[] splunkEvent)
        {
            using (var handler = APIClient.GetHttpClientHandler())
            using (var httpClient = new HttpClient(handler))
            {
                var authHeader = new AuthenticationHeaderValue("Splunk", Token);
                httpClient.DefaultRequestHeaders.Authorization = authHeader;

                try
                {
                    using (var response = await httpClient.PostAsync(String.Format("{0}:{1}/services/collector/event", this.Host, this.Port), new ByteArrayContent(splunkEvent)))
                    {
                        if (response.StatusCode == System.Net.HttpStatusCode.RequestTimeout)
                        {
                            return false;
                        }
                    }
                }
                catch (Exception ex)
                {
                    if (ex is HttpRequestException || ex is TaskCanceledException)
                    {
                        return false;
                    }
                    // We believe that this exception would be permanent, so there is no point
                    // in re-trying in this case.
                }
            }

            return true;
        }

        protected override void Write(LogEventInfo logEvent)
        {
            string logMessage = this.Layout.Render(logEvent);
            messagesToSplunk.Enqueue(logMessage);
            hasMessageToLog.Set();
        }

        protected override void CloseTarget()
        {
            isStopping = true;
            stopSign.Set();

            base.CloseTarget();
{'fraction_non_alphanumeric': 0.07211267605633803, 'fraction_numerical': 0.0030985915492957746, 'mean_word_length': 1.2768959435626102, 'pattern_counts': {'":': 0, '<': 27, '<?xml version=': 0, '>': 25, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '6852857', 'n_tokens_mistral': 3423, 'n_tokens_neox': 3129, 'n_words': 910}