sources/sighandler.c: perform grac shutdown for SIGTERM

and and several fixes for tests to live with it

Signed-off-by: roman khapov <[email protected]>

Author: rkhapov

docker/functional/bin/ody-stop

@@ -4,33 +4,37 @@ import os
 import time
 import signal
 import subprocess
+import sys
 
 
 def get_odyssey_pids():
-    print(subprocess.check_output('ps aux | grep odyssey', shell=True).decode('utf-8'))
+    print(subprocess.check_output(
+        'ps aux | grep odyssey', shell=True).decode('utf-8'))
 
     try:
         return list(map(int, subprocess.check_output(['pgrep', 'odyssey']).split()))
     except subprocess.CalledProcessError:
         # non-zero exit code means that there is no odyssey pids
         return []
 
+
 def terminate_gracefully(pid, timeout):
     try:
         os.kill(pid, signal.SIGTERM)
     except ProcessLookupError:
         print(f'Process {pid} already finished or doesnt ever existed')
         return
 
-    print(f'Waiting {timeout} seconds to {pid} finish after SIGTERM...', end='')
+    print(f'Waiting {timeout} seconds to {
+          pid} finish after SIGTERM...', end='')
 
     start = time.time()
     finished = False
 
     while time.time() - start < timeout:
         try:
             output = subprocess.check_output(['ps', '-p', str(pid)], )
-            print(output)
+            print(output.decode('utf-8'))
             if 'odyssey'.encode('utf-8') not in output:
                 finished = True
                 break
@@ -50,16 +54,21 @@ def terminate_gracefully(pid, timeout):
 def main():
     timeout = 5
 
-    pids = get_odyssey_pids()
-    print('Found odyssey pids:', ', '.join(list(map(str, pids))))
+    if len(sys.argv) > 1:
+        pids = [int(sys.argv[1])]
+    else:
+        pids = get_odyssey_pids()
+    print('Odyssey pids to stop:', ', '.join(list(map(str, pids))))
 
     for pid in pids:
         if not terminate_gracefully(pid, timeout):
             print(f'Process {pid} didnt finish within {timeout} seconds')
-            print(subprocess.check_output(['gdb', '-p', str(pid), '--batch', '-ex', 't a a bt', '-ex', 'source /gdb.py', 'mmcoro all bt']))
+            print(subprocess.check_output(['gdb', '-p', str(pid), '--batch', '-ex',
+                  't a a bt', '-ex', 'source /gdb.py', '-ex', 'mmcoro all bt']).decode('utf-8'))
             exit(1)
 
     exit(0)
 
+
 if __name__ == "__main__":
     main()

docker/functional/tests/cascade/runner.sh

@@ -26,9 +26,12 @@ mkdir -p /tmp/gateway
 mkdir -p /tmp/root1
 mkdir -p /tmp/root2
 
+/usr/bin/odyssey /tests/cascade/odyssey-gateway.conf
+sleep 1
+gateway_pid=$(pidof odyssey)
+
 /usr/bin/odyssey /tests/cascade/odyssey-root1.conf
 /usr/bin/odyssey /tests/cascade/odyssey-root2.conf
-/usr/bin/odyssey /tests/cascade/odyssey-gateway.conf
 
 sleep 1
 
@@ -80,4 +83,5 @@ exit(0 if diff < threshold else 1)' $root1_client_processed $root2_client_proces
 
 sleep 1
 
+ody-stop $gateway_pid
 ody-stop

docker/functional/tests/ody_integration_test/pkg/client-server.go

@@ -35,7 +35,8 @@ func usrReadResultWhilesigusr2Test(
 	ch := make(chan error, 1)
 
 	go func(chan error) {
-		_, err := t.Queryx("select pg_sleep(100)")
+		rows, err := t.Queryx("select pg_sleep(10)")
+		rows.Close()
 		ch <- err
 	}(ch)
 
@@ -47,7 +48,12 @@ func usrReadResultWhilesigusr2Test(
 	close(ch)
 
 	if err != nil || !ok {
-		return fmt.Errorf("connection closed or reset\n")
+		return fmt.Errorf("connection closed or reset")
+	}
+
+	err = t.Commit()
+	if err != nil {
+		return fmt.Errorf("commit failed: %w", err)
 	}
 
 	return nil
@@ -64,11 +70,13 @@ func select42(ctx context.Context, ch chan error, wg *sync.WaitGroup) {
 	}
 	defer db.Close()
 
-	if _, err := db.Query("Select 42"); err != nil {
+	rows, err := db.Query("Select 42")
+	if err != nil {
 		ch <- err
 		fmt.Println(err)
 		return
 	}
+	rows.Close()
 
 	fmt.Printf("select 42 OK\n")
 }
@@ -133,7 +141,7 @@ func selectSleep(ctx context.Context, i int, ch chan error, wg *sync.WaitGroup,
 }
 
 const (
-	sleepInterval      = 10
+	sleepInterval      = 5
 	maxCoroutineFailOk = 4
 )
 
@@ -162,6 +170,10 @@ func onlineRestartTest(ctx context.Context) error {
 		return err
 	}
 
+	if err := stopOdyssey(ctx); err != nil {
+		return err
+	}
+
 	if err := ensureOdysseyRunning(ctx); err != nil {
 		return err
 	}
@@ -190,7 +202,9 @@ func onlineRestartTest(ctx context.Context) error {
 			// to make sure previous select was in old ody
 			time.Sleep(1 * time.Second)
 
-			restartOdyssey(ctx)
+			if err = restartOdyssey(ctx); err != nil {
+				return err
+			}
 
 			for i := 0; i < coroutineSleepCnt*2; i++ {
 				wg.Add(1)
@@ -214,11 +228,14 @@ func onlineRestartTest(ctx context.Context) error {
 			return err
 		}
 		time.Sleep(1 * time.Second)
-		fmt.Println("Iter complete")
+		fmt.Printf("Iter %d complete\n", j)
 	}
-	if _, err := signalToProc(syscall.SIGINT, "odyssey"); err != nil {
+
+	if err = stopOdyssey(ctx); err != nil {
+		time.Sleep(time.Second * 1000)
 		return err
 	}
+
 	return nil
 }
 
@@ -259,7 +276,7 @@ func sigusr2Test(
 		return err
 	}
 
-	if _, err := signalToProc(syscall.SIGINT, "odyssey"); err != nil {
+	if err := stopOdyssey(ctx); err != nil {
 		return err
 	}
 	return nil
@@ -272,18 +289,21 @@ func odyClientServerInteractionsTestSet(ctx context.Context) error {
 		fmt.Println(err)
 		return err
 	}
+	logTestDone("usrReadResultWhilesigusr2Test")
 
 	if err := onlineRestartTest(ctx); err != nil {
 		err = fmt.Errorf("online restart error %w", err)
 		fmt.Println(err)
 		return err
 	}
+	logTestDone("onlineRestartTest")
 
 	if err := sigusr2Test(ctx); err != nil {
 		err = fmt.Errorf("sigusr2 error %w", err)
 		fmt.Println(err)
 		return err
 	}
+	logTestDone("sigusr2Test")
 
 	fmt.Println("odyClientServerInteractionsTestSet: Ok")
 

docker/functional/tests/ody_integration_test/pkg/cores.go

@@ -18,7 +18,7 @@ const benchTimeSec = 10
 const timeSleep = 5
 const procName = "odyssey"
 const signal = syscall.SIGTERM
-const testCount = 100
+const testCount = 10
 
 func bunchProcess(ctx context.Context, wg *sync.WaitGroup) {
 	defer wg.Done()
@@ -82,6 +82,7 @@ func odyCoresTestSet(ctx context.Context) error {
 		fmt.Println(err)
 		return err
 	}
+	logTestDone("SigTermAfterHighLoad")
 
 	fmt.Println("odyCoresTestSet: Ok")
 

docker/functional/tests/ody_integration_test/pkg/main.go

@@ -15,7 +15,6 @@ func main() {
 		odyClientServerInteractionsTestSet,
 		odyPkgSyncTestSet,
 		odyShowErrsTestSet,
-		odySignalsTestSet,
 		odyCoresTestSet,
 	} {
 		err := f(ctx)

docker/functional/tests/ody_integration_test/pkg/pkg-sync.go

@@ -52,6 +52,7 @@ func odyPkgSyncTestSet(ctx context.Context) error {
 		fmt.Println(err)
 		return err
 	}
+	logTestDone("syncPackets")
 
 	fmt.Println("odyPkgSyncTestSet: Ok")
 

docker/functional/tests/ody_integration_test/pkg/showerrs.go

@@ -4,7 +4,6 @@ import (
 	"context"
 	"database/sql"
 	"fmt"
-	"syscall"
 	"time"
 
 	"github.com/jmoiron/sqlx"
@@ -44,7 +43,7 @@ func getErrs(ctx context.Context, db *sqlx.DB) (map[string]int, error) {
 func showErrors(ctx context.Context) error {
 	// restarting odyssey drops show errs view, but we have change to request show errors to old instance
 	// so we explicitly kill old od
-	if _, err := signalToProc(syscall.SIGINT, "odyssey"); err != nil {
+	if err := stopOdyssey(ctx); err != nil {
 		return err
 	}
 
@@ -91,7 +90,7 @@ func showErrors(ctx context.Context) error {
 func showErrorsAfterPgRestart(ctx context.Context) error {
 	// restarting odyssey drops show errs view, but we have change to request show errors to old instance
 	// so we explicitly kill old od
-	if _, err := signalToProc(syscall.SIGINT, "odyssey"); err != nil {
+	if err := stopOdyssey(ctx); err != nil {
 		return err
 	}
 
@@ -132,12 +131,14 @@ func odyShowErrsTestSet(ctx context.Context) error {
 		fmt.Println(err)
 		return err
 	}
+	logTestDone("showErrors")
 
 	if err := showErrorsAfterPgRestart(ctx); err != nil {
 		err = fmt.Errorf("show errors failed: %w", err)
 		fmt.Println(err)
 		return err
 	}
+	logTestDone("showErrorsAfterPgRestart")
 
 	fmt.Println("odyShowErrsTestSet: Ok")
 

docker/functional/tests/ody_integration_test/pkg/signals.go

@@ -17,6 +17,7 @@ import (
 
 const pgCtlcluster = "/usr/lib/postgresql/16/bin/pg_ctl"
 const restartOdysseyCmd = "/usr/bin/ody-restart"
+const stopOdysseyCmd = "/usr/bin/ody-stop"
 const startOdysseyCmd = "/usr/bin/ody-start"
 
 func restartPg(ctx context.Context) error {
@@ -63,10 +64,11 @@ func ensureOdysseyRunning(ctx context.Context) error {
 
 func restartOdyssey(ctx context.Context,
 ) error {
-	_, err := exec.CommandContext(ctx, restartOdysseyCmd).Output()
+	output, err := exec.CommandContext(ctx, restartOdysseyCmd).Output()
 	if err != nil {
 		err = fmt.Errorf("error due odyssey restarting %w", err)
 		fmt.Println(err)
+		fmt.Println(string(output))
 		return err
 	}
 	fmt.Print("command restart odyssey executed\n")
@@ -75,6 +77,24 @@ func restartOdyssey(ctx context.Context,
 	return nil
 }
 
+func stopOdyssey(ctx context.Context) error {
+	output, err := exec.CommandContext(ctx, stopOdysseyCmd).Output()
+	if err != nil {
+		err = fmt.Errorf("error due odyssey stop %w", err)
+		fmt.Println(err)
+		fmt.Println(string(output))
+		return err
+	}
+	fmt.Print("command stop odyssey executed\n")
+
+	fmt.Print("stop odyssey: OK\n")
+	return nil
+}
+
+func logTestDone(name string) {
+	fmt.Printf("==== done test %s ====\n", name)
+}
+
 func pidNyName(procName string) (int, error) {
 	d, err := ioutil.ReadFile(fmt.Sprintf("/var/run/%s.pid", procName))
 	if err != nil {
@@ -91,7 +111,7 @@ func signalToProc(sig syscall.Signal, procName string) (*os.Process, error) {
 		fmt.Println(err)
 		return nil, err
 	}
-	fmt.Println(fmt.Sprintf("signalToProc: using pid %d", pid))
+	fmt.Printf("signalToProc: %v to %s using pid %d\n", sig, procName, pid)
 
 	p, err := os.FindProcess(pid)
 	if err != nil {

docker/functional/tests/ody_integration_test/pkg/util.go

@@ -312,6 +312,11 @@ keepalive_usr_timeout 0
 #
 backend_connect_timeout_ms 20000
 
+#
+# Maximum SIGTERM count before hard exit
+#
+max_sigterms_to_die 3
+
 ###
 ### GLOBAL LIMITS
 ###