#!/usr/bin/env python
# Licensed to the Apache Software Foundation (ASF) under one
# or more contributor license agreements.  See the NOTICE file
# distributed with this work for additional information
# regarding copyright ownership.  The ASF licenses this file
# to you under the Apache License, Version 2.0 (the
# "License"); you may not use this file except in compliance
# with the License.  You may obtain a copy of the License at
# 
#   http://www.apache.org/licenses/LICENSE-2.0
# 
# Unless required by applicable law or agreed to in writing,
# software distributed under the License is distributed on an
# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
# KIND, either express or implied.  See the License for the
# specific language governing permissions and limitations
# under the License.

"""
This file defines the interface that can be used to fetch and update system
configuration information.
"""
import os, copy

from gppylib.gplog import *
from gppylib.utils import checkNotNone
from gppylib.system.configurationInterface import *
from gppylib.system.ComputeCatalogUpdate import ComputeCatalogUpdate
from gppylib.gparray import GpArray, GpDB, InvalidSegmentConfiguration
from gppylib import gparray
from gppylib.db import dbconn
from gppylib.commands.gp import get_local_db_mode

logger = get_default_logger()

class GpConfigurationProviderUsingGpdbCatalog(GpConfigurationProvider) :
    """
    An implementation of GpConfigurationProvider will provide functionality to
    fetch and update gpdb system configuration information (as stored in the
    database)

    Note that the client of this is assuming that the database data is not
    changed by another party between the time segment data is loaded and when it
    is updated
    """

    def __init__(self):
        self.__masterDbUrl = None


    def initializeProvider( self, masterPort ) :
        """
        Initialize the provider to get information from the given master db, if
        it chooses to get its data from the database
     
        returns self
        """

        checkNotNone("masterPort", masterPort)

        dbUrl = dbconn.DbURL(port=masterPort, dbname='template1')

        self.__masterDbUrl = dbUrl
        return self


    def loadSystemConfig( self, useUtilityMode ) :
        """
        Load all segment information from the configuration source.

        Returns a new GpArray object
        """

        # ensure initializeProvider() was called
        checkNotNone("masterDbUrl", self.__masterDbUrl) 

        logger.info("Obtaining Segment details from master...")

        array = GpArray.initFromCatalog(self.__masterDbUrl, useUtilityMode)
        
        if get_local_db_mode(array.master.getSegmentDataDirectory()) != 'UTILITY':
            logger.debug("Validating configuration...")
            if not array.is_array_valid():
                raise InvalidSegmentConfiguration(array)
            
        return array


    def sendPgElogFromMaster( self, msg, sendAlerts):
        """
        Send a message from the master database using select pg_elog ...
        """
        # ensure initializeProvider() was called
        checkNotNone("masterDbUrl", self.__masterDbUrl)

        conn = None
        try:
            conn = dbconn.connect(self.__masterDbUrl, utility=True)
            dbconn.execSQL(conn, "SELECT GP_ELOG(" +
                        self.__toSqlCharValue(msg) + "," +
                        ("true" if sendAlerts else "false") + ")")
        finally:
            if conn:
                conn.close()


    def updateSystemConfig( self, gpArray, textForConfigTable, dbIdToForceMirrorRemoveAdd, useUtilityMode, allowPrimary) :
        """
        Update the configuration for the given segments in the underlying
        configuration store to match the current values
    
        Also resets any dirty bits on saved/updated objects

        @param textForConfigTable label to be used when adding to segment configuration history
        @param dbIdToForceMirrorRemoveAdd a map of dbid -> True for mirrors for which we should force updating the mirror
        @param useUtilityMode True if the operations we're doing are expected to run via utility moed
        @param allowPrimary True if caller authorizes add/remove primary operations (e.g. gpexpand)
        """

        # ensure initializeProvider() was called
        checkNotNone("masterDbUrl", self.__masterDbUrl) 

        logger.debug("Validating configuration changes...")

        if not gpArray.is_array_valid():
            logger.critical("Configuration is invalid")
            raise InvalidSegmentConfiguration(gpArray)

        conn = dbconn.connect(self.__masterDbUrl, useUtilityMode, allowSystemTableMods='dml')
        dbconn.execSQL(conn, "BEGIN")

        # compute what needs to be updated
        update = ComputeCatalogUpdate(gpArray, dbIdToForceMirrorRemoveAdd, useUtilityMode, allowPrimary)
        update.validate()

        # put the mirrors in a map by content id so we can update them later
        mirror_map = {}
        for seg in update.mirror_to_add:
            mirror_map[ seg.getSegmentContentId() ] = seg

        # reset dbId of new primary and mirror segments to -1
        # before invoking the operations which will assign them new ids
        for seg in update.primary_to_add:
            seg.setSegmentDbId(-1)
        for seg in update.mirror_to_add:
            seg.setSegmentDbId(-1)

        # remove mirror segments (e.g. for gpexpand rollback)
        for seg in update.mirror_to_remove:
            self.__updateSystemConfigRemoveMirror(conn, seg, textForConfigTable)

        # remove primary segments (e.g for gpexpand rollback)
        for seg in update.primary_to_remove:
            self.__updateSystemConfigRemovePrimary(conn, seg, textForConfigTable)

        # add new primary segments
        for seg in update.primary_to_add:
            self.__updateSystemConfigAddPrimary(conn, gpArray, seg, textForConfigTable, mirror_map)

        # add new mirror segments
        for seg in update.mirror_to_add:
            self.__updateSystemConfigAddMirror(conn, gpArray, seg, textForConfigTable)

        # remove and add mirror segments necessitated by catalog attribute update
        for seg in update.mirror_to_remove_and_add:
            self.__updateSystemConfigRemoveAddMirror(conn, gpArray, seg, textForConfigTable)

        # apply updates to existing segments
        for seg in update.segment_to_update:
            originalSeg = update.dbsegmap.get(seg.getSegmentDbId())
            self.__updateSystemConfigUpdateSegment(conn, gpArray, seg, originalSeg, textForConfigTable)

        # apply update to fault strategy
        if gpArray.getStrategyAsLoadedFromDb() != gpArray.getFaultStrategy():
            self.__updateSystemConfigFaultStrategy(conn, gpArray)

        # commit changes
        logger.debug("Committing configuration table changes")
        dbconn.execSQL(conn, "COMMIT")
        conn.close()

        gpArray.setStrategyAsLoadedFromDb( [gpArray.getFaultStrategy()])
        gpArray.setSegmentsAsLoadedFromDb([seg.copy() for seg in gpArray.getDbList()])


    def __updateSystemConfigRemoveMirror(self, conn, seg, textForConfigTable):
        """
        Remove a mirror segment currently in gp_segment_configuration 
        but not present in the goal configuration and record our action 
        in gp_configuration_history.
        """
        dbId   = seg.getSegmentDbId()
        self.__callSegmentRemoveMirror(conn, seg)
        self.__insertConfigHistory(conn, dbId, "%s: removed mirror segment configuration" % textForConfigTable)


    def __updateSystemConfigRemovePrimary(self, conn, seg, textForConfigTable):
        """
        Remove a primary segment currently in gp_segment_configuration 
        but not present in the goal configuration and record our action 
        in gp_configuration_history.
        """
        dbId = seg.getSegmentDbId()
        self.__callSegmentRemove(conn, seg)
        self.__insertConfigHistory(conn, dbId, "%s: removed primary segment configuration" % textForConfigTable)


    def __updateSystemConfigAddPrimary(self, conn, gpArray, seg, textForConfigTable, mirror_map):
        """
        Add a primary segment specified in our goal configuration but
        which is missing from the current gp_segment_configuration table
        and record our action in gp_configuration_history.
        """
        # lookup the mirror (if any) so that we may correct its content id
        mirrorseg = mirror_map.get( seg.getSegmentContentId() )

        # add the new segment
        dbId = self.__callSegmentAdd(conn, gpArray, seg)

        # update the segment mode, status and replication port
        self.__updateSegmentModeStatus(conn, seg)
        if gpArray.get_mirroring_enabled() == True:
            self.__updateSegmentReplicationPort(conn, seg)

        # get the newly added segment's content id
        # MPP-12393 et al WARNING: there is an unusual side effect going on here.  
        # Although gp_add_segment() executed by __callSegmentAdd() above returns 
        # the dbId of the new row in gp_segment_configuration, the following
        # select from gp_segment_configuration can return 0 rows if the updates
        # done by __updateSegmentModeStatus() and/or __updateSegmentReplicationPort()
        # are not done first.  Don't change the order of these operations unless you 
        # understand why gp_add_segment() behaves as it does.
        sql = "select content from pg_catalog.gp_segment_configuration where dbId = %s" % self.__toSqlIntValue(seg.getSegmentDbId())
        logger.debug(sql)
        sqlResult = self.__fetchSingleOutputRow(conn, sql)
        contentId = int(sqlResult[0])

        # Set the new content id for the primary as well the mirror if present.
        seg.setSegmentContentId(contentId)
        if mirrorseg is not None:
            mirrorseg.setSegmentContentId(contentId)

        self.__insertConfigHistory(conn, dbId, "%s: inserted primary segment configuration with contentid %s" % (textForConfigTable, contentId))


    def __updateSystemConfigAddMirror(self, conn, gpArray, seg, textForConfigTable):
        """
        Add a mirror segment specified in our goal configuration but
        which is missing from the current gp_segment_configuration table
        and record our action in gp_configuration_history.
        """
        dbId = self.__callSegmentAddMirror(conn, gpArray, seg)
        self.__updateSegmentModeStatus(conn, seg)
        self.__insertConfigHistory(conn, dbId, "%s: inserted mirror segment configuration" % textForConfigTable)


    def __updateSystemConfigRemoveAddMirror(self, conn, gpArray, seg, textForConfigTable):
        """
        We've been asked to update the mirror in a manner that require
        it to be removed and then re-added.   Perform the tasks
        and record our action in gp_configuration_history.
        """
        origDbId = seg.getSegmentDbId()
        self.__callSegmentRemoveMirror(conn, seg)

        dbId = self.__callSegmentAddMirror(conn, gpArray, seg)

        # now update mode/status since this is not done by gp_add_segment_mirror
        self.__updateSegmentModeStatus(conn, seg)
        self.__insertConfigHistory(conn, seg.getSegmentDbId(), 
                                   "%s: inserted segment configuration for full recovery or original dbid %s" \
                                   % (textForConfigTable, origDbId))


    def __updateSystemConfigUpdateSegment(self, conn, gpArray, seg, originalSeg, textForConfigTable):

        # update mode and status
        # when adding a mirror, the replication port may change as well
        #
        if gpArray.getFaultStrategy() == gparray.FAULT_STRATEGY_NONE:
            what = "%s: segment hostname and address"
            self.__updateSegmentAddress(conn, seg)
        else:
            what = "%s: segment mode and status"
            self.__updateSegmentModeStatus(conn, seg)

        if seg.getSegmentReplicationPort() != originalSeg.getSegmentReplicationPort():
            what = "%s: segment mode, status, and replication port"
            self.__updateSegmentReplicationPort(conn, seg)

        self.__insertConfigHistory(conn, seg.getSegmentDbId(), what % textForConfigTable)


    def __updateSystemConfigFaultStrategy(self, conn, gpArray):
        """
        Update the fault strategy.
        """
        fs  = gpArray.getFaultStrategy()
        sql = "UPDATE gp_fault_strategy\n SET fault_strategy = " + self.__toSqlCharValue(fs) + "\n"
        logger.debug(sql)
        dbconn.executeUpdateOrInsert(conn, sql, 1)


    def __callSegmentRemoveMirror(self, conn, seg):
        """
        Call gp_remove_segment_mirror() to remove the mirror.
        """
        sql = "SELECT gp_remove_segment_mirror(%s::int2)" % (self.__toSqlIntValue(seg.getSegmentContentId()))
        logger.debug(sql)
        result = self.__fetchSingleOutputRow(conn, sql)
        assert result[0] # must return True


    def __callSegmentRemove(self, conn, seg):
        """
        Call gp_remove_segment() to remove the primary.
        """
        sql = "SELECT gp_remove_segment(%s::int2)" % (self.__toSqlIntValue(seg.getSegmentDbId()))
        logger.debug(sql)
        result = self.__fetchSingleOutputRow(conn, sql)
        assert result[0]


    def __callSegmentAdd(self, conn, gpArray, seg):
        """
        Call gp_add_segment() to add the primary.
        Return the new segment's dbid.
        """
        logger.debug('callSegmentAdd %s' % repr(seg))
        filespaceMapStr = self.__toSqlFilespaceMapStr(gpArray, seg)

        sql = "SELECT gp_add_segment(%s, %s, %s, %s)" \
            % (
                self.__toSqlTextValue(seg.getSegmentHostName()), 
                self.__toSqlTextValue(seg.getSegmentAddress()), 
                self.__toSqlIntValue(seg.getSegmentPort()), 
                self.__toSqlTextValue(filespaceMapStr)
              )
        logger.debug(sql)
        sqlResult = self.__fetchSingleOutputRow(conn, sql)
        dbId = int(sqlResult[0])
        seg.setSegmentDbId(dbId)
        return dbId


    def __callSegmentAddMirror(self, conn, gpArray, seg):
        """
        Call gp_add_segment_mirror() to add the mirror.
        Return the new segment's dbid.
        """
        logger.debug('callSegmentAddMirror %s' % repr(seg))
        filespaceMapStr = self.__toSqlFilespaceMapStr(gpArray, seg)

        sql = "SELECT gp_add_segment_mirror(%s::int2, %s, %s, %s, %s, %s)" \
            % (
                self.__toSqlIntValue(seg.getSegmentContentId()),
                self.__toSqlTextValue(seg.getSegmentHostName()),
                self.__toSqlTextValue(seg.getSegmentAddress()),
                self.__toSqlIntValue(seg.getSegmentPort()),
                self.__toSqlIntValue(seg.getSegmentReplicationPort()),
                self.__toSqlTextValue(filespaceMapStr)
              )

        logger.debug(sql)
        sqlResult = self.__fetchSingleOutputRow(conn, sql)
        dbId = int(sqlResult[0])
        seg.setSegmentDbId(dbId)
        return dbId


    def __updateSegmentReplicationPort(self, conn, seg):
        # run an update
        sql = "UPDATE pg_catalog.gp_segment_configuration\n" + \
            "  SET\n" + \
            "  replication_port = " + self.__toSqlIntValue(seg.getSegmentReplicationPort()) + "\n" \
            "WHERE dbid = " + self.__toSqlIntValue(seg.getSegmentDbId())
        logger.debug(sql)
        dbconn.executeUpdateOrInsert(conn, sql, 1)


    def __updateSegmentModeStatus(self, conn, seg):
        # run an update
        sql = "UPDATE pg_catalog.gp_segment_configuration\n" + \
            "  SET\n" + \
            "  mode = " + self.__toSqlCharValue(seg.getSegmentMode()) + ",\n" \
            "  status = " + self.__toSqlCharValue(seg.getSegmentStatus()) + "\n" \
            "WHERE dbid = " + self.__toSqlIntValue(seg.getSegmentDbId())
        logger.debug(sql)
        dbconn.executeUpdateOrInsert(conn, sql, 1)


    def __updateSegmentAddress(self, conn, seg):
        # run an update
        sql = "UPDATE pg_catalog.gp_segment_configuration\n" + \
            "  SET\n" + \
            "  hostname = " + self.__toSqlCharValue(seg.getSegmentHostName()) + ",\n" \
            "  address = " + self.__toSqlCharValue(seg.getSegmentAddress()) + "\n" \
            "WHERE dbid = " + self.__toSqlIntValue(seg.getSegmentDbId())
        logger.debug(sql)
        dbconn.executeUpdateOrInsert(conn, sql, 1)


    def __fetchSingleOutputRow(self, conn, sql, retry=False):
        """
        Execute specified SQL command and return what we expect to be a single row.
        Raise an exception when more or fewer than one row is seen and when more
        than one row is seen display up to 10 rows as logger warnings.
        """
        cursor   = dbconn.execSQL(conn, sql)
        numrows  = cursor.rowcount
        numshown = 0
        res      = None
        for row in cursor:
            if numrows != 1:
                #
                # if we got back more than one row 
                # we print a few of the rows first
                # instead of immediately raising an exception
                #
                numshown += 1
                if numshown > 10:
                    break
                logger.warning('>>> %s' % row)
            else:
                assert res is None
                res = row
                assert res is not None
        cursor.close()
        if numrows != 1:
            raise Exception("SQL returned %d rows, not 1 as expected:\n%s" % (numrows, sql))
        return res


    def __insertConfigHistory(self, conn, dbId, msg ):
        # now update change history
        sql = "INSERT INTO gp_configuration_history (time, dbid, \"desc\") VALUES(\n" \
                    "now(),\n  " + \
                    self.__toSqlIntValue(dbId) + ",\n  " + \
                    self.__toSqlCharValue(msg) + "\n)"
        logger.debug(sql)
        dbconn.executeUpdateOrInsert(conn, sql, 1)


    def __toSqlFilespaceMapStr(self, gpArray, seg):
        """
        Return a string representation of the filespace map suitable
        for inclusion into the call to gp_add_segment_mirror().
        """
        filespaceArrayString = []
        for fs in gpArray.getAllFilespaces():
            path = seg.getSegmentFilespaces()[ fs.getOid() ]
            filespaceArrayString.append("{%s,%s}" % \
                        (self.__toSqlArrayStringValue(fs.getName()), \
                         self.__toSqlArrayStringValue(path)))

        filespaceMapStr = "{" + ",".join(filespaceArrayString) + "}"
        return filespaceMapStr

    def __toSqlIntValue(self, val):
        if val is None:
            return "null"
        return str(val)

    def __toSqlArrayStringValue(self, val):
        if val is None:
            return "null"
        return '"' + val.replace('"','\\"').replace('\\','\\\\') + '"'

    def __toSqlCharValue(self, val):
        return self.__toSqlTextValue(val)

    def __toSqlTextValue(self, val):
        if val is None:
            return "null"
        return "'" + val.replace("'","''").replace('\\','\\\\') + "'"
