improving on the code to limit the responsabilities of the kubernetes facade

Author: lm-sec

packages/backend/orchestrator/service/Orchestrator/Events/JobEventsProducer.cs

@@ -14,13 +14,15 @@ public enum JobStatus
     {
         Success,
         Failed,
-        Ended
+        Ended,
+        Started
     }
 
     private static readonly Dictionary<JobStatus, string> JobStatusMapping = new Dictionary<JobStatus, string>()  { 
         { JobStatus.Success, "Success" }, 
         { JobStatus.Failed, "Failed" }, 
-        { JobStatus.Ended, "Ended" }
+        { JobStatus.Ended, "Ended" },
+        { JobStatus.Started, "Started" }
     };
 
     private static Dictionary<string, JobStatus> InvertStatusMapping(Dictionary<JobStatus, string> jobStatusMapping)

packages/backend/orchestrator/service/Orchestrator/Jobs/Commands/NucleiCustomJobCommand.cs

@@ -11,7 +11,7 @@ public class NucleiCustomJobCommand : KubernetesJobCommand<CustomJobRequest>
 {
     protected override KubernetesJobTemplate JobTemplate { get; }
 
-    public NucleiCustomJobCommand(CustomJobRequest request, JobModel model, IKubernetesFacade kubernetes, IMessagesProducer<JobEventMessage> eventsProducer, JobLogsProducer jobLogsProducer, IFindingsParser parser, ILogger<NucleiCustomJobCommand> logger, IConfiguration config)
+    public NucleiCustomJobCommand(CustomJobRequest request, JobModel model, IKubernetesFacade kubernetes, JobEventsProducer eventsProducer, JobLogsProducer jobLogsProducer, IFindingsParser parser, ILogger<NucleiCustomJobCommand> logger, IConfiguration config)
         : base(request, kubernetes, eventsProducer, jobLogsProducer, parser, logger)
     {
         JobTemplate = new NucleiCustomJobTemplate(request.JobId, config, request.CustomJobParameters, model.Code, request.JobPodMilliCpuLimit, request.JobPodMemoryKbLimit, model.FindingHandler, model.Image);

packages/backend/orchestrator/service/Orchestrator/Jobs/Commands/PythonCustomJobCommand.cs

@@ -11,7 +11,7 @@ public class PythonCustomJobCommand : KubernetesJobCommand<CustomJobRequest>
 {
     protected override KubernetesJobTemplate JobTemplate { get; }
 
-    public PythonCustomJobCommand(CustomJobRequest request, JobModel model, IKubernetesFacade kubernetes, IMessagesProducer<JobEventMessage> eventsProducer, JobLogsProducer jobLogsProducer, IFindingsParser parser, ILogger<PythonCustomJobCommand> logger, IConfiguration config)
+    public PythonCustomJobCommand(CustomJobRequest request, JobModel model, IKubernetesFacade kubernetes, JobEventsProducer eventsProducer, JobLogsProducer jobLogsProducer, IFindingsParser parser, ILogger<PythonCustomJobCommand> logger, IConfiguration config)
         : base(request, kubernetes, eventsProducer, jobLogsProducer, parser, logger)
     {
         JobTemplate = new PythonCustomJobTemplate(request.JobId, config, request.CustomJobParameters, model.Code, request.JobPodMilliCpuLimit, request.JobPodMemoryKbLimit, model.Image);

packages/backend/orchestrator/service/Orchestrator/Jobs/JobFactory.cs

@@ -17,7 +17,7 @@ namespace Orchestrator.Jobs;
 public class JobFactory : IJobFactory
 {
     private IKubernetesFacade Kubernetes { get; }
-    private IMessagesProducer<JobEventMessage> EventsProducer { get; }
+    private JobEventsProducer EventsProducer { get; }
     private JobLogsProducer JobLogsProducer { get; }
     private IFindingsParser Parser { get; }
     private ILoggerFactory LoggerFactory { get; }
@@ -28,7 +28,7 @@ public JobFactory(IKubernetesFacade kubernetes, IMessagesProducer<JobEventMessag
     {
         JobLogsProducer = jobLogsProducer as JobLogsProducer;
         Kubernetes = kubernetes;
-        EventsProducer = eventsProducer;
+        EventsProducer = eventsProducer as JobEventsProducer;
         Parser = parser;
         LoggerFactory = loggerFactoryFactory;
         Logger = loggerFactoryFactory.CreateLogger<JobFactory>();

packages/backend/orchestrator/service/Orchestrator/Jobs/JobManagementCommands/TerminateJobCommand.cs

@@ -8,13 +8,21 @@ namespace Orchestrator.Jobs.JobManagementCommand;
 
 public class TerminateJobCommand : KubernetesManagementCommand<TerminateJobRequest>
 {
-    public TerminateJobCommand(TerminateJobRequest request, IKubernetesFacade kubernetes, IMessagesProducer<JobEventMessage> eventsProducer, JobLogsProducer jobLogsProducer, IFindingsParser parser, ILogger<TerminateJobCommand> logger, IConfiguration config)
+    public TerminateJobCommand(TerminateJobRequest request, IKubernetesFacade kubernetes, JobEventsProducer eventsProducer, JobLogsProducer jobLogsProducer, IFindingsParser parser, ILogger<TerminateJobCommand> logger, IConfiguration config)
         : base(request, kubernetes, eventsProducer, jobLogsProducer, parser, logger, config)
     { }
 
-    public override Task Execute()
+    public async override Task Execute()
     {
-        Kubernetes.TerminateJob(Request.JobId, Namespace);
-        return Task.CompletedTask;
+        bool terminationSuccess = await Kubernetes.TerminateJob(Request.JobId, Namespace);
+        if (terminationSuccess)
+        {
+            LogsProducer.LogDebug(Request.JobId, $"Job terminated");
+        } 
+        else
+        {
+            LogsProducer.LogDebug(Request.JobId, "Tried to terminate, but no corresponding job was runnning");
+        }
+        _ = EventsProducer.LogStatus(Request.JobId, JobEventsProducer.JobStatus.Ended);
     }
 }

packages/backend/orchestrator/service/Orchestrator/Jobs/KubernetesJobCommand.cs

@@ -8,15 +8,15 @@ namespace Orchestrator.Jobs;
 public abstract class KubernetesJobCommand<T> : JobCommand where T : JobRequest
 {
     private IKubernetesFacade Kubernetes { get; }
-    private IMessagesProducer<JobEventMessage> EventsProducer { get; }
+    private JobEventsProducer EventsProducer { get; }
     private JobLogsProducer LogsProducer { get; }
     private IFindingsParser Parser { get; }
     private ILogger Logger { get; }
     protected T Request { get; }
 
     protected abstract KubernetesJobTemplate JobTemplate { get; }
 
-    protected KubernetesJobCommand(T request, IKubernetesFacade kubernetes, IMessagesProducer<JobEventMessage> eventsProducer, JobLogsProducer jobLogsProducer, IFindingsParser parser, ILogger logger)
+    protected KubernetesJobCommand(T request, IKubernetesFacade kubernetes, JobEventsProducer eventsProducer, JobLogsProducer jobLogsProducer, IFindingsParser parser, ILogger logger)
     {
         Request = request;
         Kubernetes = kubernetes;
@@ -34,11 +34,6 @@ public override async Task Execute()
         LogsProducer.LogDebug(Request.JobId, "Job picked up by orchestrator.");
         Logger.LogInformation(Request.JobId, "Job created, listening for events.");
 
-        await EventsProducer.Produce(new JobEventMessage
-        {
-            JobId = Request.JobId,
-            FindingsJson = "{ \"findings\": [{ \"type\": \"JobStatusFinding\", \"status\": \"Started\" }]}",
-            Timestamp = TimeUtils.CurrentTimeMs(),
-        });
+        await EventsProducer.LogStatus(Request.JobId, JobEventsProducer.JobStatus.Started);
     }
 }

packages/backend/orchestrator/service/Orchestrator/Jobs/KubernetesManagementCommand.cs

@@ -10,15 +10,15 @@ namespace Orchestrator.Jobs;
 public abstract class KubernetesManagementCommand<T> : JobCommand where T : JobManagementRequest
 {
     protected IKubernetesFacade Kubernetes { get; }
-    private IMessagesProducer<JobEventMessage> EventsProducer { get; }
-    private JobLogsProducer LogsProducer { get; }
+    protected JobEventsProducer EventsProducer { get; }
+    protected JobLogsProducer LogsProducer { get; }
     private IFindingsParser Parser { get; }
     private ILogger Logger { get; }
     protected T Request { get; }
     public string Namespace { get; set; } = "default";
     protected IConfiguration Config { get; init; }
 
-    protected KubernetesManagementCommand(T request, IKubernetesFacade kubernetes, IMessagesProducer<JobEventMessage> eventsProducer, JobLogsProducer jobLogsProducer, IFindingsParser parser, ILogger logger, IConfiguration config)
+    protected KubernetesManagementCommand(T request, IKubernetesFacade kubernetes, JobEventsProducer eventsProducer, JobLogsProducer jobLogsProducer, IFindingsParser parser, ILogger logger, IConfiguration config)
     {
         Request = request;
         Kubernetes = kubernetes;

packages/backend/orchestrator/service/Orchestrator/K8s/IKubernetesFacade.cs

@@ -14,7 +14,7 @@ public interface IKubernetesFacade
     /// Creates a jobTemplate.
     /// </summary>
     /// <returns></returns>
-    Task TerminateJob(string jobId, string jobNamespace = "default");
+    Task<bool> TerminateJob(string jobId, string jobNamespace = "default");
 
     /// <summary>
     /// True if the pod is in the status "Failed" or "Succeeded", false otherwise

packages/backend/orchestrator/service/Orchestrator/K8s/KubernetesFacade.cs

@@ -8,19 +8,15 @@ namespace Orchestrator.K8s;
 public class KubernetesFacade : IKubernetesFacade
 {
     private readonly ILogger<KubernetesFacade> Logger;
-    private JobEventsProducer JobEventsProducer { get; }
-    private JobLogsProducer JobLogsProducer { get; }
 
     /// <summary>
     /// Gets or sets the Kubernetes configuration.
     /// </summary>
     //// private KubernetesClientConfiguration KubernetesConfiguration => KubernetesClientConfiguration.BuildConfigFromConfigFile(Environment.GetEnvironmentVariable("KUBECONFIG"));
     private static KubernetesClientConfiguration KubernetesConfiguration => KubernetesClientConfiguration.InClusterConfig();
 
-    public KubernetesFacade(ILogger<KubernetesFacade> logger, IMessagesProducer<JobEventMessage> jobEventsProducer, IMessagesProducer<JobLogMessage> jobLogsProducer)
+    public KubernetesFacade(ILogger<KubernetesFacade> logger)
     {
-        JobEventsProducer = jobEventsProducer as JobEventsProducer;
-        JobLogsProducer = jobLogsProducer as JobLogsProducer;
         Logger = logger;
     }
 
@@ -137,7 +133,7 @@ public async Task<bool> IsJobPodFinished(string jobName, string jobNamespace = "
     /// <param name="jobId"></param>
     /// <param name="jobNamespace"></param>
     /// <returns>true if the job was running and terminated, false if it was not running.</returns>
-    public async Task TerminateJob(string jobId, string jobNamespace = "default")
+    public async Task<bool> TerminateJob(string jobId, string jobNamespace = "default")
     {
         using var client = new Kubernetes(KubernetesConfiguration);
         string labelSelector = "red-kite.io/jobid=" + jobId;
@@ -149,14 +145,10 @@ public async Task TerminateJob(string jobId, string jobNamespace = "default")
             var options = new V1DeleteOptions(gracePeriodSeconds: 0, propagationPolicy: "Foreground");
 
             await RetryableCall(() => client.DeleteNamespacedJobAsync(runningJob.Metadata.Name, jobNamespace, options));
-            JobLogsProducer.LogDebug(jobId, $"Job terminated");
+            return true;
         }
-        else
-        {
-            JobLogsProducer.LogDebug(jobId, "Tried to terminate, but no corresponding job was runnning");
-        }
-        _ = JobEventsProducer.LogStatus(jobId, JobEventsProducer.JobStatus.Ended);
-        return;
+        
+        return false;
     }
 
     /// <summary>