add service per pod functionality
MODIFY
deploy/cr.yaml
deploy/cr.yaml
@@ -36,6 +36,18 @@ spec:
size: 3
image: percona/percona-xtradb-cluster:8.0.22-13.1
autoRecovery: true
+# expose:
+# enabled: true
+# type: LoadBalancer
+# trafficPolicy: Local
+# loadBalancerSourceRanges:
+# - 10.0.0.0/8
+# annotations:
+# networking.gke.io/load-balancer-type: "Internal"
+# replicationChannels:
+# - name: pxc1_to_pxc2
+# isSource: true
+# secretName: my-cluster-name-replication
# schedulerName: mycustom-scheduler
# readinessDelaySec: 15
# livenessDelaySec: 600

MODIFY
pkg/apis/pxc/v1/pxc_types.go
pkg/apis/pxc/v1/pxc_types.go
@@ -46,9 +46,25 @@ type PerconaXtraDBClusterSpec struct {
type PXCSpec struct {
AutoRecovery *bool `json:"autoRecovery,omitempty"`
+ ReplicationChannels []ReplicationChannel `json:"replicationChannels,omitempty"`
+ Expose ServiceExpose `json:"expose,omitempty"`
*PodSpec
}
+type ServiceExpose struct {
+ Enabled bool `json:"enabled,omitempty"`
+ Type corev1.ServiceType `json:"type,omitempty"`
+ LoadBalancerSourceRanges []string `json:"loadBalancerSourceRanges,omitempty"`
+ Annotations map[string]string `json:"annotations,omitempty"`
+ TrafficPolicy corev1.ServiceExternalTrafficPolicyType `json:"trafficPolicy,omitempty"`
+}
+
+type ReplicationChannel struct {
+ Name string `json:"name,omitempty"`
+ IsSource bool `json:"isSource,omitempty"`
+ SecretName string `json:"secretName,omitempty"`
+}
+
type TLSSpec struct {
SANs []string `json:"SANs,omitempty"`
IssuerConf *cmmeta.ObjectReference `json:"issuerConf,omitempty"`
@@ -173,6 +189,15 @@ 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 v.SecretName == "" {
+ return errors.New("pxc.replicationChannels.SecretName can't be empty")
+ }
+ }
+
if c.PMM != nil && c.PMM.Enabled {
if c.PMM.Image == "" {
return errors.New("pmm.Image can't be empty")

MODIFY
pkg/controller/pxc/controller.go
pkg/controller/pxc/controller.go
@@ -353,6 +353,18 @@ func (r *ReconcilePerconaXtraDBCluster) Reconcile(request reconcile.Request) (re
}
}
+ if o.Spec.PXC.Expose.Enabled {
+ err = r.ensurePxcPodServices(o)
+ if err != nil {
+ return rr, errors.Wrap(err, "create replication services")
+ }
+ } else {
+ err = r.removePxcPodServices(o)
+ if err != nil {
+ return rr, errors.Wrap(err, "remove pxc pod services")
+ }
+ }
+
if o.Spec.HAProxy != nil && o.Spec.HAProxy.Enabled {
err = r.updatePod(statefulset.NewHAProxy(o), o.Spec.HAProxy, o, nil)
if err != nil {
@@ -1234,6 +1246,9 @@ func (r *ReconcilePerconaXtraDBCluster) createOrUpdate(obj runtime.Object) error
switch object := obj.(type) {
case *corev1.Service:
object.Spec.ClusterIP = oldObject.(*corev1.Service).Spec.ClusterIP
+ if object.Spec.Type == corev1.ServiceTypeLoadBalancer {
+ object.Spec.HealthCheckNodePort = oldObject.(*corev1.Service).Spec.HealthCheckNodePort
+ }
}
return r.client.Update(context.TODO(), obj)

ADD
pkg/controller/pxc/replication.go
+package pxc
+
+import (
+ "context"
+ "fmt"
+
+ api "github.com/percona/percona-xtradb-cluster-operator/pkg/apis/pxc/v1"
+ "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"
+ "sigs.k8s.io/controller-runtime/pkg/client"
+)
+
+func (r *ReconcilePerconaXtraDBCluster) ensurePxcPodServices(cr *api.PerconaXtraDBCluster) error {
+ if cr.Spec.Pause {
+ return nil
+ }
+
+ isBackupRunning, err := r.isBackupRunning(cr)
+ if err != nil {
+ return errors.Wrap(err, "failed to check if backup is running")
+ }
+
+ if isBackupRunning {
+ return nil
+ }
+
+ isRestoreRunning, err := r.isRestoreRunning(cr.Name, cr.Namespace)
+ if err != nil {
+ return errors.Wrap(err, "failed to check if restore is running")
+ }
+
+ if isRestoreRunning {
+ return nil
+ }
+
+ for i := 0; i < int(cr.Spec.PXC.Size); i++ {
+ svcName := fmt.Sprintf("%s-pxc-%d", cr.Name, i)
+ svc := NewExposedPXCService(svcName, cr)
+
+ err := setControllerReference(cr, svc, r.scheme)
+ if err != nil {
+ return errors.Wrap(err, "failed to set owner to external service")
+ }
+
+ err = r.createOrUpdate(svc)
+ if err != nil {
+ return errors.Wrap(err, "failed to ensure pxc service")
+ }
+ }
+ return r.removeOutdatedServices(cr)
+}
+
+func (r *ReconcilePerconaXtraDBCluster) removeOutdatedServices(cr *api.PerconaXtraDBCluster) error {
+ //needed for labels
+ svc := NewExposedPXCService("", cr)
+
+ svcNames := make(map[string]struct{}, cr.Spec.PXC.Size)
+ for i := 0; i < int(cr.Spec.PXC.Size); i++ {
+ svcNames[fmt.Sprintf("%s-pxc-%d", cr.Name, i)] = struct{}{}
+ }
+
+ svcList := &corev1.ServiceList{}
+ err := r.client.List(context.TODO(),
+ svcList,
+ &client.ListOptions{
+ Namespace: cr.Namespace,
+ LabelSelector: labels.SelectorFromSet(svc.Labels),
+ },
+ )
+
+ if err != nil {
+ return errors.Wrap(err, "failed to list external services")
+ }
+
+ for _, service := range svcList.Items {
+ if _, ok := svcNames[service.Name]; !ok {
+ err = r.client.Delete(context.TODO(), &service)
+ if err != nil {
+ return errors.Wrapf(err, "failed to delete service %s", service.Name)
+ }
+ }
+ }
+ return nil
+}
+
+func (r *ReconcilePerconaXtraDBCluster) removePxcPodServices(cr *api.PerconaXtraDBCluster) error {
+ if cr.Spec.Pause {
+ return nil
+ }
+
+ //needed for labels
+ svc := NewExposedPXCService("", cr)
+
+ svcList := &corev1.ServiceList{}
+ err := r.client.List(context.TODO(),
+ svcList,
+ &client.ListOptions{
+ Namespace: cr.Namespace,
+ LabelSelector: labels.SelectorFromSet(svc.Labels),
+ },
+ )
+ if k8serrors.IsNotFound(err) {
+ return nil
+ }
+
+ if err != nil {
+ return errors.Wrap(err, "failed to list external services")
+ }
+
+ for _, service := range svcList.Items {
+ err = r.client.Delete(context.TODO(), &service)
+ if err != nil {
+ return errors.Wrap(err, "failed to delete external service")
+ }
+ }
+ return nil
+}
+
+func NewExposedPXCService(svcName string, cr *api.PerconaXtraDBCluster) *corev1.Service {
+ svc := &corev1.Service{
+ TypeMeta: metav1.TypeMeta{
+ APIVersion: "v1",
+ Kind: "Service",
+ },
+ ObjectMeta: metav1.ObjectMeta{
+ Name: svcName,
+ Namespace: cr.Namespace,
+ Labels: map[string]string{
+ "app.kubernetes.io/name": "percona-xtradb-cluster",
+ "app.kubernetes.io/instance": cr.Name,
+ "app.kubernetes.io/component": "external-service",
+ },
+ Annotations: cr.Spec.PXC.Expose.Annotations,
+ },
+ Spec: corev1.ServiceSpec{
+ Ports: []corev1.ServicePort{
+ {
+ Port: 3306,
+ Name: "mysql",
+ },
+ },
+ LoadBalancerSourceRanges: cr.Spec.PXC.Expose.LoadBalancerSourceRanges,
+ Selector: map[string]string{
+ "statefulset.kubernetes.io/pod-name": svcName,
+ },
+ },
+ }
+
+ switch cr.Spec.PXC.Expose.Type {
+ case corev1.ServiceTypeNodePort:
+ svc.Spec.Type = corev1.ServiceTypeNodePort
+ svc.Spec.ExternalTrafficPolicy = "Local"
+ case corev1.ServiceTypeLoadBalancer:
+ svc.Spec.Type = corev1.ServiceTypeLoadBalancer
+ svc.Spec.ExternalTrafficPolicy = "Cluster"
+ default:
+ svc.Spec.Type = corev1.ServiceTypeClusterIP
+ }
+
+ return svc
+}

MODIFY
pkg/controller/pxc/upgrade.go
pkg/controller/pxc/upgrade.go
@@ -536,6 +536,10 @@ func (r *ReconcilePerconaXtraDBCluster) isBackupRunning(cr *api.PerconaXtraDBClu
}
for _, bcp := range bcpList.Items {
+ if bcp.Spec.PXCCluster != cr.Name {
+ continue
+ }
+
if bcp.Status.State == api.BackupRunning || bcp.Status.State == api.BackupStarting {
return true, nil
}
@@ -544,6 +548,30 @@ func (r *ReconcilePerconaXtraDBCluster) isBackupRunning(cr *api.PerconaXtraDBClu
return false, nil
}
+func (r *ReconcilePerconaXtraDBCluster) isRestoreRunning(clusterName, namespace string) (bool, error) {
+ restoreList := api.PerconaXtraDBClusterRestoreList{}
+
+ err := r.client.List(context.TODO(), &restoreList, &client.ListOptions{
+ Namespace: namespace,
+ })
+ if err != nil {
+ return false, errors.Wrap(err, "failed to get restore list")
+ }
+
+ for _, v := range restoreList.Items {
+ if v.Spec.PXCCluster != clusterName {
+ continue
+ }
+
+ switch v.Status.State {
+ case api.RestoreStarting, api.RestoreStopCluster, api.RestoreRestore,
+ api.RestoreStartCluster, api.RestorePITR:
+ return true, nil
+ }
+ }
+ return false, nil
+}
+
func (r *ReconcilePerconaXtraDBCluster) getConfigHash(cr *api.PerconaXtraDBCluster, sfs api.StatefulApp) string {
configString := cr.Spec.PXC.Configuration
if sfs.Labels()["app.kubernetes.io/component"] == "haproxy" {

K8SPXC-308: add service per pod functionality (#891)