<commit_before>﻿// ------------------------------------------------------------
//  Copyright (c) Microsoft Corporation.  All rights reserved.
//  Licensed under the MIT License (MIT). See License.txt in the repo root for license information.
// ------------------------------------------------------------

namespace Microsoft.Diagnostics.EventListeners
{
    using System;
    using System.Collections.Generic;
    using System.Configuration;
    using System.Diagnostics;
    using System.Threading;
    using System.Threading.Tasks;
    using Nest;

    public class ElasticSearchListener : BufferingEventListener, IDisposable
    {
        private const string Dot = ".";
        private const string Dash = "-";
        // TODO: make it a (configuration) property of the listener
        private const string EventDocumentTypeName = "event";
        private ElasticSearchConnectionData connectionData;
        // TODO: support for multiple ES nodes/connection pools, for failover and load-balancing        

        public ElasticSearchListener(ICompositeConfigurationProvider configurationProvider, IHealthReporter healthReporter) : base(configurationProvider, healthReporter)
        {
            if (this.Disabled)
            {
                return;
            }

            Debug.Assert(configurationProvider != null);
            this.CreateConnectionData(configurationProvider);

            this.Sender = new ConcurrentEventSender<EventData>(
                eventBufferSize: 1000,
                maxConcurrency: 2,
                batchSize: 100,
                noEventsDelay: TimeSpan.FromMilliseconds(1000),
                transmitterProc: this.SendEventsAsync,
                healthReporter: healthReporter);
        }

        private ElasticClient CreateElasticClient(ICompositeConfigurationProvider configurationProvider)
        {
            ICompositeConfigurationProvider esConfigurationProvider = configurationProvider.GetConfiguration("ElasticSearchListener");
            string esServiceUriString = esConfigurationProvider.GetValue("serviceUri");
            Uri esServiceUri;
            bool serviceUriIsValid = Uri.TryCreate(esServiceUriString, UriKind.Absolute, out esServiceUri);
            if (!serviceUriIsValid)
            {
                throw new ConfigurationErrorsException("serviceUri must be a valid, absolute URI");
            }

            string userName = esConfigurationProvider.GetValue("userName");
            string password = esConfigurationProvider.GetValue("password");
            if (string.IsNullOrWhiteSpace(userName) || string.IsNullOrWhiteSpace(password))
            {
                throw new ConfigurationErrorsException("Invalid Elastic Search credentials");
            }

            ConnectionSettings config = new ConnectionSettings(esServiceUri).BasicAuthentication(userName, password);
            return new ElasticClient(config);
        }

        private void CreateConnectionData(object sender)
        {
            ICompositeConfigurationProvider configurationProvider = (ICompositeConfigurationProvider) sender;

            this.connectionData = new ElasticSearchConnectionData();
            this.connectionData.Client = this.CreateElasticClient(configurationProvider);
            this.connectionData.LastIndexName = null;
            string indexNamePrefix = configurationProvider.GetValue("indexNamePrefix");
            this.connectionData.IndexNamePrefix = string.IsNullOrWhiteSpace(indexNamePrefix) ? string.Empty : indexNamePrefix + Dash;
        }

        private async Task SendEventsAsync(IEnumerable<EventData> events, long transmissionSequenceNumber, CancellationToken cancellationToken)
        {
            if (events == null)
            {
                return;
            }

            try
            {
                string currentIndexName = this.GetIndexName(this.connectionData);
                if (!string.Equals(currentIndexName, this.connectionData.LastIndexName, StringComparison.Ordinal))
                {
                    await this.EnsureIndexExists(currentIndexName, this.connectionData.Client);
                    this.connectionData.LastIndexName = currentIndexName;
                }

                BulkRequest request = new BulkRequest();
                request.Refresh = true;

                List<IBulkOperation> operations = new List<IBulkOperation>();
                foreach (EventData eventData in events)
                {
                    BulkCreateOperation<EventData> operation = new BulkCreateOperation<EventData>(eventData);
                    operation.Index = currentIndexName;
                    operation.Type = EventDocumentTypeName;
                    operations.Add(operation);
                }

                request.Operations = operations;

                if (cancellationToken.IsCancellationRequested)
                {
                    return;
                }

                // Note: the NEST client is documented to be thread-safe so it should be OK to just reuse the this.esClient instance
                // between different SendEventsAsync callbacks.
                // Reference: https://www.elastic.co/blog/nest-and-elasticsearch-net-1-3
                IBulkResponse response = await this.connectionData.Client.BulkAsync(request);
                if (!response.IsValid)
                {
                    this.ReportEsRequestError(response, "Bulk upload");
                }

                this.ReportListenerHealthy();
            }
            catch (Exception e)
            {
                this.ReportListenerProblem("Diagnostics data upload has failed." + Environment.NewLine + e.ToString());
            }
        }

        private async Task EnsureIndexExists(string currentIndexName, ElasticClient esClient)
        {
            IExistsResponse existsResult = await esClient.IndexExistsAsync(currentIndexName);
            if (!existsResult.IsValid)
            {
                this.ReportEsRequestError(existsResult, "Index exists check");
            }

            if (existsResult.Exists)
            {
                return;
            }

            // TODO: allow the consumer to fine-tune index settings
            IndexState indexSettings = new IndexState();
            indexSettings.Settings.NumberOfReplicas = 1;<commit_msg>Fix ES NullReferenceException bug<commit_after>﻿// ------------------------------------------------------------
//  Copyright (c) Microsoft Corporation.  All rights reserved.
//  Licensed under the MIT License (MIT). See License.txt in the repo root for license information.
// ------------------------------------------------------------

namespace Microsoft.Diagnostics.EventListeners
{
    using System;
    using System.Collections.Generic;
    using System.Configuration;
    using System.Diagnostics;
    using System.Threading;
    using System.Threading.Tasks;
    using Nest;

    public class ElasticSearchListener : BufferingEventListener, IDisposable
    {
        private const string Dot = ".";
        private const string Dash = "-";
        // TODO: make it a (configuration) property of the listener
        private const string EventDocumentTypeName = "event";
        private ElasticSearchConnectionData connectionData;
        // TODO: support for multiple ES nodes/connection pools, for failover and load-balancing        

        public ElasticSearchListener(ICompositeConfigurationProvider configurationProvider, IHealthReporter healthReporter) : base(configurationProvider, healthReporter)
        {
            if (this.Disabled)
            {
                return;
            }

            Debug.Assert(configurationProvider != null);
            this.CreateConnectionData(configurationProvider);

            this.Sender = new ConcurrentEventSender<EventData>(
                eventBufferSize: 1000,
                maxConcurrency: 2,
                batchSize: 100,
                noEventsDelay: TimeSpan.FromMilliseconds(1000),
                transmitterProc: this.SendEventsAsync,
                healthReporter: healthReporter);
        }

        private ElasticClient CreateElasticClient(ICompositeConfigurationProvider configurationProvider)
        {
            ICompositeConfigurationProvider esConfigurationProvider = configurationProvider.GetConfiguration("ElasticSearchListener");
            string esServiceUriString = esConfigurationProvider.GetValue("serviceUri");
            Uri esServiceUri;
            bool serviceUriIsValid = Uri.TryCreate(esServiceUriString, UriKind.Absolute, out esServiceUri);
            if (!serviceUriIsValid)
            {
                throw new ConfigurationErrorsException("serviceUri must be a valid, absolute URI");
            }

            string userName = esConfigurationProvider.GetValue("userName");
            string password = esConfigurationProvider.GetValue("password");
            if (string.IsNullOrWhiteSpace(userName) || string.IsNullOrWhiteSpace(password))
            {
                throw new ConfigurationErrorsException("Invalid Elastic Search credentials");
            }

            ConnectionSettings config = new ConnectionSettings(esServiceUri).BasicAuthentication(userName, password);
            return new ElasticClient(config);
        }

        private void CreateConnectionData(object sender)
        {
            ICompositeConfigurationProvider configurationProvider = (ICompositeConfigurationProvider) sender;

            this.connectionData = new ElasticSearchConnectionData();
            this.connectionData.Client = this.CreateElasticClient(configurationProvider);
            this.connectionData.LastIndexName = null;
            string indexNamePrefix = configurationProvider.GetValue("indexNamePrefix");
            this.connectionData.IndexNamePrefix = string.IsNullOrWhiteSpace(indexNamePrefix) ? string.Empty : indexNamePrefix + Dash;
        }

        private async Task SendEventsAsync(IEnumerable<EventData> events, long transmissionSequenceNumber, CancellationToken cancellationToken)
        {
            if (events == null)
            {
                return;
            }

            try
            {
                string currentIndexName = this.GetIndexName(this.connectionData);
                if (!string.Equals(currentIndexName, this.connectionData.LastIndexName, StringComparison.Ordinal))
                {
                    await this.EnsureIndexExists(currentIndexName, this.connectionData.Client);
                    this.connectionData.LastIndexName = currentIndexName;
                }

                BulkRequest request = new BulkRequest();
                request.Refresh = true;

                List<IBulkOperation> operations = new List<IBulkOperation>();
                foreach (EventData eventData in events)
                {
                    BulkCreateOperation<EventData> operation = new BulkCreateOperation<EventData>(eventData);
                    operation.Index = currentIndexName;
                    operation.Type = EventDocumentTypeName;
                    operations.Add(operation);
                }

                request.Operations = operations;

                if (cancellationToken.IsCancellationRequested)
                {
                    return;
                }

                // Note: the NEST client is documented to be thread-safe so it should be OK to just reuse the this.esClient instance
                // between different SendEventsAsync callbacks.
                // Reference: https://www.elastic.co/blog/nest-and-elasticsearch-net-1-3
                IBulkResponse response = await this.connectionData.Client.BulkAsync(request);
                if (!response.IsValid)
                {
                    this.ReportEsRequestError(response, "Bulk upload");
                }

                this.ReportListenerHealthy();
            }
            catch (Exception e)
            {
                this.ReportListenerProblem("Diagnostics data upload has failed." + Environment.NewLine + e.ToString());
            }
        }

        private async Task EnsureIndexExists(string currentIndexName, ElasticClient esClient)
        {
            IExistsResponse existsResult = await esClient.IndexExistsAsync(currentIndexName);
            if (!existsResult.IsValid)
            {
                this.ReportEsRequestError(existsResult, "Index exists check");
            }

            if (existsResult.Exists)
            {
                return;
            }

            // TODO: allow the consumer to fine-tune index settings
            IndexState indexSettings = new IndexState();
            indexSettings.Settings = new IndexSettings();
            indexSettings.Settings.NumberOfReplicas = 1;
{'fraction_non_alphanumeric': 0.08689191302987642, 'fraction_numerical': 0.002346316283435007, 'mean_word_length': 1.9953150620754274, 'pattern_counts': {'":': 0, '<': 15, '<?xml version=': 0, '>': 15, 'https://': 2, 'lorem ipsum': 0, 'www.': 2, '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': '6270558', 'n_tokens_mistral': 2847, 'n_tokens_neox': 2701, 'n_words': 641}