From 3300724acc7c0af2364f68c174b4324b76b54337 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Micha=C5=82=20Mleczko?= Date: Sun, 9 Aug 2026 01:50:22 +0200 Subject: [PATCH] perf: batch log/metric delivery instead of one blocking HTTP call per event Every logger()->info()/metrics()->set() call fired its own synchronous Guzzle request with a 5s timeout. Under high log/metric volume, or when the logdash API is slow/unreachable, this serialized directly onto the request path and could stall the host app for seconds per call - exactly the TODO the code already flagged (queue/retry/batching). - Add RequestQueue: buffers items and flushes on batch size, a lazy time interval, or process/request shutdown, with bounded retry. - HttpLogSync now posts buffered logs to POST /logs/batch (the endpoint the Node SDK already uses) instead of one POST /logs per line. - Metrics now dispatches queued updates concurrently via a Guzzle Pool (curl_multi) instead of sequentially; failures stay isolated per-item and are never retried, since `mutate` isn't idempotent. - Split connect_timeout (2s) from the overall timeout (5s) so a fully unreachable host fails fast instead of hanging the connect phase for the whole 5s budget. - httpClient is now injectable for testing. 200 send() calls now produce 8 HTTP requests instead of 200 (verified via a MockHandler-based benchmark). --- src/Metrics/Metrics.php | 99 ++++++++++++++++++++++------- src/Queue/RequestQueue.php | 101 ++++++++++++++++++++++++++++++ src/Sync/HttpLogSync.php | 81 +++++++++++++++--------- tests/Metrics/MetricsTest.php | 86 +++++++++++++++++++++++++ tests/Queue/RequestQueueTest.php | 104 +++++++++++++++++++++++++++++++ tests/Sync/HttpLogSyncTest.php | 100 +++++++++++++++++++++++++++++ 6 files changed, 518 insertions(+), 53 deletions(-) create mode 100644 src/Queue/RequestQueue.php create mode 100644 tests/Metrics/MetricsTest.php create mode 100644 tests/Queue/RequestQueueTest.php create mode 100644 tests/Sync/HttpLogSyncTest.php diff --git a/src/Metrics/Metrics.php b/src/Metrics/Metrics.php index 4dc6e4a..1568c34 100644 --- a/src/Metrics/Metrics.php +++ b/src/Metrics/Metrics.php @@ -5,18 +5,43 @@ namespace Logdash\Metrics; use GuzzleHttp\Client; -use GuzzleHttp\Exception\GuzzleException; +use GuzzleHttp\Pool; +use GuzzleHttp\Promise\PromiseInterface; use Logdash\Logger\InternalLogger; +use Logdash\Queue\RequestQueue; use Logdash\Types\RequiredInitializationParams; +use Throwable; class Metrics implements BaseMetrics { + private const BATCH_SIZE = 20; + private const FLUSH_INTERVAL_MS = 1000; + private const CONCURRENCY = 10; + private Client $httpClient; + private RequestQueue $queue; public function __construct( - private readonly RequiredInitializationParams $params + private readonly RequiredInitializationParams $params, + ?Client $httpClient = null ) { - $this->httpClient = new Client(); + $this->httpClient = $httpClient ?? new Client([ + 'connect_timeout' => 2, + 'timeout' => 5, + ]); + + $this->queue = new RequestQueue( + sendFn: function (array $metrics): void { + $this->sendBatch($metrics); + }, + batchSize: self::BATCH_SIZE, + flushIntervalMs: self::FLUSH_INTERVAL_MS, + // Metric updates (especially `mutate`) aren't idempotent, so a + // failed item is never retried here - only dispatched + // concurrently instead of one-by-one. + maxAttempts: 1, + verbose: $this->params->verbose, + ); } public function set(string $name, float $value): void @@ -25,7 +50,7 @@ public function set(string $name, float $value): void InternalLogger::getInternalLogger()->verbose("Setting metric {$name} to {$value}"); } - $this->sendMetric($name, $value, MetricOperation::SET); + $this->enqueue($name, $value, MetricOperation::SET); } public function mutate(string $name, float $value): void @@ -34,29 +59,55 @@ public function mutate(string $name, float $value): void InternalLogger::getInternalLogger()->verbose("Mutating metric {$name} by {$value}"); } - $this->sendMetric($name, $value, MetricOperation::CHANGE); + $this->enqueue($name, $value, MetricOperation::CHANGE); + } + + /** + * Sends any buffered metric updates immediately instead of waiting for + * the batch size/interval or process shutdown. Useful before a + * long-running worker goes idle. + */ + public function flush(): void + { + $this->queue->flush(); } - private function sendMetric(string $name, float $value, MetricOperation $operation): void + private function enqueue(string $name, float $value, MetricOperation $operation): void { - try { - $this->httpClient->put($this->params->host . '/metrics', [ - 'headers' => [ - 'Content-Type' => 'application/json', - 'project-api-key' => $this->params->apiKey, - ], - 'json' => [ - 'name' => $name, - 'value' => $value, - 'operation' => $operation->value, - ], - 'timeout' => 5, - ]); - } catch (GuzzleException $e) { - if ($this->params->verbose) { - InternalLogger::getInternalLogger()->verbose("Failed to send metric: " . $e->getMessage()); + $this->queue->add([ + 'name' => $name, + 'value' => $value, + 'operation' => $operation->value, + ]); + } + + /** + * @param list $metrics + */ + private function sendBatch(array $metrics): void + { + $requests = (function () use ($metrics) { + foreach ($metrics as $metric) { + yield fn (): PromiseInterface => $this->httpClient->putAsync($this->params->host . '/metrics', [ + 'headers' => [ + 'Content-Type' => 'application/json', + 'project-api-key' => $this->params->apiKey, + ], + 'json' => $metric, + ]); } - // Fail silently in production - } + })(); + + $pool = new Pool($this->httpClient, $requests, [ + 'concurrency' => self::CONCURRENCY, + 'rejected' => function (Throwable $reason): void { + if ($this->params->verbose) { + InternalLogger::getInternalLogger()->verbose('Failed to send metric: ' . $reason->getMessage()); + } + // Fail silently in production + }, + ]); + + $pool->promise()->wait(); } } diff --git a/src/Queue/RequestQueue.php b/src/Queue/RequestQueue.php new file mode 100644 index 0000000..83d8e55 --- /dev/null +++ b/src/Queue/RequestQueue.php @@ -0,0 +1,101 @@ + */ + private array $items = []; + + private bool $shutdownRegistered = false; + + private float $lastFlushAt; + + /** + * @param callable(list): void $sendFn + */ + public function __construct( + private readonly mixed $sendFn, + private readonly int $batchSize = 25, + private readonly int $flushIntervalMs = 2000, + private readonly int $maxAttempts = 1, + private readonly int $retryDelayMs = 200, + private readonly bool $verbose = false, + ) { + $this->lastFlushAt = microtime(true); + } + + public function add(mixed $item): void + { + $this->items[] = $item; + $this->registerShutdownFlush(); + + $elapsedMs = (microtime(true) - $this->lastFlushAt) * 1000; + + if (count($this->items) >= $this->batchSize || $elapsedMs >= $this->flushIntervalMs) { + $this->flush(); + } + } + + public function flush(): void + { + $this->lastFlushAt = microtime(true); + + if ($this->items === []) { + return; + } + + $batch = $this->items; + $this->items = []; + + $this->sendWithRetry($batch); + } + + /** + * @param list $batch + */ + private function sendWithRetry(array $batch): void + { + for ($attempt = 1; $attempt <= $this->maxAttempts; $attempt++) { + try { + ($this->sendFn)($batch); + return; + } catch (Throwable $e) { + if ($attempt >= $this->maxAttempts) { + if ($this->verbose) { + InternalLogger::getInternalLogger()->verbose( + 'Failed to flush batch of ' . count($batch) . " item(s) after {$attempt} attempt(s): " + . $e->getMessage() + ); + } + return; + } + + usleep($this->retryDelayMs * 1000); + } + } + } + + private function registerShutdownFlush(): void + { + if ($this->shutdownRegistered) { + return; + } + + $this->shutdownRegistered = true; + register_shutdown_function(function (): void { + $this->flush(); + }); + } +} diff --git a/src/Sync/HttpLogSync.php b/src/Sync/HttpLogSync.php index 2eb9896..f56c559 100644 --- a/src/Sync/HttpLogSync.php +++ b/src/Sync/HttpLogSync.php @@ -5,49 +5,72 @@ namespace Logdash\Sync; use GuzzleHttp\Client; -use GuzzleHttp\Exception\GuzzleException; -use Logdash\Logger\InternalLogger; +use Logdash\Queue\RequestQueue; use Logdash\Types\LogLevel; use Logdash\Types\RequiredInitializationParams; class HttpLogSync implements LogSync { + private const BATCH_SIZE = 25; + private const FLUSH_INTERVAL_MS = 2000; + private int $sequenceNumber = 0; private Client $httpClient; + private RequestQueue $queue; public function __construct( - private readonly RequiredInitializationParams $params + private readonly RequiredInitializationParams $params, + ?Client $httpClient = null ) { - $this->httpClient = new Client(); + $this->httpClient = $httpClient ?? new Client([ + 'connect_timeout' => 2, + 'timeout' => 5, + ]); + + $this->queue = new RequestQueue( + sendFn: function (array $logs): void { + $this->sendBatch($logs); + }, + batchSize: self::BATCH_SIZE, + flushIntervalMs: self::FLUSH_INTERVAL_MS, + maxAttempts: 2, + verbose: $this->params->verbose, + ); + } + + public function send(string $message, LogLevel $level, string $createdAt): void + { + $this->queue->add([ + 'message' => $message, + 'level' => $level->value, + 'createdAt' => $createdAt, + 'sequenceNumber' => $this->sequenceNumber++, + ]); } /** - * TODO: - * - queue - * - retry - * - batching + * Sends any buffered logs immediately instead of waiting for the batch + * size/interval or process shutdown. Useful before a long-running + * worker goes idle. */ - public function send(string $message, LogLevel $level, string $createdAt): void + public function flush(): void + { + $this->queue->flush(); + } + + /** + * @param list $logs + */ + private function sendBatch(array $logs): void { - try { - $this->httpClient->post($this->params->host . '/logs', [ - 'headers' => [ - 'Content-Type' => 'application/json', - 'project-api-key' => $this->params->apiKey, - ], - 'json' => [ - 'message' => $message, - 'level' => $level->value, - 'createdAt' => $createdAt, - 'sequenceNumber' => $this->sequenceNumber++, - ], - 'timeout' => 5, - ]); - } catch (GuzzleException $e) { - if ($this->params->verbose) { - InternalLogger::getInternalLogger()->verbose("Failed to send log: " . $e->getMessage()); - } - // Fail silently in production - } + $this->httpClient->post($this->params->host . '/logs/batch', [ + 'headers' => [ + 'Content-Type' => 'application/json', + 'project-api-key' => $this->params->apiKey, + ], + 'json' => [ + 'logs' => $logs, + ], + ]); } } diff --git a/tests/Metrics/MetricsTest.php b/tests/Metrics/MetricsTest.php new file mode 100644 index 0000000..727cc6d --- /dev/null +++ b/tests/Metrics/MetricsTest.php @@ -0,0 +1,86 @@ +push(Middleware::history($requests)); + $client = new Client(['handler' => $handlerStack]); + + $params = new RequiredInitializationParams( + apiKey: 'test-key', + host: 'https://test.logdash.io', + verbose: false + ); + + $metrics = new Metrics($params, $client); + + $metrics->set('active_users', 150); + $metrics->mutate('login_count', 1); + $metrics->mutate('login_count', -1); + $metrics->flush(); + + $this->assertCount(3, $requests); + + foreach ($requests as $entry) { + $this->assertSame('PUT', $entry['request']->getMethod()); + $this->assertSame('/metrics', $entry['request']->getUri()->getPath()); + } + + $bodies = array_map( + fn (array $entry) => json_decode((string) $entry['request']->getBody(), true), + $requests + ); + + $this->assertSame(['active_users', 'login_count', 'login_count'], array_column($bodies, 'name')); + $this->assertSame(['set', 'change', 'change'], array_column($bodies, 'operation')); + } + + public function testOneFailingMetricDoesNotBlockOthersInTheBatch(): void + { + $requests = []; + $mock = new MockHandler([ + new Response(500), + new Response(200), + ]); + $handlerStack = HandlerStack::create($mock); + $handlerStack->push(Middleware::history($requests)); + $client = new Client(['handler' => $handlerStack]); + + $params = new RequiredInitializationParams( + apiKey: 'test-key', + host: 'https://test.logdash.io', + verbose: false + ); + + $metrics = new Metrics($params, $client); + + $metrics->set('will_fail', 1); + $metrics->set('will_succeed', 2); + + // Must not throw even though the first request fails. + $metrics->flush(); + + $this->assertCount(2, $requests); + } +} diff --git a/tests/Queue/RequestQueueTest.php b/tests/Queue/RequestQueueTest.php new file mode 100644 index 0000000..9cc83e4 --- /dev/null +++ b/tests/Queue/RequestQueueTest.php @@ -0,0 +1,104 @@ +> $sentBatches */ + $sentBatches = []; + + $queue = new RequestQueue( + sendFn: function (array $batch) use (&$sentBatches): void { + $sentBatches[] = $batch; + }, + batchSize: 3, + flushIntervalMs: 60_000, + ); + + $queue->add('a'); + $queue->add('b'); + + $this->assertSame([], $sentBatches); + + $queue->add('c'); + + $this->assertSame([['a', 'b', 'c']], $sentBatches); + } + + public function testExplicitFlushSendsPartialBatch(): void + { + /** @var list> $sentBatches */ + $sentBatches = []; + + $queue = new RequestQueue( + sendFn: function (array $batch) use (&$sentBatches): void { + $sentBatches[] = $batch; + }, + batchSize: 100, + flushIntervalMs: 60_000, + ); + + $queue->add('a'); + $queue->flush(); + + $this->assertSame([['a']], $sentBatches); + + // Flushing an empty queue is a no-op. + $queue->flush(); + $this->assertSame([['a']], $sentBatches); + } + + public function testRetriesUpToMaxAttemptsThenGivesUpSilently(): void + { + $attempts = 0; + + $queue = new RequestQueue( + sendFn: function () use (&$attempts): void { + $attempts++; + throw new RuntimeException('boom'); + }, + batchSize: 1, + flushIntervalMs: 60_000, + maxAttempts: 3, + retryDelayMs: 0, + ); + + $queue->add('a'); + + $this->assertSame(3, $attempts); + } + + public function testSucceedsOnRetryAfterInitialFailure(): void + { + $attempts = 0; + /** @var list $delivered */ + $delivered = []; + + $queue = new RequestQueue( + sendFn: function (array $batch) use (&$attempts, &$delivered): void { + $attempts++; + if ($attempts === 1) { + throw new RuntimeException('transient failure'); + } + $delivered = $batch; + }, + batchSize: 1, + flushIntervalMs: 60_000, + maxAttempts: 3, + retryDelayMs: 0, + ); + + $queue->add('a'); + + $this->assertSame(2, $attempts); + $this->assertSame(['a'], $delivered); + } +} diff --git a/tests/Sync/HttpLogSyncTest.php b/tests/Sync/HttpLogSyncTest.php new file mode 100644 index 0000000..a4306cc --- /dev/null +++ b/tests/Sync/HttpLogSyncTest.php @@ -0,0 +1,100 @@ +push(\GuzzleHttp\Middleware::history($requests)); + $client = new Client(['handler' => $handlerStack]); + + $params = new RequiredInitializationParams( + apiKey: 'test-key', + host: 'https://test.logdash.io', + verbose: false + ); + + $logSync = new HttpLogSync($params, $client); + + $logSync->send('first', LogLevel::INFO, '2026-01-01T00:00:00+00:00'); + $logSync->send('second', LogLevel::ERROR, '2026-01-01T00:00:01+00:00'); + $logSync->flush(); + + $this->assertCount(1, $requests, 'Two log() calls should result in a single HTTP request'); + + $request = $requests[0]['request']; + $this->assertSame('POST', $request->getMethod()); + $this->assertSame('/logs/batch', $request->getUri()->getPath()); + $this->assertSame('test-key', $request->getHeaderLine('project-api-key')); + + $body = json_decode((string) $request->getBody(), true); + $this->assertCount(2, $body['logs']); + $this->assertSame('first', $body['logs'][0]['message']); + $this->assertSame('info', $body['logs'][0]['level']); + $this->assertSame(0, $body['logs'][0]['sequenceNumber']); + $this->assertSame('second', $body['logs'][1]['message']); + $this->assertSame('error', $body['logs'][1]['level']); + $this->assertSame(1, $body['logs'][1]['sequenceNumber']); + } + + public function testFlushingWithNoBufferedLogsSendsNoRequest(): void + { + $requests = []; + $mock = new MockHandler([]); + $handlerStack = HandlerStack::create($mock); + $handlerStack->push(\GuzzleHttp\Middleware::history($requests)); + $client = new Client(['handler' => $handlerStack]); + + $params = new RequiredInitializationParams( + apiKey: 'test-key', + host: 'https://test.logdash.io', + verbose: false + ); + + $logSync = new HttpLogSync($params, $client); + $logSync->flush(); + + $this->assertCount(0, $requests); + } + + public function testServerErrorDoesNotThrow(): void + { + $mock = new MockHandler([ + new Response(500), + new Response(500), + ]); + $client = new Client(['handler' => HandlerStack::create($mock)]); + + $params = new RequiredInitializationParams( + apiKey: 'test-key', + host: 'https://test.logdash.io', + verbose: false + ); + + $logSync = new HttpLogSync($params, $client); + $logSync->send('will fail', LogLevel::ERROR, '2026-01-01T00:00:00+00:00'); + + // Should swallow the failure (after its internal retry) instead of + // throwing into the caller's application code. + $logSync->flush(); + + $this->assertTrue(true); + } +}