Allow remote-config to be generated from a standby

pkg/common/mysql.go

@@ -47,7 +47,9 @@ type MySQLOpts struct {
 	ErrWriter *bytes.Buffer
 }
 
-func NewMySQLOpts(f cmdutil.Factory, dbName, namespace string) (*MySQLOpts, error) {
+func NewMySQLOpts(f cmdutil.Factory, dbName, namespace string, options ...OptionFunc) (*MySQLOpts, error) {
+	cfg := buildOptionConfig(options)
+
 	config, err := f.ToRESTConfig()
 	if err != nil {
 		return nil, err
@@ -75,7 +77,7 @@ func NewMySQLOpts(f cmdutil.Factory, dbName, namespace string) (*MySQLOpts, erro
 		return nil, err
 	}
 
-	if db.Status.Phase != dbapi.DatabasePhaseReady {
+	if !cfg.skipReadinessCheck && db.Status.Phase != dbapi.DatabasePhaseReady {
 		return nil, fmt.Errorf("MySQL %s/%s is not ready", namespace, dbName)
 	}
 

pkg/common/options.go

@@ -0,0 +1,43 @@
+/*
+Copyright AppsCode Inc. and Contributors
+
+Licensed under the AppsCode Community License 1.0.0 (the "License");
+you may not use this file except in compliance with the License.
+You may obtain a copy of the License at
+
+    https://github.com/appscode/licenses/raw/1.0.0/AppsCode-Community-1.0.0.md
+
+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.
+*/
+
+package common
+
+// OptionFunc tunes how a per-engine Opts value is built.
+type OptionFunc func(*optionConfig)
+
+type optionConfig struct {
+	skipReadinessCheck bool
+}
+
+// SkipReadinessCheck drops the `status.phase == Ready` gate on the database CR.
+// Commands that only read the CR to render manifests do not need a healthy
+// database, and refusing to run against a degraded source is actively unhelpful
+// for disaster-recovery workflows, where the source being unhealthy is the
+// whole reason the command is being run.
+func SkipReadinessCheck() OptionFunc {
+	return func(cfg *optionConfig) {
+		cfg.skipReadinessCheck = true
+	}
+}
+
+func buildOptionConfig(opts []OptionFunc) optionConfig {
+	var cfg optionConfig
+	for _, opt := range opts {
+		opt(&cfg)
+	}
+	return cfg
+}

pkg/common/postgres.go

@@ -49,7 +49,9 @@ type PostgresOpts struct {
 	ErrWriter *bytes.Buffer
 }
 
-func NewPostgresOpts(f cmdutil.Factory, dbName, namespace string) (*PostgresOpts, error) {
+func NewPostgresOpts(f cmdutil.Factory, dbName, namespace string, options ...OptionFunc) (*PostgresOpts, error) {
+	cfg := buildOptionConfig(options)
+
 	config, err := f.ToRESTConfig()
 	if err != nil {
 		return nil, err
@@ -77,7 +79,7 @@ func NewPostgresOpts(f cmdutil.Factory, dbName, namespace string) (*PostgresOpts
 		return nil, err
 	}
 
-	if db.Status.Phase != dbapi.DatabasePhaseReady {
+	if !cfg.skipReadinessCheck && db.Status.Phase != dbapi.DatabasePhaseReady {
 		return nil, fmt.Errorf("postgres %s/%s is not ready", namespace, dbName)
 	}
 

pkg/remote_replica/mysql.go

@@ -21,6 +21,7 @@ import (
 	"fmt"
 	"log"
 	"os"
+	"strings"
 	"time"
 
 	dbapi "kubedb.dev/apimachinery/apis/kubedb/v1"
@@ -31,9 +32,9 @@ import (
 	core "k8s.io/api/core/v1"
 	kerr "k8s.io/apimachinery/pkg/api/errors"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
-	"k8s.io/apimachinery/pkg/labels"
 	"k8s.io/apimachinery/pkg/util/sets"
 	"k8s.io/apimachinery/pkg/util/wait"
+	"k8s.io/klog/v2"
 	cmdutil "k8s.io/kubectl/pkg/cmd/util"
 	cm_util "kmodules.xyz/cert-manager-util/certmanager/v1"
 	kutil "kmodules.xyz/client-go"
@@ -59,7 +60,7 @@ const (
 
 func MysqlAPP(f cmdutil.Factory) *cobra.Command {
 	var userName, password, dns, ns string
-	var yes bool
+	var yes, skipUserCreation bool
 	cmd := cobra.Command{
 		Use:     "mysql",
 		Short:   desLong,
@@ -71,12 +72,16 @@ func MysqlAPP(f cmdutil.Factory) *cobra.Command {
 			if len(args) == 0 {
 				log.Fatal("no database name given")
 			}
-			if err := userPrompt(yes); err != nil {
-				log.Fatal(err)
+			// Nothing on the source is altered when user creation is skipped, so
+			// the "password will be altered" confirmation has nothing to confirm.
+			if !skipUserCreation {
+				if err := userPrompt(yes); err != nil {
+					log.Fatal(err)
+				}
 			}
 			var buffer []byte
 
-			buffer, err := generateMySQLConfig(f, userName, password, dns, ns, args[0])
+			buffer, err := generateMySQLConfig(f, userName, password, dns, ns, args[0], skipUserCreation)
 			if err != nil {
 				log.Fatal(err)
 			}
@@ -111,13 +116,22 @@ func MysqlAPP(f cmdutil.Factory) *cobra.Command {
 		log.Fatal(err)
 	}
 	cmd.PersistentFlags().BoolVarP(&yes, "yes", "y", false, "permission for alter password  for the remote replica")
+	cmd.PersistentFlags().BoolVar(&skipUserCreation, "skip-user-creation", false,
+		"do not create or alter the replication user on the source; assume it already exists with the given password. "+
+			"Needed when the source has no writable primary to exec into")
 	return &cmd
 }
 
-func generateMySQLConfig(f cmdutil.Factory, userName string, password string, dns string, ns string, dbname string) ([]byte, error) {
+func generateMySQLConfig(f cmdutil.Factory, userName string, password string, dns string, ns string, dbname string, skipUserCreation bool) ([]byte, error) {
 	var buffer []byte
 
-	opts, err := common.NewMySQLOpts(f, dbname, ns)
+	var dbOpts []common.OptionFunc
+	if skipUserCreation {
+		// Rendering manifests only reads the CR; a non-Ready source should not
+		// block a disaster recovery config from being generated.
+		dbOpts = append(dbOpts, common.SkipReadinessCheck())
+	}
+	opts, err := common.NewMySQLOpts(f, dbname, ns, dbOpts...)
 	if err != nil {
 		return nil, fmt.Errorf("failed to get db %s, err:%v", dbname, err)
 	}
@@ -127,7 +141,7 @@ func generateMySQLConfig(f cmdutil.Factory, userName string, password string, dn
 		return nil, fmt.Errorf("failed to get appbinding %v", err)
 	}
 
-	authBuff, authSecretName, err := generateMySQLAuthSecret(userName, password, ns, opts)
+	authBuff, authSecretName, err := generateMySQLAuthSecret(userName, password, ns, skipUserCreation, opts)
 	if err != nil {
 		return nil, fmt.Errorf("failed to generate auth secret ,%v", err)
 	}
@@ -198,15 +212,20 @@ func generateMySQLTlsSecret(userName string, apb *appApi.AppBinding, ns string,
 	return buffer, tlsSecret.Name, nil
 }
 
-func generateMySQLAuthSecret(userName string, password string, ns string, opts *common.MySQLOpts) ([]byte, string, error) {
-	if userName != opts.Username {
+func generateMySQLAuthSecret(userName string, password string, ns string, skipUserCreation bool, opts *common.MySQLOpts) ([]byte, string, error) {
+	switch {
+	case userName == opts.Username:
+		password = opts.Pass
+	case skipUserCreation:
+		if err := verifyMySQLUser(opts, userName); err != nil {
+			return nil, "", err
+		}
+	default:
 		// generate user if not present
 		err := generateMySQLUser(opts, userName, password)
 		if err != nil {
 			return nil, "", fmt.Errorf("failed to generate user err:%v", err)
 		}
-	} else {
-		password = opts.Pass
 	}
 	// generate auth secret
 	AuthSecret := core.Secret{
@@ -235,17 +254,48 @@ func generateMySQLAuthSecret(userName string, password string, ns string, opts *
 	return buffer, AuthSecret.Name, nil
 }
 
-func generateMySQLUser(opts *common.MySQLOpts, name string, password string) error {
-	label := opts.DB.OffshootLabels()
-	if *opts.DB.Spec.Replicas > 1 {
-		label["kubedb.com/role"] = "primary"
+// verifyMySQLUser checks the replication user without writing anything. Reading
+// mysql.user works on a read replica, so any running member will do. An
+// unreachable database downgrades to a warning: generating the config is still
+// the useful outcome.
+func verifyMySQLUser(opts *common.MySQLOpts, name string) error {
+	pod, err := pickDBPod(opts.Client, opts.DB.Namespace, opts.DB.OffshootLabels(), MySQLContainerName, false)
+	if err != nil {
+		klog.Warningf("skipping verification of user %q: %v", name, err)
+		return nil
 	}
 
-	pods, err := opts.Client.CoreV1().Pods(opts.DB.Namespace).List(context.TODO(), metav1.ListOptions{
-		LabelSelector: labels.Set.String(label),
-	})
-	if err != nil || len(pods.Items) == 0 {
-		return err
+	query := fmt.Sprintf("export MYSQL_PWD='%s' && mysql -uroot -N -B -e \"SELECT Repl_slave_priv FROM mysql.user WHERE user='%s'\"", opts.Pass, name)
+	out, err := exec_util.ExecIntoPod(opts.Config, pod,
+		exec_util.Command("bash", "-c", query),
+		exec_util.Container(MySQLContainerName),
+	)
+	if err != nil {
+		klog.Warningf("skipping verification of user %q: failed to query %s: %v", name, pod.Name, err)
+		return nil
+	}
+
+	switch strings.TrimSpace(out) {
+	case "":
+		return fmt.Errorf("user %q does not exist on %s/%s; create it on the writable primary first:\n  "+
+			"CREATE USER %s IDENTIFIED BY '<password>'; GRANT REPLICATION SLAVE, CLONE_ADMIN, BACKUP_ADMIN ON *.* TO '%s'@'%%';",
+			name, opts.DB.Namespace, opts.DB.Name, name, name)
+	case "N":
+		return fmt.Errorf("user %q exists on %s/%s but lacks REPLICATION SLAVE; run on the writable primary:\n  "+
+			"GRANT REPLICATION SLAVE, CLONE_ADMIN, BACKUP_ADMIN ON *.* TO '%s'@'%%';",
+			name, opts.DB.Namespace, opts.DB.Name, name)
+	}
+
+	fmt.Printf("user %q verified on %s (replication granted); leaving the source catalog untouched\n", name, pod.Name)
+	return nil
+}
+
+func generateMySQLUser(opts *common.MySQLOpts, name string, password string) error {
+	// DDL only: a clustered source must be addressed through its primary.
+	pod, err := pickDBPod(opts.Client, opts.DB.Namespace, opts.DB.OffshootLabels(), MySQLContainerName, *opts.DB.Spec.Replicas > 1)
+	if err != nil {
+		return fmt.Errorf("%v; pass --skip-user-creation to generate the config against a standby "+
+			"whose replication user already exists", err)
 	}
 	query := fmt.Sprintf("export MYSQL_PWD='%s' && mysql -uroot -e \"create user if not exists %s; alter user %s identified by '%s';"+
 		"GRANT REPLICATION SLAVE,  CLONE_ADMIN, BACKUP_ADMIN ON *.* TO '%s'@'%%' WITH GRANT OPTION; \"", opts.Pass,
@@ -257,7 +307,7 @@ func generateMySQLUser(opts *common.MySQLOpts, name string, password string) err
 		container,
 	}
 
-	_, err = exec_util.ExecIntoPod(opts.Config, &pods.Items[0], options...)
+	_, err = exec_util.ExecIntoPod(opts.Config, pod, options...)
 	if err != nil {
 		return err
 	}

pkg/remote_replica/pods.go

@@ -0,0 +1,98 @@
+/*
+Copyright AppsCode Inc. and Contributors
+
+Licensed under the AppsCode Community License 1.0.0 (the "License");
+you may not use this file except in compliance with the License.
+You may obtain a copy of the License at
+
+    https://github.com/appscode/licenses/raw/1.0.0/AppsCode-Community-1.0.0.md
+
+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.
+*/
+
+package remote_replica
+
+import (
+	"context"
+	"fmt"
+
+	"kubedb.dev/apimachinery/apis/kubedb"
+
+	core "k8s.io/api/core/v1"
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	"k8s.io/apimachinery/pkg/labels"
+	"k8s.io/client-go/kubernetes"
+)
+
+// Containers holding the database server itself, the ones a client query has to
+// be exec-ed into.
+const (
+	PostgresContainerName = "postgres"
+	MySQLContainerName    = "mysql"
+)
+
+// pickDBPod returns a running database pod to exec into.
+//
+// primaryOnly is set when the caller intends to run DDL. CREATE ROLE / ALTER
+// ROLE / GRANT write to the shared catalog, and a hot standby rejects them with
+// "cannot execute ... in a read-only transaction", so only a pod labelled
+// kubedb.com/role=primary qualifies. Read-only catalog queries run fine on any
+// member, so verification passes primaryOnly=false and merely prefers the
+// primary when one happens to be labelled.
+func pickDBPod(client kubernetes.Interface, ns string, selector map[string]string, container string, primaryOnly bool) (*core.Pod, error) {
+	pods, err := client.CoreV1().Pods(ns).List(context.TODO(), metav1.ListOptions{
+		LabelSelector: labels.Set(selector).String(),
+	})
+	if err != nil {
+		return nil, err
+	}
+
+	pod, err := selectPod(pods.Items, container, primaryOnly)
+	if err != nil {
+		return nil, fmt.Errorf("in namespace %s: %v", ns, err)
+	}
+	return pod, nil
+}
+
+// selectPod holds the choice itself, kept free of any client so it can be
+// exercised directly. Only pods that actually carry `container` are eligible: a
+// KubeDB cluster runs helper pods under the same offshoot labels - the Postgres
+// arbiter, for one, runs pg-coordinator alone - and exec-ing a database query
+// into those fails in a way that is hard to read.
+func selectPod(pods []core.Pod, container string, primaryOnly bool) (*core.Pod, error) {
+	var usable []core.Pod
+	for i := range pods {
+		if pods[i].Status.Phase == core.PodRunning && hasRunningContainer(&pods[i], container) {
+			usable = append(usable, pods[i])
+		}
+	}
+	if len(usable) == 0 {
+		// Reported explicitly: an empty list used to be swallowed and treated as
+		// success, which produced a config for a user that was never created.
+		return nil, fmt.Errorf("none of the %d database pod(s) has a running %q container", len(pods), container)
+	}
+
+	for i := range usable {
+		if usable[i].Labels[kubedb.LabelRole] == kubedb.DatabasePodPrimary {
+			return &usable[i], nil
+		}
+	}
+	if primaryOnly {
+		return nil, fmt.Errorf("no pod is labelled %s=%s, so there is no writable primary to run DDL against",
+			kubedb.LabelRole, kubedb.DatabasePodPrimary)
+	}
+	return &usable[0], nil
+}
+
+func hasRunningContainer(pod *core.Pod, container string) bool {
+	for _, cs := range pod.Status.ContainerStatuses {
+		if cs.Name == container {
+			return cs.State.Running != nil
+		}
+	}
+	return false
+}