path: root/Software/Visual_Studio/Tango.Telemetry/TelemetryPublisher.cs

using System;
using System.Collections.Concurrent;
using System.Collections.Generic;
using System.Linq;
using System.Reflection;
using System.Text;
using System.Threading;
using System.Threading.Tasks;
using System.Timers;
using Tango.Core;
using Tango.Insights;
using Tango.Integration.Operation;
using Tango.PMR.Diagnostics;
using Tango.PMR.Insights;

namespace Tango.Telemetry
{
    public class TelemetryPublisher : ExtendedObject, IDisposable
    {
        public const int MIN_SAMPLING_INTERVAL_SECONDS = 1;

        public event EventHandler<TelemetryPackagePublishingEventArgs> PublishingPackage;
        public event EventHandler<TelemetryPackagePublishedEventArgs> PackagePublished;
        public event EventHandler<TelemetryPackagePublishFailedEventArgs> PublishPackageFailed;

        private System.Timers.Timer _diagnosticsSamplingTimer;
        private System.Timers.Timer _pendingStorageCheckTimer;

        private Thread _publishThread;
        private ProducerConsumerQueue<TelemetryPublishPackage> _publishQueue;

        private List<StartDiagnosticsResponse> _diagnosticsQueue;
        private bool _writing;
        private bool _emptyWritten;

        private IMachineOperator _machineOperator;
        private TelemetryPendingStorageManager _pendingStorageManager;

        #region Properties

        public TelemetryPublisherConfiguration Config { get; private set; }
        public bool IsStarted { get; private set; }

        #endregion

        #region Constructor

        public TelemetryPublisher(IMachineOperator machineOperator, TelemetryPendingStorageManager storageManager, TelemetryPublisherConfiguration config)
        {
            _machineOperator = machineOperator;
            _pendingStorageManager = storageManager;

            Config = config ?? new TelemetryPublisherConfiguration();

            _publishQueue = new ProducerConsumerQueue<TelemetryPublishPackage>();
            _publishThread = new Thread(PublishThreadMethod);
            _publishThread.IsBackground = true;

            _diagnosticsQueue = new List<StartDiagnosticsResponse>();

            RegisterForEvents();
        }

        #endregion

        #region Register / Unregister Events

        private void RegisterForEvents()
        {
            _machineOperator.DiagnosticsDataAvailable += MachineOperator_DiagnosticsDataAvailable;
        }

        private void UnregisterEvents()
        {
            _machineOperator.DiagnosticsDataAvailable -= MachineOperator_DiagnosticsDataAvailable;
        }

        #endregion

        #region Start / Stop

        public void Start()
        {
            if (!IsStarted)
            {
                Config.Validate();

                IsStarted = true;

                if (Config.DiagnosticsSamplingInterval.TotalSeconds < MIN_SAMPLING_INTERVAL_SECONDS)
                {
                    Config.DiagnosticsSamplingInterval = TimeSpan.FromSeconds(MIN_SAMPLING_INTERVAL_SECONDS);
                }

                if (_diagnosticsSamplingTimer == null)
                {
                    _diagnosticsSamplingTimer = new System.Timers.Timer();
                    _diagnosticsSamplingTimer.Interval = Config.DiagnosticsSamplingInterval.TotalMilliseconds;
                    _diagnosticsSamplingTimer.Elapsed += DiagnosticsSamplingTimer_Elapsed;
                }

                if (_pendingStorageCheckTimer == null)
                {
                    _pendingStorageCheckTimer = new System.Timers.Timer();
                    _pendingStorageCheckTimer.Interval = Config.PendingStorageCheckInterval.TotalMilliseconds;
                    _pendingStorageCheckTimer.Elapsed += PendingStorageCheckTimer_Elapsed;
                }

                _diagnosticsQueue.Clear();

                _writing = false;
                _diagnosticsSamplingTimer.Start();
                _pendingStorageCheckTimer.Start();

                _publishThread.Start();
            }
        }

        public void Stop()
        {
            if (IsStarted)
            {
                IsStarted = false;
                _diagnosticsSamplingTimer.Stop();
                _pendingStorageCheckTimer.Stop();
                _diagnosticsQueue.Clear();
                _publishQueue.BlockEnqueue(null);
            }
        }

        #endregion

        #region Incoming Data Event Handlers

        private void MachineOperator_DiagnosticsDataAvailable(object sender, StartDiagnosticsResponse diagnostics)
        {
            if (IsStarted && diagnostics.Monitors != null)
            {
                _diagnosticsQueue.Add(diagnostics);
            }
        }

        #endregion

        #region Timers

        private void DiagnosticsSamplingTimer_Elapsed(object sender, ElapsedEventArgs e)
        {
            if (!IsStarted || _writing) return;

            try
            {
                _writing = true;

                if (_diagnosticsQueue.Count > 0)
                {
                    var queue = _diagnosticsQueue.ToList();
                    _diagnosticsQueue.Clear();
                    _emptyWritten = false;

                    var monitorsAvg = InsightsHelper.AverageMonitors(queue.Select(x => x.Monitors).ToList());
                    queue.Clear();

                    TelemetryDiagnosticsFrame frame = new TelemetryDiagnosticsFrame();
                    frame.Monitors = monitorsAvg;
                    frame.Time = DateTime.UtcNow.Subtract(Config.DiagnosticsSamplingInterval);

                    PushTelemetryPackage(frame, TelemetrySource.Streaming);
                }
                else
                {
                    if (!_emptyWritten)
                    {
                        TelemetryDiagnosticsFrame frame = new TelemetryDiagnosticsFrame();
                        frame.Monitors = new InsightsMonitors();
                        frame.Time = DateTime.UtcNow.Subtract(Config.DiagnosticsSamplingInterval);
                        PushTelemetryPackage(frame, TelemetrySource.Streaming);
                        _emptyWritten = true;
                    }
                }
            }
            catch (Exception ex)
            {
                LogManager.Log(ex, "Error occurred on insights frame insertion.");
            }
            finally
            {
                _writing = false;
            }
        }

        private void PendingStorageCheckTimer_Elapsed(object sender, ElapsedEventArgs e)
        {
            var telemetryAll = _pendingStorageManager.GetTelemetryAll();

            foreach (var t in telemetryAll)
            {
                PushTelemetryPackage(t, TelemetrySource.PendingStorage);
            }
        }

        #endregion

        #region Push

        private void PushTelemetryPackage(ITelemetry telemetryObject, TelemetrySource source)
        {
            _publishQueue.BlockEnqueue(new TelemetryPublishPackage() { TelemetryObject = telemetryObject, Source = source });
        }

        private void PushTelemetryPackage(TelemetryPublishPackage package)
        {
            _publishQueue.BlockEnqueue(package);
        }

        #endregion

        #region Publish

        private async void PublishThreadMethod()
        {
            while (IsStarted)
            {
                TelemetryPublishPackage package = _publishQueue.BlockDequeue();
                if (package == null)
                {
                    _publishQueue = new ProducerConsumerQueue<TelemetryPublishPackage>();
                    return;
                }

                try
                {
                    await PublishTelemetryPackage(package);
                }
                catch
                {
                    Thread.Sleep(1000);
                }
            }
        }

        private async Task PublishTelemetryPackage(TelemetryPublishPackage package)
        {
            List<KeyValuePair<String, String>> properties = new List<KeyValuePair<string, string>>();

            properties.Add(new KeyValuePair<string, string>("MachineID", Config.MachineID));
            properties.Add(new KeyValuePair<string, string>("Model", Config.MachineType.ToShortName()));
            properties.Add(new KeyValuePair<string, string>("Environment", Config.Environment));

            List<String> pendingDestinations = package.TelemetryObject.PendingDestinations.ToList();

            //Add all destinations if streaming or external (They will be remove later if successfull)
            //If source is "PendingStorage" the "PendingDestination" would be already propagated from the pending storage db.
            if (package.Source == TelemetrySource.Streaming || package.Source == TelemetrySource.ExternalStorage)
            {
                pendingDestinations.AddRange(Config.TelemetryDestinations.Select(x => x.GetType().Name));
            }

            pendingDestinations = pendingDestinations.Distinct().ToList();

            foreach (var destination in Config.TelemetryDestinations.Where(x => x.SupportedSources.Contains(package.Source)))
            {
                String destinationName = destination.GetType().Name;

                if (pendingDestinations.Contains(destinationName))
                {
                    try
                    {
                        pendingDestinations.Remove(destinationName);
                        if (OnPublishingPackage(package, destination))
                        {
                            await destination.Publish(package, properties);
                            OnPackagePublished(package, destination);
                        }
                    }
                    catch (Exception ex)
                    {
                        LogManager.Log(ex, $"Error publishing telemetry package to destination {destinationName}.");

                        OnPackagePublishFailed(package, destination, ex);

                        if (destination.SupportedSources.Contains(TelemetrySource.PendingStorage))
                        {
                            if (!pendingDestinations.Contains(destinationName))
                            {
                                pendingDestinations.Add(destinationName);
                            }
                        }
                    }
                }
            }

            package.TelemetryObject.PendingDestinations = pendingDestinations;

            if (package.Source == TelemetrySource.PendingStorage && package.TelemetryObject.PendingDestinations.Count == 0)
            {
                _pendingStorageManager.DeleteTelemetryObject(package.TelemetryObject);
            }
            else
            {
                _pendingStorageManager.InsertOrUpdateTelemetryObject(package.TelemetryObject);
            }
        }

        #endregion

        #region Virtual Methods

        protected virtual bool OnPublishingPackage(TelemetryPublishPackage package, ITelemetryDestination destination)
        {
            try
            {
                var args = new TelemetryPackagePublishingEventArgs() { Package = package, Destination = destination };
                PublishingPackage?.Invoke(this, args);
                return !args.Cancel;
            }
            catch
            {
                return true;
            }
        }

        protected virtual void OnPackagePublished(TelemetryPublishPackage package, ITelemetryDestination destination)
        {
            try
            {
                PublishingPackage?.Invoke(this, new TelemetryPackagePublishingEventArgs() { Package = package, Destination = destination });
            }
            catch { }
        }

        protected virtual void OnPackagePublishFailed(TelemetryPublishPackage package, ITelemetryDestination destination, Exception exception)
        {
            try
            {
                PublishPackageFailed?.Invoke(this, new TelemetryPackagePublishFailedEventArgs() { Package = package, Destination = destination, Exception = exception });
            }
            catch { }
        }

        #endregion

        #region Dispose

        public void Dispose()
        {
            UnregisterEvents();

            if (IsStarted)
            {
                Stop();
            }

            foreach (var destination in Config.TelemetryDestinations)
            {
                destination.Dispose();
            }
        }

        #endregion
    }
}