Annotate logical backup jobs/pods, endpoints

Author: hughcapet

pkg/cluster/cluster.go

@@ -797,6 +797,12 @@ func (c *Cluster) compareLogicalBackupJob(cur, new *batchv1.CronJob) (match bool
 			newImage, curImage)
 	}
 
+	newPodAnnotation := new.Spec.JobTemplate.Spec.Template.Annotations
+	curPodAnnotation := cur.Spec.JobTemplate.Spec.Template.Annotations
+	if changed, reason := c.compareAnnotations(curPodAnnotation, newPodAnnotation); changed {
+		return false, fmt.Sprintf("new job's pod template metadata annotations does not match " + reason)
+	}
+
 	newPgVersion := getPgVersion(new)
 	curPgVersion := getPgVersion(cur)
 	if newPgVersion != curPgVersion {

pkg/cluster/k8sres.go

@@ -2048,9 +2048,10 @@ func (c *Cluster) getCustomServiceAnnotations(role PostgresRole, spec *acidv1.Po
 func (c *Cluster) generateEndpoint(role PostgresRole, subsets []v1.EndpointSubset) *v1.Endpoints {
 	endpoints := &v1.Endpoints{
 		ObjectMeta: metav1.ObjectMeta{
-			Name:      c.endpointName(role),
-			Namespace: c.Namespace,
-			Labels:    c.roleLabelsSet(true, role),
+			Name:        c.endpointName(role),
+			Namespace:   c.Namespace,
+			Annotations: c.annotationsSet(nil),
+			Labels:      c.roleLabelsSet(true, role),
 		},
 	}
 	if len(subsets) > 0 {

pkg/cluster/sync.go

@@ -239,7 +239,17 @@ func (c *Cluster) syncEndpoint(role PostgresRole) error {
 	c.setProcessName("syncing %s endpoint", role)
 
 	if ep, err = c.KubeClient.Endpoints(c.Namespace).Get(context.TODO(), c.endpointName(role), metav1.GetOptions{}); err == nil {
-		// TODO: No syncing of endpoints here, is this covered completely by updateService?
+		desiredEp := c.generateEndpoint(role, ep.Subsets)
+		if changed, _ := c.compareAnnotations(ep.Annotations, desiredEp.Annotations); changed {
+			patchData, err := metaAnnotationsPatch(desiredEp.Annotations)
+			if err != nil {
+				return fmt.Errorf("could not form patch for %s endpoint: %v", role, err)
+			}
+			ep, err = c.KubeClient.Endpoints(c.Namespace).Patch(context.TODO(), c.endpointName(role), types.MergePatchType, []byte(patchData), metav1.PatchOptions{})
+			if err != nil {
+				return fmt.Errorf("could not patch annotations of %s endpoint: %v", role, err)
+			}
+		}
 		c.Endpoints[role] = ep
 		return nil
 	}
@@ -1410,6 +1420,16 @@ func (c *Cluster) syncLogicalBackupJob() error {
 			}
 			c.logger.Info("the logical backup job is synced")
 		}
+		if changed, _ := c.compareAnnotations(job.Annotations, desiredJob.Annotations); changed {
+			patchData, err := metaAnnotationsPatch(desiredJob.Annotations)
+			if err != nil {
+				return fmt.Errorf("could not form patch for the logical backup job %q: %v", jobName, err)
+			}
+			_, err = c.KubeClient.CronJobs(c.Namespace).Patch(context.TODO(), jobName, types.MergePatchType, []byte(patchData), metav1.PatchOptions{})
+			if err != nil {
+				return fmt.Errorf("could not patch annotations of the logical backup job %q: %v", jobName, err)
+			}
+		}
 		return nil
 	}
 	if !k8sutil.ResourceNotFound(err) {

pkg/cluster/util_test.go

@@ -180,8 +180,21 @@ func checkResourcesInheritedAnnotations(cluster *Cluster, resultAnnotations map[
 		return nil
 	}
 
+	checkEndpoints := func(annotations map[string]string) error {
+		endpointsList, err := cluster.KubeClient.Endpoints(namespace).List(context.TODO(), clusterOptions)
+		if err != nil {
+			return err
+		}
+		for _, ep := range endpointsList.Items {
+			if err := containsAnnotations(annotations, ep.Annotations, ep.Name, "Endpoints"); err != nil {
+				return err
+			}
+		}
+		return nil
+	}
+
 	checkFuncs := []func(map[string]string) error{
-		checkSts, checkPods, checkSvc, checkPdb, checkPooler, checkPvc, checkSecrets,
+		checkSts, checkPods, checkSvc, checkPdb, checkPooler, checkPvc, checkSecrets, checkEndpoints,
 	}
 	for _, f := range checkFuncs {
 		if err := f(resultAnnotations); err != nil {
@@ -232,6 +245,7 @@ func newInheritedAnnotationsCluster(client k8sutil.KubernetesClient) (*Cluster,
 			Volume: acidv1.Volume{
 				Size: "1Gi",
 			},
+			NumberOfInstances: 2,
 		},
 	}
 
@@ -260,6 +274,8 @@ func newInheritedAnnotationsCluster(client k8sutil.KubernetesClient) (*Cluster,
 					PodRoleLabel:          "spilo-role",
 					ResourceCheckInterval: time.Duration(testResourceCheckInterval),
 					ResourceCheckTimeout:  time.Duration(testResourceCheckTimeout),
+					MinInstances:          -1,
+					MaxInstances:          -1,
 				},
 			},
 		}, client, pg, logger, eventRecorder)
@@ -379,6 +395,17 @@ func annotateResources(cluster *Cluster) error {
 			return err
 		}
 	}
+
+	endpoints, err := cluster.KubeClient.Endpoints(namespace).List(context.TODO(), clusterOptions)
+	if err != nil {
+		return err
+	}
+	for _, ep := range endpoints.Items {
+		ep.Annotations = externalAnnotations
+		if _, err = cluster.KubeClient.Endpoints(namespace).Update(context.TODO(), &ep, metav1.UpdateOptions{}); err != nil {
+			return err
+		}
+	}
 	return nil
 }