Merge pull request #347 from keboola/odin-SOX-407

JobMatcher for orchestration retry

Author: odinuv

.gitignore

@@ -10,3 +10,4 @@
 /build/logs
 /infection.log
 /.phpunit.result.cache
+/var/

provisioning/local/.terraform.lock.hcl

@@ -0,0 +1,10 @@
+<?php
+
+declare(strict_types=1);
+
+namespace Keboola\JobQueueInternalClient\Exception;
+
+class OrchestrationJobMatcherValidationException extends ClientException
+{
+
+}

src/Exception/OrchestrationJobMatcherValidationException.php

@@ -0,0 +1,104 @@
+<?php
+
+declare(strict_types=1);
+
+namespace Keboola\JobQueueInternalClient\Orchestration;
+
+use Keboola\JobQueueInternalClient\Client;
+use Keboola\JobQueueInternalClient\Exception\ClientException;
+use Keboola\JobQueueInternalClient\Exception\OrchestrationJobMatcherValidationException;
+use Keboola\JobQueueInternalClient\JobFactory;
+use Keboola\JobQueueInternalClient\JobFactory\Job;
+use Keboola\JobQueueInternalClient\JobFactory\JobInterface;
+use Keboola\JobQueueInternalClient\JobListOptions;
+
+// https://keboola.atlassian.net/wiki/spaces/ENGG/pages/3074195457/DRAFT+RFC-2023-011+-+Rerun+orchestration#Pair-Jobs-and-Tasks
+class OrchestrationJobMatcher
+{
+    public function __construct(
+        private readonly Client $internalClient,
+    ) {
+    }
+
+    public function matchTaskJobsForOrchestrationJob(string $jobId): OrchestrationJobMatcherResults
+    {
+        $job = $this->internalClient->getJob($jobId);
+        $childJobs = $this->getOrchestrationTaskJobs($job);
+        $configuration = $this->getCurrentOrchestrationConfiguration($job);
+        $this->validateInputs($job, $configuration);
+        $matchedTasks = [];
+        foreach ($configuration['tasks'] as $task) {
+            $matched = false;
+            foreach ($childJobs as $index => $childJob) {
+                if (((string) $task['id']) === $childJob->getOrchestrationTaskId()) {
+                    $matchedTasks[] = new OrchestrationTaskMatched(
+                        (string) $task['id'],
+                        true,
+                        $childJob->getId(),
+                        $childJob->getComponentId(),
+                        $childJob->getConfigId(),
+                        $childJob->getStatus(),
+                    );
+                    unset($childJobs[$index]);
+                    $matched = true;
+                    break;
+                }
+            }
+            if (!$matched) {
+                $matchedTasks[] = new OrchestrationTaskMatched((string) $task['id'], false, null, null, null, null);
+            }
+        }
+        return new OrchestrationJobMatcherResults(
+            $jobId,
+            $job->getConfigId(),
+            $matchedTasks,
+        );
+    }
+
+    /**
+     * @return array<JobInterface>
+     */
+    private function getOrchestrationTaskJobs(JobInterface $job): array
+    {
+        return $this->internalClient->listJobs(
+            (new JobListOptions())->setParentRunId($job->getId()),
+            true,
+        );
+    }
+
+    private function getCurrentOrchestrationConfiguration(JobInterface $job): array
+    {
+        $configuration = $job->getConfigData();
+        if (!$configuration) {
+            $configuration = $job->getComponentConfiguration()['configuration'];
+        };
+        return $configuration;
+    }
+
+    private function validateInputs(JobInterface $job, array $configuration): void
+    {
+        /* since the matcher accepts a jobId, we need to check that it is a sort of sensible input -
+            the main use case is root orchestration job, but it seems that a phaseContainer might be
+            equally valid input. */
+        if ($job->getComponentId() !== JobFactory::ORCHESTRATOR_COMPONENT) {
+            throw new OrchestrationJobMatcherValidationException(sprintf(
+                'Job "%s" is not an orchestration job.',
+                $job->getId(),
+            ));
+        }
+        if (!isset($configuration['tasks']) || !is_array($configuration['tasks'])) {
+            throw new OrchestrationJobMatcherValidationException(sprintf(
+                'Orchestration "%s" tasks must be an array.',
+                $job->getId(),
+            ));
+        }
+        array_walk($configuration['tasks'], function (array $task) {
+            if (!isset($task['id'])) {
+                throw new OrchestrationJobMatcherValidationException(sprintf(
+                    'Task does not have an id. (%s)',
+                    json_encode($task),
+                ));
+            }
+        });
+    }
+}

src/Orchestration/OrchestrationJobMatcher.php

@@ -0,0 +1,18 @@
+<?php
+
+declare(strict_types=1);
+
+namespace Keboola\JobQueueInternalClient\Orchestration;
+
+class OrchestrationJobMatcherResults
+{
+    /**
+     * @param array<OrchestrationTaskMatched> $matchedTasks
+     */
+    public function __construct(
+        public readonly string $jobId,
+        public readonly ?string $configId,
+        public readonly array $matchedTasks,
+    ) {
+    }
+}

src/Orchestration/OrchestrationJobMatcherResults.php

@@ -0,0 +1,18 @@
+<?php
+
+declare(strict_types=1);
+
+namespace Keboola\JobQueueInternalClient\Orchestration;
+
+class OrchestrationTaskMatched
+{
+    public function __construct(
+        public readonly string $taskId,
+        public readonly bool $matched,
+        public readonly ?string $jobId,
+        public readonly ?string $componentId,
+        public readonly ?string $configId,
+        public readonly ?string $status,
+    ) {
+    }
+}