diff --git a/cmd/ateapi/main.go b/cmd/ateapi/main.go index b7bc65472..f366b4c5e 100644 --- a/cmd/ateapi/main.go +++ b/cmd/ateapi/main.go @@ -196,17 +196,11 @@ func main() { mux := grpc.NewServer( grpc.Creds(serverCreds), grpc.StatsHandler(otelgrpc.NewServerHandler()), - // Bounds every connection's lifetime so round_robin clients - // periodically re-resolve DNS and pick up replicas added since they - // last connected - without this, an existing connection never - // notices new replicas on its own (see https://github.com/grpc/grpc/issues/12295). - // - // TODO: Replace with a resolver that watches Endpoints/EndpointSlices - // directly and pushes address updates, instead of relying on forced - // reconnects to trigger DNS re-resolution. See - // https://github.com/sercand/kuberesolver. + // Close connections after an hour to allow for any + // client that doesn't use Kubernetes endpoint resolvers + // to eventually reobtain backend IPs. https://github.com/grpc/grpc/issues/12295 grpc.KeepaliveParams(keepalive.ServerParameters{ - MaxConnectionAge: 1 * time.Minute, + MaxConnectionAge: 1 * time.Hour, MaxConnectionAgeGrace: maxRPCDeadline + time.Minute, }), grpc.ChainUnaryInterceptor( diff --git a/cmd/atecontroller/internal/controllers/actortemplate_controller.go b/cmd/atecontroller/internal/controllers/actortemplate_controller.go index f5778fc5d..4d39e2aa9 100644 --- a/cmd/atecontroller/internal/controllers/actortemplate_controller.go +++ b/cmd/atecontroller/internal/controllers/actortemplate_controller.go @@ -59,6 +59,7 @@ type ActorTemplateReconciler struct { //+kubebuilder:rbac:groups=core,resources=pods,verbs=get;list;watch;create;update;patch;delete //+kubebuilder:rbac:groups=core,resources=configmaps,verbs=get;list;watch //+kubebuilder:rbac:groups=core,resources=secrets,verbs=get;list;watch +//+kubebuilder:rbac:groups=discovery.k8s.io,resources=endpointslices,verbs=get;list;watch // Reconcile is part of the main kubernetes reconciliation loop which aims to // move the current state of the cluster closer to the desired state. diff --git a/cmd/atecontroller/main.go b/cmd/atecontroller/main.go index 0ffc0d4ca..3a03b1b9f 100644 --- a/cmd/atecontroller/main.go +++ b/cmd/atecontroller/main.go @@ -24,6 +24,7 @@ import ( "google.golang.org/grpc" "k8s.io/apimachinery/pkg/runtime" utilruntime "k8s.io/apimachinery/pkg/util/runtime" + "k8s.io/client-go/kubernetes" clientgoscheme "k8s.io/client-go/kubernetes/scheme" ctrl "sigs.k8s.io/controller-runtime" "sigs.k8s.io/controller-runtime/pkg/healthz" @@ -37,7 +38,7 @@ var ( scheme = runtime.NewScheme() setupLog = ctrl.Log.WithName("setup") - ateAPIConnSpec = pflag.String("ateapi-conn-spec", "dns:///api.ate-system.svc:443", "") + ateAPIConnSpec = pflag.String("ateapi-conn-spec", "k8s:///api.ate-system.svc:443", "") otelEndpoint = pflag.String("otel-exporter-otlp-endpoint", os.Getenv("OTEL_EXPORTER_OTLP_ENDPOINT"), "OTLP endpoint set on ateom worker pods so they push telemetry. Defaults to the controller's own OTEL_EXPORTER_OTLP_ENDPOINT.") @@ -57,8 +58,15 @@ func init() { func main() { pflag.Parse() ctrl.SetLogger(zap.New(zap.UseDevMode(true))) + k8sConfig := ctrl.GetConfigOrDie() + k8sClient, err := kubernetes.NewForConfig(k8sConfig) + if err != nil { + setupLog.Error(err, "creating kubernetes client for ateapi dialer") + os.Exit(1) + } dialOpts, err := ateapiauth.DialOptions(ateapiauth.ClientConfig{ + K8sClient: k8sClient, UseTokenAuth: *ateapiTokenAuth, CAFile: *ateapiCAFile, ServerName: *ateapiServerName, @@ -78,7 +86,7 @@ func main() { ateapiClient := ateapipb.NewControlClient(ateapiConn) - mgr, err := ctrl.NewManager(ctrl.GetConfigOrDie(), ctrl.Options{ + mgr, err := ctrl.NewManager(k8sConfig, ctrl.Options{ Scheme: scheme, }) if err != nil { diff --git a/cmd/atenet/internal/router/router.go b/cmd/atenet/internal/router/router.go index 117099fcc..ffea50d0f 100644 --- a/cmd/atenet/internal/router/router.go +++ b/cmd/atenet/internal/router/router.go @@ -170,6 +170,7 @@ func (s *RouterServer) Run(ctx context.Context) error { go serverboot.StartMetricsServer(ctx, serverboot.MetricsServerOptions{Addr: s.cfg.MetricsAddr}) dialOpts, err := ateapiauth.DialOptions(ateapiauth.ClientConfig{ + K8sClient: s.clientset, UseTokenAuth: s.cfg.Auth.AteapiUseTokenAuth, CAFile: s.cfg.Auth.AteapiCAFile, ServerName: s.cfg.Auth.AteapiServerName, diff --git a/cmd/benchmarking/boomer-glutton/main.go b/cmd/benchmarking/boomer-glutton/main.go index efa921ef8..413d93406 100644 --- a/cmd/benchmarking/boomer-glutton/main.go +++ b/cmd/benchmarking/boomer-glutton/main.go @@ -36,7 +36,7 @@ import ( func main() { var ( - apiEndpoint = flag.String("api-endpoint", "dns:///api.ate-system.svc.cluster.local:443", "ateapi gRPC dial target.") + apiEndpoint = flag.String("api-endpoint", "k8s:///api.ate-system.svc.cluster.local:443", "ateapi gRPC dial target.") routerURL = flag.String("router-url", "http://atenet-router.ate-system.svc.cluster.local", "atenet HTTP router base URL (no trailing slash).") atespace = flag.String("atespace", "benchmark", "Atespace every actor this worker creates lives in. Ensured (CreateAtespace, AlreadyExists is ok) at startup.") promAddr = flag.String("prometheus-addr", ":8001", "Address for the Prometheus /metrics endpoint.") diff --git a/internal/ateapiauth/client.go b/internal/ateapiauth/client.go index 70acb511b..81f9a7e40 100644 --- a/internal/ateapiauth/client.go +++ b/internal/ateapiauth/client.go @@ -23,8 +23,10 @@ import ( "strings" "github.com/agent-substrate/substrate/internal/credbundle" + "github.com/agent-substrate/substrate/internal/k8sresolver" "google.golang.org/grpc" "google.golang.org/grpc/credentials" + "k8s.io/client-go/kubernetes" ) const ( @@ -63,6 +65,10 @@ type ClientConfig struct { // and PKCS8 private key presented to the server. Required unless // UseTokenAuth is set, ignored otherwise. ClientCredBundle string + + // K8sClient is an optional Kubernetes client. When provided, an EndpointSlice + // resolver builder using this client will be attached to DialOptions. + K8sClient kubernetes.Interface } // DialOptions returns the grpc.DialOption set described by cfg, suitable to @@ -87,18 +93,25 @@ func DialOptions(cfg ClientConfig) ([]grpc.DialOption, error) { RootCAs: pool, ServerName: cfg.ServerName, } + + opts := []grpc.DialOption{ + grpc.WithDefaultServiceConfig(roundRobinServiceConfig), + } + if cfg.K8sClient != nil { + opts = append(opts, grpc.WithResolvers(k8sresolver.NewBuilder(cfg.K8sClient))) + } + if cfg.UseTokenAuth { - return []grpc.DialOption{ + opts = append(opts, grpc.WithTransportCredentials(credentials.NewTLS(tlsCfg)), grpc.WithPerRPCCredentials(&fileTokenCreds{path: cfg.TokenFile}), - grpc.WithDefaultServiceConfig(roundRobinServiceConfig), - }, nil + ) + return opts, nil } + tlsCfg.GetClientCertificate = credbundle.ClientLoader(cfg.ClientCredBundle) - return []grpc.DialOption{ - grpc.WithTransportCredentials(credentials.NewTLS(tlsCfg)), - grpc.WithDefaultServiceConfig(roundRobinServiceConfig), - }, nil + opts = append(opts, grpc.WithTransportCredentials(credentials.NewTLS(tlsCfg))) + return opts, nil } func loadCAPool(caFile string) (*x509.CertPool, error) { diff --git a/internal/ateapiauth/client_test.go b/internal/ateapiauth/client_test.go index 9109699d2..8a4ee22df 100644 --- a/internal/ateapiauth/client_test.go +++ b/internal/ateapiauth/client_test.go @@ -39,6 +39,7 @@ import ( healthpb "google.golang.org/grpc/health/grpc_health_v1" "google.golang.org/grpc/metadata" "google.golang.org/grpc/status" + "k8s.io/client-go/kubernetes/fake" ) func TestDialOptionsRequiresCAFile(t *testing.T) { @@ -56,10 +57,11 @@ func TestDialOptionsRequiresCAFile(t *testing.T) { } func TestDialOptionsRequiresModeCredential(t *testing.T) { + fakeClient := fake.NewSimpleClientset() for name, cfg := range map[string]ClientConfig{ - "cert mode without bundle": {CAFile: "ca.pem"}, - "token mode without token": {CAFile: "ca.pem", UseTokenAuth: true}, - "cert path does not satisfy token mode": {CAFile: "ca.pem", UseTokenAuth: true, ClientCredBundle: "bundle.pem"}, + "cert mode without bundle": {K8sClient: fakeClient, CAFile: "ca.pem"}, + "token mode without token": {K8sClient: fakeClient, CAFile: "ca.pem", UseTokenAuth: true}, + "cert path does not satisfy token mode": {K8sClient: fakeClient, CAFile: "ca.pem", UseTokenAuth: true, ClientCredBundle: "bundle.pem"}, } { t.Run(name, func(t *testing.T) { _, err := DialOptions(cfg) @@ -106,6 +108,7 @@ func TestDialOptionsMTLSHandshake(t *testing.T) { t.Run("with client cert", func(t *testing.T) { // TokenFile is ignored in cert mode (the default). opts, err := DialOptions(ClientConfig{ + K8sClient: fake.NewSimpleClientset(), CAFile: caFile, ClientCredBundle: clientBundle, TokenFile: filepath.Join(dir, "does-not-exist-token"), @@ -168,6 +171,7 @@ func TestDialOptionsTokenSendsBearer(t *testing.T) { // ClientCredBundle stays set (as the base manifests leave it) but is // ignored in token mode — it doesn't even need to exist on disk. opts, err := DialOptions(ClientConfig{ + K8sClient: fake.NewSimpleClientset(), CAFile: caFile, UseTokenAuth: true, TokenFile: tokenFile, diff --git a/internal/benchmarking/boomer/glutton/grpcclient.go b/internal/benchmarking/boomer/glutton/grpcclient.go index 88e9361ea..696f0a3b2 100644 --- a/internal/benchmarking/boomer/glutton/grpcclient.go +++ b/internal/benchmarking/boomer/glutton/grpcclient.go @@ -18,9 +18,13 @@ import ( "fmt" "github.com/agent-substrate/substrate/internal/ateapiauth" + _ "github.com/agent-substrate/substrate/internal/k8sresolver" "github.com/agent-substrate/substrate/pkg/proto/ateapipb" "go.opentelemetry.io/contrib/instrumentation/google.golang.org/grpc/otelgrpc" "google.golang.org/grpc" + "k8s.io/client-go/kubernetes" + "k8s.io/client-go/rest" + "k8s.io/client-go/tools/clientcmd" ) const ( @@ -42,7 +46,22 @@ const ( // and atenet-router. ClientCredBundle re-reads the bundle on every handshake, // so rotations are picked up without a restart. func DialControl(endpoint string) (*grpc.ClientConn, ateapipb.ControlClient, error) { + k8sConfig, err := rest.InClusterConfig() + if err != nil { + loadingRules := clientcmd.NewDefaultClientConfigLoadingRules() + configOverrides := &clientcmd.ConfigOverrides{} + k8sConfig, err = clientcmd.NewNonInteractiveDeferredLoadingClientConfig(loadingRules, configOverrides).ClientConfig() + if err != nil { + return nil, nil, fmt.Errorf("k8s config: %w", err) + } + } + k8sClient, err := kubernetes.NewForConfig(k8sConfig) + if err != nil { + return nil, nil, fmt.Errorf("k8s client: %w", err) + } + dialOpts, err := ateapiauth.DialOptions(ateapiauth.ClientConfig{ + K8sClient: k8sClient, CAFile: ateapiCAFile, ClientCredBundle: ateapiCredBundle, ServerName: ateapiServerName, diff --git a/internal/k8sresolver/resolver.go b/internal/k8sresolver/resolver.go new file mode 100644 index 000000000..4d93fe262 --- /dev/null +++ b/internal/k8sresolver/resolver.go @@ -0,0 +1,219 @@ +// Copyright 2026 Google LLC +// +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. + +package k8sresolver + +import ( + "context" + "fmt" + "net" + "slices" + "strconv" + "strings" + + "google.golang.org/grpc/resolver" + v1 "k8s.io/api/discovery/v1" + metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" + "k8s.io/apimachinery/pkg/labels" + "k8s.io/client-go/informers" + "k8s.io/client-go/kubernetes" + discoverylisters "k8s.io/client-go/listers/discovery/v1" + "k8s.io/client-go/rest" + "k8s.io/client-go/tools/cache" +) + +const Scheme = "k8s" + +func init() { resolver.Register(NewBuilder(nil)) } + +// Builder implements resolver.Builder for Kubernetes EndpointSlices backed by client-go Informers. +type Builder struct{ client kubernetes.Interface } + +// NewBuilder creates a new gRPC resolver Builder for EndpointSlices using client-go Informers. +func NewBuilder(client kubernetes.Interface) *Builder { return &Builder{client: client} } +func (b *Builder) Scheme() string { return Scheme } + +func (b *Builder) Build(target resolver.Target, cc resolver.ClientConn, opts resolver.BuildOptions) (resolver.Resolver, error) { + ns, svc, port, err := ParseTarget(target) + if err != nil { + return nil, fmt.Errorf("k8sresolver: %w", err) + } + + cli := b.client + if cli == nil { + config, err := rest.InClusterConfig() + if err != nil { + return nil, fmt.Errorf("k8sresolver: in-cluster config: %w", err) + } + if cli, err = kubernetes.NewForConfig(config); err != nil { + return nil, fmt.Errorf("k8sresolver: client creation: %w", err) + } + } + + selector := labels.Set{"kubernetes.io/service-name": svc}.AsSelector() + factory := informers.NewSharedInformerFactoryWithOptions(cli, 0, + informers.WithNamespace(ns), + informers.WithTweakListOptions(func(o *metav1.ListOptions) { o.LabelSelector = selector.String() }), + ) + + inf := factory.Discovery().V1().EndpointSlices() + ctx, cancel := context.WithCancel(context.Background()) + r := &k8sResolver{ + ctx: ctx, + cancel: cancel, + service: svc, + port: port, + cc: cc, + informer: inf.Informer(), + lister: inf.Lister().EndpointSlices(ns), + selector: selector, + } + + _ = inf.Informer().SetWatchErrorHandler(func(_ *cache.Reflector, err error) { + if ctx.Err() == nil { + cc.ReportError(fmt.Errorf("k8sresolver: watch error for %s/%s: %w", ns, svc, err)) + } + }) + + reg, err := inf.Informer().AddEventHandler(cache.ResourceEventHandlerFuncs{ + AddFunc: func(_ any) { r.updateState() }, + UpdateFunc: func(_, _ any) { r.updateState() }, + DeleteFunc: func(_ any) { r.updateState() }, + }) + if err != nil { + return nil, fmt.Errorf("k8sresolver: failed to add event handler for %s/%s: %w", ns, svc, err) + } + + r.registration = reg + + factory.Start(ctx.Done()) + go func() { + if cache.WaitForCacheSync(ctx.Done(), inf.Informer().HasSynced) { + r.updateState() + } else if ctx.Err() == nil { + cc.ReportError(fmt.Errorf("k8sresolver: failed to sync EndpointSlice cache for %s/%s", ns, svc)) + } + }() + return r, nil +} + +// ParseTarget parses resolver.Target in canonical format k8s:////[:port] or k8s:///.[:port]. +func ParseTarget(target resolver.Target) (ns, svc, port string, err error) { + ep := target.Endpoint() + if target.URL.Host != "" && target.URL.Host != "localhost" { + ep = target.URL.Host + "/" + strings.TrimPrefix(target.URL.Path, "/") + } + + if parts := strings.SplitN(ep, "/", 2); len(parts) == 2 { + ns, ep = parts[0], parts[1] + } else { + ns = "default" + } + + svc, port, err = net.SplitHostPort(ep) + if err != nil { + svc, port = ep, "443" + } + + if parts := strings.Split(svc, "."); len(parts) > 1 { + svc, ns = parts[0], parts[1] + } + + if ns == "" || svc == "" { + return "", "", "", fmt.Errorf("invalid target %q, expected k8s:////[:port]", target.String()) + } + return ns, svc, port, nil +} + +type k8sResolver struct { + ctx context.Context + cancel context.CancelFunc + service string + port string + cc resolver.ClientConn + informer cache.SharedIndexInformer + registration cache.ResourceEventHandlerRegistration + lister discoverylisters.EndpointSliceNamespaceLister + selector labels.Selector +} + +func (r *k8sResolver) ResolveNow(resolver.ResolveNowOptions) { r.updateState() } + +func (r *k8sResolver) Close() { + r.cancel() + if r.registration != nil { + _ = r.informer.RemoveEventHandler(r.registration) + } +} + +func (r *k8sResolver) updateState() { + if r.ctx.Err() != nil { + return + } + + eps, err := r.lister.List(r.selector) + if err != nil { + if r.ctx.Err() == nil { + r.cc.ReportError(err) + } + return + } + + targetPort, _ := strconv.Atoi(r.port) + var addrs []resolver.Address + for _, slice := range eps { + p := targetPort + if p == 0 && len(slice.Ports) > 0 && slice.Ports[0].Port != nil { + p = int(*slice.Ports[0].Port) + } + if p == 0 { + p = 443 + } + for _, ep := range slice.Endpoints { + if !isHealthy(&ep) { + continue + } + for _, ip := range ep.Addresses { + addrs = append(addrs, resolver.Address{Addr: net.JoinHostPort(ip, strconv.Itoa(p))}) + } + } + } + + slices.SortFunc(addrs, func(a, b resolver.Address) int { return strings.Compare(a.Addr, b.Addr) }) + addrs = slices.CompactFunc(addrs, func(a, b resolver.Address) bool { return a.Addr == b.Addr }) + + if err := r.cc.UpdateState(resolver.State{Addresses: addrs}); err != nil && r.ctx.Err() == nil { + r.cc.ReportError(err) + } +} + +func isHealthy(ep *v1.Endpoint) bool { + // ready indicates that this endpoint is prepared to receive traffic, + // according to whatever system is managing the endpoint. A nil value + // indicates an unknown state. In most cases consumers should interpret this + // unknown state as ready. + // More info: vendor/k8s.io/api/discovery/v1/types.go + isReady := ep.Conditions.Ready == nil || *ep.Conditions.Ready + // serving is identical to ready except that it is set regardless of the + // terminating state of endpoints. This condition should be set to true for + // a ready endpoint that is terminating. If nil, consumers should defer to + // the ready condition. + // More info: vendor/k8s.io/api/discovery/v1/types.go + isServing := (ep.Conditions.Serving == nil && isReady) || (ep.Conditions.Serving != nil && *ep.Conditions.Serving) + + // Return healthy for endpoints that are either ready or serving. We can + // still send traffic to terminating pods so don't immediately drop + // connections. + return isServing +} diff --git a/internal/k8sresolver/resolver_test.go b/internal/k8sresolver/resolver_test.go new file mode 100644 index 000000000..9927f9c6e --- /dev/null +++ b/internal/k8sresolver/resolver_test.go @@ -0,0 +1,254 @@ +// Copyright 2026 Google LLC +// +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. + +package k8sresolver + +import ( + "context" + "net/url" + "reflect" + "testing" + "time" + + "google.golang.org/grpc/resolver" + "google.golang.org/grpc/serviceconfig" + discoveryv1 "k8s.io/api/discovery/v1" + metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" + "k8s.io/client-go/kubernetes/fake" +) + +type testClientConn struct { + resolver.ClientConn + stateChan chan resolver.State + errChan chan error +} + +func newTestClientConn() *testClientConn { + return &testClientConn{ + stateChan: make(chan resolver.State, 10), + errChan: make(chan error, 10), + } +} + +func (c *testClientConn) UpdateState(state resolver.State) error { + c.stateChan <- state + return nil +} + +func (c *testClientConn) ReportError(err error) { + c.errChan <- err +} + +func (c *testClientConn) ParseServiceConfig(serviceConfigJSON string) *serviceconfig.ParseResult { + return nil +} + +func TestParseTarget(t *testing.T) { + tests := []struct { + name string + target resolver.Target + wantNamespace string + wantService string + wantPort string + wantErr bool + }{ + { + name: "slash format namespace/service:port", + target: resolver.Target{ + URL: url.URL{Scheme: "k8s", Path: "/ate-system/api:443"}, + }, + wantNamespace: "ate-system", + wantService: "api", + wantPort: "443", + }, + { + name: "host format host=ate-system path=/api:443", + target: resolver.Target{ + URL: url.URL{Scheme: "k8s", Host: "ate-system", Path: "/api:443"}, + }, + wantNamespace: "ate-system", + wantService: "api", + wantPort: "443", + }, + { + name: "fqdn format api.ate-system.svc:443", + target: resolver.Target{ + URL: url.URL{Scheme: "k8s", Path: "/api.ate-system.svc:443"}, + }, + wantNamespace: "ate-system", + wantService: "api", + wantPort: "443", + }, + { + name: "full cluster fqdn format api.ate-system.svc.cluster.local:443", + target: resolver.Target{ + URL: url.URL{Scheme: "k8s", Path: "/api.ate-system.svc.cluster.local:443"}, + }, + wantNamespace: "ate-system", + wantService: "api", + wantPort: "443", + }, + { + name: "simple service without port", + target: resolver.Target{ + URL: url.URL{Scheme: "k8s", Path: "/api"}, + }, + wantNamespace: "default", + wantService: "api", + wantPort: "443", + }, + } + + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + gotNs, gotSvc, gotPort, err := ParseTarget(tt.target) + if (err != nil) != tt.wantErr { + t.Fatalf("ParseTarget() error = %v, wantErr %v", err, tt.wantErr) + } + if gotNs != tt.wantNamespace { + t.Errorf("ParseTarget() gotNs = %v, want %v", gotNs, tt.wantNamespace) + } + if gotSvc != tt.wantService { + t.Errorf("ParseTarget() gotSvc = %v, want %v", gotSvc, tt.wantService) + } + if gotPort != tt.wantPort { + t.Errorf("ParseTarget() gotPort = %v, want %v", gotPort, tt.wantPort) + } + }) + } +} + +func TestK8sResolverEndpointSliceUpdates(t *testing.T) { + readyTrue := true + port443 := int32(443) + + slice1 := &discoveryv1.EndpointSlice{ + ObjectMeta: metav1.ObjectMeta{ + Name: "api-slice-1", + Namespace: "ate-system", + Labels: map[string]string{ + "kubernetes.io/service-name": "api", + }, + }, + Ports: []discoveryv1.EndpointPort{ + {Port: &port443}, + }, + Endpoints: []discoveryv1.Endpoint{ + { + Addresses: []string{"10.0.0.1"}, + Conditions: discoveryv1.EndpointConditions{ + Ready: &readyTrue, + }, + }, + }, + } + + fakeClient := fake.NewSimpleClientset(slice1) + builder := NewBuilder(fakeClient) + cc := newTestClientConn() + + target := resolver.Target{ + URL: url.URL{Scheme: "k8s", Path: "/ate-system/api:443"}, + } + + res, err := builder.Build(target, cc, resolver.BuildOptions{}) + if err != nil { + t.Fatalf("builder.Build() error = %v", err) + } + defer res.Close() + + select { + case state := <-cc.stateChan: + wantAddrs := []resolver.Address{{Addr: "10.0.0.1:443"}} + if !reflect.DeepEqual(state.Addresses, wantAddrs) { + t.Errorf("initial state.Addresses = %v, want %v", state.Addresses, wantAddrs) + } + case <-time.After(2 * time.Second): + t.Fatal("timed out waiting for initial state update") + } + + // Add a new EndpointSlice + slice2 := &discoveryv1.EndpointSlice{ + ObjectMeta: metav1.ObjectMeta{ + Name: "api-slice-2", + Namespace: "ate-system", + Labels: map[string]string{ + "kubernetes.io/service-name": "api", + }, + }, + Ports: []discoveryv1.EndpointPort{ + {Port: &port443}, + }, + Endpoints: []discoveryv1.Endpoint{ + { + Addresses: []string{"10.0.0.2"}, + Conditions: discoveryv1.EndpointConditions{ + Ready: &readyTrue, + }, + }, + }, + } + + _, err = fakeClient.DiscoveryV1().EndpointSlices("ate-system").Create(context.Background(), slice2, metav1.CreateOptions{}) + if err != nil { + t.Fatalf("failed to create slice2: %v", err) + } + + select { + case state := <-cc.stateChan: + wantAddrs := []resolver.Address{ + {Addr: "10.0.0.1:443"}, + {Addr: "10.0.0.2:443"}, + } + if !reflect.DeepEqual(state.Addresses, wantAddrs) { + t.Errorf("updated state.Addresses = %v, want %v", state.Addresses, wantAddrs) + } + case <-time.After(2 * time.Second): + t.Fatal("timed out waiting for state update after slice addition") + } +} + +func TestK8sResolverClose(t *testing.T) { + fakeClient := fake.NewSimpleClientset() + builder := NewBuilder(fakeClient) + cc := newTestClientConn() + + target := resolver.Target{ + URL: url.URL{Scheme: "k8s", Path: "/ate-system/api:443"}, + } + + res, err := builder.Build(target, cc, resolver.BuildOptions{}) + if err != nil { + t.Fatalf("builder.Build() error = %v", err) + } + + res.Close() + + // Drain any initial state updates that happened before Close + for len(cc.stateChan) > 0 { + <-cc.stateChan + } + + // Trigger a resolution after Close + res.ResolveNow(resolver.ResolveNowOptions{}) + + select { + case state := <-cc.stateChan: + t.Fatalf("unexpected state update after Close(): %v", state) + case err := <-cc.errChan: + t.Fatalf("unexpected error report after Close(): %v", err) + case <-time.After(100 * time.Millisecond): + // Expected: no calls to ClientConn after Close + } +} diff --git a/manifests/ate-install/atenet-router.yaml b/manifests/ate-install/atenet-router.yaml index 8da013ab0..38c4a824a 100644 --- a/manifests/ate-install/atenet-router.yaml +++ b/manifests/ate-install/atenet-router.yaml @@ -33,6 +33,14 @@ rules: - get - watch - list +- apiGroups: + - "discovery.k8s.io" + resources: + - endpointslices + verbs: + - get + - watch + - list --- apiVersion: rbac.authorization.k8s.io/v1 kind: ClusterRoleBinding @@ -137,7 +145,7 @@ spec: - "--otlp-collector-address=opentelemetry-collector.gke-managed-otel.svc.cluster.local:4317" # Client auth to ateapi (mtls): verify the serving cert against the # servicedns trust bundle and present the podidentity client cert. - - "--ateapi-address=dns:///api.ate-system.svc:443" + - "--ateapi-address=k8s:///api.ate-system.svc:443" - "--ateapi-ca-file=/run/servicedns-ca/trust-bundle.pem" - "--ateapi-client-cert=/run/podidentity.podcert.ate.dev/credential-bundle.pem" env: diff --git a/manifests/ate-install/generated/role.yaml b/manifests/ate-install/generated/role.yaml index 48f80e958..12aacd141 100644 --- a/manifests/ate-install/generated/role.yaml +++ b/manifests/ate-install/generated/role.yaml @@ -80,6 +80,14 @@ rules: - get - patch - update +- apiGroups: + - discovery.k8s.io + resources: + - endpointslices + verbs: + - get + - list + - watch - apiGroups: - networking.k8s.io resources: