Add logging of workflow steps

[MAILPOET-4463]
This commit is contained in:
John Oleksowicz
2022-08-25 14:34:47 -05:00
committed by Jan Jakeš
parent 2bcef7629c
commit cdeea173b9
8 changed files with 625 additions and 1 deletions

View File

@ -6,9 +6,11 @@ use Exception;
use MailPoet\Automation\Engine\Control\Steps\ActionStepRunner;
use MailPoet\Automation\Engine\Data\Step;
use MailPoet\Automation\Engine\Data\WorkflowRun;
use MailPoet\Automation\Engine\Data\WorkflowRunLog;
use MailPoet\Automation\Engine\Exceptions;
use MailPoet\Automation\Engine\Exceptions\InvalidStateException;
use MailPoet\Automation\Engine\Hooks;
use MailPoet\Automation\Engine\Storage\WorkflowRunLogStorage;
use MailPoet\Automation\Engine\Storage\WorkflowRunStorage;
use MailPoet\Automation\Engine\Storage\WorkflowStorage;
use MailPoet\Automation\Engine\WordPress;
@ -33,17 +35,27 @@ class StepHandler {
/** @var array<string, StepRunner> */
private $stepRunners;
/** @var WorkflowRunLogStorage */
private $workflowRunLogStorage;
/** @var Hooks */
private $hooks;
public function __construct(
ActionScheduler $actionScheduler,
ActionStepRunner $actionStepRunner,
Hooks $hooks,
WordPress $wordPress,
WorkflowRunStorage $workflowRunStorage,
WorkflowRunLogStorage $workflowRunLogStorage,
WorkflowStorage $workflowStorage
) {
$this->actionScheduler = $actionScheduler;
$this->actionStepRunner = $actionStepRunner;
$this->hooks = $hooks;
$this->wordPress = $wordPress;
$this->workflowRunStorage = $workflowRunStorage;
$this->workflowRunLogStorage = $workflowRunLogStorage;
$this->workflowStorage = $workflowStorage;
}
@ -108,7 +120,22 @@ class StepHandler {
$stepType = $step->getType();
if (isset($this->stepRunners[$stepType])) {
$log = $this->createWorkflowRunLog($workflowRun, $step, $args);
try {
$this->stepRunners[$stepType]->run($step, $workflow, $workflowRun);
$log->markCompleted();
} catch (Exception $e) {
$log->markFailed();
$log->addError($e);
throw $e;
} finally {
try {
$this->hooks->doWorkflowStepAfterRun($step, $log);
} catch (Exception $e) {
$log->addError($e);
}
$this->workflowRunLogStorage->updateWorkflowRunLog($log);
}
} else {
throw new InvalidStateException();
}
@ -137,4 +164,14 @@ class StepHandler {
// TODO: allow long-running steps (that are not done here yet)
}
private function createWorkflowRunLog(WorkflowRun $workflowRun, Step $step, array $args): WorkflowRunLog {
$log = new WorkflowRunLog($workflowRun->getId(), $step->getId(), $args);
$logId = $this->workflowRunLogStorage->createWorkflowRunLog($log);
$workflowRunLog = $this->workflowRunLogStorage->getWorkflowRunLog($logId);
if (!$workflowRunLog instanceof WorkflowRunLog) {
throw new InvalidStateException();
}
return $workflowRunLog;
}
}

View File

@ -0,0 +1,170 @@
<?php declare(strict_types = 1);
namespace MailPoet\Automation\Engine\Data;
use DateTimeImmutable;
use MailPoet\Automation\Engine\Utils\Json;
class WorkflowRunLog {
const STATUS_RUNNING = 'running';
const STATUS_COMPLETED = 'completed';
const STATUS_FAILED = 'failed';
/** @var int */
private $id;
/** @var int */
private $workflowRunId;
/** @var DateTimeImmutable */
private $createdAt;
/** @var DateTimeImmutable */
private $updatedAt;
/** @var DateTimeImmutable|null */
private $completedAt;
/** @var string */
private $status;
/** @var array */
private $errors;
/** @var array */
private $data;
/** @var string */
private $stepId;
/** @var array */
private $args;
public function __construct(
int $workflowRunId,
string $stepId,
array $args,
int $id = null
) {
$this->workflowRunId = $workflowRunId;
$this->stepId = $stepId;
$this->args = $args;
$this->status = self::STATUS_RUNNING;
if ($id) {
$this->id = $id;
}
$now = new DateTimeImmutable();
$this->createdAt = $now;
$this->updatedAt = $now;
$this->errors = [];
$this->data = [];
}
public function getId(): int {
return $this->id;
}
public function getWorkflowRunId(): int {
return $this->workflowRunId;
}
public function getStepId(): string {
return $this->stepId;
}
public function getStatus(): string {
return $this->status;
}
public function getArgs(): array {
return $this->args;
}
public function getErrors(): array {
return $this->errors;
}
public function getData(): array {
return $this->data;
}
/**
* @return DateTimeImmutable|null
*/
public function getCompletedAt() {
return $this->completedAt;
}
/**
* @param string $key
* @param mixed $value
* @return void
*/
public function setData(string $key, $value): void {
$this->data[$key] = $value;
}
public function getCreatedAt(): DateTimeImmutable {
return $this->createdAt;
}
public function getUpdatedAt(): DateTimeImmutable {
return $this->updatedAt;
}
public function toArray(): array {
return [
'workflow_run_id' => $this->workflowRunId,
'step_id' => $this->stepId,
'status' => $this->status,
'created_at' => $this->createdAt->format(DateTimeImmutable::W3C),
'updated_at' => $this->updatedAt->format(DateTimeImmutable::W3C),
'completed_at' => $this->completedAt ? $this->completedAt->format(DateTimeImmutable::W3C) : null,
'args' => Json::encode($this->args),
'errors' => Json::encode($this->errors),
'data' => Json::encode($this->data),
];
}
public function markCompleted(): void {
$this->status = self::STATUS_COMPLETED;
$this->completedAt = new DateTimeImmutable();
}
public function markFailed(): void {
$this->status = self::STATUS_FAILED;
}
public function addError(\Exception $exception, string $userFacingMessage = ''): void {
$error = [
'message' => $exception->getMessage(),
'exceptionClass' => get_class($exception),
'userFacingMessage' => $userFacingMessage,
'code' => $exception->getCode(),
'trace' => $exception->getTrace(),
];
$this->errors[] = $error;
}
public static function fromArray(array $data): self {
$workflowRunLog = new WorkflowRunLog((int)$data['workflow_run_id'], $data['step_id'], []);
$workflowRunLog->id = (int)$data['id'];
$workflowRunLog->status = $data['status'];
$workflowRunLog->errors = Json::decode($data['errors']);
$workflowRunLog->data = Json::decode($data['data']);
$workflowRunLog->args = Json::decode($data['args']);
$workflowRunLog->createdAt = new DateTimeImmutable($data['created_at']);
$workflowRunLog->updatedAt = new DateTimeImmutable($data['updated_at']);
if ($data['completed_at']) {
$workflowRunLog->completedAt = new DateTimeImmutable($data['completed_at']);
}
return $workflowRunLog;
}
}

View File

@ -4,6 +4,7 @@ namespace MailPoet\Automation\Engine;
use MailPoet\Automation\Engine\Data\Step;
use MailPoet\Automation\Engine\Data\Workflow;
use MailPoet\Automation\Engine\Data\WorkflowRunLog;
class Hooks {
/** @var WordPress */
@ -24,6 +25,8 @@ class Hooks {
public const WORKFLOW_BEFORE_SAVE = 'mailpoet/automation/workflow/before_save';
public const WORKFLOW_STEP_BEFORE_SAVE = 'mailpoet/automation/workflow/step/before_save';
public const WORKFLOW_RUN_LOG_AFTER_STEP_RUN = 'mailpoet/automation/workflow/step/after_run';
public const WORKFLOW_TEMPLATES = 'mailpoet/automation/workflow/templates';
public function doWorkflowBeforeSave(Workflow $workflow): void {
@ -37,4 +40,8 @@ class Hooks {
public function doWorkflowStepByKeyBeforeSave(Step $step): void {
$this->wordPress->doAction(self::WORKFLOW_STEP_BEFORE_SAVE . '/key=' . $step->getKey(), $step);
}
public function doWorkflowStepAfterRun(Step $step, WorkflowRunLog $workflowRunLog): void {
$this->wordPress->doAction(self::WORKFLOW_RUN_LOG_AFTER_STEP_RUN, $step, $workflowRunLog);
}
}

View File

@ -61,12 +61,29 @@ class Migrator {
PRIMARY KEY (id)
);
");
$this->runQuery("
CREATE TABLE {$this->prefix}workflow_run_logs (
id int(11) unsigned NOT NULL AUTO_INCREMENT,
workflow_run_id int(11) unsigned NOT NULL,
step_id varchar(255) NOT NULL,
status varchar(255) NOT NULL,
created_at timestamp NOT NULL,
updated_at timestamp NOT NULL,
completed_at timestamp NULL DEFAULT NULL,
args longtext,
errors longtext,
data longtext,
PRIMARY KEY (id)
);
");
}
public function deleteSchema(): void {
$this->removeOldSchema();
$this->runQuery("DROP TABLE IF EXISTS {$this->prefix}workflows");
$this->runQuery("DROP TABLE IF EXISTS {$this->prefix}workflow_runs");
$this->runQuery("DROP TABLE IF EXISTS {$this->prefix}workflow_run_logs");
$this->runQuery("DROP TABLE IF EXISTS {$this->prefix}workflow_versions");
// clean Action Scheduler data

View File

@ -0,0 +1,90 @@
<?php declare(strict_types = 1);
namespace MailPoet\Automation\Engine\Storage;
use MailPoet\Automation\Engine\Data\WorkflowRunLog;
use MailPoet\Automation\Engine\Exceptions;
use MailPoet\InvalidStateException;
use wpdb;
class WorkflowRunLogStorage {
/** @var string */
private $table;
/** @var wpdb */
private $wpdb;
public function __construct() {
global $wpdb;
$this->table = $wpdb->prefix . 'mailpoet_workflow_run_logs';
$this->wpdb = $wpdb;
}
public function createWorkflowRunLog(WorkflowRunLog $workflowRunLog): int {
$result = $this->wpdb->insert($this->table, $workflowRunLog->toArray());
if ($result === false) {
throw Exceptions::databaseError($this->wpdb->last_error);
}
return $this->wpdb->insert_id;
}
public function updateWorkflowRunLog(WorkflowRunLog $workflowRunLog): int {
$data = $workflowRunLog->toArray();
unset($data['id']);
$data['updated_at'] = (new \DateTimeImmutable())->format(\DateTimeImmutable::W3C);
$where = ['id' => $workflowRunLog->getId()];
$result = $this->wpdb->update($this->table, $data, $where);
if ($result === false) {
throw Exceptions::databaseError($this->wpdb->last_error);
}
return $result;
}
public function getWorkflowRunLog(int $id): ?WorkflowRunLog {
$table = esc_sql($this->table);
$query = $this->wpdb->prepare("SELECT * FROM $table WHERE id = %d", $id);
if (!is_string($query)) {
throw InvalidStateException::create();
}
$result = $this->wpdb->get_row($query, ARRAY_A);
if ($result) {
$data = (array)$result;
return WorkflowRunLog::fromArray($data);
}
return null;
}
/**
* @param int $workflowRunId
* @return WorkflowRunLog[]
* @throws InvalidStateException
*/
public function getLogsForWorkflowRun(int $workflowRunId): array {
$table = esc_sql($this->table);
$query = $this->wpdb->prepare("SELECT * FROM $table WHERE workflow_run_id = %d", $workflowRunId);
if (!is_string($query)) {
throw InvalidStateException::create();
}
$results = $this->wpdb->get_results($query, ARRAY_A);
if (!is_array($results)) {
throw InvalidStateException::create();
}
if ($results) {
return array_map(function($data) {
return WorkflowRunLog::fromArray($data);
}, $results);
}
return [];
}
}

View File

@ -124,6 +124,7 @@ class ContainerConfigurator implements IContainerConfigurator {
$container->autowire(\MailPoet\Automation\Engine\Migrations\Migrator::class)->setPublic(true);
$container->autowire(\MailPoet\Automation\Engine\Registry::class)->setPublic(true);
$container->autowire(\MailPoet\Automation\Engine\Storage\WorkflowRunStorage::class)->setPublic(true);
$container->autowire(\MailPoet\Automation\Engine\Storage\WorkflowRunLogStorage::class)->setPublic(true);
$container->autowire(\MailPoet\Automation\Engine\Storage\WorkflowTemplateStorage::class)->setPublic(true);
$container->autowire(\MailPoet\Automation\Engine\Storage\WorkflowStorage::class)->setPublic(true);
$container->autowire(\MailPoet\Automation\Engine\WordPress::class)->setPublic(true);

View File

@ -0,0 +1,241 @@
<?php declare(strict_types=1);
namespace MailPoet\Test\Automation\Engine\Data;
use MailPoet\Automation\Engine\Control\StepHandler;
use MailPoet\Automation\Engine\Data\Step;
use MailPoet\Automation\Engine\Data\Workflow;
use MailPoet\Automation\Engine\Data\WorkflowRun;
use MailPoet\Automation\Engine\Data\WorkflowRunLog;
use MailPoet\Automation\Engine\Hooks;
use MailPoet\Automation\Engine\Registry;
use MailPoet\Automation\Engine\Storage\WorkflowRunLogStorage;
use MailPoet\Automation\Engine\Storage\WorkflowRunStorage;
use MailPoet\Automation\Engine\Storage\WorkflowStorage;
use MailPoet\Automation\Engine\Workflows\Action;
use MailPoet\Util\Security;
use MailPoet\Validator\Builder;
use MailPoet\Validator\Schema\ObjectSchema;
use MailPoet\WP\Functions as WPFunctions;
class WorkflowRunLogTest extends \MailPoetTest {
/** @var WPFunctions */
private $wp;
/** @var StepHandler */
private $stepHandler;
/** @var WorkflowRunStorage */
private $workflowRunStorage;
/** @var WorkflowStorage */
private $workflowStorage;
/** @var Registry */
private $registry;
/** @var WorkflowRunLogStorage */
private $workflowRunLogStorage;
public function _before() {
parent::_before();
$this->workflowStorage = $this->diContainer->get(WorkflowStorage::class);
$this->workflowRunStorage = $this->diContainer->get(WorkflowRunStorage::class);
$this->workflowRunLogStorage = $this->diContainer->get(WorkflowRunLogStorage::class);
$this->stepHandler = $this->diContainer->get(StepHandler::class);
$this->registry = $this->diContainer->get(Registry::class);
$this->wp = new WPFunctions();
}
public function testItAllowsSettingData(): void {
$log = new WorkflowRunLog(1, 'step-id', []);
$this->assertSame([], $log->getData());
$log->setData('key', 'value');
$data = $log->getData();
$this->assertCount(1, $data);
$this->assertSame('value', $data['key']);
}
public function testItGetsExposedViaAction(): void {
$this->wp->addAction(Hooks::WORKFLOW_RUN_LOG_AFTER_STEP_RUN, function(Step $step, WorkflowRunLog $log) {
$log->setData('test', 'value');
}, 10, 2);
$workflowRunLogs = $this->getLogsForAction();
expect($workflowRunLogs)->count(1);
$log = $workflowRunLogs[0];
expect($log->getData()['test'])->equals('value');
}
public function testBadActionIntegrationsCannotDerailStepFromRunning() {
$this->wp->addAction(Hooks::WORKFLOW_RUN_LOG_AFTER_STEP_RUN, function(Step $step, WorkflowRunLog $log) {
throw new \Exception('bad integration');
}, 9, 2);
$workflowRunLogs = $this->getLogsForAction();
expect($workflowRunLogs)->count(1);
$log = $workflowRunLogs[0];
expect($log->getStatus())->equals(WorkflowRunLog::STATUS_COMPLETED);
}
public function testItStoresWorkflowRunAndStepIdsCorrectly() {
$testAction = $this->getRegisteredTestAction();
$actionStep = new Step('action-step-id', Step::TYPE_ACTION, $testAction->getKey());
$workflow = new Workflow('test_workflow', [$actionStep], new \WP_User());
$workflowId = $this->workflowStorage->createWorkflow($workflow);
// Reload to get additional data post-save
$workflow = $this->workflowStorage->getWorkflow($workflowId);
$this->assertInstanceOf(Workflow::class, $workflow);
$workflowRun = new WorkflowRun($workflowId, $workflow->getVersionId(), 'trigger-key', []);
$workflowRunId = $this->workflowRunStorage->createWorkflowRun($workflowRun);
$this->stepHandler->handle([
'workflow_run_id' => $workflowRunId,
'step_id' => 'action-step-id'
]);
$log = $this->workflowRunLogStorage->getLogsForWorkflowRun($workflowRunId)[0];
expect($log->getWorkflowRunId())->equals($workflowRunId);
expect($log->getStepId())->equals('action-step-id');
}
public function testItLogsCompletedStatusCorrectly(): void {
$workflowRunLogs = $this->getLogsForAction(function() {
return true;
});
expect($workflowRunLogs)->count(1);
$log = $workflowRunLogs[0];
expect($log->getStatus())->equals('completed');
}
public function testItAddsCompletedAtTimestampAfterRunningSuccessfully(): void {
$this->wp->addAction(Hooks::WORKFLOW_RUN_LOG_AFTER_STEP_RUN, function(Step $step, WorkflowRunLog $log) {
expect($log->getCompletedAt())->null();
}, 10, 2);
$workflowRunLogs = $this->getLogsForAction(function() {
return true;
});
expect($workflowRunLogs)->count(1);
$log = $workflowRunLogs[0];
expect($log->getCompletedAt())->isInstanceOf(\DateTimeImmutable::class);
}
public function testItLogsFailedStatusCorrectly(): void {
$workflowRunLogs = $this->getLogsForAction(function() {
throw new \Exception('error');
});
expect($workflowRunLogs)->count(1);
$log = $workflowRunLogs[0];
expect($log->getStatus())->equals('failed');
}
public function testItDoesNotHaveCompletedAtIfItFailsToRun(): void {
$workflowRunLogs = $this->getLogsForAction(function() {
throw new \Exception('error');
});
expect($workflowRunLogs)->count(1);
$log = $workflowRunLogs[0];
expect($log->getCompletedAt())->null();
}
public function testItIncludesErrorOnFailure(): void {
$workflowRunLogs = $this->getLogsForAction(function() {
throw new \Exception('error', 12345);
});
expect($workflowRunLogs)->count(1);
$log = $workflowRunLogs[0];
expect($log->getErrors())->count(1);
$error = $log->getErrors()[0];
expect($error['message'])->equals('error');
expect($error['code'])->equals(12345);
expect($error['exceptionClass'])->equals('Exception');
expect($error['trace'])->array();
expect(count($error['trace']))->greaterThan(0);
}
public function testItLogsStepArgs(): void {
$log = $this->getLogsForAction()[0];
expect($log->getArgs())->count(2);
expect(array_keys($log->getArgs()))->equals(['workflow_run_id', 'step_id']);
}
public function _after() {
global $wpdb;
$sql = 'truncate ' . $wpdb->prefix . 'mailpoet_workflow_run_logs';
$wpdb->query($sql);
$sql = 'truncate ' . $wpdb->prefix . 'mailpoet_workflows';
$wpdb->query($sql);
$sql = 'truncate ' . $wpdb->prefix . 'mailpoet_workflow_versions';
$wpdb->query($sql);
$sql = 'truncate ' . $wpdb->prefix . 'mailpoet_workflow_runs';
$wpdb->query($sql);
}
private function getLogsForAction($callback = null) {
$testAction = $this->getRegisteredTestAction($callback);
$actionStep = new Step('action-step-id', Step::TYPE_ACTION, $testAction->getKey());
$workflow = new Workflow('test_workflow', [$actionStep], new \WP_User());
$workflowId = $this->workflowStorage->createWorkflow($workflow);
// Reload to get additional data post-save
$workflow = $this->workflowStorage->getWorkflow($workflowId);
$this->assertInstanceOf(Workflow::class, $workflow);
$workflowRun = new WorkflowRun($workflowId, $workflow->getVersionId(), 'trigger-key', []);
$workflowRunId = $this->workflowRunStorage->createWorkflowRun($workflowRun);
try {
$this->stepHandler->handle([
'workflow_run_id' => $workflowRunId,
'step_id' => 'action-step-id'
]);
} catch (\Exception $e) {
// allow exceptions so we can test failure states
}
return $this->workflowRunLogStorage->getLogsForWorkflowRun($workflowRunId);
}
private function getRegisteredTestAction($callback = null) {
if ($callback === null) {
$callback = function() { return true; };
}
$action = new TestAction();
$action->setCallback($callback);
$this->registry->addAction($action);
return $action;
}
}
class TestAction implements Action {
private $callback;
private $key;
public function __construct() {
$this->key = Security::generateRandomString(10);
}
public function setCallback($callback) {
$this->callback = $callback;
}
public function isValid(array $subjects, Step $step, Workflow $workflow): bool {
return true;
}
public function run(Workflow $workflow, WorkflowRun $workflowRun, Step $step): void {
if ($this->callback) {
($this->callback)($workflow, $workflowRun, $step);
}
}
public function getKey(): string {
return $this->key;
}
public function getName(): string {
return 'Test Action';
}
public function getArgsSchema(): ObjectSchema {
return Builder::object();
}
}

View File

@ -0,0 +1,61 @@
<?php
namespace MailPoet\Test\Automation\Engine\Storage;
use MailPoet\Automation\Engine\Data\WorkflowRunLog;
use MailPoet\Automation\Engine\Storage\WorkflowRunLogStorage;
class WorkflowRunLogStorageTest extends \MailPoetTest {
/** @var WorkflowRunLogStorage */
private $storage;
public function _before() {
$this->storage = $this->diContainer->get(WorkflowRunLogStorage::class);
}
public function testItSavesAndRetrievesAsExpected() {
$log = new WorkflowRunLog(1, 'step-id', []);
$log->setData('key', 'value');
$log->setData('key2', ['arrayData']);
$preSave = $log->toArray();
$id = $this->storage->createWorkflowRunLog($log);
$fromDatabase = $this->storage->getWorkflowRunLog($id);
$this->assertInstanceOf(WorkflowRunLog::class, $fromDatabase);
expect($preSave)->equals($fromDatabase->toArray());
}
public function testUpdatingLogUpdatesUpdatedAtTimestamp() {
$log = new WorkflowRunLog(1, 'step-id', []);
$reflectionClass = new \ReflectionClass(WorkflowRunLog::class);
$updatedAt = $reflectionClass->getProperty('updatedAt');
$updatedAt->setAccessible(true);
$updatedAt->setValue($log, new \DateTimeImmutable('2022-09-07'));
$id = $this->storage->createWorkflowRunLog($log);
$log = $this->storage->getWorkflowRunLog($id);
$this->assertInstanceOf(WorkflowRunLog::class, $log);
$originalUpdatedAt = $log->getUpdatedAt();
$log->setData('key', 'value');
$this->assertInstanceOf(WorkflowRunLog::class, $log);
$this->storage->updateWorkflowRunLog($log);
$fromDatabase = $this->storage->getWorkflowRunLog($id);
$this->assertInstanceOf(WorkflowRunLog::class, $fromDatabase);
expect($fromDatabase->getUpdatedAt())->greaterThan($originalUpdatedAt);
}
public function testItStoresErrors() {
$log = new WorkflowRunLog(1, 'step-id', []);
$log->addError(new \Exception('test'));
$id = $this->storage->createWorkflowRunLog($log);
$log = $this->storage->getWorkflowRunLog($id);
$this->assertInstanceOf(WorkflowRunLog::class, $log);
$errors = $log->getErrors();
expect($errors)->count(1);
}
public function _after() {
global $wpdb;
$sql = 'truncate ' . $wpdb->prefix . 'mailpoet_workflow_run_logs';
$wpdb->query($sql);
}
}