chore: update logs + add parallel.for when putting updating tiles to S3

MergerCli/Dockerfile

@@ -1,10 +1,11 @@
 #See https://aka.ms/containerfastmode to understand how Visual Studio uses this Dockerfile to build your images for faster debugging.
 
 FROM mcr.microsoft.com/dotnet/runtime:6.0 AS base
+ARG VERSION="1.0.0.0"
 WORKDIR /app
 
 FROM mcr.microsoft.com/dotnet/sdk:6.0 AS build
-ARG VERSION="9.9.9.9"
+ARG VERSION="1.0.0.0"
 WORKDIR /src
 COPY ["MergerCli/MergerCli.csproj", "MergerCli/"]
 COPY ["MergerLogic/MergerLogic.csproj", "MergerLogic/"]

MergerCli/Process.cs

@@ -125,7 +125,8 @@ private void ProcessBatch(TileFormat targetFormat, IData baseData, List<Tile> ne
                 }
             }
 
-            baseData.UpdateTiles(tiles);
+            var t = Task.Run(() => baseData.UpdateTiles(tiles));
+            t.Wait();
 
             Interlocked.Add(ref tileProgressCount, tiles.Count);
             this._logger.LogInformation($"[{MethodBase.GetCurrentMethod().Name}] Tile Count: {tileProgressCount} / {totalTileCount}");

MergerCli/appsettings.json

@@ -1,6 +1,7 @@
 {
   "GENERAL": {
     "validate": false,
+    "servicePointManagerDefaultConnectionLimit": 100,
     "resumeOutputFolder": "./",
     "uploadOnly": false,
     "parallel": {

MergerLogic/Clients/S3Client.cs

@@ -31,12 +31,10 @@ public S3Client(IAmazonS3 client, IPathUtils pathUtils, IGeoUtils geoUtils, IIma
             string methodName = MethodBase.GetCurrentMethod().Name;
             try
             {
-                this._logger.LogDebug($"[{methodName}] start, key {key}");
+                this._logger.LogDebug($"[{methodName}] start, BucketName: {this._bucket}, Key: {key}");
                 var request = new GetObjectRequest() { BucketName = this._bucket, Key = key };
-                this._logger.LogDebug($"[{methodName}] start GetObjectAsync, BucketName: {request.BucketName}, Key: {request.Key}");
                 var getObjectTask = this._client.GetObjectAsync(request);
                 GetObjectResponse res = getObjectTask.Result;
-                this._logger.LogDebug($"[{methodName}] requested key {key} recieved");
 
                 byte[] image;
                 using (MemoryStream ms = new MemoryStream())
@@ -48,12 +46,12 @@ public S3Client(IAmazonS3 client, IPathUtils pathUtils, IGeoUtils geoUtils, IIma
 
                     image = ms.ToArray();
                 }
-                this._logger.LogDebug($"[{methodName}] end, key {key}");
+                this._logger.LogDebug($"[{methodName}] end, BucketName: {this._bucket}, Key: {key}");
                 return image;
             }
             catch (AggregateException e)
             {
-                this._logger.LogDebug($"[{methodName}] exception, Message: {e.Message}");
+                this._logger.LogError($"[{methodName}] exception, Message: {e.Message}, returned null");
                 return null;
             }
         }
@@ -65,6 +63,7 @@ public S3Client(IAmazonS3 client, IPathUtils pathUtils, IGeoUtils geoUtils, IIma
             var key = this.GetTileKey(z, x, y);
             if (key == null)
             {
+                this._logger.LogDebug($"[{methodName}] end z: {z}, x: {x}, y: {y} - no tile was found");
                 return null;
             }
 
@@ -99,7 +98,7 @@ public override bool TileExists(int z, int x, int y)
         public void UpdateTile(Tile tile)
         {
             string methodName = MethodBase.GetCurrentMethod().Name;
-            this._logger.LogDebug($"[{methodName}] start {tile.ToString()}");
+            this._logger.LogDebug($"[{methodName}] start {tile.ToString()} to BucketName: {this._bucket}");
             string key = this._pathUtils.GetTilePath(this.path, tile, true);
 
             var request = new PutObjectRequest()
@@ -111,23 +110,21 @@ public void UpdateTile(Tile tile)
             using (var ms = new MemoryStream(buffer))
             {
                 request.InputStream = ms;
-                this._logger.LogDebug($"[{methodName}] start PutObjectAsync BucketName: {request.BucketName}, Key: {request.Key}");
                 var task = this._client.PutObjectAsync(request);
                 var res = task.Result;
             }
-            this._logger.LogDebug($"[{methodName}] end {tile.ToString()}");
+            this._logger.LogDebug($"[{methodName}] end {tile.ToString()} to BucketName: {this._bucket}");
         }
 
         private string? GetTileKey(int z, int x, int y)
         {
             string methodName = MethodBase.GetCurrentMethod().Name;
-            this._logger.LogDebug($"[{methodName}] start z: {z}, x: {x}, y: {y}");
+            this._logger.LogDebug($"[{methodName}] start z: {z}, x: {x}, y: {y} from BucketName: {this._bucket}");
             string keyPrefix = this._pathUtils.GetTilePathWithoutExtension(this.path, z, x, y, true);
             var listRequests = new ListObjectsV2Request { BucketName = this._bucket, Prefix = keyPrefix, MaxKeys = 1 };
-            this._logger.LogDebug($"[{methodName}] start ListObjectsV2Async BucketName: {listRequests.BucketName}, Prefix: {listRequests.Prefix}");
             var listObjectsTask = this._client.ListObjectsV2Async(listRequests);
             string? result = listObjectsTask.Result.S3Objects.FirstOrDefault()?.Key;
-            this._logger.LogDebug($"[{methodName}] end z: {z}, x: {x}, y: {y}");
+            this._logger.LogDebug($"[{methodName}] end z: {z}, x: {x}, y: {y} from BucketName: {this._bucket}");
             return result;
         }
     }

MergerLogic/DataTypes/Data.cs

@@ -79,7 +79,7 @@ protected Data(IServiceProvider container, DataType type, string path, int batch
             this._container = container;
             var loggerFactory = container.GetRequiredService<ILoggerFactory>();
             this._logger = loggerFactory.CreateLogger(this.GetType());
-            this._logger.LogInformation($"[{methodName}] Ctor started");
+            this._logger.LogInformation($"[{methodName}] Data Ctor started");
             this.Type = type;
             this.Path = path;
             this.BatchSize = batchSize;
@@ -139,6 +139,7 @@ protected Data(IServiceProvider container, DataType type, string path, int batch
 
             this._logger.LogInformation($"[{methodName}] Initialize Started");
             this.Initialize();
+            this._logger.LogInformation($"[{methodName}] Initialize ended");
 
             this._logger.LogInformation($"[{methodName}] Checking if exists, {this.Type}: {this.Path}");
             bool exists = this.Exists();
@@ -152,9 +153,8 @@ protected Data(IServiceProvider container, DataType type, string path, int batch
                     throw new Exception($"{this.Type} source {path} does not exist.");
                 }
             }
-            this._logger.LogInformation($"[{methodName}] Validate Started");
             this.Validate();
-            this._logger.LogInformation($"[{methodName}] Ctor Ended");
+            this._logger.LogInformation($"[{methodName}] Data Ctor Ended");
         }
 
         protected virtual void Initialize()
@@ -256,18 +256,18 @@ public bool TileExists(Coord coord)
 
         public Tile? GetCorrespondingTile(Coord coords, bool upscale)
         {
-            this._logger.LogDebug($"[{MethodBase.GetCurrentMethod().Name}] start for coord: {coords.ToString()}, upscale: {upscale}");
+            this._logger.LogDebug($"[{MethodBase.GetCurrentMethod().Name}] start for coord: z: {coords.Z}, x: {coords.X}, y: {coords.Y}, upscale: {upscale}");
             Tile? correspondingTile = this.GetTile(coords.Z, coords.X, coords.Y);
 
             if (upscale && correspondingTile == null)
             {
                 correspondingTile = this.GetLastExistingTile(coords);
             }
-            this._logger.LogDebug($"[{MethodBase.GetCurrentMethod().Name}] end for coord: {coords.ToString()}, upscale: {upscale}");
+            this._logger.LogDebug($"[{MethodBase.GetCurrentMethod().Name}] end for coord: z: {coords.Z}, x: {coords.X}, y: {coords.Y}, upscale: {upscale}");
             return correspondingTile;
         }
 
-        public void UpdateTiles(IEnumerable<Tile> tiles)
+        public async Task UpdateTiles(IEnumerable<Tile> tiles)
         {
             this._logger.LogDebug($"[{MethodBase.GetCurrentMethod().Name}] update tiles started");
             var targetTiles = tiles.Select(tile =>
@@ -276,11 +276,11 @@ public void UpdateTiles(IEnumerable<Tile> tiles)
                 Tile? targetTile = this.FromCurrentGridTile(convertedTile);
                 return targetTile;
             }).Where(tile => tile is not null);
-            this.InternalUpdateTiles(targetTiles);
+            await this.InternalUpdateTilesAsync(targetTiles);
             this._logger.LogDebug($"[{MethodBase.GetCurrentMethod().Name}] update tiles ended");
         }
 
-        protected abstract void InternalUpdateTiles(IEnumerable<Tile> targetTiles);
+        protected abstract Task InternalUpdateTilesAsync(IEnumerable<Tile> targetTiles);
 
         public virtual void Wrapup()
         {

MergerLogic/DataTypes/Fs.cs

@@ -174,7 +174,7 @@ public override long TileCount()
             return count;
         }
 
-        protected override void InternalUpdateTiles(IEnumerable<Tile> targetTiles)
+        protected override Task InternalUpdateTilesAsync(IEnumerable<Tile> targetTiles)
         {
             foreach (Tile tile in targetTiles)
             {
@@ -190,6 +190,7 @@ protected override void InternalUpdateTiles(IEnumerable<Tile> targetTiles)
                     }
                 }
             }
+            return Task.CompletedTask;
         }
     }
 }

MergerLogic/DataTypes/Gpkg.cs

@@ -189,11 +189,12 @@ public override long TileCount()
             return result;
         }
 
-        protected override void InternalUpdateTiles(IEnumerable<Tile> targetTiles)
+        protected override Task InternalUpdateTilesAsync(IEnumerable<Tile> targetTiles)
         {
             this._logger.LogDebug($"[{MethodBase.GetCurrentMethod().Name}] started");
             this.Utils.InsertTiles(targetTiles);
             this._logger.LogDebug($"[{MethodBase.GetCurrentMethod().Name}] ended");
+            return Task.CompletedTask;
         }
     }
 }

MergerLogic/DataTypes/HttpDataSource.cs

@@ -84,7 +84,7 @@ protected IEnumerable<Tile> GetTiles()
             }
         }
 
-        protected override void InternalUpdateTiles(IEnumerable<Tile> targetTiles)
+        protected override Task InternalUpdateTilesAsync(IEnumerable<Tile> targetTiles)
         {
             throw new NotImplementedException();
         }

MergerLogic/DataTypes/IData.cs

@@ -16,7 +16,7 @@ public interface IData
         long TileCount();
         bool TileExists(Coord coord);
         bool TileExists(Tile tile);
-        void UpdateTiles(IEnumerable<Tile> tiles);
+        Task UpdateTiles(IEnumerable<Tile> tiles);
         void Wrapup();
     }
 }

MergerLogic/DataTypes/S3.cs

@@ -67,17 +67,17 @@ public override void Reset()
 
         private List<int> GetZoomLevels()
         {
-            this._logger.LogDebug($"[{MethodBase.GetCurrentMethod().Name}] start");
-            List<int> zoomLevels = new List<int>();
+            List<int> zoomLevels = new List<int>(Data<IS3Client>.MaxZoomRead);
 
             for (int zoomLevel = 0; zoomLevel < Data<IS3Client>.MaxZoomRead; zoomLevel++)
             {
+                this._logger.LogDebug($"[{MethodBase.GetCurrentMethod().Name}] Check if zoom {zoomLevel} exists");
+                // TODO: need to check if can be updated dynamically
                 if (this.FolderExists($"{zoomLevel}/"))
                 {
                     zoomLevels.Add(zoomLevel);
                 }
             }
-            this._logger.LogDebug($"[{MethodBase.GetCurrentMethod().Name}] ended");
             return zoomLevels;
         }
 
@@ -158,15 +158,16 @@ private bool FolderExists(string directory)
             };
             var task = this._client.ListObjectsV2Async(listRequests);
             var response = task.Result;
-            this._logger.LogDebug($"[{MethodBase.GetCurrentMethod().Name}] end");
-            return response.KeyCount > 0;
+            bool isExists = response.KeyCount > 0;
+            this._logger.LogDebug($"[{MethodBase.GetCurrentMethod().Name}] directory: {directory}, isExists={isExists}");
+            return isExists;
         }
 
         public override bool Exists()
         {
-            this._logger.LogInformation($"[{MethodBase.GetCurrentMethod().Name}] bucket: {this._bucket}, path: {this.Path}");
+            this._logger.LogInformation($"[{MethodBase.GetCurrentMethod().Name}] Checking if exists, bucket: {this._bucket}, path: {this.Path}");
             bool exists = FolderExists("");
-            this._logger.LogInformation($"[{MethodBase.GetCurrentMethod().Name}] ended");
+            this._logger.LogInformation($"[{MethodBase.GetCurrentMethod().Name}] exists={exists} ended");
             return exists;
         }
 
@@ -196,14 +197,14 @@ public override long TileCount()
             return tileCount;
         }
 
-        protected override void InternalUpdateTiles(IEnumerable<Tile> targetTiles)
+        protected override async Task InternalUpdateTilesAsync(IEnumerable<Tile> targetTiles)
         {
-            this._logger.LogDebug($"[{MethodBase.GetCurrentMethod().Name}] start");
-            foreach (var tile in targetTiles)
+            this._logger.LogDebug($"[{MethodBase.GetCurrentMethod().Name}] start upload tiles to S3");
+            await Parallel.ForEachAsync(targetTiles, new ParallelOptions { MaxDegreeOfParallelism = -1 }, async (tile, cancellationToken) =>
             {
-                this.Utils.UpdateTile(tile);
-            }
-            this._logger.LogDebug($"[{MethodBase.GetCurrentMethod().Name}] end");
+                await Task.Run(() => this.Utils.UpdateTile(tile), cancellationToken);
+            });
+            this._logger.LogDebug($"[{MethodBase.GetCurrentMethod().Name}] end upload tiles to S3");
         }
     }
 }

MergerLogic/Monitoring/OpenTelemetryFormattedConsoleExporter.cs

@@ -1,13 +1,16 @@
 using OpenTelemetry;
 using OpenTelemetry.Exporter;
 using OpenTelemetry.Logs;
+using System;
+using System.Net;
 
 namespace MergerLogic.Monitoring
 {
     public class OpenTelemetryFormattedConsoleExporter : ConsoleExporter<LogRecord>
     {
         private const string SERVICE_NAME_ATTRIBUTE = "service.name";
         private const string SERVICE_VERSION_ATTRIBUTE = "service.version";
+        private const string SERVICE_HOST_NAME_ATTRIBUTE = "service.host.name";
 
 
         public OpenTelemetryFormattedConsoleExporter(ConsoleExporterOptions options) : base(options)
@@ -29,9 +32,13 @@ private string MCTextFormat(LogRecord record)
             var resource = this.ParseResource();
             var serviceName = this.GetResourceAttribute(resource, SERVICE_NAME_ATTRIBUTE, "unknown_service");
             var serviceVersion = this.GetResourceAttribute(resource, SERVICE_VERSION_ATTRIBUTE, "unknown_version");
+            if (!resource.ContainsKey(SERVICE_HOST_NAME_ATTRIBUTE))
+            {
+                resource.Add(SERVICE_HOST_NAME_ATTRIBUTE, Dns.GetHostName());
+            }
+            var serviceHostName = this.GetResourceAttribute(resource, SERVICE_HOST_NAME_ATTRIBUTE, "unknown_host_name");
             var exception = record.Exception != null ? $" [{record.Exception}]" : string.Empty;
-
-            return $"[{this.FormatTime(record.Timestamp)}] [{record.LogLevel}] [{serviceName}] [{serviceVersion}] [{record.CategoryName}] {record.State}{exception}";
+            return $"[{this.FormatTime(record.Timestamp)}] [{record.LogLevel}] [{serviceName}] [{serviceHostName}] [{serviceVersion}] [{Environment.CurrentManagedThreadId}] [{record.CategoryName}] {record.State}{exception}";
         }
 
         private string FormatTime(DateTime time)

MergerLogicUnitTests/DataTypes/FSTest.cs

@@ -11,6 +11,7 @@
 using System.IO;
 using System.IO.Abstractions;
 using System.Linq;
+using System.Threading.Tasks;
 
 namespace MergerLogicUnitTests.DataTypes
 {
@@ -481,7 +482,8 @@ public void UpdateTiles(bool isBase, bool isOneXOne, GridOrigin origin)
                     fileMock.InSequence(seq).Setup(f => f.OpenWrite()).Returns(fileStream);
                 }
 
-                fsSource.UpdateTiles(testTiles);
+                var t = Task.Run(() => fsSource.UpdateTiles(testTiles));
+                t.Wait();
 
                 CollectionAssert.AreEqual(new byte[] { 0 }, fileStreams[0].ToArray());
                 if (isOneXOne)

MergerLogicUnitTests/DataTypes/GpkgTest.cs

@@ -9,6 +9,7 @@
 using System;
 using System.Collections.Generic;
 using System.Linq;
+using System.Threading.Tasks;
 
 namespace MergerLogicUnitTests.DataTypes
 {
@@ -455,7 +456,8 @@ public void UpdateTiles(bool isBase, bool isOneXOne, GridOrigin origin)
                 new Tile(1, 2, 3, new byte[] { }), new Tile(7, 7, 7, new byte[] { }),
                 new Tile(2, 2, 3, new byte[] { })
             };
-            gpkg.UpdateTiles(testTiles);
+            var t = Task.Run(() => gpkg.UpdateTiles(testTiles));
+            t.Wait();
 
             var expectedTiles = isOneXOne ? new Tile[] { testTiles[0], testTiles[2] } : testTiles;
             var tileComparer = EqualityComparerFactory.Create<Tile>((tile1, tile2) =>