diff --git a/go/core/internal/controller/predicates/discovery_disabled.go b/go/core/internal/controller/predicates/discovery_disabled.go index e9bd0034ee..3baba11c21 100644 --- a/go/core/internal/controller/predicates/discovery_disabled.go +++ b/go/core/internal/controller/predicates/discovery_disabled.go @@ -20,9 +20,9 @@ import ( "sigs.k8s.io/controller-runtime/pkg/client" "sigs.k8s.io/controller-runtime/pkg/event" "sigs.k8s.io/controller-runtime/pkg/predicate" -) -const discoveryLabel = "kagent.dev/discovery" + "github.com/kagent-dev/kagent/go/core/pkg/consts" +) // DiscoveryDisabledPredicate filters out resources with the discovery label set to disabled type DiscoveryDisabledPredicate struct { @@ -55,6 +55,6 @@ func isDiscoveryDisabled(obj client.Object) bool { return false } - discoveryLabelValue, exists := labels[discoveryLabel] - return exists && discoveryLabelValue == "disabled" + discoveryLabelValue, exists := labels[consts.DiscoveryLabel] + return exists && discoveryLabelValue == consts.DiscoveryDisabled } diff --git a/go/core/internal/controller/reconciler/reconciler.go b/go/core/internal/controller/reconciler/reconciler.go index de07032b36..207be9f1e9 100644 --- a/go/core/internal/controller/reconciler/reconciler.go +++ b/go/core/internal/controller/reconciler/reconciler.go @@ -37,6 +37,7 @@ import ( agent_translator "github.com/kagent-dev/kagent/go/core/internal/controller/translator/agent" "github.com/kagent-dev/kagent/go/core/internal/utils" "github.com/kagent-dev/kagent/go/core/internal/version" + "github.com/kagent-dev/kagent/go/core/pkg/consts" "github.com/modelcontextprotocol/go-sdk/mcp" corev1 "k8s.io/api/core/v1" "k8s.io/apimachinery/pkg/types" @@ -685,6 +686,37 @@ func (a *kagentReconciler) ReconcileKagentRemoteMCPServer(ctx context.Context, r // error path so a misconfigured RMS still surfaces a Failed condition. secretHash, secretErr := a.computeRemoteMCPServerSecretHash(ctx, server) + if remoteMCPServerDiscoveryDisabled(server) { + // The operator opted this server out of discovery (a server that + // authenticates every caller has no credential to offer the controller). + // Accept it without connecting: the status carries no tools, the DB keeps + // the server with an empty inventory, and agents resolve the tool list at + // run time with the credentials they carry. The TLS Secret hash is still + // published (agents fold it into their rollout hash), and the server + // still fails on what discovery would have rejected before dialing: a + // broken spec.tls Secret reference or a CA bundle without a valid + // certificate. A failure to persist the empty inventory fails it too, + // with the stale tools cleared from the status. + l.Info("skipping tool discovery for remote MCP server", "url", server.Spec.URL, "label", consts.DiscoveryLabel) + err := secretErr + if err == nil { + _, err = a.buildRemoteMCPServerTLSConfig(ctx, server) + } + if storeErr := a.storeToolServerWithoutTools(ctx, dbServer); storeErr != nil { + err = multierror.Append(err, storeErr) + } + if err != nil { + if err := a.reconcileRemoteMCPServerStatus(ctx, server, nil, secretHash, err); err != nil { + return fmt.Errorf("failed to reconcile remote mcp server status %s: %w", req.NamespacedName, err) + } + return nil + } + if err := a.setRemoteMCPServerStatus(ctx, server, nil, secretHash, metav1.ConditionTrue, "DiscoveryDisabled", remoteMCPServerDiscoveryDisabledMessage); err != nil { + return fmt.Errorf("failed to reconcile remote mcp server status %s: %w", req.NamespacedName, err) + } + return nil + } + l.Info("registering remote MCP server", "url", server.Spec.URL, "protocol", server.Spec.Protocol) start := time.Now() tools, err := a.upsertToolServerForRemoteMCPServer(ctx, dbServer, server) @@ -748,6 +780,17 @@ func (a *kagentReconciler) computeRemoteMCPServerSecretHash(ctx context.Context, return computeStatusSecretHash([]secretRef{{NamespacedName: nn, Secret: secret}}), nil } +// remoteMCPServerDiscoveryDisabledMessage explains an Accepted RemoteMCPServer +// that publishes no discovered tools because its operator opted out of discovery. +const remoteMCPServerDiscoveryDisabledMessage = "Tool discovery is disabled by the " + consts.DiscoveryLabel + "=" + consts.DiscoveryDisabled + + " label; agents resolve the tool list at run time. The controller did not connect: runtime authentication and reachability are not validated" + +// remoteMCPServerDiscoveryDisabled reports whether the operator opted the server +// out of tool discovery with the kagent.dev/discovery=disabled label. +func remoteMCPServerDiscoveryDisabled(server *v1alpha2.RemoteMCPServer) bool { + return server.Labels[consts.DiscoveryLabel] == consts.DiscoveryDisabled +} + func (a *kagentReconciler) reconcileRemoteMCPServerStatus( ctx context.Context, server *v1alpha2.RemoteMCPServer, @@ -769,6 +812,21 @@ func (a *kagentReconciler) reconcileRemoteMCPServerStatus( reason = "Reconciled" message = "Remote MCP server configuration accepted" } + return a.setRemoteMCPServerStatus(ctx, server, discoveredTools, secretHash, status, reason, message) +} + +// setRemoteMCPServerStatus writes the Accepted condition, the discovered tools +// and the Secret hash, skipping the update when nothing changed so the +// reconciler does not loop on its own status writes. +func (a *kagentReconciler) setRemoteMCPServerStatus( + ctx context.Context, + server *v1alpha2.RemoteMCPServer, + discoveredTools []*v1alpha2.MCPTool, + secretHash string, + status metav1.ConditionStatus, + reason string, + message string, +) error { conditionChanged := meta.SetStatusCondition(&server.Status.Conditions, metav1.Condition{ Type: v1alpha2.AgentConditionTypeAccepted, Status: status, @@ -1194,6 +1252,28 @@ func (a *kagentReconciler) upsertToolServerForRemoteMCPServer(ctx context.Contex return tools, nil } +// storeToolServerWithoutTools persists the server with an empty inventory, +// once: the periodic refresh re-enters the discovery-disabled path every +// interval and must not write to Postgres while nothing changed. The snapshot +// is taken over an empty, non-nil tool list so it differs from the row +// ensureToolServerRow writes after a failed discovery, which leaves the +// previously discovered tools in place. +func (a *kagentReconciler) storeToolServerWithoutTools(ctx context.Context, toolServer *database.ToolServer) error { + noTools := []*v1alpha2.MCPTool{} + if a.toolSnapshotUnchanged(toolServer, noTools) { + return nil + } + a.evictToolSnapshot(toolServer.Name, toolServer.GroupKind) + if _, err := a.dbClient.StoreToolServer(ctx, toolServer); err != nil { + return fmt.Errorf("failed to store toolServer %s: %w", toolServer.Name, err) + } + if err := a.dbClient.RefreshToolsForServer(ctx, toolServer.Name, toolServer.GroupKind); err != nil { + return fmt.Errorf("failed to clear tools for toolServer %s: %w", toolServer.Name, err) + } + a.rememberToolSnapshot(toolServer, noTools) + return nil +} + func toolSnapshotKey(name, groupKind string) string { return name + "\x00" + groupKind } diff --git a/go/core/internal/controller/reconciler/remote_mcp_server_discovery_test.go b/go/core/internal/controller/reconciler/remote_mcp_server_discovery_test.go new file mode 100644 index 0000000000..ff37729cf9 --- /dev/null +++ b/go/core/internal/controller/reconciler/remote_mcp_server_discovery_test.go @@ -0,0 +1,310 @@ +package reconciler + +import ( + "context" + "crypto/ecdsa" + "crypto/elliptic" + "crypto/rand" + "crypto/x509" + "crypto/x509/pkix" + "encoding/pem" + "errors" + "math/big" + "testing" + "time" + + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" + corev1 "k8s.io/api/core/v1" + "k8s.io/apimachinery/pkg/api/meta" + metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" + "k8s.io/apimachinery/pkg/runtime" + "k8s.io/apimachinery/pkg/types" + clientgoscheme "k8s.io/client-go/kubernetes/scheme" + "sigs.k8s.io/controller-runtime/pkg/client" + "sigs.k8s.io/controller-runtime/pkg/client/fake" + "sigs.k8s.io/controller-runtime/pkg/reconcile" + + "github.com/kagent-dev/kagent/go/api/database" + "github.com/kagent-dev/kagent/go/api/v1alpha2" + "github.com/kagent-dev/kagent/go/core/pkg/consts" +) + +// recordingToolServerStore is the slice of database.Client the RemoteMCPServer +// reconciler touches when discovery is disabled. Every other method panics +// (nil interface), which is the assertion that nothing else is reached. +type recordingToolServerStore struct { + database.Client + stored *database.ToolServer + refreshed *struct { + name, groupKind string + tools []*v1alpha2.MCPTool + } + writes int +} + +func (s *recordingToolServerStore) StoreToolServer(_ context.Context, toolServer *database.ToolServer) (*database.ToolServer, error) { + s.stored = toolServer + s.writes++ + return toolServer, nil +} + +func (s *recordingToolServerStore) RefreshToolsForServer(_ context.Context, serverName string, groupKind string, tools ...*v1alpha2.MCPTool) error { + s.refreshed = &struct { + name, groupKind string + tools []*v1alpha2.MCPTool + }{serverName, groupKind, tools} + s.writes++ + return nil +} + +// failingToolServerStore fails the first catalog write the opt-out performs. +type failingToolServerStore struct { + database.Client +} + +func (failingToolServerStore) StoreToolServer(context.Context, *database.ToolServer) (*database.ToolServer, error) { + return nil, errors.New("database unavailable") +} + +// selfSignedCAPEM returns a freshly generated, PEM-encoded CA certificate. +func selfSignedCAPEM(t *testing.T) []byte { + t.Helper() + key, err := ecdsa.GenerateKey(elliptic.P256(), rand.Reader) + require.NoError(t, err) + template := &x509.Certificate{ + SerialNumber: big.NewInt(1), + Subject: pkix.Name{CommonName: "test-ca"}, + NotBefore: time.Now().Add(-time.Hour), + NotAfter: time.Now().Add(time.Hour), + IsCA: true, + BasicConstraintsValid: true, + KeyUsage: x509.KeyUsageCertSign, + } + der, err := x509.CreateCertificate(rand.Reader, template, template, &key.PublicKey, key) + require.NoError(t, err) + return pem.EncodeToMemory(&pem.Block{Type: "CERTIFICATE", Bytes: der}) +} + +// TestReconcileKagentRemoteMCPServer_DiscoveryDisabled verifies that a +// RemoteMCPServer labeled kagent.dev/discovery=disabled is Accepted without the +// controller connecting to it: the URL below is unroutable, so any dial would +// fail the reconcile, and the recording store panics on anything but the two +// catalog writes the opt-out performs. +func TestReconcileKagentRemoteMCPServer_DiscoveryDisabled(t *testing.T) { + scheme := runtime.NewScheme() + require.NoError(t, clientgoscheme.AddToScheme(scheme)) + require.NoError(t, v1alpha2.AddToScheme(scheme)) + + server := &v1alpha2.RemoteMCPServer{ + ObjectMeta: metav1.ObjectMeta{ + Name: "per-caller", + Namespace: "default", + Generation: 2, + Labels: map[string]string{consts.DiscoveryLabel: consts.DiscoveryDisabled}, + }, + Spec: v1alpha2.RemoteMCPServerSpec{ + Description: "authenticates every caller", + URL: "http://192.0.2.1:1/mcp", + Protocol: v1alpha2.RemoteMCPServerProtocolStreamableHttp, + }, + Status: v1alpha2.RemoteMCPServerStatus{ + DiscoveredTools: []*v1alpha2.MCPTool{{Name: "stale"}}, + Conditions: []metav1.Condition{{ + Type: v1alpha2.AgentConditionTypeAccepted, Status: metav1.ConditionFalse, Reason: "ReconcileFailed", Message: "Unauthorized", + }}, + }, + } + kube := fake.NewClientBuilder(). + WithScheme(scheme). + WithStatusSubresource(server). + WithObjects(server). + Build() + store := &recordingToolServerStore{} + reconciler := &kagentReconciler{kube: kube, dbClient: store} + + require.NoError(t, reconciler.ReconcileKagentRemoteMCPServer(context.Background(), + reconcile.Request{NamespacedName: types.NamespacedName{Namespace: "default", Name: "per-caller"}})) + + updated := &v1alpha2.RemoteMCPServer{} + require.NoError(t, kube.Get(context.Background(), client.ObjectKeyFromObject(server), updated)) + + accepted := meta.FindStatusCondition(updated.Status.Conditions, v1alpha2.AgentConditionTypeAccepted) + require.NotNil(t, accepted) + assert.Equal(t, metav1.ConditionTrue, accepted.Status) + assert.Equal(t, "DiscoveryDisabled", accepted.Reason) + assert.Equal(t, remoteMCPServerDiscoveryDisabledMessage, accepted.Message) + assert.Equal(t, server.Generation, accepted.ObservedGeneration) + assert.Equal(t, server.Generation, updated.Status.ObservedGeneration) + assert.Empty(t, updated.Status.DiscoveredTools, "the stale inventory is cleared") + + require.NotNil(t, store.stored) + assert.Equal(t, "default/per-caller", store.stored.Name) + assert.Equal(t, "authenticates every caller", store.stored.Description) + require.NotNil(t, store.refreshed) + assert.Equal(t, store.stored.Name, store.refreshed.name) + assert.Equal(t, store.stored.GroupKind, store.refreshed.groupKind) + assert.Empty(t, store.refreshed.tools) + assert.Equal(t, 2, store.writes, "the server row and the empty inventory are written once") + + // The periodic refresh re-enters the same path; nothing changed, so the + // database is not touched again. + require.NoError(t, reconciler.ReconcileKagentRemoteMCPServer(context.Background(), + reconcile.Request{NamespacedName: types.NamespacedName{Namespace: "default", Name: "per-caller"}})) + assert.Equal(t, 2, store.writes, "an unchanged opted-out server is not rewritten") + + // A description change is a new snapshot and is written again. + require.NoError(t, kube.Get(context.Background(), client.ObjectKeyFromObject(server), updated)) + updated.Spec.Description = "renamed" + require.NoError(t, kube.Update(context.Background(), updated)) + require.NoError(t, reconciler.ReconcileKagentRemoteMCPServer(context.Background(), + reconcile.Request{NamespacedName: types.NamespacedName{Namespace: "default", Name: "per-caller"}})) + assert.Equal(t, 4, store.writes) + assert.Equal(t, "renamed", store.stored.Description) +} + +// TestReconcileKagentRemoteMCPServer_DiscoveryDisabledKeepsSecretHash verifies +// that the opt-out still publishes the TLS Secret hash agents fold into their +// rollout hash, and still fails the server on a broken spec.tls reference or a +// CA bundle discovery would reject. +func TestReconcileKagentRemoteMCPServer_DiscoveryDisabledKeepsSecretHash(t *testing.T) { + newServer := func() *v1alpha2.RemoteMCPServer { + return &v1alpha2.RemoteMCPServer{ + ObjectMeta: metav1.ObjectMeta{ + Name: "per-caller-tls", + Namespace: "default", + Generation: 1, + Labels: map[string]string{consts.DiscoveryLabel: consts.DiscoveryDisabled}, + }, + Spec: v1alpha2.RemoteMCPServerSpec{ + Description: "authenticates every caller, pinned CA", + URL: "https://192.0.2.1:1/mcp", + Protocol: v1alpha2.RemoteMCPServerProtocolStreamableHttp, + TLS: &v1alpha2.TLSConfig{CACertSecretRef: "ca", CACertSecretKey: "ca.crt"}, + }, + } + } + caSecret := func(data []byte) *corev1.Secret { + return &corev1.Secret{ + ObjectMeta: metav1.ObjectMeta{Name: "ca", Namespace: "default"}, + Data: map[string][]byte{"ca.crt": data}, + } + } + + for name, tc := range map[string]struct { + objects []client.Object + wantAccepted metav1.ConditionStatus + wantReason string + wantHash bool + wantMessage string + }{ + "CA Secret present: Accepted, hash published": { + objects: []client.Object{caSecret(selfSignedCAPEM(t))}, wantAccepted: metav1.ConditionTrue, wantReason: "DiscoveryDisabled", wantHash: true, + }, + "CA Secret missing: the broken reference still fails the server": { + wantAccepted: metav1.ConditionFalse, wantReason: "ReconcileFailed", wantMessage: "failed to get TLS secret", + }, + "CA bundle without a valid certificate: rejected as discovery would": { + objects: []client.Object{caSecret([]byte("-----BEGIN CERTIFICATE-----"))}, wantAccepted: metav1.ConditionFalse, wantReason: "ReconcileFailed", + wantHash: true, wantMessage: "does not contain valid PEM certificates", + }, + } { + t.Run(name, func(t *testing.T) { + scheme := runtime.NewScheme() + require.NoError(t, clientgoscheme.AddToScheme(scheme)) + require.NoError(t, v1alpha2.AddToScheme(scheme)) + server := newServer() + kube := fake.NewClientBuilder(). + WithScheme(scheme). + WithStatusSubresource(server). + WithObjects(append([]client.Object{server}, tc.objects...)...). + Build() + store := &recordingToolServerStore{} + reconciler := &kagentReconciler{kube: kube, dbClient: store} + + require.NoError(t, reconciler.ReconcileKagentRemoteMCPServer(context.Background(), + reconcile.Request{NamespacedName: client.ObjectKeyFromObject(server)})) + + updated := &v1alpha2.RemoteMCPServer{} + require.NoError(t, kube.Get(context.Background(), client.ObjectKeyFromObject(server), updated)) + accepted := meta.FindStatusCondition(updated.Status.Conditions, v1alpha2.AgentConditionTypeAccepted) + require.NotNil(t, accepted) + assert.Equal(t, tc.wantAccepted, accepted.Status) + assert.Equal(t, tc.wantReason, accepted.Reason) + assert.Empty(t, updated.Status.DiscoveredTools) + if tc.wantHash { + assert.NotEmpty(t, updated.Status.SecretHash, "the TLS Secret hash is published although discovery is disabled") + } else { + assert.Empty(t, updated.Status.SecretHash) + } + if tc.wantMessage != "" { + assert.Contains(t, accepted.Message, tc.wantMessage) + } + require.NotNil(t, store.stored, "the server is stored even when its TLS reference is broken") + }) + } +} + +// TestReconcileKagentRemoteMCPServer_DiscoveryDisabledStoreFailure verifies +// that a server switched to discovery disabled is not left Accepted with its +// old inventory when the empty catalog cannot be persisted. +func TestReconcileKagentRemoteMCPServer_DiscoveryDisabledStoreFailure(t *testing.T) { + scheme := runtime.NewScheme() + require.NoError(t, clientgoscheme.AddToScheme(scheme)) + require.NoError(t, v1alpha2.AddToScheme(scheme)) + + server := &v1alpha2.RemoteMCPServer{ + ObjectMeta: metav1.ObjectMeta{ + Name: "switched-off", + Namespace: "default", + Generation: 3, + Labels: map[string]string{consts.DiscoveryLabel: consts.DiscoveryDisabled}, + }, + Spec: v1alpha2.RemoteMCPServerSpec{ + URL: "http://192.0.2.1:1/mcp", + Protocol: v1alpha2.RemoteMCPServerProtocolStreamableHttp, + }, + Status: v1alpha2.RemoteMCPServerStatus{ + DiscoveredTools: []*v1alpha2.MCPTool{{Name: "discovered-before-the-switch"}}, + Conditions: []metav1.Condition{{ + Type: v1alpha2.AgentConditionTypeAccepted, Status: metav1.ConditionTrue, Reason: "Reconciled", Message: "Remote MCP server configuration accepted", + }}, + }, + } + kube := fake.NewClientBuilder(). + WithScheme(scheme). + WithStatusSubresource(server). + WithObjects(server). + Build() + reconciler := &kagentReconciler{kube: kube, dbClient: failingToolServerStore{}} + + require.NoError(t, reconciler.ReconcileKagentRemoteMCPServer(context.Background(), + reconcile.Request{NamespacedName: client.ObjectKeyFromObject(server)})) + + updated := &v1alpha2.RemoteMCPServer{} + require.NoError(t, kube.Get(context.Background(), client.ObjectKeyFromObject(server), updated)) + accepted := meta.FindStatusCondition(updated.Status.Conditions, v1alpha2.AgentConditionTypeAccepted) + require.NotNil(t, accepted) + assert.Equal(t, metav1.ConditionFalse, accepted.Status) + assert.Equal(t, "ReconcileFailed", accepted.Reason) + assert.Contains(t, accepted.Message, "database unavailable") + assert.Empty(t, updated.Status.DiscoveredTools, "the old inventory is cleared") +} + +func TestRemoteMCPServerDiscoveryDisabled(t *testing.T) { + for name, tc := range map[string]struct { + labels map[string]string + want bool + }{ + "no labels": {labels: nil, want: false}, + "other label": {labels: map[string]string{"app": "x"}, want: false}, + "other value": {labels: map[string]string{consts.DiscoveryLabel: "enabled"}, want: false}, + "disabled": {labels: map[string]string{consts.DiscoveryLabel: consts.DiscoveryDisabled}, want: true}, + "disabled, mixed": {labels: map[string]string{"app": "x", consts.DiscoveryLabel: consts.DiscoveryDisabled}, want: true}, + } { + t.Run(name, func(t *testing.T) { + server := &v1alpha2.RemoteMCPServer{ObjectMeta: metav1.ObjectMeta{Labels: tc.labels}} + assert.Equal(t, tc.want, remoteMCPServerDiscoveryDisabled(server)) + }) + } +} diff --git a/go/core/internal/controller/remote_mcp_server_controller.go b/go/core/internal/controller/remote_mcp_server_controller.go index a609fb6e36..ed549a755d 100644 --- a/go/core/internal/controller/remote_mcp_server_controller.go +++ b/go/core/internal/controller/remote_mcp_server_controller.go @@ -79,7 +79,12 @@ func (r *RemoteMCPServerController) SetupWithManager(mgr ctrl.Manager) error { WithOptions(controller.Options{ NeedLeaderElection: new(true), }). - For(&v1alpha2.RemoteMCPServer{}, builder.WithPredicates(predicate.GenerationChangedPredicate{})). + // A label change does not bump the generation, and the discovery opt-out + // (kagent.dev/discovery=disabled) is a label: watch both so the opt-out + // and its removal apply at once instead of on the next periodic refresh. + For(&v1alpha2.RemoteMCPServer{}, builder.WithPredicates(predicate.Or[client.Object]( + predicate.GenerationChangedPredicate{}, predicate.LabelChangedPredicate{}, + ))). Watches( &corev1.Secret{}, handler.EnqueueRequestsFromMapFunc(func(ctx context.Context, obj client.Object) []reconcile.Request { diff --git a/go/core/pkg/consts/labels.go b/go/core/pkg/consts/labels.go new file mode 100644 index 0000000000..9aa5360e19 --- /dev/null +++ b/go/core/pkg/consts/labels.go @@ -0,0 +1,23 @@ +package consts + +// DiscoveryLabel opts a tool server out of the controller's tool discovery. A +// RemoteMCPServer labeled `kagent.dev/discovery=disabled` is Accepted without the +// controller listing its tools: its status publishes no discovered tools, and the +// agents that reference it resolve the tool list at run time with the credentials +// they carry — for a server that authenticates every caller (a propagated caller +// token, which the controller does not hold), the controller-side listing would +// otherwise fail and mark the server as not Accepted forever. Any other value, or +// no label, keeps discovery on. The same label excludes a Service or MCPServer +// from automatic discovery (predicates.DiscoveryDisabledPredicate). +// +// Accepted=True with reason DiscoveryDisabled means only that the controller +// skipped discovery on purpose and the spec.tls configuration is usable; it +// does not validate runtime authentication or endpoint reachability. When +// switching a server to the propagated caller credential, clear any +// spec.headersFrom entry that sets Authorization: it overrides the caller's +// token. Under server-side apply or GitOps tooling that only adds fields, +// omitting the entry may leave it in place, so set `headersFrom: []` explicitly. +const DiscoveryLabel = "kagent.dev/discovery" + +// DiscoveryDisabled is the DiscoveryLabel value that turns discovery off. +const DiscoveryDisabled = "disabled"