Skip to content
Permalink
Browse files

feature #31889 [Lock] add mongodb store (kralos)

This PR was submitted for the 4.4 branch but it was squashed and merged into the 5.1-dev branch instead (closes #31889).

Discussion
----------

[Lock] add mongodb store

| Q             | A
| ------------- | ---
| Branch?       | 4.4
| Bug fix?      | no
| New feature?  | yes
| BC breaks?    | no
| Deprecations? | no
| Tests pass?   | yes (requires `ext-mongodb` and `mongodb/mongodb` to test)
| Fixed tickets | #27345
| License       | MIT
| Original Doc PR        | symfony/symfony-docs#9807
| Remove from 4.3 Doc PR | symfony/symfony-docs#11686
| Add to 4.4 Doc PR | symfony/symfony-docs#11735

Looks like I messed up `kralos:27345-lock-mongodb` with a force push (trying to fix ci issues) right before it was merged to `master` (`4.3.0`).

see #27648

**Description**
We should support Semaphore Locks with a MongoDB back end to allow those that already use MongoDB as a distributed storage engine.

Symfony already partially supports MongoDB for session storage: `Symfony\Component\HttpFoundation\Session\Storage\Handler\MongoDbSessionHandler`

**Example**

```php
$client = new MongoDb\Client();

$store = new Symfony\Component\Lock\Store\MongoDbStore(
    $client
    array(
        'database' => 'my-app',
    )
);
$lockFactory = new Symfony\Component\Lock\Factory($store);
$lock = $lockFactory->createLock('my-resource');
```

Commits
-------

a6bfa59 [Lock] add mongodb store
  • Loading branch information
fabpot committed Dec 11, 2019
2 parents 1827908 + a6bfa59 commit e8ecfe3535f68f5f284347e2339584875adfe88c
@@ -11,7 +11,8 @@ CHANGELOG
4.4.0
-----

* added InvalidTtlException
* added InvalidTtlException
* added the MongoDbStore supporting MongoDB servers >=2.2
* deprecated `StoreInterface` in favor of `BlockingStoreInterface` and `PersistingStoreInterface`
* `Factory` is deprecated, use `LockFactory` instead
* `StoreFactory::createStore` allows PDO and Zookeeper DSN.
@@ -0,0 +1,386 @@
<?php

/*
* This file is part of the Symfony package.
*
* (c) Fabien Potencier <fabien@symfony.com>
*
* For the full copyright and license information, please view the LICENSE
* file that was distributed with this source code.
*/

namespace Symfony\Component\Lock\Store;

use MongoDB\BSON\UTCDateTime;
use MongoDB\Client;
use MongoDB\Collection;
use MongoDB\Driver\Command;
use MongoDB\Driver\Exception\WriteException;
use MongoDB\Exception\DriverRuntimeException;
use MongoDB\Exception\InvalidArgumentException as MongoInvalidArgumentException;
use MongoDB\Exception\UnsupportedException;
use Symfony\Component\Lock\Exception\InvalidArgumentException;
use Symfony\Component\Lock\Exception\InvalidTtlException;
use Symfony\Component\Lock\Exception\LockAcquiringException;
use Symfony\Component\Lock\Exception\LockConflictedException;
use Symfony\Component\Lock\Exception\LockExpiredException;
use Symfony\Component\Lock\Exception\LockStorageException;
use Symfony\Component\Lock\Exception\NotSupportedException;
use Symfony\Component\Lock\Key;
use Symfony\Component\Lock\StoreInterface;

/**
* MongoDbStore is a StoreInterface implementation using MongoDB as a storage
* engine. Support for MongoDB server >=2.2 due to use of TTL indexes.
*
* CAUTION: TTL Indexes are used so this store relies on all client and server
* nodes to have synchronized clocks for lock expiry to occur at the correct
* time. To ensure locks don't expire prematurely; the TTLs should be set with
* enough extra time to account for any clock drift between nodes.
*
* CAUTION: The locked resource name is indexed in the _id field of the lock
* collection. An indexed field's value in MongoDB can be a maximum of 1024
* bytes in length inclusive of structural overhead.
*
* @see https://docs.mongodb.com/manual/reference/limits/#Index-Key-Limit
*
* @requires extension mongodb
*
* @author Joe Bennett <joe@assimtech.com>
*/
class MongoDbStore implements StoreInterface
{
private $collection;
private $client;
private $uri;
private $options;
private $initialTtl;

private $databaseVersion;

use ExpiringStoreTrait;

/**
* @param Collection|Client|string $mongo An instance of a Collection or Client or URI @see https://docs.mongodb.com/manual/reference/connection-string/
* @param array $options See below
* @param float $initialTtl The expiration delay of locks in seconds
*
* @throws InvalidArgumentException If required options are not provided
* @throws InvalidTtlException When the initial ttl is not valid
*
* Options:
* gcProbablity: Should a TTL Index be created expressed as a probability from 0.0 to 1.0 [default: 0.001]
* database: The name of the database [required when $mongo is a Client]
* collection: The name of the collection [required when $mongo is a Client]
* uriOptions: Array of uri options. [used when $mongo is a URI]
* driverOptions: Array of driver options. [used when $mongo is a URI]
*
* When using a URI string:
* the database is determined from the "database" option, otherwise the uri's path is used.
* the collection is determined from the "collection" option, otherwise the uri's "collection" querystring parameter is used.
*
* For example: mongodb://myuser:mypass@myhost/mydatabase?collection=mycollection
*
* @see https://docs.mongodb.com/php-library/current/reference/method/MongoDBClient__construct/
*
* If gcProbablity is set to a value greater than 0.0 there is a chance
* this store will attempt to create a TTL index on self::save().
* If you prefer to create your TTL Index manually you can set gcProbablity
* to 0.0 and optionally leverage
* self::createTtlIndex(int $expireAfterSeconds = 0).
*
* writeConcern, readConcern and readPreference are not specified by
* MongoDbStore meaning the collection's settings will take effect.
* @see https://docs.mongodb.com/manual/applications/replication/
*/
public function __construct($mongo, array $options = [], float $initialTtl = 300.0)
{
$this->options = array_merge([
'gcProbablity' => 0.001,
'database' => null,
'collection' => null,
'uriOptions' => [],
'driverOptions' => [],
], $options);

$this->initialTtl = $initialTtl;

if ($mongo instanceof Collection) {
$this->collection = $mongo;
} elseif ($mongo instanceof Client) {
if (null === $this->options['database']) {
throw new InvalidArgumentException(sprintf('%s() requires the "database" option when constructing with a %s', __METHOD__, Client::class));
}
if (null === $this->options['collection']) {
throw new InvalidArgumentException(sprintf('%s() requires the "collection" option when constructing with a %s', __METHOD__, Client::class));
}

$this->client = $mongo;
} elseif (\is_string($mongo)) {
if (false === $parsedUrl = parse_url($mongo)) {
throw new InvalidArgumentException(sprintf('The given MongoDB Connection URI "%s" is invalid.', $mongo));
}
$query = [];
if (isset($parsedUrl['query'])) {
parse_str($parsedUrl['query'], $query);
}
$this->options['collection'] = $this->options['collection'] ?? $query['collection'] ?? null;
$this->options['database'] = $this->options['database'] ?? ltrim($parsedUrl['path'] ?? '', '/') ?: null;
if (null === $this->options['database']) {
throw new InvalidArgumentException(sprintf('%s() requires the "database" in the uri path or option when constructing with a uri', __METHOD__));
}
if (null === $this->options['collection']) {
throw new InvalidArgumentException(sprintf('%s() requires the "collection" in the uri querystring or option when constructing with a uri', __METHOD__));
}

$this->uri = $mongo;
} else {
throw new InvalidArgumentException(sprintf('%s() requires %s or %s or URI as first argument, %s given.', __METHOD__, Collection::class, Client::class, \is_object($mongo) ? \get_class($mongo) : \gettype($mongo)));
}

if ($this->options['gcProbablity'] < 0.0 || $this->options['gcProbablity'] > 1.0) {
throw new InvalidArgumentException(sprintf('%s() gcProbablity must be a float from 0.0 to 1.0, %f given.', __METHOD__, $this->options['gcProbablity']));
}

if ($this->initialTtl <= 0) {
throw new InvalidTtlException(sprintf('%s() expects a strictly positive TTL. Got %d.', __METHOD__, $this->initialTtl));
}
}

/**
* Create a TTL index to automatically remove expired locks.
*
* If the gcProbablity option is set higher than 0.0 (defaults to 0.001);
* there is a chance this will be called on self::save().
*
* Otherwise; this should be called once manually during database setup.
*
* Alternatively the TTL index can be created manually on the database:
*
* db.lock.ensureIndex(
* { "expires_at": 1 },
* { "expireAfterSeconds": 0 }
* )
*
* Please note, expires_at is based on the application server. If the
* database time differs; a lock could be cleaned up before it has expired.
* To ensure locks don't expire prematurely; the lock TTL should be set
* with enough extra time to account for any clock drift between nodes.
*
* A TTL index MUST BE used to automatically clean up expired locks.
*
* @see http://docs.mongodb.org/manual/tutorial/expire-data/
*
* @throws UnsupportedException if options are not supported by the selected server
* @throws MongoInvalidArgumentException for parameter/option parsing errors
* @throws DriverRuntimeException for other driver errors (e.g. connection errors)
*/
public function createTtlIndex(int $expireAfterSeconds = 0)
{
$this->getCollection()->createIndex(
[ // key
'expires_at' => 1,
],
[ // options
'expireAfterSeconds' => $expireAfterSeconds,
]
);
}

/**
* {@inheritdoc}
*
* @throws LockExpiredException when save is called on an expired lock
*/
public function save(Key $key)
{
$key->reduceLifetime($this->initialTtl);

try {
$this->upsert($key, $this->initialTtl);
} catch (WriteException $e) {
if ($this->isDuplicateKeyException($e)) {
throw new LockConflictedException('Lock was acquired by someone else', 0, $e);
}
throw new LockAcquiringException('Failed to acquire lock', 0, $e);
}

if ($this->options['gcProbablity'] > 0.0
&& (
1.0 === $this->options['gcProbablity']
|| (random_int(0, PHP_INT_MAX) / PHP_INT_MAX) <= $this->options['gcProbablity']
)
) {
$this->createTtlIndex();
}

$this->checkNotExpired($key);
}

/**
* {@inheritdoc}
*/
public function waitAndSave(Key $key)
{
throw new NotSupportedException(sprintf('The store "%s" does not support blocking locks.', __CLASS__));
}

/**
* {@inheritdoc}
*
* @throws LockStorageException
* @throws LockExpiredException
*/
public function putOffExpiration(Key $key, $ttl)
{
$key->reduceLifetime($ttl);

try {
$this->upsert($key, $ttl);
} catch (WriteException $e) {
if ($this->isDuplicateKeyException($e)) {
throw new LockConflictedException('Failed to put off the expiration of the lock', 0, $e);
}
throw new LockStorageException($e->getMessage(), 0, $e);
}

$this->checkNotExpired($key);
}

/**
* {@inheritdoc}
*/
public function delete(Key $key)
{
$this->getCollection()->deleteOne([ // filter
'_id' => (string) $key,
'token' => $this->getUniqueToken($key),
]);
}

/**
* {@inheritdoc}
*/
public function exists(Key $key): bool
{
return null !== $this->getCollection()->findOne([ // filter
'_id' => (string) $key,
'token' => $this->getUniqueToken($key),
'expires_at' => [
'$gt' => $this->createMongoDateTime(microtime(true)),
],
]);
}

/**
* Update or Insert a Key.
*
* @param float $ttl Expiry in seconds from now
*/
private function upsert(Key $key, float $ttl)
{
$now = microtime(true);
$token = $this->getUniqueToken($key);

$this->getCollection()->updateOne(
[ // filter
'_id' => (string) $key,
'$or' => [
[
'token' => $token,
],
[
'expires_at' => [
'$lte' => $this->createMongoDateTime($now),
],
],
],
],
[ // update
'$set' => [
'_id' => (string) $key,
'token' => $token,
'expires_at' => $this->createMongoDateTime($now + $ttl),
],
],
[ // options
'upsert' => true,
]
);
}

private function isDuplicateKeyException(WriteException $e): bool
{
$code = $e->getCode();

$writeErrors = $e->getWriteResult()->getWriteErrors();
if (1 === \count($writeErrors)) {
$code = $writeErrors[0]->getCode();
}

// Mongo error E11000 - DuplicateKey
return 11000 === $code;
}

private function getDatabaseVersion(): string
{
if (null !== $this->databaseVersion) {
return $this->databaseVersion;
}

$command = new Command([
'buildinfo' => 1,
]);
$cursor = $this->getCollection()->getManager()->executeReadCommand(
$this->getCollection()->getDatabaseName(),
$command
);
$buildInfo = $cursor->toArray()[0];
$this->databaseVersion = $buildInfo->version;

return $this->databaseVersion;
}

private function getCollection(): Collection
{
if (null !== $this->collection) {
return $this->collection;
}

if (null === $this->client) {
$this->client = new Client($this->uri, $this->options['uriOptions'], $this->options['driverOptions']);
}

$this->collection = $this->client->selectCollection(
$this->options['database'],
$this->options['collection']
);

return $this->collection;
}

/**
* @param float $seconds Seconds since 1970-01-01T00:00:00.000Z supporting millisecond precision. Defaults to now.
*/
private function createMongoDateTime(float $seconds): UTCDateTime
{
return new UTCDateTime($seconds * 1000);
}

/**
* Retrieves an unique token for the given key namespaced to this store.
*
* @param Key lock state container
*
* @return string token
*/
private function getUniqueToken(Key $key): string
{
if (!$key->hasState(__CLASS__)) {
$token = base64_encode(random_bytes(32));
$key->setState(__CLASS__, $token);
}

return $key->getState(__CLASS__);
}
}

0 comments on commit e8ecfe3

Please sign in to comment.
You can’t perform that action at this time.