# patroni.dcs.kubernetes module

class patroni.dcs.kubernetes.CoreV1ApiProxy(use_endpoints: bool | None = False, bypass_api_service: bool | None = False)

  Bases: `object`

  Proxy class to work with k8s_client.CoreV1Api() object

  _DEFAULT_RETRIABLE_HTTP_CODES = frozenset({500, 503, 504})
  __init__(use_endpoints: bool | None = False, bypass_api_service: bool | None = False) → None
  configure_retriable_http_codes(retriable_http_codes: List[int]) → None
  configure_timeouts(loop_wait: int, retry_timeout: int | float, ttl: int) → None
  refresh_api_servers_cache() → None
  property use_endpoints: bool

class patroni.dcs.kubernetes.K8sClient

  Bases: `object`

  class ApiClient(bypass_api_service: bool | None = False)

    Bases: `object`

    _API_URL_PREFIX = '/api/v1/namespaces/'
    __init__(bypass_api_service: bool | None = False) → None
    _calculate_timeouts(api_servers: int, timeout: float | None = None) → Tuple[int, float, int]

      Calculate a request timeout and number of retries per single K8s API server node. In case if the timeout per node is too small (less than one second) we will reduce the number of nodes. For the cluster with only one API server node we will try to do 1 retry. No retries for clusters with 2 or more API server nodes. We better rely on switching to a different node.

    _do_http_request(retry: Retry | None, api_servers_cache: List[str], method: str, path: str, **kwargs: Any) → HTTPResponse
    _get_api_servers(api_servers_cache: List[str]) → List[str]
    static _handle_server_response(response: HTTPResponse, _preload_content: bool) → HTTPResponse | K8sObject
    _load_api_servers_cache() → None
    static _make_headers(headers: Dict[str, str] | None) → Dict[str, str]
    _refresh_api_servers_cache(updating_cache: bool | None = False) → None
    property api_servers_cache: List[str]
    call_api(method: str, path: str, headers: Dict[str, str] | None = None, body: Any | None = None, _retry: Retry | None = None, _preload_content: bool = True, _request_timeout: float | None = None, **kwargs: Any) → HTTPResponse | K8sObject
    refresh_api_servers_cache() → None
    request(retry: Retry | None, method: str, path: str, timeout: int | float | Tuple[int | float, int | float] | Timeout | None = None, **kwargs: Any) → HTTPResponse
    set_api_servers_cache_ttl(ttl: int) → None
    set_base_uri(value: str) → None
    set_read_timeout(timeout: int | float) → None
  class CoreV1Api(api_client: ApiClient | None = None)

    Bases: `object`

    __init__(api_client: ApiClient | None = None) → None
  class _K8sObjectTemplate(**kwargs: Any)

    Bases: `K8sObject`

    The template for objects which we create locally, e.g. k8s_client.V1ObjectMeta & co

    __init__(**kwargs: Any) → None
  __init__() → None
  class rest

    Bases: `object`

    exception ApiException(status: int | None = None, reason: str | None = None, http_resp: HTTPResponse | None = None)

      Bases: `Exception`

      __init__(status: int | None = None, reason: str | None = None, http_resp: HTTPResponse | None = None) → None

class patroni.dcs.kubernetes.K8sConfig

  Bases: `object`

  exception ConfigException

    Bases: `Exception`

  __init__() → None
  static _get_by_name(config: Dict[str, List[Dict[str, Any]]], section: str, name: str) → Dict[str, Any] | None
  _make_headers(token: str | None = None, **kwargs: Any) → None
  _pool_config_from_file_or_data(config: Dict[str, str], file_key_name: str, pool_key_name: str) → None
  _read_token_file() → str
  _set_token(token: str) → None
  property headers: Dict[str, str]
  load_incluster_config(ca_certs: str = '/var/run/secrets/kubernetes.io/serviceaccount/ca.crt', token_refresh_interval: timedelta = datetime.timedelta(seconds=60)) → None
  load_kube_config(context: str | None = None) → None
  property server: str

exception patroni.dcs.kubernetes.K8sConnectionFailed

  Bases: `K8sException`

exception patroni.dcs.kubernetes.K8sException

  Bases: `Exception`

class patroni.dcs.kubernetes.K8sObject(kwargs: Dict[str, Any])

  Bases: `object`

  __init__(kwargs: Dict[str, Any]) → None
  classmethod _wrap(parent: str | None, value: Any) → Any
  get(name: str, default: Any | None = None) → Any | None
  to_dict() → Dict[str, Any]

class patroni.dcs.kubernetes.Kubernetes(config: Dict[str, Any], mpp: AbstractMPP)

  Bases: `AbstractDCS`

  __init__(config: Dict[str, Any], mpp: AbstractMPP) → None

    Prepare DCS paths, MPP object, initial values for state information and processing dependencies.

    Parameters:

      - config – `dict`, reference to config section of selected DCS. i.e.: `zookeeper` for zookeeper, `etcd` for etcd, etc…

      - mpp – an object implementing `AbstractMPP` interface.

  __load_cluster(group: str | None, loader: Callable[[Dict[str, Any]], Cluster | Dict[int, Cluster]]) → Cluster | Dict[int, Cluster]
  __target_ref(leader_ip: str, latest_subsets: List[K8sObject], pod: K8sObject) → K8sObject
  _abc_impl = <_abc._abc_data object>
  _cluster_from_nodes(group: str, nodes: Dict[str, K8sObject], pods: Collection[K8sObject]) → Cluster
  property _config_resource_version: str | None
  _create_config_service() → None
  _delete_leader(leader: Leader) → bool

    Unused

  static _isotime() → str
  _load_cluster(path: str, loader: Callable[[Any], Cluster | Dict[int, Cluster]]) → Cluster | Dict[int, Cluster]

    Main abstract method that implements the loading of `Cluster` instance.

    Note

    Internally this method should call the loader method that will build `Cluster` object which represents current state and topology of the cluster in DCS. This method supposed to be called only by the `get_cluster()` method.

    Parameters:

      - path – the path in DCS where to load Cluster(s) from.

      - loader – one of `_postgresql_cluster_loader()` or `_mpp_cluster_loader()`.

    Raise:

      `DCSError` in case of communication problems with DCS. If the current node was running as a primary and exception raised, instance would be demoted.

  _map_subsets(endpoints: Dict[str, Any], ips: List[str]) → None
  _mpp_cluster_loader(path: Dict[str, Any]) → Dict[int, Cluster]

    Load and build all PostgreSQL clusters from a single MPP cluster.

    Parameters:

      path – the path in DCS where to load Cluster(s) from.

    Returns:

      all MPP groups as `dict`, with group IDs as keys and `Cluster` objects as values.

  _patch_or_create(name: str, annotations: Dict[str, Any], resource_version: str | None = None, patch: bool = False, retry: Callable[[...], Any] | None = None, ips: List[str] | None = None) → K8sObject

    Patch or create K8s object, Endpoint or ConfigMap.

    Parameters:

      - name – the name of the object.

      - annotations – mapping of annotations that we want to create/update.

      - resource_version – object should be updated only if the `resource_version` matches provided value.

      - patch – `True` if we know in advance that the object already exists and we should patch it.

      - retry – a callable that will take care of retries

      - ips –

        IP address that we want to put to the subsets of the endpoint. Could have following values:

        - `None` - when we don’t need to touch subset;

        - `[]` - to set subsets to the empty list, when `delete_leader()` method is called;

        - `['ip.add.re.ss']` - when we want to make sure that the subsets of the leader endpoint contains the IP address of the leader, that we get from the `kubernetes.pod_ip`;

        - `['']` - when we want to make sure that the subsets of the leader endpoint contains the IP address of the leader, but `kubernetes.pod_ip` configuration is missing. In this case we will try to take the IP address of the Pod which name matches `name` from the config file.

    Returns:

      the new `V1Endpoints` or `V1ConfigMap` object, that was created or updated.

  _postgresql_cluster_loader(path: Dict[str, Any]) → Cluster

    Load and build the `Cluster` object from DCS, which represents a single PostgreSQL cluster.

    Parameters:

      path – the path in DCS where to load `Cluster` from.

    Returns:

      `Cluster` instance.

  _update_leader(leader: Leader) → bool

    Unused

  _update_leader_with_retry(annotations: Dict[str, Any], resource_version: str | None, ips: List[str]) → bool
  _wait_caches(stop_time: float) → None
  _write_failsafe(value: str) → bool

    Unused

  _write_leader_optime(last_lsn: str) → bool

    Unused

  _write_status(value: str) → bool

    Unused

  attempt_to_acquire_leader() → bool

    Attempt to acquire leader lock.

    Note

    This method should create `/leader` key with the value `_name`.

    The key must be created atomically. In case the key already exists it should not be overwritten and `False` must be returned.

    If key creation fails due to DCS not being accessible or because it is not able to process requests (hopefully temporary), the `DCSError` exception should be raised.

    Returns:

      `True` if key has been created successfully.

  cancel_initialization() → bool

    Removes the `initialize` key for a cluster.

    Returns:

      `True` if successfully committed to DCS.

  client_path(path: str) → str

    Construct the absolute key name from appropriate parts for the DCS type.

    Parameters:

      path – The key name within the current Patroni cluster.

    Returns:

      absolute key name for the current Patroni cluster.

  static compare_ports(p1: K8sObject, p2: K8sObject) → bool
  delete_cluster(*args: Any, **kwargs: Any) → Any

    Delete cluster from DCS.

    Returns:

      `True` if successfully committed to DCS.

  delete_leader(leader: Leader | None, last_lsn: int | None = None) → bool

    Update `optime/leader` and voluntarily remove leader key from DCS.

    This method should remove leader key if current instance is the leader.

    Parameters:

      - leader – `Leader` object with information about the leader.

      - last_lsn – latest checkpoint location in bytes.

    Returns:

      boolean result of called abstract `_delete_leader()`.

  delete_sync_state(version: str | None = None) → bool

    Patch annotations of $SCOPE-sync Endpoint or ConfigMap with empty values.

    Effectively it removes “leader” and “sync_standby” annotations from the object. :param version: last known resource_version for conditional update of the object :returns: True if “delete” was successful

  get_mpp_coordinator() → Cluster | None

    Load the PostgreSQL cluster for the MPP Coordinator.

    Note

    This method is only executed on the worker nodes to find the coordinator.

    Returns:

      Select `Cluster` instance associated with the MPP Coordinator group ID.

  initialize(create_new: bool = True, sysid: str = '') → bool

    Race for cluster initialization.

    This method should atomically create `initialize` key and return `True`, otherwise it should return `False`.

    Parameters:

      - create_new – `False` if the key should already exist (in the case we are setting the system_id).

      - sysid – PostgreSQL cluster system identifier, if specified, is written to the key.

    Returns:

      `True` if key has been created successfully.

  property leader_path: str

    Get the client path for `leader`.

  manual_failover(leader: str | None, candidate: str | None, scheduled_at: datetime | None = None, version: str | None = None) → bool

    Prepare dictionary with given values and set `/failover` key in DCS.

    Parameters:

      - leader – value to set for `leader`.

      - candidate – value to set for `member`.

      - scheduled_at – value converted to ISO date format for `scheduled_at`.

      - version – for conditional update of the key/object.

    Returns:

      `True` if successfully committed to DCS.

  static member(pod: K8sObject) → Member
  patch_or_create(*args: Any, **kwargs: Any) → Any
  patch_or_create_config(annotations: Dict[str, Any], resource_version: str | None = None, patch: bool = False, retry: bool = True) → bool
  reload_config(config: Config | Dict[str, Any]) → None

    Handles dynamic config changes.

    Either cause by changes in the local configuration file + SIGHUP or by changes of dynamic configuration

  retry(method: Callable[[...], Any], *args: Any, **kwargs: Any) → Any
  set_config_value(value: str, version: str | None = None) → bool

    Create or update `/config` key in DCS.

    Parameters:

      - value – new value to set in the `config` key.

      - version – for conditional update of the key/object.

    Returns:

      `True` if successfully committed to DCS.

  set_failover_value(value: str, version: str | None = None) → bool

    Unused

  set_history_value(value: str) → bool

    Set value for `history` in DCS.

    Parameters:

      value – new value of `history` key/object.

    Returns:

      `True` if successfully committed to DCS.

  set_retry_timeout(retry_timeout: int) → None

    Set the new value for retry_timeout.

  set_sync_state_value(value: str, version: str | None = None) → bool

    Unused

  set_ttl(ttl: int) → bool | None

    Set the new ttl value for DCS keys.

  static subsets_changed(last_observed_subsets: List[K8sObject], ip: str, ports: List[K8sObject]) → bool

        >>> ip = '1.2.3.4'
        >>> a = [k8s_client.V1EndpointAddress(ip=ip)]
        >>> s = [k8s_client.V1EndpointSubset(addresses=a)]
        >>> Kubernetes.subsets_changed(s, '1.2.3.5', [])
        True
        >>> s = [k8s_client.V1EndpointSubset(addresses=a, ports=[k8s_client.V1EndpointPort(protocol='TCP', port=1)])]
        >>> Kubernetes.subsets_changed(s, '1.2.3.4', [k8s_client.V1EndpointPort(port=5432)])
        True
        >>> p1 = k8s_client.V1EndpointPort(name='port1', port=1)
        >>> p2 = k8s_client.V1EndpointPort(name='port2', port=2)
        >>> p3 = k8s_client.V1EndpointPort(name='port3', port=3)
        >>> s = [k8s_client.V1EndpointSubset(addresses=a, ports=[p1, p2])]
        >>> Kubernetes.subsets_changed(s, ip, [p2, p3])
        True
        >>> s2 = [k8s_client.V1EndpointSubset(addresses=a, ports=[p2, p1])]
        >>> Kubernetes.subsets_changed(s, ip, [p2, p1])
        False

  take_leader() → bool

    Establish a new leader in DCS.

    Note

    This method should create leader key with value of `_name` and `ttl` of `ttl`.

    Since it could be called only on initial cluster bootstrap it could create this key regardless, overwriting the key if necessary.

    Returns:

      `True` if successfully committed to DCS.

  touch_member(*args: Any, **kwargs: Any) → Any

    Update member key in DCS.

    Note

    This method should create or update key with the name with `/members/` + `_name` and the value of data in a given DCS.

    Parameters:

      data – information about an instance (including connection strings).

    Returns:

      `True` if successfully committed to DCS.

  property ttl: int

    Get current `ttl` value.

  update_leader(cluster: Cluster, last_lsn: int | None, slots: Dict[str, int] | None = None, failsafe: Dict[str, str] | None = None) → bool

    Update `leader` key (or session) ttl, `/status`, and `/failsafe` keys.

    Parameters:

      - cluster – `Cluster` object with information about the current cluster state.

      - last_lsn – absolute WAL LSN in bytes.

      - slots – dictionary with permanent slots `confirmed_flush_lsn`.

      - failsafe – if defined dictionary passed to `write_failsafe()`.

    Returns:

      `True` if `leader` key (or session) has been updated successfully.

  watch(leader_version: str | None, timeout: float) → bool

    Sleep if the current node is a leader, otherwise, watch for changes of leader key with a given timeout.

    Parameters:

      - leader_version – version of a leader key.

      - timeout – timeout in seconds.

    Returns:

      if `True` this will reschedule the next run of the HA cycle.

  write_leader_optime(last_lsn: int) → None

    Write value for WAL LSN to `optime` annotation of the leader object.

    Parameters:

      last_lsn – absolute WAL LSN in bytes.

  write_sync_state(leader: str | None, sync_standby: Collection[str] | None, quorum: int | None, version: str | None = None) → SyncState | None

    Prepare and write annotations to $SCOPE-sync Endpoint or ConfigMap.

    Parameters:

      - leader – name of the leader node that manages /sync key

      - sync_standby – collection of currently known synchronous standby node names

      - quorum – if the node from sync_standby list is doing a leader race it should see at least quorum other nodes from the sync_standby + leader list

      - version – last known resource_version for conditional update of the object

    Returns:

      the new `SyncState` object or None

exception patroni.dcs.kubernetes.KubernetesError(value: Any)

  Bases: `DCSError`

exception patroni.dcs.kubernetes.KubernetesRetriableException(orig: ApiException)

  Bases: `ApiException`

  __init__(orig: ApiException) → None
  property sleeptime: int | None

class patroni.dcs.kubernetes.ObjectCache(dcs: Kubernetes, func: Callable[[...], Any], retry: Retry, condition: Condition, name: str | None = None)

  Bases: `Thread`

  __init__(dcs: Kubernetes, func: Callable[[...], Any], retry: Retry, condition: Condition, name: str | None = None) → None

    This constructor should always be called with keyword arguments. Arguments are:

    group should be None; reserved for future extension when a ThreadGroup class is implemented.

    target is the callable object to be invoked by the run() method. Defaults to None, meaning nothing is called.

    name is the thread name. By default, a unique name is constructed of the form “Thread-N” where N is a small decimal number.

    args is a list or tuple of arguments for the target invocation. Defaults to ().

    kwargs is a dictionary of keyword arguments for the target invocation. Defaults to {}.

    If a subclass overrides the constructor, it must make sure to invoke the base class constructor (Thread.__init__()) before doing anything else to the thread.

  _build_cache() → None
  _do_watch(resource_version: str) → None
  static _finish_response(response: HTTPResponse) → None
  _list() → K8sObject
  _process_event(event: Dict[str, Any]) → None
  _watch(resource_version: str) → HTTPResponse
  copy() → Dict[str, K8sObject]
  delete(name: str, resource_version: str) → Tuple[bool, K8sObject | None]
  get(name: str) → K8sObject | None
  is_ready() → bool

    Must be called only when holding the lock on _condition

  kill_stream() → None
  run() → None

    Method representing the thread’s activity.

    You may override this method in a subclass. The standard run() method invokes the callable object passed to the object’s constructor as the target argument, if any, with sequential and keyword arguments taken from the args and kwargs arguments, respectively.

  set(name: str, value: K8sObject) → Tuple[bool, K8sObject | None]

patroni.dcs.kubernetes._cleanup_temp_files() → None

patroni.dcs.kubernetes._create_temp_file(content: bytes) → str

patroni.dcs.kubernetes._run_and_handle_exceptions(method: Callable[[...], Any], *args: Any, **kwargs: Any) → Any

patroni.dcs.kubernetes.catch_kubernetes_errors(func: Callable[[...], Any]) → Callable[[...], Any]

patroni.dcs.kubernetes.to_camel_case(value: str) → str