Enhance K8s extn for node failure cases (#2033)

Signed-off-by: Yunus Qureshi <[email protected]>

Author: yunus-qureshi

OracleDatabase/SingleInstance/extensions/k8s/lock.py

@@ -18,42 +18,84 @@
 import argparse
 import fcntl
 import tempfile
+import threading, subprocess
 from multiprocessing.connection import Listener, Client
 
 # Multiprocess communication auth key
 AUTHKEY = 'vkidSQkgAHc='
+DIR_LOCK_FILE = os.sep + '.dirlock'
 
-
-def acquire_lock(lock_file, sock_file, block):
+def acquire_lock(lock_file, sock_file, block, heartbeat):
     """
     Acquire a lock on the passed file, block if needed
     :param lock_file:
     :param sock_file:
     :param block:
     :return:
     """
-    print('[%s]: Acquiring lock on %s' % (time.strftime('%Y:%m:%d %H:%M:%S'), lock_file))
-    lock_handle = open(lock_file, 'w')
+
+    # get dir lock first to check lock file existence
+    with open(os.path.dirname(lock_file) + DIR_LOCK_FILE, 'w') as dir_lh:
+        fcntl.flock(dir_lh, fcntl.LOCK_EX)
+        if not os.path.exists(lock_file):
+            print('[%s]: Creating %s' % (time.strftime('%Y:%m:%d %H:%M:%S'), os.path.basename(lock_file)))
+            open(lock_file, 'w').close()
+
+    lock_handle = open(lock_file)
+    print('[%s]: Acquiring lock %s with heartbeat %s secs' %
+         (time.strftime('%Y:%m:%d %H:%M:%S'), os.path.basename(lock_file), heartbeat))
     while True:
         try:
             fcntl.flock(lock_handle, fcntl.LOCK_EX | fcntl.LOCK_NB)
-            print('[%s]: Lock acquired on %s' % (time.strftime('%Y:%m:%d %H:%M:%S'), lock_file))
+            print('[%s]: Lock acquired' % (time.strftime('%Y:%m:%d %H:%M:%S')))
+            with open(os.path.dirname(lock_file) + DIR_LOCK_FILE, 'w') as dir_lh:
+                fcntl.flock(dir_lh, fcntl.LOCK_EX)
+                print('[%s]: Starting heartbeat' % (time.strftime('%Y:%m:%d %H:%M:%S')))
+                os.utime(lock_file, None)
             break
         except IOError as e:
             if not block:
                 print(e)
                 return 1
+
             time.sleep(0.1)
 
+            # to handle stale NFS locks
+            pulse = int(time.time() - os.path.getmtime(lock_file))
+            if heartbeat < pulse:
+                # something is wrong
+                print('[%s]: Lost heartbeat by %s secs' % (time.strftime('%Y:%m:%d %H:%M:%S'), pulse))
+                lock_handle.close()
+                # get dir lock
+                with open(os.path.dirname(lock_file) + DIR_LOCK_FILE, 'w') as dir_lh:
+                    fcntl.flock(dir_lh, fcntl.LOCK_EX)
+                    # pulse check again after acquring dir lock
+                    if heartbeat < int(time.time() - os.path.getmtime(lock_file)):
+                        print('[%s]: Recreating %s' % (time.strftime('%Y:%m:%d %H:%M:%S'), os.path.basename(lock_file)))
+                        os.remove(lock_file)
+                        open(lock_file, 'w').close()
+
+                lock_handle = open(lock_file)
+                print('[%s]: Reacquiring lock %s' %
+                     (time.strftime('%Y:%m:%d %H:%M:%S'), os.path.basename(lock_file)))
+
+
     if os.fork():
         return 0
     else:
         # Spawn a child process to hold on to the lock
         if os.path.exists(sock_file):
             os.remove(sock_file)
-        print('[%s]: Holding on to the lock using %s' % (time.strftime('%Y:%m:%d %H:%M:%S'), sock_file))
+        print('[%s]: Lock held %s' % (time.strftime('%Y:%m:%d %H:%M:%S'), os.path.basename(lock_file)))
         listener = Listener(address=sock_file, authkey=AUTHKEY)
 
+        def listen():
+            while True:
+                conn = listener.accept()
+                if conn.recv():
+                    break
+            release()
+
         def release(sig=None, frame=None):
             """
             Release if the process is stopped/terminated
@@ -67,15 +109,15 @@ def release(sig=None, frame=None):
                 time.sleep(30)
             lock_handle.close()
             listener.close()
-            print('[%s]: Lock released on %s' % (time.strftime('%Y:%m:%d %H:%M:%S'), lock_file))
+            print('[%s]: Lock released %s' % (time.strftime('%Y:%m:%d %H:%M:%S'), os.path.basename(lock_file)))
 
         signal.signal(signal.SIGTERM, release)
         signal.signal(signal.SIGINT, release)
-        while True:
-            conn = listener.accept()
-            if conn.recv():
-                break
-        release()
+        threading.Thread(target=listen).start()
+
+        while not lock_handle.closed:
+            os.utime(lock_file, None)
+            time.sleep(5)
 
 
 def check_lock(sock_file):
@@ -90,7 +132,7 @@ def check_lock(sock_file):
     cl = Client(address=sock_file, authkey=AUTHKEY)
     cl.send(False)
     cl.close()
-    print('[%s]: Lock held' % (time.strftime('%Y:%m:%d %H:%M:%S')))
+    print('[%s]: Lock held %s' % (time.strftime('%Y:%m:%d %H:%M:%S'), os.path.basename(sock_file)))
     return 0
 
 
@@ -102,7 +144,7 @@ def release_lock(sock_file):
     """
     if not os.path.exists(sock_file):
         return 1
-    print('[%s]: Connecting to the lock process %s' % (time.strftime('%Y:%m:%d %H:%M:%S'), sock_file))
+    print('[%s]: Releasing lock %s' % (time.strftime('%Y:%m:%d %H:%M:%S'), os.path.basename(sock_file)))
     cl = Client(address=sock_file, authkey=AUTHKEY)
     cl.send(True)
     cl.close()
@@ -120,14 +162,16 @@ def main():
     parser.add_argument('--release', action='store_true', dest='release')
     parser.add_argument('--file', dest='lock_file')
     parser.add_argument('--block', action='store_true', dest='block')
+    # heartbeat in secs
+    parser.add_argument('--heartbeat', type=int, dest='heartbeat', default=30)
     args = parser.parse_args()
     if not args.lock_file:
         parser.print_help()
         sys.exit()
     # Derive sock_file name from lock_file
     sock_file = os.path.join(tempfile.gettempdir(), os.path.basename(args.lock_file))
     if args.acquire:
-        sys.exit(acquire_lock(args.lock_file, sock_file, args.block))
+        sys.exit(acquire_lock(args.lock_file, sock_file, args.block, args.heartbeat))
     elif args.check:
         sys.exit(check_lock(sock_file))
     elif args.release:

OracleDatabase/SingleInstance/extensions/k8s/startDB.sh

@@ -20,7 +20,9 @@ fi;
 
 export ORACLE_SID=$(grep "$ORACLE_HOME" /etc/oratab | cut -d: -f1)
 
-# Start database in nomount mode, shutdown first to abort any zombie procs on restart
+# Clean up any left over zombie procs from container crash, start database in nomount mode
+ipcs -m | awk ' /[0-9]/ {print $2}' | xargs -n1 ipcrm -m 2> /dev/null
+ipcs -s | awk ' /[0-9]/ {print $2}' | xargs -n1 ipcrm -s 2> /dev/null
 for i in {1..10}; do
   sqlplus / as sysdba << EOF
    shutdown abort;