diff options
Diffstat (limited to 'ansible_collections/community/general/plugins/cache/memcached.py')
-rw-r--r-- | ansible_collections/community/general/plugins/cache/memcached.py | 241 |
1 files changed, 241 insertions, 0 deletions
diff --git a/ansible_collections/community/general/plugins/cache/memcached.py b/ansible_collections/community/general/plugins/cache/memcached.py new file mode 100644 index 000000000..0bc5256b3 --- /dev/null +++ b/ansible_collections/community/general/plugins/cache/memcached.py @@ -0,0 +1,241 @@ +# -*- coding: utf-8 -*- +# Copyright (c) 2014, Brian Coca, Josh Drake, et al +# Copyright (c) 2017 Ansible Project +# GNU General Public License v3.0+ (see LICENSES/GPL-3.0-or-later.txt or https://www.gnu.org/licenses/gpl-3.0.txt) +# SPDX-License-Identifier: GPL-3.0-or-later + +from __future__ import (absolute_import, division, print_function) +__metaclass__ = type + +DOCUMENTATION = ''' + author: Unknown (!UNKNOWN) + name: memcached + short_description: Use memcached DB for cache + description: + - This cache uses JSON formatted, per host records saved in memcached. + requirements: + - memcache (python lib) + options: + _uri: + description: + - List of connection information for the memcached DBs + default: ['127.0.0.1:11211'] + type: list + elements: string + env: + - name: ANSIBLE_CACHE_PLUGIN_CONNECTION + ini: + - key: fact_caching_connection + section: defaults + _prefix: + description: User defined prefix to use when creating the DB entries + default: ansible_facts + env: + - name: ANSIBLE_CACHE_PLUGIN_PREFIX + ini: + - key: fact_caching_prefix + section: defaults + _timeout: + default: 86400 + description: Expiration timeout in seconds for the cache plugin data. Set to 0 to never expire + env: + - name: ANSIBLE_CACHE_PLUGIN_TIMEOUT + ini: + - key: fact_caching_timeout + section: defaults + type: integer +''' + +import collections +import os +import time +from multiprocessing import Lock +from itertools import chain + +from ansible.errors import AnsibleError +from ansible.module_utils.common._collections_compat import MutableSet +from ansible.plugins.cache import BaseCacheModule +from ansible.utils.display import Display + +try: + import memcache + HAS_MEMCACHE = True +except ImportError: + HAS_MEMCACHE = False + +display = Display() + + +class ProxyClientPool(object): + """ + Memcached connection pooling for thread/fork safety. Inspired by py-redis + connection pool. + + Available connections are maintained in a deque and released in a FIFO manner. + """ + + def __init__(self, *args, **kwargs): + self.max_connections = kwargs.pop('max_connections', 1024) + self.connection_args = args + self.connection_kwargs = kwargs + self.reset() + + def reset(self): + self.pid = os.getpid() + self._num_connections = 0 + self._available_connections = collections.deque(maxlen=self.max_connections) + self._locked_connections = set() + self._lock = Lock() + + def _check_safe(self): + if self.pid != os.getpid(): + with self._lock: + if self.pid == os.getpid(): + # bail out - another thread already acquired the lock + return + self.disconnect_all() + self.reset() + + def get_connection(self): + self._check_safe() + try: + connection = self._available_connections.popleft() + except IndexError: + connection = self.create_connection() + self._locked_connections.add(connection) + return connection + + def create_connection(self): + if self._num_connections >= self.max_connections: + raise RuntimeError("Too many memcached connections") + self._num_connections += 1 + return memcache.Client(*self.connection_args, **self.connection_kwargs) + + def release_connection(self, connection): + self._check_safe() + self._locked_connections.remove(connection) + self._available_connections.append(connection) + + def disconnect_all(self): + for conn in chain(self._available_connections, self._locked_connections): + conn.disconnect_all() + + def __getattr__(self, name): + def wrapped(*args, **kwargs): + return self._proxy_client(name, *args, **kwargs) + return wrapped + + def _proxy_client(self, name, *args, **kwargs): + conn = self.get_connection() + + try: + return getattr(conn, name)(*args, **kwargs) + finally: + self.release_connection(conn) + + +class CacheModuleKeys(MutableSet): + """ + A set subclass that keeps track of insertion time and persists + the set in memcached. + """ + PREFIX = 'ansible_cache_keys' + + def __init__(self, cache, *args, **kwargs): + self._cache = cache + self._keyset = dict(*args, **kwargs) + + def __contains__(self, key): + return key in self._keyset + + def __iter__(self): + return iter(self._keyset) + + def __len__(self): + return len(self._keyset) + + def add(self, value): + self._keyset[value] = time.time() + self._cache.set(self.PREFIX, self._keyset) + + def discard(self, value): + del self._keyset[value] + self._cache.set(self.PREFIX, self._keyset) + + def remove_by_timerange(self, s_min, s_max): + for k in list(self._keyset.keys()): + t = self._keyset[k] + if s_min < t < s_max: + del self._keyset[k] + self._cache.set(self.PREFIX, self._keyset) + + +class CacheModule(BaseCacheModule): + + def __init__(self, *args, **kwargs): + connection = ['127.0.0.1:11211'] + + super(CacheModule, self).__init__(*args, **kwargs) + if self.get_option('_uri'): + connection = self.get_option('_uri') + self._timeout = self.get_option('_timeout') + self._prefix = self.get_option('_prefix') + + if not HAS_MEMCACHE: + raise AnsibleError("python-memcached is required for the memcached fact cache") + + self._cache = {} + self._db = ProxyClientPool(connection, debug=0) + self._keys = CacheModuleKeys(self._db, self._db.get(CacheModuleKeys.PREFIX) or []) + + def _make_key(self, key): + return "{0}{1}".format(self._prefix, key) + + def _expire_keys(self): + if self._timeout > 0: + expiry_age = time.time() - self._timeout + self._keys.remove_by_timerange(0, expiry_age) + + def get(self, key): + if key not in self._cache: + value = self._db.get(self._make_key(key)) + # guard against the key not being removed from the keyset; + # this could happen in cases where the timeout value is changed + # between invocations + if value is None: + self.delete(key) + raise KeyError + self._cache[key] = value + + return self._cache.get(key) + + def set(self, key, value): + self._db.set(self._make_key(key), value, time=self._timeout, min_compress_len=1) + self._cache[key] = value + self._keys.add(key) + + def keys(self): + self._expire_keys() + return list(iter(self._keys)) + + def contains(self, key): + self._expire_keys() + return key in self._keys + + def delete(self, key): + del self._cache[key] + self._db.delete(self._make_key(key)) + self._keys.discard(key) + + def flush(self): + for key in self.keys(): + self.delete(key) + + def copy(self): + return self._keys.copy() + + def __getstate__(self): + return dict() + + def __setstate__(self, data): + self.__init__() |