[MVP] gprebalance

README.Ubuntu.bash

@@ -43,6 +43,7 @@ apt-get install -y \
 	python3-psycopg2 \
 	python3-psutil \
 	python3-yaml \
+	python3-transitions \
 	zlib1g-dev
 
 tee -a /etc/sysctl.conf << EOF

gpMgmt/bin/Makefile

@@ -7,7 +7,7 @@ ifneq "$(wildcard $(top_builddir)/src/Makefile.global)" ""
 include $(top_builddir)/src/Makefile.global
 endif
 
-SUBDIRS = stream gpcheckcat_modules gpconfig_modules gpssh_modules gppylib lib
+SUBDIRS = stream gpcheckcat_modules gpconfig_modules gpssh_modules gppylib lib gprebalance_modules
 SUBDIRS += ifaddrs
 
 $(recurse)
@@ -16,7 +16,8 @@ PROGRAMS= analyzedb gpactivatestandby gpaddmirrors gpcheckcat gpcheckperf \
 	gpcheckresgroupimpl gpconfig gpdeletesystem gpexpand gpinitstandby \
 	gpinitsystem gpload gpload.py gplogfilter gpmovemirrors \
 	gprecoverseg gpreload gpsync gpsd gpssh gpssh-exkeys gpstart \
-	gpstate gpstop minirepro gpmemwatcher gpmemreport gpcheckresgroupv2impl
+	gpstate gpstop minirepro gpmemwatcher gpmemreport gpcheckresgroupv2impl \
+	gprebalance ggrebalance
 
 installdirs:
 	$(MKDIR_P) '$(DESTDIR)$(bindir)/lib'
@@ -109,7 +110,8 @@ unitdevel:
 	@echo "Running pure unit tests..."
 	PYTHONPATH=$(SERVER_SRC):$(SERVER_SBIN):$(PYTHONPATH):$(PYTHONSRC_INSTALL_PYTHON_PATH):$(SRC)/ext:$(SBIN_DIR):$(LIB_DIR):$(PYLIB_DIR)/mock-1.0.1 \
 	    python3 -m unittest discover --verbose -s $(SRC)/gppylib -p "test_unit*.py"
-
+	PYTHONPATH=$(SERVER_SRC):$(SERVER_SBIN):$(PYTHONPATH):$(PYTHONSRC_INSTALL_PYTHON_PATH):$(SRC)/ext:$(SBIN_DIR):$(LIB_DIR):$(PYLIB_DIR)/mock-1.0.1 \
+	    python3 -m unittest discover --verbose -s $(SRC)/gprebalance_modules -t $(SRC)
 
 .PHONY: installcheck-bash
 installcheck-bash:

gpMgmt/bin/gpmovemirrors

@@ -400,7 +400,7 @@ try:
     recoversegOptions = "-i " + newConfig.inputFile + " -B " + str(options.batch_size) + \
                         " -b " + str(options.segment_batch_size) + " -v -a -d " + options.coordinator_data_directory
     if options.logfile_directory != None:
-        recoversegOptions = recoversegOptions + " -l " + str(options.logfile_directory)
+        recoversegOptions = recoversegOptions + f' -l "{str(options.logfile_directory)}"'
     logger.info('About to run gprecoverseg with options: ' + recoversegOptions)
     cmd = GpRecoverSeg("Running gprecoverseg", options=recoversegOptions)
     cmd.run(validateAfter=True)

gpMgmt/bin/gppylib/Makefile

@@ -10,6 +10,7 @@ $(recurse)
 PROGRAMS= gparray.py gpunit unit2
 
 DATA= __init__.py \
+	fault_injection.py \
 	datetimeutils.py \
 	gp_era.py \
 	gpcatalog.py \

gpMgmt/bin/gppylib/commands/gp.py

@@ -453,7 +453,7 @@ def __init__(self,name,directory,ctxt=LOCAL,remoteHost=None):
     def is_shutdown(self):
         for key, value in self.results.split_stdout():
             if key == 'Database cluster state':
-                return value.strip() == 'shut down'
+                return value.strip() == 'shut down' or value.strip() == 'shut down in recovery'
         return False
 
     @staticmethod
@@ -1688,6 +1688,20 @@ def __init__(self, name, options = "", ctxt = LOCAL, remoteHost = None):
        cmdStr = "$GPHOME/bin/gprecoverseg %s" % (options)
        Command.__init__(self,name,cmdStr,ctxt,remoteHost)
 
+class GpMoveMirrors(Command):
+   """
+   This command will execute the gpmovemirror utility
+   """
+
+   def __init__(self, name, options = "", ctxt = LOCAL, remoteHost = None):
+       self.name = name
+       self.options = options
+       self.ctxt = ctxt
+       self.remoteHost = remoteHost
+
+       cmdStr = "$GPHOME/bin/gpmovemirrors %s" % (options)
+       Command.__init__(self,name,cmdStr,ctxt,remoteHost)
+
 class IfAddrs:
     @staticmethod
     def list_addrs(hostname=None, include_loopback=False):

gpMgmt/bin/gppylib/commands/unix.py

@@ -691,6 +691,29 @@ def get_hostname(self):
             raise Exception('Command not yet executed')
         return self.results.stdout.strip()
 
+# --------------port is not busy--------------------
+class PortIsAvailable(Command):
+    def __init__(self, name, port, ctxt=REMOTE, remoteHost=None):
+        self.port = port
+        # Check if port is listening: return code 0 if is in use,
+        # 1 if is available
+        cmdStr = (
+            f"(command -v ss >/dev/null 2>&1 && "
+            f"ss -tuln | grep -q ':{port} ') && "
+            f"echo 'IN_USE' || echo 'AVAILABLE'"
+        )
+        Command.__init__(self, name, cmdStr, ctxt, remoteHost)
+
+    def is_port_available(self) -> bool:
+        """
+        Check if port is available based on command results
+        """
+        if not self.results:
+            return False
+
+        output = self.results.stdout.strip()
+
+        return output == 'AVAILABLE' 
 
 # --------------tcp port is active -----------------------
 class PgPortIsActive(Command):

gpMgmt/bin/gppylib/fault_injection.py

@@ -0,0 +1,41 @@
+#!/usr/bin/env python3
+
+import os
+import threading
+import time
+import signal
+
+GPMGMT_FAULT_POINT = 'GPMGMT_FAULT_POINT'
+GPMGMT_FAULT_DELAY_MS = 'GPMGMT_FAULT_DELAY_MS'
+GPMGMT_FAULT_TYPE = 'GPMGMT_FAULT_TYPE'
+GPMGMT_FAULT_FILE_FLAG = 'GPMGMT_FAULT_FILE_FLAG'
+
+GPMGMT_FAULT_TYPE_SYSPEND = 'suspend'
+
+def inject_fault(fault_point):
+    if GPMGMT_FAULT_POINT in os.environ and fault_point == os.environ[GPMGMT_FAULT_POINT]:
+        if GPMGMT_FAULT_TYPE in os.environ and os.environ[GPMGMT_FAULT_TYPE] == GPMGMT_FAULT_TYPE_SYSPEND:
+            while GPMGMT_FAULT_FILE_FLAG in os.environ and os.path.exists(os.environ[GPMGMT_FAULT_FILE_FLAG]):
+                time.sleep(0.1)
+            return
+        if GPMGMT_FAULT_DELAY_MS in os.environ and int(os.environ[GPMGMT_FAULT_DELAY_MS]) > 0:
+            delay_ms = int(os.environ[GPMGMT_FAULT_DELAY_MS])
+
+            def raise_exception(delay: int):
+                time.sleep(delay / 1000)
+                print('Fault Injection %s' % os.environ[GPMGMT_FAULT_POINT])
+                os.kill(os.getpid(), signal.SIGINT)
+
+            thread = threading.Thread(target=raise_exception, kwargs={"delay": delay_ms})
+            thread.daemon = True
+            thread.start()
+        else:
+            raise Exception('Fault Injection %s' % os.environ[GPMGMT_FAULT_POINT])
+
+# decorator for test purposes
+def wrap_state_func_with_faults(func):
+    def func_with_faults(*args):
+        inject_fault(f'{func.__name__}_begin')
+        func(*args)
+        inject_fault(f'{func.__name__}_end')
+    return func_with_faults