Itops 300/feature/update repman (#13)

Co-authored-by: apple <[email protected]>
Co-authored-by: emma <[email protected]>
Co-authored-by: emmaloubersac <[email protected]>

Author: 3 people

arbitrator/arbitrator_cmd.go

@@ -56,6 +56,7 @@ func init() {
 	rootCmd.Flags().StringVar(&conf.KeyPath, "keypath", "/etc/replication-manager/.replication-manager.key", "Encryption key file path")
 	rootCmd.PersistentFlags().BoolVar(&conf.Verbose, "verbose", false, "Print detailed execution info")
 	rootCmd.PersistentFlags().StringVar(&memprofile, "memprofile", "/tmp/repmgr.mprof", "Write a memory profile to a file readable by pprof")
+	rootCmd.PersistentFlags().StringVar(&conf.WorkingDir, "monitoring-datadir", "/var/lib/replication-manager", "Path to write temporary and persistent files")
 
 }
 

cluster/cluster.go

@@ -30,8 +30,7 @@ import (
 	"github.com/signal18/replication-manager/utils/s18log"
 	"github.com/signal18/replication-manager/utils/state"
 	log "github.com/sirupsen/logrus"
-	logsqlerr "github.com/sirupsen/logrus"
-	logsqlgen "github.com/sirupsen/logrus"
+	logsql "github.com/sirupsen/logrus"
 	"golang.org/x/crypto/ssh"
 )
 
@@ -171,6 +170,11 @@ type Cluster struct {
 	WaitingFailover               int                         `json:"waitingFailover"`
 	Configurator                  configurator.Configurator   `json:"configurator"`
 	DiffVariables                 []VariableDiff              `json:"diffVariables"`
+	inInitNodes                   bool                        `json:"-"`
+	CanInitNodes                  bool                        `json:"canInitNodes"`
+	errorInitNodes                error                       `json:"-"`
+	SqlErrorLog                   *logsql.Logger              `json:"-"`
+	SqlGeneralLog                 *logsql.Logger              `json:"-"`
 	sync.Mutex
 	crcTable *crc64.Table
 }
@@ -251,7 +255,8 @@ const (
 
 // Init initial cluster definition
 func (cluster *Cluster) Init(conf config.Config, cfgGroup string, tlog *s18log.TermLog, log *s18log.HttpLog, termlength int, runUUID string, repmgrVersion string, repmgrHostname string, key []byte) error {
-
+	cluster.SqlErrorLog = logsql.New()
+	cluster.SqlGeneralLog = logsql.New()
 	cluster.crcTable = crc64.MakeTable(crc64.ECMA) // http://golang.org/pkg/hash/crc64/#pkg-constants
 	cluster.switchoverChan = make(chan bool)
 	// should use buffered channels or it will block
@@ -263,6 +268,7 @@ func (cluster *Cluster) Init(conf config.Config, cfgGroup string, tlog *s18log.T
 	cluster.addtableCond = nbc.New()
 	cluster.altertableCond = nbc.New()
 	cluster.canFlashBack = true
+	cluster.CanInitNodes = true
 	cluster.runOnceAfterTopology = true
 	cluster.testStopCluster = true
 	cluster.testStartCluster = true
@@ -309,34 +315,34 @@ func (cluster *Cluster) Init(conf config.Config, cfgGroup string, tlog *s18log.T
 		MaxSize:    cluster.Conf.LogRotateMaxSize,
 		MaxBackups: cluster.Conf.LogRotateMaxBackup,
 		MaxAge:     cluster.Conf.LogRotateMaxAge,
-		Level:      logsqlerr.DebugLevel,
-		Formatter: &logsqlerr.TextFormatter{
+		Level:      logsql.DebugLevel,
+		Formatter: &logsql.TextFormatter{
 			DisableColors:   true,
 			TimestampFormat: "2006-01-02 15:04:05",
 			FullTimestamp:   true,
 		},
 	})
 	if err != nil {
-		logsqlerr.WithError(err).Error("Can't init error sql log file")
+		cluster.SqlErrorLog.WithError(err).Error("Can't init error sql log file")
 	}
-	logsqlerr.AddHook(hookerr)
+	cluster.SqlErrorLog.AddHook(hookerr)
 
 	hookgen, err := s18log.NewRotateFileHook(s18log.RotateFileConfig{
 		Filename:   cluster.WorkingDir + "/sql_general.log",
 		MaxSize:    cluster.Conf.LogRotateMaxSize,
 		MaxBackups: cluster.Conf.LogRotateMaxBackup,
 		MaxAge:     cluster.Conf.LogRotateMaxAge,
-		Level:      logsqlerr.DebugLevel,
-		Formatter: &logsqlgen.TextFormatter{
+		Level:      logsql.DebugLevel,
+		Formatter: &logsql.TextFormatter{
 			DisableColors:   true,
 			TimestampFormat: "2006-01-02 15:04:05",
 			FullTimestamp:   true,
 		},
 	})
 	if err != nil {
-		logsqlgen.WithError(err).Error("Can't init general sql log file")
+		cluster.SqlGeneralLog.WithError(err).Error("Can't init general sql log file")
 	}
-	logsqlgen.AddHook(hookgen)
+	cluster.SqlGeneralLog.AddHook(hookgen)
 	cluster.LoadAPIUsers()
 	// createKeys do nothing yet
 	cluster.createKeys()
@@ -359,18 +365,25 @@ func (cluster *Cluster) Init(conf config.Config, cfgGroup string, tlog *s18log.T
 
 	return nil
 }
+
 func (cluster *Cluster) initOrchetratorNodes() {
+	if cluster.inInitNodes {
+		return
+	}
+	cluster.inInitNodes = true
+	defer func() { cluster.inInitNodes = false }()
 
+	//defer cluster.insideInitNodes = false
 	//cluster.LogPrintf(LvlInfo, "Loading nodes from orchestrator %s", cluster.Conf.ProvOrchestrator)
 	switch cluster.GetOrchestrator() {
 	case config.ConstOrchestratorOpenSVC:
-		cluster.Agents, _ = cluster.OpenSVCGetNodes()
+		cluster.Agents, cluster.errorInitNodes = cluster.OpenSVCGetNodes()
 	case config.ConstOrchestratorKubernetes:
-		cluster.Agents, _ = cluster.K8SGetNodes()
+		cluster.Agents, cluster.errorInitNodes = cluster.K8SGetNodes()
 	case config.ConstOrchestratorSlapOS:
-		cluster.Agents, _ = cluster.SlapOSGetNodes()
+		cluster.Agents, cluster.errorInitNodes = cluster.SlapOSGetNodes()
 	case config.ConstOrchestratorLocalhost:
-		cluster.Agents, _ = cluster.LocalhostGetNodes()
+		cluster.Agents, cluster.errorInitNodes = cluster.LocalhostGetNodes()
 	case config.ConstOrchestratorOnPremise:
 	default:
 		log.Fatalln("prov-orchestrator not supported", cluster.Conf.ProvOrchestrator)
@@ -424,8 +437,8 @@ func (cluster *Cluster) Run() {
 					for k, v := range cluster.Servers {
 						cluster.LogPrintf(LvlDbg, "Server [%d]: URL: %-15s State: %6s PrevState: %6s", k, v.URL, v.State, v.PrevState)
 					}
-					if cluster.master != nil {
-						cluster.LogPrintf(LvlDbg, "Master [ ]: URL: %-15s State: %6s PrevState: %6s", cluster.master.URL, cluster.master.State, cluster.master.PrevState)
+					if cluster.GetMaster() != nil {
+						cluster.LogPrintf(LvlDbg, "Master [ ]: URL: %-15s State: %6s PrevState: %6s", cluster.master.URL, cluster.GetMaster().State, cluster.GetMaster().PrevState)
 						for k, v := range cluster.slaves {
 							cluster.LogPrintf(LvlDbg, "Slave  [%d]: URL: %-15s State: %6s PrevState: %6s", k, v.URL, v.State, v.PrevState)
 						}
@@ -444,7 +457,7 @@ func (cluster *Cluster) Run() {
 					if !cluster.IsInFailover() {
 						cluster.initProxies()
 					}
-					cluster.initOrchetratorNodes()
+					go cluster.initOrchetratorNodes()
 					cluster.ResticFetchRepo()
 					cluster.runOnceAfterTopology = false
 				} else {
@@ -461,7 +474,7 @@ func (cluster *Cluster) Run() {
 						cluster.InjectProxiesTraffic()
 					}
 					if cluster.sme.GetHeartbeats()%30 == 0 {
-						cluster.initOrchetratorNodes()
+						go cluster.initOrchetratorNodes()
 						cluster.MonitorQueryRules()
 						cluster.MonitorVariablesDiff()
 						cluster.ResticFetchRepo()
@@ -471,9 +484,12 @@ func (cluster *Cluster) Run() {
 						cluster.sme.PreserveState("WARN0093")
 						cluster.sme.PreserveState("WARN0084")
 						cluster.sme.PreserveState("WARN0095")
-						cluster.sme.PreserveState("ERR00082")
 						cluster.sme.PreserveState("WARN0101")
 					}
+					if !cluster.CanInitNodes {
+						cluster.SetState("ERR00082", state.State{ErrType: "WARNING", ErrDesc: fmt.Sprintf(clusterError["ERR00082"], cluster.errorInitNodes), ErrFrom: "OPENSVC"})
+					}
+
 					if cluster.sme.GetHeartbeats()%36000 == 0 {
 						cluster.ResticPurgeRepo()
 					} else {
@@ -561,8 +577,13 @@ func (cluster *Cluster) StateProcessing() {
 			}
 			//		cluster.statecloseChan <- s
 		}
+		var states []string
+		if cluster.runOnceAfterTopology {
+			states = cluster.sme.GetFirstStates()
 
-		states := cluster.sme.GetStates()
+		} else {
+			states = cluster.sme.GetStates()
+		}
 		for i := range states {
 			cluster.LogPrintf("STATE", states[i])
 		}
@@ -573,7 +594,9 @@ func (cluster *Cluster) StateProcessing() {
 		}
 
 		for _, s := range cluster.sme.GetLastOpenedStates() {
+
 			cluster.CheckAlert(s)
+
 		}
 
 		cluster.sme.ClearState()
@@ -715,7 +738,7 @@ func (cluster *Cluster) FailoverForce() error {
 
 		}
 	}
-	if cluster.master == nil {
+	if cluster.GetMaster() == nil {
 		cluster.LogPrintf(LvlErr, "Could not find a failed server in the hosts list")
 		return errors.New("ERROR: Could not find a failed server in the hosts list")
 	}
@@ -861,21 +884,21 @@ func (cluster *Cluster) MonitorSchema() {
 	if !cluster.Conf.MonitorSchemaChange {
 		return
 	}
-	if cluster.master == nil {
+	if cluster.GetMaster() == nil {
 		return
 	}
-	if cluster.master.State == stateFailed || cluster.master.State == stateMaintenance || cluster.master.State == stateUnconn {
+	if cluster.GetMaster().State == stateFailed || cluster.GetMaster().State == stateMaintenance || cluster.GetMaster().State == stateUnconn {
 		return
 	}
-	if cluster.master.Conn == nil {
+	if cluster.GetMaster().Conn == nil {
 		return
 	}
 	cluster.sme.SetMonitorSchemaState()
-	cluster.master.Conn.SetConnMaxLifetime(3595 * time.Second)
+	cluster.GetMaster().Conn.SetConnMaxLifetime(3595 * time.Second)
 
-	tables, tablelist, logs, err := dbhelper.GetTables(cluster.master.Conn, cluster.master.DBVersion)
-	cluster.LogSQL(logs, err, cluster.master.URL, "Monitor", LvlErr, "Could not fetch master tables %s", err)
-	cluster.master.Tables = tablelist
+	tables, tablelist, logs, err := dbhelper.GetTables(cluster.GetMaster().Conn, cluster.GetMaster().DBVersion)
+	cluster.LogSQL(logs, err, cluster.GetMaster().URL, "Monitor", LvlErr, "Could not fetch master tables %s", err)
+	cluster.GetMaster().Tables = tablelist
 
 	var tableCluster []string
 	var duplicates []*ServerMonitor
@@ -889,7 +912,7 @@ func (cluster *Cluster) MonitorSchema() {
 
 		duplicates = append(duplicates, cluster.GetMaster())
 		tableCluster = append(tableCluster, cluster.GetName())
-		oldtable, err := cluster.master.GetTableFromDict(t.TableSchema + "." + t.TableName)
+		oldtable, err := cluster.GetMaster().GetTableFromDict(t.TableSchema + "." + t.TableName)
 		haschanged := false
 		if err != nil {
 			if err.Error() == "Empty" {
@@ -936,7 +959,7 @@ func (cluster *Cluster) MonitorSchema() {
 	}
 	cluster.DBIndexSize = totindexsize
 	cluster.DBTableSize = tottablesize
-	cluster.master.DictTables = tables
+	cluster.GetMaster().DictTables = tables
 	cluster.sme.RemoveMonitorSchemaState()
 }
 
@@ -993,7 +1016,7 @@ func (cluster *Cluster) LostArbitration(realmasterurl string) {
 	}
 	if cluster.Conf.ArbitrationFailedMasterScript != "" {
 		cluster.LogPrintf(LvlInfo, "Calling abitration failed for master script")
-		out, err := exec.Command(cluster.Conf.ArbitrationFailedMasterScript, cluster.master.Host, cluster.master.Port).CombinedOutput()
+		out, err := exec.Command(cluster.Conf.ArbitrationFailedMasterScript, cluster.GetMaster().Host, cluster.GetMaster().Port).CombinedOutput()
 		if err != nil {
 			cluster.LogPrintf(LvlErr, "%s", err)
 		}

cluster/cluster_chk.go

@@ -130,7 +130,7 @@ func (cluster *Cluster) isMasterFailed() bool {
 func (cluster *Cluster) isMaxMasterFailedCountReached() bool {
 	// no illimited failed count
 
-	if cluster.master.FailCount >= cluster.Conf.MaxFail {
+	if cluster.GetMaster().FailCount >= cluster.Conf.MaxFail {
 		cluster.sme.AddState("WARN0023", state.State{ErrType: "WARNING", ErrDesc: fmt.Sprintf(clusterError["WARN0023"]), ErrFrom: "CHECK"})
 		return true
 	} else {
@@ -256,6 +256,9 @@ func (cluster *Cluster) isMaxscaleSupectRunning() bool {
 }
 
 func (cluster *Cluster) isFoundCandidateMaster() bool {
+	if cluster.GetTopology() == topoActivePassive {
+		return true
+	}
 	key := -1
 	if cluster.Conf.MultiMasterGrouprep {
 		key = cluster.electSwitchoverGroupReplicationCandidate(cluster.slaves, true)

cluster/cluster_fail.go

@@ -343,12 +343,12 @@ func (cluster *Cluster) MasterFailover(fail bool) bool {
 		}
 
 		if cluster.Conf.ReadOnly {
-			logs, err = dbhelper.SetReadOnly(cluster.oldMaster.Conn, true)
+			logs, err = cluster.oldMaster.SetReadOnly()
 			cluster.LogSQL(logs, err, cluster.oldMaster.URL, "MasterFailover", LvlErr, "Could not set old master as read-only, %s", err)
-
-		} else {
-			logs, err = dbhelper.SetReadOnly(cluster.oldMaster.Conn, false)
-			cluster.LogSQL(logs, err, cluster.oldMaster.URL, "MasterFailover", LvlErr, "Could not set old master as read-write, %s", err)
+			/*	} else {
+				logs, err = cluster.oldMaster.SetReadWrite()
+				cluster.LogSQL(logs, err, cluster.oldMaster.URL, "MasterFailover", LvlErr, "Could not set old master as read-write, %s", err)
+			*/
 		}
 		if cluster.Conf.SwitchDecreaseMaxConn {
 
@@ -1216,7 +1216,7 @@ func (cluster *Cluster) VMasterFailover(fail bool) bool {
 	cluster.failoverPreScript(fail)
 
 	// Phase 2: Reject updates and sync slaves on switchover
-	if fail == false {
+	if fail == false && cluster.GetTopology() != topoMultiMasterWsrep {
 		cluster.LogPrintf(LvlInfo, "Rejecting updates on %s (old master)", cluster.oldMaster.URL)
 		cluster.oldMaster.freeze()
 	}
@@ -1323,7 +1323,8 @@ func (cluster *Cluster) VMasterFailover(fail bool) bool {
 				cluster.LogPrintf(LvlErr, "Could not set old master as read-write, %s", err)
 			}
 		}
-		if cluster.Conf.SwitchDecreaseMaxConn {
+		// Galara does not freeze old master because of bug https://jira.mariadb.org/browse/MDEV-9134
+		if cluster.Conf.SwitchDecreaseMaxConn && cluster.GetTopology() != topoMultiMasterWsrep {
 			logs, err := dbhelper.SetMaxConnections(cluster.oldMaster.Conn, cluster.oldMaster.maxConn, cluster.oldMaster.DBVersion)
 			cluster.LogSQL(logs, err, cluster.oldMaster.URL, "MasterFailover", LvlErr, "Could not set max connections on %s %s", cluster.oldMaster.URL, err)
 		}

cluster/cluster_get.go

@@ -296,6 +296,14 @@ func (cluster *Cluster) GetGComm() string {
 		} else {
 			gcomms = append(gcomms, server.Host+":"+strconv.Itoa(cluster.Conf.MultiMasterGrouprepPort))
 		}
+
+	}
+	//	For bootrap galera cluster on first node
+	if cluster.AllServersFailed() && cluster.GetTopology() == topoMultiMasterWsrep {
+		return ""
+	}
+	if cluster.GetTopology() == topoMultiMasterWsrep {
+		return strings.Join(gcomms, ",") + "?pc.wait_prim=yes"
 	}
 	return strings.Join(gcomms, ",")
 }
@@ -308,7 +316,7 @@ func (cluster *Cluster) getOnePreferedMaster() *ServerMonitor {
 		if cluster.Conf.LogLevel > 2 {
 			cluster.LogPrintf(LvlDbg, "Lookup if server: %s is preferred master: %s", server.URL, cluster.Conf.PrefMaster)
 		}
-		if strings.Contains(cluster.Conf.PrefMaster, server.URL) {
+		if server.IsPrefered() {
 			return server
 		}
 	}
@@ -514,6 +522,8 @@ func (cluster *Cluster) GetTopology() string {
 	} else if cluster.Conf.MasterSlavePgLogical {
 		cluster.Conf.Topology = topoMasterSlavePgLog
 		cluster.IsPostgres = true
+	} else if cluster.Conf.ActivePassive {
+		cluster.Conf.Topology = topoActivePassive
 	} else {
 		relay := cluster.GetRelayServer()
 		if relay != nil && cluster.Conf.ReplicationNoRelay == false {

cluster/cluster_has.go

@@ -166,6 +166,29 @@ func (cluster *Cluster) HasAllDbUp() bool {
 	return true
 }
 
+func (cluster *Cluster) HasNoDbUnconnected() bool {
+	if cluster.Servers == nil {
+		return false
+	}
+	for _, s := range cluster.Servers {
+		if s != nil {
+			if s.State == stateFailed || s.State == stateUnconn /*&& misc.Contains(cluster.ignoreList, s.URL) == false*/ {
+				return false
+			}
+			if s.State == stateSuspect && cluster.GetTopology() != topoUnknown {
+				//supect is used to reload config and avoid backend state change to failed that would disable servers in proxies and cause glinch in cluster traffic
+				// at the same time to enbale bootstrap replication we need to know when server are up
+				return false
+			}
+			if s.Conn == nil {
+				return false
+			}
+		}
+	}
+
+	return true
+}
+
 func (cluster *Cluster) HasRequestDBRestart() bool {
 	if cluster.Servers == nil {
 		return false