Jobs script check and upgrade (#1248)

* feat: Add dbjobs_launcher script for cleanup and job execution - Modify `dbjobs_launcher.sh` to clean up old `.run` directories before continuously execute `dbjobs_new.sh`. - Enhanced `dbjobs_new.sh` with improved encryption functions, allowing for custom passwords and better API request handling. - Added generic functions for sending encrypted data to APIs with retry logic. - Implemented job check and upgrade functions to manage job execution and script updates. - Updated logging mechanisms for better traceability and error handling. fix: Trim whitespace in checksum generation - Modified `GenerateChecksum` function to ignore comments more robustly by trimming whitespace before checking for comment lines. * feat: Add API swagger for jobs upgrade and job checks on servers * refactor: Remove verbose logging for server needs check * Refactor jobsCheck and jobsUpgrade functions for improved process handling - Ensured environment variables LOG_DIR, CHECKPOINT_DIR, and LOCK_DIR are exported for subshells. - Reintroduced jobsCheck and jobsUpgrade calls in the main execution flow for better clarity and structure. * Enhance database table validation and version comparison logic - Updated the `table_exists` function to perform a more comprehensive check on the 'jobs' table structure in the 'replication_manager_schema' database, ensuring all required columns are present with correct types and constraints. - Introduced a new `create_jobs_table` function that interacts with an API to create the jobs table if it does not exist, improving the robustness of the table creation process. - Added a `check_jobs_table` function to streamline the process of checking for the jobs table and creating it if necessary. - Modified the version parsing logic in `version.go` to handle case-insensitive flavor detection for MariaDB, PostgreSQL, and Percona. - Implemented a new `Compare` method for version comparison, allowing for more precise version checks between different database flavors. - Added a `HasVersionChanged` function to determine if the database version has changed, considering flavor differences and providing error handling for nil versions. * Update API endpoints for refreshing app templates and checking cluster/server error states * Implement rolling jobs upgrade functionality and related API endpoints * Add support for rolling jobs upgrade in API and Redux slice * Implement rolling jobs upgrade functionality and update related state management * Refactor RollingJobsUpgrade to return error on timeout and improve logging * Add timeout error handling for jobs upgrade in RollingJobsUpgrade

Author: caffeinated92

cluster/cluster.go

@@ -583,7 +583,7 @@ func (cluster *Cluster) InitFromConf() {
 //cluster.Conf.PrintConf()
 cluster.initScheduler()
 cluster.CheckDefaultUser(true)
-cluster.SetToolVersions()
+cluster.RefreshToolVersions()
 cluster.StartResticRepo()
 
 cluster.Conf.TopologyTarget = cluster.GetTopologyFromConf()
@@ -642,6 +642,22 @@ func (cluster *Cluster) initScheduler() {
 
 }
 
+var pstates30 = []string{
+"WARN0084", // Variables diff
+"ERR00090", "WARN0102", // Config related
+"WARN0093", "WARN0095", "WARN0134", "WARN0145", // Restic related
+"WARN0101", "WARN0111", "WARN0112", // Backup related
+"WARN0139", "WARN0140", "WARN0141", "WARN0142", "WARN0143", "WARN0150", "WARN0151", // Tresholds
+"WARN0147", "WARN0148", "WARN0153", "WARN0154", // Jobs related
+"CREDIT01", // Credit related
+}
+
+var pstates3600 = []string{
+"WARN0094", // Restic
+"WARN0132", "WARN0137", // App templates
+"WARN0117", "WARN0118", "WARN0119", "WARN0120", "WARN0121", // Tools versions
+}
+
 func (cluster *Cluster) Run() {
 defer cluster.LogPanicToFile("cluster")
 interval := time.Second
@@ -739,23 +755,25 @@ func (cluster *Cluster) Run() {
 cluster.CheckAllBackupFreeSpace()
 cluster.CheckAvailableCredit()
 cluster.CheckOpenSVCTresholds()
+cluster.CheckJobsVersion()
+cluster.JobsCheckSchedulerTable()
 } else {
-cluster.StateMachine.PreserveState("WARN0093", "WARN0084", "WARN0095", "WARN0101", "WARN0111", "WARN0112", "ERR00090", "WARN0102", "WARN0134", "WARN0139", "WARN0140", "WARN0141", "WARN0142", "WARN0143", "WARN0145", "WARN0150", "WARN0151", "CREDIT01")
+cluster.StateMachine.PreserveState(pstates30...)
 }
 if !cluster.CanInitNodes {
 cluster.SetState("ERR00082", state.State{ErrType: "WARNING", ErrDesc: fmt.Sprintf(clusterError["ERR00082"], cluster.errorInitNodes), ErrFrom: "OPENSVC"})
 }
 if !cluster.CanConnectVault {
 cluster.SetState("ERR00089", state.State{ErrType: "WARNING", ErrDesc: fmt.Sprintf(clusterError["ERR00089"], cluster.errorConnectVault), ErrFrom: "OPENSVC"})
 }
-if cluster.StateMachine.GetHeartbeats()%36000 == 0 {
+if cluster.StateMachine.GetHeartbeats()%3600 == 0 {
 // Set in parallel since it will wait for fetch to finish
 go cluster.ResticPurgeRepo()
 go cluster.RefreshAllAppTemplateMD5()
+cluster.RefreshToolVersions()
 } else {
 // Preserve tools if not installed or has problem
-cluster.StateMachine.PreserveState("WARN0094", "WARN0117", "WARN0118", "WARN0119", "WARN0120", "WARN0121")
-cluster.StateMachine.PreserveState("WARN0132", "WARN0137")
+cluster.StateMachine.PreserveState(pstates3600...)
 }
 if cluster.SlavesOldestMasterFile.Suffix == 0 {
 go cluster.CheckSlavesReplicationsPurge()
@@ -910,6 +928,13 @@ func (cluster *Cluster) StateProcessing() {
 }
 }
 }
+if s.ErrKey == "WARN0148" && servertoreseed != nil {
+go servertoreseed.UpgradeJobsScript()
+}
+
+if s.ErrKey == "WARN0155" {
+go cluster.RollingJobsUpgrade()
+}
 
 //	cluster.statecloseChan <- s
 cluster.CheckAlert(s, true)

cluster/cluster_acl.go

@@ -413,7 +413,7 @@ func (cluster *Cluster) IsURLPassDatabasesACL(strUser string, URL string) bool {
 return true
 }
 }
-if cluster.APIUsers[strUser].Grants[config.GrantProxyConfigFlag] {
+if cluster.APIUsers[strUser].Grants[config.GrantDBConfigFlag] {
 if strings.Contains(URL, "/config") {
 return true
 }
@@ -507,6 +507,9 @@ func (cluster *Cluster) IsURLPassDatabasesACL(strUser string, URL string) bool {
 if strings.Contains(URL, "/actions/wait-innodb-purge") {
 return true
 }
+if strings.Contains(URL, "/actions/jobs-upgrade") {
+return true
+}
 }
 /*	if cluster.APIUsers[strUser].Grants[config.GrantDBConfigCreate] {
 if strings.Contains(URL, "/kill") {
@@ -796,6 +799,9 @@ func (cluster *Cluster) IsURLPassACL(strUser string, URL string, errorPrint bool
 if strings.Contains(URL, "/api/clusters/"+cluster.Name+"/actions/cancel-rolling-reprov") {
 return true
 }
+if strings.Contains(URL, "/api/clusters/"+cluster.Name+"/actions/jobs-upgrade") {
+return true
+}
 }
 if cluster.APIUsers[strUser].Grants[config.GrantClusterRotatePasswords] {
 if strings.Contains(URL, "/api/clusters/"+cluster.Name+"/actions/rotate-passwords") {

cluster/cluster_bck.go

@@ -44,7 +44,7 @@ func (cluster *Cluster) ResticGetEnv() []string {
 
 func (cluster *Cluster) CheckResticInstallation() {
 if cluster.Conf.BackupRestic && cluster.VersionsMap.Get("restic") == nil {
-if err := cluster.SetResticVersion(); err != nil {
+if err := cluster.RefreshResticVersion(); err != nil {
 cluster.SetState("WARN0121", state.State{ErrType: "WARNING", ErrDesc: fmt.Sprintf(clusterError["WARN0121"], err), ErrFrom: "CLUSTER"})
 } else {
 cluster.LogModulePrintf(cluster.Conf.Verbose, config.ConstLogModGeneral, config.LvlInfo, "Restic version: %s", cluster.VersionsMap.Get("restic").ToString())

cluster/cluster_chk.go

@@ -1044,3 +1044,20 @@ func (cluster *Cluster) CheckDBCredentials() {
 cluster.SetState("ERR00101", state.State{ErrType: "ERROR", ErrDesc: config.ClusterError["ERR00101"], ErrFrom: "CLUSTER"})
 }
 }
+
+func (cluster *Cluster) CheckJobsVersion() {
+for _, server := range cluster.Servers {
+server.CheckJobsVersion()
+}
+}
+
+func (cluster *Cluster) JobsCheckSchedulerTable() {
+for _, server := range cluster.Servers {
+ok, err := server.JobsCheckSchedulerTable()
+if err != nil {
+cluster.SetState("WARN0153", state.State{ErrType: "WARNING", ErrDesc: fmt.Sprintf(clusterError["WARN0153"], server.URL), ErrFrom: "CLUSTER"})
+} else if !ok {
+cluster.SetState("WARN0154", state.State{ErrType: "WARNING", ErrDesc: fmt.Sprintf(clusterError["WARN0154"], server.URL, ""), ErrFrom: "CLUSTER"})
+}
+}
+}

cluster/cluster_get.go

@@ -32,6 +32,7 @@ import (
 "github.com/signal18/replication-manager/utils/misc"
 "github.com/signal18/replication-manager/utils/state"
 "github.com/signal18/replication-manager/utils/tty"
+"github.com/signal18/replication-manager/utils/version"
 )
 
 func (cluster *Cluster) GetCrcTable() *crc64.Table {
@@ -1677,3 +1678,8 @@ func (cluster *Cluster) GetStagingServerFromConfig() *ServerMonitor {
 
 return nil
 }
+
+func (cluster *Cluster) GetToolsVersion(name string) (*version.Version, bool) {
+toolVer, ok := cluster.VersionsMap.CheckAndGet(name)
+return toolVer, ok
+}

cluster/cluster_has.go

@@ -606,3 +606,13 @@ func (cluster *Cluster) HasDiscoverTopologyReachTarget() bool {
 func (cluster *Cluster) IsTopologyTargetEqual(target string) bool {
 return cluster.Conf.TopologyTarget == target
 }
+
+func (cluster *Cluster) IsInErrorState(key, serverURL string) bool {
+if key == "" {
+return false
+} else if serverURL == "" {
+return cluster.StateMachine.IsInState(key)
+} else {
+return cluster.StateMachine.IsInState(fmt.Sprintf("%s@%s", key, serverURL))
+}
+}

cluster/cluster_job.go

@@ -20,6 +20,7 @@ import (
 
 "github.com/signal18/replication-manager/config"
 "github.com/signal18/replication-manager/utils/dbhelper"
+"github.com/signal18/replication-manager/utils/state"
 )
 
 func (cluster *Cluster) JobAnalyzeSQL(persistent bool) error {
@@ -271,3 +272,7 @@ func (cluster *Cluster) JobParseMyDumperMetaOld(dir string) (config.MyDumperMeta
 
 return m, nil
 }
+
+func (cluster *Cluster) SetJobsUpgradeSender(server *ServerMonitor) {
+cluster.SetState("WARN0148", state.State{ErrDesc: fmt.Sprintf(config.ClusterError["WARN0148"], server.URL), ErrType: config.LvlWarn, ErrFrom: "JOBS", ServerUrl: server.URL})
+}

cluster/cluster_roll.go

@@ -185,3 +185,46 @@ func (cluster *Cluster) RollingOptimize() {
 cluster.LogModulePrintf(cluster.Conf.Verbose, config.ConstLogModGeneral, config.LvlInfo, "Optimize job id %d on %s ", jobid, s.URL)
 }
 }
+
+func (cluster *Cluster) RollingJobsUpgrade() error {
+cluster.LogModulePrintf(cluster.Conf.Verbose, config.ConstLogModGeneral, config.LvlInfo, "Rolling jobs upgrade")
+var ts time.Time
+
+for _, s := range cluster.slaves {
+ts = time.Now()
+s.SetWaitJobsUpgradeCookie()
+cluster.LogModulePrintf(cluster.Conf.Verbose, config.ConstLogModGeneral, config.LvlInfo, "Set jobs upgrade cookie on %s ", s.URL)
+
+// Wait for the server to clear the cookie
+for s.HasRollingJobsUpgradeCookie() {
+if time.Since(ts) > 5*time.Minute {
+cluster.LogModulePrintf(cluster.Conf.Verbose, config.ConstLogModGeneral, config.LvlErr, "Timeout waiting for jobs upgrade on %s ", s.URL)
+return errors.New("Timeout waiting for jobs upgrade")
+}
+
+time.Sleep(2 * time.Second)
+}
+
+cluster.LogModulePrintf(cluster.Conf.Verbose, config.ConstLogModGeneral, config.LvlInfo, "Jobs upgrade completed on %s ", s.URL)
+}
+
+ts = time.Now()
+cluster.master.SetWaitJobsUpgradeCookie()
+cluster.LogModulePrintf(cluster.Conf.Verbose, config.ConstLogModGeneral, config.LvlInfo, "Set jobs upgrade cookie on master %s ", cluster.master.URL)
+
+// Wait for the server to clear the cookie
+for cluster.master.HasRollingJobsUpgradeCookie() {
+if time.Since(ts) > 5*time.Minute {
+cluster.LogModulePrintf(cluster.Conf.Verbose, config.ConstLogModGeneral, config.LvlErr, "Timeout waiting for jobs upgrade on master %s ", cluster.master.URL)
+return errors.New("Timeout waiting for jobs upgrade on master")
+}
+
+time.Sleep(2 * time.Second)
+}
+
+cluster.LogModulePrintf(cluster.Conf.Verbose, config.ConstLogModGeneral, config.LvlInfo, "Jobs upgrade completed on master %s ", cluster.master.URL)
+
+cluster.LogModulePrintf(cluster.Conf.Verbose, config.ConstLogModGeneral, config.LvlInfo, "Rolling jobs upgrade completed")
+
+return nil
+}

cluster/cluster_sec.go

@@ -9,15 +9,18 @@ package cluster
 import (
 "context"
 "crypto/tls"
+"encoding/json"
 "errors"
 "fmt"
+"io"
 "net"
 "net/smtp"
 "strings"
 
 vault "github.com/hashicorp/vault/api"
 "github.com/jordan-wright/email"
 "github.com/signal18/replication-manager/config"
+"github.com/signal18/replication-manager/utils/crypto"
 "github.com/signal18/replication-manager/utils/dbhelper"
 "github.com/signal18/replication-manager/utils/misc"
 "github.com/sirupsen/logrus"
@@ -583,3 +586,46 @@ func (cluster *Cluster) RevokeDBUserGrants(user string) error {
 }
 return nil
 }
+
+type DecodedData struct {
+Data string `json:"data"`
+}
+
+func (cluster *Cluster) SecretLoginCheck(vars map[string]string, rbody io.ReadCloser) (*ServerMonitor, error, int) {
+var decodedData DecodedData
+body, err := io.ReadAll(rbody)
+if err != nil {
+return nil, fmt.Errorf("Decode reading body :%s", err.Error()), 500
+}
+
+err = json.Unmarshal(body, &decodedData)
+if err != nil {
+return nil, fmt.Errorf("Decode body :%s. Err: %s", string(body), err.Error()), 400
+}
+
+var node *ServerMonitor
+if vars["serverPort"] == "" {
+node = cluster.GetServerFromName(vars["serverName"])
+} else {
+node = cluster.GetServerFromURL(vars["serverName"] + ":" + vars["serverPort"])
+}
+if node == nil {
+return nil, fmt.Errorf("No server"), 500
+}
+// Decrypt the encrypted data
+key := crypto.GetSHA256Hash(node.Pass)
+iv := crypto.GetMD5Hash(node.Pass)
+
+cluster.LogModulePrintf(cluster.Conf.Verbose, config.ConstLogModTask, config.LvlDbg, "Received login with encrypted secret %s", decodedData.Data)
+
+decrypted, err := node.DecodeSecret(decodedData.Data, key, iv)
+if err != nil {
+return nil, fmt.Errorf("Error decrypting data : %s", err.Error()), 500
+}
+
+if decrypted != cluster.GetDbPass() {
+return nil, fmt.Errorf("Invalid secret"), 401
+}
+
+return node, nil, 200
+}