Skip to content

Commit e62e491

Browse files
pepamartinecpepamartinec
authored
Merge pull request #159 from keboola/pepa_PS-3696_lazyEncryptor
PS-3696 Lazy encryptor
2 parents 8ebb61b + 27da4a2 commit e62e491

28 files changed

+2274
-1287
lines changed

‎src/Client.php‎

Lines changed: 9 additions & 14 deletions
Original file line numberDiff line numberDiff line change
@@ -37,12 +37,12 @@ class Client
3737
private const JSON_DEPTH = 512;
3838

3939
protected GuzzleClient $guzzle;
40-
private JobFactory $jobFactory;
40+
private ExistingJobFactory $existingJobFactory;
4141
private LoggerInterface $logger;
4242

4343
public function __construct(
4444
LoggerInterface $logger,
45-
JobFactory $jobFactory,
45+
ExistingJobFactory $existingJobFactory,
4646
string $internalQueueApiUrl,
4747
string $internalQueueToken,
4848
array $options = []
@@ -70,7 +70,7 @@ public function __construct(
7070
throw new ClientException('Invalid parameters when creating client: ' . $messages);
7171
}
7272
$this->guzzle = $this->initClient($internalQueueApiUrl, $internalQueueToken, $options);
73-
$this->jobFactory = $jobFactory;
73+
$this->existingJobFactory = $existingJobFactory;
7474
$this->logger = $logger;
7575
}
7676

@@ -88,7 +88,7 @@ public function createJob(JobInterface $job): JobInterface
8888
throw new ClientException('Invalid job data: ' . $e->getMessage(), $e->getCode(), $e);
8989
}
9090
$result = $this->sendRequest($request);
91-
return $this->jobFactory->loadFromExistingJobData($result);
91+
return $this->existingJobFactory->loadFromExistingJobData($result);
9292
}
9393

9494
public function createJobsBatch(array $jobs): array
@@ -101,12 +101,7 @@ public function createJobsBatch(array $jobs): array
101101
}
102102
$result = $this->sendRequest($request);
103103

104-
return array_map(fn(array $jobData) => $this->jobFactory->loadFromExistingJobData($jobData), $result);
105-
}
106-
107-
public function getJobFactory(): JobFactory
108-
{
109-
return $this->jobFactory;
104+
return array_map(fn(array $jobData) => $this->existingJobFactory->loadFromExistingJobData($jobData), $result);
110105
}
111106

112107
public function getJob(string $jobId): JobInterface
@@ -117,7 +112,7 @@ public function getJob(string $jobId): JobInterface
117112

118113
$request = new Request('GET', 'jobs/' . $jobId);
119114
$result = $this->sendRequest($request);
120-
return $this->jobFactory->loadFromExistingJobData($result);
115+
return $this->existingJobFactory->loadFromExistingJobData($result);
121116
}
122117

123118
/**
@@ -209,7 +204,7 @@ public function patchJob(string $jobId, JobPatchData $patchData): JobInterface
209204
[],
210205
json_encode($patchData->jsonSerialize(), JSON_THROW_ON_ERROR)
211206
);
212-
return $this->jobFactory->loadFromExistingJobData($this->sendRequest($request));
207+
return $this->existingJobFactory->loadFromExistingJobData($this->sendRequest($request));
213208
}
214209

215210
public function postJobResult(
@@ -231,7 +226,7 @@ public function postJobResult(
231226
JSON_THROW_ON_ERROR
232227
)
233228
);
234-
return $this->jobFactory->loadFromExistingJobData($this->sendRequest($request));
229+
return $this->existingJobFactory->loadFromExistingJobData($this->sendRequest($request));
235230
}
236231

237232
/**
@@ -300,7 +295,7 @@ private function mapJobsFromResponse(array $responseBody): array
300295
{
301296
$jobs = array_map(function (array $jobData): ?JobInterface {
302297
try {
303-
return $this->jobFactory->loadFromExistingJobData($jobData);
298+
return $this->existingJobFactory->loadFromExistingJobData($jobData);
304299
} catch (Throwable $e) {
305300
$this->logger->error('Failed to parse Job data: ' . $e->getMessage());
306301
// ignore invalid job

‎src/ClientFactory.php‎

Lines changed: 4 additions & 4 deletions
Original file line numberDiff line numberDiff line change
@@ -10,26 +10,26 @@ class ClientFactory
1010
{
1111
private string $internalApiUrl;
1212
private string $internalApiToken;
13-
private JobFactory $jobFactory;
13+
private ExistingJobFactory $existingJobFactory;
1414
private LoggerInterface $logger;
1515

1616
public function __construct(
1717
string $internalApiUrl,
1818
string $internalApiToken,
19-
JobFactory $jobFactory,
19+
ExistingJobFactory $existingJobFactory,
2020
LoggerInterface $logger
2121
) {
2222
$this->internalApiUrl = $internalApiUrl;
2323
$this->internalApiToken = $internalApiToken;
24-
$this->jobFactory = $jobFactory;
24+
$this->existingJobFactory = $existingJobFactory;
2525
$this->logger = $logger;
2626
}
2727

2828
public function getClient(): Client
2929
{
3030
return new Client(
3131
$this->logger,
32-
$this->jobFactory,
32+
$this->existingJobFactory,
3333
$this->internalApiUrl,
3434
$this->internalApiToken,
3535
[

‎src/ExistingJobFactory.php‎

Lines changed: 33 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,33 @@
1+
<?php
2+
3+
declare(strict_types=1);
4+
5+
namespace Keboola\JobQueueInternalClient;
6+
7+
use Keboola\JobQueueInternalClient\JobFactory\FullJobDefinition;
8+
use Keboola\JobQueueInternalClient\JobFactory\Job;
9+
use Keboola\JobQueueInternalClient\JobFactory\JobInterface;
10+
use Keboola\JobQueueInternalClient\JobFactory\ObjectEncryptorProvider\ObjectEncryptorProviderInterface;
11+
use Keboola\StorageApiBranch\Factory\StorageClientPlainFactory;
12+
13+
class ExistingJobFactory extends JobFactory
14+
{
15+
private StorageClientPlainFactory $storageClientFactory;
16+
private ObjectEncryptorProviderInterface $objectEncryptorProvider;
17+
18+
public function __construct(
19+
StorageClientPlainFactory $storageClientFactory,
20+
ObjectEncryptorProviderInterface $objectEncryptorProvider
21+
) {
22+
$this->storageClientFactory = $storageClientFactory;
23+
$this->objectEncryptorProvider = $objectEncryptorProvider;
24+
}
25+
26+
public function loadFromExistingJobData(array $data): JobInterface
27+
{
28+
$data = $this->validateJobData($data, FullJobDefinition::class);
29+
30+
$encryptor = $this->objectEncryptorProvider->getJobEncryptor($data);
31+
return new Job($encryptor, $this->storageClientFactory, $data);
32+
}
33+
}

‎src/JobFactory.php‎

Lines changed: 2 additions & 157 deletions
Original file line numberDiff line numberDiff line change
@@ -4,21 +4,12 @@
44

55
namespace Keboola\JobQueueInternalClient;
66

7-
use Keboola\JobQueueInternalClient\DataPlane\DataPlaneConfigRepository;
87
use Keboola\JobQueueInternalClient\Exception\ClientException;
9-
use Keboola\JobQueueInternalClient\JobFactory\Behavior;
108
use Keboola\JobQueueInternalClient\JobFactory\FullJobDefinition;
11-
use Keboola\JobQueueInternalClient\JobFactory\Job;
12-
use Keboola\JobQueueInternalClient\JobFactory\JobInterface;
13-
use Keboola\JobQueueInternalClient\JobFactory\JobRuntimeResolver;
149
use Keboola\JobQueueInternalClient\JobFactory\NewJobDefinition;
15-
use Keboola\ObjectEncryptor\ObjectEncryptor;
16-
use Keboola\StorageApi\ClientException as StorageClientException;
17-
use Keboola\StorageApiBranch\Factory\ClientOptions;
18-
use Keboola\StorageApiBranch\Factory\StorageClientPlainFactory;
1910
use Symfony\Component\Config\Definition\Exception\InvalidConfigurationException;
2011

21-
class JobFactory
12+
abstract class JobFactory
2213
{
2314
public const STATUS_CANCELLED = 'cancelled';
2415
public const STATUS_CREATED = 'created';
@@ -43,26 +34,6 @@ class JobFactory
4334

4435
public const PAY_AS_YOU_GO_FEATURE = 'pay-as-you-go';
4536

46-
private StorageClientPlainFactory $storageClientFactory;
47-
private JobRuntimeResolver $jobRuntimeResolver;
48-
private ObjectEncryptor $controlPlaneObjectEncryptor;
49-
private DataPlaneConfigRepository $dataPlaneConfigRepository;
50-
private bool $supportsDataPlanes;
51-
52-
public function __construct(
53-
StorageClientPlainFactory $storageClientFactory,
54-
JobRuntimeResolver $jobRuntimeResolver,
55-
ObjectEncryptor $controlPlaneEncryptor,
56-
DataPlaneConfigRepository $dataPlaneConfigRepository,
57-
bool $supportsDataPlanes
58-
) {
59-
$this->storageClientFactory = $storageClientFactory;
60-
$this->jobRuntimeResolver = $jobRuntimeResolver;
61-
$this->controlPlaneObjectEncryptor = $controlPlaneEncryptor;
62-
$this->dataPlaneConfigRepository = $dataPlaneConfigRepository;
63-
$this->supportsDataPlanes = $supportsDataPlanes;
64-
}
65-
6637
public static function getFinishedStatuses(): array
6738
{
6839
return [self::STATUS_SUCCESS, self::STATUS_WARNING, self::STATUS_ERROR, self::STATUS_CANCELLED,
@@ -102,141 +73,15 @@ public static function getAllowedParallelismValues(): array
10273
return array_merge($intValues, ['infinity', null]);
10374
}
10475

105-
public function createNewJob(array $data): JobInterface
106-
{
107-
$data = $this->validateJobData($data, NewJobDefinition::class);
108-
109-
try {
110-
$client = $this->storageClientFactory->createClientWrapper(new ClientOptions(
111-
null,
112-
$data['#tokenString']
113-
))->getBasicClient();
114-
$tokenInfo = $client->verifyToken();
115-
$jobId = $client->generateId();
116-
$runId = empty($data['parentRunId']) ? $jobId :
117-
$data['parentRunId'] . JobInterface::RUN_ID_DELIMITER . $jobId;
118-
} catch (StorageClientException $e) {
119-
throw new ClientException(
120-
'Cannot create job: "' . $e->getMessage() . '".',
121-
$e->getCode(),
122-
$e
123-
);
124-
}
125-
126-
if (!empty($data['variableValuesId']) && !empty($data['variableValuesData']['values'])) {
127-
throw new ClientException(
128-
'Provide either "variableValuesId" or "variableValuesData", but not both.'
129-
);
130-
}
131-
132-
if ($this->supportsDataPlanes) {
133-
$dataPlaneConfig = $this->dataPlaneConfigRepository->fetchProjectDataPlane(
134-
(string) $tokenInfo['owner']['id'],
135-
);
136-
} else {
137-
$dataPlaneConfig = null;
138-
}
139-
140-
if ($dataPlaneConfig !== null) {
141-
$objectEncryptor = $dataPlaneConfig->getEncryption()->createEncryptor();
142-
} else {
143-
$objectEncryptor = $this->controlPlaneObjectEncryptor;
144-
}
145-
146-
$jobData = [
147-
'id' => $jobId,
148-
'runId' => $runId,
149-
'projectId' => $tokenInfo['owner']['id'],
150-
'projectName' => $tokenInfo['owner']['name'],
151-
'dataPlaneId' => $dataPlaneConfig ? $dataPlaneConfig->getId() : null,
152-
'tokenId' => $tokenInfo['id'],
153-
'#tokenString' => $data['#tokenString'],
154-
'tokenDescription' => $tokenInfo['description'],
155-
'status' => self::STATUS_CREATED,
156-
'desiredStatus' => self::DESIRED_STATUS_PROCESSING,
157-
'mode' => $data['mode'],
158-
'componentId' => $data['componentId'],
159-
'configId' => $data['configId'] ?? null,
160-
'configData' => $data['configData'] ?? null,
161-
'configRowIds' => $data['configRowIds'] ?? null,
162-
'tag' => $data['tag'] ?? null,
163-
'parallelism' => $data['parallelism'] ?? null,
164-
'backend' => $data['backend'] ?? null,
165-
'behavior' => $data['behavior'] ?? (new Behavior())->toDataArray(),
166-
'result' => [],
167-
'usageData' => [],
168-
'isFinished' => false,
169-
'branchId' => $data['branchId'] ?? null,
170-
'variableValuesId' => $data['variableValuesId'] ?? null,
171-
'variableValuesData' => $data['variableValuesData'] ?? [],
172-
'orchestrationJobId' => $data['orchestrationJobId'] ?? null,
173-
];
174-
175-
$jobData = $this->jobRuntimeResolver->resolveJobData($jobData, $tokenInfo);
176-
// set type after resolving parallelism
177-
$jobData['type'] = $data['type'] ?? $this->getJobType($jobData);
178-
179-
$data = $objectEncryptor->encryptForProject(
180-
$jobData,
181-
(string) $data['componentId'],
182-
(string) $tokenInfo['owner']['id']
183-
);
184-
185-
$data = $this->validateJobData($data, FullJobDefinition::class);
186-
return new Job($objectEncryptor, $this->storageClientFactory, $data);
187-
}
188-
189-
public function loadFromExistingJobData(array $data): JobInterface
190-
{
191-
$data = $this->validateJobData($data, FullJobDefinition::class);
192-
193-
// combination $this->supportsDataPlanes === false && data['dataPlaneId'] !== null should be considered an error
194-
// in the future, but we can't do that now as Job Runner does use this now but knows nothing about the data
195-
// plane
196-
197-
if ($this->supportsDataPlanes && ($data['dataPlaneId'] ?? null)) {
198-
$dataPlaneConfig = $this->dataPlaneConfigRepository->fetchDataPlaneConfig($data['dataPlaneId']);
199-
$objectEncryptor = $dataPlaneConfig->getEncryption()->createEncryptor();
200-
} else {
201-
$objectEncryptor = $this->controlPlaneObjectEncryptor;
202-
}
203-
204-
return new Job($objectEncryptor, $this->storageClientFactory, $data);
205-
}
206-
207-
public function modifyJob(JobInterface $job, array $patchData): JobInterface
208-
{
209-
$data = $job->jsonSerialize();
210-
$data = array_replace_recursive($data, $patchData);
211-
212-
return $this->loadFromExistingJobData($data);
213-
}
214-
21576
/**
21677
* @param class-string<FullJobDefinition|NewJobDefinition> $validatorClass
21778
*/
218-
private function validateJobData(array $data, string $validatorClass): array
79+
protected function validateJobData(array $data, string $validatorClass): array
21980
{
22081
try {
22182
return (new $validatorClass())->processData($data);
22283
} catch (InvalidConfigurationException $e) {
22384
throw new ClientException($e->getMessage(), $e->getCode(), $e);
22485
}
22586
}
226-
227-
private function getJobType(array $data): string
228-
{
229-
if ((intval($data['parallelism']) > 0) || $data['parallelism'] === self::PARALLELISM_INFINITY) {
230-
return self::TYPE_ROW_CONTAINER;
231-
} else {
232-
if ($data['componentId'] === self::ORCHESTRATOR_COMPONENT) {
233-
if (isset($data['configData']['phaseId']) && (string) ($data['configData']['phaseId']) !== '') {
234-
return self::TYPE_PHASE_CONTAINER;
235-
} else {
236-
return self::TYPE_ORCHESTRATION_CONTAINER;
237-
}
238-
}
239-
}
240-
return self::TYPE_STANDARD;
241-
}
24287
}

0 commit comments

Comments
 (0)