Duplicate the service as well as the Deployment.

config/default/kustomization.yaml

@@ -1,3 +1,11 @@
+apiVersion: kustomize.config.k8s.io/v1beta1
+kind: Kustomization
+patches:
+# Protect the /metrics endpoint by putting it behind auth.
+# If you want your controller-manager to expose the /metrics
+# endpoint w/o any authn/z, please comment the following line.
+- path: manager_auth_proxy_patch.yaml
+
 # Adds namespace to all resources.
 namespace: flux-system
 
@@ -26,14 +34,6 @@ resources:
 # [PROMETHEUS] To enable prometheus monitor, uncomment all sections with 'PROMETHEUS'.
 #- ../prometheus
 
-patchesStrategicMerge:
-# Protect the /metrics endpoint by putting it behind auth.
-# If you want your controller-manager to expose the /metrics
-# endpoint w/o any authn/z, please comment the following line.
-- manager_auth_proxy_patch.yaml
-
-
-
 # [WEBHOOK] To enable webhook, uncomment all the sections with [WEBHOOK] prefix including the one in
 # crd/kustomization.yaml
 #- manager_webhook_patch.yaml

config/rbac/role.yaml

@@ -5,6 +5,18 @@ metadata:
   creationTimestamp: null
   name: manager-role
 rules:
+- apiGroups:
+  - ""
+  resources:
+  - services
+  verbs:
+  - create
+  - delete
+  - get
+  - list
+  - patch
+  - update
+  - watch
 - apiGroups:
   - apps
   resources:

go.mod

@@ -3,6 +3,7 @@ module github.com/weaveworks/flux-shard-controller
 go 1.19
 
 require (
+	dario.cat/mergo v1.0.0
 	github.com/fluxcd/pkg/apis/meta v1.1.0
 	github.com/fluxcd/pkg/runtime v0.38.1
 	github.com/gitops-tools/pkg v0.1.0

go.sum

@@ -1,5 +1,8 @@
 cloud.google.com/go v0.26.0/go.mod h1:aQUYkXzVsufM+DwF1aE+0xfcU+56JwCaLick0ClmMTw=
 cloud.google.com/go v0.34.0/go.mod h1:aQUYkXzVsufM+DwF1aE+0xfcU+56JwCaLick0ClmMTw=
+dario.cat/mergo v1.0.0 h1:AGCNq9Evsj31mOgNPcLyXc+4PNABt905YmuqPYYpBWk=
+dario.cat/mergo v1.0.0/go.mod h1:uNxQE+84aUszobStD9th8a29P2fMDhsBdgRYvZOxGmk=
+github.com/AdaLogics/go-fuzz-headers v0.0.0-20230106234847-43070de90fa1 h1:EKPd1INOIyr5hWOWhvpmQpY6tKjeG0hT1s3AMC/9fic=
 github.com/Azure/go-ansiterm v0.0.0-20210617225240-d185dfc1b5a1 h1:UQHMgLO+TxOElx5B5HZ4hJQsoJ/PvUvKRhJHDQXO8P8=
 github.com/BurntSushi/toml v0.3.1/go.mod h1:xHWCNGjB5oqiDr8zfno3MHue2Ht5sIBksp03qcyfWMU=
 github.com/MakeNowJust/heredoc v1.0.0 h1:cXCdzVdstXyiTqTvfqk9SDHpKNjxuom+DOlyEeQ4pzQ=
@@ -23,6 +26,7 @@ github.com/cncf/udpa/go v0.0.0-20191209042840-269d4d468f6f/go.mod h1:M8M6+tZqaGX
 github.com/cncf/udpa/go v0.0.0-20201120205902-5459f2c99403/go.mod h1:WmhPx2Nbnhtbo57+VJT5O0JRkEi1Wbu0z5j0R8u5Hbk=
 github.com/cncf/xds/go v0.0.0-20210312221358-fbca930ec8ed/go.mod h1:eXthEFrGJvWHgFFCl3hGmgk+/aYT6PnTQLykKQRLhEs=
 github.com/creack/pty v1.1.9/go.mod h1:oKZEueFk5CKHvIhNR5MUki03XCEU+Q6VDXinZuGJ33E=
+github.com/cyphar/filepath-securejoin v0.2.3 h1:YX6ebbZCZP7VkM3scTTokDgBL2TY741X51MTk3ycuNI=
 github.com/davecgh/go-spew v1.1.0/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38=
 github.com/davecgh/go-spew v1.1.1 h1:vj9j/u1bqnvCEfJOwUhtlOARqs3+rkHYY13jYWTU97c=
 github.com/davecgh/go-spew v1.1.1/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38=

internal/controller/fluxshardset_controller.go

@@ -19,7 +19,11 @@ package controller
 import (
 	"context"
 	"fmt"
+	"reflect"
 
+	"dario.cat/mergo"
+	"github.com/fluxcd/pkg/runtime/patch"
+	corev1 "k8s.io/api/core/v1"
 	apierrors "k8s.io/apimachinery/pkg/api/errors"
 	"k8s.io/apimachinery/pkg/api/meta"
 	"k8s.io/apimachinery/pkg/runtime"
@@ -38,6 +42,7 @@ import (
 	deploys "github.com/weaveworks/flux-shard-controller/internal/deploys"
 	appsv1 "k8s.io/api/apps/v1"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	"k8s.io/apimachinery/pkg/apis/meta/v1/unstructured"
 )
 
 var accessor = meta.NewAccessor()
@@ -53,6 +58,7 @@ type FluxShardSetReconciler struct {
 // +kubebuilder:rbac:groups=templates.weave.works,resources=fluxshardsets,verbs=get;list;watch;create;update;patch;delete
 // +kubebuilder:rbac:groups=templates.weave.works,resources=fluxshardsets/status,verbs=get;update;patch
 // +kubebuilder:rbac:groups=apps,resources=deployments,verbs=get;list;watch;create;update;patch;delete
+// +kubebuilder:rbac:groups="",resources=services,verbs=get;list;watch;create;update;patch;delete
 
 // Reconcile is part of the main kubernetes reconciliation loop which aims to
 // move the current state of the cluster closer to the desired state.
@@ -89,7 +95,7 @@ func (r *FluxShardSetReconciler) Reconcile(ctx context.Context, req ctrl.Request
 
 	if inventory != nil {
 		templatesv1.SetReadyWithInventory(&shardSet, inventory, templatesv1.ReconciliationSucceededReason,
-			fmt.Sprintf("%d shard(s) created", len(inventory.Entries)))
+			fmt.Sprintf("%d resources created", len(inventory.Entries)))
 
 		if err := r.patchStatus(ctx, req, shardSet.Status); client.IgnoreNotFound(err) != nil {
 			templatesv1.SetFluxShardSetReadiness(&shardSet, metav1.ConditionFalse, templatesv1.ReconciliationFailedReason, err.Error())
@@ -104,7 +110,7 @@ func (r *FluxShardSetReconciler) Reconcile(ctx context.Context, req ctrl.Request
 func (r *FluxShardSetReconciler) removeResourceRefs(ctx context.Context, deletions []templatesv1.ResourceRef) error {
 	logger := log.FromContext(ctx)
 	for _, v := range deletions {
-		d, err := deploymentFromResourceRef(v)
+		d, err := unstructuredFromResourceRef(v)
 		if err != nil {
 			return err
 		}
@@ -143,7 +149,12 @@ func (r *FluxShardSetReconciler) reconcileResources(ctx context.Context, fluxSha
 		return nil, client.IgnoreNotFound(err)
 	}
 
-	generatedDeployments, err := deploys.GenerateDeployments(fluxShardSet, srcDeploy)
+	srcService, err := r.getSourceService(ctx, srcDeploy)
+	if err != nil {
+		return nil, fmt.Errorf("failed to find Service for Deployment %s: %w", client.ObjectKeyFromObject(srcDeploy), err)
+	}
+
+	generatedResources, err := deploys.GenerateDeployments(fluxShardSet, srcDeploy, srcService)
 	if err != nil {
 		return nil, fmt.Errorf("failed to generate deployments: %w", err)
 	}
@@ -156,43 +167,49 @@ func (r *FluxShardSetReconciler) reconcileResources(ctx context.Context, fluxSha
 	// newInventory holds the resource refs for the generated resources.
 	newInventory := sets.New[templatesv1.ResourceRef]()
 
-	for _, newDeployment := range generatedDeployments {
-		ref, err := templatesv1.ResourceRefFromObject(newDeployment)
+	for _, newResource := range generatedResources {
+		ref, err := templatesv1.ResourceRefFromObject(newResource)
 		if err != nil {
 			return nil, fmt.Errorf("failed to update inventory: %w", err)
 		}
-
 		if existingInventory.Has(ref) {
 			newInventory.Insert(ref)
-			existing := &appsv1.Deployment{}
-			err = r.Client.Get(ctx, client.ObjectKeyFromObject(newDeployment), existing)
+			existing := runtimeObjectFromObject(newResource)
+			err = r.Client.Get(ctx, client.ObjectKeyFromObject(newResource), existing)
 			if err == nil {
-				newDeployment = copyDeploymentContent(existing, newDeployment)
-				if err := r.Client.Patch(ctx, newDeployment, client.MergeFrom(existing)); err != nil {
-					return nil, fmt.Errorf("failed to update Deployment: %w", err)
+				patchHelper, err := patch.NewHelper(existing, r.Client)
+				if err != nil {
+					return nil, err
+				}
+				if err := mergo.Merge(existing, newResource, mergo.WithOverride); err != nil {
+					return nil, err
 				}
 
-				if err := logResourceMessage(logger, "updated deployment", newDeployment); err != nil {
+				if err := patchHelper.Patch(ctx, existing); err != nil {
+					return nil, err
+				}
+
+				if err := logResourceMessage(logger, "updated resource", newResource); err != nil {
 					return nil, err
 				}
 				continue
 			}
 
 			if !apierrors.IsNotFound(err) {
-				return nil, fmt.Errorf("failed to load existing Deployment: %w", err)
+				return nil, fmt.Errorf("failed to load existing resource %s: %w", client.ObjectKeyFromObject(newResource), err)
 			}
 
 		}
 
-		if err := controllerutil.SetOwnerReference(fluxShardSet, newDeployment, r.Scheme); err != nil {
+		if err := controllerutil.SetOwnerReference(fluxShardSet, newResource, r.Scheme); err != nil {
 			return nil, fmt.Errorf("failed to set owner reference: %w", err)
 		}
 
-		if err := r.Client.Create(ctx, newDeployment); err != nil {
-			return nil, fmt.Errorf("failed to create Deployment: %w", err)
+		if err := r.Client.Create(ctx, newResource); err != nil {
+			return nil, fmt.Errorf("failed to create resource: %w", err)
 		}
 		newInventory.Insert(ref)
-		if err := logResourceMessage(logger, "created new deployment", newDeployment); err != nil {
+		if err := logResourceMessage(logger, "created new resource", newResource); err != nil {
 			return nil, err
 		}
 	}
@@ -228,6 +245,19 @@ func (r *FluxShardSetReconciler) getSourceDeployment(ctx context.Context, fluxSh
 	return srcDeploy, nil
 }
 
+func (r *FluxShardSetReconciler) getSourceService(ctx context.Context, deploy *appsv1.Deployment) (*corev1.Service, error) {
+	srcServiceKey := client.ObjectKey{
+		Name:      deploy.GetName(),
+		Namespace: deploy.GetNamespace(),
+	}
+	srcService := &corev1.Service{}
+	if err := r.Client.Get(ctx, srcServiceKey, srcService); err != nil {
+		return nil, client.IgnoreNotFound(err)
+	}
+
+	return srcService, nil
+}
+
 func (r *FluxShardSetReconciler) patchStatus(ctx context.Context, req ctrl.Request, newStatus templatesv1.FluxShardSetStatus) error {
 	var set templatesv1.FluxShardSet
 	if err := r.Client.Get(ctx, req.NamespacedName, &set); err != nil {
@@ -256,19 +286,6 @@ func (r *FluxShardSetReconciler) deploymentsToFluxShardSet(ctx context.Context,
 	return result
 }
 
-func deploymentFromResourceRef(ref templatesv1.ResourceRef) (*appsv1.Deployment, error) {
-	objMeta, err := object.ParseObjMetadata(ref.ID)
-	if err != nil {
-		return nil, fmt.Errorf("failed to parse object ID %s: %w", ref.ID, err)
-	}
-	d := appsv1.Deployment{}
-
-	d.Namespace = objMeta.Namespace
-	d.Name = objMeta.Name
-
-	return &d, nil
-}
-
 func logResourceMessage(logger logr.Logger, msg string, obj runtime.Object) error {
 	namespace, err := accessor.Namespace(obj)
 	if err != nil {
@@ -307,3 +324,20 @@ func indexDeployments(o client.Object) []string {
 
 	return []string{fmt.Sprintf("%s/%s", fss.GetNamespace(), fss.Spec.SourceDeploymentRef.Name)}
 }
+
+func runtimeObjectFromObject(o client.Object) client.Object {
+	return reflect.New(reflect.TypeOf(o).Elem()).Interface().(client.Object)
+}
+
+func unstructuredFromResourceRef(ref templatesv1.ResourceRef) (*unstructured.Unstructured, error) {
+	objMeta, err := object.ParseObjMetadata(ref.ID)
+	if err != nil {
+		return nil, fmt.Errorf("failed to parse object ID %s: %w", ref.ID, err)
+	}
+	u := unstructured.Unstructured{}
+	u.SetGroupVersionKind(objMeta.GroupKind.WithVersion(ref.Version))
+	u.SetName(objMeta.Name)
+	u.SetNamespace(objMeta.Namespace)
+
+	return &u, nil
+}