Startup probe (#648)

* Update chkmqstarted to check in-sync for Native-HA

Author: sdmarshall79

cmd/chkmqstarted/main.go

@@ -23,16 +23,65 @@ import (
 	"os"
 	"os/exec"
 	"os/signal"
+	"strconv"
 	"strings"
+	"time"
 
+	"github.com/ibm-messaging/mq-container/internal/ready"
 	"github.com/ibm-messaging/mq-container/pkg/name"
 )
 
 func queueManagerStarted(ctx context.Context) (bool, error) {
+
 	name, err := name.GetQueueManagerName()
 	if err != nil {
 		return false, err
 	}
+
+	readyStrings := []string{
+		"(RUNNING)",
+		"(RUNNING AS STANDBY)",
+		"(RECOVERY GROUP LEADER)",
+		"(STARTING)",
+		"(REPLICA)",
+	}
+
+	// For Native-HA only, check if the queue manager instance is in-sync with one or more replicas
+	// - If not in-sync within the expected time period, revert to checking on queue manager 'ready' status
+	// - This ensures we do not block indefinitely for breaking changes (i.e. protocol changes)
+	if os.Getenv("MQ_NATIVE_HA") == "true" {
+
+		// Check if the Native-HA queue manager instance is currently in-sync
+		isReadyToSync, isInSync, err := isInSyncWithReplicas(ctx, name, readyStrings)
+		if err != nil {
+			return false, err
+		} else if isInSync {
+			return true, nil
+		}
+
+		// Check if the Native-HA queue manager instance is ready-to-sync
+		// - A successful queue manager 'ready' status indicates that we are ready-to-sync
+		if !isReadyToSync {
+			return false, nil
+		}
+		err = ready.SetReadyToSync()
+		if err != nil {
+			return false, err
+		}
+
+		// Check if the time period for checking in-sync has now expired
+		// - We have already confirmed a successful queue manager 'ready' status
+		// - Therefore the expiration of the in-sync time period will result in success
+		expired, err := hasInSyncTimePeriodExpired()
+		if err != nil {
+			return false, err
+		} else if expired {
+			return true, nil
+		}
+
+		return false, nil
+	}
+
 	// Specify the queue manager name, just in case someone's created a second queue manager
 	// #nosec G204
 	cmd := exec.CommandContext(ctx, "dspmq", "-n", "-m", name)
@@ -42,18 +91,60 @@ func queueManagerStarted(ctx context.Context) (bool, error) {
 		fmt.Println(err)
 		return false, err
 	}
-	readyStrings := []string{
-		"(RUNNING)",
-		"(RUNNING AS STANDBY)",
-		"(RECOVERY GROUP LEADER)",
-		"(STARTING)",
-		"(REPLICA)",
-	}
+
 	for _, checkString := range readyStrings {
 		if strings.Contains(string(out), checkString) {
 			return true, nil
 		}
 	}
+
+	return false, nil
+}
+
+// isInSyncWithReplicas returns the in-sync status for a Native-HA queue manager instance
+func isInSyncWithReplicas(ctx context.Context, name string, readyStrings []string) (bool, bool, error) {
+
+	cmd := exec.CommandContext(ctx, "dspmq", "-n", "-o", "nativeha", "-m", name)
+	out, err := cmd.CombinedOutput()
+	if err != nil {
+		return false, false, err
+	} else if strings.Contains(string(out), "INSYNC(YES)") {
+		return true, true, nil
+	}
+
+	for _, checkString := range readyStrings {
+		if strings.Contains(string(out), checkString) {
+			return true, false, nil
+		}
+	}
+
+	return false, false, nil
+}
+
+// hasInSyncTimePeriodExpired returns true if a Native-HA queue manager instance is not in-sync within the expected time period, otherwise false
+func hasInSyncTimePeriodExpired() (bool, error) {
+
+	// Default timeout 5 seconds
+	var timeout int64 = 5
+	var err error
+
+	// Check if a timeout override has been set
+	customTimeout := os.Getenv("MQ_NATIVE_HA_IN_SYNC_TIMEOUT")
+	if customTimeout != "" {
+		timeout, err = strconv.ParseInt(customTimeout, 10, 64)
+		if err != nil {
+			return false, err
+		}
+	}
+
+	isReadyToSync, readyToSyncStartTime, err := ready.GetReadyToSyncStartTime()
+	if err != nil {
+		return false, err
+	}
+	if isReadyToSync && time.Now().Unix()-readyToSyncStartTime.Unix() >= timeout {
+		return true, nil
+	}
+
 	return false, nil
 }
 

internal/ready/ready.go

@@ -20,14 +20,17 @@ package ready
 import (
 	"context"
 	"os"
+	"strconv"
 	"strings"
+	"time"
 
 	"github.com/ibm-messaging/mq-container/internal/command"
 )
 
-const fileName string = "/run/runmqserver/ready"
+const readyFile string = "/run/runmqserver/ready"
+const readyToSyncFile string = "/run/runmqserver/ready-to-sync"
 
-func fileExists() (bool, error) {
+func fileExists(fileName string) (bool, error) {
 	_, err := os.Stat(fileName)
 	if err != nil {
 		if !os.IsNotExist(err) {
@@ -40,7 +43,21 @@ func fileExists() (bool, error) {
 
 // Clear ensures that any readiness state is cleared
 func Clear() error {
-	exist, err := fileExists()
+	err := clearFile(readyFile)
+	if err != nil {
+		return err
+	}
+	err = clearFile(readyToSyncFile)
+	if err != nil {
+		return err
+	}
+
+	return nil
+}
+
+// clearFile removes the specified file if it exists
+func clearFile(fileName string) error {
+	exist, err := fileExists(fileName)
 	if err != nil {
 		return err
 	}
@@ -54,19 +71,57 @@ func Clear() error {
 // manager has finished its configuration step
 func Set() error {
 	// #nosec G306 - this gives permissions to owner/s group only.
-	return os.WriteFile(fileName, []byte("1"), 0770)
+	return os.WriteFile(readyFile, []byte("1"), 0770)
 }
 
 // Check checks whether or not the queue manager has finished its
 // configuration steps
 func Check() (bool, error) {
-	exists, err := fileExists()
+	exists, err := fileExists(readyFile)
 	if err != nil {
 		return false, err
 	}
 	return exists, nil
 }
 
+// SetReadyToSync is used to indicate that a Native-HA queue manager instance is ready-to-sync
+func SetReadyToSync() error {
+
+	exists, err := fileExists(readyToSyncFile)
+	if err != nil {
+		return err
+	} else if exists {
+		return nil
+	}
+
+	readyToSyncStartTime := strconv.FormatInt(time.Now().Unix(), 10)
+	// #nosec G306 - required permissions
+	return os.WriteFile(readyToSyncFile, []byte(readyToSyncStartTime), 0660)
+}
+
+// GetReadyToSyncStartTime returns the start-time a Native-HA queue manager instance was ready-to-sync
+func GetReadyToSyncStartTime() (bool, time.Time, error) {
+
+	exists, err := fileExists(readyToSyncFile)
+	if err != nil {
+		return exists, time.Time{}, err
+	}
+
+	if exists {
+		buf, err := os.ReadFile(readyToSyncFile)
+		if err != nil {
+			return true, time.Time{}, err
+		}
+		readyToSyncStartTime, err := strconv.ParseInt(string(buf), 10, 64)
+		if err != nil {
+			return true, time.Time{}, err
+		}
+		return true, time.Unix(readyToSyncStartTime, 0), nil
+	}
+
+	return false, time.Time{}, nil
+}
+
 // Status returns an enum representing the current running status of the queue manager
 func Status(ctx context.Context, name string) (QMStatus, error) {
 	out, _, err := command.RunContext(ctx, "dspmq", "-n", "-m", name)