implement pxc replication
MODIFY
build/pxc-configure-pxc.sh
build/pxc-configure-pxc.sh
@@ -39,7 +39,8 @@ function mysql_root_exec() {
NODE_IP=$(hostname -I | awk ' { print $1 } ')
CLUSTER_NAME="$(hostname -f | cut -d'.' -f2)"
-SERVER_ID=${HOSTNAME/$CLUSTER_NAME-}
+SERVER_NUM=${HOSTNAME/$CLUSTER_NAME-/}
+SERVER_ID=${CLUSTER_HASH}${SERVER_NUM}
NODE_NAME=$(hostname -f)
NODE_PORT=3306
@@ -80,7 +81,7 @@ fi
grep -E -q "^[#]?wsrep_sst_donor" "$CFG" || sed '/^\[mysqld\]/a wsrep_sst_donor=\n' ${CFG} 1<>${CFG}
grep -E -q "^[#]?wsrep_node_incoming_address" "$CFG" || sed '/^\[mysqld\]/a wsrep_node_incoming_address=\n' ${CFG} 1<>${CFG}
grep -E -q "^[#]?wsrep_provider_options" "$CFG" || sed '/^\[mysqld\]/a wsrep_provider_options="pc.weight=10"\n' ${CFG} 1<>${CFG}
-sed -r "s|^[#]?server_id=.*$|server_id=1${SERVER_ID}|" ${CFG} 1<>${CFG}
+sed -r "s|^[#]?server_id=.*$|server_id=${SERVER_ID}|" ${CFG} 1<>${CFG}
sed -r "s|^[#]?coredumper$|coredumper|" ${CFG} 1<>${CFG}
sed -r "s|^[#]?wsrep_node_address=.*$|wsrep_node_address=${NODE_IP}|" ${CFG} 1<>${CFG}
sed -r "s|^[#]?wsrep_cluster_name=.*$|wsrep_cluster_name=${CLUSTER_NAME}|" ${CFG} 1<>${CFG}

MODIFY
build/pxc-entrypoint.sh
build/pxc-entrypoint.sh
@@ -148,6 +148,7 @@ function join {
}
MYSQL_VERSION=$(mysqld -V | awk '{print $3}' | awk -F'.' '{print $1"."$2}')
+MYSQL_PATCH_VERSION=$(mysqld -V | awk '{print $3}' | awk -F'.' '{print $3}' | awk -F'-' '{print $1}')
# if vault secret file exists we assume we need to turn on encryption
vault_secret="/etc/mysql/vault-keyring-secret/keyring_vault.conf"
@@ -179,6 +180,13 @@ fi
grep -q "^progress=" $CFG && sed -i "s|^progress=.*|progress=1|" $CFG
grep -q "^\[sst\]" "$CFG" || printf '[sst]\n' >>"$CFG"
grep -q "^cpat=" "$CFG" || sed '/^\[sst\]/a cpat=.*\\.pem$\\|.*init\\.ok$\\|.*galera\\.cache$\\|.*wsrep_recovery_verbose\\.log$\\|.*readiness-check\\.sh$\\|.*liveness-check\\.sh$\\|.*sst_in_progress$\\|.*sst-xb-tmpdir$\\|.*\\.sst$\\|.*gvwstate\\.dat$\\|.*grastate\\.dat$\\|.*\\.err$\\|.*\\.log$\\|.*RPM_UPGRADE_MARKER$\\|.*RPM_UPGRADE_HISTORY$\\|.*pxc-entrypoint\\.sh$\\|.*unsafe-bootstrap\\.sh$\\|.*pxc-configure-pxc\\.sh\\|.*peer-list$' "$CFG" 1<>"$CFG"
+if [[ "$MYSQL_VERSION" == '8.0' ]]; then
+ if [[ $MYSQL_PATCH_VERSION -ge 26 ]]; then
+ grep -q "^skip_replica_start=ON" "$CFG" || sed -i "/\[mysqld\]/a skip_replica_start=ON" $CFG
+ else
+ grep -q "^skip_slave_start=ON" "$CFG" || sed -i "/\[mysqld\]/a skip_slave_start=ON" $CFG
+ fi
+fi
file_env 'XTRABACKUP_PASSWORD' 'xtrabackup' 'xtrabackup'
file_env 'CLUSTERCHECK_PASSWORD' '' 'clustercheck'
@@ -302,8 +310,13 @@ if [ -z "$CLUSTER_JOIN" ] && [ "$1" = 'mysqld' -a -z "$wantHelp" ]; then
fi
if [ -z "$MYSQL_INITDB_SKIP_TZINFO" ]; then
+ (
+ echo "set wsrep_on=0;"
+ echo "SET @@SESSION.SQL_LOG_BIN = off;"
# sed is for https://bugs.mysql.com/bug.php?id=20545
- mysql_tzinfo_to_sql /usr/share/zoneinfo | sed 's/Local time zone must be set--see zic manual page/FCTY/' | "${mysql[@]}" mysql
+ mysql_tzinfo_to_sql /usr/share/zoneinfo | sed 's/Local time zone must be set--see zic manual page/FCTY/'
+ echo "set wsrep_on=1;"
+ ) | "${mysql[@]}" mysql
fi
{ set +x; } 2>/dev/null

MODIFY
deploy/cr.yaml
deploy/cr.yaml
@@ -34,7 +34,7 @@ spec:
schedule: "0 4 * * *"
pxc:
size: 3
- image: percona/percona-xtradb-cluster:8.0.22-13.1
+ image: percona/percona-xtradb-cluster:8.0.23-14.1
autoRecovery: true
# expose:
# enabled: true
@@ -47,7 +47,12 @@ spec:
# replicationChannels:
# - name: pxc1_to_pxc2
# isSource: true
-# secretName: my-cluster-name-replication
+# - name: pxc2_to_pxc1
+# isSource: false
+# sourcesList:
+# - host: 10.95.251.101
+# port: 3306
+# weight: 100
# schedulerName: mycustom-scheduler
# readinessDelaySec: 15
# livenessDelaySec: 600

MODIFY
e2e-tests/functions
e2e-tests/functions
@@ -319,6 +319,7 @@ compare_kubectl() {
| yq d - '**.namespace' \
| yq d - '**.uid' \
| yq d - 'metadata.resourceVersion' \
+ | yq d - '**.env.(name==CLUSTER_HASH)' \
| yq d - 'metadata.selfLink' \
| yq d - 'metadata.deletionTimestamp' \
| yq d - 'metadata.annotations."k8s.v1.cni.cncf.io*"' \

MODIFY
pkg/apis/pxc/v1/pxc_types.go
pkg/apis/pxc/v1/pxc_types.go
@@ -62,7 +62,13 @@ type ServiceExpose struct {
type ReplicationChannel struct {
Name string `json:"name,omitempty"`
IsSource bool `json:"isSource,omitempty"`
- SecretName string `json:"secretName,omitempty"`
+ SourcesList []ReplicationSource `json:"sourcesList,omitempty"`
+}
+
+type ReplicationSource struct {
+ Host string `json:"host,omitempty"`
+ Port int `json:"port,omitempty"`
+ Weight int `json:"weight,omitempty"`
}
type TLSSpec struct {
@@ -194,12 +200,36 @@ func (cr *PerconaXtraDBCluster) Validate() error {
return errors.New("pxc.Image can't be empty")
}
- for _, v := range c.PXC.ReplicationChannels {
- if v.Name == "" {
- return errors.New("pxc.replicationChannels.Name can't be empty")
+ if len(c.PXC.ReplicationChannels) > 0 {
+ // since we do not allow multimaster
+ // isSource field should be equal everywhere
+ isSrc := c.PXC.ReplicationChannels[0].IsSource
+ for _, channel := range c.PXC.ReplicationChannels {
+ // this restrictions coming from mysql itself
+ if len(channel.Name) > 64 || channel.Name == "" || channel.Name == "group_replication_applier" || channel.Name == "group_replication_recovery" {
+ return errors.Errorf("invalid replication channel name %s, please see channel naming conventions", channel.Name)
+ }
+
+ if isSrc != channel.IsSource {
+ return errors.New("you can specify only one type of replication please specify equal values for isSource field")
+ }
+
+ if channel.IsSource {
+ continue
+ }
+
+ if len(channel.SourcesList) == 0 {
+ return errors.Errorf("sources list for replication channel %s should be empty, because it's replica", channel.Name)
+ }
+
+ for _, src := range channel.SourcesList {
+ if src.Weight == 0 {
+ src.Weight = 100
+ }
+ if src.Port == 0 {
+ src.Port = 3306
+ }
}
- if v.SecretName == "" {
- return errors.New("pxc.replicationChannels.SecretName can't be empty")
}
}

MODIFY
pkg/controller/pxc/controller.go
pkg/controller/pxc/controller.go
@@ -236,7 +236,7 @@ func (r *ReconcilePerconaXtraDBCluster) Reconcile(request reconcile.Request) (re
if o.CompareVersionWith("1.7.0") >= 0 && *o.Spec.PXC.AutoRecovery {
err = r.recoverFullClusterCrashIfNeeded(o)
if err != nil {
- reqLogger.Error(err, "Failed to check if cluster needs to recover")
+ reqLogger.Info("Failed to check if cluster needs to recover", "err", err.Error())
}
}
@@ -510,6 +510,13 @@ func (r *ReconcilePerconaXtraDBCluster) Reconcile(request reconcile.Request) (re
}
}
+ if o.CompareVersionWith("1.9.0") >= 0 {
+ err = r.reconcileReplication(o)
+ if err != nil {
+ reqLogger.Info("reconcile replication error", "err", err.Error())
+ }
+ }
+
err = r.reconcileBackups(o)
if err != nil {
return reconcile.Result{}, err

MODIFY
pkg/controller/pxc/replication.go
pkg/controller/pxc/replication.go
@@ -3,16 +3,26 @@ package pxc
import (
"context"
"fmt"
+ "strings"
+ "github.com/go-logr/logr"
+ "github.com/hashicorp/go-version"
api "github.com/percona/percona-xtradb-cluster-operator/pkg/apis/pxc/v1"
+ "github.com/percona/percona-xtradb-cluster-operator/pkg/pxc/app/statefulset"
+ "github.com/percona/percona-xtradb-cluster-operator/pkg/pxc/queries"
"github.com/pkg/errors"
corev1 "k8s.io/api/core/v1"
k8serrors "k8s.io/apimachinery/pkg/api/errors"
metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
"k8s.io/apimachinery/pkg/labels"
+ "k8s.io/apimachinery/pkg/types"
"sigs.k8s.io/controller-runtime/pkg/client"
)
+const replicationPodLabel = "percona.com/replicationPod"
+
+var minReplicationVersion = version.Must(version.NewVersion("8.0.23"))
+
func (r *ReconcilePerconaXtraDBCluster) ensurePxcPodServices(cr *api.PerconaXtraDBCluster) error {
if cr.Spec.Pause {
return nil
@@ -86,6 +96,325 @@ func (r *ReconcilePerconaXtraDBCluster) removeOutdatedServices(cr *api.PerconaXt
return nil
}
+func (r *ReconcilePerconaXtraDBCluster) reconcileReplication(cr *api.PerconaXtraDBCluster) error {
+ if cr.Status.PXC.Ready < 1 || cr.Spec.Pause {
+ return nil
+ }
+
+ logger := r.logger(cr.Name, cr.Namespace)
+
+ sfs := statefulset.NewNode(cr)
+
+ listRaw := corev1.PodList{}
+ err := r.client.List(context.TODO(),
+ &listRaw,
+ &client.ListOptions{
+ Namespace: cr.Namespace,
+ LabelSelector: labels.SelectorFromSet(sfs.Labels()),
+ },
+ )
+ if k8serrors.IsNotFound(err) {
+ return nil
+ } else if err != nil {
+ return errors.Wrap(err, "get pod list")
+ }
+
+ // we need only running pods, because we unable to
+ // connect to failed/pending pods
+ podList := make([]corev1.Pod, 0)
+ for _, pod := range listRaw.Items {
+ if isPodReady(pod) {
+ podList = append(podList, pod)
+ }
+ }
+
+ primary, err := r.getPrimaryPod(cr)
+ if err != nil {
+ return errors.Wrap(err, "get primary pxc pod")
+ }
+
+ var primaryPod *corev1.Pod
+ for _, pod := range podList {
+ if pod.Status.PodIP == primary || pod.Name == primary || strings.HasPrefix(primary, fmt.Sprintf("%s.%s.%s", pod.Name, sfs.StatefulSet().Name, cr.Namespace)) {
+ primaryPod = &pod
+ break
+ }
+ }
+
+ if primaryPod == nil {
+ logger.Info("Unable to find primary pod for replication. No pod with name or ip like this", "primary name", primary)
+ return nil
+ }
+
+ user := "operator"
+ port := int32(33062)
+
+ primaryDB, err := queries.New(r.client, cr.Namespace, internalPrefix+cr.Name, user, primaryPod.Name+"."+cr.Name+"-pxc."+cr.Namespace, port)
+ if err != nil {
+ return errors.Wrapf(err, "failed to connect to pod %s", primaryPod.Name)
+ }
+
+ defer primaryDB.Close()
+
+ dbVer, err := primaryDB.Version()
+ if err != nil {
+ return errors.Wrap(err, "failed to get current db version")
+ }
+
+ if minReplicationVersion.Compare(version.Must(version.NewVersion(dbVer))) < 0 {
+ logger.Info("Failed to reconcile replication. Unsupported mysql version", "minimum version", minReplicationVersion.String())
+ return nil
+ }
+
+ err = removeOutdatedChannels(primaryDB, cr.Spec.PXC.ReplicationChannels)
+ if err != nil {
+ return errors.Wrap(err, "remove outdated replication channels")
+ }
+
+ err = checkReadonlyStatus(cr.Spec.PXC.ReplicationChannels, podList, cr, r.client)
+ if err != nil {
+ return errors.Wrap(err, "failed to ensure cluster readonly status")
+ }
+
+ if len(cr.Spec.PXC.ReplicationChannels) == 0 {
+ return deleteReplicaLabels(r.client, podList)
+ }
+
+ if cr.Spec.PXC.ReplicationChannels[0].IsSource {
+ return deleteReplicaLabels(r.client, podList)
+ }
+
+ // if primary pod is not a replica, we need to make it as replica, and stop replication on other pods
+ _, ok := primaryPod.Labels[replicationPodLabel]
+ if !ok {
+ for _, pod := range podList {
+ if pod.Name == primaryPod.Name {
+ continue
+ }
+ if _, ok := pod.Labels[replicationPodLabel]; ok {
+ r.logger(cr.Name, cr.Namespace).Info("Replication pod has changed", "new replication pod", pod.Name)
+ db, err := queries.New(r.client, cr.Namespace, internalPrefix+cr.Name, user, pod.Name+"."+cr.Name+"-pxc."+cr.Namespace, port)
+ if err != nil {
+ return errors.Wrapf(err, "failed to connect to pod %s", pod.Name)
+ }
+ err = db.StopAllReplication()
+ db.Close()
+ if err != nil {
+ return errors.Wrapf(err, "stop replication on pod %s", pod.Name)
+ }
+ delete(pod.Labels, replicationPodLabel)
+ err = r.client.Update(context.TODO(), &pod)
+ if err != nil {
+ return errors.Wrap(err, "failed to remove primary label from secondary pod")
+ }
+ }
+ }
+ primaryPod.Labels[replicationPodLabel] = "true"
+ err = r.client.Update(context.TODO(), primaryPod)
+ if err != nil {
+ return errors.Wrap(err, "add label to main replica pod")
+ }
+ }
+
+ sysUsersSecretObj := corev1.Secret{}
+ err = r.client.Get(context.TODO(),
+ types.NamespacedName{
+ Namespace: cr.Namespace,
+ Name: internalPrefix + cr.Name,
+ },
+ &sysUsersSecretObj,
+ )
+ if err != nil {
+ return errors.Wrap(err, "get secrets")
+ }
+
+ for _, channels := range cr.Spec.PXC.ReplicationChannels {
+ if channels.IsSource {
+ continue
+ }
+ err = manageReplicationChannel(r.log, primaryDB, channels, string(sysUsersSecretObj.Data["replication"]))
+ if err != nil {
+ return errors.Wrapf(err, "manage replication channel %s", channels.Name)
+ }
+ }
+
+ return nil
+}
+
+func checkReadonlyStatus(channels []api.ReplicationChannel, pods []corev1.Pod, cr *api.PerconaXtraDBCluster, client client.Client) error {
+ isReplica := false
+ if len(channels) > 0 {
+ isReplica = !channels[0].IsSource
+ }
+
+ for _, pod := range pods {
+ db, err := queries.New(client, cr.Namespace, internalPrefix+cr.Name, "operator", pod.Name+"."+cr.Name+"-pxc."+cr.Namespace, 33062)
+ if err != nil {
+ return errors.Wrapf(err, "connect to pod %s", pod.Name)
+ }
+ defer db.Close()
+ readonly, err := db.IsReadonly()
+ if err != nil {
+ return errors.Wrap(err, "check readonly status")
+ }
+
+ if isReplica && readonly || (!isReplica && !readonly) {
+ continue
+ }
+
+ if isReplica && !readonly {
+ err = db.EnableReadonly()
+ }
+
+ if !isReplica && readonly {
+ err = db.DisableReadonly()
+ }
+ if err != nil {
+ return errors.Wrap(err, "enable or disable readonly mode")
+ }
+
+ }
+ return nil
+}
+
+func removeOutdatedChannels(db queries.Database, currentChannels []api.ReplicationChannel) error {
+ dbChannels, err := db.CurrentReplicationChannels()
+ if err != nil {
+ return errors.Wrap(err, "get current replication channels")
+ }
+
+ if len(dbChannels) == 0 {
+ return nil
+ }
+
+ toRemove := make(map[string]struct{})
+ for _, v := range dbChannels {
+ toRemove[v] = struct{}{}
+ }
+
+ for _, v := range currentChannels {
+ if !v.IsSource {
+ delete(toRemove, v.Name)
+ }
+ }
+
+ if len(toRemove) == 0 {
+ return nil
+ }
+
+ for channelToRemove := range toRemove {
+ err = db.StopReplication(channelToRemove)
+ if err != nil {
+ return errors.Wrapf(err, "stop replication for channel %s", channelToRemove)
+ }
+
+ srcList, err := db.ReplicationChannelSources(channelToRemove)
+ if err != nil && err != queries.ErrNotFound {
+ return errors.Wrapf(err, "get src list for outdated channel %s", channelToRemove)
+ }
+ for _, v := range srcList {
+ err = db.DeleteReplicationSource(channelToRemove, v.Host, v.Port)
+ if err != nil {
+ return errors.Wrapf(err, "delete replication source for outdated channel %s", channelToRemove)
+ }
+ }
+ }
+ return nil
+}
+
+func manageReplicationChannel(log logr.Logger, primaryDB queries.Database, channel api.ReplicationChannel, replicaPW string) error {
+ currentSources, err := primaryDB.ReplicationChannelSources(channel.Name)
+ if err != nil && err != queries.ErrNotFound {
+ return errors.Wrapf(err, "get current replication sources for channel %s", channel.Name)
+ }
+
+ replicationStatus, err := primaryDB.ReplicationStatus(channel.Name)
+ if err != nil {
+ return errors.Wrap(err, "failed to check replication status")
+ }
+
+ if !isSourcesChanged(channel.SourcesList, currentSources) {
+ if replicationStatus == queries.ReplicationStatusError {
+ log.Info("Replication for channel is not running. Please, check the replication status", "channel", channel.Name)
+ return nil
+ }
+
+ if replicationStatus == queries.ReplicationStatusActive {
+ return nil
+ }
+ }
+
+ if replicationStatus == queries.ReplicationStatusActive {
+ err = primaryDB.StopReplication(channel.Name)
+ if err != nil {
+ return errors.Wrapf(err, "stop replication for channel %s", channel.Name)
+ }
+ }
+
+ for _, src := range currentSources {
+ err = primaryDB.DeleteReplicationSource(channel.Name, src.Host, src.Port)
+ if err != nil {
+ return errors.Wrapf(err, "delete replication source for channel %s", channel.Name)
+ }
+ }
+
+ maxWeight := 0
+ maxWeightSrc := channel.SourcesList[0]
+
+ for _, src := range channel.SourcesList {
+ if src.Weight > maxWeight {
+ maxWeightSrc = src
+ }
+ err := primaryDB.AddReplicationSource(channel.Name, src.Host, src.Port, src.Weight)
+ if err != nil {
+ return errors.Wrapf(err, "add replication source for channel %s", channel.Name)
+ }
+ }
+
+ return primaryDB.StartReplication(replicaPW, queries.ReplicationChannelSource{
+ Name: channel.Name,
+ Host: maxWeightSrc.Host,
+ Port: maxWeightSrc.Port,
+ })
+}
+
+func isSourcesChanged(new []api.ReplicationSource, old []queries.ReplicationChannelSource) bool {
+ if len(new) != len(old) {
+ return true
+ }
+
+ oldSrc := make(map[string]queries.ReplicationChannelSource)
+ for _, src := range old {
+ oldSrc[src.Host] = src
+ }
+
+ for _, v := range new {
+ oldSource, ok := oldSrc[v.Host]
+ if !ok {
+ return true
+ }
+ if oldSource.Port != v.Port || oldSource.Weight != v.Weight {
+ return true
+ }
+ delete(oldSrc, v.Host)
+ }
+
+ return len(oldSrc) != 0
+}
+
+func deleteReplicaLabels(client client.Client, pods []corev1.Pod) error {
+ for _, pod := range pods {
+ if _, ok := pod.Labels[replicationPodLabel]; ok {
+ delete(pod.Labels, replicationPodLabel)
+ err := client.Update(context.TODO(), &pod)
+ if err != nil {
+ return errors.Wrap(err, "failed to remove replication label from pod")
+ }
+ }
+ }
+ return nil
+}
+
func (r *ReconcilePerconaXtraDBCluster) removePxcPodServices(cr *api.PerconaXtraDBCluster) error {
if cr.Spec.Pause {
return nil
@@ -162,3 +491,16 @@ func NewExposedPXCService(svcName string, cr *api.PerconaXtraDBCluster) *corev1.
return svc
}
+
+// isPodReady returns a boolean reflecting if a pod is in a "ready" state
+func isPodReady(pod corev1.Pod) bool {
+ for _, condition := range pod.Status.Conditions {
+ if condition.Status != corev1.ConditionTrue {
+ continue
+ }
+ if condition.Type == corev1.PodReady {
+ return true
+ }
+ }
+ return false
+}

MODIFY
pkg/controller/pxc/upgrade.go
pkg/controller/pxc/upgrade.go
@@ -5,6 +5,7 @@ import (
"crypto/md5"
"fmt"
"sort"
+ "strconv"
"strings"
"time"
@@ -13,9 +14,6 @@ import (
"github.com/pkg/errors"
- "strconv"
-
- "github.com/percona/percona-xtradb-cluster-operator/pkg/apis/pxc/v1"
api "github.com/percona/percona-xtradb-cluster-operator/pkg/apis/pxc/v1"
"github.com/percona/percona-xtradb-cluster-operator/pkg/pxc"
"github.com/percona/percona-xtradb-cluster-operator/pkg/pxc/queries"
@@ -203,7 +201,7 @@ func (r *ReconcilePerconaXtraDBCluster) updatePod(sfs api.StatefulApp, podSpec *
return errors.Wrap(err, "update error")
}
- if cr.Spec.UpdateStrategy != v1.SmartUpdateStatefulSetStrategyType {
+ if cr.Spec.UpdateStrategy != api.SmartUpdateStatefulSetStrategyType {
return nil
}

MODIFY
pkg/controller/pxc/version.go
pkg/controller/pxc/version.go
@@ -266,14 +266,16 @@ func (r *ReconcilePerconaXtraDBCluster) fetchVersionFromPXC(cr *apiv1.PerconaXtr
user := "root"
port := int32(3306)
+ secrets := cr.Spec.SecretsName
if cr.CompareVersionWith("1.6.0") >= 0 {
port = int32(33062)
+ secrets = "internal-" + cr.Name
}
logger := r.logger(cr.Name, cr.Namespace)
for _, pod := range list.Items {
- database, err := queries.New(r.client, cr.Namespace, cr.Spec.SecretsName, user, pod.Name+"."+cr.Name+"-pxc."+cr.Namespace, port)
+ database, err := queries.New(r.client, cr.Namespace, secrets, user, pod.Name+"."+cr.Name+"-pxc."+cr.Namespace, port)
if err != nil {
logger.Error(err, "failed to create db instance")
continue

MODIFY
pkg/pxc/app/statefulset/node.go
pkg/pxc/app/statefulset/node.go
@@ -2,6 +2,7 @@ package statefulset
import (
"fmt"
+ "hash/fnv"
appsv1 "k8s.io/api/apps/v1"
corev1 "k8s.io/api/core/v1"
@@ -189,6 +190,21 @@ func (c *Node) AppContainer(spec *api.PodSpec, secrets string, cr *api.PerconaXt
Optional: &fvar,
},
})
+ serverIDHash := fnv.New32()
+ serverIDHash.Write([]byte(string(cr.UID)))
+
+ // we cut first 3 symbols to give a space for hostname(actially, pod number)
+ // which is appended to all server ids. If we do not do this, it
+ // can cause a int32 overflow
+ // P.S max value is 4294967295
+ serverIDHashStr := fmt.Sprint(serverIDHash.Sum32())
+ if len(serverIDHashStr) > 7 {
+ serverIDHashStr = serverIDHashStr[:7]
+ }
+ appc.Env = append(appc.Env, corev1.EnvVar{
+ Name: "CLUSTER_HASH",
+ Value: serverIDHashStr,
+ })
}
if cr.CompareVersionWith("1.3.0") >= 0 {
for k, v := range appc.VolumeMounts {

MODIFY
pkg/pxc/queries/queries.go
pkg/pxc/queries/queries.go
@@ -3,8 +3,10 @@ package queries
import (
"context"
"database/sql"
- "errors"
"fmt"
+ "strings"
+
+ "github.com/pkg/errors"
_ "github.com/go-sql-driver/mysql"
corev1 "k8s.io/api/core/v1"
@@ -12,6 +14,14 @@ import (
"sigs.k8s.io/controller-runtime/pkg/client"
)
+type ReplicationStatus int8
+
+const (
+ ReplicationStatusActive ReplicationStatus = iota
+ ReplicationStatusError
+ ReplicationStatusNotInitiated
+)
+
// value of writer group is hardcoded in ProxySQL config inside docker image
// https://github.com/percona/percona-docker/blob/pxc-operator-1.3.0/proxysql/dockerdir/etc/proxysql-admin.cnf#L23
const writerID = 11
@@ -20,6 +30,13 @@ type Database struct {
db *sql.DB
}
+type ReplicationChannelSource struct {
+ Name string
+ Host string
+ Port int
+ Weight int
+}
+
var ErrNotFound = errors.New("not found")
func New(client client.Client, namespace, secretName, user, host string, port int32) (Database, error) {
@@ -52,8 +69,156 @@ func New(client client.Client, namespace, secretName, user, host string, port in
}, nil
}
+func (p *Database) CurrentReplicationChannels() ([]string, error) {
+ rows, err := p.db.Query(`SELECT DISTINCT(Channel_name) from replication_asynchronous_connection_failover`)
+ if err != nil {
+ if errors.Is(err, sql.ErrNoRows) {
+ return nil, nil
+ }
+ return nil, errors.Wrap(err, "select current replication channels")
+ }
+
+ defer rows.Close()
+
+ result := make([]string, 0)
+ for rows.Next() {
+ src := ""
+ err = rows.Scan(&src)
+ if err != nil {
+ return nil, errors.Wrap(err, "scan channel name")
+ }
+ result = append(result, src)
+ }
+ return result, nil
+}
+
+func (p *Database) ReplicationStatus(channel string) (ReplicationStatus, error) {
+ rows, err := p.db.Query(`SHOW REPLICA STATUS FOR CHANNEL ?`, channel)
+ if err != nil {
+ if strings.HasSuffix(err.Error(), "does not exist.") || errors.Is(err, sql.ErrNoRows) {
+ return ReplicationStatusNotInitiated, nil
+ }
+ return ReplicationStatusError, errors.Wrap(err, "get current replica status")
+ }
+
+ defer rows.Close()
+ cols, err := rows.Columns()
+ if err != nil {
+ return ReplicationStatusError, errors.Wrap(err, "get columns")
+ }
+ vals := make([]interface{}, len(cols))
+ for i := range cols {
+ vals[i] = new(sql.RawBytes)
+ }
+
+ for rows.Next() {
+ err = rows.Scan(vals...)
+ if err != nil {
+ return ReplicationStatusError, errors.Wrap(err, "scan replication status")
+ }
+ }
+
+ IORunning := string(*vals[10].(*sql.RawBytes))
+ SQLRunning := string(*vals[11].(*sql.RawBytes))
+ LastErrNo := string(*vals[18].(*sql.RawBytes))
+ if IORunning == "Yes" && SQLRunning == "Yes" {
+ return ReplicationStatusActive, nil
+ }
+
+ if IORunning == "No" && SQLRunning == "No" && LastErrNo == "0" {
+ return ReplicationStatusNotInitiated, nil
+ }
+
+ return ReplicationStatusError, nil
+}
+
+func (p *Database) StopAllReplication() error {
+ _, err := p.db.Exec("STOP REPLICA")
+ return errors.Wrap(err, "failed to stop replication")
+}
+
+func (p *Database) AddReplicationSource(name, host string, port, weight int) error {
+ _, err := p.db.Exec("SELECT asynchronous_connection_failover_add_source(?, ?, ?, null, ?)", name, host, port, weight)
+ return errors.Wrap(err, "add replication source")
+}
+
+func (p *Database) ReplicationChannelSources(channelName string) ([]ReplicationChannelSource, error) {
+ rows, err := p.db.Query(`
+ SELECT host,
+ port,
+ weight
+ FROM replication_asynchronous_connection_failover
+ WHERE channel_name = ?
+ `, channelName)
+ if err != nil {
+ if errors.Is(err, sql.ErrNoRows) {
+ return nil, ErrNotFound
+ }
+ return nil, errors.Wrap(err, "get replication channels")
+ }
+ defer rows.Close()
+ result := make([]ReplicationChannelSource, 0)
+ for rows.Next() {
+ r := ReplicationChannelSource{}
+ err = rows.Scan(&r.Host, &r.Port, &r.Weight)
+ if err != nil {
+ return nil, errors.Wrap(err, "read replication channel info")
+ }
+ result = append(result, r)
+ }
+ return result, nil
+}
+
+func (p *Database) StopReplication(name string) error {
+ _, err := p.db.Exec("STOP REPLICA FOR CHANNEL ?", name)
+ return errors.Wrap(err, "stop replication for channel "+name)
+}
+
+func (p *Database) EnableReadonly() error {
+ _, err := p.db.Exec("SET GLOBAL READ_ONLY=1")
+ return errors.Wrap(err, "set global read_only param to 1")
+}
+
+func (p *Database) DisableReadonly() error {
+ _, err := p.db.Exec("SET GLOBAL READ_ONLY=0")
+ return errors.Wrap(err, "set global read_only param to 0")
+}
+
+func (p *Database) IsReadonly() (bool, error) {
+ readonly := 0
+ err := p.db.QueryRow("select @@read_only").Scan(&readonly)
+ return readonly == 1, errors.Wrap(err, "select global read_only param")
+}
+
+func (p *Database) StartReplication(replicaPass string, src ReplicationChannelSource) error {
+ _, err := p.db.Exec(`
+ CHANGE REPLICATION SOURCE TO
+ master_user='replication',
+ master_password=?,
+ master_host=?,
+ master_port=?,
+ source_connection_auto_failover=1,
+ master_auto_position=1,
+ master_retry_count=3,
+ master_connect_retry=60
+ FOR CHANNEL ?
+`, replicaPass, src.Host, src.Port, src.Name)
+ if err != nil {
+ return errors.Wrap(err, "change source for channel "+src.Name)
+ }
+
+ _, err = p.db.Exec(`START REPLICA FOR CHANNEL ?`, src.Name)
+ return errors.Wrap(err, "start replica for source "+src.Name)
+
+}
+
+func (p *Database) DeleteReplicationSource(name, host string, port int) error {
+ _, err := p.db.Exec("SELECT asynchronous_connection_failover_delete_source(?, ?, ?, null)", name, host, port)
+ return errors.Wrap(err, "delete replication source "+name)
+}
+
func (p *Database) Status(host, ip string) ([]string, error) {
- rows, err := p.db.Query("select status from mysql_servers where hostname like ? or hostname = ?;", host+"%", ip)
+ rows, err := p.db.Query("SELECT status FROM mysql_servers WHERE hostname LIKE ? OR hostname = ?;", host+"%", ip)
if err != nil {
if errors.Is(err, sql.ErrNoRows) {
return nil, ErrNotFound

K8SPXC-308: implement pxc replication (#941)

https://jira.percona.com/browse/K8SPXC-308