Testing enhancements (#98)

Fixes and enhancements based on the testing
conductor-sdk · Jan 2, 2024 · aaf6405 · aaf6405
1 parent ac99179
commit aaf6405
Show file tree

Hide file tree

Showing 14 changed files with 196 additions and 161 deletions.
diff --git a/.github/workflows/pull_request.yml b/.github/workflows/pull_request.yml
@@ -1,6 +1,6 @@
-name: Continuous Integration
+name: CI Build
 
-on: pull_request
+on: [push, pull_request,workflow_dispatch]
 
 jobs:
   lint:

diff --git a/.gitignore b/.gitignore
@@ -16,3 +16,4 @@ build
 **/*.jfm
 **/*.user
 **/key-*.xml
+csharp-examples/csharp-examples.sln
diff --git a/Conductor/Client/Authentication/TokenHandler.cs b/Conductor/Client/Authentication/TokenHandler.cs
@@ -48,9 +48,9 @@ private string GetTokenFromServer(OrkesAuthenticationSettings authenticationSett
                 {
                     return tokenClient.GenerateToken(tokenRequest)._token;
                 }
-                catch (ApiException e)
+                catch (Exception e)
                 {
-                    Console.WriteLine($"Failed to refresh authentication token, reason: {e}, request: {tokenRequest.ToJson()}");
+                    Console.WriteLine($"Failed to refresh authentication token, attempt = {attempt}, error = {e.Message}");
                 }
             }
             throw new Exception("Failed to refresh authentication token");

diff --git a/Conductor/Client/Extensions/DependencyInjectionExtensions.cs b/Conductor/Client/Extensions/DependencyInjectionExtensions.cs
@@ -9,17 +9,10 @@ namespace Conductor.Client.Extensions
 {
     public static class DependencyInjectionExtensions
     {
-        public static IServiceCollection AddConductorWorkflowTask<T>(this IServiceCollection services) where T : IWorkflowTask
-        {
-            services.AddTransient(typeof(IWorkflowTask), typeof(T));
-            services.AddTransient(typeof(T));
-            return services;
-        }
 
         public static IServiceCollection AddConductorWorkflowTask<T>(this IServiceCollection services, T worker) where T : IWorkflowTask
         {
-            services.AddTransient(typeof(IWorkflowTask), typeof(T));
-            services.AddTransient(typeof(T));
+            services.AddSingleton<IWorkflowTask>(worker);
             return services;
         }
 

diff --git a/Conductor/Client/Worker/WorkflowTaskExecutor.cs b/Conductor/Client/Worker/WorkflowTaskExecutor.cs
@@ -47,7 +47,6 @@ public WorkflowTaskExecutor(
 
         public System.Threading.Tasks.Task Start(CancellationToken token)
         {
-
             if (token != CancellationToken.None)
                 token.ThrowIfCancellationRequested();
 
@@ -68,7 +67,6 @@ public System.Threading.Tasks.Task Start(CancellationToken token)
 
         private void Work4Ever(CancellationToken token)
         {
-
             while (true)
             {
                 try
@@ -78,9 +76,21 @@ private void Work4Ever(CancellationToken token)
 
                     WorkOnce(token);
                 }
+                catch (System.OperationCanceledException canceledException)
+                {
+                    //Do nothing the operation was cancelled
+                    _logger.LogTrace(
+                        $"[{_workerSettings.WorkerId}] Operation Cancelled: {canceledException.Message}"
+                        + $", taskName: {_worker.TaskType}"
+                        + $", domain: {_worker.WorkerSettings.Domain}"
+                        + $", batchSize: {_workerSettings.BatchSize}"
+                    );
+                    Sleep(SLEEP_FOR_TIME_SPAN_ON_WORKER_ERROR);
+                }
                 catch (Exception e)
                 {
-                    _logger.LogDebug(
+
+                    _logger.LogError(
                         $"[{_workerSettings.WorkerId}] worker error: {e.Message}"
                         + $", taskName: {_worker.TaskType}"
                         + $", domain: {_worker.WorkerSettings.Domain}"
@@ -102,18 +112,19 @@ private async void WorkOnce(CancellationToken token)
                 Sleep(_workerSettings.PollInterval);
                 return;
             }
+
             var uniqueBatchId = Guid.NewGuid();
             _logger.LogTrace(
-                  $"[{_workerSettings.WorkerId}] Processing tasks batch"
-                  + $", Task batch unique Id: {uniqueBatchId}"
-                  );
+                $"[{_workerSettings.WorkerId}] Processing tasks batch"
+                + $", Task batch unique Id: {uniqueBatchId}"
+            );
 
             await System.Threading.Tasks.Task.Run(() => ProcessTasks(tasks, token));
 
             _logger.LogTrace(
-                  $"[{_workerSettings.WorkerId}] Completed tasks batch"
-                  + $", Task batch unique Id: {uniqueBatchId}"
-                  );
+                $"[{_workerSettings.WorkerId}] Completed tasks batch"
+                + $", Task batch unique Id: {uniqueBatchId}"
+            );
         }
 
         private List<Models.Task> PollTasks()
@@ -127,30 +138,47 @@ private async void WorkOnce(CancellationToken token)
             var availableWorkerCounter = _workerSettings.BatchSize - _workflowTaskMonitor.GetRunningWorkers();
             if (availableWorkerCounter < 1)
             {
-                throw new Exception("no worker available");
+                _logger.LogDebug("All workers are busy");
+                return new List<Task>();
             }
-            var tasks = _taskClient.PollTask(_worker.TaskType, _workerSettings.WorkerId, _workerSettings.Domain, availableWorkerCounter);
-            if (tasks == null)
+
+            try
             {
-                tasks = new List<Models.Task>();
+                var tasks = _taskClient.PollTask(_worker.TaskType, _workerSettings.WorkerId, _workerSettings.Domain,
+                    availableWorkerCounter);
+                if (tasks == null)
+                {
+                    tasks = new List<Models.Task>();
+                }
+
+                _logger.LogTrace(
+                    $"[{_workerSettings.WorkerId}] Polled {tasks.Count} tasks"
+                    + $", taskType: {_worker.TaskType}"
+                    + $", domain: {_workerSettings.Domain}"
+                    + $", batchSize: {_workerSettings.BatchSize}"
+                );
+                return tasks;
+            }
+            catch (Exception e)
+            {
+                _logger.LogTrace(
+                    $"[{_workerSettings.WorkerId}] Polling error: {e.Message} "
+                    + $", taskType: {_worker.TaskType}"
+                    + $", domain: {_workerSettings.Domain}"
+                    + $", batchSize: {_workerSettings.BatchSize}"
+                );
+                return new List<Task>();
             }
-            _logger.LogTrace(
-                $"[{_workerSettings.WorkerId}] Polled {tasks.Count} tasks"
-                + $", taskType: {_worker.TaskType}"
-                + $", domain: {_workerSettings.Domain}"
-                + $", batchSize: {_workerSettings.BatchSize}"
-            );
-            return tasks;
         }
 
         private async void ProcessTasks(List<Models.Task> tasks, CancellationToken token)
         {
-
             List<System.Threading.Tasks.Task> threads = new List<System.Threading.Tasks.Task>();
             if (tasks == null || tasks.Count == 0)
             {
                 return;
             }
+
             foreach (var task in tasks)
             {
                 if (token != CancellationToken.None)
@@ -159,6 +187,7 @@ private async void ProcessTasks(List<Models.Task> tasks, CancellationToken token
                 _workflowTaskMonitor.IncrementRunningWorker();
                 threads.Add(System.Threading.Tasks.Task.Run(() => ProcessTask(task, token)));
             }
+
             await System.Threading.Tasks.Task.WhenAll(threads);
         }
 
@@ -178,25 +207,26 @@ private async void ProcessTask(Models.Task task, CancellationToken token)
 
             try
             {
-                TaskResult taskResult = new TaskResult(taskId: task.TaskId, workflowInstanceId: task.WorkflowInstanceId);
+                TaskResult taskResult =
+                    new TaskResult(taskId: task.TaskId, workflowInstanceId: task.WorkflowInstanceId);
 
                 if (token == CancellationToken.None)
                     taskResult = _worker.Execute(task);
                 else
                     taskResult = await _worker.Execute(task, token);
                 _logger.LogTrace(
-                   $"[{_workerSettings.WorkerId}] Done processing task for worker"
-                   + $", taskType: {_worker.TaskType}"
-                   + $", domain: {_workerSettings.Domain}"
-                   + $", taskId: {task.TaskId}"
-                   + $", workflowId: {task.WorkflowInstanceId}"
-                   + $", CancelToken: {token}"
-               );
+                    $"[{_workerSettings.WorkerId}] Done processing task for worker"
+                    + $", taskType: {_worker.TaskType}"
+                    + $", domain: {_workerSettings.Domain}"
+                    + $", taskId: {task.TaskId}"
+                    + $", workflowId: {task.WorkflowInstanceId}"
+                    + $", CancelToken: {token}"
+                );
                 UpdateTask(taskResult);
             }
             catch (Exception e)
             {
-                _logger.LogDebug(
+                _logger.LogError(
                     $"[{_workerSettings.WorkerId}] Failed to process task for worker, reason: {e.Message}"
                     + $", taskType: {_worker.TaskType}"
                     + $", domain: {_workerSettings.Domain}"
@@ -206,7 +236,6 @@ private async void ProcessTask(Models.Task task, CancellationToken token)
                 );
                 var taskResult = task.Failed(e.Message);
                 UpdateTask(taskResult);
-
             }
             finally
             {
@@ -228,6 +257,7 @@ private void UpdateTask(Models.TaskResult taskResult)
                     {
                         Sleep(TimeSpan.FromSeconds(1 << attemptCounter));
                     }
+
                     _taskClient.UpdateTask(taskResult);
                     _logger.LogTrace(
                         $"[{_workerSettings.WorkerId}] Done updating task"
@@ -240,7 +270,7 @@ private void UpdateTask(Models.TaskResult taskResult)
                 }
                 catch (Exception e)
                 {
-                    _logger.LogTrace(
+                    _logger.LogError(
                         $"[{_workerSettings.WorkerId}] Failed to update task, reason: {e.Message}"
                         + $", taskType: {_worker.TaskType}"
                         + $", domain: {_workerSettings.Domain}"
@@ -249,6 +279,7 @@ private void UpdateTask(Models.TaskResult taskResult)
                     );
                 }
             }
+
             throw new Exception("Failed to update task after retries");
         }
 
@@ -260,7 +291,6 @@ private void Sleep(TimeSpan timeSpan)
 
         private void LogInfo()
         {
-
         }
     }
-}
+}
diff --git a/Tests/Worker/WorkerTests.cs b/Tests/Worker/WorkerTests.cs
@@ -30,7 +30,7 @@ public async System.Threading.Tasks.Task TestWorkflowAsyncExecution()
         {
             var workflow = GetConductorWorkflow();
             ApiExtensions.GetWorkflowExecutor().RegisterWorkflow(workflow, true);
-            var workflowIdList = await StartWorkflows(workflow, quantity: 35);
+            var workflowIdList = await StartWorkflows(workflow, quantity: 15);
             await ExecuteWorkflowTasks(workflowCompletionTimeout: TimeSpan.FromSeconds(20));
             await ValidateWorkflowCompletion(workflowIdList.ToArray());
         }
@@ -50,16 +50,15 @@ private async System.Threading.Tasks.Task<ConcurrentBag<string>> StartWorkflows(
             var startedWorkflows = await WorkflowExtensions.StartWorkflows(
                 _workflowClient,
                 startWorkflowRequest,
-                maxAllowedInParallel: 10,
+                maxAllowedInParallel: 3,
                 total: quantity
             );
             return startedWorkflows;
         }
 
         private async System.Threading.Tasks.Task ExecuteWorkflowTasks(TimeSpan workflowCompletionTimeout)
         {
-            var host = WorkflowTaskHost.CreateWorkerHost(Microsoft.Extensions.Logging.LogLevel.Debug);
-            host = WorkflowTaskHost.CreateWorkerHost(Microsoft.Extensions.Logging.LogLevel.Debug, new ClassWorker());
+            var host = WorkflowTaskHost.CreateWorkerHost(Microsoft.Extensions.Logging.LogLevel.Information, new ClassWorker());
             await host.StartAsync();
             Thread.Sleep(workflowCompletionTimeout);
             await host.StopAsync();

diff --git a/Tests/Worker/Workers.cs b/Tests/Worker/Workers.cs
@@ -30,7 +30,6 @@ public static TaskResult SimpleWorker(Conductor.Client.Models.Task task)
         public TaskResult LazyWorker(Conductor.Client.Models.Task task)
         {
             var timeSpan = System.TimeSpan.FromMilliseconds(_random.Next(128, 2048));
-            Console.WriteLine($"Lazy worker is going to rest for {timeSpan.Milliseconds} ms");
             System.Threading.Tasks.Task.Delay(timeSpan).GetAwaiter().GetResult();
             return task.Completed();
         }

diff --git a/csharp-examples/Dockerfile b/csharp-examples/Dockerfile
@@ -1,14 +1,15 @@
-FROM mcr.microsoft.com/dotnet/runtime:8.0-nanoserver-1809 AS base
+FROM mcr.microsoft.com/dotnet/runtime:8.0.0-alpine3.18-amd64 AS base
+RUN mkdir /app
 WORKDIR /app
 
-FROM mcr.microsoft.com/dotnet/sdk:8.0-nanoserver-1809 AS build
+FROM mcr.microsoft.com/dotnet/sdk:8.0.100-1-alpine3.18-amd64 AS build
 ARG BUILD_CONFIGURATION=Release
-WORKDIR /src
+
 COPY ["csharp-examples/csharp-examples.csproj", "csharp-examples/"]
 COPY ["Conductor/conductor-csharp.csproj", "Conductor/"]
 RUN dotnet restore "./csharp-examples/./csharp-examples.csproj"
 COPY . .
-WORKDIR "/src/csharp-examples"
+WORKDIR "csharp-examples"
 RUN dotnet build "./csharp-examples.csproj" -c %BUILD_CONFIGURATION% -o /app/build
 
 FROM build AS publish
@@ -18,4 +19,4 @@ RUN dotnet publish "./csharp-examples.csproj" -c %BUILD_CONFIGURATION% -o /app/p
 FROM base AS final
 WORKDIR /app
 COPY --from=publish /app/publish .
-ENTRYPOINT ["dotnet", "csharp-examples.dll"]
+ENTRYPOINT ["dotnet", "csharp-examples.dll"]
diff --git a/csharp-examples/DockerfileMacArm b/csharp-examples/DockerfileMacArm
@@ -0,0 +1,22 @@
+FROM mcr.microsoft.com/dotnet/runtime:8.0.0-alpine3.18-arm64v8 AS base
+RUN mkdir /app
+WORKDIR /app
+
+FROM mcr.microsoft.com/dotnet/sdk:8.0.100-alpine3.18-arm64v8 AS build
+ARG BUILD_CONFIGURATION=Release
+
+COPY ["csharp-examples/csharp-examples.csproj", "csharp-examples/"]
+COPY ["Conductor/conductor-csharp.csproj", "Conductor/"]
+RUN dotnet restore "./csharp-examples/./csharp-examples.csproj"
+COPY . .
+WORKDIR "csharp-examples"
+RUN dotnet build "./csharp-examples.csproj" -c %BUILD_CONFIGURATION% -o /app/build
+
+FROM build AS publish
+ARG BUILD_CONFIGURATION=Release
+RUN dotnet publish "./csharp-examples.csproj" -c %BUILD_CONFIGURATION% -o /app/publish /p:UseAppHost=false
+
+FROM base AS final
+WORKDIR /app
+COPY --from=publish /app/publish .
+ENTRYPOINT ["dotnet", "csharp-examples.dll"]
diff --git a/csharp-examples/Program.cs b/csharp-examples/Program.cs
@@ -1,13 +1,10 @@
 using csharp_examples;
 
-class Program
+internal class Program
 {
-    static void Main()
+    private static void Main()
     {
-        Runner runner = new Runner();
-
-        // call specific method based on testing scenario
-        runner.RunMultiSimpleTask();
-        //runner.RunSimpleTask();
+        var runner = new Runner();
+        runner.StartTasks();
     }
 }