Refactoring: extract process handling from async job.
This commit is contained in:
parent
6524fade3e
commit
e74c464857
@ -2,35 +2,20 @@
|
||||
|
||||
namespace App\Jobs;
|
||||
|
||||
use App\Services\ProcessStatus;
|
||||
use Illuminate\Support\Facades\DB;
|
||||
use App\Services\RemoteProcess\RemoteProcess;
|
||||
use Illuminate\Bus\Queueable;
|
||||
use Illuminate\Contracts\Queue\ShouldBeUnique;
|
||||
use Illuminate\Contracts\Queue\ShouldQueue;
|
||||
use Illuminate\Foundation\Bus\Dispatchable;
|
||||
use Illuminate\Process\InvokedProcess;
|
||||
use Illuminate\Process\ProcessResult;
|
||||
use Illuminate\Queue\InteractsWithQueue;
|
||||
use Illuminate\Queue\SerializesModels;
|
||||
use Illuminate\Support\Facades\Process;
|
||||
use Spatie\Activitylog\Contracts\Activity;
|
||||
|
||||
class ExecuteCoolifyProcess implements ShouldQueue
|
||||
{
|
||||
use Dispatchable, InteractsWithQueue, Queueable, SerializesModels;
|
||||
|
||||
protected $throttleIntervalMS = 500;
|
||||
|
||||
protected $timeStart;
|
||||
|
||||
protected $currentTime;
|
||||
|
||||
protected $lastWriteAt = 0;
|
||||
|
||||
protected string $stdOutIncremental = '';
|
||||
|
||||
protected string $stdErrIncremental = '';
|
||||
|
||||
/**
|
||||
* Create a new job instance.
|
||||
*/
|
||||
@ -41,92 +26,12 @@ public function __construct(
|
||||
/**
|
||||
* Execute the job.
|
||||
*/
|
||||
public function handle(): ProcessResult
|
||||
public function handle(): void
|
||||
{
|
||||
$this->timeStart = hrtime(true);
|
||||
|
||||
$user = $this->activity->getExtraProperty('user');
|
||||
$destination = $this->activity->getExtraProperty('destination');
|
||||
$port = $this->activity->getExtraProperty('port');
|
||||
$command = $this->activity->getExtraProperty('command');
|
||||
|
||||
$delimiter = 'EOF-COOLIFY-SSH';
|
||||
|
||||
$sshCommand = 'ssh '
|
||||
. '-o StrictHostKeyChecking=no -o UserKnownHostsFile=/dev/null '
|
||||
. '-o PasswordAuthentication=no '
|
||||
. '-o RequestTTY=no '
|
||||
// Quiet mode. Causes most warning and diagnostic messages to be suppressed.
|
||||
// Errors are still out put. This is to silence for example, that warning
|
||||
// Permanently added <host and key type> to the list of known hosts.
|
||||
. '-q '
|
||||
. "-p {$port} "
|
||||
. "{$user}@{$destination} "
|
||||
. " 'bash -se' << \\$delimiter" . PHP_EOL
|
||||
. $command . PHP_EOL
|
||||
. $delimiter;
|
||||
|
||||
$process = Process::start($sshCommand, $this->handleOutput(...));
|
||||
|
||||
$processResult = $process->wait();
|
||||
|
||||
$status = match ($processResult->exitCode()) {
|
||||
0 => ProcessStatus::FINISHED,
|
||||
default => ProcessStatus::ERROR,
|
||||
};
|
||||
|
||||
$this->activity->properties = $this->activity->properties->merge([
|
||||
'exitCode' => $processResult->exitCode(),
|
||||
'stdout' => $processResult->output(),
|
||||
'stderr' => $processResult->errorOutput(),
|
||||
'status' => $status,
|
||||
$remoteProcess = resolve(RemoteProcess::class, [
|
||||
'activity' => $this->activity,
|
||||
]);
|
||||
|
||||
$this->activity->save();
|
||||
|
||||
return $processResult;
|
||||
}
|
||||
|
||||
protected function handleOutput(string $type, string $output)
|
||||
{
|
||||
$this->currentTime = $this->elapsedTime();
|
||||
|
||||
if ($type === 'out') {
|
||||
$this->stdOutIncremental .= $output;
|
||||
} else {
|
||||
$this->stdErrIncremental .= $output;
|
||||
}
|
||||
|
||||
$this->activity->description .= $output;
|
||||
|
||||
if ($this->isAfterLastThrottle()) {
|
||||
// Let's write to database.
|
||||
DB::transaction(function () {
|
||||
$this->activity->save();
|
||||
$this->lastWriteAt = $this->currentTime;
|
||||
});
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* Decides if it's time to write again to database.
|
||||
*
|
||||
* @return bool
|
||||
*/
|
||||
protected function isAfterLastThrottle()
|
||||
{
|
||||
// If DB was never written, then we immediately decide we have to write.
|
||||
if ($this->lastWriteAt === 0) {
|
||||
return true;
|
||||
}
|
||||
|
||||
return ($this->currentTime - $this->throttleIntervalMS) > $this->lastWriteAt;
|
||||
}
|
||||
|
||||
protected function elapsedTime(): int
|
||||
{
|
||||
$timeMs = (hrtime(true) - $this->timeStart) / 1_000_000;
|
||||
|
||||
return intval($timeMs);
|
||||
$remoteProcess();
|
||||
}
|
||||
}
|
||||
|
@ -27,17 +27,13 @@ public function __construct(
|
||||
'command' => $this->command,
|
||||
'status' => ProcessStatus::HOLDING,
|
||||
])
|
||||
->log("Awaiting to start command...\n\n");
|
||||
->log("Awaiting command to start...\n\n");
|
||||
}
|
||||
|
||||
public function __invoke(): Activity|ProcessResult
|
||||
public function __invoke(): Activity
|
||||
{
|
||||
$job = new ExecuteCoolifyProcess($this->activity);
|
||||
|
||||
if (app()->environment('testing')) {
|
||||
return $job->handle();
|
||||
}
|
||||
|
||||
dispatch($job);
|
||||
|
||||
return $this->activity;
|
||||
|
121
app/Services/RemoteProcess/RemoteProcess.php
Normal file
121
app/Services/RemoteProcess/RemoteProcess.php
Normal file
@ -0,0 +1,121 @@
|
||||
<?php
|
||||
|
||||
namespace App\Services\RemoteProcess;
|
||||
|
||||
use App\Services\ProcessStatus;
|
||||
use Illuminate\Process\ProcessResult;
|
||||
use Illuminate\Support\Facades\DB;
|
||||
use Illuminate\Support\Facades\Process;
|
||||
use Spatie\Activitylog\Contracts\Activity;
|
||||
|
||||
class RemoteProcess
|
||||
{
|
||||
protected $timeStart;
|
||||
|
||||
protected $currentTime;
|
||||
|
||||
protected $lastWriteAt = 0;
|
||||
|
||||
protected $throttleIntervalMS = 500;
|
||||
|
||||
protected string $stdOutIncremental = '';
|
||||
|
||||
protected string $stdErrIncremental = '';
|
||||
|
||||
/**
|
||||
* Create a new job instance.
|
||||
*/
|
||||
public function __construct(
|
||||
public Activity $activity,
|
||||
){}
|
||||
|
||||
public function __invoke(): ProcessResult
|
||||
{
|
||||
$this->timeStart = hrtime(true);
|
||||
|
||||
$processResult = Process::run($this->getCommand(), $this->handleOutput(...));
|
||||
|
||||
$status = match ($processResult->exitCode()) {
|
||||
0 => ProcessStatus::FINISHED,
|
||||
default => ProcessStatus::ERROR,
|
||||
};
|
||||
|
||||
$this->activity->properties = $this->activity->properties->merge([
|
||||
'exitCode' => $processResult->exitCode(),
|
||||
'stdout' => $processResult->output(),
|
||||
'stderr' => $processResult->errorOutput(),
|
||||
'status' => $status,
|
||||
]);
|
||||
|
||||
$this->activity->save();
|
||||
|
||||
return $processResult;
|
||||
}
|
||||
|
||||
protected function getCommand(): string
|
||||
{
|
||||
$user = $this->activity->getExtraProperty('user');
|
||||
$destination = $this->activity->getExtraProperty('destination');
|
||||
$port = $this->activity->getExtraProperty('port');
|
||||
$command = $this->activity->getExtraProperty('command');
|
||||
|
||||
$delimiter = 'EOF-COOLIFY-SSH';
|
||||
|
||||
return 'ssh '
|
||||
. '-o StrictHostKeyChecking=no -o UserKnownHostsFile=/dev/null '
|
||||
. '-o PasswordAuthentication=no '
|
||||
. '-o RequestTTY=no '
|
||||
// Quiet mode. Causes most warning and diagnostic messages to be suppressed.
|
||||
// Errors are still out put. This is to silence for example, that warning
|
||||
// Permanently added <host and key type> to the list of known hosts.
|
||||
. '-q '
|
||||
. "-p {$port} "
|
||||
. "{$user}@{$destination} "
|
||||
. " 'bash -se' << \\$delimiter" . PHP_EOL
|
||||
. $command . PHP_EOL
|
||||
. $delimiter;
|
||||
}
|
||||
|
||||
protected function handleOutput(string $type, string $output)
|
||||
{
|
||||
$this->currentTime = $this->elapsedTime();
|
||||
|
||||
if ($type === 'out') {
|
||||
$this->stdOutIncremental .= $output;
|
||||
} else {
|
||||
$this->stdErrIncremental .= $output;
|
||||
}
|
||||
|
||||
$this->activity->description .= $output;
|
||||
|
||||
if ($this->isAfterLastThrottle()) {
|
||||
// Let's write to database.
|
||||
DB::transaction(function () {
|
||||
$this->activity->save();
|
||||
$this->lastWriteAt = $this->currentTime;
|
||||
});
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* Determines if it's time to write again to database.
|
||||
*
|
||||
* @return bool
|
||||
*/
|
||||
protected function isAfterLastThrottle()
|
||||
{
|
||||
// If DB was never written, then we immediately decide we have to write.
|
||||
if ($this->lastWriteAt === 0) {
|
||||
return true;
|
||||
}
|
||||
|
||||
return ($this->currentTime - $this->throttleIntervalMS) > $this->lastWriteAt;
|
||||
}
|
||||
|
||||
protected function elapsedTime(): int
|
||||
{
|
||||
$timeMs = (hrtime(true) - $this->timeStart) / 1_000_000;
|
||||
|
||||
return intval($timeMs);
|
||||
}
|
||||
}
|
@ -1,7 +1,6 @@
|
||||
<?php
|
||||
|
||||
use App\Services\CoolifyProcess;
|
||||
use Illuminate\Process\ProcessResult;
|
||||
use Spatie\Activitylog\Contracts\Activity;
|
||||
|
||||
if (! function_exists('coolifyProcess')) {
|
||||
@ -10,7 +9,7 @@
|
||||
* Run a Coolify Process, which SSH's into a machine to run the command(s).
|
||||
*
|
||||
*/
|
||||
function coolifyProcess($command, $destination): Activity|ProcessResult
|
||||
function coolifyProcess($command, $destination): Activity
|
||||
{
|
||||
$process = resolve(CoolifyProcess::class, [
|
||||
'destination' => $destination,
|
||||
|
@ -56,8 +56,7 @@ class="bg-indigo-500 rounded py-2 px-4 disabled:bg-gray-300"
|
||||
flex-direction: column-reverse;
|
||||
"
|
||||
placeholder="Build output"
|
||||
>
|
||||
{{ data_get($activity, 'description') }}
|
||||
>{{ data_get($activity, 'description') }}
|
||||
</pre>
|
||||
|
||||
<div>
|
||||
|
@ -13,20 +13,21 @@
|
||||
$host = 'testing-host';
|
||||
|
||||
// Assert there's no containers start with coolify_test_*
|
||||
$processResult = coolifyProcess($areThereCoolifyTestContainers, $host);
|
||||
$containers = Output::containerList($processResult->output());
|
||||
$activity = coolifyProcess($areThereCoolifyTestContainers, $host);
|
||||
ray($activity);
|
||||
$containers = Output::containerList($activity->getExtraProperty('stdout'));
|
||||
expect($containers)->toBeEmpty();
|
||||
|
||||
// start a container nginx -d --name = $containerName
|
||||
$processResult = coolifyProcess("docker run -d --name {$containerName} nginx", $host);
|
||||
expect($processResult->successful())->toBeTrue();
|
||||
$activity = coolifyProcess("docker run -d --name {$containerName} nginx", $host);
|
||||
expect($activity->getExtraProperty('exitCode'))->toBe(0);
|
||||
|
||||
// docker ps name = $container
|
||||
$processResult = coolifyProcess($areThereCoolifyTestContainers, $host);
|
||||
$containers = Output::containerList($processResult->output());
|
||||
$activity = coolifyProcess($areThereCoolifyTestContainers, $host);
|
||||
$containers = Output::containerList($activity->getExtraProperty('stdout'));
|
||||
expect($containers->where('Names', $containerName)->count())->toBe(1);
|
||||
|
||||
// Stop testing containers
|
||||
$processResult = coolifyProcess("docker stop $(docker ps --filter='name={$coolifyNamePrefix}*' -q)", $host);
|
||||
expect($processResult->successful())->toBeTrue();
|
||||
$activity = coolifyProcess("docker stop $(docker ps --filter='name={$coolifyNamePrefix}*' -q)", $host);
|
||||
expect($activity->getExtraProperty('exitCode'))->toBe(0);
|
||||
});
|
||||
|
Loading…
Reference in New Issue
Block a user