//******************************************************************************************************//  InfluxDBOutputAdapter.cs - Gbtc////  Copyright © 2010, Grid Protection Alliance.  All Rights Reserved.////  Licensed to the Grid Protection Alliance (GPA) under one or more contributor license agreements. See//  the NOTICE file distributed with this work for additional information regarding copyright ownership.//  The GPA licenses this file to you under the MIT License (MIT), the "License"; you may//  not use this file except in compliance with the License. You may obtain a copy of the License at:////      http://www.opensource.org/licenses/MIT////  Unless agreed to in writing, the subject software distributed under the License is distributed on an//  "AS-IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. Refer to the//  License for the specific language governing permissions and limitations.////  Code Modification History://  ----------------------------------------------------------------------------------------------------//  07/02/2014 - J. Ritchie Carroll//       Generated original version of source code.////****************************************************************************************************** using System;using System.Collections.Generic;using System.ComponentModel;using System.IO;using System.Linq;using System.Net;using System.Text;using System.Threading;using System.Threading.Tasks;using GSF;using GSF.Diagnostics;using GSF.Threading;using GSF.TimeSeries;using GSF.TimeSeries.Adapters; namespace InfluxDBAdapters{    /// <summary>    /// Represents an output adapter that archives measurements to a local archive.    /// </summary>    [Description("InfluxDB: Archives measurements to an InfluxDB instance")]    public class InfluxDBOutputAdapter : OutputAdapterBase    {        #region [ Members ]         // Constants         /// <summary>        /// Default value for <see cref="UserName"/>.        /// </summary>        public const string DefaultUserName = "root";         /// <summary>        /// Default value for <see cref="Password"/>.        /// </summary>        public const string DefaultPassword = "root";         /// <summary>        /// Default value for <see cref="UseParallelPosting"/>.        /// </summary>        public const bool DefaultUseParallelPosting = true;         /// <summary>        /// Default value for <see cref="ValuesPerPost"/>.        /// </summary>        public const int DefaultValuesPerPost = 50;         // Fields        private readonly ShortSynchronizedOperation m_requestRestart;         private Uri m_requestUri;               // Data posting URI for InfluxDB connection        private string m_serverUri;             // Base server URI for InfluxDB connection        private string m_databaseName;          // Database name forInfluxDB connection        private string m_userName;              // Username for InfluxDB connection        private string m_password;              // Password for InfluxDB connection        private bool m_useParallelPosting;      // Enable parallel posting        private int m_valuesPerPost;            // Maximum values to send per post (when using parallel posting)        private string m_connectionResponse;    // Response from connection attempt        private long m_totalValues;             // Total archived values        private long m_totalPosts;              // Total post to the InfluxDB connection        private long m_totalParallelGroups;     // Total measurement groups processed in parallel        private long m_ignoredErrors;           // Total ignored 400 and 500 HTTP status code errors         #endregion         #region [ Constructors ]         /// <summary>        /// Creates a new <see cref="InfluxDBOutputAdapter"/>;        /// </summary>        public InfluxDBOutputAdapter()        {            m_requestRestart = new ShortSynchronizedOperation(() =>            {                if (Enabled)                    Start();            });        }         #endregion         #region [ Properties ]         /// <summary>        /// Gets or sets the base server URI for the InfluxDB connection.        /// </summary>        [ConnectionStringParameter, Description("Defines the base server URI for the InfluxDB connection (e.g., http://localhost:8086).")]        public string ServerUri        {            get            {                return m_serverUri;            }            set            {                m_serverUri = value;            }        }         /// <summary>        /// Gets or sets the database name for the InfluxDB connection.        /// </summary>        [ConnectionStringParameter, Description("Defines the database name for the InfluxDB connection. Database will not be created, it is expected to already exist.")]        public string DatabaseName        {            get            {                return m_databaseName;            }            set            {                m_databaseName = value;            }        }         /// <summary>        /// Gets or sets the user name for the InfluxDB connection.        /// </summary>        [ConnectionStringParameter, Description("Defines the the user name for the InfluxDB connection."), DefaultValue(DefaultUserName)]        public string UserName        {            get            {                return m_userName;            }            set            {                m_userName = value;            }        }         /// <summary>        /// Gets or sets the password for the InfluxDB connection.        /// </summary>        [ConnectionStringParameter, Description("Defines the password for the InfluxDB connection."), DefaultValue(DefaultPassword)]        public string Password        {            get            {                return m_password;            }            set            {                m_password = value;            }        }         /// <summary>        /// Gets or sets flag that determines if multiple posts to InfluxDB should be made in parallel.        /// </summary>        [ConnectionStringParameter, Description("Defines flag that determines if multiple posts to InfluxDB should be made in parallel."), DefaultValue(DefaultUseParallelPosting)]        public bool UseParallelPosting        {            get            {                return m_useParallelPosting;            }            set            {                m_useParallelPosting = value;            }        }         /// <summary>        /// Gets or sets the maximum values to send per post when <see cref="UseParallelPosting"/> is <c>true</c> for the InfluxDB connection.        /// </summary>        [ConnectionStringParameter, Description("When parallel posting is enabled, defines the maximum values to send per post for the InfluxDB connection."), DefaultValue(DefaultValuesPerPost)]        public int ValuesPerPost        {            get            {                return m_valuesPerPost;            }            set            {                m_valuesPerPost = value;            }        }         /// <summary>        /// Gets or sets whether or not to automatically place measurements back into the processing        /// queue if an exception occurs while processing.  Defaults to false.        /// </summary>        [ConnectionStringParameter, Description("Defines whether or not to automatically place measurements back into the processing queue if an exception occurs while processing.  For InfluxDB adapter this defaults to true."), DefaultValue(true)]        public override bool RequeueOnException        {            get            {                return base.RequeueOnException;            }            set            {                base.RequeueOnException = value;            }        }         /// <summary>        /// Returns a flag that determines if measurements sent to this <see cref="InfluxDBOutputAdapter"/> are destined for archival.        /// </summary>        public override bool OutputIsForArchive        {            get            {                return true;            }        }         /// <summary>        /// Gets flag that determines if this <see cref="InfluxDBOutputAdapter"/> uses an asynchronous connection.        /// </summary>        protected override bool UseAsyncConnect        {            get            {                return false;            }        }         /// <summary>        /// Gets a detailed status for this <see cref="InfluxDBOutputAdapter"/>.        /// </summary>        public override string Status        {            get            {                StringBuilder status = new StringBuilder();                 status.Append(base.Status);                 status.AppendFormat("         InfluxDB base URI: {0}", m_serverUri);                status.AppendLine();                status.AppendFormat("      Use parallel posting: {0}", m_useParallelPosting);                status.AppendLine();                 if (m_useParallelPosting)                {                    status.AppendFormat("   Maximum values per post: {0:N0}", m_valuesPerPost);                    status.AppendLine();                    status.AppendFormat("  Average parallel threads: {0:0.0}", m_totalParallelGroups / (double)InternalProcessQueue.TotalFunctionCalls);                    status.AppendLine();                }                 status.AppendFormat("     Total archived values: {0:N0}", m_totalValues);                status.AppendLine();                status.AppendFormat("               Total posts: {0:N0}", m_totalPosts);                status.AppendLine();                status.AppendFormat("   Average values per post: {0:R}", Math.Round(m_totalValues / (double)m_totalPosts, 2));                status.AppendLine();                status.AppendFormat("       Connection response: {0}", m_connectionResponse);                status.AppendLine();                status.AppendFormat("Ignored error status codes: {0:N0} total 400 or 500", m_ignoredErrors);                status.AppendLine();                 return status.ToString();            }        }         #endregion         #region [ Methods ]         /// <summary>        /// Returns a brief status of this <see cref="InfluxDBOutputAdapter"/>        /// </summary>        /// <param name="maxLength">Maximum number of characters in the status string</param>        /// <returns>Status</returns>        public override string GetShortStatus(int maxLength)        {            return string.Format("Archived {0:N0} measurements via {1:N0} posts to \"{2}\".", m_totalValues, m_totalPosts, m_databaseName).CenterText(maxLength);        }         /// <summary>        /// Initializes <see cref="InfluxDBOutputAdapter"/>.        /// </summary>        public override void Initialize()        {            base.Initialize();             Dictionary<string, string> settings = Settings;            string setting;             if (!settings.TryGetValue("serverUri", out m_serverUri))                throw new InvalidOperationException("Base server URI is a required setting for the InfluxDB connection. Please add a server URI setting in the format of \"serverUri=http://localhost:8086\" to the connection string.");             if (!settings.TryGetValue("databaseName", out m_databaseName))                throw new InvalidOperationException("Database name is a required setting for the InfluxDB connection. Please add a database name setting in the format of \"databaseName=MyDB\" to the connection string.");             if (settings.TryGetValue("userName", out setting))                m_userName = setting;            else                m_userName = DefaultUserName;             if (settings.TryGetValue("password", out setting))                m_password = setting;            else                m_password = DefaultPassword;             if (settings.TryGetValue("useParallelPosting", out setting))                m_useParallelPosting = setting.ParseBoolean();            else                m_useParallelPosting = DefaultUseParallelPosting;             if (!settings.TryGetValue("valuesPerPost", out setting) || !int.TryParse(setting, out m_valuesPerPost))                m_valuesPerPost = DefaultValuesPerPost;             // Since InfluxDB API is web based, you generally want to retry posts in case of issues            if (!settings.ContainsKey("requeueOnException"))                RequeueOnException = true;             // Define request URI            UriBuilder requestUri = new UriBuilder(m_serverUri);            requestUri.Path = string.Format("db/{0}/series", m_databaseName.UriEncode());            requestUri.Query = string.Format("u={0}&p={1}", m_userName.UriEncode(), m_password.UriEncode());            m_requestUri = requestUri.Uri;        }         /// <summary>        /// Attempts to connect to InfluxDB database.        /// </summary>        protected override void AttemptConnection()        {            try            {                // Setup an authenticate request                UriBuilder requestUri = new UriBuilder(m_serverUri);                requestUri.Path = string.Format("db/{0}/authenticate", m_databaseName.UriEncode());                requestUri.Query = string.Format("u={0}&p={1}", m_userName.UriEncode(), m_password.UriEncode());                 // Create a web request for authentication that will at least make sure the server is available                HttpWebRequest request = WebRequest.Create(requestUri.Uri) as HttpWebRequest;                 if ((object)request != null)                {                    request.ContentType = "application/json";                     // Attempt query - if this doesn't throw an exception, query succeeded                    using (WebResponse response = request.GetResponse())                    {                        m_connectionResponse = ((HttpWebResponse)response).StatusDescription;                    }                }                else                {                    throw new InvalidOperationException("Failed to create an HTTP request from " + requestUri);                }            }            catch (Exception ex)            {                m_connectionResponse = ex.Message;                 // Rethrow any captured exceptions, this will restart connection cycle in OutputAdapterBase                throw;            }        }         /// <summary>        /// Attempts to disconnect from InfluxDB.        /// </summary>        protected override void AttemptDisconnection()        {            // Connections are stateless, nothing to do here...        }         /// <summary>        /// Serializes measurements to InfluxDB.        /// </summary>        protected override void ProcessMeasurements(IMeasurement[] measurements)        {            if (measurements.Length == 0)                return;             if (m_useParallelPosting)            {                List<IMeasurement[]> measurementGroups = new List<IMeasurement[]>();                IMeasurement[] measurementGroup = measurements.Take(m_valuesPerPost).ToArray();                int skipCount = measurementGroup.Length;                 while (measurementGroup.Length > 0)                {                    measurementGroups.Add(measurementGroup);                    measurementGroup = measurements.Skip(skipCount).Take(m_valuesPerPost).ToArray();                    skipCount += measurementGroup.Length;                }                 Parallel.ForEach(measurementGroups, PostMeasurementsToArchive);                m_totalParallelGroups += measurementGroups.Count;            }            else            {                PostMeasurementsToArchive(measurements);            }        }         private void PostMeasurementsToArchive(IMeasurement[] measurements)        {            const string PostFormat = "{{\"name\":\"{0}\",\"columns\":[\"time\",\"value\",\"quality\"],\"points\":[[{1},{2},\"{3}\"]]}}";             try            {                HttpWebRequest request = WebRequest.Create(m_requestUri) as HttpWebRequest;                 if ((object)request == null)                    throw new InvalidOperationException("Failed to create an HTTP request from " + m_requestUri);                 request.Method = "POST";                request.KeepAlive = true;                request.SendChunked = true;                request.AllowWriteStreamBuffering = true;                request.ContentType = "application/json";                 // Build a JSON post expression with measurement values to use as post data                StringBuilder jsonData = new StringBuilder();                 foreach (IMeasurement measurement in measurements)                {                    if (jsonData.Length > 0)                        jsonData.Append(',');                    else                        jsonData.Append('[');                     jsonData.AppendFormat(PostFormat, measurement.Key, GetEpochMilliseconds(measurement.Timestamp), measurement.AdjustedValue, measurement.StateFlags);                }                 jsonData.Append(']');                 // Encode JSON data as UTF8                byte[] postData = Encoding.UTF8.GetBytes(jsonData.ToString());                 // Write data to request stream                request.ContentLength = postData.Length;                 using (Stream requestStream = request.GetRequestStream())                {                    requestStream.Write(postData, 0, postData.Length);                }                 // Post data                WebResponse response = request.GetResponse();                response.Dispose();                 Interlocked.Add(ref m_totalValues, measurements.Length);                Interlocked.Increment(ref m_totalPosts);            }            catch (Exception ex)            {                if (RequeueOnException)                    InternalProcessQueue.InsertRange(0, measurements);                 bool ignoreError = false;                 // Under load, 400 and 500 status code errors seem to be occasionally reported - we ignore these                WebException webex = ex as WebException;                 if ((object)webex != null)                {                    HttpWebResponse webResponse = webex.Response as HttpWebResponse;                     if ((object)webResponse != null)                    {                        HttpStatusCode statusCode = webResponse.StatusCode;                        ignoreError = (statusCode == HttpStatusCode.BadRequest || statusCode == HttpStatusCode.InternalServerError);                    }                }                 if (!ignoreError)                {                    OnProcessException(MessageLevel.Warning, ex);                     // So long as user hasn't requested to stop, restart connection cycle when exceptions occur                    m_requestRestart.RunOnceAsync();                }                else                {                    Interlocked.Increment(ref m_ignoredErrors);                }            }        }

Lines 1–500 of 509. Load more