allow to configure and change replication parameters
Allow tuning sourceConnectRetry and sourceRetryCount parameters
at live cluster
MODIFY
pkg/apis/pxc/v1/pxc_types.go
pkg/apis/pxc/v1/pxc_types.go
@@ -63,6 +63,12 @@ type ReplicationChannel struct {
Name string `json:"name,omitempty"`
IsSource bool `json:"isSource,omitempty"`
SourcesList []ReplicationSource `json:"sourcesList,omitempty"`
+ Config *ReplicationChannelConfig `json:"configuration,omitempty"`
+}
+
+type ReplicationChannelConfig struct {
+ SourceRetryCount uint `json:"sourceRetryCount,omitempty"`
+ SourceConnectRetry uint `json:"sourceConnectRetry,omitempty"`
}
type ReplicationSource struct {
@@ -124,6 +130,7 @@ const (
// PerconaXtraDBClusterStatus defines the observed state of PerconaXtraDBCluster
type PerconaXtraDBClusterStatus struct {
PXC AppStatus `json:"pxc,omitempty"`
+ PXCReplication *ReplicationStatus `json:"pxcReplication,omitempty"`
ProxySQL AppStatus `json:"proxysql,omitempty"`
HAProxy AppStatus `json:"haproxy,omitempty"`
Backup ComponentStatus `json:"backup,omitempty"`
@@ -138,6 +145,16 @@ type PerconaXtraDBClusterStatus struct {
Ready int32 `json:"ready"`
}
+// TODO: add replication status(error,active and etc)
+type ReplicationStatus struct {
+ Channels []ReplicationChannelStatus `json:"replicationChannels,omitempty"`
+}
+
+type ReplicationChannelStatus struct {
+ Name string `json:"name,omitempty"`
+ ReplicationChannelConfig
+}
+
type ConditionStatus string
const (
@@ -589,6 +606,12 @@ func (cr *PerconaXtraDBCluster) CheckNSetDefaults(serverVersion *version.ServerV
c.PXC.ReplicationChannels[chIdx].SourcesList[srcIdx].Port = 3306
}
}
+ if !channel.IsSource && channel.Config == nil {
+ c.PXC.ReplicationChannels[chIdx].Config = &ReplicationChannelConfig{
+ SourceRetryCount: 3,
+ SourceConnectRetry: 60,
+ }
+ }
}
setSafeDefaults(c, logger)

MODIFY
pkg/controller/pxc/replication.go
pkg/controller/pxc/replication.go
@@ -234,17 +234,21 @@ func (r *ReconcilePerconaXtraDBCluster) reconcileReplication(cr *api.PerconaXtra
}
}
- for _, channels := range cr.Spec.PXC.ReplicationChannels {
- if channels.IsSource {
+ for _, channel := range cr.Spec.PXC.ReplicationChannels {
+ if channel.IsSource {
continue
}
- err = manageReplicationChannel(r.log, primaryDB, channels, string(sysUsersSecretObj.Data["replication"]))
+
+ currConf := currentReplicaConfig(channel.Name, cr.Status.PXCReplication)
+
+ err = manageReplicationChannel(r.log, primaryDB, channel, currConf, string(sysUsersSecretObj.Data["replication"]))
if err != nil {
- return errors.Wrapf(err, "manage replication channel %s", channels.Name)
+ return errors.Wrapf(err, "manage replication channel %s", channel.Name)
}
+ setReplicationChannelStatus(cr, channel)
}
- return nil
+ return r.updateStatus(cr, nil)
}
func handleReplicaPasswordChange(db queries.Database, newPass string) error {
@@ -343,7 +347,7 @@ func removeOutdatedChannels(db queries.Database, currentChannels []api.Replicati
return nil
}
-func manageReplicationChannel(log logr.Logger, primaryDB queries.Database, channel api.ReplicationChannel, replicaPW string) error {
+func manageReplicationChannel(log logr.Logger, primaryDB queries.Database, channel api.ReplicationChannel, currConf api.ReplicationChannelConfig, 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)
@@ -360,7 +364,8 @@ func manageReplicationChannel(log logr.Logger, primaryDB queries.Database, chann
return nil
}
- if replicationStatus == queries.ReplicationStatusActive {
+ if replicationStatus == queries.ReplicationStatusActive &&
+ *channel.Config == currConf {
return nil
}
}
@@ -392,10 +397,14 @@ func manageReplicationChannel(log logr.Logger, primaryDB queries.Database, chann
}
}
- return primaryDB.StartReplication(replicaPW, queries.ReplicationChannelSource{
+ return primaryDB.StartReplication(replicaPW, queries.ReplicationConfig{
+ Source: queries.ReplicationChannelSource{
Name: channel.Name,
Host: maxWeightSrc.Host,
Port: maxWeightSrc.Port,
+ },
+ SourceRetryCount: channel.Config.SourceRetryCount,
+ SourceConnectRetry: channel.Config.SourceConnectRetry,
})
}
@@ -525,3 +534,40 @@ func isPodReady(pod corev1.Pod) bool {
}
return false
}
+
+func currentReplicaConfig(name string, status *api.ReplicationStatus) api.ReplicationChannelConfig {
+ res := api.ReplicationChannelConfig{}
+ if status == nil {
+ return res
+ }
+
+ for _, v := range status.Channels {
+ if v.Name == name {
+ return v.ReplicationChannelConfig
+ }
+ }
+ return res
+}
+
+func setReplicationChannelStatus(cr *api.PerconaXtraDBCluster, channel api.ReplicationChannel) {
+ status := api.ReplicationChannelStatus{
+ Name: channel.Name,
+ ReplicationChannelConfig: *channel.Config,
+ }
+
+ if cr.Status.PXCReplication == nil {
+ cr.Status.PXCReplication = &api.ReplicationStatus{
+ Channels: []api.ReplicationChannelStatus{status},
+ }
+ return
+ }
+
+ for k, v := range cr.Status.PXCReplication.Channels {
+ if channel.Name == v.Name {
+ cr.Status.PXCReplication.Channels[k] = status
+ return
+ }
+ }
+
+ cr.Status.PXCReplication.Channels = append(cr.Status.PXCReplication.Channels, status)
+}

MODIFY
pkg/pxc/queries/queries.go
pkg/pxc/queries/queries.go
@@ -30,6 +30,12 @@ type Database struct {
db *sql.DB
}
+type ReplicationConfig struct {
+ Source ReplicationChannelSource
+ SourceRetryCount uint
+ SourceConnectRetry uint
+}
+
type ReplicationChannelSource struct {
Name string
Host string
@@ -222,7 +228,7 @@ func (p *Database) IsReadonly() (bool, error) {
return readonly == 1, errors.Wrap(err, "select global read_only param")
}
-func (p *Database) StartReplication(replicaPass string, src ReplicationChannelSource) error {
+func (p *Database) StartReplication(replicaPass string, config ReplicationConfig) error {
_, err := p.db.Exec(`
CHANGE REPLICATION SOURCE TO
SOURCE_USER='replication',
@@ -231,16 +237,16 @@ func (p *Database) StartReplication(replicaPass string, src ReplicationChannelSo
SOURCE_PORT=?,
SOURCE_CONNECTION_AUTO_FAILOVER=1,
SOURCE_AUTO_POSITION=1,
- SOURCE_RETRY_COUNT=3,
- SOURCE_CONNECT_RETRY=60
+ SOURCE_RETRY_COUNT=?,
+ SOURCE_CONNECT_RETRY=?
FOR CHANNEL ?
-`, replicaPass, src.Host, src.Port, src.Name)
+`, replicaPass, config.Source.Host, config.Source.Port, config.SourceRetryCount, config.SourceConnectRetry, config.Source.Name)
if err != nil {
- return errors.Wrap(err, "change source for channel "+src.Name)
+ return errors.Wrapf(err, "change source for channel %s", config.Source.Name)
}
- _, err = p.db.Exec(`START REPLICA FOR CHANNEL ?`, src.Name)
- return errors.Wrap(err, "start replica for source "+src.Name)
+ _, err = p.db.Exec(`START REPLICA FOR CHANNEL ?`, config.Source.Name)
+ return errors.Wrapf(err, "start replica for source %s", config.Source.Name)
}

K8SPXC-789: allow to configure and change replication parameters  (#1002)

Allow tuning sourceConnectRetry and sourceRetryCount parameters
at live cluster