Added support for tasks with an external id

Author: dei79

dotNet/CoreHelpers.Extensions.Logging.Tasks/TaskLoggerExtension.cs

@@ -24,10 +24,7 @@ public static ILoggingBuilder AddTaskLogger(this ILoggingBuilder builder)
 
         public static ITaskLoggerScope? BeginNewTaskScope(this ILogger logger, string taskType, string taskSource, string taskWorker, string metaDataString, TimeSpan? cacheTimeSpan = null)
             => BeginTypedTaskScope(logger, new TaskLoggerState() { TaskId = string.Empty, TaskType = taskType, TaskSource = taskSource, TaskWorker = taskWorker, IsTaskAnnounced = false, MetaData = metaDataString, CacheTimeSpan = cacheTimeSpan ?? TimeSpan.FromSeconds(30) });
-
-        public static ITaskLoggerScope? BeginNewTaskScopeWithExternalId(this ILogger logger, string taskId, string taskType, string taskSource, string taskWorker, string metaDataString, TimeSpan? cacheTimeSpan = null)
-            => BeginTypedTaskScope(logger, new TaskLoggerState() { TaskId = taskId, TaskType = taskType, TaskSource = taskSource, TaskWorker = taskWorker, IsTaskAnnounced = false, MetaData = metaDataString, CacheTimeSpan = cacheTimeSpan ?? TimeSpan.FromSeconds(30) });
-
+        
         private static ITaskLoggerScope? BeginTypedTaskScope(ILogger logger, TaskLoggerState taskLoggerState)
         {
             var innerDisposable = logger.BeginScope<TaskLoggerState>(taskLoggerState);

dotNet/CoreHelpers.TaskLogging.Abstractions/ITaskLoggerFactory.cs

@@ -81,6 +81,22 @@ public interface ITaskLoggerFactory
         /// <param name="taskWorker"></param>
         /// <returns></returns>
         Task UpdateTaskWorker(string taskId, string taskWorker);
+        
+        /// <summary>
+        /// Resolve an external id into the task id or returns null if not found
+        /// </summary>
+        /// <param name="externalTaskId"></param>
+        /// <returns></returns>
+        Task<string?> LookupTaskIdByExternalId(string externalTaskId);
+        
+        
+        /// <summary>
+        /// Registeres an external id for a given task id
+        /// </summary>
+        /// <param name="taskId"></param>
+        /// <param name="externalTaskId"></param>
+        /// <returns></returns>
+        Task RegisterExternlIdForTask(string taskId, string externalTaskId);
     }
 }
 

dotNet/CoreHelpers.TaskLogging.Sample/Program.cs

@@ -27,7 +27,8 @@
 using IHost host = builder.Build();
 
 // get the worker
-var worker = host.Services.GetService<WorkerParallelTasks>();
+//var worker = host.Services.GetService<WorkerParallelTasks>();
+var worker = host.Services.GetService<Worker>();
 if (worker == null)
     throw new NullReferenceException();
 

dotNet/CoreHelpers.TaskLogging.Sample/Worker.cs

@@ -43,9 +43,15 @@ public async Task Process()
             };
 
             var taskId = await _taskLoggerFactory.AnnounceTask("announcejob", "q", "w", metaData);
+			var externalId = Guid.NewGuid().ToString();
 
+			await _taskLoggerFactory.RegisterExternlIdForTask(taskId, externalId);
             using (_logger.BeginTaskScope(taskId))
             {
+	            var lookedUpTaskId = await _taskLoggerFactory.LookupTaskIdByExternalId(externalId);
+	            if (lookedUpTaskId != taskId)
+		            throw new Exception($"Task with id {lookedUpTaskId} was not found");
+	            
                 await _processors.Where(p => p is ProcessorSuccess).First().Execute();
             }
 

dotNet/CoreHelpers.TaskLogging.Sample/WorkerParallelTasks.cs

@@ -41,10 +41,7 @@ private Task SpawnTask(int iTaskNumber)
 				_logger.LogInformation($"Executing task {iTaskNumber}");
 
 				// create a task logger scope
-				using var _ =
-					iTaskNumber == 0 ?
-						_logger.BeginNewTaskScopeWithExternalId(Guid.NewGuid().ToString(), "SpawnedTask", "WorkerParallelTasks", $"TaskWorker-{iTaskNumber}", string.Empty, TimeSpan.FromSeconds(1)) :
-						_logger.BeginNewTaskScope("SpawnedTask", "WorkerParallelTasks", $"TaskWorker-{iTaskNumber}", TimeSpan.FromSeconds(1));
+				using var _ = _logger.BeginNewTaskScope("SpawnedTask", "WorkerParallelTasks", $"TaskWorker-{iTaskNumber}", TimeSpan.FromSeconds(1));
 
 				// log something in the context of the task
 				_logger.LogInformation($"T{iTaskNumber}: Started task {iTaskNumber}");

dotNet/CoreHelpers.TaskLogging/AzureStorageTableTaskLoggerFactory.cs

@@ -125,6 +125,34 @@ public async Task UpdateTaskWorker(string taskId, string taskWorker)
             await UpdateEntityInTable<AzureTableTaskEntity>(tableName, taskEntity);
         }
 
+        public async Task<string?> LookupTaskIdByExternalId(string externalTaskId)
+        {
+            // get the table name
+            var tableName = GetExternalTaskIdLookupTable();
+
+            // lookup 
+            var externalIdEnties = await QueryEntitiyFromTableByPartitionKey<AzureTableTaskEntity>(tableName, externalTaskId);
+            if (externalIdEnties.Length == 0)
+                return null;
+
+            return externalIdEnties.First().RowKey;
+        }
+
+        public async Task RegisterExternlIdForTask(string taskId, string externalTaskId)
+        {
+            // build the task entity
+            var externalIdForTaskEntity = new AzureTableTaskEntity()
+            {
+                PartitionKey = externalTaskId,
+                RowKey = taskId,
+            };
+            
+            // get the table name
+            var tableName = GetExternalTaskIdLookupTable();
+            
+            await AddEntityToTable<AzureTableTaskEntity>(tableName, externalIdForTaskEntity);
+        }
+
         public async Task<string[]> MergePendingMessagesIfNeeded(DateTimeOffset flushTime, bool force, string taskKey, string[] messages)
         {
             // check if the cache limit is exceeded
@@ -184,6 +212,9 @@ private string GetTableName(string tableName)
 
         private string GetTaskTable()
             => GetTableName("Tasks");
+        
+        private string GetExternalTaskIdLookupTable()
+            => GetTableName("TasksExternalIdLookup");
 
         private string GetRunningTaskTable()
             => $"{_environmentPrefix}TasksRunning";
@@ -218,6 +249,21 @@ private async Task UpdateEntityInTable<T>(string tableName, T entity) where T :
         private async Task DeleteEntityByKeys(string tableName, string pKey, string rowKey)
             => await ExecuteEntityToTableOperation(tableName, (TableClient tc) => tc.DeleteEntityAsync(pKey, rowKey, Azure.ETag.All));
 
+        private async Task<T[]> QueryEntitiyFromTableByPartitionKey<T>(string tableName, string partitionKey) where T : class, ITableEntity
+        {
+            var result = new List<T>();
+            
+            await ExecuteEntityToTableOperation(tableName,  async (TableClient tc) =>
+            {
+                var entities = tc.QueryAsync<T>(filter: $"PartitionKey eq '{partitionKey}'");
+                
+                await foreach (var entity in entities)
+                    result.Add(entity);
+            });
+
+            return result.ToArray();
+        }
+
         private async Task ExecuteEntityToTableOperation(string tableName, Func<TableClient, Task> operation)
         {
             // get the table client
@@ -232,7 +278,17 @@ private async Task ExecuteEntityToTableOperation(string tableName, Func<TableCli
             {
                 if (e.ErrorCode != null && e.ErrorCode.Equals("TableNotFound"))
                 {
-                    await tableClient.CreateAsync();
+                    try
+                    {
+                        await tableClient.CreateAsync();
+                        
+                    // double check patter because of race conditions    
+                    } catch(Azure.RequestFailedException e2)
+                    {
+                        if (e2.ErrorCode != null && !e2.ErrorCode.Equals("TableAlreadyExists"))
+                            throw;
+                    }
+                    
                     await operation(tableClient);
                 }
             }

dotNet/CoreHelpers.TaskLogging/AzureTableExternalTaskIdEntity.cs

@@ -0,0 +1,14 @@
+using System;
+using Azure;
+using Azure.Data.Tables;
+
+namespace CoreHelpers.TaskLogging
+{
+    public class AzureTableExternalTaskIdEntity : ITableEntity
+    {
+        public string PartitionKey { get; set; } = default!;
+        public string RowKey { get; set; } = default!;
+        public ETag ETag { get; set; } = default!;
+        public DateTimeOffset? Timestamp { get; set; } = default!;
+    }
+}