SessionAnalysisProcessor.cs

        /// <summary>
        /// Check every stinkin' session event to see if they're affected by the latest redaction rules
        /// </summary>
        public void ProcessRedactionRuleChange(CentralRepository repository)
        {
            Log.Write(LogMessageSeverity.Information, LogCategory, "Checking existing session events for candidates to update based on newest redaction rules",
                "We will check the caption of the log event for each session event to determine if it may be affected by a redaction rule and if so queue it to be recalculated.");
            //we need to be ready for millions and millions of log events, all of which could match.. So we have to
            //be cautious about memory and locks.
            using (var context = repository.GetContext())
            {
                context.Database.CommandTimeout = 600;
                context.Configuration.AutoDetectChangesEnabled = false;

                //get the list of application ids - we process these one at a time so we can sort within them. 
                //More painful than I'd like, but we need to sort by the clustered index but work in batches.
                var applicationSummaries = context.Set<Application>()
                    .OrderBy(a => a.Product.Name)
                    .ThenBy(a => a.Name)
                    .Select(a => new { a.Id, ProductName = a.Product.Name, ApplicationName = a.Name }).ToList();

                int batchSize = 1000;

                var cancellationTokenSource = new CancellationTokenSource();
                var captionRewriters = new ConcurrentDictionary<Guid, CentralRepositoryLogMessageRewriter>();
                var sessionEventQueue = new BlockingCollection<SessionEventSummaryModel>(batchSize); //limit how much we'll keep in RAM at a go..

                //set up our processing threads.  This is pretty lightweight so we'll be aggressive.
                int numberOfThreads = Math.Max(2, Math.Min(Environment.ProcessorCount, 8)); //between 2 and 8 threads.
                var tasks = new Task[numberOfThreads];
                for (int i = 0; i < tasks.Length; i++)
                {
                    tasks[i] = Task.Factory.StartNew(AsyncCheckSessionEventsForChange, new object[] { repository, captionRewriters, sessionEventQueue, cancellationTokenSource }, cancellationTokenSource.Token, TaskCreationOptions.LongRunning, TaskScheduler.Default);
                }

                //now pull the data and dump it into the queue
                try
                {
                    foreach (var applicationSummary in applicationSummaries)
                    {
                        Log.Write(LogMessageSeverity.Information, LogCategory, string.Format("Checking Session Events from {0} {1} for captions to be rewritten", applicationSummary.ProductName, applicationSummary.ApplicationName),
                            "We pull the first log event from each session event, see if it should be rewritten and if so queue the session event to be examined asynchronously.");
                        SessionEventSummaryModel lastSummary = null;
                        do
                        {
                            var rawData = context.Set<SessionEvent>()
                                .Where(se => se.ApplicationId == applicationSummary.Id);

                            if (lastSummary != null)
                            {
                                //we want to start just greater than the last one we checked.. so add that where clause..
                                var lastFingerprintChecked = lastSummary.FingerprintHash;
                                rawData = rawData.Where(se => se.FingerprintHash.CompareTo(lastFingerprintChecked) > 0);
                                lastSummary = null;
                            }

                            var nextBatch = rawData
                                .OrderBy(se => se.FingerprintHash) //we need an order by to use skip/take, this is an efficient one because it aligns with the clustered index (after our appid filter)
                                .Select(se => new SessionEventSummaryModel
                                {
                                    Id = se.Id,
                                    Caption = se.LastLogEvent.Caption,
                                    ProductName = se.Application.Product.Name,
                                    ApplicationName = se.Application.Name,
                                    ApplicationId = se.ApplicationId,
                                    FingerprintHash = se.FingerprintHash
                                });

                            foreach (var summary in nextBatch.Take(batchSize))
                            {
                                sessionEventQueue.Add(summary, cancellationTokenSource.Token);
                                lastSummary = summary;
                            }

                        } while (lastSummary != null);
                    }
                }
                finally
                {
                    //we need to be sure to let all our tasks know there are no more items.
                    sessionEventQueue.CompleteAdding();
                }
            }

            Log.Write(LogMessageSeverity.Information, LogCategory, "Completed queuing log events for signature recalculation",
                "At this point every log event that may be affected by a redaction rule has been queued to have its signature recalculated.");
        }


        /// <summary>
        /// Work with a blocking collection of simple session events to see if they would likely be changed by a redaction rule
        /// </summary>
        /// <param name="state"></param>
        private void AsyncCheckSessionEventsForChange(object state)
        {
            //unbundle our arguments
            var stateArray = (object[])state;
            var repository = (CentralRepository)stateArray[0];
            var logMessageRewriters = (ConcurrentDictionary<Guid, CentralRepositoryLogMessageRewriter>)stateArray[1];
            var sessionEventQueue = (BlockingCollection<SessionEventSummaryModel>)stateArray[2];
            var cancellationTokenSource = (CancellationTokenSource)stateArray[3];

            bool completed = false;
            do
            {
                try
                {
                    foreach (var sessionEventSummary in sessionEventQueue.GetConsumingEnumerable(cancellationTokenSource.Token))
                    {
                        CentralRepositoryLogMessageRewriter logMessageRewriter = null;
                        if (logMessageRewriters.TryGetValue(sessionEventSummary.ApplicationId, out logMessageRewriter) == false)
                        {
                            logMessageRewriter = new CentralRepositoryLogMessageRewriter(repository, sessionEventSummary.ProductName, sessionEventSummary.ApplicationName);
                            logMessageRewriters.TryAdd(sessionEventSummary.ApplicationId, logMessageRewriter); //these are expensive to load up so we want to cache them.
                        }

                        string newCaption;
                        if (logMessageRewriter.RewriteLogMessageCaption(sessionEventSummary.Caption, out newCaption))
                        {
                            //queue this one for rewriting
                            m_ServerContext.EventQueue.QueueEvent(repository.Id, QueuedEvent.SourceSessionEvent, QueuedEvent.EventSessionEventSignatureRecalculate, sessionEventSummary.Id, null, false);
                        }
                    }

                    completed = true; //the consuming enumerable is complete
                }
                catch (OperationCanceledException)
                {
                    // we can safely ignore this - our cancellation token was canceled
                    completed = true;
                }
                catch (Exception ex)
                {
                    // make sure to catch any exceptions that occur... letting this bubble could crash the process
                    //if we did fail, it is likely a transient problem, wait a minute before  rolling around to try again.
                    Log.Write(LogMessageSeverity.Warning, LogWriteMode.Queued, ex, true, LogCategory, "Session event signature recalculation interrupted due to " + ex.GetType(), "Due to an exception we stopped checking session events.  We'll wait a minute before attempting to start checking again.  Original Exception:\r\n{0}: {1}", ex.GetType(), ex.Message);
                    cancellationTokenSource.SleepUnlessCanceled(60);
                }
            } while (completed == false);
        }