_deploy.py 41.8 KB
Newer Older
O
oceanbase-admin 已提交
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24
# coding: utf-8
# OceanBase Deploy.
# Copyright (C) 2021 OceanBase
#
# This file is part of OceanBase Deploy.
#
# OceanBase Deploy is free software: you can redistribute it and/or modify
# it under the terms of the GNU General Public License as published by
# the Free Software Foundation, either version 3 of the License, or
# (at your option) any later version.
#
# OceanBase Deploy is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
# GNU General Public License for more details.
#
# You should have received a copy of the GNU General Public License
# along with OceanBase Deploy.  If not, see <https://www.gnu.org/licenses/>.


from __future__ import absolute_import, division, print_function

import os
import re
R
Rongfeng Fu 已提交
25
import pickle
O
oceanbase-admin 已提交
26 27 28 29
import getpass
from copy import deepcopy
from enum import Enum

R
Rongfeng Fu 已提交
30 31 32
from ruamel.yaml.comments import CommentedMap

from tool import ConfigUtil, FileUtil, YamlLoader, OrderedDict
O
oceanbase-admin 已提交
33 34 35 36 37
from _manager import Manager
from _repository import Repository


yaml = YamlLoader()
R
Rongfeng Fu 已提交
38 39 40 41 42
DEFAULT_CONFIG_PARSER_MANAGER = None


class ParserError(Exception):
    pass
O
oceanbase-admin 已提交
43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98


class UserConfig(object):

    DEFAULT = {
        'username': getpass.getuser(),
        'password': None,
        'key_file': None,
        'port': 22,
        'timeout': 30
    }

    def __init__(self, username=None, password=None, key_file=None, port=None, timeout=None):
        self.username = username if username else self.DEFAULT['username']
        self.password = password
        self.key_file = key_file if key_file else self.DEFAULT['key_file']
        self.port = port if port else self.DEFAULT['port']
        self.timeout = timeout if timeout else self.DEFAULT['timeout']


class ServerConfig(object):

    def __init__(self, ip, name=None):
        self.ip = ip
        self._name = name

    @property
    def name(self):
        return self._name if self._name else self.ip

    def __str__(self):
        return '%s(%s)' % (self._name, self.ip) if self._name else self.ip

    def __hash__(self):
        return hash(self.__str__())

    def __eq__(self, other):
        if isinstance(other, self.__class__):
            return self.ip == other.ip and self.name == other.name
        if isinstance(other, dict):
            return self.ip == other['ip'] and self.name == other['name']


class ServerConfigFlyweightFactory(object):

    _CACHE = {}

    @staticmethod
    def get_instance(ip, name=None):
        server = ServerConfig(ip, name)
        _key = server.__str__()
        if _key not in ServerConfigFlyweightFactory._CACHE:
            ServerConfigFlyweightFactory._CACHE[_key] = server
        return ServerConfigFlyweightFactory._CACHE[_key]


R
Rongfeng Fu 已提交
99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 174 175 176 177 178 179 180 181 182 183 184 185 186 187 188 189 190 191 192 193 194 195 196 197 198 199 200 201 202 203 204 205 206 207 208 209 210 211 212 213 214 215 216 217 218 219 220 221 222 223 224 225 226 227 228 229 230 231 232 233 234 235 236 237 238 239 240 241 242 243 244 245 246 247 248 249 250 251 252 253 254 255 256 257 258 259 260 261 262 263 264 265 266 267 268 269 270 271 272 273 274 275 276 277 278 279 280 281 282 283 284 285 286 287 288 289 290 291 292 293 294 295 296 297 298 299 300 301
class InnerConfigItem(str):
    pass


class InnerConfig(object):

    def __init__(self, path, yaml_loader):
        self.path = path
        self.yaml_loader = yaml_loader
        self.config = {}
        self._load()

    def _load(self):
        self.config = {}
        try:
            with FileUtil.open(self.path, 'rb') as f:
                config = self.yaml_loader.load(f)
                for component_name in config:
                    self.config[component_name] = {}
                    c_config = config[component_name]
                    for server in c_config:
                        i_config = {}
                        s_config = c_config[server]
                        for key in s_config:
                            i_config[InnerConfigItem(key)] = s_config[key]
                        self.config[component_name][server] = i_config
        except:
            pass

    def _dump(self):
        try:
            stdio = self.yaml_loader.stdio if self.yaml_loader else None
            with FileUtil.open(self.path, 'w', stdio=stdio) as f:
                self.yaml_loader.dump(self.config, f)
            return True
        except:
            pass
        return False

    def dump(self):
        return self._dump()

    def get_component_config(self, component_name):
        return self.config.get(component_name, {})

    def get_server_config(self, component_name, server):
        return self.get_component(component_name).get(server, {})

    def update_component_config(self, component_name, config):
        self.config[component_name] = {}
        for server in config:
            c_config = {}
            data = config[server]
            for key in data:
                if not isinstance(key, InnerConfigItem):
                    key = InnerConfigItem(key)
                c_config[key] = data[key]
            self.config[component_name][server] = c_config


class ConfigParser(object):

    STYLE = ''
    INNER_CONFIG_MAP = {}
    PREFIX = '$_'

    @classmethod
    def _is_inner_item(cls, key):
        return isinstance(key, InnerConfigItem) and key.startswith(cls.PREFIX)
    
    @classmethod
    def extract_inner_config(cls, cluster_config, config):
        return {}
    
    @classmethod
    def _to_cluster_config(cls, component_name, config):
        raise NotImplementedError

    @classmethod
    def to_cluster_config(cls, component_name, config):
        cluster_config = cls._to_cluster_config(component_name, config)
        cluster_config.parser = cls
        return cluster_config

    @classmethod
    def _from_cluster_config(cls, conf, cluster_config):
        raise NotImplementedError
        
    @classmethod
    def from_cluster_config(cls, cluster_config):
        if not cls.STYLE:
            raise NotImplementedError('undefined Style ConfigParser')
        
        conf = CommentedMap()
        conf['style'] = cls.STYLE
        if cluster_config.origin_package_hash:
            conf['package_hash'] = cluster_config.origin_package_hash
        if cluster_config.origin_version:
            conf['version'] = cluster_config.origin_version
        if cluster_config.origin_tag:
            conf['tag'] = cluster_config.origin_tag
        if cluster_config.depends:
            conf['depends'] = cluster_config.depends
        conf = cls._from_cluster_config(conf, cluster_config)
        inner_config = cls.extract_inner_config(cluster_config, conf)
        return {
            'inner_config': inner_config,
            'config': conf
        }
    
    @classmethod
    def get_server_src_conf(cls, cluster_config, component_config, server):
        if server.name not in component_config:
            component_config[server.name] = {}
        return component_config[server.name]

    @classmethod
    def get_global_src_conf(cls, cluster_config, component_config):
        if 'global' not in component_config:
            component_config['global'] = {}
        return component_config['global']


class DefaultConfigParser(ConfigParser):

    STYLE = 'default'

    @classmethod
    def _to_cluster_config(cls, component_name, conf):
        if 'servers' in conf and isinstance(conf['servers'], list):
            servers = []
            for server in conf['servers']:
                if isinstance(server, dict):
                    ip = ConfigUtil.get_value_from_dict(server, 'ip', transform_func=str)
                    name = ConfigUtil.get_value_from_dict(server, 'name', transform_func=str)
                else:
                    ip = server
                    name = None
                if not re.match('^\d{1,3}(\\.\d{1,3}){3}$', ip):
                    continue
                server = ServerConfigFlyweightFactory.get_instance(ip, name)
                if server not in servers:
                    servers.append(server)
        else:
            servers = []
        cluster_config = ClusterConfig(
            servers,
            component_name,
            ConfigUtil.get_value_from_dict(conf, 'version', None, str),
            ConfigUtil.get_value_from_dict(conf, 'tag', None, str),
            ConfigUtil.get_value_from_dict(conf, 'package_hash', None, str)
        )
        if 'global' in conf:
            cluster_config.set_global_conf(conf['global'])
        for server in servers:
            if server.name in conf:
                cluster_config.add_server_conf(server, conf[server.name])
        return cluster_config

    @classmethod
    def extract_inner_config(cls, cluster_config, config):
        inner_config = cluster_config.get_inner_config()
        for server in cluster_config.servers:
            if server.name not in inner_config:
                inner_config[server.name] = {}

        global_config = config.get('global', {})
        keys = list(global_config.keys())
        for key in keys:
            if cls._is_inner_item(key):
                for server in cluster_config.servers:
                    inner_config[server.name][key] = global_config[key]
                del global_config[key]
        
        for server in cluster_config.servers:
            if server.name not in config:
                continue
            server_config = config[server.name]
            keys = list(server_config.keys())
            for key in keys:
                if cls._is_inner_item(key):
                    inner_config[server.name][key] = server_config[key]
                    del server_config[key]
        return inner_config

    @classmethod
    def _from_cluster_config(cls, conf, cluster_config):
        conf['servers'] = []
        for server in cluster_config.servers:
            if server.name == server.ip:
                conf['servers'].append(server.name)
            else:
                conf['servers'].append({'name': server.name, 'ip': server.ip})

        if cluster_config.get_original_global_conf():
            conf['global'] = cluster_config.get_original_global_conf()
        for server in cluster_config.servers:
            server_config = cluster_config.get_original_server_conf(server)
            if server_config:
                conf[server.name] = server_config
        return conf


O
oceanbase-admin 已提交
302 303
class ClusterConfig(object):

R
Rongfeng Fu 已提交
304
    def __init__(self, servers, name, version, tag, package_hash, parser=None):
O
oceanbase-admin 已提交
305
        self.version = version
R
Rongfeng Fu 已提交
306
        self.origin_version = version
O
oceanbase-admin 已提交
307
        self.tag = tag
R
Rongfeng Fu 已提交
308
        self.origin_tag = tag
O
oceanbase-admin 已提交
309
        self.name = name
R
Rongfeng Fu 已提交
310
        self.origin_package_hash = package_hash
O
oceanbase-admin 已提交
311 312 313 314 315 316
        self.package_hash = package_hash
        self._temp_conf = {}
        self._default_conf = {}
        self._global_conf = {}
        self._server_conf = {}
        self._cache_server = {}
R
Rongfeng Fu 已提交
317
        self._original_global_conf = {}
R
Rongfeng Fu 已提交
318 319
        self._inner_config = {}
        servers = list(servers)
O
oceanbase-admin 已提交
320
        self.servers = servers
R
Rongfeng Fu 已提交
321
        self._original_servers = servers # 保证顺序
O
oceanbase-admin 已提交
322 323 324 325
        for server in servers:
            self._server_conf[server] = {}
            self._cache_server[server] = None
        self._deploy_config = None
R
Rongfeng Fu 已提交
326
        self._depends = {}
R
Rongfeng Fu 已提交
327
        self.parser = parser
O
oceanbase-admin 已提交
328 329 330 331 332 333

    def __eq__(self, other):
        if not isinstance(other, self.__class__):
            return False
        return self._global_conf == other._global_conf and self._server_conf == other._server_conf

R
Rongfeng Fu 已提交
334 335 336 337 338 339 340 341 342 343
    def __deepcopy__(self, memo):
        cluster_config = self.__class__(deepcopy(self.servers), self.name, self.version, self.tag, self.package_hash, self.parser)
        copy_attrs = ['origin_tag', 'origin_version', 'origin_package_hash', 'parser']
        deepcopy_attrs = ['_temp_conf', '_default_conf', '_global_conf', '_server_conf', '_cache_server', '_original_global_conf', '_depends', '_original_servers', '_inner_config']
        for attr in copy_attrs:
            setattr(cluster_config, attr, getattr(self, attr))
        for attr in deepcopy_attrs:
            setattr(cluster_config, attr, deepcopy(getattr(self, attr)))
        return cluster_config

O
oceanbase-admin 已提交
344 345 346 347 348
    def set_deploy_config(self, _deploy_config):
        if self._deploy_config is None:
            self._deploy_config = _deploy_config
            return True
        return False
R
Rongfeng Fu 已提交
349 350 351 352 353 354 355 356
    @property
    def original_servers(self):
        return self._original_servers

    @property
    def depends(self):
        return self._depends.keys()

R
Rongfeng Fu 已提交
357 358 359 360 361 362 363 364 365 366 367
    def _clear_cache_server(self):
        for server in self._cache_server:
            self._cache_server[server] = None

    def get_inner_config(self):
        return self._inner_config

    def apply_inner_config(self, config):
        self._inner_config = config
        self._clear_cache_server()

R
Rongfeng Fu 已提交
368
    def add_depend(self, name, cluster_conf):
R
Rongfeng Fu 已提交
369 370 371 372
        if self.name == name:
            raise Exception('Can not set %s as %s\'s dependency' % (name, name))
        if self.name in cluster_conf.depends:
            raise Exception('Circular Dependency: %s and %s' % (self.name, name))
R
Rongfeng Fu 已提交
373 374 375 376 377 378
        self._depends[name] = cluster_conf

    def del_depend(self, name, component_name):
        if component_name in self._depends:
            del self._depends[component_name]

R
Rongfeng Fu 已提交
379
    def get_depend_servers(self, name):
R
Rongfeng Fu 已提交
380 381 382 383 384
        if name not in self._depends:
            return None
        cluster_config = self._depends[name]
        return deepcopy(cluster_config.original_servers)
        
R
Rongfeng Fu 已提交
385
    def get_depend_config(self, name, server=None):
R
Rongfeng Fu 已提交
386 387 388 389 390
        if name not in self._depends:
            return None
        cluster_config = self._depends[name]
        config = cluster_config.get_server_conf_with_default(server) if server else cluster_config.get_global_conf()
        return deepcopy(config)
R
Rongfeng Fu 已提交
391

O
oceanbase-admin 已提交
392 393 394
    def update_server_conf(self, server, key, value, save=True):
        if self._deploy_config is None:
            return False
R
Rongfeng Fu 已提交
395 396
        if server not in self._server_conf:
            return False
O
oceanbase-admin 已提交
397 398 399 400 401 402 403 404 405 406 407 408
        if not self._deploy_config.update_component_server_conf(self.name, server, key, value, save):
            return False
        self._server_conf[server][key] = value
        if self._cache_server[server] is not None:
            self._cache_server[server][key] = value
        return True

    def update_global_conf(self, key, value, save=True):
        if self._deploy_config is None:
            return False
        if not self._deploy_config.update_component_global_conf(self.name, key, value, save):
            return False
R
Rongfeng Fu 已提交
409
        self._update_global_conf(key, value)
O
oceanbase-admin 已提交
410 411 412 413 414
        for server in self._cache_server:
            if self._cache_server[server] is not None:
                self._cache_server[server][key] = value
        return True

R
Rongfeng Fu 已提交
415 416 417 418
    def _update_global_conf(self, key, value):
        self._original_global_conf[key] = value
        self._global_conf[key] = value

O
oceanbase-admin 已提交
419 420 421
    def get_unconfigured_require_item(self, server):
        items = []
        config = self.get_server_conf(server)
R
Rongfeng Fu 已提交
422 423 424 425 426 427 428
        if config is not None:
            for key in self._temp_conf:
                if not self._temp_conf[key].require:
                    continue
                if key in config:
                    continue
                items.append(key)
O
oceanbase-admin 已提交
429 430 431
        return items

    def get_server_conf_with_default(self, server):
R
Rongfeng Fu 已提交
432 433
        if server not in self._server_conf:
            return None
O
oceanbase-admin 已提交
434 435 436 437 438 439 440 441
        config = {}
        for key in self._temp_conf:
            if self._temp_conf[key].default is not None:
                config[key] = self._temp_conf[key].default
        config.update(self.get_server_conf(server))
        return config

    def get_need_redeploy_items(self, server):
R
Rongfeng Fu 已提交
442 443
        if server not in self._server_conf:
            return None
O
oceanbase-admin 已提交
444 445 446 447 448 449 450 451
        items = {}
        config = self.get_server_conf(server)
        for key in config:
            if key in self._temp_conf and self._temp_conf[key].need_redeploy:
                items[key] = config[key]
        return items

    def get_need_restart_items(self, server):
R
Rongfeng Fu 已提交
452 453
        if server not in self._server_conf:
            return None
O
oceanbase-admin 已提交
454 455 456 457 458 459 460 461 462 463 464
        items = {}
        config = self.get_server_conf(server)
        for key in config:
            if key in self._temp_conf and self._temp_conf[key].need_restart:
                items[key] = config[key]
        return items
        
    def update_temp_conf(self, temp_conf):
        self._default_conf = {}
        self._temp_conf = temp_conf
        for key in self._temp_conf:
R
Rongfeng Fu 已提交
465
            if self._temp_conf[key].require and self._temp_conf[key].default is not None:
O
oceanbase-admin 已提交
466 467 468
                self._default_conf[key] = self._temp_conf[key].default
        self.set_global_conf(self._global_conf) # 更新全局配置

R
Rongfeng Fu 已提交
469 470 471 472 473 474
    def get_temp_conf_item(self, key):
        if self._temp_conf:
            return self._temp_conf.get(key)
        else:
            return None
    
R
Rongfeng Fu 已提交
475 476 477 478 479 480 481 482 483 484 485 486 487 488 489 490 491 492 493
    def check_param(self):
        error = []
        if self._temp_conf:
            error += self._check_param(self._global_conf)
            for server in self._server_conf:
                error += self._check_param(self._server_conf[server])
        return not error, set(error)

    def _check_param(self, config):
        error = []
        for key in config:
            item = self._temp_conf.get(key)
            if item:
                try:
                    item.check_value(config[key])
                except Exception as e:
                    error.append(str(e))
        return error

O
oceanbase-admin 已提交
494
    def set_global_conf(self, conf):
R
Rongfeng Fu 已提交
495
        self._original_global_conf = deepcopy(conf)
O
oceanbase-admin 已提交
496
        self._global_conf = deepcopy(self._default_conf)
R
Rongfeng Fu 已提交
497
        self._global_conf.update(self._original_global_conf)
R
Rongfeng Fu 已提交
498
        self._clear_cache_server()
O
oceanbase-admin 已提交
499 500 501 502

    def add_server_conf(self, server, conf):
        if server not in self.servers:
            self.servers.append(server)
R
Rongfeng Fu 已提交
503 504
        if server not in self._original_servers:
            self._original_servers.append(server)
O
oceanbase-admin 已提交
505 506 507 508 509 510 511 512 513 514
        self._server_conf[server] = conf
        self._cache_server[server] = None

    def get_global_conf(self):
        return self._global_conf

    def get_server_conf(self, server):
        if server not in self._server_conf:
            return None
        if self._cache_server[server] is None:
R
Rongfeng Fu 已提交
515 516
            conf = deepcopy(self._inner_config.get(server.name, {}))
            conf.update(self._global_conf)
O
oceanbase-admin 已提交
517 518 519 520
            conf.update(self._server_conf[server])
            self._cache_server[server] = conf
        return self._cache_server[server]

R
Rongfeng Fu 已提交
521 522 523
    def get_original_global_conf(self):
        return self._original_global_conf

R
Rongfeng Fu 已提交
524 525 526
    def get_original_server_conf(self, server):
        return self._server_conf.get(server)

O
oceanbase-admin 已提交
527 528 529 530 531 532 533 534 535 536 537 538

class DeployStatus(Enum):

    STATUS_CONFIGUREING = 'configuring'
    STATUS_CONFIGURED = 'configured'
    STATUS_DEPLOYING = 'delopying'
    STATUS_DEPLOYED = 'deployed'
    STATUS_RUNNING = 'running'
    STATUS_STOPING = 'stoping'
    STATUS_STOPPED = 'stopped'
    STATUS_DESTROYING = 'destroying'
    STATUS_DESTROYED = 'destroyed'
R
Rongfeng Fu 已提交
539
    STATUS_UPRADEING = 'upgrading'
O
oceanbase-admin 已提交
540 541 542 543 544 545 546 547 548 549 550 551


class DeployConfigStatus(Enum):

    UNCHNAGE = 'unchange'
    NEED_RELOAD = 'need reload'
    NEED_RESTART = 'need restart'
    NEED_REDEPLOY = 'need redeploy'


class DeployInfo(object):

R
Rongfeng Fu 已提交
552
    def __init__(self, name, status, components=OrderedDict(), config_status=DeployConfigStatus.UNCHNAGE):
O
oceanbase-admin 已提交
553 554 555 556 557 558 559 560 561 562 563 564 565 566
        self.status = status
        self.name = name
        self.components = components
        self.config_status = config_status

    def __str__(self):
        info = ['%s (%s)' % (self.name, self.status.value)]
        for name in self.components:
            info.append('%s-%s' % (name, self.components[name]))
        return '\n'.join(info)


class DeployConfig(object):

R
Rongfeng Fu 已提交
567
    def __init__(self, yaml_path, yaml_loader=yaml, inner_config=None, config_parser_manager=None):
O
oceanbase-admin 已提交
568 569
        self._user = None
        self.unuse_lib_repository = False
R
Rongfeng Fu 已提交
570
        self.auto_create_tenant = False
R
Rongfeng Fu 已提交
571
        self._inner_config = inner_config
R
Rongfeng Fu 已提交
572
        self.components = OrderedDict()
O
oceanbase-admin 已提交
573 574 575
        self._src_data = None
        self.yaml_path = yaml_path
        self.yaml_loader = yaml_loader
R
Rongfeng Fu 已提交
576 577 578
        self.config_parser_manager = config_parser_manager if config_parser_manager else DEFAULT_CONFIG_PARSER_MANAGER
        if self.config_parser_manager is None:
            raise ParserError('ConfigParaserManager Not Set')
O
oceanbase-admin 已提交
579 580 581 582 583 584
        self._load()

    @property
    def user(self):
        return self._user

R
Rongfeng Fu 已提交
585 586 587 588 589 590 591 592 593 594 595 596 597 598 599 600
    @property
    def inner_config(self):
        return self._inner_config

    @inner_config.setter
    def inner_config(self, inner_config):
        if inner_config:
            def get_inner_config(component_name):
                return inner_config.get_component_config(component_name)
        else:
            def get_inner_config(component_name):
                return {}
        for component_name in self.components:
            self.components[component_name].apply_inner_config(get_inner_config(component_name))
        self._inner_config = inner_config

O
oceanbase-admin 已提交
601 602 603 604 605 606 607
    def set_unuse_lib_repository(self, status):
        if self.unuse_lib_repository != status:
            self.unuse_lib_repository = status
            self._src_data['unuse_lib_repository'] = status
            return self._dump()
        return True

R
Rongfeng Fu 已提交
608 609 610 611 612 613 614
    def set_auto_create_tenant(self, status):
        if self.auto_create_tenant != status:
            self.auto_create_tenant = status
            self._src_data['auto_create_tenant'] = status
            return self._dump()
        return True

R
Rongfeng Fu 已提交
615 616 617 618 619 620 621 622 623 624 625 626 627 628 629 630 631 632 633 634 635 636 637
    def update_component_package_hash(self, component, package_hash, version=None):
        if component not in self.components:
            return False
        ori_data = self._src_data[component]
        src_data = deepcopy(ori_data)
        src_data['package_hash'] = package_hash
        if version:
            src_data['version'] = version
        elif 'version' in src_data:
            del src_data['version']
        if 'tag' in src_data:
            del src_data['tag']
        
        self._src_data[component] = src_data
        if self._dump():
            cluster_config = self.components[component]
            cluster_config.package_hash = src_data.get('package_hash')
            cluster_config.version = src_data.get('version')
            cluster_config.tag = None
            return True
        self._src_data[component] = ori_data
        return False

O
oceanbase-admin 已提交
638
    def _load(self):
R
Rongfeng Fu 已提交
639 640 641 642 643 644 645 646 647 648 649 650 651 652 653 654 655 656 657 658 659 660 661 662 663 664
        with open(self.yaml_path, 'rb') as f:
            depends = {}
            self._src_data = self.yaml_loader.load(f)
            for key in self._src_data:
                if key == 'user':
                    self.set_user_conf(UserConfig(
                        ConfigUtil.get_value_from_dict(self._src_data[key], 'username'),
                        ConfigUtil.get_value_from_dict(self._src_data[key], 'password'),
                        ConfigUtil.get_value_from_dict(self._src_data[key], 'key_file'),
                        ConfigUtil.get_value_from_dict(self._src_data[key], 'port', 0, int),
                        ConfigUtil.get_value_from_dict(self._src_data[key], 'timeout', 0, int),
                    ))
                elif key == 'unuse_lib_repository':
                    self.unuse_lib_repository = self._src_data['unuse_lib_repository']
                elif key == 'auto_create_tenant':
                    self.auto_create_tenant = self._src_data['auto_create_tenant']
                elif issubclass(type(self._src_data[key]), dict):
                    self._add_component(key, self._src_data[key])
                    depends[key] = self._src_data[key].get('depends', [])
            for comp in depends:
                conf = self.components[comp]
                for name in depends[comp]:
                    if name == comp:
                        continue
                    if name in self.components:
                        conf.add_depend(name, self.components[name])
O
oceanbase-admin 已提交
665 666 667
        if not self.user:
            self.set_user_conf(UserConfig())

R
Rongfeng Fu 已提交
668 669 670 671 672 673 674 675 676 677 678 679 680 681 682
    def _separate_config(self):
        if self.inner_config:
            for component_name in self.components:
                cluster_config = self.components[component_name]
                src_data = self._src_data[component_name]
                parser = cluster_config.parser
                if parser:
                    inner_config = parser.extract_inner_config(cluster_config, src_data)
                    self.inner_config.update_component_config(component_name, inner_config)

    def _dump_inner_config(self):
        if self.inner_config:
            self._separate_config()
            self.inner_config.dump()

O
oceanbase-admin 已提交
683 684 685
    def _dump(self):
        try:
            with open(self.yaml_path, 'w') as f:
R
Rongfeng Fu 已提交
686
                self._dump_inner_config()
O
oceanbase-admin 已提交
687 688 689
                self.yaml_loader.dump(self._src_data, f)
            return True
        except:
R
Rongfeng Fu 已提交
690 691
            import logging
            logging.exception('')
O
oceanbase-admin 已提交
692 693 694 695 696 697 698 699 700
            pass
        return False

    def dump(self):
        return self._dump()

    def set_user_conf(self, conf):
        self._user = conf

R
Rongfeng Fu 已提交
701 702 703 704 705 706 707 708 709 710 711 712 713 714 715 716 717 718 719 720 721 722 723 724 725 726 727 728
    def add_depend_for_component(self, component_name, depend_component_name, save=True):
        if component_name not in self.components:
            return False
        if depend_component_name not in self.components:
            return False
        cluster_config = self.components[component_name]
        if depend_component_name in cluster_config.depends:
            return True
        cluster_config.add_depend(depend_component_name, self.components[depend_component_name])
        component_config = self._src_data[component_name]
        if 'depends' not in component_config:
            component_config['depends'] = []
        component_config['depends'].append(depend_component_name)
        return self.dump() if save else True

    def del_depend_for_component(self, component_name, depend_component_name, save=True):
        if component_name not in self.components:
            return False
        if depend_component_name not in self.components:
            return False
        cluster_config = self.components[component_name]
        if depend_component_name not in cluster_config.depends:
            return True
        cluster_config.del_depend(depend_component_name, depend_component_name)
        component_config = self._src_data[component_name]
        component_config['depends'] = cluster_config.depends
        return self.dump() if save else True

O
oceanbase-admin 已提交
729 730 731 732 733 734
    def update_component_server_conf(self, component_name, server, key, value, save=True):
        if component_name not in self.components:
            return False
        cluster_config = self.components[component_name]
        if server not in cluster_config.servers:
            return False
R
Rongfeng Fu 已提交
735 736
        component_config = cluster_config.parser.get_server_src_conf(cluster_config, self._src_data[component_name], server)
        component_config[key] = value
O
oceanbase-admin 已提交
737 738 739 740 741
        return self.dump() if save else True

    def update_component_global_conf(self, component_name, key, value, save=True):
        if component_name not in self.components:
            return False
R
Rongfeng Fu 已提交
742 743 744
        cluster_config = self.components[component_name]
        component_config = cluster_config.parser.get_global_src_conf(cluster_config, self._src_data[component_name])
        component_config[key] = value
O
oceanbase-admin 已提交
745 746
        return self.dump() if save else True

R
Rongfeng Fu 已提交
747 748 749 750 751 752 753 754 755 756 757 758
    def _set_component(self, cluster_config):
        cluster_config.set_deploy_config(self)
        if self.inner_config:
            inner_config = self.inner_config.get_component_config(cluster_config.name)
            if inner_config:
                cluster_config.apply_inner_config(inner_config)
        self.components[cluster_config.name] = cluster_config

    def update_component(self, cluster_config):
        component_name = cluster_config.name
        if cluster_config.parser:
            parser = cluster_config.parser
O
oceanbase-admin 已提交
759
        else:
R
Rongfeng Fu 已提交
760 761 762 763 764 765 766 767 768 769 770 771 772 773 774 775 776 777 778 779 780 781 782 783 784 785 786 787 788 789 790 791 792 793
            conf = self._src_data.get('component_name', {})
            style = conf.get('style', 'default')
            parser = self.config_parser_manager.get_parser(component_name, style)

        conf = parser.from_cluster_config(cluster_config)
        if component_name in self.components:
            self.components[component_name].set_deploy_config(None)
        cluster_config = deepcopy(cluster_config)
        cluster_config.apply_inner_config(conf['inner_config'])
        if self.inner_config:
            self.inner_config.update_component_config(component_name, conf['inner_config'])
        self._src_data[component_name] = conf['config']
        self._set_component(cluster_config)
        return True

    def _add_component(self, component_name, conf):
        parser = self.config_parser_manager.get_parser(component_name, conf.get('style'))
        cluster_config = parser.to_cluster_config(component_name, conf)
        self._set_component(cluster_config)

    def change_component_config_style(self, component_name, style):
        if component_name not in self.components:
            return False

        parser = self.config_parser_manager.get_parser(component_name, style)
        cluster_config = self.components[component_name]
        if cluster_config.parser != parser:
            new_config = parser.from_cluster_config(cluster_config)
            self._add_component(component_name, new_config)
            self.components[component_name].apply_inner_config(new_config['inner_config'])
            if self.inner_config:
                self.inner_config.update_component_config(component_name, new_config['inner_config'])
            self._src_data[component_name] = new_config['config']
        return True
O
oceanbase-admin 已提交
794 795 796 797 798 799


class Deploy(object):

    DEPLOY_STATUS_FILE = '.data'
    DEPLOY_YAML_NAME = 'config.yaml'
R
Rongfeng Fu 已提交
800
    INNER_CONFIG_NAME = 'inner_config.yaml'
R
Rongfeng Fu 已提交
801
    UPRADE_META_NAME = '.upgrade'
O
oceanbase-admin 已提交
802

R
Rongfeng Fu 已提交
803
    def __init__(self, config_dir, config_parser_manager=None, stdio=None):
O
oceanbase-admin 已提交
804 805 806 807 808
        self.config_dir = config_dir
        self.name = os.path.split(config_dir)[1]
        self._info = None
        self._config = None
        self.stdio = stdio
R
Rongfeng Fu 已提交
809
        self.config_parser_manager = config_parser_manager
R
Rongfeng Fu 已提交
810
        self._uprade_meta = None
O
oceanbase-admin 已提交
811 812 813 814 815 816 817 818 819 820 821 822 823 824 825 826

    def use_model(self, name, repository, dump=True):
        self.deploy_info.components[name] = {
            'hash': repository.hash,
            'version': repository.version,
        }
        return self._dump_deploy_info() if dump else True

    @staticmethod
    def get_deploy_file_path(path):
        return os.path.join(path, Deploy.DEPLOY_STATUS_FILE)

    @staticmethod
    def get_deploy_yaml_path(path):
        return os.path.join(path, Deploy.DEPLOY_YAML_NAME)

R
Rongfeng Fu 已提交
827 828 829 830
    @staticmethod
    def get_upgrade_meta_path(path):
        return os.path.join(path, Deploy.UPRADE_META_NAME)

R
Rongfeng Fu 已提交
831 832 833 834
    @staticmethod
    def get_inner_config_path(path):
        return os.path.join(path, Deploy.INNER_CONFIG_NAME)

O
oceanbase-admin 已提交
835 836 837 838 839 840 841 842 843 844 845 846 847 848
    @staticmethod
    def get_temp_deploy_yaml_path(path):
        return os.path.join(path, 'tmp_%s' % Deploy.DEPLOY_YAML_NAME)

    @property
    def deploy_info(self):
        if self._info is None:
            try:
                path = self.get_deploy_file_path(self.config_dir)
                with open(path, 'rb') as f:
                    data = yaml.load(f)
                    self._info = DeployInfo(
                        data['name'],
                        getattr(DeployStatus, data['status'], DeployStatus.STATUS_CONFIGURED),
R
Rongfeng Fu 已提交
849
                        ConfigUtil.get_value_from_dict(data, 'components', OrderedDict()),
O
oceanbase-admin 已提交
850 851 852 853 854 855
                        getattr(DeployConfigStatus, ConfigUtil.get_value_from_dict(data, 'config_status', '_'), DeployConfigStatus.UNCHNAGE),
                    )
            except:
                self._info = DeployInfo(self.name, DeployStatus.STATUS_CONFIGURED)
        return self._info

R
Rongfeng Fu 已提交
856 857 858 859 860 861 862 863 864 865 866 867 868 869 870 871 872 873 874 875 876 877
    def _load_deploy_config(self, path):
        yaml_loader = YamlLoader(stdio=self.stdio)
        deploy_config = DeployConfig(path, yaml_loader=yaml_loader, config_parser_manager=self.config_parser_manager)
        deploy_info = self.deploy_info
        for component_name in deploy_info.components:
            if component_name not in deploy_config.components:
                continue
            config = deploy_info.components[component_name]
            cluster_config = deploy_config.components[component_name]
            if 'version' in config and config['version']:
                cluster_config.version = config['version']
            if 'hash' in config and config['hash']:
                cluster_config.package_hash = config['hash']

        deploy_config.inner_config = InnerConfig(self.get_inner_config_path(self.config_dir), yaml_loader=yaml_loader)
        return deploy_config
                
    @property
    def temp_deploy_config(self):
        path = self.get_temp_deploy_yaml_path(self.config_dir)
        return self._load_deploy_config(path)

O
oceanbase-admin 已提交
878 879 880
    @property
    def deploy_config(self):
        if self._config is None:
R
Rongfeng Fu 已提交
881 882
            path = self.get_deploy_yaml_path(self.config_dir)
            self._config = self._load_deploy_config(path)
O
oceanbase-admin 已提交
883 884
        return self._config

R
Rongfeng Fu 已提交
885 886 887 888 889 890 891 892 893 894 895 896 897 898 899 900 901 902 903 904
    def _get_uprade_meta(self):
        if self._uprade_meta is None and self.deploy_info.status == DeployStatus.STATUS_UPRADEING:
            try:
                path = self.get_upgrade_meta_path(self.config_dir)
                with open(path) as f:
                    self._uprade_meta = yaml.load(f)
            except:
                self.stdio and getattr(self.stdio, 'exception', print)('fail to load uprade meta data')
        return self._uprade_meta

    @property
    def upgrade_ctx(self):
        uprade_meta = self._get_uprade_meta()
        return uprade_meta.get('uprade_ctx') if uprade_meta else None

    @property
    def upgrading_component(self):
        uprade_meta = self._get_uprade_meta()
        return uprade_meta.get('component') if uprade_meta else None

O
oceanbase-admin 已提交
905
    def apply_temp_deploy_config(self):
R
Rongfeng Fu 已提交
906
        self.stdio and getattr(self.stdio, 'verbose', print)('%s apply temp config' % self.name)
O
oceanbase-admin 已提交
907 908 909 910 911 912 913 914 915 916 917 918 919 920 921 922 923 924 925 926 927 928 929 930 931 932 933 934
        src_yaml_path = self.get_temp_deploy_yaml_path(self.config_dir)
        target_src_path = self.get_deploy_yaml_path(self.config_dir)
        try:
            FileUtil.move(src_yaml_path, target_src_path)
            self._config = None
            self.update_deploy_config_status(DeployConfigStatus.UNCHNAGE)
            return True
        except Exception as e:
            self.stdio and getattr(self.stdio, 'exception', print)('mv %s to %s failed, error: \n%s' % (src_yaml_path, target_src_path, e))
        return False

    def _dump_deploy_info(self):
        path = self.get_deploy_file_path(self.config_dir)
        self.stdio and getattr(self.stdio, 'verbose', print)('dump deploy info to %s' % path)
        try:
            with open(path, 'w') as f:
                data = {
                    'name': self.deploy_info.name,
                    'components': self.deploy_info.components,
                    'status': self.deploy_info.status.name,
                    'config_status': self.deploy_info.config_status.name,
                }
                yaml.dump(data, f)
            return True
        except:
            self.stdio and getattr(self.stdio, 'exception', print)('dump deploy info to %s failed' % path)
        return False

R
Rongfeng Fu 已提交
935 936 937 938 939 940 941 942 943 944 945 946 947 948 949 950 951 952 953 954 955 956 957 958 959 960 961 962 963 964 965 966 967 968 969 970 971 972 973 974 975 976 977 978 979 980 981 982 983 984 985 986
    def _update_deploy_status(self, status):
        old = self.deploy_info.status
        self.deploy_info.status = status
        if self._dump_deploy_info():
            return True
        self.deploy_info.status = old
        return False

    def _update_deploy_config_status(self, status):
        old = self.deploy_info.config_status
        self.deploy_info.config_status = status
        if self._dump_deploy_info():
            return True
        self.deploy_info.config_status = old
        return False

    def _dump_upgrade_meta_data(self):
        path = self.get_upgrade_meta_path(self.config_dir)
        self.stdio and getattr(self.stdio, 'verbose', print)('dump upgrade meta data to %s' % path)
        try:
            if self._uprade_meta:
                with open(path, 'wb') as f:
                    yaml.dump(self._uprade_meta, f)
            else:
                FileUtil.rm(path, self.stdio)
            return True
        except:
            self.stdio and getattr(self.stdio, 'exception', print)('dump upgrade meta data to %s failed' % path)
        return False

    def start_upgrade(self, component, **uprade_ctx):
        if self.deploy_info.status != DeployStatus.STATUS_RUNNING:
            return False
        self._uprade_meta = {
            'component': component,
            'uprade_ctx': uprade_ctx
        }
        if self._dump_upgrade_meta_data() and self._update_deploy_status(DeployStatus.STATUS_UPRADEING):
            return True
        self._uprade_meta = None
        return False

    def update_upgrade_ctx(self, **uprade_ctx):
        if self.deploy_info.status != DeployStatus.STATUS_UPRADEING:
            return False
        uprade_meta = deepcopy(self._get_uprade_meta())
        self._uprade_meta['uprade_ctx'].update(uprade_ctx)
        if self._dump_upgrade_meta_data():
            return True
        self._uprade_meta = uprade_meta
        return False

R
Rongfeng Fu 已提交
987 988
    def update_component_repository(self, repository):
        if not self.deploy_config.update_component_package_hash(repository.name, repository.hash, repository.version):
R
Rongfeng Fu 已提交
989
            return False
R
Rongfeng Fu 已提交
990
        self.use_model(repository.name, repository)
R
Rongfeng Fu 已提交
991 992 993 994 995 996 997
        return True

    def stop_upgrade(self, dest_repository=None):
        if self._update_deploy_status(DeployStatus.STATUS_RUNNING):
            self._uprade_meta = None
            self._dump_upgrade_meta_data()
            if dest_repository:
R
Rongfeng Fu 已提交
998
                self.update_component_repository(dest_repository)
R
Rongfeng Fu 已提交
999 1000 1001
            return True
        return False

O
oceanbase-admin 已提交
1002 1003
    def update_deploy_status(self, status):
        if isinstance(status, DeployStatus):
R
Rongfeng Fu 已提交
1004 1005 1006 1007 1008
            if self._update_deploy_status(status):
                if DeployStatus.STATUS_DESTROYED == status:
                    self.deploy_info.components = {}
                    self._dump_deploy_info()
                return True
O
oceanbase-admin 已提交
1009 1010 1011 1012
        return False

    def update_deploy_config_status(self, status):
        if isinstance(status, DeployConfigStatus):
R
Rongfeng Fu 已提交
1013
            return self._update_deploy_config_status(status)
O
oceanbase-admin 已提交
1014 1015
        return False

R
Rongfeng Fu 已提交
1016 1017 1018 1019 1020 1021 1022 1023 1024 1025 1026 1027 1028 1029 1030 1031 1032 1033 1034 1035 1036 1037
    def effect_tip(self):
        cmd_map = {
            DeployConfigStatus.NEED_RELOAD: 'obd cluster reload %s' % self.name,
            DeployConfigStatus.NEED_RESTART: 'obd cluster restart %s --wp' % self.name,
            DeployConfigStatus.NEED_REDEPLOY: 'obd cluster redeploy %s' % self.name,
        }
        if self.deploy_info.config_status in cmd_map:
            return '\nUse `%s` to make changes take effect.' % cmd_map[self.deploy_info.config_status]
        return ''


class ConfigParserManager(Manager):

    RELATIVE_PATH = 'config_parser/'
    
    def __init__(self, home_path, stdio=None):
        super(ConfigParserManager, self).__init__(home_path, stdio)
        self.global_parsers = {
            'default': DefaultConfigParser,
        }
        self.component_parsers = {}

R
Rongfeng Fu 已提交
1038
    def _format_paraser_name(self, style):
R
Rongfeng Fu 已提交
1039 1040 1041 1042 1043 1044 1045 1046 1047 1048 1049 1050
        style = style.title()
        return '%sConfigParser' % style.replace('-', '').replace(' ', '').replace('_', '')

    def _load_paraser(self, lib_path, style):
        from tool import DynamicLoading
        module_name = '%s_config_parser' % style
        file_name = '%s.py' % module_name
        path = os.path.join(lib_path, file_name)
        if os.path.isfile(path):
            DynamicLoading.add_lib_path(lib_path)
            self.stdio and getattr(self.stdio, 'verbose', 'print')('load config parser: %s' % path)
            module = DynamicLoading.import_module(module_name, self.stdio)
R
Rongfeng Fu 已提交
1051
            clz_name = self._format_paraser_name(style)
R
Rongfeng Fu 已提交
1052 1053 1054 1055 1056 1057 1058 1059 1060 1061 1062 1063 1064 1065 1066 1067 1068 1069 1070 1071 1072 1073 1074 1075 1076 1077 1078 1079 1080 1081 1082 1083 1084 1085
            try:
                return getattr(module, clz_name)
            except:
                self.stdio and getattr(self.stdio, 'exception', 'print')('')
                return None
            finally:
                DynamicLoading.remove_lib_path(lib_path)
        else:
            self.stdio and getattr(self.stdio, 'verbose', 'print')('No config parser: %s' % path)
            return None

    def _get_component_parser(self, component, style):
        if component not in self.component_parsers:
            self.component_parsers[component] = {}
        component_parsers = self.component_parsers[component]
        if style not in component_parsers:
            lib_path = os.path.join(self.path, component)
            component_parsers[style] = self._load_paraser(lib_path, style)
        return component_parsers[style]

    def _get_global_parser(self, style):
        if style not in self.global_parsers:
            self.global_parsers[style] = self._load_paraser(self.path, style)
        return self.global_parsers[style]

    def get_parser(self, component='', style=''):
        if not style or style == 'default':
            return self.global_parsers['default']
        if component:
            parser = self._get_component_parser(component, style)
            if parser:
                return parser
        parser = self._get_global_parser(style)
        if not parser:
F
v1.4.0  
frf12 已提交
1086
            raise ParserError('Unsupported configuration style: %s %s' % (component, style))
R
Rongfeng Fu 已提交
1087 1088
        return parser

O
oceanbase-admin 已提交
1089 1090 1091 1092 1093

class DeployManager(Manager):

    RELATIVE_PATH = 'cluster/'
    
R
Rongfeng Fu 已提交
1094
    def __init__(self, home_path, lock_manager=None, stdio=None):
O
oceanbase-admin 已提交
1095
        super(DeployManager, self).__init__(home_path, stdio)
R
Rongfeng Fu 已提交
1096
        self.lock_manager = lock_manager
R
Rongfeng Fu 已提交
1097
        self.config_parser_manager = ConfigParserManager(home_path, stdio)
R
Rongfeng Fu 已提交
1098 1099 1100 1101 1102 1103 1104 1105

    def _lock(self, name, read_only=False):
        if self.lock_manager:
            if read_only:
                return self.lock_manager.deploy_sh_lock(name)
            else:
                return self.lock_manager.deploy_ex_lock(name)
        return True
O
oceanbase-admin 已提交
1106

R
Rongfeng Fu 已提交
1107
    def get_deploy_configs(self, read_only=True):
O
oceanbase-admin 已提交
1108
        configs = []
R
Rongfeng Fu 已提交
1109 1110
        for name in os.listdir(self.path):
            path = os.path.join(self.path, name)
O
oceanbase-admin 已提交
1111
            if os.path.isdir(path):
R
Rongfeng Fu 已提交
1112
                self._lock(name, read_only)
R
Rongfeng Fu 已提交
1113
                configs.append(Deploy(path, config_parser_manager=self.config_parser_manager, stdio=self.stdio))
O
oceanbase-admin 已提交
1114 1115
        return configs

R
Rongfeng Fu 已提交
1116 1117
    def get_deploy_config(self, name, read_only=False):
        self._lock(name, read_only)
O
oceanbase-admin 已提交
1118 1119
        path = os.path.join(self.path, name)
        if os.path.isdir(path):
R
Rongfeng Fu 已提交
1120
            return Deploy(path, config_parser_manager=self.config_parser_manager, stdio=self.stdio)
O
oceanbase-admin 已提交
1121 1122 1123
        return None

    def create_deploy_config(self, name, src_yaml_path):
R
Rongfeng Fu 已提交
1124
        self._lock(name)
O
oceanbase-admin 已提交
1125 1126 1127 1128
        config_dir = os.path.join(self.path, name)
        target_src_path = Deploy.get_deploy_yaml_path(config_dir)
        self._mkdir(config_dir)
        if FileUtil.copy(src_yaml_path, target_src_path, self.stdio):
R
Rongfeng Fu 已提交
1129
            return Deploy(config_dir, config_parser_manager=self.config_parser_manager, stdio=self.stdio)
O
oceanbase-admin 已提交
1130 1131 1132 1133 1134
        else:
            self._rm(config_dir)
            return None
        
    def remove_deploy_config(self, name):
R
Rongfeng Fu 已提交
1135
        self._lock(name)
O
oceanbase-admin 已提交
1136 1137
        config_dir = os.path.join(self.path, name)
        self._rm(config_dir)