feat(triggers): schema-aware match conditions WIP

Pushes CRD OpenAPI schemas from agent to Control Plane and adds the
inventory + match-condition validation pieces on the agent side.

  - clusterdiscovery: extract openAPIV3Schema from CRDs and ship it on
    each ClusterResource (CRDs only; built-in K8s types are still nil).
  - internal/inventory/grpc + controller: gRPC client wrapping the
    AgentInventoryService.PutClusterResources RPC, plus a controller
    that pushes on startup, on a periodic interval (default 1h), and on
    CRD informer events (debounced 5s to coalesce helm-install bursts).
  - api/testtriggers/v1: TestTriggerSpec.Validate now checks each
    match[].path against MatchPathPattern (canonical regex) and
    MatchPathBracketPattern (bracket suffixes are rejected because the
    expression engine in pkg/expressions can't tokenize [*] or [N];
    array-path support is a follow-up).
  - pkg/triggers tests: 32 CRD-shape matcher cases + 19 spec.Validate
    cases + a TestFieldOperatorParity fixture that locks the DTO and
    CRD operator enums in lockstep.
  - internal/inventory/controller tests: 10 cases covering watchable
    filtering, retry on push failure, ctx-cancel cleanup, debounced
    notifier coalescing, and the no-notifier path.

Companion: testkube-cloud-api PR receives + caches + serves the
snapshot, validates match[] against the schema, and surfaces the UI.

Author: dejanzele

api/testtriggers/v1/validation.go

@@ -2,10 +2,23 @@ package v1
 
 import (
 	"fmt"
+	"regexp"
 
 	workflowtriggersv1 "github.com/kubeshop/testkube/api/workflowtriggers/v1"
 )
 
+// MatchPathPattern is the canonical regex for match-condition paths. Exported
+// so cp-api and the UI can pin to the same shape. Accepts dot-separated
+// identifiers, with optional `[*]` or `[N]` suffixes per segment for forward
+// compatibility with array-aware backends.
+var MatchPathPattern = regexp.MustCompile(`^\.[A-Za-z0-9_-]+(\[\*\]|\[\d+\])?(\.[A-Za-z0-9_-]+(\[\*\]|\[\d+\])?)*$`)
+
+// MatchPathBracketPattern detects array-segment suffixes (`[*]` or `[N]`).
+// The matcher's expression engine can't tokenize either today, so paths with
+// bracket suffixes silently no-op at fire time. We reject them at save time
+// to surface the gap loudly. Lift this once pkg/expressions gains support.
+var MatchPathBracketPattern = regexp.MustCompile(`\[\*\]|\[\d+\]`)
+
 // Validate checks the TestTriggerSpec for logical errors that can't be caught
 // by CRD schema validation alone. Called from REST create/update handlers.
 // Currently scoped to the match[] field, mirroring WorkflowTriggerSpec.Validate.
@@ -17,6 +30,12 @@ func (s *TestTriggerSpec) Validate() []error {
 			errs = append(errs, fmt.Errorf("match[%d].path is required", i))
 			continue
 		}
+		if !MatchPathPattern.MatchString(cond.Path) {
+			errs = append(errs, fmt.Errorf("match[%d].path %q is not a valid dot-path (e.g. .status.phase)", i, cond.Path))
+		}
+		if MatchPathBracketPattern.MatchString(cond.Path) {
+			errs = append(errs, fmt.Errorf("match[%d].path %q contains an array index/wildcard ([*] or [N]) — array-path matching is not supported yet; please match on a scalar field", i, cond.Path))
+		}
 
 		switch cond.Operator {
 		case workflowtriggersv1.FieldOperatorEquals,

cmd/api-server/main.go

@@ -19,6 +19,7 @@ import (
 	"google.golang.org/grpc/metadata"
 	"google.golang.org/grpc/status"
 	corev1 "k8s.io/api/core/v1"
+	apiextclient "k8s.io/apiextensions-apiserver/pkg/client/clientset/clientset"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"k8s.io/apimachinery/pkg/runtime"
 	"k8s.io/client-go/dynamic"
@@ -42,6 +43,8 @@ import (
 	intconfig "github.com/kubeshop/testkube/internal/config"
 	"github.com/kubeshop/testkube/internal/cronjob/robfig"
 	cronjobtestworkflow "github.com/kubeshop/testkube/internal/cronjob/testworkflow"
+	inventorycontroller "github.com/kubeshop/testkube/internal/inventory/controller"
+	inventorygrpc "github.com/kubeshop/testkube/internal/inventory/grpc"
 	synccontroller "github.com/kubeshop/testkube/internal/sync/controller"
 	syncgrpc "github.com/kubeshop/testkube/internal/sync/grpc"
 	"github.com/kubeshop/testkube/pkg/agent"
@@ -51,11 +54,11 @@ import (
 	cloudartifacts "github.com/kubeshop/testkube/pkg/cloud/data/artifact"
 	cloudtestworkflow "github.com/kubeshop/testkube/pkg/cloud/data/testworkflow"
 	cloudwebhook "github.com/kubeshop/testkube/pkg/cloud/data/webhook"
+	"github.com/kubeshop/testkube/pkg/clusterdiscovery"
 	"github.com/kubeshop/testkube/pkg/configmap"
 	"github.com/kubeshop/testkube/pkg/controller"
 	"github.com/kubeshop/testkube/pkg/controlplane"
 	"github.com/kubeshop/testkube/pkg/controlplane/scheduling"
-	"github.com/kubeshop/testkube/pkg/clusterdiscovery"
 	"github.com/kubeshop/testkube/pkg/controlplaneclient"
 	"github.com/kubeshop/testkube/pkg/coordination/leader"
 	"github.com/kubeshop/testkube/pkg/cronjob"
@@ -155,6 +158,8 @@ func main() {
 	commons.ExitOnError("creating k8s clientset", err)
 	dynamicClient, err := dynamic.NewForConfig(kubeConfig)
 	commons.ExitOnError("creating k8s dynamic client", err)
+	apiextClient, err := apiextclient.NewForConfig(kubeConfig)
+	commons.ExitOnError("creating k8s apiextensions client", err)
 
 	log.DefaultLogger.Infow("connected to Kubernetes cluster successfully", "namespace", cfg.TestkubeNamespace)
 
@@ -321,6 +326,7 @@ func main() {
 	// This setup can be moved back down to just before the controller initialisation
 	// when the SuperAgent migration has been removed.
 	syncStore := syncgrpc.NewClient(grpcConn, log.DefaultLogger, proContext.APIKey, proContext.OrgID, grpcTLSEnabled)
+	inventoryClient := inventorygrpc.NewClient(grpcConn, log.DefaultLogger, proContext.APIKey, proContext.OrgID, grpcTLSEnabled)
 	// SUPER AGENT DEPRECATION MIGRATION
 	// Run the migration function blocking further processing. We want the migration to run and succeed or to fail and
 	// kill the program before any additional processing occurs to avoid any conflicts with the migration process and
@@ -713,9 +719,25 @@ func main() {
 		testWorkflowExecutor,
 		cfg.ExportArchiveMaxSize,
 	)
-	api.ClusterDiscoverer = clusterdiscovery.New(clientset)
+	api.ClusterDiscoverer = clusterdiscovery.New(clientset).WithSchemas(apiextClient)
 	api.Init(httpServer)
 
+	// Push watchable cluster-resources snapshot to CP on startup + every hour.
+	// Control Plane needs this to render the TestTrigger resourceRef picker —
+	// new architecture: agent pushes, CP caches (see AgentInventoryService).
+	if proContext.APIKey != "" {
+		// Start a CRD informer so adds/updates/deletes trigger an immediate
+		// push to CP. The hourly tick is the safety net for missed events.
+		crdNotifier := inventorycontroller.StartCRDChangeNotifier(ctx, apiextClient, log.DefaultLogger)
+		clusterResourcesController := &inventorycontroller.ClusterResourcesController{
+			Discoverer: api.ClusterDiscoverer,
+			Pusher:     inventoryClient,
+			Notifier:   crdNotifier,
+			Log:        log.DefaultLogger,
+		}
+		g.Go(func() error { return clusterResourcesController.Run(ctx) })
+	}
+
 	log.DefaultLogger.Info("starting agent service")
 
 	shouldRunAgent := shouldRunDefaultAgent(cfg, proContext)

internal/inventory/controller/cluster_resources.go

@@ -0,0 +1,124 @@
+// Package controller runs the Agent-side loop that pushes cluster-environment
+// inventory to the Control Plane. See architecture/new-architecture.md — the
+// Agent is authoritative about what it sees in its cluster; the Control Plane
+// caches a snapshot.
+//
+// Triggers:
+//   - on startup (push immediately after the loop begins)
+//   - periodic ticker (default 1h; covers RBAC drift and missed CRD events)
+//
+// A future extension is a controller-runtime reconciler on CustomResourceDefinition
+// to push immediately when a CRD is installed/removed.
+package controller
+
+import (
+	"context"
+	"time"
+
+	"go.uber.org/zap"
+
+	"github.com/kubeshop/testkube/pkg/api/v1/testkube"
+	"github.com/kubeshop/testkube/pkg/clusterdiscovery"
+)
+
+// ClusterResourcesPusher is the subset of the inventory gRPC client we need.
+type ClusterResourcesPusher interface {
+	PutClusterResources(ctx context.Context, resources []testkube.ClusterResource) error
+}
+
+// ClusterResourcesDiscoverer is the subset of pkg/clusterdiscovery.Discoverer we need.
+type ClusterResourcesDiscoverer interface {
+	List(ctx context.Context) ([]testkube.ClusterResource, error)
+}
+
+// ClusterResourcesController pushes watchable cluster GVKs to the Control Plane
+// on startup, on a periodic interval, and (optionally) when an external
+// Notifier signals a change. Safe to cancel via context.
+type ClusterResourcesController struct {
+	Discoverer ClusterResourcesDiscoverer
+	Pusher     ClusterResourcesPusher
+	Interval   time.Duration
+	// Notifier is fired by external sources (e.g. a CRD informer) when
+	// inventory may have changed. Multiple signals in quick succession are
+	// coalesced into one push via Debounce.
+	Notifier <-chan struct{}
+	// Debounce is the quiet period after the last Notifier event before the
+	// push fires. Defaults to 5s — long enough to coalesce a burst (e.g. helm
+	// install of an operator that registers ten CRDs at once) without delaying
+	// a single-CRD update too noticeably in the UI.
+	Debounce time.Duration
+	Log      *zap.SugaredLogger
+}
+
+// Run blocks until ctx is canceled. Errors are logged but not returned — a
+// transient push failure should not kill the agent; the next tick will retry.
+func (c *ClusterResourcesController) Run(ctx context.Context) error {
+	interval := c.Interval
+	if interval <= 0 {
+		interval = time.Hour
+	}
+	debounce := c.Debounce
+	if debounce <= 0 {
+		debounce = 5 * time.Second
+	}
+
+	c.pushOnce(ctx)
+
+	ticker := time.NewTicker(interval)
+	defer ticker.Stop()
+
+	// debounceC starts nil so the select arm is inactive until the first
+	// notifier event arms the timer. Subsequent events extend the timer.
+	var debounceTimer *time.Timer
+	var debounceC <-chan time.Time
+
+	for {
+		select {
+		case <-ctx.Done():
+			if debounceTimer != nil {
+				debounceTimer.Stop()
+			}
+			return nil
+		case <-ticker.C:
+			c.pushOnce(ctx)
+		case <-c.Notifier:
+			if debounceTimer == nil {
+				debounceTimer = time.NewTimer(debounce)
+				debounceC = debounceTimer.C
+			} else if !debounceTimer.Stop() {
+				// timer was about to fire; drain channel before reset.
+				select {
+				case <-debounceC:
+				default:
+				}
+			}
+			debounceTimer.Reset(debounce)
+		case <-debounceC:
+			debounceTimer = nil
+			debounceC = nil
+			c.pushOnce(ctx)
+		}
+	}
+}
+
+func (c *ClusterResourcesController) pushOnce(ctx context.Context) {
+	resources, err := c.Discoverer.List(ctx)
+	if err != nil {
+		c.Log.Warnw("inventory: cluster discovery failed; skipping push", "error", err)
+		return
+	}
+	watchable := resources[:0]
+	for _, r := range resources {
+		if r.CanWatch {
+			watchable = append(watchable, r)
+		}
+	}
+	if err := c.Pusher.PutClusterResources(ctx, watchable); err != nil {
+		c.Log.Warnw("inventory: push cluster resources to CP failed", "error", err, "count", len(watchable))
+		return
+	}
+	c.Log.Infow("inventory: pushed cluster resources snapshot to CP", "count", len(watchable))
+}
+
+// compile-time check: pkg/clusterdiscovery.Discoverer satisfies our interface.
+var _ ClusterResourcesDiscoverer = (*clusterdiscovery.Discoverer)(nil)