Fix Kubernetes locker discovery with EndpointSlices

docs/user_guide/HA.md

@@ -91,7 +91,7 @@ clustering:
   # locker is used to configure the KV store used for 
   # service registration, service discovery, leader election and targets locks
   locker:
-    # type of locker, only consul is supported currently
+    # type of locker: consul, k8s, or redis
     type: consul
     # address of the locker server
     address: localhost:8500

docs/user_guide/ha_kubernetes.md

@@ -0,0 +1,95 @@
+# Kubernetes locker
+
+The `k8s` locker uses Kubernetes Leases for leader election and target ownership,
+and EndpointSlices for peer discovery. gNMIc runs inside the cluster and uses its
+Pod's ServiceAccount. The locker does not require a separate Redis or Consul service.
+
+Configure the namespace containing both the gNMIc Pods and their API Service:
+
+```yaml
+api-server:
+  address: :7890
+clustering:
+  cluster-name: telemetry
+  instance-name: ${POD_NAME}
+  locker:
+    type: k8s
+    namespace: telemetry
+```
+
+Set `POD_NAME` from the Pod's `metadata.name` using the downward API. The instance
+name must match the Pod name so that discovered peers match the instance names
+stored with their Leases.
+The API Service name is `<cluster-name>-gnmic-api`; its selector must match the
+collector Pods. Expose a single TCP API port on this Service:
+
+```yaml
+apiVersion: v1
+kind: Service
+metadata:
+  name: telemetry-gnmic-api
+  namespace: telemetry
+spec:
+  selector:
+    app.kubernetes.io/name: gnmic
+    app.kubernetes.io/instance: telemetry
+  ports:
+    - name: api
+      port: 7890
+      targetPort: api
+      protocol: TCP
+```
+
+Use an API readiness probe so Kubernetes only advertises running API servers.
+Discovery combines all `discovery.k8s.io/v1` EndpointSlices labeled
+`kubernetes.io/service-name=telemetry-gnmic-api`. It excludes endpoints explicitly
+marked not ready, not serving, or terminating. Unspecified readiness and serving
+conditions are accepted. Empty results remove the previously discovered peers.
+Duplicate Pod endpoints across slices produce one stable API address, including
+when both IPv4 and IPv6 addresses are present.
+
+Grant the ServiceAccount access to EndpointSlices and Leases in that namespace:
+
+```yaml
+apiVersion: v1
+kind: ServiceAccount
+metadata:
+  name: gnmic
+  namespace: telemetry
+---
+apiVersion: rbac.authorization.k8s.io/v1
+kind: Role
+metadata:
+  name: gnmic-locker
+  namespace: telemetry
+rules:
+  - apiGroups: [discovery.k8s.io]
+    resources: [endpointslices]
+    verbs: [get, list, watch]
+  - apiGroups: [coordination.k8s.io]
+    resources: [leases]
+    verbs: [get, list, watch, create, update, delete]
+---
+apiVersion: rbac.authorization.k8s.io/v1
+kind: RoleBinding
+metadata:
+  name: gnmic-locker
+  namespace: telemetry
+subjects:
+  - kind: ServiceAccount
+    name: gnmic
+    namespace: telemetry
+roleRef:
+  apiGroup: rbac.authorization.k8s.io
+  kind: Role
+  name: gnmic-locker
+```
+
+Set `serviceAccountName: gnmic` in the collector Pod template. Existing installations
+must grant EndpointSlice permissions before upgrading gNMIc; core/v1 Endpoints
+permissions are no longer required by the locker. Lease permissions and renewal
+settings are unchanged by this discovery migration.
+
+Each active target has a Lease that is periodically renewed through the Kubernetes
+API. Size the deployment using measured renewal latency and API request capacity,
+and verify target ownership and leader recovery during Pod replacement.

mkdocs.yml

@@ -105,6 +105,7 @@ nav:
       - Caching: user_guide/caching.md
 
       - Clustering: user_guide/HA.md
+      - Kubernetes locker: user_guide/ha_kubernetes.md
 
       - REST API: 
           - Introduction: user_guide/api/api_intro.md

pkg/lockers/k8s_locker/k8s_discovery.go

@@ -0,0 +1,166 @@
+// © 2022 Nokia.
+//
+// This code is a Contribution to the gNMIc project (“Work”) made under the Google Software Grant and Corporate Contributor License Agreement (“CLA”) and governed by the Apache License 2.0.
+// No other rights or licenses in or to any of Nokia’s intellectual property are granted for any other purpose.
+// This code is provided on an “as is” basis without any warranties of any kind.
+//
+// SPDX-License-Identifier: Apache-2.0
+
+package k8s_locker
+
+import (
+	"context"
+	"net"
+	"reflect"
+	"sort"
+	"strconv"
+	"time"
+
+	corev1 "k8s.io/api/core/v1"
+	discoveryv1 "k8s.io/api/discovery/v1"
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	"k8s.io/apimachinery/pkg/labels"
+	"k8s.io/apimachinery/pkg/runtime"
+	"k8s.io/apimachinery/pkg/watch"
+	discoverylisters "k8s.io/client-go/listers/discovery/v1"
+	"k8s.io/client-go/tools/cache"
+
+	"github.com/openconfig/gnmic/pkg/lockers"
+)
+
+const defaultWatchTimeout = 10 * time.Second
+
+func serviceSelector(serviceName string) string {
+	return labels.Set{discoveryv1.LabelServiceName: serviceName}.String()
+}
+
+func (k *k8sLocker) GetServices(ctx context.Context, serviceName string, _ []string) ([]*lockers.Service, error) {
+	list, err := k.clientset.DiscoveryV1().EndpointSlices(k.Cfg.Namespace).List(ctx, metav1.ListOptions{
+		LabelSelector: serviceSelector(serviceName),
+	})
+	if err != nil {
+		return nil, err
+	}
+	slices := make([]*discoveryv1.EndpointSlice, len(list.Items))
+	for i := range list.Items {
+		slices[i] = &list.Items[i]
+	}
+	return endpointSliceServices(slices), nil
+}
+
+func (k *k8sLocker) WatchServices(ctx context.Context, serviceName string, _ []string, sChan chan<- []*lockers.Service, watchTimeout time.Duration) error {
+	if watchTimeout <= 0 {
+		watchTimeout = defaultWatchTimeout
+	}
+	timeoutSeconds := max(int64(watchTimeout.Seconds()), 1)
+	client := k.clientset.DiscoveryV1().EndpointSlices(k.Cfg.Namespace)
+	source := &cache.ListWatch{
+		ListWithContextFunc: func(ctx context.Context, opts metav1.ListOptions) (runtime.Object, error) {
+			opts.LabelSelector = serviceSelector(serviceName)
+			return client.List(ctx, opts)
+		},
+		WatchFuncWithContext: func(ctx context.Context, opts metav1.ListOptions) (watch.Interface, error) {
+			opts.LabelSelector = serviceSelector(serviceName)
+			opts.TimeoutSeconds = &timeoutSeconds
+			return client.Watch(ctx, opts)
+		},
+	}
+	informer := cache.NewSharedIndexInformer(cache.ToListWatcherWithWatchListSemantics(source, k.clientset), &discoveryv1.EndpointSlice{}, 0, cache.Indexers{})
+	changes := make(chan struct{}, 1)
+	notify := func() {
+		select {
+		case changes <- struct{}{}:
+		default:
+		}
+	}
+	_, err := informer.AddEventHandler(cache.ResourceEventHandlerFuncs{
+		AddFunc:    func(interface{}) { notify() },
+		UpdateFunc: func(interface{}, interface{}) { notify() },
+		DeleteFunc: func(interface{}) { notify() },
+	})
+	if err != nil {
+		return err
+	}
+	ctx, cancel := context.WithCancel(ctx)
+	stopped := make(chan struct{})
+	go func() {
+		defer close(stopped)
+		informer.Run(ctx.Done())
+	}()
+	defer func() {
+		cancel()
+		<-stopped
+	}()
+	if !cache.WaitForCacheSync(ctx.Done(), informer.HasSynced) {
+		return ctx.Err()
+	}
+	lister := discoverylisters.NewEndpointSliceLister(informer.GetIndexer())
+	var previous []*lockers.Service
+	initial := true
+	for {
+		slices, err := lister.List(labels.Everything())
+		if err != nil {
+			return err
+		}
+		services := endpointSliceServices(slices)
+		if initial || !reflect.DeepEqual(previous, services) {
+			select {
+			case sChan <- services:
+				previous, initial = services, false
+			case <-ctx.Done():
+				return ctx.Err()
+			}
+		}
+		select {
+		case <-ctx.Done():
+			return ctx.Err()
+		case <-changes:
+		}
+	}
+}
+
+func endpointSliceServices(slices []*discoveryv1.EndpointSlice) []*lockers.Service {
+	peers := make(map[string]*lockers.Service)
+	for _, slice := range slices {
+		port := int32(0)
+		for _, p := range slice.Ports {
+			if p.Port != nil && *p.Port > 0 && (p.Protocol == nil || *p.Protocol == corev1.ProtocolTCP) {
+				port = *p.Port
+				break
+			}
+		}
+		if port == 0 {
+			continue
+		}
+		for _, endpoint := range slice.Endpoints {
+			c := endpoint.Conditions
+			if c.Ready != nil && !*c.Ready || c.Serving != nil && !*c.Serving || c.Terminating != nil && *c.Terminating {
+				continue
+			}
+			for _, address := range endpoint.Addresses {
+				if address == "" {
+					continue
+				}
+				name := address
+				if endpoint.TargetRef != nil && endpoint.TargetRef.Name != "" {
+					name = endpoint.TargetRef.Name
+				}
+				peer := &lockers.Service{
+					ID:      name + "-api",
+					Address: net.JoinHostPort(address, strconv.Itoa(int(port))),
+					Tags:    []string{"instance-name=" + name},
+				}
+				// A Pod can occur in overlapping or dual-stack slices; keep one stable API address.
+				if previous, ok := peers[peer.ID]; !ok || peer.Address < previous.Address {
+					peers[peer.ID] = peer
+				}
+			}
+		}
+	}
+	services := make([]*lockers.Service, 0, len(peers))
+	for _, peer := range peers {
+		services = append(services, peer)
+	}
+	sort.Slice(services, func(i, j int) bool { return services[i].ID < services[j].ID })
+	return services
+}

pkg/lockers/k8s_locker/k8s_discovery_test.go

@@ -0,0 +1,116 @@
+package k8s_locker
+
+import (
+	"context"
+	"errors"
+	"reflect"
+	"testing"
+
+	corev1 "k8s.io/api/core/v1"
+	discoveryv1 "k8s.io/api/discovery/v1"
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	"k8s.io/apimachinery/pkg/runtime"
+	"k8s.io/client-go/kubernetes/fake"
+	ktesting "k8s.io/client-go/testing"
+	"k8s.io/utils/ptr"
+
+	"github.com/openconfig/gnmic/pkg/lockers"
+)
+
+const testNamespace = "telemetry"
+const testService = "test-gnmic-api"
+
+func testSlice(name string, endpoints ...discoveryv1.Endpoint) *discoveryv1.EndpointSlice {
+	return &discoveryv1.EndpointSlice{
+		ObjectMeta:  metav1.ObjectMeta{Name: name, Namespace: testNamespace, Labels: map[string]string{discoveryv1.LabelServiceName: testService}},
+		AddressType: discoveryv1.AddressTypeIPv4,
+		Ports:       []discoveryv1.EndpointPort{{Port: ptr.To(int32(7890)), Protocol: ptr.To(corev1.ProtocolTCP)}},
+		Endpoints:   endpoints,
+	}
+}
+
+func testEndpoint(name, address string) discoveryv1.Endpoint {
+	return discoveryv1.Endpoint{
+		Addresses:  []string{address},
+		TargetRef:  &corev1.ObjectReference{Kind: "Pod", Name: name},
+		Conditions: discoveryv1.EndpointConditions{Ready: ptr.To(true)},
+	}
+}
+
+func testPeer(name, address string) *lockers.Service {
+	return &lockers.Service{ID: name + "-api", Address: address, Tags: []string{"instance-name=" + name}}
+}
+
+func TestGetServicesEndpointSlices(t *testing.T) {
+	a := testEndpoint("gnmic-0", "10.0.0.1")
+	b := testEndpoint("gnmic-1", "10.0.0.2")
+	unready := testEndpoint("unready", "10.0.0.3")
+	unready.Conditions.Ready = ptr.To(false)
+	notServing := testEndpoint("not-serving", "10.0.0.4")
+	notServing.Conditions.Serving = ptr.To(false)
+	terminating := testEndpoint("terminating", "10.0.0.5")
+	terminating.Conditions.Terminating = ptr.To(true)
+	unknown := testEndpoint("unknown", "10.0.0.6")
+	unknown.Conditions = discoveryv1.EndpointConditions{}
+	foreign := testSlice("foreign", testEndpoint("foreign", "10.0.1.1"))
+	foreign.Labels[discoveryv1.LabelServiceName] = "another-service"
+	otherNamespace := testSlice("other-namespace", testEndpoint("other-namespace", "10.0.1.2"))
+	otherNamespace.Namespace = "other"
+	tests := []struct {
+		name    string
+		objects []runtime.Object
+		want    []*lockers.Service
+	}{
+		{name: "absent", want: []*lockers.Service{}},
+		{name: "empty", objects: []runtime.Object{testSlice("empty")}, want: []*lockers.Service{}},
+		{name: "single", objects: []runtime.Object{testSlice("one", a)}, want: []*lockers.Service{testPeer("gnmic-0", "10.0.0.1:7890")}},
+		{name: "multi-slice", objects: []runtime.Object{testSlice("two", b, a), testSlice("one", a), foreign, otherNamespace},
+			want: []*lockers.Service{testPeer("gnmic-0", "10.0.0.1:7890"), testPeer("gnmic-1", "10.0.0.2:7890")}},
+		{name: "conditions", objects: []runtime.Object{testSlice("conditions", unready, notServing, terminating, unknown)},
+			want: []*lockers.Service{testPeer("unknown", "10.0.0.6:7890")}},
+	}
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			client := fake.NewClientset(tt.objects...)
+			k := &k8sLocker{clientset: client, Cfg: &config{Namespace: testNamespace}}
+			got, err := k.GetServices(t.Context(), testService, nil)
+			if err != nil || !reflect.DeepEqual(got, tt.want) {
+				t.Fatalf("GetServices() = %#v, %v; want %#v", got, err, tt.want)
+			}
+			for _, action := range client.Actions() {
+				if action.GetResource().Group != discoveryv1.GroupName || action.GetResource().Resource != "endpointslices" {
+					t.Fatalf("unexpected API action: %#v", action)
+				}
+			}
+		})
+	}
+}
+
+func TestEndpointSliceAddressesAndPorts(t *testing.T) {
+	ipv6 := testSlice("ipv6", testEndpoint("gnmic-0", "2001:db8::1"), testEndpoint("gnmic-1", "2001:db8::2"))
+	ipv6.AddressType = discoveryv1.AddressTypeIPv6
+	ipv4 := testSlice("ipv4", testEndpoint("gnmic-0", "10.0.0.1"))
+	ipv4.Ports = []discoveryv1.EndpointPort{
+		{Port: ptr.To(int32(53)), Protocol: ptr.To(corev1.ProtocolUDP)},
+		{Port: ptr.To(int32(7890))},
+	}
+	missingPort := testSlice("missing-port", testEndpoint("missing-port", "10.0.0.3"))
+	missingPort.Ports[0].Port = nil
+	missingRef := testSlice("missing-ref", discoveryv1.Endpoint{Addresses: []string{"10.0.0.4", ""}})
+	want := []*lockers.Service{testPeer("10.0.0.4", "10.0.0.4:7890"), testPeer("gnmic-0", "10.0.0.1:7890"), testPeer("gnmic-1", "[2001:db8::2]:7890")}
+	for _, slices := range [][]*discoveryv1.EndpointSlice{{ipv6, ipv4, missingPort, missingRef}, {missingRef, missingPort, ipv4, ipv6}} {
+		if got := endpointSliceServices(slices); !reflect.DeepEqual(got, want) {
+			t.Fatalf("services = %#v; want %#v", got, want)
+		}
+	}
+}
+
+func TestGetServicesListError(t *testing.T) {
+	client := fake.NewClientset()
+	expected := errors.New("list unavailable")
+	client.PrependReactor("list", "endpointslices", func(ktesting.Action) (bool, runtime.Object, error) { return true, nil, expected })
+	k := &k8sLocker{clientset: client, Cfg: &config{Namespace: testNamespace}}
+	if _, err := k.GetServices(context.Background(), testService, nil); !errors.Is(err, expected) {
+		t.Fatalf("error = %v; want %v", err, expected)
+	}
+}