Add support for RateLimitingSampler (#72)

* Add support for Tracing headers configuration

* Add support for RateLimitingSampler

* Use previous config, that follows JaegerConfig docs

* Increase code coverage & Fix some bugs

* Trying to find problem in Travis CI tests

* Update tests

* Fix test

* Commit RateLimitSamplerTest to be less dependend on test machine speed

* Fix 7.2 specific error

README.md

@@ -31,7 +31,7 @@ use OpenTracing\GlobalTracer;
 $config = new Config(
     [
         'sampler' => [
-            'type' => 'const',
+            'type' => Jaeger\SAMPLER_TYPE_CONST,
             'param' => true,
         ],
         'logging' => true,
@@ -48,6 +48,52 @@ $scope->close();
 $tracer->flush();
 ```
 
+### Samplers
+
+List of supported samplers, for more info about samplers, please read [Jaeger Sampling](https://www.jaegertracing.io/docs/1.9/sampling/) guide.
+
+#### Const sampler
+This sampler either samples everything, or nothing.
+
+##### Configuration
+```
+'sampler' => [
+    'type' => Jaeger\SAMPLER_TYPE_CONST,
+    'param' => true, // boolean wheter to trace or not
+],
+```
+
+#### Probabilistic sampler
+This sampler samples request by given rate.
+
+##### Configuration
+```
+'sampler' => [
+    'type' => Jaeger\SAMPLER_TYPE_PROBABILISTIC,
+    'param' => 0.5, // float [0.0, 1.0]
+],
+```
+
+#### Rate limiting sampler
+Samples maximum specified number of traces (requests) per second.
+
+##### Requirements
+* `psr/cache` PSR-6 cache component to store and retrieve sampler state between requests.
+Cache component is passed to `Jaeger\Config` trough its constructor.
+* `hrtime()` function, that can retrieve time in nanoseconds. You need either `php 7.3` or [PECL/hrtime](http://pecl.php.net/package/hrtime) extension.
+
+##### Configuration
+```
+'sampler' => [
+    'type' => Jaeger\SAMPLER_TYPE_RATE_LIMITING,
+    'param' => 100 // integer maximum number of traces per second,
+    'cache' => [
+        'currentBalanceKey' => 'rate.currentBalance' // string
+        'lastTickKey' => 'rate.lastTick' // string
+    ]
+],
+```
+
 ## Testing
 
 Tests are located in the `tests` directory. See [tests/README.md](./tests/README.md).

composer.json

@@ -28,14 +28,17 @@
         "opentracing/opentracing": "1.0.0-beta5",
         "packaged/thrift": "^0.10",
         "phlib/base_convert": "^1.0",
+        "psr/cache": "^1.0",
         "psr/log": "^1.0"
     },
     "provide": {
         "opentracing/opentracing": "1.0.0-beta5"
     },
     "require-dev": {
         "phpunit/phpunit": "^6.4",
-        "squizlabs/php_codesniffer": "3.*"
+        "squizlabs/php_codesniffer": "3.*",
+        "cache/array-adapter": "^1.0",
+        "symfony/polyfill-php73": "^1.10"
     },
     "config": {
         "optimize-autoloader": true,

src/Jaeger/Codec/TextCodec.php

@@ -130,7 +130,7 @@ public function extract($carrier)
             return null;
         }
 
-        return new SpanContext($traceId, $spanId, $parentId, $flags);
+        return new SpanContext($traceId, $spanId, $parentId, $flags, $baggage);
     }
 
     /**

src/Jaeger/Config.php

@@ -9,10 +9,13 @@
 use Jaeger\Reporter\ReporterInterface;
 use Jaeger\Sampler\ConstSampler;
 use Jaeger\Sampler\ProbabilisticSampler;
+use Jaeger\Sampler\RateLimitingSampler;
 use Jaeger\Sampler\SamplerInterface;
 use Jaeger\Sender\UdpSender;
 use Jaeger\Thrift\Agent\AgentClient;
+use Jaeger\Util\RateLimiter;
 use OpenTracing\GlobalTracer;
+use Psr\Cache\CacheItemPoolInterface;
 use Psr\Log\LoggerInterface;
 use Psr\Log\NullLogger;
 use Thrift\Exception\TTransportException;
@@ -41,15 +44,25 @@ class Config
      */
     private $logger;
 
+    /**
+     * @var CacheItemPoolInterface
+     */
+    private $cache;
+
     /**
      * Config constructor.
      * @param array $config
      * @param string|null $serviceName
      * @param LoggerInterface|null $logger
+     * @param CacheItemPoolInterface|null $cache
      * @throws Exception
      */
-    public function __construct(array $config, string $serviceName = null, LoggerInterface $logger = null)
-    {
+    public function __construct(
+        array $config,
+        string $serviceName = null,
+        LoggerInterface $logger = null,
+        CacheItemPoolInterface $cache = null
+    ) {
         $this->config = $config;
 
         $this->serviceName = $config['service_name'] ?? $serviceName;
@@ -58,6 +71,7 @@ public function __construct(array $config, string $serviceName = null, LoggerInt
         }
 
         $this->logger = $logger ?: new NullLogger();
+        $this->cache = $cache;
     }
 
     /**
@@ -143,6 +157,7 @@ private function getReporter(): ReporterInterface
 
     /**
      * @return SamplerInterface
+     * @throws \Psr\Cache\InvalidArgumentException
      * @throws Exception
      */
     private function getSampler(): SamplerInterface
@@ -157,8 +172,20 @@ private function getSampler(): SamplerInterface
             return new ConstSampler($samplerParam ?? false);
         } elseif ($samplerType === SAMPLER_TYPE_PROBABILISTIC) {
             return new ProbabilisticSampler((float)$samplerParam);
+        } elseif ($samplerType === SAMPLER_TYPE_RATE_LIMITING) {
+            if (!$this->cache) {
+                throw new Exception('You cannot use RateLimitingSampler without cache component');
+            }
+            $cacheConfig = $samplerConfig['cache'] ?? [];
+            return new RateLimitingSampler(
+                $samplerParam ?? 0,
+                new RateLimiter(
+                    $this->cache,
+                    $cacheConfig['currentBalanceKey'] ?? 'rate.currentBalance',
+                    $cacheConfig['lastTickKey'] ?? 'rate.lastTick'
+                )
+            );
         }
-
         throw new Exception('Unknown sampler type ' . $samplerType);
     }
 

src/Jaeger/Sampler/RateLimitingSampler.php

@@ -0,0 +1,62 @@
+<?php
+
+namespace Jaeger\Sampler;
+
+use Jaeger\Util\RateLimiter;
+
+use const Jaeger\SAMPLER_PARAM_TAG_KEY;
+use const Jaeger\SAMPLER_TYPE_RATE_LIMITING;
+use const Jaeger\SAMPLER_TYPE_TAG_KEY;
+
+class RateLimitingSampler implements SamplerInterface
+{
+    /**
+     * @var RateLimiter
+     */
+    private $rateLimiter;
+
+    /**
+     * A list of the sampler tags.
+     *
+     * @var array
+     */
+    private $tags = [];
+
+    public function __construct($maxTracesPerSecond, RateLimiter $rateLimiter)
+    {
+        $this->tags = [
+            SAMPLER_TYPE_TAG_KEY => SAMPLER_TYPE_RATE_LIMITING,
+            SAMPLER_PARAM_TAG_KEY => $maxTracesPerSecond,
+        ];
+
+        $maxTracesPerNanosecond = $maxTracesPerSecond / 1000000000.0;
+        $this->rateLimiter = $rateLimiter;
+        $this->rateLimiter->initialize($maxTracesPerNanosecond, $maxTracesPerSecond > 1.0 ? 1.0 : $maxTracesPerSecond);
+    }
+
+    /**
+     * Whether or not the new trace should be sampled.
+     *
+     * Implementations should return an array in the format [$decision, $tags].
+     *
+     * @param string $traceId The traceId on the span.
+     * @param string $operation The operation name set on the span.
+     * @return array
+     */
+    public function isSampled(string $traceId = '', string $operation = '')
+    {
+        return [$this->rateLimiter->checkCredit(1.0), $this->tags];
+    }
+
+    /**
+     * {@inheritdoc}
+     *
+     * Only implemented to satisfy the sampler interface.
+     *
+     * @return void
+     */
+    public function close()
+    {
+        // nothing to do
+    }
+}

src/Jaeger/Tracer.php

@@ -273,6 +273,7 @@ public function extract($format, $carrier)
      */
     public function flush()
     {
+        $this->sampler->close();
         $this->reporter->close();
     }
 

src/Jaeger/Util/RateLimiter.php

@@ -0,0 +1,128 @@
+<?php
+
+namespace Jaeger\Util;
+
+use Psr\Cache\CacheItemInterface;
+use Psr\Cache\CacheItemPoolInterface;
+
+class RateLimiter
+{
+    /**
+     * @var CacheItemPoolInterface
+     */
+    private $cache;
+
+    /**
+     * @var CacheItemInterface
+     */
+    private $balance;
+
+    /**
+     * @var CacheItemInterface
+     */
+    private $lastTick;
+
+    /**
+     * @var float
+     */
+    private $creditsPerNanosecond = 0;
+
+    /**
+     * @var float
+     */
+    private $maxBalance = 0;
+
+    /**
+     * RateLimiter constructor.
+     *
+     * @param CacheItemPoolInterface $cache
+     * @param string $currentBalanceKey key of current balance value in $cache
+     * @param string $lastTickKey key of last tick value in $cache
+     * @throws \Psr\Cache\InvalidArgumentException
+     */
+    public function __construct(
+        CacheItemPoolInterface $cache,
+        string $currentBalanceKey,
+        string $lastTickKey
+    ) {
+        $this->cache = $cache;
+        $this->balance = $this->cache->getItem($currentBalanceKey);
+        $this->lastTick = $this->cache->getItem($lastTickKey);
+    }
+
+    /**
+     * @param $itemCost
+     * @return bool
+     */
+    public function checkCredit($itemCost)
+    {
+        if (!$this->creditsPerNanosecond) {
+            return false;
+        }
+
+        list($lastTick, $balance) = $this->getState();
+
+        if (!$lastTick) {
+            $this->saveState(hrtime(true), 0);
+            return true;
+        }
+
+        $currentTick = hrtime(true);
+        $elapsedTime = $currentTick - $lastTick;
+        $balance += $elapsedTime * $this->creditsPerNanosecond;
+        if ($balance > $this->maxBalance) {
+            $balance = $this->maxBalance;
+        }
+
+        $result = false;
+        if ($balance >= $itemCost) {
+            $balance -= $itemCost;
+            $result = true;
+        }
+
+        $this->saveState($currentTick, $balance);
+
+        return $result;
+    }
+
+
+    /**
+     * Initializes limiter costs and boundaries
+     *
+     * @param float $creditsPerNanosecond
+     * @param float $maxBalance
+     */
+    public function initialize(float $creditsPerNanosecond, float $maxBalance)
+    {
+        $this->creditsPerNanosecond = $creditsPerNanosecond;
+        $this->maxBalance = $maxBalance;
+    }
+
+    /**
+     * Method loads last tick and current balance from cache
+     *
+     * @return array [$lastTick, $balance]
+     */
+    private function getState() : array
+    {
+        return [
+            $this->lastTick->get(),
+            $this->balance->get()
+        ];
+    }
+
+    /**
+     * Method saves last tick and current balance into cache
+     *
+     * @param integer $lastTick
+     * @param float $balance
+     */
+    private function saveState($lastTick, $balance)
+    {
+        $this->lastTick->set($lastTick);
+        $this->balance->set($balance);
+        $this->cache->saveDeferred($this->lastTick);
+        $this->cache->saveDeferred($this->balance);
+        $this->cache->commit();
+    }
+}