Merge pull request #487 from keboola/PST-2444-ondra

keboola.flow is orchestration container job type

Author: ondrajodas

src/JobFactory.php

@@ -12,6 +12,9 @@
 abstract class JobFactory
 {
     public const ORCHESTRATOR_COMPONENT = 'keboola.orchestrator';
+
+    public const FLOW_COMPONENT = 'keboola.flow';
+
     public const PROTECTED_DEFAULT_BRANCH_FEATURE = 'protected-default-branch';
 
     /**

src/JobFactory/JobRuntimeResolver.php

@@ -275,7 +275,9 @@ private function resolveJobType(array $jobData): JobType
         if ((intval($jobData['parallelism']) > 0) || $jobData['parallelism'] === JobInterface::PARALLELISM_INFINITY) {
             return JobType::ROW_CONTAINER;
         } else {
-            if ($jobData['componentId'] === JobFactory::ORCHESTRATOR_COMPONENT) {
+            if ($jobData['componentId'] === JobFactory::FLOW_COMPONENT) {
+                return JobType::ORCHESTRATION_CONTAINER;
+            } elseif ($jobData['componentId'] === JobFactory::ORCHESTRATOR_COMPONENT) {
                 if (isset($jobData['configData']['phaseId']) && (string) ($jobData['configData']['phaseId']) !== '') {
                     return JobType::PHASE_CONTAINER;
                 } else {

src/Orchestration/OrchestrationJobMatcher.php

@@ -92,7 +92,7 @@ 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) {
+        if (!in_array($job->getComponentId(), [JobFactory::ORCHESTRATOR_COMPONENT, JobFactory::FLOW_COMPONENT], true)) {
             throw new OrchestrationJobMatcherValidationException(sprintf(
                 'Job "%s" is not an orchestration job.',
                 $job->getId(),

tests/JobFactory/JobRuntimeResolverTest.php

@@ -69,6 +69,12 @@ public function resolveDefaultBackendContextData(): Generator
             JobType::ORCHESTRATION_CONTAINER,
             null,
         ];
+        yield 'flow job' => [
+            JobFactory::FLOW_COMPONENT,
+            [],
+            JobType::ORCHESTRATION_CONTAINER,
+            null,
+        ];
     }
 
     /**

tests/Orchestration/OrchestrationJobMatcherTest.php

@@ -293,6 +293,106 @@ public function testMatcherPartial(): void
         $componentsApi->deleteConfiguration(JobFactory::ORCHESTRATOR_COMPONENT, $orchestrationConfigurationId);
     }
 
+    public function testMatcherFlowComponent(): void
+    {
+        $client = $this->getClient();
+        $configuration = $this->getOrchestrationConfiguration();
+
+        // Vytvoříme nový job s FLOW_COMPONENT
+        $storageClient = new StorageClient([
+            'token' => self::getRequiredEnv('TEST_STORAGE_API_TOKEN_MASTER'),
+            'url' => self::getRequiredEnv('TEST_STORAGE_API_URL'),
+        ]);
+
+        // create token for job
+        $tokenOptions = (new TokenCreateOptions())
+            ->setDescription(__CLASS__)
+            ->setCanManageBuckets(true)
+            ->setExpiresIn(60 * 5)
+        ;
+
+        $tokens = new Tokens($storageClient);
+        $token = $tokens->createToken($tokenOptions);
+
+        $componentsApi = new Components($storageClient);
+        $componentConfig = new Configuration();
+        $componentConfig->setConfiguration($configuration);
+        $componentConfig->setName('testMatcherFlowComponent');
+        $componentConfig->setComponentId(JobFactory::FLOW_COMPONENT);
+        $this->componentId = JobFactory::FLOW_COMPONENT;
+        $this->configurationId = $componentsApi->addConfiguration($componentConfig)['id'];
+
+        $flowJob = $this->getNewJobFactory()->createNewJob(
+            [
+                '#tokenString' => $token['token'],
+                'configData' => [],
+                'componentId' => JobFactory::FLOW_COMPONENT,
+                'configId' => $this->configurationId,
+                'mode' => 'run',
+                'parentRunId' => '',
+                'orchestrationJobId' => null,
+            ],
+        );
+        $client->createJob($flowJob);
+
+        // Vytvoříme child jobs
+        $jobIds = [];
+        foreach ($configuration['tasks'] as $task) {
+            $jobIds[] = $client->createJob($this->getNewJobFactory()->createNewJob(
+                [
+                    '#tokenString' => $token['token'],
+                    'componentId' => $task['task']['componentId'],
+                    'configData' => $task['task']['configData'],
+                    'mode' => $task['task']['mode'],
+                    'orchestrationJobId' => $flowJob->getId(),
+                    'orchestrationTaskId' => (string) $task['id'],
+                    'parentRunId' => $flowJob->getId(),
+                ],
+            ))->getId();
+        }
+
+        $tokens->dropToken($token['id']);
+
+        $matcher = new OrchestrationJobMatcher($client, $this->createStorageClientPlainFactory());
+        $results = $matcher->matchTaskJobsForOrchestrationJob(
+            $flowJob->getId(),
+            self::getRequiredEnv('TEST_STORAGE_API_TOKEN'),
+        );
+        self::assertEquals(
+            new OrchestrationJobMatcherResults(
+                $flowJob->getId(),
+                $this->configurationId,
+                [
+                    new OrchestrationTaskMatched(
+                        '30679',
+                        true,
+                        $jobIds[0],
+                        'keboola.ex-db-snowflake',
+                        null,
+                        'created',
+                    ),
+                    new OrchestrationTaskMatched(
+                        '92543',
+                        true,
+                        $jobIds[1],
+                        'keboola.snowflake-transformation',
+                        null,
+                        'created',
+                    ),
+                    new OrchestrationTaskMatched(
+                        '25052',
+                        true,
+                        $jobIds[2],
+                        'keboola.ex-sample-data',
+                        null,
+                        'created',
+                    ),
+                ],
+            ),
+            $results,
+        );
+    }
+
     /**
      * @dataProvider invalidConfigurationProvider
      */