Discovery end2end implementation + tests

.gitignore

@@ -33,4 +33,9 @@ notes/
 docs/public
 docs/resources/_gen/
 docs/.hugo_build.lock
-test/integration/clab-*
+test/integration/**/clab-*
+
+# Only for development and testing purposes
+# To be removed after development of targetsource
+# ignored in order to not add unnecassary logging messages
+lab/dev/resources/targetsources

Makefile

@@ -308,9 +308,10 @@ delete-targetsources-dev-lab: ## Delete the target sources for the development l
 ##@ Testing Lab
 
 .PHONY: run-integration-tests
-run-integration-tests: docker-build undeploy-test-cluster deploy-test-cluster install-test-cluster-dependencies load-test-image deploy install-kubectl install-gnmic install-containerlab deploy-test-topology apply-test-resources
+run-integration-tests: docker-build undeploy-test-cluster deploy-test-cluster install-test-cluster-dependencies load-test-image deploy deploy-test-http-server install-kubectl install-gnmic install-containerlab deploy-test-topology apply-test-resources
 	kubectl wait --for=condition=Ready cluster --all --timeout=180s
 	kubectl wait --for=condition=Ready pipeline --all --timeout=180s
+	kubectl wait --for=jsonpath='{.status.targetsCount}'=3 targetsource --all --timeout=180s
 	kubectl wait --for=jsonpath='{.status.connectionState}'=READY target --all --timeout=180s
 	kubectl get subscriptions -o yaml
 	kubectl get outputs -o yaml

api/v1alpha1/targetsource_types.go

@@ -24,7 +24,8 @@ import (
 // +kubebuilder:validation:Required
 type TargetSourceSpec struct {
 	Provider *ProviderSpec `json:"provider"`
-	//
+
+	// +kubebuilder:validation:Optional
 	TargetLabels map[string]string `json:"targetLabels,omitempty"`
 
 	// +kubebuilder:validation:MinLength=1
@@ -39,7 +40,10 @@ type ProviderSpec struct {
 
 type HTTPConfig struct {
 	// +kubebuilder:validation:MinLength=1
-	URL string `json:"url"`
+	URL   string `json:"url"`
+	Token string `json:"token,omitempty"`
+	// +kubebuilder:validation:Optional
+	AcceptPush bool `json:"acceptPush,omitempty"`
 }
 
 type ConsulConfig struct {
@@ -49,9 +53,10 @@ type ConsulConfig struct {
 
 // TargetSourceStatus defines the observed state of TargetSource
 type TargetSourceStatus struct {
-	Status       string      `json:"status"`
-	TargetsCount int32       `json:"targetsCount"`
-	LastSync     metav1.Time `json:"lastSync"`
+	Status             string      `json:"status,omitempty"`
+	ObservedGeneration int64       `json:"observedGeneration"`
+	TargetsCount       int32       `json:"targetsCount,omitempty"`
+	LastSync           metav1.Time `json:"lastSync,omitempty"`
 }
 
 //+kubebuilder:object:root=true

cmd/main.go

@@ -28,6 +28,7 @@ import (
 
 	certmanagerv1 "github.com/cert-manager/cert-manager/pkg/apis/certmanager/v1"
 	"k8s.io/apimachinery/pkg/runtime"
+	"k8s.io/apimachinery/pkg/types"
 	utilruntime "k8s.io/apimachinery/pkg/util/runtime"
 	clientgoscheme "k8s.io/client-go/kubernetes/scheme"
 	ctrl "sigs.k8s.io/controller-runtime"
@@ -40,6 +41,8 @@ import (
 	operatorv1alpha1 "github.com/gnmic/operator/api/v1alpha1"
 	"github.com/gnmic/operator/internal/apiserver"
 	"github.com/gnmic/operator/internal/controller"
+	"github.com/gnmic/operator/internal/controller/discovery"
+	"github.com/gnmic/operator/internal/controller/discovery/core"
 	webhookv1alpha1 "github.com/gnmic/operator/internal/webhook/v1alpha1"
 	//+kubebuilder:scaffold:imports
 )
@@ -64,13 +67,17 @@ func main() {
 	var probeAddr string
 	var devMode bool
 	var apiAddr string
+	var discoveryChunkSize int
+	var discoveryBufferSize int
 	flag.StringVar(&apiAddr, "api-bind-address", "", "The address the operator API endpoint binds to. Disabled if empty.")
 	flag.BoolVar(&devMode, "dev-mode", false, "Enable development mode.")
 	flag.StringVar(&metricsAddr, "metrics-bind-address", ":8080", "The address the metric endpoint binds to.")
 	flag.StringVar(&probeAddr, "health-probe-bind-address", ":8081", "The address the probe endpoint binds to.")
 	flag.BoolVar(&enableLeaderElection, "leader-elect", false,
 		"Enable leader election for controller manager. "+
 			"Enabling this will ensure there is only one active controller manager.")
+	flag.IntVar(&discoveryChunkSize, "discovery-chunk-size", 100, "Maximum number of targets/events sent in a single discovery message.")
+	flag.IntVar(&discoveryBufferSize, "discovery-buffer-size", 10, "Amount of discovery messages that can be queued in the channel buffer.")
 	opts := zap.Options{
 		Development: devMode,
 	}
@@ -79,6 +86,8 @@ func main() {
 
 	ctrl.SetLogger(zap.New(zap.UseFlagOptions(&opts)))
 
+	discoveryRegistry := discovery.NewRegistry[types.NamespacedName, core.DiscoveryRegistryValue]()
+
 	mgr, err := ctrl.NewManager(ctrl.GetConfigOrDie(), ctrl.Options{
 		Scheme:                 scheme,
 		Metrics:                metricsserver.Options{BindAddress: metricsAddr},
@@ -117,8 +126,11 @@ func main() {
 		os.Exit(1)
 	}
 	if err := (&controller.TargetSourceReconciler{
-		Client: mgr.GetClient(),
-		Scheme: mgr.GetScheme(),
+		Client:            mgr.GetClient(),
+		Scheme:            mgr.GetScheme(),
+		BufferSize:        discoveryBufferSize,
+		ChunkSize:         discoveryChunkSize,
+		DiscoveryRegistry: discoveryRegistry,
 	}).SetupWithManager(mgr); err != nil {
 		setupLog.Error(err, "unable to create controller", "controller", "TargetSource")
 		os.Exit(1)
@@ -220,6 +232,7 @@ func main() {
 
 	if apiAddr != "" {
 		apiServer := apiserver.New(apiAddr, clusterReconciler)
+		apiServer.DiscoveryRegistry = discoveryRegistry
 		err = mgr.Add(manager.RunnableFunc(func(ctx context.Context) error {
 			errCh := make(chan error)
 			go func() {

config/crd/bases/operator.gnmic.dev_targetsources.yaml

@@ -49,6 +49,10 @@ spec:
                     type: object
                   http:
                     properties:
+                      acceptPush:
+                        type: boolean
+                      token:
+                        type: string
                       url:
                         minLength: 1
                         type: string
@@ -77,15 +81,16 @@ spec:
               lastSync:
                 format: date-time
                 type: string
+              observedGeneration:
+                format: int64
+                type: integer
               status:
                 type: string
               targetsCount:
                 format: int32
                 type: integer
             required:
-            - lastSync
-            - status
-            - targetsCount
+            - observedGeneration
             type: object
         type: object
     served: true

go.mod

@@ -5,6 +5,7 @@ go 1.25.5
 require (
 	github.com/cert-manager/cert-manager v1.19.3
 	github.com/go-logr/logr v1.4.3
+	github.com/google/uuid v1.6.0
 	github.com/onsi/ginkgo/v2 v2.27.3
 	github.com/onsi/gomega v1.38.3
 	github.com/openconfig/gnmic/pkg/api v0.1.10
@@ -47,7 +48,6 @@ require (
 	github.com/google/gnostic-models v0.7.1 // indirect
 	github.com/google/go-cmp v0.7.0 // indirect
 	github.com/google/pprof v0.0.0-20251213031049-b05bdaca462f // indirect
-	github.com/google/uuid v1.6.0 // indirect
 	github.com/json-iterator/go v1.1.12 // indirect
 	github.com/modern-go/concurrent v0.0.0-20180306012644-bacd9c7ef1dd // indirect
 	github.com/modern-go/reflect2 v1.0.3-0.20250322232337-35a7c28c31ee // indirect

internal/apiserver/apiserver.go

@@ -5,11 +5,16 @@ import (
 	"net/http"
 
 	"github.com/gnmic/operator/internal/controller"
+	"github.com/gnmic/operator/internal/controller/discovery"
+	"github.com/gnmic/operator/internal/controller/discovery/core"
+	"k8s.io/apimachinery/pkg/types"
 )
 
 type APIServer struct {
 	Server            *http.Server
 	clusterReconciler *controller.ClusterReconciler
+
+	DiscoveryRegistry *discovery.Registry[types.NamespacedName, core.DiscoveryRegistryValue]
 }
 
 func New(addr string, clusterReconciler *controller.ClusterReconciler) *APIServer {

internal/controller/const.go

@@ -21,6 +21,8 @@ const (
 	LabelCertType            = "operator.gnmic.dev/cert-type"
 	LabelValueCertTypeClient = "client"
 	LabelValueCertTypeTunnel = "tunnel"
+
+	LabelTargetSourceFinalizer = "operator.gnmic.dev/targetsource-finalizer"
 )
 
 const (

internal/controller/discovery/client.go

@@ -1,22 +1,28 @@
 package discovery
 
-// File may become obsolete, depends on how the logic to compare desired vs. existing state will get implemented
-
 import (
 	"context"
 
+	apierrors "k8s.io/apimachinery/pkg/api/errors"
+	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"
 
 	gnmicv1alpha1 "github.com/gnmic/operator/api/v1alpha1"
 )
 
-func FetchExistingTargets(ctx context.Context, c client.Client, ts gnmicv1alpha1.TargetSource) ([]gnmicv1alpha1.Target, error) {
+func fetchExistingTargets(ctx context.Context, c client.Client, ts *gnmicv1alpha1.TargetSource) ([]gnmicv1alpha1.Target, error) {
 	var targetList gnmicv1alpha1.TargetList
 
-	err := c.List(ctx, &targetList,
+	err := c.List(
+		ctx,
+		&targetList,
 		client.InNamespace(ts.Namespace),
 		client.MatchingLabels{
-			"gnmic.io/source": ts.Name,
+			LabelTargetSourceName: ts.Name,
 		},
 	)
 	if err != nil {
@@ -25,3 +31,59 @@ func FetchExistingTargets(ctx context.Context, c client.Client, ts gnmicv1alpha1
 
 	return targetList.Items, nil
 }
+
+func applyTarget(ctx context.Context, c client.Client, s *runtime.Scheme, desired *gnmicv1alpha1.Target, ts *gnmicv1alpha1.TargetSource) error {
+	existing := &gnmicv1alpha1.Target{
+		ObjectMeta: metav1.ObjectMeta{
+			Name:      desired.Name,
+			Namespace: desired.Namespace,
+		},
+	}
+
+	_, err := controllerutil.CreateOrUpdate(ctx, c, existing, func() error {
+		existing.Spec = desired.Spec
+		existing.Labels = desired.Labels
+
+		return controllerutil.SetControllerReference(ts, existing, s)
+	})
+
+	return err
+}
+
+func deleteTarget(ctx context.Context, c client.Client, name string, namespace string) error {
+	existing := &gnmicv1alpha1.Target{}
+
+	err := c.Get(ctx, types.NamespacedName{
+		Name:      name,
+		Namespace: namespace,
+	}, existing)
+	if apierrors.IsNotFound(err) {
+		return nil
+	} else if err != nil {
+		return err
+	}
+
+	err = c.Delete(ctx, existing)
+	if apierrors.IsNotFound(err) {
+		return nil
+	}
+
+	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
+}

internal/controller/discovery/const.go

@@ -0,0 +1,13 @@
+package discovery
+
+const (
+	// Kubernetes Side Labels
+	LabelTargetSourceName = "operator.gnmic.dev/targetsource"
+)
+
+const (
+	// Prefix and Labels for external systems
+	ExternalLabelPrefix = "gnmic_operator_"
+
+	ExternalLabelTargetProfile = ExternalLabelPrefix + "target_profile"
+)

internal/controller/discovery/core/loader_interface.go

@@ -12,12 +12,7 @@ type Loader interface {
 	// Name returns the unique loader identifier e.g. "pull"
 	Name() string
 
-	// Start begins discovery and pushes target snapshots into the out channel
-	// The loader must stop cleanly when ctx is cancelled
-	Start(
-		ctx context.Context,
-		targetsourceName string,
-		spec gnmicv1alpha1.TargetSourceSpec,
-		out chan<- []DiscoveryMessage,
-	) error
+	// Run begins discovery and pushes target snapshots or events into the out channel
+	// The loader must stop cleanly when ctx is canceled
+	Run(ctx context.Context, out chan<- []DiscoveryMessage, spec gnmicv1alpha1.TargetSourceSpec) error
 }

internal/controller/discovery/core/message_interface.go

@@ -0,0 +1,8 @@
+package core
+
+type DiscoveryMessage interface {
+	isDiscoveryMessage()
+}
+
+func (DiscoveryEvent) isDiscoveryMessage()    {}
+func (DiscoverySnapshot) isDiscoveryMessage() {}