Merge pull request ceph#57817 from phlogistonjohn/jjm-smb-sqlite

mgr/smb: add sqlite internal store backend for smb mgr module

Reviewed-by: Adam King <[email protected]>

Author: adk3798

qa/suites/orch/cephadm/smb/tasks/deploy_smb_mgr_basic.yaml

@@ -17,6 +17,7 @@ tasks:
 - cephadm.configure_samba_client_container:
     role: host.b
 - cephadm:
+    single_host_defaults: true
 
 - cephadm.shell:
     host.a:

qa/suites/orch/cephadm/smb/tasks/deploy_smb_mgr_domain.yaml

@@ -17,6 +17,7 @@ tasks:
 - cephadm.deploy_samba_ad_dc:
     role: host.b
 - cephadm:
+    single_host_defaults: true
 
 - cephadm.shell:
     host.a:

qa/suites/orch/cephadm/smb/tasks/deploy_smb_mgr_res_basic.yaml

@@ -17,6 +17,7 @@ tasks:
 - cephadm.configure_samba_client_container:
     role: host.b
 - cephadm:
+    single_host_defaults: true
 
 - cephadm.shell:
     host.a:

qa/suites/orch/cephadm/smb/tasks/deploy_smb_mgr_res_dom.yaml

@@ -17,6 +17,7 @@ tasks:
 - cephadm.deploy_samba_ad_dc:
     role: host.b
 - cephadm:
+    single_host_defaults: true
 
 - cephadm.shell:
     host.a:

src/pybind/mgr/mgr_module.py

@@ -1,7 +1,22 @@
 import ceph_module  # noqa
 
-from typing import cast, Tuple, Any, Dict, Generic, Optional, Callable, List, \
-    Mapping, NamedTuple, Sequence, Union, Set, TYPE_CHECKING
+from typing import (
+    Any,
+    Callable,
+    Dict,
+    Generic,
+    Iterator,
+    List,
+    Mapping,
+    NamedTuple,
+    Optional,
+    Sequence,
+    Set,
+    TYPE_CHECKING,
+    Tuple,
+    Union,
+    cast,
+)
 if TYPE_CHECKING:
     import sys
     if sys.version_info >= (3, 8):
@@ -17,6 +32,7 @@
 import subprocess
 import threading
 from collections import defaultdict
+from contextlib import contextmanager
 from enum import IntEnum, Enum
 import os
 import rados
@@ -175,6 +191,13 @@ class MgrDBNotReady(RuntimeError):
     pass
 
 
+class MgrDBNotAllowed(MgrDBNotReady):
+    """A more specific subclass of MgrDBNotReady raised when mgr_pool option
+    disabled.
+    """
+    pass
+
+
 class OSDMap(ceph_module.BasePyOSDMap):
     def get_epoch(self) -> int:
         return self._get_epoch()
@@ -1307,6 +1330,7 @@ def close_db(self) -> None:
     def open_db(self) -> Optional[sqlite3.Connection]:
         if not self.pool_exists(self.MGR_POOL_NAME):
             if not self.have_enough_osds():
+                self.log.warning('not enough osds to create mgr pool')
                 return None
             self.create_mgr_pool()
         uri = f"file:///{self.MGR_POOL_NAME}:{self.module_name}/main.db?vfs=ceph"
@@ -1340,12 +1364,29 @@ def db(self) -> sqlite3.Connection:
             return self._db
         db_allowed = self.get_ceph_option("mgr_pool")
         if not db_allowed:
-            raise MgrDBNotReady()
+            raise MgrDBNotAllowed()
         self._db = self.open_db()
         if self._db is None:
             raise MgrDBNotReady()
         return self._db
 
+    @contextmanager
+    def exclusive_db_access(self) -> Iterator[sqlite3.Connection]:
+        """Context manager that grants exclusive access to the manager module sqlite3
+        db connection, while establishing a new db transaction.
+        """
+        with self._db_lock, self.db:
+            yield self.db
+
+    @contextmanager
+    def exclusive_db_cursor(self) -> Iterator[sqlite3.Cursor]:
+        """Context manager that yields a db cursor after getting exclusive
+        access to the manager module sqlite3 connection and a new db
+        transaction.
+        """
+        with self.exclusive_db_access() as db:
+            yield db.cursor()
+
     @property
     def release_name(self) -> str:
         """

src/pybind/mgr/smb/config_store.py

@@ -1,6 +1,6 @@
-from typing import Collection, Dict, Iterator
+from typing import Collection, Dict, Iterator, Optional
 
-from .proto import ConfigEntry, EntryKey, Simplified
+from .proto import ConfigEntry, ConfigStore, EntryKey, FindParams, Simplified
 
 
 class MemConfigEntry:
@@ -97,3 +97,83 @@ def contents(self, ns: str) -> Collection[str]:
 
     def __iter__(self) -> Iterator[EntryKey]:
         return iter(self._entries.keys())
+
+
+class ObjectCachingEntry:
+    """A config entry that wraps a different ConfigEntry and caches the
+    simplified object. If the object is set the cache will be updated. If the
+    object is removed the cached object will be forgotten. The cached object
+    can be manually reset with the `clear_cached_obj` method.
+    """
+
+    def __init__(
+        self, base_entry: ConfigEntry, *, obj: Optional[Simplified] = None
+    ) -> None:
+        self._base = base_entry
+        self._obj = obj
+
+    def clear_cached_obj(self) -> None:
+        self._obj = None
+
+    def set(self, obj: Simplified) -> None:
+        self._obj = None  # if base.set fails, obj will be left unset
+        self._base.set(obj)
+        self._obj = obj
+
+    def get(self) -> Simplified:
+        if self._obj is not None:
+            return self._obj
+        self._obj = self._base.get()
+        return self._obj
+
+    def remove(self) -> bool:
+        self._obj = None
+        return self._base.remove()
+
+    def exists(self) -> bool:
+        return self._base.exists()
+
+    @property
+    def uri(self) -> str:
+        return self._base.uri
+
+    @property
+    def full_key(self) -> EntryKey:
+        return self._base.full_key
+
+
+def find_in_store(
+    store: ConfigStore, ns: str, params: FindParams
+) -> Collection[ConfigEntry]:
+    """Given a ConfigStore and namespace within that store, search the stored
+    objects for matching parameters. Params is a dict that will be compared to
+    the same keys/attributes of the objects being searched. Only exact matches
+    will be returned.
+    If the store implements the FindingConfigStore protocol the operation
+    of finding
+    """
+    # is it a FindingConfigStore?
+    _find_entries = getattr(store, 'find_entries', None)
+    if _find_entries:
+        try:
+            return _find_entries(ns, params)
+        except NotImplementedError:
+            # Allow the store to reject any of the ns/params/whatnot with a
+            # NotImplementedError even if it implements the find_entries
+            # function. This will fall back to the simple-but-slow approach of
+            # deserializing and examining every object.
+            pass
+    return _find_in_store(store, ns, params)
+
+
+def _find_in_store(
+    store: ConfigStore, ns: str, params: FindParams
+) -> Collection[ConfigEntry]:
+    """Fallback mode for find_in_store."""
+    found = []
+    for sub_key in store.contents(ns):
+        entry = store[(ns, sub_key)]
+        obj = entry.get()
+        if all(obj[pkey] == pval for pkey, pval in params.items()):
+            found.append(ObjectCachingEntry(entry, obj=obj))
+    return found

src/pybind/mgr/smb/handler.py

@@ -12,7 +12,9 @@
     cast,
 )
 
+import contextlib
 import logging
+import operator
 import time
 
 from ceph.deployment.service_spec import SMBSpec
@@ -21,6 +23,7 @@
 from .enums import (
     AuthMode,
     CephFSStorageProvider,
+    ConfigNS,
     Intent,
     JoinSourceType,
     LoginAccess,
@@ -355,24 +358,33 @@ def apply(
             incoming = order_resources(inputs)
             for resource in incoming:
                 staging.stage(resource)
-            for resource in incoming:
-                results.append(
-                    self._check(resource, staging, create_only=create_only)
-                )
+            with _store_transaction(staging.destination_store):
+                for resource in incoming:
+                    results.append(
+                        self._check(
+                            resource, staging, create_only=create_only
+                        )
+                    )
         except ErrorResult as err:
             results.append(err)
         except Exception as err:
             log.exception("error updating resource")
-            result = ErrorResult(resource, msg=str(err))
+            msg = str(err)
+            if not msg:
+                # handle the case where the exception has no text
+                msg = f"error updating resource: {type(err)} (see logs for details)"
+            result = ErrorResult(resource, msg=msg)
             results.append(result)
         if results.success:
             log.debug(
                 'successfully updated %s resources. syncing changes to public stores',
                 len(list(results)),
             )
-            results = staging.save()
-            _prune_linked_entries(staging)
-            self._sync_modified(results)
+            with _store_transaction(staging.destination_store):
+                results = staging.save()
+                _prune_linked_entries(staging)
+            with _store_transaction(staging.destination_store):
+                self._sync_modified(results)
         return results
 
     def cluster_ids(self) -> List[str]:
@@ -394,13 +406,15 @@ def user_and_group_ids(self) -> List[str]:
         return list(UsersAndGroupsEntry.ids(self.internal_store))
 
     def all_resources(self) -> List[SMBResource]:
-        return self._search_resources(_Matcher())
+        with _store_transaction(self.internal_store):
+            return self._search_resources(_Matcher())
 
     def matching_resources(self, names: List[str]) -> List[SMBResource]:
         matcher = _Matcher()
         for name in names:
             matcher.parse(name)
-        return self._search_resources(matcher)
+        with _store_transaction(self.internal_store):
+            return self._search_resources(matcher)
 
     def _search_resources(self, matcher: _Matcher) -> List[SMBResource]:
         log.debug("performing search with matcher: %s", matcher)
@@ -822,6 +836,58 @@ def _check_share(
         raise ErrorResult(
             share, msg="path is not a valid directory in volume"
         )
+    name_used_by = _share_name_in_use(staging, share)
+    if name_used_by:
+        raise ErrorResult(
+            share,
+            msg="share name already in use",
+            status={"conflicting_share_id": name_used_by},
+        )
+
+
+def _share_name_in_use(
+    staging: _Staging, share: resources.Share
+) -> Optional[str]:
+    """Returns the share_id value if the share's name is already in
+    use by a different share in the cluster. Returns None if no other
+    shares are using the name.
+    """
+    share_ids = (share.cluster_id, share.share_id)
+    share_ns = str(ConfigNS.SHARES)
+    # look for any duplicate names in the staging area.
+    # these items are already in memory
+    for ekey, res in staging.incoming.items():
+        if ekey[0] != share_ns:
+            continue  # not a share
+        assert isinstance(res, resources.Share)
+        if (res.cluster_id, res.share_id) == share_ids:
+            continue  # this share
+        if (res.cluster_id, res.name) == (share.cluster_id, share.name):
+            return res.share_id
+    # look for any duplicate names in the underyling store
+    found = config_store.find_in_store(
+        staging.destination_store,
+        share_ns,
+        {'cluster_id': share.cluster_id, 'name': share.name},
+    )
+    # remove any shares that are deleted in staging
+    found_curr = [
+        entry for entry in found if entry.full_key not in staging.deleted
+    ]
+    # remove self-share from list
+    id_pair = operator.itemgetter('cluster_id', 'share_id')
+    found_curr = [
+        entry for entry in found_curr if id_pair(entry.get()) != share_ids
+    ]
+    if not found_curr:
+        return None
+    if len(found_curr) != 1:
+        # this should not normally happen
+        log.warning(
+            'multiple shares with one name in cluster: %s',
+            ' '.join(s.get()['share_id'] for s in found_curr),
+        )
+    return found_curr[0].get()['share_id']
 
 
 def _check_join_auths(
@@ -1244,3 +1310,15 @@ def _cephx_data_entity(cluster_id: str) -> str:
     use for data access.
     """
     return f'client.smb.fs.cluster.{cluster_id}'
+
+
+@contextlib.contextmanager
+def _store_transaction(store: ConfigStore) -> Iterator[None]:
+    transaction = getattr(store, 'transaction', None)
+    if not transaction:
+        log.debug("No transaction support for store")
+        yield None
+        return
+    log.debug("Using store transaction")
+    with transaction():
+        yield None