Оптимизация обращений к хранилищу и синхронизации потоков

* Отправка данных из разных источников логов теперь выполняется общими запросами. что значительно уменьшает количество обращений к хранилищу (в текущей версии к ClickHouse) * Оптимизация синхронизации потоков. Теперь поток отправки данных в хранилище не блокируем работу потокв чтения логов. Ранее могли возникать кратковременные ожидания.
2026-06-19 21:46:53 +02:00 · 2021-05-01 10:46:11 +05:00
parent a21401fb8b
commit 4c63ccb338
13 changed files with 311 additions and 191 deletions
@@ -27,20 +27,16 @@ namespace YY.TechJournalExportAssistantWithSharedBufferConsoleApp
            Console.WriteLine("Good luck & bye!");
        }

-        private static void OnError(
-            TechJournalSettings.LogSourceSettings settings,
-            OnErrorExportSharedBufferEventArgs args)
+        private static void OnError(OnErrorExportSharedBufferEventArgs args)
        {
-            Console.WriteLine($"Ошибка при экспорте логов ({settings?.Name ?? "<>"}): {args.Exception}");
+            Console.WriteLine($"Ошибка при экспорте логов: {args.Exception}");
        }

-        private static void OnSend(
-            TechJournalSettings.LogSourceSettings settings,
-            OnSendLogFromSharedBufferEventArgs args)
+        private static void OnSend(OnSendLogFromSharedBufferEventArgs args)
        {
-            Console.WriteLine($"Отправка данных в хранилище ({settings?.Name ?? "<>"}):\n" +
-                              $"Записей: {args._rows.Select(e => e.Value.Count).Sum()}\n" +
-                              $"Актуальных позиций чтения: {args._positions.Count}");
+            Console.WriteLine($"Отправка данных в хранилище:\n" +
+                              $"Записей: {args.DataFromBuffer.Values.SelectMany(i => i.LogRows).Select(i => i.Value).Count()}\n" +
+                              $"Актуальных позиций чтения: {args.DataFromBuffer.Values.Select(i => i.LogPosition).Count() }");
        }
    }
 }
@@ -45,6 +45,132 @@ namespace YY.TechJournalExportAssistant.ClickHouse

        #region RowsData

+        public void SaveRowsData(Dictionary<LogBufferItemKey, LogBufferItem> sourceDataFromBuffer)
+        {
+            List<object[]> rowsForInsert = new List<object[]>();
+            List<object[]> positionsForInsert = new List<object[]>();
+            Dictionary<string, LastRowsInfoByLogFile> maxPeriodByDirectories = new Dictionary<string, LastRowsInfoByLogFile>();
+
+            var dataFromBuffer = sourceDataFromBuffer
+                .OrderBy(i => i.Key.Period)
+                .ThenBy(i => i.Value.LogPosition.EventNumber)
+                .ToList();
+            long itemNumber = 0;
+            foreach (var dataItem in dataFromBuffer)
+            {
+                itemNumber++;
+                FileInfo logFileInfo = new FileInfo(dataItem.Key.LogFile);
+
+                positionsForInsert.Add(new object[]
+                {
+                    dataItem.Key.Settings.TechJournalLog.Name,
+                    logFileInfo.Directory?.Name ?? string.Empty,
+                    DateTime.Now.Ticks + itemNumber,
+                    logFileInfo.Name,
+                    logFileInfo.CreationTimeUtc,
+                    logFileInfo.LastWriteTimeUtc,
+                    dataItem.Value.LogPosition.EventNumber,
+                    dataItem.Value.LogPosition.CurrentFileData.Replace("\\", "\\\\"),
+                    dataItem.Value.LogPosition.StreamPosition ?? 0
+                });
+
+                foreach (var rowData in dataItem.Value.LogRows)
+                {
+                    if (!maxPeriodByDirectories.TryGetValue(logFileInfo.FullName, out LastRowsInfoByLogFile lastInfo))
+                    {
+                        if (logFileInfo.Directory != null)
+                        {
+                            GetRowsDataMaxPeriodAndId(
+                                dataItem.Key.Settings.TechJournalLog,
+                                logFileInfo.Directory.Name,
+                                logFileInfo.Name,
+                                rowData.Value.Period,
+                                out var maxPeriod,
+                                out var maxId
+                            );
+                            lastInfo = new LastRowsInfoByLogFile(maxPeriod, maxId);
+                            maxPeriodByDirectories.Add(logFileInfo.FullName, lastInfo);
+                        }
+                    }
+
+                    bool existByPeriod = lastInfo.MaxPeriod > ClickHouseHelpers.MinDateTimeValue &&
+                                         rowData.Value.Period.Truncate(TimeSpan.FromSeconds(1)) <= lastInfo.MaxPeriod;
+                    bool existById = lastInfo.MaxId > 0 &&
+                                     rowData.Value.Id <= lastInfo.MaxId;
+                    if (existByPeriod && existById)
+                        continue;
+
+                    var eventItem = rowData.Value;
+                    rowsForInsert.Add(new object[]
+                        {
+                            dataItem.Key.Settings.TechJournalLog.Name,
+                            logFileInfo.Directory.Name,
+                            logFileInfo.Name,
+                            eventItem.Id,
+                            eventItem.Period,
+                            eventItem.Level,
+                            eventItem.Duration,
+                            eventItem.DurationSec,
+                            eventItem.EventName ?? string.Empty,
+                            eventItem.ServerContextName ?? string.Empty,
+                            eventItem.ProcessName ?? string.Empty,
+                            eventItem.SessionId ?? 0,
+                            eventItem.ApplicationName ?? string.Empty,
+                            eventItem.ClientId ?? 0,
+                            eventItem.ComputerName ?? string.Empty,
+                            eventItem.ConnectionId ?? 0,
+                            eventItem.UserName ?? string.Empty,
+                            eventItem.ApplicationId ?? 0,
+                            eventItem.Context ?? string.Empty,
+                            eventItem.ActionType.GetDescription() ?? string.Empty,
+                            eventItem.Database ?? string.Empty,
+                            eventItem.DatabaseCopy ?? string.Empty,
+                            eventItem.DBMS.GetPresentation() ?? string.Empty,
+                            eventItem.DatabasePID ?? string.Empty,
+                            eventItem.PlanSQLText ?? string.Empty,
+                            eventItem.Rows ?? 0,
+                            eventItem.RowsAffected ?? 0,
+                            eventItem.SQLText ?? string.Empty,
+                            eventItem.SQLQueryOnly ?? string.Empty,
+                            eventItem.SQLQueryParametersOnly ?? string.Empty,
+                            eventItem.SQLQueryHash ?? string.Empty,
+                            eventItem.SDBL ?? string.Empty,
+                            eventItem.Description ?? string.Empty,
+                            eventItem.Message ?? string.Empty,
+                            eventItem.GetCustomFieldsAsJSON() ?? string.Empty
+                        });
+                }
+            }
+
+            if (rowsForInsert.Count > 0)
+            {
+                using (ClickHouseBulkCopy bulkCopyInterface = new ClickHouseBulkCopy(_connection)
+                {
+                    DestinationTableName = "EventData",
+                    BatchSize = 100000,
+                    MaxDegreeOfParallelism = 4
+                })
+                {
+                    var bulkResult = bulkCopyInterface.WriteToServerAsync(rowsForInsert);
+                    bulkResult.Wait();
+                    rowsForInsert.Clear();
+                }
+            }
+
+            if (positionsForInsert.Count > 0)
+            {
+                using (ClickHouseBulkCopy bulkCopyInterface = new ClickHouseBulkCopy(_connection)
+                {
+                    DestinationTableName = "LogFiles",
+                    BatchSize = 100000
+                })
+                {
+                    var bulkResult = bulkCopyInterface.WriteToServerAsync(positionsForInsert);
+                    bulkResult.Wait();
+                }
+            }
+        }
+
        public void SaveRowsData(TechJournalLogBase techJournalLog, 
            List<EventData> eventData, 
            string fileName,
@@ -55,7 +181,7 @@ namespace YY.TechJournalExportAssistant.ClickHouse

            SaveRowsData(techJournalLog, eventDataToInsert);
        }
-
+        
        public void SaveRowsData(TechJournalLogBase techJournalLog, 
            IDictionary<string, List<EventData>> eventData,
            Dictionary<string, LastRowsInfoByLogFile> maxPeriodByFiles = null)
@@ -195,9 +321,7 @@ namespace YY.TechJournalExportAssistant.ClickHouse
        }
        
        public IDictionary<string, TechJournalPosition> GetCurrentLogPositions(
-            TechJournalLogBase techJournalLog,
-            TechJournalSettings settings,
-            KeyValuePair<TechJournalSettings.LogSourceSettings, LogBufferItem> logBufferItem)
+            TechJournalLogBase techJournalLog)
        {
            var cmdGetLastLogFileInfo = _connection.CreateCommand();
            cmdGetLastLogFileInfo.CommandText = Resource.Query_GetActualPositions;
@@ -2,7 +2,6 @@
 using System.Linq;
 using Microsoft.Extensions.Configuration;
 using YY.TechJournalExportAssistant.Core;
-using YY.TechJournalExportAssistant.Core.SharedBuffer;
 using YY.TechJournalReaderAssistant;
 using EventData = YY.TechJournalReaderAssistant.Models.EventData;

@@ -118,13 +117,13 @@ namespace YY.TechJournalExportAssistant.ClickHouse
            }
        }

-        public override IDictionary<string, TechJournalPosition> GetCurrentLogPositions(TechJournalSettings settings, KeyValuePair<TechJournalSettings.LogSourceSettings, LogBufferItem> logBufferItem)
+        public override IDictionary<string, TechJournalPosition> GetCurrentLogPositions(TechJournalSettings settings)
        {
            IDictionary<string, TechJournalPosition> positions;

            using (var context = new ClickHouseContext(_connectionString))
            {
-                positions = context.GetCurrentLogPositions(_techJournalLog, settings, logBufferItem);
+                positions = context.GetCurrentLogPositions(_techJournalLog);
            }

            return positions;
@@ -7,23 +7,38 @@ namespace YY.TechJournalExportAssistant.ClickHouse
 {
    public class TechJournalOnClickHouseTargetBuilder : ITechJournalOnTargetBuilder
    {
-        public ITechJournalOnTarget CreateTarget(TechJournalSettings settings, KeyValuePair<TechJournalSettings.LogSourceSettings, LogBufferItem> logBufferItem)
+        public ITechJournalOnTarget CreateTarget(TechJournalSettings settings, KeyValuePair<LogBufferItemKey, LogBufferItem> logBufferItem)
        {
            ITechJournalOnTarget target = new TechJournalOnClickHouse(settings.ConnectionString,
-                logBufferItem.Key.Portion);
-            target.SetInformationSystem(new TechJournalLogBase()
-            {
-                Name = logBufferItem.Key.Name,
-                Description = logBufferItem.Key.Description
-            });
+                logBufferItem.Key.Settings.Portion);
+            target.SetInformationSystem(logBufferItem.Key.Settings.TechJournalLog);

            return target;
        }

-        public IDictionary<string, TechJournalPosition> GetCurrentLogPositions(TechJournalSettings settings, KeyValuePair<TechJournalSettings.LogSourceSettings, LogBufferItem> logBufferItem)
+        public IDictionary<string, TechJournalPosition> GetCurrentLogPositions(TechJournalSettings settings, KeyValuePair<LogBufferItemKey, LogBufferItem> logBufferItem)
        {
            ITechJournalOnTarget target = CreateTarget(settings, logBufferItem);
-            return target.GetCurrentLogPositions(settings, logBufferItem);
+            return target.GetCurrentLogPositions(settings);
+        }
+
+        public IDictionary<string, TechJournalPosition> GetCurrentLogPositions(TechJournalSettings settings, TechJournalLogBase techJournalLog)
+        {
+            Dictionary<string, TechJournalPosition> allPositions = new Dictionary<string, TechJournalPosition>();
+            using (ClickHouseContext context = new ClickHouseContext(settings.ConnectionString))
+            {
+                return context.GetCurrentLogPositions(techJournalLog);
+            }
+        }
+
+        public void SaveRowsData(
+            TechJournalSettings settings,
+            Dictionary<LogBufferItemKey, LogBufferItem> dataFromBuffer)
+        {
+            using (ClickHouseContext context = new ClickHouseContext(settings.ConnectionString))
+            {
+                context.SaveRowsData(dataFromBuffer);
+            }
        }
    }
 }
@@ -15,7 +15,6 @@ namespace YY.TechJournalExportAssistant.Core
        void Save(EventData eventData, string fileName);
        void Save(IList<EventData> eventData, string fileName);
        void Save(IDictionary<string, List<EventData>> rowsData);
-        IDictionary<string, TechJournalPosition> GetCurrentLogPositions(TechJournalSettings settings,
-            KeyValuePair<TechJournalSettings.LogSourceSettings, LogBufferItem> logBufferItem);
+        IDictionary<string, TechJournalPosition> GetCurrentLogPositions(TechJournalSettings settings);
    }
 }
@@ -1,23 +1,15 @@
 using System.Collections.Generic;
-using YY.TechJournalReaderAssistant;
-using YY.TechJournalReaderAssistant.Models;

 namespace YY.TechJournalExportAssistant.Core.SharedBuffer.EventArgs
 {
    public sealed class OnSendLogFromSharedBufferEventArgs : System.EventArgs
    {
-        public TechJournalSettings.LogSourceSettings _settings { get; } 
-        public IDictionary<string, List<EventData>> _rows { get; }
-        public IReadOnlyDictionary<string, TechJournalPosition> _positions { get; }
+        public IReadOnlyDictionary<LogBufferItemKey, LogBufferItem> DataFromBuffer { get; }

        public OnSendLogFromSharedBufferEventArgs(
-            TechJournalSettings.LogSourceSettings settings,
-            IDictionary<string, List<EventData>> rows, 
-            IReadOnlyDictionary<string, TechJournalPosition> positions)
+            IReadOnlyDictionary<LogBufferItemKey, LogBufferItem> dataFromBuffer)
        {
-            _settings = settings;
-            _rows = rows;
-            _positions = positions;
+            DataFromBuffer = dataFromBuffer;
        }
    }    
 }
@@ -5,7 +5,8 @@ namespace YY.TechJournalExportAssistant.Core.SharedBuffer
 {
    public interface ITechJournalOnTargetBuilder
    {
-        ITechJournalOnTarget CreateTarget(TechJournalSettings settings, KeyValuePair<TechJournalSettings.LogSourceSettings, LogBufferItem> logBufferItem);
-        IDictionary<string, TechJournalPosition> GetCurrentLogPositions(TechJournalSettings settings, KeyValuePair<TechJournalSettings.LogSourceSettings, LogBufferItem> logBufferItem);
+        ITechJournalOnTarget CreateTarget(TechJournalSettings settings, KeyValuePair<LogBufferItemKey, LogBufferItem> logBufferItem);
+        IDictionary<string, TechJournalPosition> GetCurrentLogPositions(TechJournalSettings settings, TechJournalLogBase techJournalLog);
+        void SaveRowsData(TechJournalSettings settings, Dictionary<LogBufferItemKey, LogBufferItem> dataFromBuffer);
    }
 }
@@ -26,20 +26,19 @@ namespace YY.TechJournalExportAssistant.Core.SharedBuffer
        public long ItemsCount => LogRows.Count;

        /// <summary>
-        /// Актуальная позиция чтения файлов лога
+        /// Актуальная позиция чтения файла лога
        /// </summary>
-        public ConcurrentDictionary<string, TechJournalPosition> LogPositions { get; set; }
+        public TechJournalPosition LogPosition { get; set; }

        /// <summary>
        /// Записи логов
        /// </summary>
        public ConcurrentDictionary<EventKey, EventData> LogRows { get; set; }
-
+        
        public LogBufferItem()
        {
            Created = DateTime.Now;
            LastUpdate = DateTime.MinValue;
-            LogPositions = new ConcurrentDictionary<string, TechJournalPosition>();
            LogRows = new ConcurrentDictionary<EventKey, EventData>();
        }
    }
@@ -0,0 +1,21 @@
+using System;
+
+namespace YY.TechJournalExportAssistant.Core.SharedBuffer
+{
+    public class LogBufferItemKey
+    {
+        public TechJournalSettings.LogSourceSettings Settings { get; }
+        public DateTime Period { get; }
+        public string LogFile { get; }
+
+        public LogBufferItemKey(
+            TechJournalSettings.LogSourceSettings setting,
+            DateTime period,
+            string logFile)
+        {
+            Settings = setting;
+            Period = period;
+            LogFile = logFile;
+        }
+    }
+}
@@ -2,6 +2,7 @@
 using System.Collections.Concurrent;
 using System.Collections.Generic;
 using System.IO;
+using System.Linq;
 using YY.TechJournalReaderAssistant;
 using YY.TechJournalReaderAssistant.Models;

@@ -9,13 +10,14 @@ namespace YY.TechJournalExportAssistant.Core.SharedBuffer
 {
    public class LogBuffers
    {
-        public readonly ConcurrentDictionary<TechJournalSettings.LogSourceSettings, LogBufferItem> LogBuffer;
-        public readonly object LockObject;
-
+        public readonly ConcurrentDictionary<LogBufferItemKey, LogBufferItem> LogBuffer;
+        
+        public ConcurrentDictionary<TechJournalSettings.LogSourceSettings, ConcurrentDictionary<string, TechJournalPosition>> LogPositions { get; }
+        
        public LogBuffers()
        {
-            LogBuffer = new ConcurrentDictionary<TechJournalSettings.LogSourceSettings, LogBufferItem>();
-            LockObject = new object();
+            LogBuffer = new ConcurrentDictionary<LogBufferItemKey, LogBufferItem>();
+            LogPositions = new ConcurrentDictionary<TechJournalSettings.LogSourceSettings, ConcurrentDictionary<string, TechJournalPosition>>();
        }

        /// <summary>
@@ -25,17 +27,9 @@ namespace YY.TechJournalExportAssistant.Core.SharedBuffer
        {
            get
            {
-                long totalItemsCount = 0;
-
-                foreach (var logBufferItem in LogBuffer)
-                {
-                    //lock (logBufferItem.Key.LockObject)
-                    //{
-                        totalItemsCount += logBufferItem.Value.ItemsCount;
-                    //}
-                }
-
-                return totalItemsCount;
+                return LogBuffer
+                    .Select(e => e.Value.ItemsCount)
+                    .Sum();
            }
        }

@@ -47,16 +41,14 @@ namespace YY.TechJournalExportAssistant.Core.SharedBuffer

                foreach (var logBufferItem in LogBuffer)
                {
-                    //lock (logBufferItem.Key.LockObject)
-                    //{
-                        if (bufferCreated == DateTime.MinValue)
-                        {
-                            bufferCreated = logBufferItem.Value.Created;
-                        } else if (bufferCreated < logBufferItem.Value.Created)
-                        {
-                            bufferCreated = logBufferItem.Value.Created;
-                        }
-                    //}
+                    if (bufferCreated == DateTime.MinValue)
+                    {
+                        bufferCreated = logBufferItem.Value.Created;
+                    }
+                    else if (bufferCreated < logBufferItem.Value.Created)
+                    {
+                        bufferCreated = logBufferItem.Value.Created;
+                    }
                }

                return bufferCreated;
@@ -72,92 +64,61 @@ namespace YY.TechJournalExportAssistant.Core.SharedBuffer
            lock (logSettings.LockObject)
            {
                var logFileInfo = new FileInfo(position.CurrentFileData);
-                SaveLogPosition(logSettings, logFileInfo, position);
-                SaveLogs(logSettings, rowsData, logFileInfo);
+                SaveLogs(logSettings, position, rowsData, logFileInfo);
            }
        }
-
-        private void SaveLogPosition(
-            TechJournalSettings.LogSourceSettings logSettings, 
-            FileInfo logFileInfo, 
-            TechJournalPosition position)
-        {
-            LogBuffer.AddOrUpdate(
-                logSettings,
-                (settings) =>
-                {
-                    var newLogBufferItem = new LogBufferItem();
-
-                    if (logFileInfo.Directory != null)
-                        newLogBufferItem.LogPositions.TryAdd(
-                            logFileInfo.Directory.Name,
-                            position);
-
-                    return newLogBufferItem;
-                },
-                (settings, logBufferItem) =>
-                {
-                    if (logFileInfo.Directory != null)
-                        logBufferItem.LogPositions.AddOrUpdate(
-                            logFileInfo.Directory.Name,
-                            (fullDirectoryName) => position,
-                            (fullDirectoryName, oldPosition) => position);
-
-                    return logBufferItem;
-                });
-        }
-
+        
        private void SaveLogs(
            TechJournalSettings.LogSourceSettings logSettings,
+            TechJournalPosition position,
            IList<EventData> rowsData,
            FileInfo logFileInfo)
        {
-            LogBuffer.AddOrUpdate(
-                logSettings,
+            var newBufferItem = new LogBufferItem();
+            newBufferItem.LastUpdate = DateTime.Now;
+            newBufferItem.LogPosition = position;
+            newBufferItem.Created = DateTime.Now;
+            foreach (var rowData in rowsData)
+            {
+                newBufferItem.LogRows.TryAdd(new EventKey()
+                {
+                    Id = Guid.NewGuid(),
+                    File = logFileInfo
+                }, rowData);
+            }
+
+            LogBuffer.TryAdd(new LogBufferItemKey(logSettings, DateTime.Now, logFileInfo.FullName), 
+                newBufferItem);
+
+            LogPositions.AddOrUpdate(logSettings,
                (settings) =>
                {
-                    var newBufferItem = new LogBufferItem();
-
-                    newBufferItem.LastUpdate = DateTime.Now;
-                    foreach (var rowData in rowsData)
-                    {
-                        newBufferItem.LogRows.TryAdd(new EventKey()
-                        {
-                            Id = Guid.NewGuid(),
-                            File = logFileInfo
-                        }, rowData);
-                    }
-
-                    return newBufferItem;
+                    var newPositions = new ConcurrentDictionary<string, TechJournalPosition>();
+                    if (logFileInfo.Directory != null)
+                        newPositions.AddOrUpdate(logFileInfo.Directory.Name,
+                            (dirName) => position, 
+                            (dirName, oldPosition) => position);
+                    return newPositions;
                },
                (settings, logBufferItem) =>
                {
-                    DateTime operationDate = DateTime.Now;
-                    if (logBufferItem.Created == DateTime.MinValue)
-                        logBufferItem.Created = operationDate;
-                    logBufferItem.LastUpdate = operationDate;
-                    foreach (var rowData in rowsData)
-                    {
-                        logBufferItem.LogRows.TryAdd(new EventKey()
-                        {
-                            Id = Guid.NewGuid(),
-                            File = logFileInfo
-                        }, rowData);
-                    }
-
+                    if (logFileInfo.Directory != null)
+                        logBufferItem.AddOrUpdate(logFileInfo.Directory.Name,
+                            (dirName) => position,
+                            (dirName, oldPosition) => position);
                    return logBufferItem;
                });
        }
-
+        
        public TechJournalPosition GetLastPosition(
            TechJournalSettings.LogSourceSettings logSettings, 
            TechJournalLogBase techJournalLog,
            string directoryName)
        {
            TechJournalPosition position = null;
-            if (LogBuffer.TryGetValue(logSettings, out LogBufferItem bufferItem))
+            if (LogPositions.TryGetValue(logSettings, out ConcurrentDictionary<string, TechJournalPosition> settingPositions))
            {
-                bufferItem.LogPositions.TryGetValue(directoryName, out position);
+                settingPositions.TryGetValue(directoryName, out position);
            }

            return position;
@@ -1,6 +1,7 @@
 using System;
 using System.Collections.Concurrent;
 using System.Collections.Generic;
+using System.IO;
 using System.Linq;
 using System.Threading;
 using System.Threading.Tasks;
@@ -29,12 +30,32 @@ namespace YY.TechJournalExportAssistant.Core.SharedBuffer

            foreach (var logSourceSettings in _settings.LogSources)
            {
-                _logBuffers.LogBuffer.TryAdd(logSourceSettings, new LogBufferItem());
-            }
-            foreach (var logBufferItem in _logBuffers.LogBuffer)
-            {
-                var logPositions = techJournalTargetBuilder.GetCurrentLogPositions(settings, logBufferItem);
-                logBufferItem.Value.LogPositions = new ConcurrentDictionary<string, TechJournalPosition>(logPositions);
+                _logBuffers.LogPositions.TryAdd(logSourceSettings,
+                    new ConcurrentDictionary<string, TechJournalPosition>());
+
+                var logPositions = techJournalTargetBuilder.GetCurrentLogPositions(settings, logSourceSettings.TechJournalLog);
+                foreach (var logPosition in logPositions)
+                {
+                    FileInfo logFileInfo = new FileInfo(logPosition.Value.CurrentFileData);
+                    _logBuffers.LogPositions.AddOrUpdate(logSourceSettings,
+                        (settingsKey) =>
+                        {
+                            var newPositions = new ConcurrentDictionary<string, TechJournalPosition>();
+                            if (logFileInfo.Directory != null)
+                                newPositions.AddOrUpdate(logFileInfo.Directory.Name,
+                                    (dirName) => logPosition.Value,
+                                    (dirName, oldPosition) => logPosition.Value);
+                            return newPositions;
+                        },
+                        (settingsKey, logBufferItemOld) =>
+                        {
+                            if (logFileInfo.Directory != null)
+                                logBufferItemOld.AddOrUpdate(logFileInfo.Directory.Name,
+                                    (dirName) => logPosition.Value,
+                                    (dirName, oldPosition) => logPosition.Value);
+                            return logBufferItemOld;
+                        });
+                }
            }
        }

@@ -143,7 +164,7 @@ namespace YY.TechJournalExportAssistant.Core.SharedBuffer
                }
                catch (Exception e)
                {
-                    RaiseOnError(settings, new OnErrorExportSharedBufferEventArgs(e));
+                    RaiseOnError(new OnErrorExportSharedBufferEventArgs(e));
                    await Task.Delay(60000, cancellationToken);
                }
            }
@@ -176,50 +197,30 @@ namespace YY.TechJournalExportAssistant.Core.SharedBuffer
                {
                    if (cancellationToken.IsCancellationRequested)
                        break;
-                    TechJournalSettings.LogSourceSettings lastSettings = null;

                    try
                    {
-                        foreach (var logBufferItem in _logBuffers.LogBuffer)
+                        var itemsToUpload = _logBuffers.LogBuffer
+                            .Select(i => i.Key)
+                            .OrderBy(i => i.Period)
+                            .ToList();
+
+                        var dataToUpload = _logBuffers.LogBuffer
+                            .Where(i => itemsToUpload.Contains(i.Key))
+                            .ToDictionary(k => k.Key, v => v.Value);
+
+                        OnSend(new OnSendLogFromSharedBufferEventArgs(dataToUpload));
+
+                        _techJournalTargetBuilder.SaveRowsData(_settings, dataToUpload);
+
+                        foreach (var itemToUpload in itemsToUpload)
                        {
-                            lastSettings = logBufferItem.Key;
-                            if (cancellationToken.IsCancellationRequested)
-                                break;
-                            if (logBufferItem.Value.LogRows.Count == 0)
-                                continue;
-
-                            lock (logBufferItem.Key.LockObject)
-                            {
-                                var eventsByFile = logBufferItem.Value.LogRows
-                                    .Select(b => new
-                                    {
-                                        FileName = b.Key.File.FullName,
-                                        EventData = b.Value
-                                    })
-                                    .GroupBy(g => g.FileName)
-                                    .ToDictionary(
-                                        g => g.Key,
-                                        g => g.Select(e => e.EventData).ToList());
-
-                                OnSend(logBufferItem.Key, new OnSendLogFromSharedBufferEventArgs(
-                                    logBufferItem.Key, eventsByFile, logBufferItem.Value.LogPositions));
-
-                                ITechJournalOnTarget target = _techJournalTargetBuilder.CreateTarget(_settings, logBufferItem);
-                                
-                                target.Save(eventsByFile);
-                                eventsByFile.Clear();
-                                logBufferItem.Value.LogRows.Clear();
-
-                                target.SaveLogPositions(logBufferItem.Value.LogPositions.Select(l => l.Value).ToList());
-                                
-                                logBufferItem.Value.Created = DateTime.MinValue;
-                                logBufferItem.Value.LastUpdate = DateTime.MinValue;
-                            }
+                            _logBuffers.LogBuffer.TryRemove(itemToUpload, out _);
                        }
                    }
                    catch (Exception e)
                    {
-                        RaiseOnError(lastSettings, new OnErrorExportSharedBufferEventArgs(e));
+                        RaiseOnError(new OnErrorExportSharedBufferEventArgs(e));
                        await Task.Delay(1000, cancellationToken);
                    }
                }
@@ -238,26 +239,22 @@ namespace YY.TechJournalExportAssistant.Core.SharedBuffer

        #region Events

-        public delegate void OnSendLogFromSharedBufferEventArgsHandler(TechJournalSettings.LogSourceSettings settings, OnSendLogFromSharedBufferEventArgs args);
-        public delegate void OnErrorExportSharedBufferEventArgsHandler(TechJournalSettings.LogSourceSettings settings, OnErrorExportSharedBufferEventArgs args);
+        public delegate void OnSendLogFromSharedBufferEventArgsHandler(OnSendLogFromSharedBufferEventArgs args);
+        public delegate void OnErrorExportSharedBufferEventArgsHandler(OnErrorExportSharedBufferEventArgs args);
        public event OnSendLogFromSharedBufferEventArgsHandler OnSendLogEvent;
        public event OnErrorExportSharedBufferEventArgsHandler OnErrorEvent;

-        protected void OnSend(
-            TechJournalSettings.LogSourceSettings settings,
-            OnSendLogFromSharedBufferEventArgs args)
+        protected void OnSend(OnSendLogFromSharedBufferEventArgs args)
        {
-            OnSendLogEvent?.Invoke(settings, args);
+            OnSendLogEvent?.Invoke(args);
        }
-        protected void RaiseOnError(
-            TechJournalSettings.LogSourceSettings settings,
-            OnErrorExportSharedBufferEventArgs args)
+        protected void RaiseOnError(OnErrorExportSharedBufferEventArgs args)
        {
-            OnErrorEvent?.Invoke(settings, args);
+            OnErrorEvent?.Invoke(args);
        }
        private void OnErrorExportDataToBuffer(OnErrorExportDataEventArgs e)
        {
-            RaiseOnError(null, new OnErrorExportSharedBufferEventArgs(e.Exception));
+            RaiseOnError(new OnErrorExportSharedBufferEventArgs(e.Exception));
        }

        #endregion
@@ -1,6 +1,5 @@
 using System;
 using System.Collections.Generic;
-using YY.TechJournalExportAssistant.Core.SharedBuffer;
 using YY.TechJournalReaderAssistant;
 using YY.TechJournalReaderAssistant.Models;

@@ -50,7 +49,7 @@ namespace YY.TechJournalExportAssistant.Core
            throw new NotImplementedException();
        }

-        public virtual IDictionary<string, TechJournalPosition> GetCurrentLogPositions(TechJournalSettings settings, KeyValuePair<TechJournalSettings.LogSourceSettings, LogBufferItem> logBufferItem)
+        public virtual IDictionary<string, TechJournalPosition> GetCurrentLogPositions(TechJournalSettings settings)
        {
            throw new NotImplementedException();
        }
@@ -81,8 +81,25 @@ namespace YY.TechJournalExportAssistant.Core

        public class LogSourceSettings
        {
+            private TechJournalLogBase _techJournalLog;
            public string Name { get; }
            public string Description { get; }
+            public TechJournalLogBase TechJournalLog
+            {
+                get
+                {
+                    if (_techJournalLog == null)
+                    {
+                        _techJournalLog = new TechJournalLogBase()
+                        {
+                            Name = Name,
+                            Description = Description
+                        };
+                    }
+
+                    return _techJournalLog;
+                }
+            }
            public string SourcePath { get; }
            public int Portion { get; }
            public TimeZoneInfo TimeZone { get; }