TargetLoader + TargetReconciler architecture

.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

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
@@ -40,6 +41,8 @@ type ProviderSpec struct {
 type HTTPConfig struct {
 	// +kubebuilder:validation:MinLength=1
 	URL string `json:"url"`
+	// +kubebuilder:validation:Optional
+	AcceptPush bool `json:"acceptPush,omitempty"`
 }
 
 type ConsulConfig struct {

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/core"
+	"github.com/gnmic/operator/internal/controller/discovery"
 	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,8 @@ spec:
                     type: object
                   http:
                     properties:
+                      acceptPush:
+                        type: boolean
                       url:
                         minLength: 1
                         type: string

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,7 +1,5 @@
 package discovery
 
-// File may become obsolete, depends on how the logic to compare desired vs. existing state will get implemented
-
 import (
 	"context"
 
@@ -10,13 +8,20 @@ import (
 	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 {

internal/controller/discovery/const.go

@@ -0,0 +1,6 @@
+package discovery
+
+const (
+	// Labels
+	LabelTargetSourceName = "operator.gnmic.dev/targetsource"
+)

internal/controller/discovery/core/loader_interface.go

@@ -2,8 +2,6 @@ package core
 
 import (
 	"context"
-
-	gnmicv1alpha1 "github.com/gnmic/operator/api/v1alpha1"
 )
 
 // Loader defines a pluggable TargetSource loader interface
@@ -12,12 +10,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) error
 }

internal/controller/discovery/core/message_interface.go

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

internal/controller/discovery/core/types.go

@@ -1,5 +1,39 @@
 package core
 
+import (
+	"context"
+
+	"k8s.io/apimachinery/pkg/types"
+)
+
+// DiscoveryRegistryValue represents the controller-owned runtime state
+// with its configuration for a single TargetSource
+type DiscoveryRegistryValue struct {
+	// Channel is the outbound communication channel used by discovery
+	// components (loaders, webhooks, etc.) to emit discovery messages
+	Channel chan<- []DiscoveryMessage
+	// Stop cancels the discovery context associated with this registry entry
+	Stop context.CancelFunc
+
+	CommonLoaderConfig *CommonLoaderConfig
+}
+
+type CommonLoaderConfig struct {
+	TargetsourceNN types.NamespacedName
+	ChunkSize      int
+	AcceptPush     bool
+}
+
+// EventAction represents the type of a discovery event
+type EventAction int
+
+const (
+	// EventDelete indicates that a target should be removed
+	EventDelete EventAction = iota
+	// EventApply indicates that a target should be applied (created or updated)
+	EventApply
+)
+
 // DiscoveredTarget represents a target discovered from an external source
 // before it is materialized as a Kubernetes Target CR
 type DiscoveredTarget struct {
@@ -8,15 +42,14 @@ type DiscoveredTarget struct {
 	Labels  map[string]string
 }
 
-const (
-	DELETE DiscoveryEvent = 0
-	CREATE DiscoveryEvent = 1
-	UPDATE DiscoveryEvent = 2
-)
-
-type DiscoveryEvent int
-
-type DiscoveryMessage struct {
+type DiscoveryEvent struct {
 	Target DiscoveredTarget
-	Event  DiscoveryEvent
+	Event  EventAction
+}
+
+type DiscoverySnapshot struct {
+	SnapshotID  string
+	ChunkIndex  int
+	TotalChunks int
+	Targets     []DiscoveredTarget
 }

internal/controller/discovery/discovery.go

@@ -0,0 +1,15 @@
+package discovery
+
+// Package discovery implements the discovery runtime subsystem.
+//
+// The discovery subsystem is responsible for:
+// - Receiving discovery data from external providers (loaders, webhooks).
+// - Applying discovered state to Kubernetes Targets.
+//
+// The package is structured into the following subpackages:
+// - core: message contracts, snapshot/event types, and transport helpers.
+// - message processor: snapshot + event target state application logic.
+// - loaders: target discovery providers (HTTP, webhook, etc.).
+// - registry: key -> channel registry.
+//
+// At the moment, the targetsource controller imports specific subpackages explicitly.

internal/controller/discovery/loaders.go

@@ -0,0 +1,24 @@
+package discovery
+
+import (
+	"fmt"
+
+	gnmicv1alpha1 "github.com/gnmic/operator/api/v1alpha1"
+	"github.com/gnmic/operator/internal/controller/discovery/core"
+	http "github.com/gnmic/operator/internal/controller/discovery/loaders/http"
+)
+
+// NewLoader creates a loader by name
+func NewLoader(cfg *core.CommonLoaderConfig, spec gnmicv1alpha1.TargetSourceSpec) (core.Loader, error) {
+
+	switch {
+	case spec.Provider.HTTP != nil:
+		cfg.AcceptPush = spec.Provider.HTTP.AcceptPush
+		return http.New(*cfg), nil
+	case spec.Provider.Consul != nil:
+		return nil, fmt.Errorf("Unimplemented targetsource provider, check TargetSource CRD for %s", cfg.TargetsourceNN)
+	default:
+		return nil, fmt.Errorf("unknown targetsource provider, check TargetSource CRD for %s", cfg.TargetsourceNN)
+	}
+
+}