diff --git a/api/v1alpha1/targetsource_types.go b/api/v1alpha1/targetsource_types.go index 27c28e1..917a30d 100644 --- a/api/v1alpha1/targetsource_types.go +++ b/api/v1alpha1/targetsource_types.go @@ -401,6 +401,13 @@ type TargetSourceStatus struct { ObservedGeneration int64 `json:"observedGeneration"` TargetsCount int32 `json:"targetsCount,omitempty"` LastSync metav1.Time `json:"lastSync,omitempty"` + + // Conditions represent the latest available observations of the + // TargetSource's state. + // +listType=map + // +listMapKey=type + // +optional + Conditions []metav1.Condition `json:"conditions,omitempty"` } //+kubebuilder:object:root=true diff --git a/api/v1alpha1/zz_generated.deepcopy.go b/api/v1alpha1/zz_generated.deepcopy.go index 55495a3..8c6d682 100644 --- a/api/v1alpha1/zz_generated.deepcopy.go +++ b/api/v1alpha1/zz_generated.deepcopy.go @@ -1517,6 +1517,13 @@ func (in *TargetSourceSpec) DeepCopy() *TargetSourceSpec { func (in *TargetSourceStatus) DeepCopyInto(out *TargetSourceStatus) { *out = *in in.LastSync.DeepCopyInto(&out.LastSync) + if in.Conditions != nil { + in, out := &in.Conditions, &out.Conditions + *out = make([]metav1.Condition, len(*in)) + for i := range *in { + (*in)[i].DeepCopyInto(&(*out)[i]) + } + } } // DeepCopy is an autogenerated deepcopy function, copying the receiver, creating a new TargetSourceStatus. diff --git a/config/crd/bases/operator.gnmic.dev_targetsources.yaml b/config/crd/bases/operator.gnmic.dev_targetsources.yaml index f9e3248..1d9414b 100644 --- a/config/crd/bases/operator.gnmic.dev_targetsources.yaml +++ b/config/crd/bases/operator.gnmic.dev_targetsources.yaml @@ -451,6 +451,68 @@ spec: status: description: TargetSourceStatus defines the observed state of TargetSource properties: + conditions: + description: |- + Conditions represent the latest available observations of the + TargetSource's state. + items: + description: Condition contains details for one aspect of the current + state of this API Resource. + properties: + lastTransitionTime: + description: |- + lastTransitionTime is the last time the condition transitioned from one status to another. + This should be when the underlying condition changed. If that is not known, then using the time when the API field changed is acceptable. + format: date-time + type: string + message: + description: |- + message is a human readable message indicating details about the transition. + This may be an empty string. + maxLength: 32768 + type: string + observedGeneration: + description: |- + observedGeneration represents the .metadata.generation that the condition was set based upon. + For instance, if .metadata.generation is currently 12, but the .status.conditions[x].observedGeneration is 9, the condition is out of date + with respect to the current state of the instance. + format: int64 + minimum: 0 + type: integer + reason: + description: |- + reason contains a programmatic identifier indicating the reason for the condition's last transition. + Producers of specific condition types may define expected values and meanings for this field, + and whether the values are considered a guaranteed API. + The value should be a CamelCase string. + This field may not be empty. + maxLength: 1024 + minLength: 1 + pattern: ^[A-Za-z]([A-Za-z0-9_,:]*[A-Za-z0-9_])?$ + type: string + status: + description: status of the condition, one of True, False, Unknown. + enum: + - "True" + - "False" + - Unknown + type: string + type: + description: type of condition in CamelCase or in foo.example.com/CamelCase. + maxLength: 316 + pattern: ^([a-z0-9]([-a-z0-9]*[a-z0-9])?(\.[a-z0-9]([-a-z0-9]*[a-z0-9])?)*/)?(([A-Za-z0-9][-A-Za-z0-9_.]*)?[A-Za-z0-9])$ + type: string + required: + - lastTransitionTime + - message + - reason + - status + - type + type: object + type: array + x-kubernetes-list-map-keys: + - type + x-kubernetes-list-type: map lastSync: format: date-time type: string diff --git a/internal/controller/discovery/client.go b/internal/controller/discovery/client.go index 169166a..bc7b38b 100644 --- a/internal/controller/discovery/client.go +++ b/internal/controller/discovery/client.go @@ -9,7 +9,6 @@ import ( metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" "k8s.io/apimachinery/pkg/runtime" "k8s.io/apimachinery/pkg/types" - "k8s.io/client-go/util/retry" "sigs.k8s.io/controller-runtime/pkg/client" "sigs.k8s.io/controller-runtime/pkg/controller/controllerutil" @@ -78,23 +77,6 @@ func deleteTarget(ctx context.Context, c client.Client, name string, namespace s return err } -// updateTargetSourceStatus updates the status of the TargetSource Object ts. The only fields updated are targetCount and LastSync, which takes the current timestamp. -func updateTargetSourceStatus(ctx context.Context, c client.Client, ts *gnmicv1alpha1.TargetSource, targetCount int32) error { - err := retry.RetryOnConflict(retry.DefaultRetry, func() error { - latest := &gnmicv1alpha1.TargetSource{} - if err := c.Get(ctx, client.ObjectKeyFromObject(ts), latest); err != nil { - return err - } - - latest.Status.TargetsCount = targetCount - latest.Status.LastSync = metav1.Now() - - return c.Status().Update(ctx, latest) - }) - - return err -} - // Helper: GetSecretValues returns values from a secret // If keys are provided -> returns only those keys // If keys is empty -> returns entire secret data diff --git a/internal/controller/discovery/core/status_updater_interface.go b/internal/controller/discovery/core/status_updater_interface.go new file mode 100644 index 0000000..35a7c3d --- /dev/null +++ b/internal/controller/discovery/core/status_updater_interface.go @@ -0,0 +1,33 @@ +package core + +import ( + "context" + + metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" +) + +const ( + ConditionTypeReady = "Ready" + ConditionTypeReconciling = "Reconciling" + ConditionTypeDegraded = "Degraded" + ConditionTypeStalled = "Stalled" + + ReasonWaitingForSync Reason = "WaitingForSync" + ReasonSyncStarted Reason = "SyncStarted" + ReasonSyncSucceeded Reason = "SyncSucceeded" + ReasonSyncCompleted Reason = "SyncCompleted" + ReasonSyncWithErrors Reason = "SyncSucceededWithErrors" + ReasonSyncFailed Reason = "SyncFailed" +) + +type Reason string + +type StatusUpdate struct { + Conditions []metav1.Condition + TargetsCount *int32 +} + +// StatusUpdater defines the interface for TargetLoaders and MessageProcessor to update the status of the TargetSource +type StatusUpdater interface { + UpdateStatus(context.Context, StatusUpdate) error +} diff --git a/internal/controller/discovery/core/types.go b/internal/controller/discovery/core/types.go index 69bf65e..ac9c87c 100644 --- a/internal/controller/discovery/core/types.go +++ b/internal/controller/discovery/core/types.go @@ -26,6 +26,7 @@ type CommonLoaderConfig struct { PushConfig *v1alpha1.PushSpec Router *gin.Engine ResourceFetcher ResourceFetcher + Updater StatusUpdater } // EventAction represents the type of a discovery event diff --git a/internal/controller/discovery/loaders/http/loader.go b/internal/controller/discovery/loaders/http/loader.go index 4ad3861..618f37a 100644 --- a/internal/controller/discovery/loaders/http/loader.go +++ b/internal/controller/discovery/loaders/http/loader.go @@ -13,6 +13,7 @@ import ( "github.com/go-logr/logr" "github.com/google/uuid" + metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" "sigs.k8s.io/controller-runtime/pkg/log" gnmicv1alpha1 "github.com/gnmic/operator/api/v1alpha1" @@ -39,6 +40,17 @@ func (l *Loader) Name() string { return "http" } +// reportStatus emits a status update through the configured StatusUpdater, +// if one is set. It is a no-op when no updater is configured (e.g. in tests). +func (l *Loader) reportStatus(ctx context.Context, update core.StatusUpdate) { + if l.loaderCfg.Updater == nil { + return + } + if err := l.loaderCfg.Updater.UpdateStatus(ctx, update); err != nil { + log.FromContext(ctx).Error(err, "failed to update TargetSource status") + } +} + // Run starts the HTTP discovery loop // It performs an immediate fetch and then continues polling at a fixed interval func (l *Loader) Run(ctx context.Context, out chan<- []core.DiscoveryMessage) error { @@ -79,6 +91,17 @@ func (l *Loader) Run(ctx context.Context, out chan<- []core.DiscoveryMessage) er // helper function to fetch targets and emit discovery messages fetchAndEmit := func() { + l.reportStatus(ctx, core.StatusUpdate{ + Conditions: []metav1.Condition{ + { + Type: core.ConditionTypeReconciling, + Status: metav1.ConditionTrue, + Reason: string(core.ReasonSyncStarted), + Message: "Fetching targets from HTTP endpoint", + }, + }, + }) + // Fetch targets from HTTP endpoint targets, err := l.fetchTargetsFromHTTPEndpoint(ctx, client, logger) if err != nil { @@ -87,6 +110,16 @@ func (l *Loader) Run(ctx context.Context, out chan<- []core.DiscoveryMessage) er "Failed to fetch targets from HTTP endpoint", "url", l.spec.URL, ) + l.reportStatus(ctx, core.StatusUpdate{ + Conditions: []metav1.Condition{ + { + Type: core.ConditionTypeStalled, + Status: metav1.ConditionTrue, + Reason: string(core.ReasonSyncFailed), + Message: err.Error(), + }, + }, + }) return } diff --git a/internal/controller/discovery/message_processor.go b/internal/controller/discovery/message_processor.go index f573b1b..cd4519d 100644 --- a/internal/controller/discovery/message_processor.go +++ b/internal/controller/discovery/message_processor.go @@ -4,6 +4,7 @@ import ( "context" "fmt" + metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" "k8s.io/apimachinery/pkg/runtime" "sigs.k8s.io/controller-runtime/pkg/client" "sigs.k8s.io/controller-runtime/pkg/log" @@ -31,15 +32,17 @@ type MessageProcessor struct { // Events are deferred while snapshot is in progress deferredEvents []core.DiscoveryEvent targetCount int32 + updater core.StatusUpdater } // NewMessageProcessor wires a MessageProcessor instance -func NewMessageProcessor(c client.Client, s *runtime.Scheme, ts *gnmicv1alpha1.TargetSource, in <-chan []core.DiscoveryMessage) *MessageProcessor { +func NewMessageProcessor(c client.Client, s *runtime.Scheme, ts *gnmicv1alpha1.TargetSource, in <-chan []core.DiscoveryMessage, u core.StatusUpdater) *MessageProcessor { return &MessageProcessor{ client: c, scheme: s, targetSource: ts, in: in, + updater: u, } } @@ -366,7 +369,22 @@ func (m *MessageProcessor) applyEvent(ctx context.Context, event core.DiscoveryE } func (m *MessageProcessor) updateStatus(ctx context.Context, logger logr.Logger) { - if err := updateTargetSourceStatus(ctx, m.client, m.targetSource, m.targetCount); err != nil { + if m.updater == nil { + return + } + count := m.targetCount + update := core.StatusUpdate{ + Conditions: []metav1.Condition{ + { + Type: core.ConditionTypeReady, + Status: metav1.ConditionTrue, + Reason: string(core.ReasonSyncSucceeded), + Message: "Successfully synced targets", + }, + }, + TargetsCount: &count, + } + if err := m.updater.UpdateStatus(ctx, update); err != nil { logger.Error(err, "error updating TargetSource status") } else { logger.Info("updated target source status", diff --git a/internal/controller/discovery/message_processor_test.go b/internal/controller/discovery/message_processor_test.go index 6e805ad..67eab29 100644 --- a/internal/controller/discovery/message_processor_test.go +++ b/internal/controller/discovery/message_processor_test.go @@ -39,6 +39,7 @@ func mockMessageProcessor(opts ...func(*MessageProcessor)) *MessageProcessor { scheme, &targetSource, targetChannel, + nil, ) for _, opt := range opts { diff --git a/internal/controller/discovery/status_updater_client.go b/internal/controller/discovery/status_updater_client.go new file mode 100644 index 0000000..1642536 --- /dev/null +++ b/internal/controller/discovery/status_updater_client.go @@ -0,0 +1,69 @@ +package discovery + +import ( + "context" + + metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" + "k8s.io/apimachinery/pkg/runtime" + "k8s.io/client-go/util/retry" + "sigs.k8s.io/controller-runtime/pkg/client" + + gnmicv1alpha1 "github.com/gnmic/operator/api/v1alpha1" + "github.com/gnmic/operator/internal/controller/discovery/core" +) + +// k8sStatusUpdater is a client which fulfills the StatusUpdater interface +type k8sStatusUpdater struct { + client client.Client + scheme *runtime.Scheme + targetSource *gnmicv1alpha1.TargetSource +} + +// Returns an instance of k8sStatusUpdater +func NewK8sStatusUpdater(c client.Client, s *runtime.Scheme, ts *gnmicv1alpha1.TargetSource) *k8sStatusUpdater { + return &k8sStatusUpdater{ + client: c, + scheme: s, + targetSource: ts, + } +} + +// UpdateStatus takes a StatusUpdate holding Conditions and a pointer referencing the TargetsCount. +// If TargetsCount is set, the LastSync time gets set to metav1.Now(). +// Replaces LastTransitionTime of each Condition with metav1.Now(). +func (c *k8sStatusUpdater) UpdateStatus(ctx context.Context, update core.StatusUpdate) error { + + return c.patchStatus(ctx, func( + ts *gnmicv1alpha1.TargetSource, + ) { + now := metav1.Now() + + // Update status fields: Replace all Conditions and set TargetsCount and LastSync if pointer != nil + for i := range update.Conditions { + update.Conditions[i].LastTransitionTime = now + } + ts.Status.Conditions = update.Conditions + + if update.TargetsCount != nil { + ts.Status.TargetsCount = *update.TargetsCount + ts.Status.LastSync = now + } + }) +} + +// patchStatus is an internal function to update the Kubernetes object +func (c *k8sStatusUpdater) patchStatus(ctx context.Context, mutate func(*gnmicv1alpha1.TargetSource)) error { + err := retry.RetryOnConflict(retry.DefaultRetry, func() error { + latest := &gnmicv1alpha1.TargetSource{} + if err := c.client.Get(ctx, client.ObjectKeyFromObject(c.targetSource), latest); err != nil { + return err + } + + patch := client.MergeFrom(latest.DeepCopy()) + mutate(latest) + + return c.client.Status().Patch(ctx, latest, patch) + }) + + return err +} diff --git a/internal/controller/targetsource_controller.go b/internal/controller/targetsource_controller.go index 59394db..5f39fd2 100644 --- a/internal/controller/targetsource_controller.go +++ b/internal/controller/targetsource_controller.go @@ -192,9 +192,11 @@ func (r *TargetSourceReconciler) startDiscovery( ) error { targetChannel := make(chan []discoveryTypes.DiscoveryMessage, r.BufferSize) ctx, cancel := context.WithCancel(context.Background()) + statusUpdater := discovery.NewK8sStatusUpdater(r.Client, r.Scheme, targetSource) loaderConfig := discoveryTypes.CommonLoaderConfig{ TargetsourceNN: key, ChunkSize: r.ChunkSize, + Updater: statusUpdater, } // Cleanup function to cleanup discovery runtime of targetsource @@ -208,6 +210,7 @@ func (r *TargetSourceReconciler) startDiscovery( r.Scheme, targetSource, targetChannel, + statusUpdater, ) loader, err := discovery.NewLoader(reconcileCtx, r.Client, &loaderConfig, targetSource.Spec) if err != nil {