Use Patch objects for small updates

JSON Merge Patch cannot describe partial changes to JSON arrays, and CustomResources are never going to support Strategic Merge Patch (https://issue.k8s.io/58414). In preparation for more deeply structured CustomResources, we should get into the habit of using JSON Patches. This commit introduces two new types, `kubeapi.JSON6902` and `kubeapi.Merge7386`, to represent entire JSON Patch and JSON Merge Patch documents. The former escapes individual tokens of the operation paths, making it easier and safer to use than the existing `util.JSONPatchOperation` type. If we ever start using the `sigs.k8s.io/controller-runtime/pkg/client` package, these types are positioned to implement its `Patch` interface. Additionally, using these new types eliminates our direct dependency on `github.com/evanphx/json-patch` which has been a Go module since v4.6.0 and should be imported as `github.com/evanphx/json-patch/v4`. This reduces some of the effort needed to switch to Go modules. Issue: [ch9453]

Author: cbandy

Gopkg.lock

@@ -11,10 +11,6 @@ required = [
  "k8s.io/code-generator",
 ]
 
-[[constraint]]
- name = "github.com/evanphx/json-patch"
- version = "4.6.0"
-
 [[constraint]]
  name = "github.com/fatih/color"
  version = "1.9.0"

Gopkg.toml

@@ -16,19 +16,16 @@ limitations under the License.
 */
 
 import (
-"encoding/json"
 "errors"
 "strings"
 
 "github.com/crunchydata/postgres-operator/internal/apiserver"
 "github.com/crunchydata/postgres-operator/internal/config"
+"github.com/crunchydata/postgres-operator/internal/kubeapi"
 crv1 "github.com/crunchydata/postgres-operator/pkg/apis/crunchydata.com/v1"
 msgs "github.com/crunchydata/postgres-operator/pkg/apiservermsgs"
 "github.com/crunchydata/postgres-operator/pkg/events"
-jsonpatch "github.com/evanphx/json-patch"
 log "github.com/sirupsen/logrus"
-v1 "k8s.io/api/apps/v1"
-"k8s.io/apimachinery/pkg/api/meta"
 metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 "k8s.io/apimachinery/pkg/types"
 "k8s.io/apimachinery/pkg/util/validation"
@@ -117,12 +114,18 @@ func Label(request *msgs.LabelRequest, ns, pgouser string) msgs.LabelResponse {
 }
 
 func addLabels(items []crv1.Pgcluster, DryRun bool, LabelCmdLabel string, newLabels map[string]string, ns, pgouser string) {
+patchBytes, err := kubeapi.NewMergePatch().Add("metadata", "labels")(newLabels).Bytes()
+if err != nil {
+log.Error(err.Error())
+return
+}
+
 for i := 0; i < len(items); i++ {
 if DryRun {
 log.Debug("dry run only")
 } else {
-log.Debugf("adding label to cluster %s", items[i].Spec.Name)
-err := PatchPgcluster(newLabels, items[i], ns)
+log.Debugf("patching cluster %s: %s", items[i].Spec.Name, patchBytes)
+_, err := apiserver.Clientset.CrunchydataV1().Pgclusters(ns).Patch(items[i].Spec.Name, types.MergePatchType, patchBytes)
 if err != nil {
 log.Error(err.Error())
 }
@@ -163,8 +166,8 @@ func addLabels(items []crv1.Pgcluster, DryRun bool, LabelCmdLabel string, newLab
 for _, d := range deployments.Items {
 //update Deployment with the label
 if !DryRun {
-//err := updateLabels(&d, items[i].Spec.Name, newLabels)
-err := updateLabels(&d, d.Name, newLabels, ns)
+log.Debugf("patching deployment %s: %s", d.Name, patchBytes)
+_, err := apiserver.Clientset.AppsV1().Deployments(ns).Patch(d.Name, types.MergePatchType, patchBytes)
 if err != nil {
 log.Error(err.Error())
 }
@@ -174,82 +177,6 @@ func addLabels(items []crv1.Pgcluster, DryRun bool, LabelCmdLabel string, newLab
 }
 }
 
-func updateLabels(deployment *v1.Deployment, clusterName string, newLabels map[string]string, ns string) error {
-
-var err error
-
-log.Debugf("%v are the labels to apply", newLabels)
-
-var patchBytes, newData, origData []byte
-origData, err = json.Marshal(deployment)
-if err != nil {
-return err
-}
-
-accessor, err2 := meta.Accessor(deployment)
-if err2 != nil {
-return err2
-}
-
-objLabels := accessor.GetLabels()
-if objLabels == nil {
-objLabels = make(map[string]string)
-}
-
-//update the deployment labels
-for key, value := range newLabels {
-objLabels[key] = value
-}
-log.Debugf("updated labels are %v", objLabels)
-
-accessor.SetLabels(objLabels)
-newData, err = json.Marshal(deployment)
-if err != nil {
-return err
-}
-
-patchBytes, err = jsonpatch.CreateMergePatch(origData, newData)
-if err != nil {
-return err
-}
-
-_, err = apiserver.Clientset.AppsV1().Deployments(ns).Patch(clusterName, types.MergePatchType, patchBytes, "")
-if err != nil {
-log.Debugf("error updating patching deployment %s", err.Error())
-}
-return err
-
-}
-
-func PatchPgcluster(newLabels map[string]string, oldCRD crv1.Pgcluster, ns string) error {
-
-oldData, err := json.Marshal(oldCRD)
-if err != nil {
-return err
-}
-if oldCRD.ObjectMeta.Labels == nil {
-oldCRD.ObjectMeta.Labels = make(map[string]string)
-}
-for key, value := range newLabels {
-oldCRD.ObjectMeta.Labels[key] = value
-}
-var newData, patchBytes []byte
-newData, err = json.Marshal(oldCRD)
-if err != nil {
-return err
-}
-patchBytes, err = jsonpatch.CreateMergePatch(oldData, newData)
-if err != nil {
-return err
-}
-
-log.Debug(string(patchBytes))
-_, err6 := apiserver.Clientset.CrunchydataV1().Pgclusters(ns).Patch(oldCRD.Spec.Name, types.MergePatchType, patchBytes)
-
-return err6
-
-}
-
 func validateLabel(LabelCmdLabel, ns string) (map[string]string, error) {
 var err error
 labelMap := make(map[string]string)
@@ -361,11 +288,19 @@ func DeleteLabel(request *msgs.DeleteLabelRequest, ns string) msgs.LabelResponse
 }
 
 func deleteLabels(items []crv1.Pgcluster, LabelCmdLabel string, labelsMap map[string]string, ns string) error {
-var err error
+patch := kubeapi.NewMergePatch()
+for key := range labelsMap {
+patch.Remove("metadata", "labels", key)
+}
+patchBytes, err := patch.Bytes()
+if err != nil {
+log.Error(err.Error())
+return err
+}
 
 for i := 0; i < len(items); i++ {
-log.Debugf("deleting label from %s", items[i].Spec.Name)
-err = deletePatchPgcluster(labelsMap, items[i], ns)
+log.Debugf("patching cluster %s: %s", items[i].Spec.Name, patchBytes)
+_, err = apiserver.Clientset.CrunchydataV1().Pgclusters(ns).Patch(items[i].Spec.Name, types.MergePatchType, patchBytes)
 if err != nil {
 log.Error(err.Error())
 return err
@@ -383,7 +318,8 @@ func deleteLabels(items []crv1.Pgcluster, LabelCmdLabel string, labelsMap map[st
 }
 
 for _, d := range deployments.Items {
-err = deleteTheLabel(&d, items[i].Spec.Name, labelsMap, ns)
+log.Debugf("patching deployment %s: %s", d.Name, patchBytes)
+_, err = apiserver.Clientset.AppsV1().Deployments(ns).Patch(d.Name, types.MergePatchType, patchBytes)
 if err != nil {
 log.Error(err.Error())
 return err
@@ -393,79 +329,3 @@ func deleteLabels(items []crv1.Pgcluster, LabelCmdLabel string, labelsMap map[st
 }
 return err
 }
-
-func deletePatchPgcluster(labelsMap map[string]string, oldCRD crv1.Pgcluster, ns string) error {
-
-oldData, err := json.Marshal(oldCRD)
-if err != nil {
-return err
-}
-if oldCRD.ObjectMeta.Labels == nil {
-oldCRD.ObjectMeta.Labels = make(map[string]string)
-}
-for k := range labelsMap {
-delete(oldCRD.ObjectMeta.Labels, k)
-}
-
-var newData, patchBytes []byte
-newData, err = json.Marshal(oldCRD)
-if err != nil {
-return err
-}
-patchBytes, err = jsonpatch.CreateMergePatch(oldData, newData)
-if err != nil {
-return err
-}
-
-log.Debug(string(patchBytes))
-_, err6 := apiserver.Clientset.CrunchydataV1().Pgclusters(ns).Patch(oldCRD.Spec.Name, types.MergePatchType, patchBytes)
-
-return err6
-
-}
-
-func deleteTheLabel(deployment *v1.Deployment, clusterName string, labelsMap map[string]string, ns string) error {
-
-var err error
-
-log.Debugf("%v are the labels to delete", labelsMap)
-
-var patchBytes, newData, origData []byte
-origData, err = json.Marshal(deployment)
-if err != nil {
-return err
-}
-
-accessor, err2 := meta.Accessor(deployment)
-if err2 != nil {
-return err2
-}
-
-objLabels := accessor.GetLabels()
-if objLabels == nil {
-objLabels = make(map[string]string)
-}
-
-for k := range labelsMap {
-delete(objLabels, k)
-}
-log.Debugf("revised labels after delete are %v", objLabels)
-
-accessor.SetLabels(objLabels)
-newData, err = json.Marshal(deployment)
-if err != nil {
-return err
-}
-
-patchBytes, err = jsonpatch.CreateMergePatch(origData, newData)
-if err != nil {
-return err
-}
-
-_, err = apiserver.Clientset.AppsV1().Deployments(ns).Patch(deployment.Name, types.MergePatchType, patchBytes, "")
-if err != nil {
-log.Debugf("error patching deployment: %v", err.Error())
-}
-return err
-
-}

internal/apiserver/labelservice/labelimpl.go

@@ -21,8 +21,8 @@ import (
 "time"
 
 "github.com/crunchydata/postgres-operator/internal/apiserver"
-"github.com/crunchydata/postgres-operator/internal/apiserver/labelservice"
 "github.com/crunchydata/postgres-operator/internal/config"
+"github.com/crunchydata/postgres-operator/internal/kubeapi"
 "github.com/crunchydata/postgres-operator/internal/util"
 crv1 "github.com/crunchydata/postgres-operator/pkg/apis/crunchydata.com/v1"
 msgs "github.com/crunchydata/postgres-operator/pkg/apiservermsgs"
@@ -32,6 +32,7 @@ import (
 v1 "k8s.io/api/apps/v1"
 kerrors "k8s.io/apimachinery/pkg/api/errors"
 metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+"k8s.io/apimachinery/pkg/types"
 )
 
 // CreatePolicy ...
@@ -212,6 +213,13 @@ func ApplyPolicy(request *msgs.ApplyPolicyRequest, ns, pgouser string) msgs.Appl
 labels := make(map[string]string)
 labels[request.Name] = "pgpolicy"
 
+patch, err := kubeapi.NewMergePatch().Add("metadata", "labels")(labels).Bytes()
+if err != nil {
+resp.Status.Code = msgs.Error
+resp.Status.Msg = err.Error()
+return resp
+}
+
 for _, d := range allDeployments {
 if d.ObjectMeta.Labels[config.LABEL_SERVICE_NAME] != d.ObjectMeta.Labels[config.LABEL_PG_CLUSTER] {
 log.Debugf("skipping apply policy on deployment %s", d.Name)
@@ -238,13 +246,15 @@ func ApplyPolicy(request *msgs.ApplyPolicyRequest, ns, pgouser string) msgs.Appl
 return resp
 }
 
-err = util.UpdatePolicyLabels(apiserver.Clientset, d.ObjectMeta.Name, ns, labels)
+log.Debugf("patching deployment %s: %s", d.ObjectMeta.Name, patch)
+_, err = apiserver.Clientset.AppsV1().Deployments(ns).Patch(d.ObjectMeta.Name, types.MergePatchType, patch)
 if err != nil {
 log.Error(err)
 }
 
 //update the pgcluster crd labels with the new policy
-err = labelservice.PatchPgcluster(map[string]string{request.Name: config.LABEL_PGPOLICY}, *cl, ns)
+log.Debugf("patching cluster %s: %s", cl.Name, patch)
+_, err = apiserver.Clientset.CrunchydataV1().Pgclusters(ns).Patch(cl.Name, types.MergePatchType, patch)
 if err != nil {
 log.Error(err)
 }

internal/apiserver/policyservice/policyimpl.go

@@ -22,9 +22,11 @@ import (
 log "github.com/sirupsen/logrus"
 apiv1 "k8s.io/api/batch/v1"
 metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+"k8s.io/apimachinery/pkg/types"
 
 "github.com/crunchydata/postgres-operator/internal/config"
 "github.com/crunchydata/postgres-operator/internal/controller"
+"github.com/crunchydata/postgres-operator/internal/kubeapi"
 "github.com/crunchydata/postgres-operator/internal/operator/backrest"
 backrestoperator "github.com/crunchydata/postgres-operator/internal/operator/backrest"
 clusteroperator "github.com/crunchydata/postgres-operator/internal/operator/cluster"
@@ -87,7 +89,12 @@ func (c *Controller) handleCloneBackrestRestoreUpdate(job *apiv1.Job) error {
 
 // first, make sure the Pgtask resource knows that the job is complete,
 // which is using this legacy bit of code
-if err := util.Patch(c.Client.CrunchydataV1().RESTClient(), patchURL, crv1.JobCompletedStatus, patchResource, job.Name, namespace); err != nil {
+patch, err := kubeapi.NewJSONPatch().Add("spec", "status")(crv1.JobCompletedStatus).Bytes()
+if err == nil {
+log.Debugf("patching task %s: %s", job.Name, patch)
+_, err = c.Client.CrunchydataV1().Pgtasks(namespace).Patch(job.Name, types.JSONPatchType, patch)
+}
+if err != nil {
 log.Warn(err)
 // we can continue on, even if this fails...
 }
@@ -132,8 +139,12 @@ func (c *Controller) handleBackrestBackupUpdate(job *apiv1.Job) error {
 log.Debugf("got a backrest job status=%d", job.Status.Succeeded)
 log.Debugf("update the status to completed here for backrest %s job %s", labels[config.LABEL_PG_CLUSTER], job.Name)
 
-if err := util.Patch(c.Client.CrunchydataV1().RESTClient(), patchURL, crv1.JobCompletedStatus, patchResource, job.Name,
-job.ObjectMeta.Namespace); err != nil {
+patch, err := kubeapi.NewJSONPatch().Add("spec", "status")(crv1.JobCompletedStatus).Bytes()
+if err == nil {
+log.Debugf("patching task %s: %s", job.Name, patch)
+_, err = c.Client.CrunchydataV1().Pgtasks(job.Namespace).Patch(job.Name, types.JSONPatchType, patch)
+}
+if err != nil {
 log.Errorf("error in patching pgtask %s: %s", job.ObjectMeta.SelfLink, err.Error())
 }
 publishBackupComplete(labels[config.LABEL_PG_CLUSTER], job.ObjectMeta.Labels[config.LABEL_PG_CLUSTER_IDENTIFIER], job.ObjectMeta.Labels[config.LABEL_PGOUSER], "pgbackrest", job.ObjectMeta.Namespace, "")

internal/controller/job/backresthandler.go

@@ -30,11 +30,6 @@ type Controller struct {
 Informer batchinformers.JobInformer
 }
 
-const (
-patchResource = "pgtasks"
-patchURL = "/spec/status"
-)
-
 // onAdd is called when a postgresql operator job is created and an associated add event is
 // generated
 func (c *Controller) onAdd(obj interface{}) {