feat(reconcile): update existing deployments

.husky/hooks/pre-commit

@@ -0,0 +1,12 @@
+#!/usr/bin/env bash
+set -euo pipefail
+
+cd "$(dirname ${BASH_SOURCE})"/../..
+
+gofmt -s -w .
+
+# This could be a simple "goimports -w ." without find&grep, if only .goimportsignore would work on a per-project basis.
+# See https://github.com/golang/go/issues/42965.
+find . -iname \*.go | grep -v zz_generated.deepcopy.go | xargs goimports -w
+
+git add --patch

.husky/hooks/pre-push

@@ -0,0 +1,9 @@
+#!/usr/bin/env bash
+set -euo pipefail
+
+cd "$(dirname ${BASH_SOURCE})"/../..
+
+make
+make lint
+make test
+make test-e2e

Dockerfile

@@ -12,10 +12,9 @@ COPY go.sum go.sum
 RUN go mod download
 
 # Copy the go source
-COPY cmd/main.go cmd/main.go
+COPY cmd/ cmd/
 COPY api/ api/
-COPY internal/controller/ internal/controller/
-COPY internal/webhook/ internal/webhook/
+COPY internal/ internal/
 
 # Build
 # the GOARCH has not a default value to allow the binary be built according to the host where the command

api/v1alpha1/dash0_types.go

@@ -37,6 +37,17 @@ type Dash0List struct {
 	Items           []Dash0 `json:"items"`
 }
 
+type ConditionType string
+type Reason string
+
+const (
+	ConditionTypeAvailable ConditionType = "Available"
+	ConditionTypeDegraded  ConditionType = "Degraded"
+
+	ReasonSuccessfulInstrumentation Reason = "SuccessfulInstrumentation"
+	ReasonFailedInstrumentation     Reason = "FailedInstrumentation"
+)
+
 func init() {
 	SchemeBuilder.Register(&Dash0{}, &Dash0List{})
 }

cmd/main.go

@@ -15,6 +15,7 @@ import (
 
 	"k8s.io/apimachinery/pkg/runtime"
 	utilruntime "k8s.io/apimachinery/pkg/util/runtime"
+	"k8s.io/client-go/kubernetes"
 	clientgoscheme "k8s.io/client-go/kubernetes/scheme"
 	ctrl "sigs.k8s.io/controller-runtime"
 	"sigs.k8s.io/controller-runtime/pkg/healthz"
@@ -131,18 +132,24 @@ func startOperatorManager(
 		return fmt.Errorf("unable to create the manager: %w", err)
 	}
 
+	clientSet, err := kubernetes.NewForConfig(mgr.GetConfig())
+	if err != nil {
+		return fmt.Errorf("unable to create the clientset client")
+	}
+
 	if err = (&controller.Dash0Reconciler{
-		Client:   mgr.GetClient(),
-		Scheme:   mgr.GetScheme(),
-		Recorder: mgr.GetEventRecorderFor("dash0-controller"),
+		Client:    mgr.GetClient(),
+		ClientSet: clientSet,
+		Scheme:    mgr.GetScheme(),
+		Recorder:  mgr.GetEventRecorderFor("dash0-controller"),
 	}).SetupWithManager(mgr); err != nil {
 		return fmt.Errorf("unable to set up the Dash0 reconciler: %w", err)
 	}
 	setupLog.Info("Dash0 reconciler has been set up.")
 
 	if os.Getenv("ENABLE_WEBHOOKS") != "false" {
-		if err = (&dash0webhook.WebhookHandler{
-			EventRecorder: mgr.GetEventRecorderFor("dash0-webhook"),
+		if err = (&dash0webhook.Handler{
+			Recorder: mgr.GetEventRecorderFor("dash0-webhook"),
 		}).SetupWebhookWithManager(mgr); err != nil {
 			return fmt.Errorf("unable to create the Dash0 webhook: %w", err)
 		}
@@ -153,15 +160,15 @@ func startOperatorManager(
 
 	//+kubebuilder:scaffold:builder
 
-	if err := mgr.AddHealthzCheck("healthz", healthz.Ping); err != nil {
+	if err = mgr.AddHealthzCheck("healthz", healthz.Ping); err != nil {
 		return fmt.Errorf("unable to set up the health check: %w", err)
 	}
-	if err := mgr.AddReadyzCheck("readyz", healthz.Ping); err != nil {
+	if err = mgr.AddReadyzCheck("readyz", healthz.Ping); err != nil {
 		return fmt.Errorf("unable to set up the ready check: %w", err)
 	}
 
 	setupLog.Info("starting manager")
-	if err := mgr.Start(ctrl.SetupSignalHandler()); err != nil {
+	if err = mgr.Start(ctrl.SetupSignalHandler()); err != nil {
 		return fmt.Errorf("unable to set up the signal handler: %w", err)
 	}
 	return nil

config/samples/kustomization.yaml

@@ -0,0 +1,4 @@
+## Append samples of your project ##
+resources:
+- operator_v1alpha1_dash0.yaml
+#+kubebuilder:scaffold:manifestskustomizesamples

config/samples/operator_v1alpha1_dash0.yaml

@@ -0,0 +1,10 @@
+apiVersion: operator.dash0.com/v1alpha1
+kind: Dash0
+metadata:
+  labels:
+    app.kubernetes.io/name: dash0
+    app.kubernetes.io/instance: dash0-sample
+    app.kubernetes.io/part-of: dash0-operator
+    app.kubernetes.io/managed-by: kustomize
+    app.kubernetes.io/created-by: dash0-operator
+  name: dash0-sample

internal/controller/suite_test.go → internal/controller/controller_suite_test.go

@@ -12,8 +12,11 @@ import (
 	. "github.com/onsi/ginkgo/v2"
 	. "github.com/onsi/gomega"
 
+	"k8s.io/client-go/kubernetes"
 	"k8s.io/client-go/kubernetes/scheme"
 	"k8s.io/client-go/rest"
+	"k8s.io/client-go/tools/record"
+	ctrl "sigs.k8s.io/controller-runtime"
 	"sigs.k8s.io/controller-runtime/pkg/client"
 	"sigs.k8s.io/controller-runtime/pkg/envtest"
 	logf "sigs.k8s.io/controller-runtime/pkg/log"
@@ -23,9 +26,13 @@ import (
 	//+kubebuilder:scaffold:imports
 )
 
-var cfg *rest.Config
-var k8sClient client.Client
-var testEnv *envtest.Environment
+var (
+	cfg       *rest.Config
+	k8sClient client.Client
+	clientset *kubernetes.Clientset
+	recorder  record.EventRecorder
+	testEnv   *envtest.Environment
+)
 
 func TestControllers(t *testing.T) {
 	RegisterFailHandler(Fail)
@@ -65,6 +72,16 @@ var _ = BeforeSuite(func() {
 	Expect(err).NotTo(HaveOccurred())
 	Expect(k8sClient).NotTo(BeNil())
 
+	clientset, err = kubernetes.NewForConfig(cfg)
+	Expect(err).NotTo(HaveOccurred())
+	Expect(clientset).NotTo(BeNil())
+
+	mgr, err := ctrl.NewManager(cfg, ctrl.Options{
+		Scheme: scheme.Scheme,
+	})
+	Expect(err).NotTo(HaveOccurred())
+	Expect(mgr).NotTo(BeNil())
+	recorder = mgr.GetEventRecorderFor("dash0-controller")
 })
 
 var _ = AfterSuite(func() {

internal/controller/dash0_controller.go

@@ -5,28 +5,30 @@ package controller
 
 import (
 	"context"
+	"fmt"
 
+	"github.com/go-logr/logr"
 	appsv1 "k8s.io/api/apps/v1"
 	apierrors "k8s.io/apimachinery/pkg/api/errors"
 	"k8s.io/apimachinery/pkg/api/meta"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"k8s.io/apimachinery/pkg/runtime"
+	"k8s.io/client-go/kubernetes"
 	"k8s.io/client-go/tools/record"
 	ctrl "sigs.k8s.io/controller-runtime"
 	"sigs.k8s.io/controller-runtime/pkg/client"
 	"sigs.k8s.io/controller-runtime/pkg/log"
 
 	operatorv1alpha1 "github.com/dash0hq/dash0-operator/api/v1alpha1"
-)
-
-const (
-	conditionTypeAvailable = "Available"
+	"github.com/dash0hq/dash0-operator/internal/k8sresources"
+	. "github.com/dash0hq/dash0-operator/internal/util"
 )
 
 type Dash0Reconciler struct {
 	client.Client
-	Scheme   *runtime.Scheme
-	Recorder record.EventRecorder
+	ClientSet *kubernetes.Clientset
+	Scheme    *runtime.Scheme
+	Recorder  record.EventRecorder
 }
 
 func (r *Dash0Reconciler) SetupWithManager(mgr ctrl.Manager) error {
@@ -59,50 +61,146 @@ func (r *Dash0Reconciler) SetupWithManager(mgr ctrl.Manager) error {
 // - About Controllers: https://kubernetes.io/docs/concepts/architecture/controller/
 // - https://pkg.go.dev/sigs.k8s.io/[email protected]/pkg/reconcile
 func (r *Dash0Reconciler) Reconcile(ctx context.Context, req ctrl.Request) (ctrl.Result, error) {
-	log := log.FromContext(ctx).WithValues("namespace", req.NamespacedName.Namespace, "name", req.NamespacedName.Name)
+	log := log.FromContext(ctx)
+	log.Info("Processig reconcile request")
 
 	// Check whether the Dash0 custom resource exists.
 	dash0CustomResource := &operatorv1alpha1.Dash0{}
 	err := r.Get(ctx, req.NamespacedName, dash0CustomResource)
 	if err != nil {
 		if apierrors.IsNotFound(err) {
-			log.Info("The Dash0 custom resource has not been found, either it hasn't been installed or it has been deleted. Ignoring the reconciliation request.")
+			log.Info("The Dash0 custom resource has not been found, either it hasn't been installed or it has been deleted. Ignoring the reconcile request.")
 			// stop the reconciliation
 			return ctrl.Result{}, nil
 		}
-		log.Error(err, "Failed to get the Dash0 custom resource, requeuing reconciliation request.")
+		log.Error(err, "Failed to get the Dash0 custom resource, requeuing reconcile request.")
 		// requeue the request.
 		return ctrl.Result{}, err
 	}
 
-	if dash0CustomResource.Status.Conditions == nil || len(dash0CustomResource.Status.Conditions) == 0 {
-		// No status is available, assume unknown.
-		meta.SetStatusCondition(&dash0CustomResource.Status.Conditions, metav1.Condition{Type: conditionTypeAvailable, Status: metav1.ConditionUnknown, Reason: "Reconciling", Message: "Starting reconciliation"})
-		if err = r.Status().Update(ctx, dash0CustomResource); err != nil {
-			log.Error(err, "Cannot update the status of the Dash0 custom resource, requeuing reconciliation request.")
-			return ctrl.Result{}, err
-		}
-
-		// Re-fetch the Dash0 custom resource after updating the status. This also helps to avoid triggering
-		// "the object has been modified, please apply your changes to the latest version and try again".
-		if err := r.Get(ctx, req.NamespacedName, dash0CustomResource); err != nil {
-			log.Error(err, "Failed to re-fetch the Dash0 custom resource after updating its status, requeuing reconciliation request.")
-			return ctrl.Result{}, err
-		}
+	isFirstReconcile, err := r.initStatusConditions(ctx, dash0CustomResource, &log)
+	if err != nil {
+		return ctrl.Result{}, err
 	}
 
-	// If a deletion timestamp is set this indicates that the Dash0 custom resource is about to be deleted.
-	isMarkedForDeletion := dash0CustomResource.GetDeletionTimestamp() != nil
+	isMarkedForDeletion := r.handleFinalizers(dash0CustomResource)
 	if isMarkedForDeletion {
+		// Dash0 is marked for deletion, no further reconciliation is necessary.
 		return ctrl.Result{}, nil
 	}
 
-	// TODO inject Dash0 instrumentations into _existing_ resources (later)
+	if isFirstReconcile {
+		r.handleFirstReconcile(ctx, dash0CustomResource, &log)
+	}
 
+	ensureResourceIsMarkedAsAvailable(dash0CustomResource)
 	if err := r.Status().Update(ctx, dash0CustomResource); err != nil {
-		log.Error(err, "Failed to update Dash0 status")
+		log.Error(err, "Failed to update Dash0 status conditions, requeuing reconcile request.")
 		return ctrl.Result{}, err
 	}
 
 	return ctrl.Result{}, nil
 }
+
+func (r *Dash0Reconciler) initStatusConditions(ctx context.Context, dash0CustomResource *operatorv1alpha1.Dash0, log *logr.Logger) (bool, error) {
+	firstReconcile := false
+	needsRefresh := false
+	if dash0CustomResource.Status.Conditions == nil || len(dash0CustomResource.Status.Conditions) == 0 {
+		setAvailableConditionToUnknown(dash0CustomResource)
+		firstReconcile = true
+		needsRefresh = true
+	} else if availableCondition := meta.FindStatusCondition(dash0CustomResource.Status.Conditions, string(operatorv1alpha1.ConditionTypeAvailable)); availableCondition == nil {
+		setAvailableConditionToUnknown(dash0CustomResource)
+		needsRefresh = true
+	}
+	if needsRefresh {
+		err := r.refreshStatus(ctx, dash0CustomResource, log)
+		if err != nil {
+			return firstReconcile, err
+		}
+	}
+	return firstReconcile, nil
+}
+
+func (r *Dash0Reconciler) handleFinalizers(dash0CustomResource *operatorv1alpha1.Dash0) bool {
+	isMarkedForDeletion := dash0CustomResource.GetDeletionTimestamp() != nil
+	// if !isMarkedForDeletion {
+	//   add finalizers here
+	// } else /* if has finalizers */ {
+	//   execute finalizers here
+	// }
+	return isMarkedForDeletion
+}
+
+func (r *Dash0Reconciler) handleFirstReconcile(ctx context.Context, dash0CustomResource *operatorv1alpha1.Dash0, log *logr.Logger) {
+	log.Info("Initial reconcile in progress.")
+	instrumentationEnabled := true
+	instrumentingExistingResourcesEnabled := true
+	if !instrumentationEnabled {
+		log.Info(
+			"Instrumentation is not enabled, neither new nor existing resources will be modified to send telemetry to Dash0.",
+		)
+		return
+	}
+
+	if !instrumentingExistingResourcesEnabled {
+		log.Info(
+			"Instrumenting existing resources is not enabled, only new resources will be modified (at deploy time) to send telemetry to Dash0.",
+		)
+		return
+	}
+
+	log.Info("Modifying existing resources to make them send telemetry to Dash0.")
+	if err := r.modifyExistingResources(ctx, dash0CustomResource); err != nil {
+		log.Error(err, "Modifying existing resources failed.")
+	}
+}
+
+func (r *Dash0Reconciler) refreshStatus(ctx context.Context, dash0CustomResource *operatorv1alpha1.Dash0, log *logr.Logger) error {
+	if err := r.Status().Update(ctx, dash0CustomResource); err != nil {
+		log.Error(err, "Cannot update the status of the Dash0 custom resource, requeuing reconcile request.")
+		return err
+	}
+	return nil
+}
+
+func (r *Dash0Reconciler) modifyExistingResources(ctx context.Context, dash0CustomResource *operatorv1alpha1.Dash0) error {
+	namespace := dash0CustomResource.Namespace
+
+	listOptions := metav1.ListOptions{
+		LabelSelector: fmt.Sprintf("!%s", Dash0AutoInstrumentationLabel),
+	}
+
+	deploymentsInNamespace, err := r.ClientSet.AppsV1().Deployments(namespace).List(ctx, listOptions)
+	if err != nil {
+		return fmt.Errorf("error when querying deployments: %w", err)
+	}
+
+	for _, deployment := range deploymentsInNamespace.Items {
+		logger := log.FromContext(ctx).WithValues("resource type", "deployment", "resource namespace", deployment.GetNamespace(), "resource name", deployment.GetName())
+		operationLabel := "Modifying deployment"
+		err := Retry(operationLabel, func() error {
+			if err := r.Client.Get(ctx, client.ObjectKey{
+				Namespace: deployment.GetNamespace(),
+				Name:      deployment.GetName(),
+			}, &deployment); err != nil {
+				return fmt.Errorf("error when fetching deployment %s/%s: %w", deployment.GetNamespace(), deployment.GetName(), err)
+			}
+			hasBeenModified := k8sresources.ModifyPodSpec(&deployment.Spec.Template.Spec, logger)
+			if hasBeenModified {
+				return r.Client.Update(ctx, &deployment)
+			} else {
+				return nil
+			}
+		}, &logger)
+
+		if err != nil {
+			QueueFailedInstrumentationEvent(r.Recorder, &deployment, "controller", err)
+			return fmt.Errorf("Error when modifying deployment %s/%s: %w", deployment.GetNamespace(), deployment.GetName(), err)
+		} else {
+			QueueSuccessfulInstrumentationEvent(r.Recorder, &deployment, "controller")
+			logger.Info("Added instrumentation to deployment")
+		}
+	}
+	return nil
+}