V2.3.1 (#179)

Author: frf12

_deploy.py

@@ -299,7 +299,7 @@ def from_cluster_config(cls, cluster_config):
         if cluster_config.origin_tag:
             conf['tag'] = cluster_config.origin_tag
         if cluster_config.depends:
-            conf['depends'] = cluster_config.depends
+            conf['depends'] = list(cluster_config.depends)
         conf = cls._from_cluster_config(conf, cluster_config)
         inner_config = cls.extract_inner_config(cluster_config, conf)
         return {
@@ -546,6 +546,8 @@ def is_ln_install_mode(self):
         return self._deploy_config.is_ln_install_mode()
 
     def apply_inner_config(self, config):
+        if not isinstance(config, ComponentInnerConfig):
+            config = ComponentInnerConfig(self.name, {} if config is None else config)
         self._inner_config = config
         self._clear_cache_server()
 
@@ -987,7 +989,7 @@ def get_inner_config(component_name):
                 return inner_config.get_component_config(component_name)
         else:
             def get_inner_config(component_name):
-                return {}
+                return ComponentInnerConfig(component_name, {})
 
         self._inner_config = inner_config
         base_dir = self.get_base_dir()

_errno.py

@@ -184,8 +184,8 @@ class InitDirFailedErrorMessage(object):
 EC_OBC_CONNECTION_URL_EMPTY = OBDErrorCodeTemplate(4402, '{server} ob-configserver config error: connection_url is empty')
 EC_OBC_CONNECTION_URL_ERROR = OBDErrorCodeTemplate(4402, '{server} ob-configserver config error: connection_url must be an absolute path')
 EC_OBC_DATABASE_TYPE_ERROR = OBDErrorCodeTemplate(4402, '{server} ob-configserver config error: database_type can only be set to `mysql` or `sqlite3`, and must be in lowercase. ')
-EC_OBC_SQLITE_PERMISSION_DENIED = OBDErrorCodeTemplate(4403, '{ip}: {path}: permission denied.')
-EC_OBC_DATABASE_CONNECT_ERROR = OBDErrorCodeTemplate(4404, '{server}: failed to connect to database: {url}')
+EC_OBC_SQLITE_PERMISSION_DENIED = OBDErrorCodeTemplate(4403, 'ob-configserver connect to sqlite failed: {ip}: {path}: permission denied.')
+EC_OBC_DATABASE_CONNECT_ERROR = OBDErrorCodeTemplate(4404, 'ob-configserver connect to mysql failed: {server}: failed url to connect to database: {url}')
 
 
 

core.py

@@ -2510,7 +2510,7 @@ def redeploy_cluster(self, name, search_repo=True, need_confirm=False):
                 return False
         if not getattr(self.options, 'ignore_standby', False):
             # check if the current cluster's tenant has a standby tenant in other cluster
-            self._call_stdio('start_loading', 'check is exit standby tenant')
+            self._call_stdio('start_loading', 'Check for standby tenant')
             connect_plugins = self.search_py_script_plugin(repositories, 'connect')
             get_standbys_plugins = self.search_py_script_plugin(repositories, 'get_standbys', no_found_act='ignore')
             get_deployment_connections_plugins = self.search_py_script_plugin(repositories, 'get_deployment_connections', no_found_act='ignore')
@@ -2580,9 +2580,9 @@ def destroy_cluster(self, name):
                 self._call_stdio('error', err.EC_UNEXPECTED_EXCEPTION)
                 return False
         if not getattr(self.options, 'ignore_standby', False):
-            self._call_stdio('verbose', 'Check exit standby tenant')
+            self._call_stdio('verbose', 'Check for standby tenant')
             # check if the current cluster's tenant has a standby tenant in other cluster
-            self._call_stdio('start_loading', 'check is exit standby tenant')
+            self._call_stdio('start_loading', 'Check for standby tenant')
             connect_plugins = self.search_py_script_plugin(repositories, 'connect')
             get_standbys_plugins = self.search_py_script_plugin(repositories, 'get_standbys', no_found_act='ignore')
             get_deployment_connections_plugins = self.search_py_script_plugin(repositories, 'get_deployment_connections', no_found_act='ignore')
@@ -2954,7 +2954,6 @@ def upgrade_cluster(self, name):
                             return False
             upgrade_repositories.append(dest_repository)
 
-            self.set_repositories(upgrade_repositories)
             upgrade_check_plugins = self.search_py_script_plugin(upgrade_repositories, 'upgrade_check', no_found_act='warn')
             if current_repository in upgrade_check_plugins:
                 connect_plugin = self.search_py_script_plugin(upgrade_repositories, 'connect')[current_repository]
@@ -2969,6 +2968,7 @@ def upgrade_cluster(self, name):
                 if not self.call_plugin(
                     upgrade_check_plugins[current_repository], current_repository,
                     current_repository=current_repository,
+                    upgrade_repositories=upgrade_repositories,
                     route=route,
                     cursor=cursor
                 ):

example/all-components-min.yaml

@@ -7,7 +7,7 @@
 #   timeout: ssh connection timeout (second), default 30
 oceanbase-ce:
   depends:
-    ob-configserver
+    - ob-configserver
   servers:
     - name: server1
       # Please don't use hostname, only IP can be supported
@@ -80,6 +80,7 @@ obproxy-ce:
   # When the associated configurations are not done, OBD will automatically get the these configurations from the dependent components.
   depends:
     - oceanbase-ce
+    - ob-configserver
   servers:
     - 172.19.33.6
   global:

example/all-components.yaml

@@ -7,7 +7,7 @@
 #   timeout: ssh connection timeout (second), default 30
 oceanbase-ce:
   depends:
-    ob-configserver
+    - ob-configserver
   servers:
     - name: server1
       # Please don't use hostname, only IP can be supported
@@ -80,6 +80,7 @@ obproxy-ce:
   # When the associated configurations are not done, OBD will automatically get the these configurations from the dependent components.
   depends:
     - oceanbase-ce
+    - ob-configserver
   servers:
     - 172.19.33.6
   global:

example/ob-configserver/distributed-with-obproxy-and-configserver-example.yaml

@@ -7,7 +7,7 @@
 #   timeout: ssh connection timeout (second), default 30
 oceanbase-ce:
   depends:
-    ob-configserver
+    - ob-configserver
   servers:
     - name: server1
       # Please don't use hostname, only IP can be supported
@@ -54,6 +54,7 @@ obproxy-ce:
   # When the associated configurations are not done, OBD will automatically get the these configurations from the dependent components.
   depends:
     - oceanbase-ce
+    - ob-configserver
   servers:
     - 192.168.1.5
   global:

plugins/oceanbase-diagnostic-tool/1.0/gather_all.py

@@ -19,12 +19,16 @@
 
 
 from __future__ import absolute_import, division, print_function
-from ssh import LocalClient
+
+import os
 import datetime
-from tool import TimeUtils
+
 from subprocess import call, Popen, PIPE
+
+from ssh import LocalClient
+from tool import TimeUtils
+from _rpm import Version
 import _errno as err
-import os
 
 
 def gather_all(plugin_context, *args, **kwargs):
@@ -101,7 +105,8 @@ def run():
         if not server_config.get('redo_dir'):
             server_config['redo_dir'] = server_config['data_dir']
         if not server_config.get('slog_dir'):
-            server_config['slog_dir'] = '%s/slog' % server_config['redo_dir']
+            mount_key = 'redo_dir' if Version('4.0') > cluster_config.version else 'data_dir'
+            server_config['slog_dir'] = '%s/slog' % server_config[mount_key]
         if not server_config.get('clog_dir'):
             server_config['clog_dir'] = '%s/clog' % server_config['redo_dir']
         clog_dir = server_config['clog_dir']

plugins/oceanbase-diagnostic-tool/1.0/gather_slog.py

@@ -19,12 +19,16 @@
 
 
 from __future__ import absolute_import, division, print_function
-from ssh import LocalClient
+
+import os
 import datetime
-from tool import TimeUtils
+
 from subprocess import call, Popen, PIPE
+
+from ssh import LocalClient
+from tool import TimeUtils
+from _rpm import Version
 import _errno as err
-import os
 
 
 def gather_slog(plugin_context, *args, **kwargs):
@@ -72,7 +76,6 @@ def run():
     options = plugin_context.options
     obdiag_bin = "obdiag"
     cluster_config = plugin_context.cluster_config
-    cluster_name = cluster_config.name
     stdio = plugin_context.stdio
     global_conf = cluster_config.get_global_conf()
     from_option = get_option('from')
@@ -91,7 +94,8 @@ def run():
         if not server_config.get('redo_dir'):
             server_config['redo_dir'] = server_config['data_dir']
         if not server_config.get('slog_dir'):
-            server_config['slog_dir'] = '%s/slog' % server_config['redo_dir']
+            mount_key = 'redo_dir' if Version('4.0') > cluster_config.version else 'data_dir'
+            server_config['slog_dir'] = '%s/slog' % server_config[mount_key]
         data_dir = server_config['slog_dir']
 
     try:

plugins/oceanbase/3.1.0/upgrade_check.py

@@ -23,9 +23,8 @@
 import os
 
 
-def upgrade_check(plugin_context, current_repository, route, cursor, *args, **kwargs):
+def upgrade_check(plugin_context, current_repository, upgrade_repositories, route, cursor, *args, **kwargs):
 
-    repositories = plugin_context.repositories
     options = plugin_context.options
     stdio = plugin_context.stdio
     cluster_config = plugin_context.cluster_config
@@ -56,7 +55,7 @@ def upgrade_check(plugin_context, current_repository, route, cursor, *args, **kw
     while i < n:
         cant_use = False
         node = route[i]
-        repository = repositories[i]
+        repository = upgrade_repositories[i]
         stdio.verbose('route %s-%s use %s. file check begin.' % (node.get('version'), node.get('release'), repository))
         script_dir = os.path.join(repository.repository_dir, 'etc/direct_upgrade') if node.get('direct_upgrade') else os.path.join(repository.repository_dir, 'etc')
         if skip_check is False:

plugins/oceanbase/3.1.0/upgrade_route.py

@@ -106,6 +106,8 @@ def get_node(self, repository):
 
     def findShortestUpgradePath(self, current_repository, dest_repository, stdio):
         start_node = self.get_node(current_repository)
+        if not start_node:
+            return
         queue = [start_node]
         visited = set([start_node])
         finalNode = None

plugins/oceanbase/4.0.0.0/create_tenant.py

@@ -23,10 +23,11 @@
 
 import re
 import time
+from collections import defaultdict
 
 from _errno import EC_OBSERVER_CAN_NOT_MIGRATE_IN
 
-tenant_cursor = None
+tenant_cursor_cache = defaultdict(dict)
 
 
 def parse_size(size):
@@ -57,16 +58,29 @@ def format_size(size, precision=1):
     return format % (size, units[idx])
 
 
-def exec_sql_in_tenant(sql, cursor, tenant, mode, retries=10):
-    global tenant_cursor
-    if not tenant_cursor:
+def exec_sql_in_tenant(sql, cursor, tenant, mode, user='', password='', print_exception=True, retries=20):
+    if not user:
         user = 'SYS' if mode == 'oracle' else 'root'
-        tenant_cursor = cursor.new_cursor(tenant=tenant, user=user)
-        if not tenant_cursor and retries:
-            retries -= 1
-            time.sleep(2)
-            return exec_sql_in_tenant(sql, cursor, tenant, mode, retries)
-    return tenant_cursor.execute(sql)
+    # find tenant ip, port
+    tenant_cursor = None
+    if cursor in tenant_cursor_cache and tenant in tenant_cursor_cache[cursor] and user in tenant_cursor_cache[cursor][tenant]:
+        tenant_cursor = tenant_cursor_cache[cursor][tenant][user]
+    else:
+        query_sql = "select a.SVR_IP,c.SQL_PORT from oceanbase.DBA_OB_UNITS as a, oceanbase.DBA_OB_TENANTS as b, oceanbase.DBA_OB_SERVERS as c  where a.TENANT_ID=b.TENANT_ID and a.SVR_IP=c.SVR_IP and a.svr_port=c.SVR_PORT and TENANT_NAME=%s"
+        tenant_server_ports = cursor.fetchall(query_sql, (tenant, ), raise_exception=False, exc_level='verbose')
+        for tenant_server_port in tenant_server_ports:
+            tenant_ip = tenant_server_port['SVR_IP']
+            tenant_port = tenant_server_port['SQL_PORT']
+            tenant_cursor = cursor.new_cursor(tenant=tenant, user=user, password=password, ip=tenant_ip, port=tenant_port, print_exception=print_exception)
+            if tenant_cursor:
+                if tenant not in tenant_cursor_cache[cursor]:
+                    tenant_cursor_cache[cursor][tenant] = {}
+                tenant_cursor_cache[cursor][tenant][user] = tenant_cursor
+                break
+    if not tenant_cursor and retries:
+        time.sleep(1)
+        return exec_sql_in_tenant(sql, cursor, tenant, mode, user, password, print_exception=print_exception, retries=retries-1)
+    return tenant_cursor.execute(sql, raise_exception=False, exc_level='verbose') if tenant_cursor else False
 
 
 def create_tenant(plugin_context, cursor, create_tenant_options=None, *args, **kwargs):

plugins/oceanbase/4.0.0.0/start_check.py

@@ -264,7 +264,7 @@ def system_memory_check():
             if not server_config.get('ilog_dir'):
                 server_config['ilog_dir'] = '%s/ilog' % server_config['redo_dir']
             if not server_config.get('slog_dir'):
-                server_config['slog_dir'] = '%s/slog' % server_config['redo_dir']
+                server_config['slog_dir'] = '%s/slog' % server_config['data_dir']
             if server_config['redo_dir'] == server_config['data_dir']:
                 keys = ['home_path', 'data_dir', 'clog_dir', 'ilog_dir', 'slog_dir']
             else:

plugins/oceanbase/4.0.0.0/upgrade_check.py

@@ -23,9 +23,8 @@
 import os
 
 
-def upgrade_check(plugin_context, current_repository, route, cursor, *args, **kwargs):
+def upgrade_check(plugin_context, current_repository, upgrade_repositories, route, cursor, *args, **kwargs):
 
-    repositories = plugin_context.repositories
     options = plugin_context.options
     stdio = plugin_context.stdio
     cluster_config = plugin_context.cluster_config
@@ -41,14 +40,14 @@ def upgrade_check(plugin_context, current_repository, route, cursor, *args, **kw
         zones.add(zone)
 
     if len(zones) > 2:
-        tenants = cursor.fetchall("""select a.tenant_name, a.tenant_id, zone_list from oceanbase.DBA_OB_TENANTS as a left join (
+        tenants = cursor.fetchall("""select a.tenant_name, a.tenant_id, group_concat(zone_list separator ';') as zone_list from oceanbase.DBA_OB_TENANTS as a left join (
                                   select zone_list, tenant_id from oceanbase.DBA_OB_RESOURCE_POOLS ) as b 
-                                  on a.tenant_id = b.tenant_id where a.tenant_name not like 'META$%'""")
+                                  on a.tenant_id = b.tenant_id where a.tenant_name not like 'META$%' group by tenant_id""")
         if tenants is False:
             return
         for tenant in tenants:
             zone_list = tenant.get('zone_list', '').split(';')
-            if len(zone_list) < 3:
+            if len(set(zone_list)) < 3:
                 stdio.error('Tenant %s does not meet rolling upgrade conditions (zone number greater than 2).' % tenant.get('tenant_name'))
                 return
 
@@ -57,7 +56,7 @@ def upgrade_check(plugin_context, current_repository, route, cursor, *args, **kw
     while i < n:
         cant_use = False
         node = route[i]
-        repository = repositories[i]
+        repository = upgrade_repositories[i]
         stdio.verbose('route %s-%s use %s. file check begin.' % (node.get('version'), node.get('release'), repository))
         script_dir = os.path.join(repository.repository_dir, 'etc/direct_upgrade') if node.get('direct_upgrade') else os.path.join(repository.repository_dir, 'etc')
         if skip_check is False:

plugins/oceanbase/4.0.0.0/upgrade_route.py

@@ -106,6 +106,8 @@ def get_node(self, repository):
 
     def findShortestUpgradePath(self, current_repository, dest_repository, stdio):
         start_node = self.get_node(current_repository)
+        if not start_node:
+            return
         queue = [start_node]
         visited = set([start_node])
         finalNode = None

plugins/oceanbase/4.2.0.0/start_check.py

@@ -264,7 +264,7 @@ def system_memory_check():
             if not server_config.get('ilog_dir'):
                 server_config['ilog_dir'] = '%s/ilog' % server_config['redo_dir']
             if not server_config.get('slog_dir'):
-                server_config['slog_dir'] = '%s/slog' % server_config['redo_dir']
+                server_config['slog_dir'] = '%s/slog' % server_config['data_dir']
             if server_config['redo_dir'] == server_config['data_dir']:
                 keys = ['home_path', 'data_dir', 'clog_dir', 'ilog_dir', 'slog_dir']
             else:

plugins/oceanbase/4.2.0.0/upgrade_check.py

@@ -26,8 +26,7 @@
 from _stdio import FormtatText
 
 
-def upgrade_check(plugin_context, current_repository, route, cursor, *args, **kwargs):
-    repositories = plugin_context.repositories
+def upgrade_check(plugin_context, current_repository, upgrade_repositories, route, cursor, *args, **kwargs):
     options = plugin_context.options
     stdio = plugin_context.stdio
     cluster_config = plugin_context.cluster_config
@@ -58,14 +57,14 @@ def upgrade_check(plugin_context, current_repository, route, cursor, *args, **kw
         zones.add(zone)
 
     if len(zones) > 2:
-        tenants = cursor.fetchall("""select a.tenant_name, a.tenant_id, zone_list from oceanbase.DBA_OB_TENANTS as a left join (
+        tenants = cursor.fetchall("""select a.tenant_name, a.tenant_id, group_concat(zone_list separator ';') as zone_list from oceanbase.DBA_OB_TENANTS as a left join (
                                   select zone_list, tenant_id from oceanbase.DBA_OB_RESOURCE_POOLS ) as b 
-                                  on a.tenant_id = b.tenant_id where a.tenant_name not like 'META$%'""")
+                                  on a.tenant_id = b.tenant_id where a.tenant_name not like 'META$%' group by tenant_id""")
         if tenants is False:
             return
         for tenant in tenants:
             zone_list = tenant.get('zone_list', '').split(';')
-            if len(zone_list) < 3:
+            if len(set(zone_list)) < 3:
                 stdio.error('Tenant %s does not meet rolling upgrade conditions (zone number greater than 2).' % tenant.get('tenant_name'))
                 return
 
@@ -74,7 +73,7 @@ def upgrade_check(plugin_context, current_repository, route, cursor, *args, **kw
     while i < n:
         cant_use = False
         node = route[i]
-        repository = repositories[i]
+        repository = upgrade_repositories[i]
         stdio.verbose('route %s-%s use %s. file check begin.' % (node.get('version'), node.get('release'), repository))
         script_dir = os.path.join(repository.repository_dir, 'etc/direct_upgrade') if node.get('direct_upgrade') else os.path.join(repository.repository_dir, 'etc')
         if skip_check is False:

plugins/ocp-express/1.0.1/connect.py

@@ -26,6 +26,11 @@
 import _errno as err
 
 
+class ConnectCursor(dict):
+    def close(self):
+        pass
+
+
 class OcpExpressCursor(object):
 
     class Response(object):
@@ -88,7 +93,7 @@ def connect(plugin_context, target_server=None, *args, **kwargs):
     else:
         servers = cluster_config.servers
         stdio.start_loading('Connect to ocp-express')
-    cursors = {}
+    cursors = ConnectCursor()
     for server in servers:
         config = cluster_config.get_server_conf(server)
         username = 'system'