Source code for tooz.drivers.kubernetes

#
#    Licensed under the Apache License, Version 2.0 (the "License"); you may
#    not use this file except in compliance with the License. You may obtain
#    a copy of the License at
#
#         http://www.apache.org/licenses/LICENSE-2.0
#
#    Unless required by applicable law or agreed to in writing, software
#    distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
#    WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
#    License for the specific language governing permissions and limitations
#    under the License.

from __future__ import annotations

from typing import Any
import warnings

from kubernetes.client import exceptions as k8s_exc
import sherlock

import tooz
from tooz import coordination
from tooz import locking
from tooz import utils


class KubernetesLock(locking.Lock):
    def __init__(self, name: bytes, namespace: bytes, lock: Any) -> None:
        super().__init__(name)
        self._name = name
        self._namespace = namespace
        self._lock = lock
        self._client = lock.client

    def is_still_owner(self) -> bool:
        if not self._lock.locked():
            return False
        try:
            holder = self._client.read_namespaced_lease(
                self._name, self._namespace
            ).spec.holder_identity
            if holder == self._lock._owner:
                return True
        except k8s_exc.ApiException as e:
            if "Reason: Not Found" not in str(e):
                utils.raise_with_cause(
                    tooz.ToozError, f"operation error: {str(e)}", cause=e
                )
        return False

    # FIXME(stephenfin): We are missing a timeout parameter here
    def acquire(
        self,
        blocking: bool = True,
        shared: bool = False,
        expire: int | None = None,
    ) -> bool:
        if shared:
            raise tooz.NotImplemented("not implemented")
        blocking, timeout = utils.convert_blocking(blocking)
        sherlock.configure(
            expire=expire, timeout=int(timeout) if timeout else timeout
        )
        return bool(self._lock.acquire(blocking=blocking))

    def release(self) -> bool:
        if self._lock.locked():
            try:
                self._lock.release()
            except sherlock.lock.LockException as le:
                msg = "Lock was not set by this process."
                if msg in str(le):
                    return True
                else:
                    raise
            return True
        else:
            return False

    @property
    def acquired(self) -> bool:
        return self._lock.locked() and self.is_still_owner()




class SherlockDriver(coordination.CoordinationDriverCachedRunWatchers):
    """This driver uses the `sherlock`_ client against `kubernetes`_ servers.

    The Kubernetes coordinator url should look like::

      kubernetes://[[?OPTION1=VALUE1[&OPTION2=VALUE2[&...]]]

    Currently the following options will be proxied to the contained client:

    ================  ===============================  ====================
    Name              Source                           Default
    ================  ===============================  ====================
    namespace         'namespace' options key          openstack
    ================  ===============================  ====================

    .. _kubernetes: https://kubernetes.io/
    .. _sherlock: https://sher-lock.readthedocs.io/en/latest/
    """

    #: Default namespace when none is provided.
    K8S_NAMESPACE = "openstack"

    CHARACTERISTICS = (
        coordination.Characteristics.NON_TIMEOUT_BASED,
        coordination.Characteristics.DISTRIBUTED_ACROSS_THREADS,
        coordination.Characteristics.DISTRIBUTED_ACROSS_PROCESSES,
        coordination.Characteristics.DISTRIBUTED_ACROSS_HOSTS,
    )
    """
    Tuple of :py:class:`~tooz.coordination.Characteristics` introspectable
    enum member(s) that can be used to interogate how this driver works.
    """



    def __init__(
        self, member_id: bytes, parsed_url: Any, options: dict[str, Any]
    ) -> None:
        warnings.warn(
            'The kubernetes driver is deprecated and will be removed in '
            'a future release',
            category=DeprecationWarning,
            stacklevel=2,
        )
        super().__init__(member_id, parsed_url, options)
        options = utils.collapse(options)
        self._namespace = options.get('namespace', self.K8S_NAMESPACE)




    def get_lock(self, name: bytes) -> KubernetesLock:
        lock = sherlock.KubernetesLock(
            lock_name=name, k8s_namespace=self._namespace
        )
        return KubernetesLock(name=name, namespace=self._namespace, lock=lock)