appliance: reconciliation helpers (#61932)

* appliance: set pseudo-CRD name values

* appliance: receive and store requested version

Make the version part of the appliance-driven config, and store it on a
ConfigMap annotation. Service-specific upgrade logic then has access to
both the old and the new version, supporting complex multiversion
upgrades as needed.

The "Sourcegraph" config object is a kubebuilder-scaffolded custom type,
but we don't actually use CRDs. We drive its status field using this
ConfigMap annotation, so that lower-level code can behave as if we do
use CRDs. This is similar to what we do with the namespace field.

* appliance: add some standard labels to default deployment

Notably the version, which might be useful for debugging. Label keys
lifted from the helm chart.

* appliance: deploy blobstore deployment using new reconciliation logic

Author: Craig Furman

internal/appliance/BUILD.bazel

@@ -5,6 +5,7 @@ go_library(
     name = "appliance",
     srcs = [
         "blobstore.go",
+        "kubernetes.go",
         "reconcile.go",
         "spec.go",
     ],
@@ -17,7 +18,6 @@ go_library(
         "//internal/k8s/resource/pod",
         "//internal/k8s/resource/pvc",
         "//internal/k8s/resource/service",
-        "//internal/maps",
         "//lib/errors",
         "//lib/pointers",
         "@io_k8s_api//apps/v1:apps",

internal/appliance/blobstore.go

@@ -3,35 +3,30 @@ package appliance
 import (
 	"context"
 
-	appsv1 "k8s.io/api/apps/v1"
-	corev1 "k8s.io/api/core/v1"
-	apierrors "k8s.io/apimachinery/pkg/api/errors"
-	"k8s.io/apimachinery/pkg/api/resource"
-	"k8s.io/apimachinery/pkg/util/intstr"
-	"sigs.k8s.io/controller-runtime/pkg/client"
-
-	"github.com/sourcegraph/sourcegraph/lib/errors"
-	"github.com/sourcegraph/sourcegraph/lib/pointers"
-
-	"github.com/sourcegraph/sourcegraph/internal/appliance/hash"
 	"github.com/sourcegraph/sourcegraph/internal/k8s/resource/container"
 	"github.com/sourcegraph/sourcegraph/internal/k8s/resource/deployment"
 	"github.com/sourcegraph/sourcegraph/internal/k8s/resource/pod"
 	"github.com/sourcegraph/sourcegraph/internal/k8s/resource/pvc"
 	"github.com/sourcegraph/sourcegraph/internal/k8s/resource/service"
-	"github.com/sourcegraph/sourcegraph/internal/maps"
+	"github.com/sourcegraph/sourcegraph/lib/errors"
+	"github.com/sourcegraph/sourcegraph/lib/pointers"
+	appsv1 "k8s.io/api/apps/v1"
+	corev1 "k8s.io/api/core/v1"
+	"k8s.io/apimachinery/pkg/api/resource"
+	"k8s.io/apimachinery/pkg/util/intstr"
+	"sigs.k8s.io/controller-runtime/pkg/client"
 )
 
-func (r *Reconciler) reconcileBlobstore(ctx context.Context, sg *Sourcegraph) error {
-	if err := r.reconcileBlobstorePersistentVolumeClaims(ctx, sg); err != nil {
+func (r *Reconciler) reconcileBlobstore(ctx context.Context, sg *Sourcegraph, owner client.Object) error {
+	if err := r.reconcileBlobstorePersistentVolumeClaims(ctx, sg, owner); err != nil {
 		return err
 	}
 
-	if err := r.reconcileBlobstoreServices(ctx, sg); err != nil {
+	if err := r.reconcileBlobstoreServices(ctx, sg, owner); err != nil {
 		return err
 	}
 
-	if err := r.reconcileBlobstoreDeployments(ctx, sg); err != nil {
+	if err := r.reconcileBlobstoreDeployments(ctx, sg, owner); err != nil {
 		return err
 	}
 
@@ -72,42 +67,13 @@ func buildBlobstorePersistentVolumeClaim(sg *Sourcegraph) (corev1.PersistentVolu
 	return p, nil
 }
 
-func (r *Reconciler) reconcileBlobstorePersistentVolumeClaims(ctx context.Context, sg *Sourcegraph) error {
+func (r *Reconciler) reconcileBlobstorePersistentVolumeClaims(ctx context.Context, sg *Sourcegraph, owner client.Object) error {
 	p, err := buildBlobstorePersistentVolumeClaim(sg)
 	if err != nil {
 		return err
 	}
 
-	p.Labels = hash.SetTemplateHashLabel(p.Labels, p.Spec)
-
-	var existing corev1.PersistentVolumeClaim
-	if r.IsObjectFound(ctx, p.Name, p.Namespace, &existing) {
-		if sg.Spec.Blobstore.Disabled {
-			return nil
-		}
-
-		// Object exists update if needed
-		if hash.GetTemplateHashLabel(existing.Labels) == hash.GetTemplateHashLabel(p.Labels) {
-			// no updates needed
-			return nil
-		}
-
-		// need to update
-		existing.Labels = maps.Merge(existing.Labels, p.Labels)
-		existing.Annotations = maps.Merge(existing.Annotations, p.Annotations)
-		existing.Spec = p.Spec
-
-		return r.Update(ctx, &existing)
-	}
-
-	if sg.Spec.Blobstore.Disabled {
-		return nil
-	}
-
-	// Note: we don't set a controller reference here as we want PVCs to persist if blobstore is deleted.
-	// This helps to protect against accidental data deletions.
-
-	return r.Create(ctx, &p)
+	return reconcileBlobStoreObject(ctx, r, &p, &corev1.PersistentVolumeClaim{}, sg, owner)
 }
 
 func buildBlobstoreService(sg *Sourcegraph) (corev1.Service, error) {
@@ -131,53 +97,12 @@ func buildBlobstoreService(sg *Sourcegraph) (corev1.Service, error) {
 	return s, nil
 }
 
-func (r *Reconciler) reconcileBlobstoreServices(ctx context.Context, sg *Sourcegraph) error {
+func (r *Reconciler) reconcileBlobstoreServices(ctx context.Context, sg *Sourcegraph, owner client.Object) error {
 	s, err := buildBlobstoreService(sg)
 	if err != nil {
 		return err
 	}
-
-	s.Labels = hash.SetTemplateHashLabel(s.Labels, s.Spec)
-
-	var existing corev1.Service
-	if r.IsObjectFound(ctx, s.Name, sg.Namespace, &existing) {
-		if sg.Spec.Blobstore.Disabled {
-			// blobstore service exists, but has been disabled. Delete the service.
-			//
-			// Using a precondition to make sure the version of the resource that is deleted
-			// is the version we intend, and not a resource that was already resgeated.
-			err = r.Delete(ctx, &existing, client.Preconditions{
-				UID:             &existing.UID,
-				ResourceVersion: &existing.ResourceVersion,
-			})
-
-			if err != nil && !apierrors.IsNotFound(err) {
-				return err
-			}
-
-			return nil
-		}
-		// Object exists update if needed
-		if hash.GetTemplateHashLabel(existing.Labels) == hash.GetTemplateHashLabel(s.Labels) {
-			// no updates needed
-			return nil
-		}
-
-		// need to update
-		existing.Labels = maps.Merge(existing.Labels, s.Labels)
-		existing.Annotations = maps.Merge(existing.Annotations, s.Annotations)
-		existing.Spec = s.Spec
-
-		return r.Update(ctx, &existing)
-	}
-
-	if sg.Spec.Blobstore.Disabled {
-		return nil
-	}
-
-	// TODO set owner ref
-
-	return r.Create(ctx, &s)
+	return reconcileBlobStoreObject(ctx, r, &s, &corev1.Service{}, sg, owner)
 }
 
 func buildBlobstoreDeployment(sg *Sourcegraph) (appsv1.Deployment, error) {
@@ -272,6 +197,7 @@ func buildBlobstoreDeployment(sg *Sourcegraph) (appsv1.Deployment, error) {
 	defaultDeployment, err := deployment.NewDeployment(
 		name,
 		sg.Namespace,
+		sg.Spec.RequestedVersion,
 		deployment.WithPodTemplateSpec(podTemplate.Template),
 	)
 
@@ -282,51 +208,29 @@ func buildBlobstoreDeployment(sg *Sourcegraph) (appsv1.Deployment, error) {
 	return defaultDeployment, nil
 }
 
-func (r *Reconciler) reconcileBlobstoreDeployments(ctx context.Context, sg *Sourcegraph) error {
+func (r *Reconciler) reconcileBlobstoreDeployments(ctx context.Context, sg *Sourcegraph, owner client.Object) error {
 	d, err := buildBlobstoreDeployment(sg)
 	if err != nil {
 		return err
 	}
+	return reconcileBlobStoreObject(ctx, r, &d, &appsv1.Deployment{}, sg, owner)
+}
 
-	d.Labels = hash.SetTemplateHashLabel(d.Labels, d.Spec)
-
-	var existing appsv1.Deployment
-	if r.IsObjectFound(ctx, d.Name, sg.Namespace, &existing) {
-		if sg.Spec.Blobstore.Disabled {
-			// blobstore deployment exists, but has been disabled. Delete the deployment.
-			//
-			// Using a precondition to make sure the version of the resource that is deleted
-			// is the version we intend, and not a resource that was already recreated.
-			err = r.Delete(ctx, &existing, client.Preconditions{
-				UID:             &existing.UID,
-				ResourceVersion: &existing.ResourceVersion,
-			})
-
-			if err != nil && !apierrors.IsNotFound(err) {
-				return err
-			}
-
-			return nil
-		}
-		// Object exists update if needed
-		if hash.GetTemplateHashLabel(existing.Labels) == hash.GetTemplateHashLabel(d.Labels) {
-			// no updates needed
-			return nil
-		}
-
-		// need to update
-		existing.Labels = maps.Merge(existing.Labels, d.Labels)
-		existing.Annotations = maps.Merge(existing.Annotations, d.Annotations)
-		existing.Spec = d.Spec
-
-		return r.Update(ctx, &existing)
-	}
-
+func reconcileBlobStoreObject[T client.Object](ctx context.Context, r *Reconciler, obj, objKind T, sg *Sourcegraph, owner client.Object) error {
 	if sg.Spec.Blobstore.Disabled {
-		return nil
+		return r.ensureObjectDeleted(ctx, obj)
 	}
 
-	// TODO set owner ref
+	// Any secrets (or other configmaps) referenced in BlobStoreSpec can be
+	// added to this struct so that they are hashed, and cause an update to the
+	// Deployment if changed.
+	updateIfChanged := struct {
+		BlobstoreSpec
+		Version string
+	}{
+		BlobstoreSpec: sg.Spec.Blobstore,
+		Version:       sg.Spec.RequestedVersion,
+	}
 
-	return r.Create(ctx, &d)
+	return createOrUpdateObject(ctx, r, updateIfChanged, owner, obj, objKind)
 }

internal/appliance/kubernetes.go

@@ -0,0 +1,104 @@
+package appliance
+
+import (
+	"context"
+	"crypto/sha256"
+	"encoding/hex"
+	"encoding/json"
+
+	kerrors "k8s.io/apimachinery/pkg/api/errors"
+	"k8s.io/apimachinery/pkg/types"
+	ctrl "sigs.k8s.io/controller-runtime"
+	"sigs.k8s.io/controller-runtime/pkg/client"
+	"sigs.k8s.io/controller-runtime/pkg/log"
+
+	"github.com/sourcegraph/sourcegraph/lib/errors"
+)
+
+// Upsert a Kubernetes object.
+//
+// obj is the object you want to reconcile, updating an existing cluster object
+// if it has changed, or creating it if none existed before.
+//
+// objKind should be the same type as obj, usually an instantiated
+// struct-pointer to a particular Kubernetes object type, e.g.
+// `&appsv1.Deployment{}`. It is used to hold data about any existing object of
+// the same name, to compare it to obj, and possibly be replaced by obj.
+//
+// updateIfChanged is the object whose hash we store in an annotation to
+// determine whether an existing in-cluster object is out of date and needs to
+// be replaced.
+//
+// Takes the reconciler as a parameter rather than being a method on it due to
+// limitations of Go generics.
+func createOrUpdateObject[R client.Object](
+	ctx context.Context, r *Reconciler, updateIfChanged any,
+	owner client.Object, obj client.Object, objKind R,
+) error {
+	logger := log.FromContext(ctx).WithValues("kind", obj.GetObjectKind().GroupVersionKind(), "namespace", obj.GetNamespace(), "name", obj.GetName())
+	namespacedName := types.NamespacedName{Namespace: obj.GetNamespace(), Name: obj.GetName()}
+
+	cfgHash, err := configHash(updateIfChanged)
+	if err != nil {
+		return err
+	}
+	annotations := obj.GetAnnotations()
+	if annotations == nil {
+		annotations = map[string]string{}
+	}
+	annotations[annotationKeyConfigHash] = cfgHash
+	obj.SetAnnotations(annotations)
+
+	existingRes := objKind
+	if err := r.Client.Get(ctx, namespacedName, existingRes); err != nil {
+		if kerrors.IsNotFound(err) {
+			logger.Info("didn't find existing object, creating it")
+			if err := r.Client.Create(ctx, obj); err != nil {
+				logger.Error(err, "error creating object")
+				return err
+			}
+			return nil
+		}
+
+		logger.Error(err, "unexpected error getting object")
+		return err
+	}
+
+	if err := ctrl.SetControllerReference(owner, obj, r.Scheme); err != nil {
+		return errors.Newf("setting controller reference: %w", err)
+	}
+
+	if cfgHash != existingRes.GetAnnotations()[annotationKeyConfigHash] {
+		logger.Info("Found existing object with spec that does not match desired state. Clobbering it.")
+		if err := r.Client.Update(ctx, obj); err != nil {
+			logger.Error(err, "error updating object")
+			return err
+		}
+		return nil
+	}
+
+	logger.Info("Found existing object with spec that matches the desired state. Will do nothing.")
+	return nil
+}
+
+func (r *Reconciler) ensureObjectDeleted(ctx context.Context, obj client.Object) error {
+	logger := log.FromContext(ctx).WithValues("kind", obj.GetObjectKind().GroupVersionKind(), "namespace", obj.GetNamespace(), "name", obj.GetName())
+	if err := r.Client.Delete(ctx, obj); err != nil {
+		if kerrors.IsNotFound(err) {
+			return nil
+		}
+
+		logger.Error(err, "unexpected error deleting resource")
+		return err
+	}
+	return nil
+}
+
+func configHash(configElement any) (string, error) {
+	cfgBytes, err := json.Marshal(configElement)
+	if err != nil {
+		return "", err
+	}
+	hash := sha256.Sum256(cfgBytes)
+	return hex.EncodeToString(hash[:]), nil
+}

internal/appliance/reconcile.go

@@ -21,6 +21,11 @@ import (
 	"github.com/sourcegraph/sourcegraph/internal/appliance/hash"
 )
 
+const (
+	annotationKeyCurrentVersion = "appliance.sourcegraph.com/currentVersion"
+	annotationKeyConfigHash     = "appliance.sourcegraph.com/configHash"
+)
+
 var _ reconcile.Reconciler = &Reconciler{}
 
 type Reconciler struct {
@@ -56,6 +61,28 @@ func (r *Reconciler) Reconcile(ctx context.Context, req ctrl.Request) (ctrl.Resu
 		return reconcile.Result{}, err
 	}
 
+	// Sourcegraph is a kubebuilder-scaffolded custom type, but we do not
+	// actually ask operators to install CRDs. Therefore we set its namespace
+	// based on the actual object being reconciled, so that more deeply-nested
+	// code can treat it like a CRD.
+	sourcegraph.Namespace = applianceSpec.GetNamespace()
+
+	// Similarly, we simulate a CRD status using an annotation. ConfigMaps don't
+	// have Statuses, so we must use annotations to drive this.
+	// This can be empty string.
+	sourcegraph.Status.CurrentVersion = applianceSpec.GetAnnotations()[annotationKeyCurrentVersion]
+
+	// Reconcile services here
+	if err := r.reconcileBlobstore(ctx, &sourcegraph, &applianceSpec); err != nil {
+		return ctrl.Result{}, errors.Newf("failed to reconcile blobstore: %w", err)
+	}
+
+	// Set the current version annotation in case migration logic depends on it.
+	applianceSpec.Annotations[annotationKeyCurrentVersion] = sourcegraph.Spec.RequestedVersion
+	if err := r.Client.Update(ctx, &applianceSpec); err != nil {
+		return ctrl.Result{}, errors.Newf("failed to update current version annotation: %w", err)
+	}
+
 	return ctrl.Result{}, nil
 }
 

internal/appliance/spec.go

@@ -349,6 +349,9 @@ type StorageClassSpec struct {
 
 // SourcegraphSpec defines the desired state of Sourcegraph
 type SourcegraphSpec struct {
+	// RequestedVersion is the user-requested version of Sourcegraph to deploy.
+	RequestedVersion string `json:"requestedVersion"`
+
 	// ManagementState defines if Sourcegraph should be managed by the operator or not.
 	// Default is managed.
 	ManagementState ManagementStateType `json:"managementState,omitempty"`