schema: Caching on schema on client

Store schema in per user cache. Together with schemas also information
about mapping between server and fingerprint is stored to reduce traffic.

https://fedorahosted.org/freeipa/ticket/4739

Reviewed-By: Jan Cholasta <jcholast@redhat.com>

1 files changed, 220 insertions, 10 deletions

diff --git a/ipaclient/remote_plugins/schema.py b/ipaclient/remote_plugins/schema.py
index bf44f9f8e..a54d4eb77 100644
--- a/ipaclient/remote_plugins/schema.py
+++ b/ipaclient/remote_plugins/schema.py
@@ -3,20 +3,28 @@
 #
 
 import collections
-import os.path
+import errno
+import fcntl
+import glob
+import json
+import os
+import re
 import sys
+import time
 import types
+import zipfile
 
 import six
 
 from ipaclient.plugins.rpcclient import rpcclient
-from ipalib import parameters, plugable
+from ipalib import errors, parameters, plugable
 from ipalib.frontend import Command, Method, Object
 from ipalib.output import Output
 from ipalib.parameters import DefaultFrom, Flag, Password, Str
 from ipalib.text import _
 from ipapython.dn import DN
 from ipapython.dnsutil import DNSName
+from ipapython.ipa_log_manager import log_mgr
 
 if six.PY3:
     unicode = str
@@ -46,6 +54,21 @@ _PARAMS = {
     'str': parameters.Str,
 }
 
+USER_CACHE_PATH = (
+    os.environ.get('XDG_CACHE_HOME') or
+    os.path.join(
+        os.environ.get(
+            'HOME',
+            os.path.expanduser('~')
+        ),
+        '.cache'
+    )
+)
+SCHEMA_DIR = os.path.join(USER_CACHE_PATH, 'ipa', 'schema')
+SERVERS_DIR = os.path.join(USER_CACHE_PATH, 'ipa', 'servers')
+
+logger = log_mgr.get_logger(__name__)
+
 
 class _SchemaCommand(Command):
     def get_options(self):
@@ -340,22 +363,209 @@ class _SchemaObjectPlugin(_SchemaPlugin):
     schema_key = 'classes'
 
 
-def get_package(api):
+def _ensure_dir_created(d):
     try:
-        schema = api._schema
-    except AttributeError:
-        client = rpcclient(api)
-        client.finalize()
+        os.makedirs(d)
+    except OSError as e:
+        if e.errno != errno.EEXIST:
+            raise RuntimeError("Unable to create cache directory: {}"
+                               "".format(e))
+
+
+class _LockedZipFile(zipfile.ZipFile):
+    """ Add locking to zipfile.ZipFile
+    Shared lock is used with read mode, exclusive with write mode.
+    """
+    def __enter__(self):
+        if 'r' in self.mode:
+            fcntl.flock(self.fp, fcntl.LOCK_SH)
+        elif 'w' in self.mode or 'a' in self.mode:
+            fcntl.flock(self.fp, fcntl.LOCK_EX)
+
+        return super(_LockedZipFile, self).__enter__()
+
+    def __exit__(self, type_, value, traceback):
+        fcntl.flock(self.fp, fcntl.LOCK_UN)
+
+        return super(_LockedZipFile, self).__exit__(type_, value, traceback)
+
+
+class _SchemaNameSpace(collections.Mapping):
+
+    def __init__(self, schema, name):
+        self.name = name
+        self._schema = schema
+
+    def __getitem__(self, key):
+        return self._schema.read_namespace_member(self.name, key)
+
+    def __iter__(self):
+        for key in self._schema.iter_namespace(self.name):
+            yield key
+
+    def __len__(self):
+        return len(list(self._schema.iter_namespace(self.name)))
+
+
+class Schema(object):
+    """
+    Store and provide schema for commands and topics
 
+    Create api instance
+    >>> from ipalib import api
+    >>> api.bootstrap(context='cli')
+    >>> api.finalize()
+
+    Get schema object
+    >>> m = Schema(api)
+
+    From now on we can access schema for commands stored in cache
+    >>> m['commands'][u'ping'][u'doc']
+    u'Ping a remote server.'
+
+    >>> m['topics'][u'ping'][u'doc']
+    u'Ping the remote IPA server to ...'
+
+    """
+    schema_path_template = os.path.join(SCHEMA_DIR, '{}')
+    servers_path_template = os.path.join(SERVERS_DIR, '{}')
+    ns_member_pattern_template = '^{}/(?P<name>.+)$'
+    ns_member_path_template = '{}/{}'
+    namespaces = {'classes', 'commands', 'topics'}
+    schema_info_path = 'schema'
+
+    @classmethod
+    def _list(cls):
+        for f in glob.glob(cls.schema_path_template.format('*')):
+            yield os.path.splitext(os.path.basename(f))[0]
+
+    @classmethod
+    def _in_cache(cls, fingeprint):
+        return os.path.exists(cls.schema_path_template.format(fingeprint))
+
+    def __init__(self, api):
+        self._api = api
+        self._dict = {}
+
+    def _open_server_info(self, hostname, mode):
+        encoded_hostname = DNSName(hostname).ToASCII()
+        path = self.servers_path_template.format(encoded_hostname)
+        return open(path, mode)
+
+    def _get_schema(self):
+        client = rpcclient(self._api)
+        client.finalize()
         client.connect(verbose=False)
+
+        fps = [unicode(f) for f in Schema._list()]
+        kwargs = {u'version': u'2.170'}
+        if fps:
+            kwargs[u'known_fingerprints'] = fps
         try:
-            schema = client.forward(u'schema', version=u'2.170')['result']
+            schema = client.forward(u'schema', **kwargs)['result']
+        except errors.SchemaUpToDate as e:
+            fp = e.fingerprint
+            ttl = e.ttl
+        else:
+            fp = schema['fingerprint']
+            ttl = schema['ttl']
+            self._store(fp, schema)
         finally:
             client.disconnect()
 
-        for key in ('commands', 'classes', 'topics'):
-            schema[key] = {str(s['full_name']): s for s in schema[key]}
+        exp = ttl + time.time()
+        return (fp, exp)
+
+    def _ensure_cached(self):
+        no_info = False
+        try:
+            # pylint: disable=access-member-before-definition
+            fp = self._server_schema_fingerprint
+            exp = self._server_schema_expiration
+        except AttributeError:
+            try:
+                with self._open_server_info(self._api.env.server, 'r') as sc:
+                    si = json.load(sc)
+
+                fp = si['fingerprint']
+                exp = si['expiration']
+            except Exception as e:
+                no_info = True
+                if not (isinstance(e, EnvironmentError) and
+                        e.errno == errno.ENOENT):  # pylint: disable=no-member
+                    logger.warning('Failed to load server properties: {}'
+                                   ''.format(e))
+
+        if no_info or exp < time.time() or not Schema._in_cache(fp):
+            (fp, exp) = self._get_schema()
+            _ensure_dir_created(SERVERS_DIR)
+            try:
+                with self._open_server_info(self._api.env.server, 'w') as sc:
+                    json.dump(dict(fingerprint=fp, expiration=exp), sc)
+            except Exception as e:
+                logger.warning('Failed to store server properties: {}'
+                               ''.format(e))
+
+        if not self._dict:
+            self._dict['fingerprint'] = fp
+            schema_info = self._read(self.schema_info_path)
+            self._dict['version'] = schema_info['version']
+            for ns in self.namespaces:
+                self._dict[ns] = _SchemaNameSpace(self, ns)
+
+        self._server_schema_fingerprintr = fp
+        self._server_schema_expiration = exp
+
+    def __getitem__(self, key):
+        self._ensure_cached()
+        return self._dict[key]
+
+    def _open_archive(self, mode, fp=None):
+        if not fp:
+            fp = self['fingerprint']
+        arch_path = self.schema_path_template.format(fp)
+        return _LockedZipFile(arch_path, mode)
+
+    def _store(self, fingerprint, schema={}):
+        _ensure_dir_created(SCHEMA_DIR)
+
+        schema_info = dict(version=schema['version'],
+                           fingerprint=schema['fingerprint'])
+
+        with self._open_archive('w', fingerprint) as zf:
+            # store schema information
+            zf.writestr(self.schema_info_path, json.dumps(schema_info))
+            # store namespaces
+            for namespace in self.namespaces:
+                for member in schema[namespace]:
+                    path = self.ns_member_path_template.format(
+                        namespace,
+                        member['full_name']
+                    )
+                    zf.writestr(path, json.dumps(member))
+
+    def _read(self, path):
+        with self._open_archive('r') as zf:
+            return json.loads(zf.read(path))
+
+    def read_namespace_member(self, namespace, member):
+        path = self.ns_member_path_template.format(namespace, member)
+        return self._read(path)
+
+    def iter_namespace(self, namespace):
+        pattern = self.ns_member_pattern_template.format(namespace)
+        with self._open_archive('r') as zf:
+            for name in zf.namelist():
+                r = re.match(pattern, name)
+                if r:
+                    yield r.groups('name')[0]
+
 
+def get_package(api):
+    try:
+        schema = api._schema
+    except AttributeError:
+        schema = Schema(api)
         object.__setattr__(api, '_schema', schema)
 
     fingerprint = str(schema['fingerprint'])