handle replication password change
MODIFY
clientcmd/clientcmd.go
clientcmd/clientcmd.go
@@ -24,7 +24,9 @@ func NewClient() (*Client, error) {
// Instantiate loader for kubeconfig file.
kubeconfig := clientcmd.NewNonInteractiveDeferredLoadingClientConfig(
clientcmd.NewDefaultClientConfigLoadingRules(),
- &clientcmd.ConfigOverrides{},
+ &clientcmd.ConfigOverrides{
+ Timeout: "10s",
+ },
)
// Get a rest.Config from the kubeconfig file. This will be passed into all

MODIFY
pkg/controller/pxc/controller.go
pkg/controller/pxc/controller.go
@@ -274,12 +274,17 @@ func (r *ReconcilePerconaXtraDBCluster) Reconcile(request reconcile.Request) (re
if err != nil {
return reconcile.Result{}, errors.Wrap(err, "reconcile users secret")
}
- var pxcAnnotations, proxysqlAnnotations map[string]string
+
+ userReconcileResult := &ReconcileUsersResult{}
+
if o.CompareVersionWith("1.5.0") >= 0 {
- pxcAnnotations, proxysqlAnnotations, err = r.reconcileUsers(o)
+ urr, err := r.reconcileUsers(o)
if err != nil {
return rr, errors.Wrap(err, "reconcile users")
}
+ if urr != nil {
+ userReconcileResult = urr
+ }
}
r.resyncPXCUsersWithProxySQL(o)
@@ -335,7 +340,7 @@ func (r *ReconcilePerconaXtraDBCluster) Reconcile(request reconcile.Request) (re
}
pxcSet := statefulset.NewNode(o)
- pxc.MergeTemplateAnnotations(pxcSet.StatefulSet(), pxcAnnotations)
+ pxc.MergeTemplateAnnotations(pxcSet.StatefulSet(), userReconcileResult.pxcAnnotations)
err = r.updatePod(pxcSet, o.Spec.PXC.PodSpec, o, inits)
if err != nil {
return reconcile.Result{}, errors.Wrap(err, "pxc upgrade error")
@@ -428,7 +433,7 @@ func (r *ReconcilePerconaXtraDBCluster) Reconcile(request reconcile.Request) (re
}
proxysqlSet := statefulset.NewProxy(o)
- pxc.MergeTemplateAnnotations(proxysqlSet.StatefulSet(), proxysqlAnnotations)
+ pxc.MergeTemplateAnnotations(proxysqlSet.StatefulSet(), userReconcileResult.proxysqlAnnotations)
if o.Spec.ProxySQL != nil && o.Spec.ProxySQL.Enabled {
err = r.updatePod(proxysqlSet, o.Spec.ProxySQL, o, nil)
@@ -511,7 +516,7 @@ func (r *ReconcilePerconaXtraDBCluster) Reconcile(request reconcile.Request) (re
}
if o.CompareVersionWith("1.9.0") >= 0 {
- err = r.reconcileReplication(o)
+ err = r.reconcileReplication(o, userReconcileResult.updateReplicationPassword)
if err != nil {
reqLogger.Info("reconcile replication error", "err", err.Error())
}

MODIFY
pkg/controller/pxc/replication.go
pkg/controller/pxc/replication.go
@@ -96,7 +96,7 @@ func (r *ReconcilePerconaXtraDBCluster) removeOutdatedServices(cr *api.PerconaXt
return nil
}
-func (r *ReconcilePerconaXtraDBCluster) reconcileReplication(cr *api.PerconaXtraDBCluster) error {
+func (r *ReconcilePerconaXtraDBCluster) reconcileReplication(cr *api.PerconaXtraDBCluster, replicaPassUpdated bool) error {
if cr.Status.PXC.Ready < 1 || cr.Spec.Pause {
return nil
}
@@ -227,6 +227,13 @@ func (r *ReconcilePerconaXtraDBCluster) reconcileReplication(cr *api.PerconaXtra
return errors.Wrap(err, "get secrets")
}
+ if replicaPassUpdated {
+ err = handleReplicaPasswordChange(primaryDB, string(sysUsersSecretObj.Data["replication"]))
+ if err != nil {
+ return errors.Wrap(err, "failed to change replication password")
+ }
+ }
+
for _, channels := range cr.Spec.PXC.ReplicationChannels {
if channels.IsSource {
continue
@@ -240,6 +247,21 @@ func (r *ReconcilePerconaXtraDBCluster) reconcileReplication(cr *api.PerconaXtra
return nil
}
+func handleReplicaPasswordChange(db queries.Database, newPass string) error {
+ channels, err := db.CurrentReplicationChannels()
+ if err != nil {
+ return errors.Wrap(err, "get current replication channels")
+ }
+
+ for _, channel := range channels {
+ err := db.ChangeChannelPassword(channel, newPass)
+ if err != nil {
+ return errors.Wrapf(err, "change password for channel %s", channel)
+ }
+ }
+ return nil
+}
+
func checkReadonlyStatus(channels []api.ReplicationChannel, pods []corev1.Pod, cr *api.PerconaXtraDBCluster, client client.Client) error {
isReplica := false
if len(channels) > 0 {

MODIFY
pkg/controller/pxc/status_test.go
pkg/controller/pxc/status_test.go
@@ -319,57 +319,57 @@ func TestClusterStatus(t *testing.T) {
want: api.AppStateInit,
},
"PXC init": {
- status: api.PerconaXtraDBClusterStatus{PXC: api.AppStatus{Status: api.AppStateInit}},
+ status: api.PerconaXtraDBClusterStatus{PXC: api.AppStatus{ComponentStatus: api.ComponentStatus{Status: api.AppStateInit}}},
want: api.AppStateInit,
},
"PXC ready": {
- status: api.PerconaXtraDBClusterStatus{PXC: api.AppStatus{Status: api.AppStateReady}},
+ status: api.PerconaXtraDBClusterStatus{PXC: api.AppStatus{ComponentStatus: api.ComponentStatus{Status: api.AppStateReady}}},
want: api.AppStateReady,
},
"PXC stopping": {
- status: api.PerconaXtraDBClusterStatus{PXC: api.AppStatus{Status: api.AppStateStopping}},
+ status: api.PerconaXtraDBClusterStatus{PXC: api.AppStatus{ComponentStatus: api.ComponentStatus{Status: api.AppStateStopping}}},
want: api.AppStateStopping,
},
"PXC paused, HAProxy stopping": {
status: api.PerconaXtraDBClusterStatus{
- PXC: api.AppStatus{Status: api.AppStatePaused},
- HAProxy: api.AppStatus{Status: api.AppStateStopping},
+ PXC: api.AppStatus{ComponentStatus: api.ComponentStatus{Status: api.AppStatePaused}},
+ HAProxy: api.AppStatus{ComponentStatus: api.ComponentStatus{Status: api.AppStateStopping}},
},
want: api.AppStateStopping,
},
"PXC paused, ProxySQL stopping": {
status: api.PerconaXtraDBClusterStatus{
- PXC: api.AppStatus{Status: api.AppStatePaused},
- ProxySQL: api.AppStatus{Status: api.AppStateStopping},
+ PXC: api.AppStatus{ComponentStatus: api.ComponentStatus{Status: api.AppStatePaused}},
+ ProxySQL: api.AppStatus{ComponentStatus: api.ComponentStatus{Status: api.AppStateStopping}},
},
want: api.AppStateStopping,
},
"PXC paused, HAProxy paused": {
status: api.PerconaXtraDBClusterStatus{
- PXC: api.AppStatus{Status: api.AppStatePaused},
- HAProxy: api.AppStatus{Status: api.AppStatePaused},
+ PXC: api.AppStatus{ComponentStatus: api.ComponentStatus{Status: api.AppStatePaused}},
+ HAProxy: api.AppStatus{ComponentStatus: api.ComponentStatus{Status: api.AppStatePaused}},
},
want: api.AppStatePaused,
},
"HAProxy init": {
- status: api.PerconaXtraDBClusterStatus{HAProxy: api.AppStatus{Status: api.AppStateInit}},
+ status: api.PerconaXtraDBClusterStatus{HAProxy: api.AppStatus{ComponentStatus: api.ComponentStatus{Status: api.AppStateInit}}},
want: api.AppStateInit,
},
"HAProxy ready": {
status: api.PerconaXtraDBClusterStatus{
- PXC: api.AppStatus{Status: api.AppStateReady},
- HAProxy: api.AppStatus{Status: api.AppStateReady},
+ PXC: api.AppStatus{ComponentStatus: api.ComponentStatus{Status: api.AppStateReady}},
+ HAProxy: api.AppStatus{ComponentStatus: api.ComponentStatus{Status: api.AppStateReady}},
},
want: api.AppStateReady,
},
"ProxySQL init": {
- status: api.PerconaXtraDBClusterStatus{ProxySQL: api.AppStatus{Status: api.AppStateInit}},
+ status: api.PerconaXtraDBClusterStatus{ProxySQL: api.AppStatus{ComponentStatus: api.ComponentStatus{Status: api.AppStateInit}}},
want: api.AppStateInit,
},
"ProxySQL ready": {
status: api.PerconaXtraDBClusterStatus{
- PXC: api.AppStatus{Status: api.AppStateReady},
- ProxySQL: api.AppStatus{Status: api.AppStateReady},
+ PXC: api.AppStatus{ComponentStatus: api.ComponentStatus{Status: api.AppStateReady}},
+ ProxySQL: api.AppStatus{ComponentStatus: api.ComponentStatus{Status: api.AppStateReady}},
},
want: api.AppStateReady,
},
@@ -377,7 +377,7 @@ func TestClusterStatus(t *testing.T) {
for name, test := range tests {
t.Run(name, func(tt *testing.T) {
- got := test.status.ClusterStatus(false)
+ got := test.status.ClusterStatus(false, false)
if got != test.want {
t.Errorf("AppState got %#v, want %#v", got, test.want)

MODIFY
pkg/controller/pxc/users.go
pkg/controller/pxc/users.go
@@ -19,9 +19,21 @@ import (
const internalPrefix = "internal-"
-func (r *ReconcilePerconaXtraDBCluster) reconcileUsers(cr *api.PerconaXtraDBCluster) (pxcAnnotations, proxysqlAnnotations map[string]string, err error) {
+type userUpdateRestart struct {
+ restartPXC bool
+ restartProxy bool
+ updateReplicationPass bool
+}
+
+type ReconcileUsersResult struct {
+ pxcAnnotations map[string]string
+ proxysqlAnnotations map[string]string
+ updateReplicationPassword bool
+}
+
+func (r *ReconcilePerconaXtraDBCluster) reconcileUsers(cr *api.PerconaXtraDBCluster) (*ReconcileUsersResult, error) {
sysUsersSecretObj := corev1.Secret{}
- err = r.client.Get(context.TODO(),
+ err := r.client.Get(context.TODO(),
types.NamespacedName{
Namespace: cr.Namespace,
Name: cr.Spec.SecretsName,
@@ -29,9 +41,9 @@ func (r *ReconcilePerconaXtraDBCluster) reconcileUsers(cr *api.PerconaXtraDBClus
&sysUsersSecretObj,
)
if err != nil && k8serrors.IsNotFound(err) {
- return nil, nil, nil
+ return nil, nil
} else if err != nil {
- return nil, nil, errors.Wrapf(err, "get sys users secret '%s'", cr.Spec.SecretsName)
+ return nil, errors.Wrapf(err, "get sys users secret '%s'", cr.Spec.SecretsName)
}
secretName := internalPrefix + cr.Name
@@ -46,7 +58,7 @@ func (r *ReconcilePerconaXtraDBCluster) reconcileUsers(cr *api.PerconaXtraDBClus
&internalSysSecretObj,
)
if err != nil && !k8serrors.IsNotFound(err) {
- return nil, nil, errors.Wrap(err, "get internal sys users secret")
+ return nil, errors.Wrap(err, "get internal sys users secret")
}
if k8serrors.IsNotFound(err) {
@@ -57,78 +69,80 @@ func (r *ReconcilePerconaXtraDBCluster) reconcileUsers(cr *api.PerconaXtraDBClus
}
err = r.client.Create(context.TODO(), internalSysUsersSecret)
if err != nil {
- return nil, nil, errors.Wrap(err, "create internal sys users secret")
+ return nil, errors.Wrap(err, "create internal sys users secret")
}
- return nil, nil, nil
+ return nil, nil
}
if cr.Status.PXC.Ready > 0 {
err := r.manageOperatorAdminUser(cr, &sysUsersSecretObj, &internalSysSecretObj)
if err != nil {
- return nil, nil, errors.Wrap(err, "manage operator admin user")
+ return nil, errors.Wrap(err, "manage operator admin user")
}
if cr.CompareVersionWith("1.6.0") >= 0 {
// monitor user need more grants for work in version more then 1.6.0
err = r.manageMonitorUser(cr, &internalSysSecretObj)
if err != nil {
- return nil, nil, errors.Wrap(err, "manage monitor user")
+ return nil, errors.Wrap(err, "manage monitor user")
}
}
if cr.CompareVersionWith("1.7.0") >= 0 {
// xtrabackup user need more grants for work in version more then 1.7.0
err = r.manageXtrabackupUser(cr, &internalSysSecretObj)
if err != nil {
- return nil, nil, errors.Wrap(err, "manage xtrabackup user")
+ return nil, errors.Wrap(err, "manage xtrabackup user")
}
}
if cr.CompareVersionWith("1.9.0") >= 0 {
err = r.manageReplicationUser(cr, &sysUsersSecretObj, &internalSysSecretObj)
if err != nil {
- return nil, nil, errors.Wrap(err, "manage replication user")
+ return nil, errors.Wrap(err, "manage replication user")
}
}
}
if cr.Status.Status != api.AppStateReady {
- return nil, nil, nil
+ return nil, nil
}
newSysData, err := json.Marshal(sysUsersSecretObj.Data)
if err != nil {
- return nil, nil, errors.Wrap(err, "marshal sys secret data")
+ return nil, errors.Wrap(err, "marshal sys secret data")
}
newSecretDataHash := sha256Hash(newSysData)
dataChanged, err := sysUsersSecretDataChanged(newSecretDataHash, &internalSysSecretObj)
if err != nil {
- return nil, nil, errors.Wrap(err, "check sys users data changes")
+ return nil, errors.Wrap(err, "check sys users data changes")
}
if !dataChanged {
- return nil, nil, nil
+ return nil, nil
}
- restartPXC, restartProxy, err := r.manageSysUsers(cr, &sysUsersSecretObj, &internalSysSecretObj)
+ restarts, err := r.manageSysUsers(cr, &sysUsersSecretObj, &internalSysSecretObj)
if err != nil {
- return nil, nil, errors.Wrap(err, "manage sys users")
+ return nil, errors.Wrap(err, "manage sys users")
}
internalSysSecretObj.Data = sysUsersSecretObj.Data
err = r.client.Update(context.TODO(), &internalSysSecretObj)
if err != nil {
- return nil, nil, errors.Wrap(err, "update internal sys users secret")
+ return nil, errors.Wrap(err, "update internal sys users secret")
}
- if restartProxy {
- proxysqlAnnotations = make(map[string]string)
- proxysqlAnnotations["last-applied-secret"] = newSecretDataHash
+ result := &ReconcileUsersResult{
+ updateReplicationPassword: restarts.updateReplicationPass,
}
- if restartPXC {
- pxcAnnotations = make(map[string]string)
- pxcAnnotations["last-applied-secret"] = newSecretDataHash
+
+ if restarts.restartProxy {
+ result.proxysqlAnnotations = map[string]string{"last-applied-secret": newSecretDataHash}
+ }
+ if restarts.restartPXC {
+ result.pxcAnnotations = map[string]string{"last-applied-secret": newSecretDataHash}
}
- return pxcAnnotations, proxysqlAnnotations, nil
+ return result, nil
}
func (r *ReconcilePerconaXtraDBCluster) manageMonitorUser(cr *api.PerconaXtraDBCluster, internalSysSecretObj *corev1.Secret) error {
@@ -223,7 +237,7 @@ func (r *ReconcilePerconaXtraDBCluster) manageXtrabackupUser(cr *api.PerconaXtra
return nil
}
-func (r *ReconcilePerconaXtraDBCluster) manageSysUsers(cr *api.PerconaXtraDBCluster, sysUsersSecretObj, internalSysSecretObj *corev1.Secret) (bool, bool, error) {
+func (r *ReconcilePerconaXtraDBCluster) manageSysUsers(cr *api.PerconaXtraDBCluster, sysUsersSecretObj, internalSysSecretObj *corev1.Secret) (*userUpdateRestart, error) {
type action int
const (
@@ -232,6 +246,7 @@ func (r *ReconcilePerconaXtraDBCluster) manageSysUsers(cr *api.PerconaXtraDBClus
rProxy
rProxyifPMM
syncProxyUsers
+ syncReplicaUser
)
type user struct {
@@ -277,6 +292,7 @@ func (r *ReconcilePerconaXtraDBCluster) manageSysUsers(cr *api.PerconaXtraDBClus
requiredUsers = append(requiredUsers, user{
name: "replication",
hosts: []string{"%"},
+ action: syncReplicaUser,
})
}
@@ -294,11 +310,14 @@ func (r *ReconcilePerconaXtraDBCluster) manageSysUsers(cr *api.PerconaXtraDBClus
})
}
- var sysUsers, proxyUsers []users.SysUser
- var todo action
+ var (
+ sysUsers, proxyUsers []users.SysUser
+ todo action
+ )
+
for _, user := range requiredUsers {
if len(sysUsersSecretObj.Data[user.name]) == 0 {
- return false, false, errors.New("undefined or not exist user " + user.name)
+ return nil, errors.New("undefined or not exist user " + user.name)
}
if bytes.Equal(sysUsersSecretObj.Data[user.name], internalSysSecretObj.Data[user.name]) {
@@ -327,8 +346,11 @@ func (r *ReconcilePerconaXtraDBCluster) manageSysUsers(cr *api.PerconaXtraDBClus
todo &^= rPXCifPMM | rProxyifPMM
}
- restartPXC := todo&(rPXC|rPXCifPMM) != 0
- restartProxy := todo&(rProxy|rProxyifPMM) != 0
+ res := &userUpdateRestart{
+ restartPXC: todo&(rPXC|rPXCifPMM) != 0,
+ restartProxy: todo&(rProxy|rProxyifPMM) != 0,
+ updateReplicationPass: todo&syncReplicaUser != 0,
+ }
pxcUser := "root"
pxcPass := string(internalSysSecretObj.Data["root"])
@@ -343,28 +365,28 @@ func (r *ReconcilePerconaXtraDBCluster) manageSysUsers(cr *api.PerconaXtraDBClus
}
um, err := users.NewManager(addr, pxcUser, pxcPass)
if err != nil {
- return false, false, errors.Wrap(err, "new users manager")
+ return res, errors.Wrap(err, "new users manager")
}
defer um.Close()
err = um.UpdateUsersPass(sysUsers)
if err != nil {
- return false, false, errors.Wrap(err, "update sys users pass")
+ return res, errors.Wrap(err, "update sys users pass")
}
if cr.Spec.ProxySQL != nil && cr.Spec.ProxySQL.Enabled {
err = r.updateProxyUsers(proxyUsers, internalSysSecretObj, cr)
if err != nil {
- return false, false, errors.Wrap(err, "update Proxy users pass")
+ return res, errors.Wrap(err, "update Proxy users pass")
}
}
- if todo&syncProxyUsers != 0 && !restartProxy {
+ if todo&syncProxyUsers != 0 && !res.restartProxy {
err = r.syncPXCUsersWithProxySQL(cr)
if err != nil {
- return false, false, errors.Wrap(err, "sync users")
+ return res, errors.Wrap(err, "sync users")
}
}
- return restartPXC, restartProxy, nil
+ return res, nil
}
func (r *ReconcilePerconaXtraDBCluster) syncPXCUsersWithProxySQL(cr *api.PerconaXtraDBCluster) error {

MODIFY
pkg/pxc/queries/queries.go
pkg/pxc/queries/queries.go
@@ -92,6 +92,38 @@ func (p *Database) CurrentReplicationChannels() ([]string, error) {
return result, nil
}
+func (p *Database) ChangeChannelPassword(channel, password string) error {
+ tx, err := p.db.Begin()
+ if err != nil {
+ return errors.Wrap(err, "start transaction for updating channel password")
+ }
+ _, err = tx.Exec(`STOP REPLICA IO_THREAD FOR CHANNEL ?`, channel)
+ if err != nil {
+ errT := tx.Rollback()
+ if errT != nil {
+ return errors.Wrapf(err, "rollback STOP REPLICA IO_THREAD FOR CHANNEL %s", channel)
+ }
+ return errors.Wrapf(err, "stop replication IO thread for channel %s", channel)
+ }
+ _, err = tx.Exec(`CHANGE REPLICATION SOURCE TO SOURCE_PASSWORD=? FOR CHANNEL ?`, password, channel)
+ if err != nil {
+ errT := tx.Rollback()
+ if errT != nil {
+ return errors.Wrapf(err, "rollback CHANGE SOURCE_PASSWORD FOR CHANNEL %s", channel)
+ }
+ return errors.Wrapf(err, "change master password for channel %s", channel)
+ }
+ _, err = tx.Exec(`START REPLICA IO_THREAD FOR CHANNEL ?`, channel)
+ if err != nil {
+ errT := tx.Rollback()
+ if errT != nil {
+ return errors.Wrapf(err, "rollback START REPLICA IO_THREAD FOR CHANNEL %s", channel)
+ }
+ return errors.Wrapf(err, "start io thread for channel %s", channel)
+ }
+ return tx.Commit()
+}
+
func (p *Database) ReplicationStatus(channel string) (ReplicationStatus, error) {
rows, err := p.db.Query(`SHOW REPLICA STATUS FOR CHANNEL ?`, channel)
if err != nil {
@@ -193,14 +225,14 @@ func (p *Database) IsReadonly() (bool, error) {
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
+ SOURCE_USER='replication',
+ SOURCE_PASSWORD=?,
+ SOURCE_HOST=?,
+ SOURCE_PORT=?,
+ SOURCE_CONNECTION_AUTO_FAILOVER=1,
+ SOURCE_AUTO_POSITION=1,
+ SOURCE_RETRY_COUNT=3,
+ SOURCE_CONNECT_RETRY=60
FOR CHANNEL ?
`, replicaPass, src.Host, src.Port, src.Name)
if err != nil {

K8SPXC-851: handle replication password change (#981)