同じ API で複数のタスクを実行する必要があります。API はレート制限されているため、ロックアウトされないように呼び出し回数 (100 x 120 秒) を調整する必要があります。各ロックは 30 分間続き、チーム全体をフリーズさせ、恐ろしい問題を引き起こす可能性があります。
システムは Laravel 上に構築されており、キューを使用してこの API 呼び出しを行っています。必要な最初の作業は、システムが毎晩 1 回、API に対して約 950 回の呼び出しを行い、API の内容をデータベースと同期することです。2 番目と 3 番目のジョブはオンデマンドで、ユーザーが必要とするたびに、それぞれ約 100 から 200 の呼び出しを実行する必要がある場合があります。(複数のユーザーが 1 日中同時にこれを必要とする可能性があるため、キューの列によりレート制限を超えないようにする必要があります)。
スーパーバイザーで 1 人のワーカーだけですべてをセットアップしましたが、問題なく動作しているように見えました。しかし、その後、すべてのジョブをオンデマンドで呼び出しているわけではないことに気付きました (2 番目と 3 番目の種類)。ユーザーが 8 つのことに対してアクションを実行する必要がある場合、それらはキュー ラインに表示されますが、2 つまたは 3 つしか実行されません。残りは取得されます。拾ったり、走らせたり、失敗したりしませんでした。これは、ワーカーの数を 8 に増やすことである程度修正されました。これを行った後、ユーザーが 8 つのアクションを実行する必要がある場合は機能しましたが、毎晩 900 のタスクが失敗し始めました。
また、8 つのジョブで機能することにも気付きましたが、ユーザーに最大 50 のタスクを移動させると、これも失敗します。
/etc/supervisor/conf.d のスーパーバイザー構成:
[unix_http_server]
file = /tmp/supervisor.sock
chmod = 0777
chown= ubuntu:ubuntu
[supervisord]
logfile = /home/ubuntu/otmas/supervisord.log
logfile_maxbytes = 50MB
logfile_backups=10
loglevel = info
pidfile = /tmp/supervisord.pid
nodaemon = false
minfds = 1024
minprocs = 200
umask = 022
user = ubuntu
identifier = supervisor
directory = /tmp
nocleanup = true
childlogdir = /tmp
strip_ansi = false
[rpcinterface:supervisor]
supervisor.rpcinterface_factory = supervisor.rpcinterface:make_main_rpcinterface
[supervisorctl]
serverurl = unix:///tmp/supervisor.sock
[program:laravel-worker]
process_name=%(program_name)s_%(process_num)02d
command=php /home/ubuntu/www/artisan horizon
autostart=true
autorestart=true
user=ubuntu
numprocs=8
redirect_stderr=true
stdout_logfile=/home/ubuntu/otmas/worker.log
私たちのキュー設定:
'connections' => [
'redis' => [
'driver' => 'redis',
'connection' => 'default',
'queue' => 'default',
'retry_after' => 460,
'block_for' => 140,
],
],
私たちの地平線設定:
'environments' => [
'production' => [
'supervisor-1' => [
'connection' => 'redis',
'queue' => ['default'],
'balance' => 'simple',
'processes' => 10,
'tries' => 3,
],
],
'local' => [
'supervisor-1' => [
'connection' => 'redis',
'queue' => ['default'],
'balance' => 'simple',
'processes' => 3,
'tries' => 3,
],
],
'develop' => [
'supervisor-1' => [
'connection' => 'redis',
'queue' => ['default','email'],
'balance' => 'auto',
'processes' => 3,
'tries' => 3,
'timeout' => 3,
'delay' => 3,
],
],
],
database.php の redis 構成:
'redis' => [
'client' => 'predis',
'default' => [
'host' => env('REDIS_HOST', '127.0.0.1'),
'password' => env('REDIS_PASSWORD', null),
'port' => env('REDIS_PORT', 6379),
'database' => env('REDIS_DB', 0),
'read_write_timeout' => -1,
],
'cache' => [
'host' => env('REDIS_HOST', '127.0.0.1'),
'password' => env('REDIS_PASSWORD', null),
'port' => env('REDIS_PORT', 6379),
'database' => env('REDIS_CACHE_DB', 1),
'read_write_timeout' => -1,
],
],
毎晩の仕事:
class GetProjectTasks implements ShouldQueue
{
use Dispatchable, InteractsWithQueue, Queueable, SerializesModels;
protected $project;
public $tries = 540;
public $timeout = 15;
public function __construct(ZohoProject $project)
{
$this->project = $project;
}
public function handle()
{
\Log::info("before redis get tasks for project: ".$this->project->id);
try {
$client = new GuzzleClient();
$accessToken = Setting::where('name','zoho_access_token')->get()->first();
$requestHeaders = [
'cache-control' => 'no-cache',
'Authorization' => 'Bearer '.$accessToken->value
];
$requestOptions = [
'headers' => $requestHeaders
];
$zohoTasksRestApi = Setting::where('name','zoho_projectapi_restapi')->get()->first()->value;
$project = $this->project->id;
} catch(Exception $e) {
Log::notice('Exception caught');
throw $e;
}
Redis::throttle('zohoprojecttasksget')->allow(85)->every(120)->then(function () use ($client, $zohoTasksRestApi, $portalId, $project, $requestOptions) {
try {
$zohoTasksResponse = $client->request('GET', $zohoTasksRestApi.'portal/'.$portalId.'/projects/'.$project.'/tasks/', $requestOptions);
$zohoTasksResult = json_decode((string) $zohoTasksResponse->getBody(), true);
if($zohoTasksResult) {
foreach ($zohoTasksResult['tasks'] as $key => $task) {
if (array_has($task, 'start_date') && array_has($task, 'end_date')) {
$taskStartDate = \Carbon\Carbon::createFromTimestamp((int)$task['start_date_long']/1000, 'America/Santiago');
$taskEndDate = \Carbon\Carbon::createFromTimestamp((int)$task['end_date_long']/1000, 'America/Santiago');
$zohoTask = ZohoTask::updateOrCreate(['id' => $task['id']], [
'name' => $task['name'],
'zoho_project_id' => $project,
'registry' => json_encode($task),
'start_date' => $taskStartDate,
'end_date' => $taskEndDate
]);
if(array_has($task, ['details.owners'])) {
if($task['details']['owners'][0]['name'] != 'Unassigned') {
foreach ($task['details']['owners'] as $key => $owner) {
$user = ZohoUser::find($owner['id']);
$zohoTask->assignees()->save($user);
}
}
}
} else {
$zohoTask = ZohoTask::updateOrCreate(['id' => $task['id']], [
'name' => $task['name'],
'zoho_project_id' => $project,
'registry' => json_encode($task),
]);
}
}
}
} catch(Exception $e) {
Log::notice('Exception caught');
throw $e;
}
}, function () use ($project) {
return $this->release(85);
});
}
}
オンデマンド ジョブの例:
class ReplaceTask implements ShouldQueue
{
use Dispatchable, InteractsWithQueue, Queueable, SerializesModels;
protected $task, $taskToReplace;
public $tries = 140;
public $timeout = 15;
public function __construct(ZohoTask $task, ZohoTask $taskToReplace)
{
$this->task = $task;
$this->taskToReplace = $taskToReplace;
}
public function handle()
{
\Log::info("before redis replace task: ".$this->taskToReplace.' with task:: '.$this->task);
try {
$client = new GuzzleClient();
$accessToken = Setting::where('name','zoho_access_token')->get()->first();
$requestHeaders = [
'cache-control' => 'no-cache',
'Authorization' => 'Bearer '.$accessToken->value
];
$ZohoProjectsRestApi = Setting::where('name','zoho_projectapi_restapi')->get()->first()->value;
$task = $this->task;
$taskToReplace = $this->taskToReplace;
$project = $task->project;
} catch(Exception $e) {
Log::notice('Exception caught');
throw $e;
}
\Log::info("before redis task: ".$task->id.' to replace '.$taskToReplace->id.' Project: '.$project->id);
Redis::throttle('zohoupdatetask')->allow(50)->every(120)->then(function () use ($client, $ZohoProjectsRestApi, $portalId, $project, $requestHeaders, $task, $taskToReplace) {
try {
$assignee = $taskToReplace->assignees->last();
$assignResponse = $client->request('POST', $ZohoProjectsRestApi.'portal/'.$portalId.'/projects/'.$project->id.'/tasks/'.$task->id.'/', [
'headers' => $requestHeaders,
'form_params' => [
'start_date' => $taskToReplace->start_date->format('m-d-Y'),
'start_time' => $taskToReplace->start_date->format('h:i A'),
'person_responsible' => $assignee->id
]
]);
\Log::info($assignResponse->getBody());
if($assignResponse->getStatusCode() == 200) {
$task->assignees()->detach();
$assignResult = json_decode((string) $assignResponse->getBody(), true);
$taskRegistry = $assignResult['tasks'][0];
$taskStartDate = \Carbon\Carbon::createFromTimestamp((int)$taskRegistry['start_date_long']/1000, 'America/Santiago');
$taskEndDate = \Carbon\Carbon::createFromTimestamp((int)$taskRegistry['end_date_long']/1000, 'America/Santiago');
$task->start_date = $taskStartDate;
$task->end_date = $taskEndDate;
$task->registry = json_encode($taskRegistry);
$task->assignees()->save($assignee);
$task->save();
\Log::info("Task Ok move");
}
} catch(Exception $e) {
Log::notice('Exception caught');
throw $e;
}
}, function () use ($project, $task, $taskToReplace) {
\Log::info("Task hit throttle");
return $this->release(120);
});
\Log::info("after redis");
}
}
Horizon タスク: (保留中のタスクは実行もピックアップも何も行われません) タスクはキューに表示されますが、実行も失敗もしないようです。
