~jfb-tempo-consulting/unifield-toolbox/ct_psql14

import heapq

import threading

 

class Queue():

    lock = threading.RLock()

    queue = []

    list_items = {}

 

    def add(self, item, prio=1):

        with self.lock:

            if item not in self.list_items:

                self.list_items[item] = prio

                heapq.heappush(self.queue, [prio, item])

            elif self.list_items[item] <= prio:

                return

            else:

                for t in self.queue:

                    if t[1] == item:

                        t[0] = 0 # delete item

                heapq.heappush(self.queue, [prio, item])

 

    def pop(self):

        with self.lock:

            prio = 0

            while prio == 0:

                try:

                    prio, item = heapq.heappop(self.queue)

                except IndexError:

                    # empty queue

                    return False, False

                try:

                    del self.list_items[item]

                except KeyError:

                    pass

            return prio, item

 

    def resfresh(self):

        with self.lock:

            forced_path = os.path.join(SRC_DIR, 'forced_instance')

            if os.path.exists(forced_path):

                with open(forced_path) as forced_path_desc:

                    instance = forced_path_desc.read()

                self.add(instance, -2)

                os.remove(forced_path)

 

            for instance in sorted(os.listdir(SRC_DIR)):

                if instance.startswith('.'):

                    continue

                newbase = os.path.isfile(os.path.join(SRC_DIR, instance, 'base', 'base.tar.7z'))

                if newbase:

                    log('Put %s base on top' % (instance, ))

                    self.add(instance, -1)

                else:

                    self.add(instance)

 

def stopped(delete=False):

    stop_service = os.path.join(SRC_DIR, 'stop_service')

    if os.path.exists(stop_service):

        if delete:

            os.remove(stop_service)

            log('Stopped')

            return True

    return False

 

def process_directory(thread, queue):

 

    psql_port = 5432 + thread

    while True:

        queue.resfresh()

 

        nb, instance = queue.pop()

        if not instance or instance == 'INIT':

            log('thread %s, sleep' % (thread,))

            time.sleep(60)

            if instance == 'INIT':

                with open(TOUCH_FILE_LOOP, 'w') as t_file:

                    t_file.write(time.strftime('%Y-%m-%d%H%M%S'))

                queue.add('INIT')

            continue

 

 

        if nb < 0:

                if stopped():

                    return False

                pg_wal = os.path.join(dest_basebackup, 'pg_wal')

                    for del_recreate in [pg_wal, pg_xlog, os.path.join(dest_basebackup, 'pg_log')]:

                if not os.path.exists(oldwal):

                    log('Unable to copy WAL, wal directory not found %s' % oldwal)

                    continue

                for wal in os.listdir(full_name):

                    full_path_wal = os.path.join(full_name, wal)

                    if wal.endswith('7z') and not wal.startswith('.'):

                        try:

                            un7zip(full_path_wal, oldwal)

                            wal_moved += 1

                        except subprocess.CalledProcessError as e:

                            # try to extract all WAL: UC when new bb generated to unlock a dump

                            error(e.output or e.stderr)

                        except Exception:

                            logger.exception('ERROR')

                    elif wal == 'force_dump':

                        os.remove(full_path_wal)

                        forced_dump = True

                if forced_dump or wal_moved or basebackup_found:

                        psql_start = [os.path.join(PSQL_DIR, 'pg_ctl.exe'),'-o', '-p %s'%psql_port, '-D', to_win(dest_basebackup), '-t', '1200', '-w', 'start']

                        db = psycopg2.connect('dbname=template1 host=127.0.0.1 user=openpg port=%s'%psql_port)

                        if last_wal_date and last_wal_date == restore_date.strftime('%Y%m%d-%H%M%S'):

                                    db = psycopg2.connect('dbname=%s host=127.0.0.1 user=openpg port=%s' % (x[0], psql_port))

                                    pg_dump = [os.path.join(PSQL_DIR, 'pg_dump.exe'), '-h', '127.0.0.1', '-p', psql_port, '-U', 'openpg', '-Fc', '--lock-wait-timeout=120000',  '-f', to_win(dump_file), x[0]]

            thread_touch = '%s-%s' % (TOUCH_FILE_DUMP, thread)

            with open(thread_touch, 'w') as t_file:

    log('Check directories')

    nb_threads = 2

    threads = []

    q = Queue()

    q.add('INIT')

    for x in range(0, nb_threads):

        t = threading.Thread(target=process_directory, args=(x, q))

        threads.append(t)

        t.run()

 

    for t in threads:

        t.join()

        log('Thread ends')

 

    stopped(True)

    log('Process ends')

    sys.exit(0)