<commit_before>#! /usr/bin/env python

"""
Initialize, start, stop, or destroy WAL replication mirror segments.

============================= DISCLAIMER =============================
This is a developer tool to assist with development of WAL replication
for mirror segments.  This tool is not meant to be used in production.
It is suggested to only run this tool against a gpdemo cluster that
was initialized with no FileRep mirrors.

Example:
    WITH_MIRRORS=false make create-demo-cluster
======================================================================

Assumptions:
1. Greenplum cluster was compiled with --enable-segwalrep
2. Greenplum cluster was initialized without mirror segments.
3. Cluster is all on one host
4. Greenplum environment is all setup (greenplum_path.sh, MASTER_DATA_DIRECTORY, PGPORT, etc.)
5. Greenplum environment is started
6. Greenplum environment is the same throughout tool usage

Assuming all of the above, you can just run the tool as so:
    ./gpsegwalrep.py [init|start|stop|destroy]
"""

import argparse
import os
import sys
import subprocess
import threading
import datetime
import time
from sets import Set
from gppylib.db import dbconn

PRINT_LOCK = threading.Lock()
THREAD_LOCK = threading.Lock()

def runcommands(commands, thread_name, command_finish, exit_on_error=True):
    output = []

    for command in commands:
        try:
            output.append('%s: Running command... %s' % (datetime.datetime.now(), command))
            with THREAD_LOCK:
                output = output + subprocess.check_output(command, stderr=subprocess.STDOUT, shell=True).split('\n')
        except subprocess.CalledProcessError, e:
            output.append(str(e))
            output.append(e.output)
            if exit_on_error:
                with PRINT_LOCK:
                    for line in output:
                        print '%s:  %s' % (thread_name, line)
                    print ''

                sys.exit(e.returncode)

    output.append('%s: %s' % (datetime.datetime.now(), command_finish))
    with PRINT_LOCK:
        for line in output:
            print '%s:  %s' % (thread_name, line)
        print ''

def displaySegmentConfiguration():
    commands = []
    commands.append("psql postgres -c \"select * from gp_segment_configuration order by content, dbid\"")
    runcommands(commands, "", "")

class InitMirrors():
    ''' Initialize the WAL replication mirror segment '''

    def __init__(self, cluster_config, hostname, init=True):
        self.clusterconfig = cluster_config
        self.segconfigs = cluster_config.get_seg_configs()
        self.hostname = hostname
        self.init = init

    def initThread(self, segconfig, user):
        commands = []
        if self.init:
            primary_port = segconfig.port
            primary_dir = segconfig.datadir
            mirror_dir = cluster_config.get_pair_dir(segconfig)
            mirror_port = cluster_config.get_pair_port(segconfig)
        else:
            primary_port = cluster_config.get_pair_port(segconfig)
            primary_dir = cluster_config.get_pair_dir(segconfig)
            mirror_dir = segconfig.datadir
            mirror_port = segconfig.port

        mirror_contentid = segconfig.content

        if self.init:
            commands.append("echo 'host  replication  %s  samenet  trust' >> %s/pg_hba.conf" % (user, primary_dir))
            commands.append("pg_ctl -D %s reload" % primary_dir)

        # 1. create base backup
        commands.append("rm -rf %s" % mirror_dir);
        commands.append("pg_basebackup -x -R -c fast -E ./pg_log -E ./db_dumps -E ./gpperfmon/data -E ./gpperfmon/logs -D %s -h %s -p %d" % (mirror_dir, self.hostname, primary_port))
        commands.append("mkdir %s/pg_log; mkdir %s/pg_xlog/archive_status" % (mirror_dir, mirror_dir))

        if self.init:
            # 2. update catalog
            catalog_update_query = "select pg_catalog.gp_add_segment_mirror(%d::int2, '%s', '%s', %d, '%s')" % (mirror_contentid, self.hostname, self.hostname, mirror_port, mirror_dir)
            commands.append("PGOPTIONS=\"-c gp_session_role=utility\" psql postgres -c \"%s\"" % catalog_update_query)

        thread_name = 'Mirror content %d' % mirror_contentid
        command_finish = 'Initialized mirror at %s' % mirror_dir
        runcommands(commands, thread_name, command_finish)

    def run(self):
        # Assume db user is current user
        user = subprocess.check_output(["whoami"]).rstrip('\n')

        initThreads = []
        for segconfig in self.segconfigs:
            assert(segconfig.content != GpSegmentConfiguration.MASTER_CONTENT_ID)
            if self.init:
                assert(segconfig.role == GpSegmentConfiguration.ROLE_PRIMARY)
            else:
                assert(segconfig.role == GpSegmentConfiguration.ROLE_MIRROR)
            thread = threading.Thread(target=self.initThread, args=(segconfig, user))
            thread.start()
            initThreads.append(thread)

        for thread in initThreads:
            thread.join()

class StartInstances():
    ''' Start a greenplum segment '''

    def __init__(self, cluster_config, host, wait=False):
        self.clusterconfig = cluster_config
        self.segconfigs = cluster_config.get_seg_configs()
        self.host = host
        self.wait = wait

    def startThread(self, segconfig):
        commands = []
        waitstring = ''
        dbid = segconfig.dbid
        contentid = segconfig.content
        segment_port = segconfig.port
        segment_dir = segconfig.datadir
        segment_role = StartInstances.getRole(contentid)

        # Need to set the dbid to 0 on segments to prevent use in mmxlog records
        if contentid != GpSegmentConfiguration.MASTER_CONTENT_ID:
            dbid = 0

        opts = ("-p %d --gp_dbid=%d --silent-mode=true -i -M %s --gp_contentid=%d --gp_num_contents_in_cluster=%d" %
                (segment_port, dbid, segment_role, contentid, self.clusterconfig.get_num_contents()))

        # Arguments for the master.
        #
        # -E in GPDB will set Gp_entry_postmaster = true;
        # to start master in utility mode, need to remove -E and add -c gp_role=utility
        #
        # we automatically assume people want to start in master only utility mode
        # if the self.clusterconfig.get_num_contents() is 0
        if contentid == GpSegmentConfiguration.MASTER_CONTENT_ID:
            if self.clusterconfig.get_num_contents() == 0:
                opts += " -c gp_role=utility"
            else:
                opts += " -E"

        if self.wait:
            waitstring = "-w -t 180"

        commands.append("pg_ctl -D %s %s -o '%s' start" % (segment_dir, waitstring, opts))
        commands.append("pg_ctl -D %s status" % segment_dir)

        if contentid == GpSegmentConfiguration.MASTER_CONTENT_ID:
            segment_label = 'master'
        elif segconfig.preferred_role == GpSegmentConfiguration.ROLE_PRIMARY:
            segment_label = 'primary'
        else:
            segment_label = 'mirror'

        thread_name = 'Segment %s content %d' % (segment_label, contentid)
        command_finish = 'Started %s segment with content %d and port %d at %s' % (segment_label, contentid, segment_port, segment_dir)
        runcommands(commands, thread_name, command_finish)

    @staticmethod
    def getRole(contentid):
        if contentid == GpSegmentConfiguration.MASTER_CONTENT_ID:
            return 'master'
        else:
            return 'mirrorless'

    def run(self):
        startThreads = []
        for segconfig in self.segconfigs:
            thread = threading.Thread(target=self.startThread, args=(segconfig,))
            thread.start()
            startThreads.append(thread)

        for thread in startThreads:
            thread.join()

class StopInstances():
    ''' Stop all segments'''

    def __init__(self, cluster_config):
        self.clusterconfig = cluster_config
        self.segconfigs = cluster_config.get_seg_configs()

    def stopThread(self, segconfig):
        commands = []
        segment_contentid = segconfig.content
        segment_dir = segconfig.datadir

        if segment_contentid == GpSegmentConfiguration.MASTER_CONTENT_ID:
            segment_type = 'master'
        elif segconfig.preferred_role == GpSegmentConfiguration.ROLE_PRIMARY:
            segment_type = 'primary'
        else:
            segment_type = 'mirror'

        commands.append("pg_ctl -D %s stop" % segment_dir)

        thread_name = 'Segment %s content %d' % (segment_type, segment_contentid)
        command_finish = 'Stopped %s segment at %s' % (segment_type, segment_dir)
        runcommands(commands, thread_name, command_finish)

    def run(self):
        stopThreads = []
        for segconfig in self.segconfigs:
            thread = threading.Thread(target=self.stopThread, args=(segconfig,))
            thread.start()
            stopThreads.append(thread)

        for thread in stopThreads:
            thread.join()

class DestroyMirrors():
    ''' Destroy the WAL replication mirror segment '''

    def __init__(self, cluster_config):
        self.clusterconfig = cluster_config
        self.segconfigs = cluster_config.get_seg_configs()

    def destroyThread(self, segconfig):
        commands = []
        mirror_contentid = segconfig.content
        mirror_dir = segconfig.datadir

        commands.append("pg_ctl -D %s stop" % mirror_dir)
        commands.append("rm -rf %s" % mirror_dir)
        thread_name = 'Mirror content %d' % mirror_contentid
        command_finish = 'Destroyed mirror at %s' % mirror_dir
        runcommands(commands, thread_name, command_finish, False)

        # Let FTS recognize that mirrors are gone.  As a result,
        # primaries will be marked not-in-sync.  If this step is
        # omitted, FTS will stop probing as soon as mirrors are
        # removed from catalog and primaries will be left "in-sync"
        # without mirrors.
        #
        # FIXME: enhance gp_remove_segment_mirror() to do this, so
        # that utility remains simplified.  Remove this stopgap
        # thereafter.
        ForceFTSProbeScan(self.clusterconfig,
                          GpSegmentConfiguration.STATUS_DOWN,
                          GpSegmentConfiguration.NOT_IN_SYNC)

        commands = []
        catalog_update_query = "select pg_catalog.gp_remove_segment_mirror(%d::int2)" % (mirror_contentid)
        commands.append("PGOPTIONS=\"-c gp_session_role=utility\" psql postgres -c \"%s\"" % catalog_update_query)

        command_finish = 'Removed mirror %s from catalog' % mirror_dir
        runcommands(commands, thread_name, command_finish, False)

    def run(self):
        destroyThreads = []
        for segconfig in self.segconfigs:
            assert(segconfig.preferred_role == GpSegmentConfiguration.ROLE_MIRROR)
            thread = threading.Thread(target=self.destroyThread, args=(segconfig,))
            thread.start()
            destroyThreads.append(thread)

        for thread in destroyThreads:
            thread.join()

class GpSegmentConfiguration():
    ROLE_PRIMARY = 'p'
    ROLE_MIRROR = 'm'
    STATUS_DOWN = 'd'
    STATUS_UP = 'u'<commit_msg>gpsegwalrep.py: Bug fix for InitMirrors() and DestroyMirrors()<commit_after>#! /usr/bin/env python

"""
Initialize, start, stop, or destroy WAL replication mirror segments.

============================= DISCLAIMER =============================
This is a developer tool to assist with development of WAL replication
for mirror segments.  This tool is not meant to be used in production.
It is suggested to only run this tool against a gpdemo cluster that
was initialized with no FileRep mirrors.

Example:
    WITH_MIRRORS=false make create-demo-cluster
======================================================================

Assumptions:
1. Greenplum cluster was compiled with --enable-segwalrep
2. Greenplum cluster was initialized without mirror segments.
3. Cluster is all on one host
4. Greenplum environment is all setup (greenplum_path.sh, MASTER_DATA_DIRECTORY, PGPORT, etc.)
5. Greenplum environment is started
6. Greenplum environment is the same throughout tool usage

Assuming all of the above, you can just run the tool as so:
    ./gpsegwalrep.py [init|start|stop|destroy]
"""

import argparse
import os
import sys
import subprocess
import threading
import datetime
import time
from sets import Set
from gppylib.db import dbconn

PRINT_LOCK = threading.Lock()
THREAD_LOCK = threading.Lock()

def runcommands(commands, thread_name, command_finish, exit_on_error=True):
    output = []

    for command in commands:
        try:
            output.append('%s: Running command... %s' % (datetime.datetime.now(), command))
            with THREAD_LOCK:
                output = output + subprocess.check_output(command, stderr=subprocess.STDOUT, shell=True).split('\n')
        except subprocess.CalledProcessError, e:
            output.append(str(e))
            output.append(e.output)
            if exit_on_error:
                with PRINT_LOCK:
                    for line in output:
                        print '%s:  %s' % (thread_name, line)
                    print ''

                sys.exit(e.returncode)

    output.append('%s: %s' % (datetime.datetime.now(), command_finish))
    with PRINT_LOCK:
        for line in output:
            print '%s:  %s' % (thread_name, line)
        print ''

def displaySegmentConfiguration():
    commands = []
    commands.append("psql postgres -c \"select * from gp_segment_configuration order by content, dbid\"")
    runcommands(commands, "", "")

class InitMirrors():
    ''' Initialize the WAL replication mirror segment '''

    def __init__(self, cluster_config, hostname, init=True):
        self.clusterconfig = cluster_config
        self.segconfigs = cluster_config.get_seg_configs()
        self.hostname = hostname
        self.init = init

    def initThread(self, segconfig, user):
        commands = []
        if self.init:
            primary_port = segconfig.port
            primary_dir = segconfig.datadir
            mirror_dir = self.clusterconfig.get_pair_dir(segconfig)
            mirror_port = self.clusterconfig.get_pair_port(segconfig)
        else:
            primary_port = self.clusterconfig.get_pair_port(segconfig)
            primary_dir = self.clusterconfig.get_pair_dir(segconfig)
            mirror_dir = segconfig.datadir
            mirror_port = segconfig.port

        mirror_contentid = segconfig.content

        if self.init:
            commands.append("echo 'host  replication  %s  samenet  trust' >> %s/pg_hba.conf" % (user, primary_dir))
            commands.append("pg_ctl -D %s reload" % primary_dir)

        # 1. create base backup
        commands.append("rm -rf %s" % mirror_dir);
        commands.append("pg_basebackup -x -R -c fast -E ./pg_log -E ./db_dumps -E ./gpperfmon/data -E ./gpperfmon/logs -D %s -h %s -p %d" % (mirror_dir, self.hostname, primary_port))
        commands.append("mkdir %s/pg_log; mkdir %s/pg_xlog/archive_status" % (mirror_dir, mirror_dir))

        if self.init:
            # 2. update catalog
            catalog_update_query = "select pg_catalog.gp_add_segment_mirror(%d::int2, '%s', '%s', %d, '%s')" % (mirror_contentid, self.hostname, self.hostname, mirror_port, mirror_dir)
            commands.append("PGOPTIONS=\"-c gp_session_role=utility\" psql postgres -c \"%s\"" % catalog_update_query)

        thread_name = 'Mirror content %d' % mirror_contentid
        command_finish = 'Initialized mirror at %s' % mirror_dir
        runcommands(commands, thread_name, command_finish)

    def run(self):
        # Assume db user is current user
        user = subprocess.check_output(["whoami"]).rstrip('\n')

        initThreads = []
        for segconfig in self.segconfigs:
            assert(segconfig.content != GpSegmentConfiguration.MASTER_CONTENT_ID)
            if self.init:
                assert(segconfig.role == GpSegmentConfiguration.ROLE_PRIMARY)
            else:
                assert(segconfig.role == GpSegmentConfiguration.ROLE_MIRROR)
            thread = threading.Thread(target=self.initThread, args=(segconfig, user))
            thread.start()
            initThreads.append(thread)

        for thread in initThreads:
            thread.join()

class StartInstances():
    ''' Start a greenplum segment '''

    def __init__(self, cluster_config, host, wait=False):
        self.clusterconfig = cluster_config
        self.segconfigs = cluster_config.get_seg_configs()
        self.host = host
        self.wait = wait

    def startThread(self, segconfig):
        commands = []
        waitstring = ''
        dbid = segconfig.dbid
        contentid = segconfig.content
        segment_port = segconfig.port
        segment_dir = segconfig.datadir
        segment_role = StartInstances.getRole(contentid)

        # Need to set the dbid to 0 on segments to prevent use in mmxlog records
        if contentid != GpSegmentConfiguration.MASTER_CONTENT_ID:
            dbid = 0

        opts = ("-p %d --gp_dbid=%d --silent-mode=true -i -M %s --gp_contentid=%d --gp_num_contents_in_cluster=%d" %
                (segment_port, dbid, segment_role, contentid, self.clusterconfig.get_num_contents()))

        # Arguments for the master.
        #
        # -E in GPDB will set Gp_entry_postmaster = true;
        # to start master in utility mode, need to remove -E and add -c gp_role=utility
        #
        # we automatically assume people want to start in master only utility mode
        # if the self.clusterconfig.get_num_contents() is 0
        if contentid == GpSegmentConfiguration.MASTER_CONTENT_ID:
            if self.clusterconfig.get_num_contents() == 0:
                opts += " -c gp_role=utility"
            else:
                opts += " -E"

        if self.wait:
            waitstring = "-w -t 180"

        commands.append("pg_ctl -D %s %s -o '%s' start" % (segment_dir, waitstring, opts))
        commands.append("pg_ctl -D %s status" % segment_dir)

        if contentid == GpSegmentConfiguration.MASTER_CONTENT_ID:
            segment_label = 'master'
        elif segconfig.preferred_role == GpSegmentConfiguration.ROLE_PRIMARY:
            segment_label = 'primary'
        else:
            segment_label = 'mirror'

        thread_name = 'Segment %s content %d' % (segment_label, contentid)
        command_finish = 'Started %s segment with content %d and port %d at %s' % (segment_label, contentid, segment_port, segment_dir)
        runcommands(commands, thread_name, command_finish)

    @staticmethod
    def getRole(contentid):
        if contentid == GpSegmentConfiguration.MASTER_CONTENT_ID:
            return 'master'
        else:
            return 'mirrorless'

    def run(self):
        startThreads = []
        for segconfig in self.segconfigs:
            thread = threading.Thread(target=self.startThread, args=(segconfig,))
            thread.start()
            startThreads.append(thread)

        for thread in startThreads:
            thread.join()

class StopInstances():
    ''' Stop all segments'''

    def __init__(self, cluster_config):
        self.clusterconfig = cluster_config
        self.segconfigs = cluster_config.get_seg_configs()

    def stopThread(self, segconfig):
        commands = []
        segment_contentid = segconfig.content
        segment_dir = segconfig.datadir

        if segment_contentid == GpSegmentConfiguration.MASTER_CONTENT_ID:
            segment_type = 'master'
        elif segconfig.preferred_role == GpSegmentConfiguration.ROLE_PRIMARY:
            segment_type = 'primary'
        else:
            segment_type = 'mirror'

        commands.append("pg_ctl -D %s stop" % segment_dir)

        thread_name = 'Segment %s content %d' % (segment_type, segment_contentid)
        command_finish = 'Stopped %s segment at %s' % (segment_type, segment_dir)
        runcommands(commands, thread_name, command_finish)

    def run(self):
        stopThreads = []
        for segconfig in self.segconfigs:
            thread = threading.Thread(target=self.stopThread, args=(segconfig,))
            thread.start()
            stopThreads.append(thread)

        for thread in stopThreads:
            thread.join()

class DestroyMirrors():
    ''' Destroy the WAL replication mirror segment '''

    def __init__(self, cluster_config):
        self.clusterconfig = cluster_config
        self.segconfigs = cluster_config.get_seg_configs()

    def destroyThread(self, segconfig):
        commands = []
        mirror_contentid = segconfig.content
        mirror_dir = segconfig.datadir

        commands.append("pg_ctl -D %s stop" % mirror_dir)
        commands.append("rm -rf %s" % mirror_dir)
        thread_name = 'Mirror content %d' % mirror_contentid
        command_finish = 'Destroyed mirror at %s' % mirror_dir
        runcommands(commands, thread_name, command_finish, False)

        # Let FTS recognize that mirrors are gone.  As a result,
        # primaries will be marked not-in-sync.  If this step is
        # omitted, FTS will stop probing as soon as mirrors are
        # removed from catalog and primaries will be left "in-sync"
        # without mirrors.
        #
        # FIXME: enhance gp_remove_segment_mirror() to do this, so
        # that utility remains simplified.  Remove this stopgap
        # thereafter.
        ForceFTSProbeScan(self.clusterconfig,
                          GpSegmentConfiguration.STATUS_DOWN,
                          GpSegmentConfiguration.NOT_IN_SYNC)

        commands = []
        catalog_update_query = "select pg_catalog.gp_remove_segment_mirror(%d::int2)" % (mirror_contentid)
        commands.append("PGOPTIONS=\"-c gp_session_role=utility\" psql postgres -c \"%s\"" % catalog_update_query)

        command_finish = 'Removed mirror %s from catalog' % mirror_dir
        runcommands(commands, thread_name, command_finish, False)

    def run(self):
        destroyThreads = []
        for segconfig in self.segconfigs:
            assert(segconfig.role == GpSegmentConfiguration.ROLE_MIRROR)
            thread = threading.Thread(target=self.destroyThread, args=(segconfig,))
            thread.start()
            destroyThreads.append(thread)

        for thread in destroyThreads:
            thread.join()

class GpSegmentConfiguration():
    ROLE_PRIMARY = 'p'
    ROLE_MIRROR = 'm'
    STATUS_DOWN = 'd'
    STATUS_UP = 'u'
{'fraction_non_alphanumeric': 0.10357572249187336, 'fraction_numerical': 0.001514004542013626, 'mean_word_length': 2.588686481303931, 'pattern_counts': {'":': 0, '<': 3, '<?xml version=': 0, '>': 7, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '1524322', 'n_tokens_mistral': 6541, 'n_tokens_neox': 6384, 'n_words': 1677}