feat!: modify task completed to notify job track but discrete overseer (MAPCO-4798)

MergerLogic/Clients/HeartbeatClient.cs

@@ -1,7 +1,10 @@
 using MergerLogic.Utils;
 using Microsoft.Extensions.Logging;
+using Newtonsoft.Json;
 using System.Reflection;
+using System.Text;
 using System.Timers;
+using System.Net.Http.Json;
 
 namespace MergerLogic.Clients
 {
@@ -27,14 +30,16 @@ public HeartbeatClient(ILogger<GpkgClient> logger, IConfigurationManager configu
             this._timer.Elapsed += this.Send;
         }
 
-        ~HeartbeatClient() {
+        ~HeartbeatClient()
+        {
             this._timer.Elapsed -= this.Send;
             this._timer.Dispose();
         }
 
         public void Start(string taskId)
         {
-            if (this._timer.Enabled) {
+            if (this._timer.Enabled)
+            {
                 this.Stop();
             }
             this._logger.LogInformation($"[{MethodBase.GetCurrentMethod().Name}] Starting heartbeat for task={taskId}");
@@ -50,7 +55,23 @@ public void Stop()
             }
             this._logger.LogInformation($"[{MethodBase.GetCurrentMethod().Name}] Stops heartbeats for taskId={this._taskId}");
             this._timer.Stop();
-            this._taskId = null;
+            try
+            {
+                string relativeUri = $"heartbeat/remove";
+                string url = new Uri(new Uri(this._baseUrl), relativeUri).ToString();
+                var content = JsonContent.Create(new[] { this._taskId });
+                this._httpClient.PostData(url, content);
+            }
+            catch (Exception e)
+            {
+                string message = $"[{MethodBase.GetCurrentMethod().Name}] Could not delete heartbeat for task={this._taskId}, {e.Message}";
+                this._logger.LogError(message);
+                throw new Exception(message, e);
+            }
+            finally
+            {
+                this._taskId = null;
+            }
         }
 
         public void Send(object? sender, ElapsedEventArgs elapsedEventArgs)
@@ -66,7 +87,6 @@ public void Send(object? sender, ElapsedEventArgs elapsedEventArgs)
                 this._logger.LogError($"[{MethodBase.GetCurrentMethod().Name}] Could not send heartbeat for task={this._taskId}, {e.Message}");
                 throw;
             }
-
         }
     }
-}
+}

MergerService/Models/Jobs/JobMergeMetadata.cs

@@ -10,19 +10,19 @@ public class JobMergeMetadata
         [JsonInclude] public string[] FileNames { get; }
         [JsonInclude] public string OriginDirectory { get; }
         [JsonInclude] public string LayerRelativePath { get; }
-        [JsonInclude] public string? ManagerCallbackUrl { get; }
+        [JsonInclude] public AdditionalParams? AdditionalParams { get; }
 
         [System.Text.Json.Serialization.JsonIgnore]
         private JsonSerializerSettings _jsonSerializerSettings;
 
-        public JobMergeMetadata(JobMetadata metadata, string[] fileNames,string originDirectory, string layerRelativePath,
-            string managerCallbackUrl)
+        public JobMergeMetadata(JobMetadata metadata, string[] fileNames, string originDirectory, string layerRelativePath,
+            AdditionalParams additionalParams)
         {
             this.Metadata = metadata;
             this.FileNames = fileNames;
             this.OriginDirectory = originDirectory;
             this.LayerRelativePath = layerRelativePath;
-            this.ManagerCallbackUrl = managerCallbackUrl;
+            this.AdditionalParams = additionalParams;
 
             this._jsonSerializerSettings = new JsonSerializerSettings();
             this._jsonSerializerSettings.Converters.Add(new StringEnumConverter());
@@ -38,5 +38,5 @@ public override string ToString()
             return JsonConvert.SerializeObject(this, this._jsonSerializerSettings)!;
         }
     }
-    
+
 }

MergerService/Models/Jobs/JobParamersAdditiomalParams.cs

@@ -0,0 +1,23 @@
+using Newtonsoft.Json;
+using Newtonsoft.Json.Converters;
+using System.Text.Json.Serialization;
+
+namespace MergerService.Models.Jobs
+{
+     public class AdditionalParams
+    {
+        [JsonInclude] public string? JobTrackerServiceURL { get; }
+
+        [System.Text.Json.Serialization.JsonIgnore]
+        private JsonSerializerSettings _jsonSerializerSettings;
+
+        public AdditionalParams(string jobTrackerServiceURL)
+        {
+            this.JobTrackerServiceURL = jobTrackerServiceURL;
+
+            this._jsonSerializerSettings = new JsonSerializerSettings();
+            this._jsonSerializerSettings.Converters.Add(new StringEnumConverter());
+        }
+    }
+
+}

MergerService/Runners/TaskRunner.cs

@@ -87,7 +87,7 @@ public bool RunTask(MergeTask? task)
             }
 
             this._logger.LogInformation($"[{methodName}] Run Task: jobId {task.JobId}, taskId {task.Id}");
-            string? managerCallbackUrl = this._jobUtils.GetJob(task.JobId)?.Parameters.ManagerCallbackUrl;
+            string? managerCallbackUrl = this._jobUtils.GetJob(task.JobId)?.Parameters.AdditionalParams?.JobTrackerServiceURL;
             string log = managerCallbackUrl == null ? "managerCallbackUrl not provided as job parameter" : $"managerCallback url: {managerCallbackUrl}";
             this._logger.LogDebug($"[{methodName}]{log}");
 

MergerService/Utils/TaskUtils.cs

@@ -67,13 +67,13 @@ public TaskUtils(MergerLogic.Utils.IConfigurationManager configuration, IHttpReq
             }
         }
 
-        private void NotifyOnStatusChange(string jobId, string taskId, string managerCallbackUrl)
+        private void NotifyOnStatusChange(string taskId, string managerCallbackUrl)
         {
             using (this._activitySource.StartActivity("notify Manager on task completion"))
             {
                 // Notify Manager on task completion
-                this._logger.LogInformation($"[{MethodBase.GetCurrentMethod().Name}] Notifying Manager on completion, job: {jobId}, task: {taskId}");
-                string relativeUri = $"jobs/{jobId}/{taskId}/completed";
+                this._logger.LogInformation($"[{MethodBase.GetCurrentMethod().Name}] Notifying Manager on completion, task: {taskId}");
+                string relativeUri = $"tasks/{taskId}/notify";
                 string url = new Uri(new Uri(managerCallbackUrl), relativeUri).ToString();
                 _ = this._httpClient.PostData(url, null);
             }
@@ -116,7 +116,7 @@ public void UpdateCompletion(string jobId, string taskId, string? managerCallbac
             if (managerCallbackUrl is not null)
             {
                 // Update Manager on task completion
-                NotifyOnStatusChange(jobId, taskId, managerCallbackUrl);
+                NotifyOnStatusChange(taskId, managerCallbackUrl);
             }
         }
 
@@ -160,7 +160,7 @@ private void UpdateFailed(string jobId, string taskId, int attempts, string reas
             if (managerCallbackUrl is not null)
             {
                 // Notify Manager on task failure
-                NotifyOnStatusChange(jobId, taskId, managerCallbackUrl);
+                NotifyOnStatusChange(taskId, managerCallbackUrl);
             }
         }
     }

helm/Chart.yaml

@@ -2,5 +2,5 @@ apiVersion: v2
 name: gpkg-merger
 description: A Helm chart for gpkg-merger service # refers to MergerService
 type: application
-version: 1.4.1
-appVersion: 1.4.1
+version: 2.0.0
+appVersion: 2.0.0