From 92a5d658fa2cdb4b418e5e2ab0bdbaf7e67bdd0b Mon Sep 17 00:00:00 2001 From: pood1e Date: Sun, 6 Sep 2026 16:22:38 +0800 Subject: [PATCH 1/2] fix: discover Kubernetes locker peers through EndpointSlices --- docs/user_guide/HA.md | 2 +- docs/user_guide/ha_kubernetes.md | 97 +++++++++ mkdocs.yml | 1 + pkg/lockers/k8s_locker/k8s_discovery.go | 166 +++++++++++++++ pkg/lockers/k8s_locker/k8s_discovery_test.go | 116 ++++++++++ .../k8s_locker/k8s_discovery_watch_test.go | 199 ++++++++++++++++++ pkg/lockers/k8s_locker/k8s_locker.go | 2 +- pkg/lockers/k8s_locker/k8s_registration.go | 128 ----------- 8 files changed, 581 insertions(+), 130 deletions(-) create mode 100644 docs/user_guide/ha_kubernetes.md create mode 100644 pkg/lockers/k8s_locker/k8s_discovery.go create mode 100644 pkg/lockers/k8s_locker/k8s_discovery_test.go create mode 100644 pkg/lockers/k8s_locker/k8s_discovery_watch_test.go diff --git a/docs/user_guide/HA.md b/docs/user_guide/HA.md index 926e0ee94..b30ae99b7 100644 --- a/docs/user_guide/HA.md +++ b/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 diff --git a/docs/user_guide/ha_kubernetes.md b/docs/user_guide/ha_kubernetes.md new file mode 100644 index 000000000..e16ed535a --- /dev/null +++ b/docs/user_guide/ha_kubernetes.md @@ -0,0 +1,97 @@ +# 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 + lease-duration: 10s + renew-period: 5s + retry-timer: 2s +``` + +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 their Lease holders. +The API Service name is `-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, 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. diff --git a/mkdocs.yml b/mkdocs.yml index caa8c45fc..30ab90083 100644 --- a/mkdocs.yml +++ b/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 diff --git a/pkg/lockers/k8s_locker/k8s_discovery.go b/pkg/lockers/k8s_locker/k8s_discovery.go new file mode 100644 index 000000000..fa99ca869 --- /dev/null +++ b/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 +} diff --git a/pkg/lockers/k8s_locker/k8s_discovery_test.go b/pkg/lockers/k8s_locker/k8s_discovery_test.go new file mode 100644 index 000000000..897644466 --- /dev/null +++ b/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) + } +} diff --git a/pkg/lockers/k8s_locker/k8s_discovery_watch_test.go b/pkg/lockers/k8s_locker/k8s_discovery_watch_test.go new file mode 100644 index 000000000..c22e6a90c --- /dev/null +++ b/pkg/lockers/k8s_locker/k8s_discovery_watch_test.go @@ -0,0 +1,199 @@ +package k8s_locker + +import ( + "context" + "errors" + "reflect" + "testing" + "time" + + discoveryv1 "k8s.io/api/discovery/v1" + metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" + "k8s.io/apimachinery/pkg/runtime" + "k8s.io/apimachinery/pkg/watch" + "k8s.io/client-go/kubernetes/fake" + ktesting "k8s.io/client-go/testing" + "k8s.io/utils/ptr" + + "github.com/openconfig/gnmic/pkg/lockers" +) + +func receivePeers(t *testing.T, ch <-chan []*lockers.Service, want ...*lockers.Service) { + t.Helper() + if want == nil { + want = []*lockers.Service{} + } + select { + case got := <-ch: + if !reflect.DeepEqual(got, want) { + t.Fatalf("snapshot = %#v; want %#v", got, want) + } + case <-time.After(10 * time.Second): + t.Fatal("no peer snapshot received") + } +} + +func startDiscovery(t *testing.T, client *fake.Clientset) (chan []*lockers.Service, context.CancelFunc, chan error) { + t.Helper() + k := &k8sLocker{clientset: client, Cfg: &config{Namespace: testNamespace}} + ctx, cancel := context.WithCancel(t.Context()) + changes := make(chan []*lockers.Service) + done := make(chan error, 1) + go func() { done <- k.WatchServices(ctx, testService, nil, changes, time.Second) }() + t.Cleanup(cancel) + return changes, cancel, done +} + +func stoppedDiscovery(t *testing.T, done <-chan error) { + t.Helper() + select { + case err := <-done: + if !errors.Is(err, context.Canceled) { + t.Fatalf("WatchServices() error = %v", err) + } + case <-time.After(5 * time.Second): + t.Fatal("discovery did not stop") + } +} + +func TestWatchServicesSliceLifecycle(t *testing.T) { + a := testSlice("a", testEndpoint("gnmic-0", "10.0.0.1")) + b := testSlice("b", testEndpoint("gnmic-1", "10.0.0.2")) + client := fake.NewClientset(a, b) + changes, cancel, done := startDiscovery(t, client) + receivePeers(t, changes, testPeer("gnmic-0", "10.0.0.1:7890"), testPeer("gnmic-1", "10.0.0.2:7890")) + endpointSlices := client.DiscoveryV1().EndpointSlices(testNamespace) + a = a.DeepCopy() + a.Endpoints[0].Conditions.Ready = ptr.To(false) + if _, err := endpointSlices.Update(t.Context(), a, metav1.UpdateOptions{}); err != nil { + t.Fatal(err) + } + receivePeers(t, changes, testPeer("gnmic-1", "10.0.0.2:7890")) + c := testSlice("c", testEndpoint("gnmic-2", "10.0.0.3")) + if _, err := endpointSlices.Create(t.Context(), c, metav1.CreateOptions{}); err != nil { + t.Fatal(err) + } + receivePeers(t, changes, testPeer("gnmic-1", "10.0.0.2:7890"), testPeer("gnmic-2", "10.0.0.3:7890")) + if err := endpointSlices.Delete(t.Context(), b.Name, metav1.DeleteOptions{}); err != nil { + t.Fatal(err) + } + receivePeers(t, changes, testPeer("gnmic-2", "10.0.0.3:7890")) + if err := endpointSlices.Delete(t.Context(), c.Name, metav1.DeleteOptions{}); err != nil { + t.Fatal(err) + } + receivePeers(t, changes) + cancel() + stoppedDiscovery(t, done) +} + +func TestWatchServicesInitiallyEmpty(t *testing.T) { + client := fake.NewClientset() + changes, cancel, done := startDiscovery(t, client) + receivePeers(t, changes) + if _, err := client.DiscoveryV1().EndpointSlices(testNamespace).Create(t.Context(), testSlice("a", testEndpoint("gnmic-0", "10.0.0.1")), metav1.CreateOptions{}); err != nil { + t.Fatal(err) + } + receivePeers(t, changes, testPeer("gnmic-0", "10.0.0.1:7890")) + cancel() + stoppedDiscovery(t, done) +} + +func TestWatchServicesRelistsExpiredVersion(t *testing.T) { + a := testSlice("a", testEndpoint("gnmic-0", "10.0.0.1")) + client := fake.NewClientset(a) + watches := make(chan *watch.RaceFreeFakeWatcher, 10) + client.PrependWatchReactor("endpointslices", func(action ktesting.Action) (bool, watch.Interface, error) { + opts := action.(ktesting.WatchAction).GetWatchRestrictions() + if opts.Labels.String() != serviceSelector(testService) { + t.Errorf("watch selector = %s", opts.Labels) + } + w := watch.NewRaceFreeFake() + watches <- w + return true, w, nil + }) + changes, cancel, done := startDiscovery(t, client) + receivePeers(t, changes, testPeer("gnmic-0", "10.0.0.1:7890")) + var first *watch.RaceFreeFakeWatcher + select { + case first = <-watches: + case <-time.After(5 * time.Second): + t.Fatal("watch did not start") + } + a = a.DeepCopy() + a.Endpoints = []discoveryv1.Endpoint{testEndpoint("gnmic-1", "10.0.0.2")} + if err := client.Tracker().Update(discoveryv1.SchemeGroupVersion.WithResource("endpointslices"), a, testNamespace); err != nil { + t.Fatal(err) + } + first.Error(&metav1.Status{Status: metav1.StatusFailure, Reason: metav1.StatusReasonExpired, Code: 410}) + first.Stop() + receivePeers(t, changes, testPeer("gnmic-1", "10.0.0.2:7890")) + var second *watch.RaceFreeFakeWatcher + select { + case second = <-watches: + case <-time.After(5 * time.Second): + t.Fatal("watch did not restart") + } + second.Delete(a) + receivePeers(t, changes) + cancel() + stoppedDiscovery(t, done) +} + +func TestWatchServicesCancellation(t *testing.T) { + for _, name := range []string{"blocked-publisher", "list-retry"} { + t.Run(name, func(t *testing.T) { + client := fake.NewClientset() + attempted := make(chan struct{}, 1) + client.PrependReactor("list", "endpointslices", func(ktesting.Action) (bool, runtime.Object, error) { + select { + case attempted <- struct{}{}: + default: + } + if name == "list-retry" { + return true, nil, errors.New("temporary list error") + } + return false, nil, nil + }) + _, cancel, done := startDiscovery(t, client) + select { + case <-attempted: + case <-time.After(5 * time.Second): + t.Fatal("list did not start") + } + cancel() + stoppedDiscovery(t, done) + }) + } +} + +func TestWatchServicesReconnectsWithoutDroppingPeers(t *testing.T) { + a := testSlice("a", testEndpoint("gnmic-0", "10.0.0.1")) + client := fake.NewClientset(a) + watches := make(chan *watch.RaceFreeFakeWatcher, 10) + client.PrependWatchReactor("endpointslices", func(ktesting.Action) (bool, watch.Interface, error) { + w := watch.NewRaceFreeFake() + watches <- w + return true, w, nil + }) + changes, cancel, done := startDiscovery(t, client) + receivePeers(t, changes, testPeer("gnmic-0", "10.0.0.1:7890")) + var first *watch.RaceFreeFakeWatcher + select { + case first = <-watches: + case <-time.After(5 * time.Second): + t.Fatal("watch did not start") + } + first.Stop() + var second *watch.RaceFreeFakeWatcher + select { + case second = <-watches: + case <-time.After(5 * time.Second): + t.Fatal("watch did not reconnect") + } + a = a.DeepCopy() + a.Endpoints[0].Addresses = []string{"10.0.0.2"} + second.Modify(a) + receivePeers(t, changes, testPeer("gnmic-0", "10.0.0.2:7890")) + cancel() + stoppedDiscovery(t, done) +} diff --git a/pkg/lockers/k8s_locker/k8s_locker.go b/pkg/lockers/k8s_locker/k8s_locker.go index 51fb74e5d..073fedc32 100644 --- a/pkg/lockers/k8s_locker/k8s_locker.go +++ b/pkg/lockers/k8s_locker/k8s_locker.go @@ -52,7 +52,7 @@ func init() { type k8sLocker struct { Cfg *config - clientset *kubernetes.Clientset + clientset kubernetes.Interface logger *slog.Logger m *sync.RWMutex acquiredlocks map[string]*lock diff --git a/pkg/lockers/k8s_locker/k8s_registration.go b/pkg/lockers/k8s_locker/k8s_registration.go index 1d0823543..775a1823e 100644 --- a/pkg/lockers/k8s_locker/k8s_registration.go +++ b/pkg/lockers/k8s_locker/k8s_registration.go @@ -10,21 +10,15 @@ package k8s_locker import ( "context" - "fmt" "strings" "time" - corev1 "k8s.io/api/core/v1" "k8s.io/apimachinery/pkg/api/errors" metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" - "k8s.io/apimachinery/pkg/fields" - "k8s.io/apimachinery/pkg/watch" "github.com/openconfig/gnmic/pkg/lockers" ) -const defaultWatchTimeout = 10 * time.Second - func (k *k8sLocker) Register(ctx context.Context, s *lockers.ServiceRegistration) error { return nil } @@ -33,128 +27,6 @@ func (k *k8sLocker) Deregister(s string) error { return nil } -func (k *k8sLocker) WatchServices(ctx context.Context, serviceName string, tags []string, sChan chan<- []*lockers.Service, watchTimeout time.Duration) error { - if watchTimeout <= 0 { - watchTimeout = defaultWatchTimeout - } - resourceVersion := "" - var err error - for { - select { - case <-ctx.Done(): - return ctx.Err() - default: - resourceVersion, err = k.watch(ctx, serviceName, tags, sChan, watchTimeout, resourceVersion) - if err != nil { - k.logger.Warn("watch ended with error", "err", err) - time.Sleep(k.Cfg.RetryTimer) - } else if k.Cfg.Debug { - k.logger.Debug("watch timed out") - } - } - } -} - -func (k *k8sLocker) watch(ctx context.Context, serviceName string, _ []string, sChan chan<- []*lockers.Service, watchTimeout time.Duration, resourceVersion string) (string, error) { - timeoutSeconds := int64(watchTimeout.Seconds()) - listopts := metav1.ListOptions{ - FieldSelector: fields.OneTermEqualSelector(metav1.ObjectNameField, serviceName).String(), - ResourceVersion: resourceVersion, - TimeoutSeconds: &timeoutSeconds, - } - if k.Cfg.Debug { - if resourceVersion == "" { - k.logger.Debug("starting watch beginning with unspecified resource version") - } else { - k.logger.Debug("starting watch", "resource_version", resourceVersion) - } - } - watched, err := k.clientset.CoreV1().Endpoints(k.Cfg.Namespace).Watch(ctx, listopts) - if err != nil { - return "", err - } - defer watched.Stop() - - watchChan := watched.ResultChan() - - for { - select { - case <-ctx.Done(): - return "", ctx.Err() - case event := <-watchChan: - switch event.Type { - case watch.Modified, watch.Added: - endpoints, ok := event.Object.(*corev1.Endpoints) - if !ok { - // this ought not to happen, but we should probably - // start from scratch next time in case it does - return "", fmt.Errorf("error converting watch result to an endpoint") - } - resourceVersion = endpoints.ResourceVersion - if k.Cfg.Debug { - k.logger.Debug("received watch event", "event_type", event.Type, "resource_version", resourceVersion) - } - svcs, err := parseEndpoint(endpoints) - if err != nil { - return "", err - } - sChan <- svcs - case "": - // reached the timeout. return the version we last saw so - // we can resume watching - return resourceVersion, nil - default: - // something else happened, including maybe the object we - // were watching being deleted. we'll need to start the - // next watch from scratch, so don't return the resource - // version - return "", fmt.Errorf("unexpected watch event: %s", event.Type) - } - } - } -} - -func parseEndpoint(endpoint *corev1.Endpoints) ([]*lockers.Service, error) { - // the service should only have a single port number assigned, so - // all subsets should have the port number we're looking for - if len(endpoint.Subsets) <= 0 { - return nil, fmt.Errorf("no subsets found in endpoint for service %s", endpoint.Name) - } - if len(endpoint.Subsets[0].Ports) <= 0 { - return nil, fmt.Errorf("no ports found for service %s", endpoint.Name) - } - port := endpoint.Subsets[0].Ports[0].Port - - services := make([]*lockers.Service, 0, len(endpoint.Subsets[0].Addresses)) - for _, subset := range endpoint.Subsets { - for _, addr := range subset.Addresses { - targetName := addr.IP - if addr.TargetRef != nil { - targetName = addr.TargetRef.Name - } - ls := &lockers.Service{ - ID: fmt.Sprintf("%s-api", targetName), - Address: fmt.Sprintf("%s:%d", addr.IP, port), - Tags: []string{ - fmt.Sprintf("instance-name=%s", targetName), - }, - } - services = append(services, ls) - } - } - - return services, nil -} - -func (k *k8sLocker) GetServices(ctx context.Context, serviceName string, tags []string) ([]*lockers.Service, error) { - ep, err := k.clientset.CoreV1().Endpoints(k.Cfg.Namespace).Get(ctx, serviceName, metav1.GetOptions{}) - if err != nil { - return nil, err - } - - return parseEndpoint(ep) -} - func (k *k8sLocker) IsLocked(ctx context.Context, key string) (bool, error) { key = strings.ReplaceAll(key, "/", "-") ol, err := k.clientset.CoordinationV1().Leases(k.Cfg.Namespace).Get(ctx, key, metav1.GetOptions{}) From 9ea52f6610a5083c3dbd485fd80b0c848f3c90c6 Mon Sep 17 00:00:00 2001 From: pood1e Date: Sun, 6 Sep 2026 18:12:27 +0800 Subject: [PATCH 2/2] Keep Kubernetes discovery example focused on identity and RBAC --- docs/user_guide/ha_kubernetes.md | 8 +++----- 1 file changed, 3 insertions(+), 5 deletions(-) diff --git a/docs/user_guide/ha_kubernetes.md b/docs/user_guide/ha_kubernetes.md index e16ed535a..db2efd808 100644 --- a/docs/user_guide/ha_kubernetes.md +++ b/docs/user_guide/ha_kubernetes.md @@ -15,13 +15,11 @@ clustering: locker: type: k8s namespace: telemetry - lease-duration: 10s - renew-period: 5s - retry-timer: 2s ``` 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 their Lease holders. +name must match the Pod name so that discovered peers match the instance names +stored with their Leases. The API Service name is `-gnmic-api`; its selector must match the collector Pods. Expose a single TCP API port on this Service: @@ -70,7 +68,7 @@ rules: verbs: [get, list, watch] - apiGroups: [coordination.k8s.io] resources: [leases] - verbs: [get, list, create, update, delete] + verbs: [get, list, watch, create, update, delete] --- apiVersion: rbac.authorization.k8s.io/v1 kind: RoleBinding