Elastic APM Server: Set status.ObservedGeneration from metadata.Generation

config/crds/v1/all-crds.yaml

@@ -1462,6 +1462,15 @@ spec:
                 description: KibanaAssociationStatus is the status of any auto-linking
                   to Kibana.
                 type: string
+              observedGeneration:
+                description: ObservedGeneration represents the .metadata.generation
+                  that the status is based upon. It corresponds to the metadata generation,
+                  which is updated on mutation by the API Server. If the generation
+                  observed in status diverges from the generation in metadata, the
+                  APM Server controller has not yet processed the changes contained
+                  in the APM Server specification.
+                format: int64
+                type: integer
               secretTokenSecret:
                 description: SecretTokenSecretName is the name of the Secret that
                   contains the secret token

config/crds/v1/bases/apm.k8s.elastic.co_apmservers.yaml

@@ -7686,6 +7686,15 @@ spec:
                 description: KibanaAssociationStatus is the status of any auto-linking
                   to Kibana.
                 type: string
+              observedGeneration:
+                description: ObservedGeneration represents the .metadata.generation
+                  that the status is based upon. It corresponds to the metadata generation,
+                  which is updated on mutation by the API Server. If the generation
+                  observed in status diverges from the generation in metadata, the
+                  APM Server controller has not yet processed the changes contained
+                  in the APM Server specification.
+                format: int64
+                type: integer
               secretTokenSecret:
                 description: SecretTokenSecretName is the name of the Secret that
                   contains the secret token

deploy/eck-operator/charts/eck-operator-crds/templates/all-crds.yaml

@@ -1474,6 +1474,15 @@ spec:
                 description: KibanaAssociationStatus is the status of any auto-linking
                   to Kibana.
                 type: string
+              observedGeneration:
+                description: ObservedGeneration represents the .metadata.generation
+                  that the status is based upon. It corresponds to the metadata generation,
+                  which is updated on mutation by the API Server. If the generation
+                  observed in status diverges from the generation in metadata, the
+                  APM Server controller has not yet processed the changes contained
+                  in the APM Server specification.
+                format: int64
+                type: integer
               secretTokenSecret:
                 description: SecretTokenSecretName is the name of the Secret that
                   contains the secret token

pkg/apis/apm/v1/apmserver_types.go

@@ -63,14 +63,25 @@ type ApmServerSpec struct {
 // ApmServerStatus defines the observed state of ApmServer
 type ApmServerStatus struct {
 	commonv1.DeploymentStatus `json:",inline"`
+
 	// ExternalService is the name of the service the agents should connect to.
 	ExternalService string `json:"service,omitempty"`
+
 	// SecretTokenSecretName is the name of the Secret that contains the secret token
 	SecretTokenSecretName string `json:"secretTokenSecret,omitempty"`
+
 	// ElasticsearchAssociationStatus is the status of any auto-linking to Elasticsearch clusters.
 	ElasticsearchAssociationStatus commonv1.AssociationStatus `json:"elasticsearchAssociationStatus,omitempty"`
+
 	// KibanaAssociationStatus is the status of any auto-linking to Kibana.
 	KibanaAssociationStatus commonv1.AssociationStatus `json:"kibanaAssociationStatus,omitempty"`
+
+	// ObservedGeneration represents the .metadata.generation that the status is based upon.
+	// It corresponds to the metadata generation, which is updated on mutation by the API Server.
+	// If the generation observed in status diverges from the generation in metadata, the APM Server
+	// controller has not yet processed the changes contained in the APM Server specification.
+	// +kubebuilder:validation:Optional
+	ObservedGeneration int64 `json:"observedGeneration,omitempty"`
 }
 
 // +kubebuilder:object:root=true
@@ -186,6 +197,11 @@ func (as *ApmServer) SetAssociationStatusMap(typ commonv1.AssociationType, statu
 	}
 }
 
+// GetObservedGeneration will return the observedGeneration from the Elastic APM Server's status.
+func (as *ApmServer) GetObservedGeneration() int64 {
+	return as.Status.ObservedGeneration
+}
+
 // ApmEsAssociation helps to manage the APMServer / Elasticsearch association
 type ApmEsAssociation struct {
 	*ApmServer

pkg/controller/apmserver/controller.go

@@ -88,9 +88,8 @@ func Add(mgr manager.Manager, params operator.Parameters) error {
 
 // newReconciler returns a new reconcile.Reconciler
 func newReconciler(mgr manager.Manager, params operator.Parameters) *ReconcileApmServer {
-	client := mgr.GetClient()
 	return &ReconcileApmServer{
-		Client:         client,
+		Client:         mgr.GetClient(),
 		recorder:       mgr.GetEventRecorderFor(controllerName),
 		dynamicWatches: watches.NewDynamicWatches(),
 		Parameters:     params,
@@ -153,14 +152,18 @@ type ReconcileApmServer struct {
 	iteration uint64
 }
 
+// K8sClient returns the kubernetes client from the APM Server reconciler.
 func (r *ReconcileApmServer) K8sClient() k8s.Client {
 	return r.Client
 }
 
+// DynamicWatches returns the set of dynamic watches from the APM Server reconciler.
 func (r *ReconcileApmServer) DynamicWatches() watches.DynamicWatches {
 	return r.dynamicWatches
 }
 
+// Recorder returns the Kubernetes recorder that is responsible for recording and reporting
+// events from the APM Server reconciler.
 func (r *ReconcileApmServer) Recorder() record.EventRecorder {
 	return r.recorder
 }
@@ -200,30 +203,34 @@ func (r *ReconcileApmServer) Reconcile(ctx context.Context, request reconcile.Re
 		return reconcile.Result{}, r.onDelete(k8s.ExtractNamespacedName(&as))
 	}
 
+	results, state := r.doReconcile(ctx, &as)
+
+	return results.WithError(r.updateStatus(ctx, state)).Aggregate()
+}
+
+func (r *ReconcileApmServer) doReconcile(ctx context.Context, as *apmv1.ApmServer) (*reconciler.Results, State) {
+	state := NewState(as)
+	results := reconciler.NewResult(ctx)
+
 	areAssocsConfigured, err := association.AreConfiguredIfSet(as.GetAssociations(), r.recorder)
 	if err != nil {
-		return reconcile.Result{}, tracing.CaptureError(ctx, err)
+		return results.WithError(tracing.CaptureError(ctx, err)), state
 	}
 	if !areAssocsConfigured {
-		return reconcile.Result{}, nil
+		return results, state
 	}
 
-	return r.doReconcile(ctx, request, &as)
-}
-
-func (r *ReconcileApmServer) doReconcile(ctx context.Context, request reconcile.Request, as *apmv1.ApmServer) (reconcile.Result, error) {
 	// Run validation in case the webhook is disabled
 	if err := r.validate(ctx, as); err != nil {
-		return reconcile.Result{}, err
+		return results.WithError(err), state
 	}
 
-	state := NewState(request, as)
 	svc, err := common.ReconcileService(ctx, r.Client, NewService(*as), as)
 	if err != nil {
-		return reconcile.Result{}, err
+		return results.WithError(err), state
 	}
 
-	_, results := certificates.Reconciler{
+	_, results = certificates.Reconciler{
 		K8sClient:             r.K8sClient(),
 		DynamicWatches:        r.DynamicWatches(),
 		Owner:                 as,
@@ -237,22 +244,22 @@ func (r *ReconcileApmServer) doReconcile(ctx context.Context, request reconcile.
 		GarbageCollectSecrets: true,
 	}.ReconcileCAAndHTTPCerts(ctx)
 	if results.HasError() {
-		res, err := results.Aggregate()
+		_, err := results.Aggregate()
 		k8s.EmitErrorEvent(r.recorder, err, as, events.EventReconciliationError, "Certificate reconciliation error: %v", err)
-		return res, err
+		return results, state
 	}
 
 	asVersion, err := version.Parse(as.Spec.Version)
 	if err != nil {
-		return reconcile.Result{}, err
+		return results.WithError(err), state
 	}
 	logger := log.WithValues("namespace", as.Namespace, "as_name", as.Name)
 	assocAllowed, err := association.AllowVersion(asVersion, as, logger, r.recorder)
 	if err != nil {
-		return reconcile.Result{}, tracing.CaptureError(ctx, err)
+		return results.WithError(tracing.CaptureError(ctx, err)), state
 	}
 	if !assocAllowed {
-		return reconcile.Result{}, nil // will eventually retry
+		return results, state // will eventually retry
 	}
 
 	r.warnIfDeprecated(asVersion, as)
@@ -261,23 +268,17 @@ func (r *ReconcileApmServer) doReconcile(ctx context.Context, request reconcile.
 	if err != nil {
 		if apierrors.IsConflict(err) {
 			log.V(1).Info("Conflict while updating status")
-			return reconcile.Result{Requeue: true}, nil
+			return results.WithResult(reconcile.Result{Requeue: true}), state
 		}
 		k8s.EmitErrorEvent(r.recorder, err, as, events.EventReconciliationError, "Deployment reconciliation error: %v", err)
-		return state.Result, tracing.CaptureError(ctx, err)
+		return results.WithError(tracing.CaptureError(ctx, err)), state
 	}
 
 	state.UpdateApmServerExternalService(*svc)
 
-	// update status
-	err = r.updateStatus(ctx, state)
-	if err != nil && apierrors.IsConflict(err) {
-		log.V(1).Info("Conflict while updating status", "namespace", as.Namespace, "as", as.Name)
-		return reconcile.Result{Requeue: true}, nil
-	}
-	res, err := results.WithError(err).Aggregate()
+	_, err = results.WithError(err).Aggregate()
 	k8s.EmitErrorEvent(r.recorder, err, as, events.EventReconciliationError, "Reconciliation error: %v", err)
-	return res, err
+	return results, state
 }
 
 func (r *ReconcileApmServer) warnIfDeprecated(version semver.Version, as *apmv1.ApmServer) {
@@ -345,12 +346,12 @@ func (r *ReconcileApmServer) updateStatus(ctx context.Context, state State) erro
 	span, _ := apm.StartSpan(ctx, "update_status", tracing.SpanTypeApp)
 	defer span.End()
 
-	current := state.originalApmServer
-	if reflect.DeepEqual(current.Status, state.ApmServer.Status) {
+	original := state.originalApmServer
+	if reflect.DeepEqual(original.Status, state.ApmServer.Status) {
 		return nil
 	}
-	if state.ApmServer.Status.IsDegraded(current.Status.DeploymentStatus) {
-		r.recorder.Event(current, corev1.EventTypeWarning, events.EventReasonUnhealthy, "Apm Server health degraded")
+	if state.ApmServer.Status.IsDegraded(original.Status.DeploymentStatus) {
+		r.recorder.Event(original, corev1.EventTypeWarning, events.EventReasonUnhealthy, "Apm Server health degraded")
 	}
 	log.V(1).Info("Updating status",
 		"iteration", atomic.LoadUint64(&r.iteration),
@@ -361,6 +362,7 @@ func (r *ReconcileApmServer) updateStatus(ctx context.Context, state State) erro
 	return common.UpdateStatus(r.Client, state.ApmServer)
 }
 
+// NewService returns the service used by the APM Server.
 func NewService(as apmv1.ApmServer) *corev1.Service {
 	svc := corev1.Service{
 		ObjectMeta: as.Spec.HTTP.Service.ObjectMeta,