Implement endpoint slice resolution for ate api connections for HA

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(

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.

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 {

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,

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.")

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) {

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,

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,