[feat] Support buffer mechanism in standalone process of metric (#6030)

Co-authored-by: 李铭昕 <[email protected]>

publish/metric.php

@@ -19,6 +19,9 @@
  'use_standalone_process' => env('METRIC_USE_STANDALONE_PROCESS', true),
  'enable_default_metric' => env('METRIC_ENABLE_DEFAULT_METRIC', true),
  'default_metric_interval' => env('DEFAULT_METRIC_INTERVAL', 5),
+ // only available when use_standalone_process is true
+ 'buffer_interval' => env('METRIC_BUFFER_INTERVAL', 5),
+ 'buffer_size' => env('METRIC_BUFFER_SIZE', 200),
  'metric' => [
  'prometheus' => [
  'driver' => Hyperf\Metric\Adapter\Prometheus\MetricFactory::class,

src/Adapter/RemoteProxy/Counter.php

@@ -11,13 +11,12 @@
  */
 namespace Hyperf\Metric\Adapter\RemoteProxy;
 
+use Hyperf\Context\ApplicationContext;
 use Hyperf\Metric\Contract\CounterInterface;
-use Hyperf\Process\ProcessCollector;
+use Hyperf\Metric\Contract\MetricCollectorInterface;
 
 class Counter implements CounterInterface
 {
- protected const TARGET_PROCESS_NAME = 'metric';
-
  /**
  * @var string[]
  */
@@ -38,7 +37,9 @@ public function with(string ...$labelValues): static
  public function add(int $delta): void
  {
  $this->delta = $delta;
- $process = ProcessCollector::get(static::TARGET_PROCESS_NAME)[0];
- $process->write(serialize($this));
+
+ ApplicationContext::getContainer()
+ ->get(MetricCollectorInterface::class)
+ ->add($this);
  }
 }

src/Adapter/RemoteProxy/Gauge.php

@@ -11,13 +11,12 @@
  */
 namespace Hyperf\Metric\Adapter\RemoteProxy;
 
+use Hyperf\Context\ApplicationContext;
 use Hyperf\Metric\Contract\GaugeInterface;
-use Hyperf\Process\ProcessCollector;
+use Hyperf\Metric\Contract\MetricCollectorInterface;
 
 class Gauge implements GaugeInterface
 {
- protected const TARGET_PROCESS_NAME = 'metric';
-
  /**
  * @var string[]
  */
@@ -41,15 +40,19 @@ public function set(float $value): void
  {
  $this->value = $value;
  $this->delta = null;
- $process = ProcessCollector::get(static::TARGET_PROCESS_NAME)[0];
- $process->write(serialize($this));
+
+ ApplicationContext::getContainer()
+ ->get(MetricCollectorInterface::class)
+ ->add($this);
  }
 
  public function add(float $delta): void
  {
  $this->delta = $delta;
  $this->value = null;
- $process = ProcessCollector::get(static::TARGET_PROCESS_NAME)[0];
- $process->write(serialize($this));
+
+ ApplicationContext::getContainer()
+ ->get(MetricCollectorInterface::class)
+ ->add($this);
  }
 }

src/Adapter/RemoteProxy/Histogram.php

@@ -11,13 +11,12 @@
  */
 namespace Hyperf\Metric\Adapter\RemoteProxy;
 
+use Hyperf\Context\ApplicationContext;
 use Hyperf\Metric\Contract\HistogramInterface;
-use Hyperf\Process\ProcessCollector;
+use Hyperf\Metric\Contract\MetricCollectorInterface;
 
 class Histogram implements HistogramInterface
 {
- protected const TARGET_PROCESS_NAME = 'metric';
-
  /**
  * @var string[]
  */
@@ -38,7 +37,9 @@ public function with(string ...$labelValues): static
  public function put(float $sample): void
  {
  $this->sample = $sample;
- $process = ProcessCollector::get(static::TARGET_PROCESS_NAME)[0];
- $process->write(serialize($this));
+
+ ApplicationContext::getContainer()
+ ->get(MetricCollectorInterface::class)
+ ->add($this);
  }
 }

src/Adapter/RemoteProxy/MetricCollector.php

@@ -0,0 +1,49 @@
+<?php
+
+declare(strict_types=1);
+/**
+ * This file is part of Hyperf.
+ *
+ * @link https://www.hyperf.io
+ * @document https://hyperf.wiki
+ * @contact [email protected]
+ * @license https://github.com/hyperf/hyperf/blob/master/LICENSE
+ */
+namespace Hyperf\Metric\Adapter\RemoteProxy;
+
+use Hyperf\Metric\Contract\MetricCollectorInterface;
+use Hyperf\Process\ProcessCollector;
+
+class MetricCollector implements MetricCollectorInterface
+{
+ protected const TARGET_PROCESS_NAME = 'metric';
+
+ protected array $buffer = [];
+
+ public function __construct(
+ protected int $bufferSize = 200
+ ) {
+ }
+
+ public function add(object $data): void
+ {
+ $this->buffer[] = $data;
+
+ if (count($this->buffer) >= $this->bufferSize) {
+ $this->flush();
+ }
+ }
+
+ public function flush(): void
+ {
+ $process = ProcessCollector::get(static::TARGET_PROCESS_NAME)[0];
+ $buffer = $this->buffer;
+ $this->buffer = [];
+ $process->write(serialize($buffer));
+ }
+
+ public function getBuffer(): array
+ {
+ return $this->buffer;
+ }
+}

src/Adapter/RemoteProxy/MetricCollectorFactory.php

@@ -0,0 +1,34 @@
+<?php
+
+declare(strict_types=1);
+/**
+ * This file is part of Hyperf.
+ *
+ * @link https://www.hyperf.io
+ * @document https://hyperf.wiki
+ * @contact [email protected]
+ * @license https://github.com/hyperf/hyperf/blob/master/LICENSE
+ */
+namespace Hyperf\Metric\Adapter\RemoteProxy;
+
+use Hyperf\Contract\ConfigInterface;
+use Hyperf\Metric\Contract\MetricCollectorInterface;
+use Psr\Container\ContainerExceptionInterface;
+use Psr\Container\ContainerInterface;
+use Psr\Container\NotFoundExceptionInterface;
+
+class MetricCollectorFactory
+{
+ /**
+ * @throws ContainerExceptionInterface
+ * @throws NotFoundExceptionInterface
+ */
+ public function __invoke(ContainerInterface $container): MetricCollectorInterface
+ {
+ $config = $container->get(ConfigInterface::class);
+
+ return new MetricCollector(
+ (int) $config->get('metric.buffer_size', 200)
+ );
+ }
+}

src/ConfigProvider.php

@@ -13,9 +13,12 @@
 
 use Domnikl\Statsd\Connection;
 use Domnikl\Statsd\Connection\UdpSocket;
+use Hyperf\Metric\Adapter\RemoteProxy\MetricCollectorFactory;
 use Hyperf\Metric\Aspect\CounterAnnotationAspect;
 use Hyperf\Metric\Aspect\HistogramAnnotationAspect;
+use Hyperf\Metric\Contract\MetricCollectorInterface;
 use Hyperf\Metric\Contract\MetricFactoryInterface;
+use Hyperf\Metric\Listener\MetricBufferWatcher;
 use Hyperf\Metric\Listener\OnBeforeHandle;
 use Hyperf\Metric\Listener\OnCoroutineServerStart;
 use Hyperf\Metric\Listener\OnMetricFactoryReady;
@@ -37,6 +40,7 @@ public function __invoke(): array
  Adapter::class => InMemory::class,
  Connection::class => UdpSocket::class,
  DriverInterface::class => Guzzle::class,
+ MetricCollectorInterface::class => MetricCollectorFactory::class,
  ],
  'aspects' => [
  CounterAnnotationAspect::class,
@@ -56,6 +60,7 @@ public function __invoke(): array
  OnBeforeHandle::class,
  OnWorkerStart::class,
  OnCoroutineServerStart::class,
+ MetricBufferWatcher::class,
  ],
  'processes' => [
  MetricProcess::class,

src/Contract/MetricCollectorInterface.php

@@ -0,0 +1,19 @@
+<?php
+
+declare(strict_types=1);
+/**
+ * This file is part of Hyperf.
+ *
+ * @link https://www.hyperf.io
+ * @document https://hyperf.wiki
+ * @contact [email protected]
+ * @license https://github.com/hyperf/hyperf/blob/master/LICENSE
+ */
+namespace Hyperf\Metric\Contract;
+
+interface MetricCollectorInterface
+{
+ public function add(object $data): void;
+
+ public function flush(): void;
+}

src/Listener/MetricBufferWatcher.php

@@ -0,0 +1,80 @@
+<?php
+
+declare(strict_types=1);
+/**
+ * This file is part of Hyperf.
+ *
+ * @link https://www.hyperf.io
+ * @document https://hyperf.wiki
+ * @contact [email protected]
+ * @license https://github.com/hyperf/hyperf/blob/master/LICENSE
+ */
+namespace Hyperf\Metric\Listener;
+
+use Hyperf\Contract\ConfigInterface;
+use Hyperf\Coordinator\Constants;
+use Hyperf\Coordinator\CoordinatorManager;
+use Hyperf\Coordinator\Timer;
+use Hyperf\Coroutine\Coroutine;
+use Hyperf\Event\Contract\ListenerInterface;
+use Hyperf\Framework\Event\BeforeWorkerStart;
+use Hyperf\Metric\Contract\MetricCollectorInterface;
+use Psr\Container\ContainerInterface;
+
+/**
+ * Collect and handle metrics before worker start.
+ * Only used for swoole process mode.
+ */
+class MetricBufferWatcher implements ListenerInterface
+{
+ private ConfigInterface $config;
+
+ private MetricCollectorInterface $collector;
+
+ private Timer $timer;
+
+ public function __construct(protected ContainerInterface $container)
+ {
+ $this->config = $container->get(ConfigInterface::class);
+ $this->collector = $container->get(MetricCollectorInterface::class);
+ $this->timer = new Timer();
+ }
+
+ /**
+ * @return string[] returns the events that you want to listen
+ */
+ public function listen(): array
+ {
+ return [
+ BeforeWorkerStart::class,
+ ];
+ }
+
+ /**
+ * Handle the Event when the event is triggered, all listeners will
+ * complete before the event is returned to the EventDispatcher.
+ */
+ public function process(object $event): void
+ {
+ if ($event->workerId === null) {
+ return;
+ }
+
+ /*
+ * Only start buffer watcher in standalone process mode
+ */
+ if (! $this->config->get('metric.use_standalone_process', true)) {
+ return;
+ }
+
+ $timerInterval = $this->config->get('metric.buffer_interval', 5);
+ $timerId = $this->timer->tick($timerInterval, function () {
+ $this->collector->flush();
+ });
+ // Clean up timer on worker exit;
+ Coroutine::create(function () use ($timerId) {
+ CoordinatorManager::until(Constants::WORKER_EXIT)->yield();
+ $this->timer->clear($timerId);
+ });
+ }
+}