QQGoblin

class Ha(object):
   """
    在DCS上更新Postgresql节点的状态
   """
    def touch_member(self):
        with self._member_state_lock:
            data = {
                'conn_url': self.state_handler.connection_string,
                'api_url': self.patroni.api.connection_string,
                'state': self.state_handler.state,

QQGoblin

class Patroni(AbstractPatroniDaemon):
  
      def __init__(self, config):

        super(Patroni, self).__init__(config)
        self.version = __version__
        # 通过配置文件中的配置，动态加载dcs的lib，并初始换
        self.dcs = get_dcs(self.config)
        # 初始化watchdog服务，PS：作为容器运行时不需要
        self.watchdog = Watchdog(self.config)

QQGoblin

def _load_cluster(self):
    stop_time = time.time() + self._retry.deadline
    # 刷新kube-apiserver的地址信息
    self._api.refresh_api_servers_cache()
    try:

        with self._condition:
            # 等待podsCache和kindCache的缓存建立完毕
            self._wait_caches(stop_time)
            # 填充集群中的member信息（Pod信息）以及Endpoint信息

QQGoblin

def get_dcs(config):
    # 获取所有DCS实现的名称，dcs_modules()通过pkgutil查询dcs目录下所有python文件的名称
    # 最终返回一个List，内容为：patroni.dcs.kubernetes 等等。
    modules = dcs_modules()

    for module_name in modules:
        name = module_name.split('.')[-1] # dcs名称：如etcd、kubernetes等等
        if name in config:  # 判断配置文件中是否有对应的名称的Section配置段
            try:
                module = importlib.import_module(module_name)

QQGoblin

class ApiClient(object):

    _API_URL_PREFIX = '/api/v1/namespaces/'

    def __init__(self, bypass_api_service=False):
        self._bypass_api_service = bypass_api_service
        self.pool_manager = urllib3.PoolManager(**k8s_config.pool_config)
        self._base_uri = k8s_config.server
        self._api_servers_cache = [k8s_config.server]
        self._api_servers_cache_updated = 0

QQGoblin

#!/usr/bin/env python3
# -*- coding: utf-8 -*-

import argparse
import json
import logging
import re
import os
from json.decoder import JSONDecodeError