From ba4831a6901c4d0672472a603e4b44248afc3fb0 Mon Sep 17 00:00:00 2001 From: Ed Bartosh Date: Mon, 14 Sep 2026 10:37:17 +0300 Subject: [PATCH 1/3] refactor: move pkg/agent/watch to pkg/kubernetes/watch Kubernetes-related helpers are being consolidated under pkg/kubernetes so they can be shared beyond the agent. Move the watch package there first, as a plain rename, ahead of adding the new kubernetes client package. Signed-off-by: Ed Bartosh --- pkg/agent/agent.go | 2 +- pkg/{agent => kubernetes}/watch/file.go | 0 pkg/{agent => kubernetes}/watch/object.go | 0 pkg/{agent => kubernetes}/watch/watch.go | 2 +- pkg/kubernetes/watch/watch_test.go | 196 ++++++++++++++++++++++ 5 files changed, 198 insertions(+), 2 deletions(-) rename pkg/{agent => kubernetes}/watch/file.go (100%) rename pkg/{agent => kubernetes}/watch/object.go (100%) rename pkg/{agent => kubernetes}/watch/watch.go (97%) create mode 100644 pkg/kubernetes/watch/watch_test.go diff --git a/pkg/agent/agent.go b/pkg/agent/agent.go index f577e7b87..8a3be0245 100644 --- a/pkg/agent/agent.go +++ b/pkg/agent/agent.go @@ -33,8 +33,8 @@ import ( nrtapi "github.com/containers/nri-plugins/pkg/agent/nrtapi" "github.com/containers/nri-plugins/pkg/agent/podresapi" - "github.com/containers/nri-plugins/pkg/agent/watch" cfgapi "github.com/containers/nri-plugins/pkg/apis/config/v1alpha1" + "github.com/containers/nri-plugins/pkg/kubernetes/watch" k8sclient "k8s.io/client-go/kubernetes" logger "github.com/containers/nri-plugins/pkg/log" diff --git a/pkg/agent/watch/file.go b/pkg/kubernetes/watch/file.go similarity index 100% rename from pkg/agent/watch/file.go rename to pkg/kubernetes/watch/file.go diff --git a/pkg/agent/watch/object.go b/pkg/kubernetes/watch/object.go similarity index 100% rename from pkg/agent/watch/object.go rename to pkg/kubernetes/watch/object.go diff --git a/pkg/agent/watch/watch.go b/pkg/kubernetes/watch/watch.go similarity index 97% rename from pkg/agent/watch/watch.go rename to pkg/kubernetes/watch/watch.go index 66735213f..473725c7d 100644 --- a/pkg/agent/watch/watch.go +++ b/pkg/kubernetes/watch/watch.go @@ -35,5 +35,5 @@ const ( ) var ( - log = logger.Get("agent") + log = logger.Get("watch") ) diff --git a/pkg/kubernetes/watch/watch_test.go b/pkg/kubernetes/watch/watch_test.go new file mode 100644 index 000000000..0cdad4631 --- /dev/null +++ b/pkg/kubernetes/watch/watch_test.go @@ -0,0 +1,196 @@ +// Copyright The NRI Plugins Authors. All Rights Reserved. +// +// 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 watch + +import ( + "context" + "os" + "path/filepath" + "testing" + "time" + + metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" + "k8s.io/apimachinery/pkg/runtime" + k8swatch "k8s.io/apimachinery/pkg/watch" +) + +// Compile-time assertions: the re-exported aliases really are the same +// types and values as the upstream ones. These lines would fail to +// compile if the aliases drifted. +var ( + _ Interface = k8swatch.Interface(nil) + _ EventType = k8swatch.EventType("") + _ Event = k8swatch.Event{} +) + +func TestEventTypeConstants(t *testing.T) { + tests := []struct { + name string + local EventType + want k8swatch.EventType + }{ + {"Added", Added, k8swatch.Added}, + {"Modified", Modified, k8swatch.Modified}, + {"Deleted", Deleted, k8swatch.Deleted}, + {"Bookmark", Bookmark, k8swatch.Bookmark}, + {"Error", Error, k8swatch.Error}, + } + for _, tc := range tests { + if tc.local != tc.want { + t.Errorf("%s: got %q, want %q", tc.name, tc.local, tc.want) + } + } +} + +// TestObject_HappyPath confirms events sent to the fake Interface +// returned by CreateFn make it out through Object's ResultChan. +func TestObject_HappyPath(t *testing.T) { + fake := k8swatch.NewFake() + create := func(ctx context.Context, ns, name string) (Interface, error) { + return fake, nil + } + + ctx, cancel := context.WithCancel(t.Context()) + defer cancel() + + ow, err := Object(ctx, "default", "cm-x", create) + if err != nil { + t.Fatalf("Object returned error: %v", err) + } + defer ow.Stop() + + // Push an Added and a Modified event through the fake; expect both. + go func() { + fake.Add(&metav1.Status{Message: "added"}) + fake.Modify(&metav1.Status{Message: "modified"}) + }() + + got := drainEvents(t, ow.ResultChan(), 2, 2*time.Second) + if len(got) < 2 { + t.Fatalf("expected at least 2 events, got %d: %+v", len(got), got) + } + if got[0].Type != Added { + t.Errorf("event[0].Type = %q, want %q", got[0].Type, Added) + } + if got[1].Type != Modified { + t.Errorf("event[1].Type = %q, want %q", got[1].Type, Modified) + } +} + +// TestObject_StopIdempotent verifies Stop() can be called more than +// once without panicking. The existing implementation uses sync.Once +// on the internal stop path; this is a defensive guard. +func TestObject_StopIdempotent(t *testing.T) { + fake := k8swatch.NewFake() + create := func(ctx context.Context, ns, name string) (Interface, error) { + return fake, nil + } + + ow, err := Object(t.Context(), "default", "cm-x", create) + if err != nil { + t.Fatalf("Object returned error: %v", err) + } + ow.Stop() + ow.Stop() // must not panic +} + +// TestFile_CreateVsWriteEventTypes is a defensive guard against +// regressing the Create-emits-Added / Write-emits-Modified distinction. +// PR #536's parallel implementation ships a copy-paste bug where both +// cases emit Added; we don't have that bug today, and this test +// ensures we don't accidentally introduce it in a future edit. +func TestFile_CreateVsWriteEventTypes(t *testing.T) { + dir := t.TempDir() + file := filepath.Join(dir, "watched.yaml") + + unmarshal := func(data []byte, name string) (runtime.Object, error) { + return &metav1.Status{Message: string(data)}, nil + } + + fw, err := File(file, unmarshal) + if err != nil { + t.Fatalf("File returned error: %v", err) + } + defer fw.Stop() + + // Step 1: create the file — expect an Added event from the fsnotify + // Create. + if err := os.WriteFile(file, []byte("v1"), 0o600); err != nil { + t.Fatalf("initial write: %v", err) + } + + // Step 2: modify the file using O_WRONLY|O_APPEND so fsnotify + // emits only Write (not Create). os.WriteFile uses O_CREATE|O_TRUNC + // which fires a Create event even for an existing file — that would + // give us a second Added instead of the Modified we're testing for. + time.Sleep(150 * time.Millisecond) + f, err := os.OpenFile(file, os.O_WRONLY|os.O_APPEND, 0) + if err != nil { + t.Fatalf("open for append: %v", err) + } + if _, err := f.Write([]byte("+v2")); err != nil { + _ = f.Close() + t.Fatalf("append write: %v", err) + } + if err := f.Close(); err != nil { + t.Fatalf("close after append: %v", err) + } + + // Drain up to 4 events (initial-Added-from-run + Create-Added + Write-Modified + slack). + got := drainEvents(t, fw.ResultChan(), 4, 3*time.Second) + sawAdded, sawModified := false, false + for _, ev := range got { + if ev.Type == Added { + sawAdded = true + } + if ev.Type == Modified { + sawModified = true + } + } + if !sawAdded { + t.Errorf("expected at least one Added event; got types: %v", eventTypes(got)) + } + if !sawModified { + t.Errorf("expected at least one Modified event (write path); got types: %v", eventTypes(got)) + } +} + +// drainEvents receives up to `want` events from ch or times out. +// Returns whatever it received. +func drainEvents(t *testing.T, ch <-chan Event, want int, timeout time.Duration) []Event { + t.Helper() + got := make([]Event, 0, want) + deadline := time.After(timeout) + for len(got) < want { + select { + case ev, ok := <-ch: + if !ok { + return got + } + got = append(got, ev) + case <-deadline: + return got + } + } + return got +} + +func eventTypes(evs []Event) []EventType { + out := make([]EventType, len(evs)) + for i, ev := range evs { + out[i] = ev.Type + } + return out +} From a7cef880c9eac04ef7869d33c1d806d2b1cd06c6 Mon Sep 17 00:00:00 2001 From: Ed Bartosh Date: Mon, 14 Sep 2026 10:37:40 +0300 Subject: [PATCH 2/3] kubernetes/client: add wrapped kubernetes client Add a pkg/kubernetes/client package that bundles the REST config, HTTP client, and clientset construction (from a kubeconfig file or in-cluster config) behind a single Client type, so callers no longer need to wire up all three separately. Not yet used by anything; pkg/agent will be switched over next. Signed-off-by: Ed Bartosh --- pkg/kubernetes/client/client.go | 219 +++++++++++ pkg/kubernetes/client/client_test.go | 370 ++++++++++++++++++ .../client/testdata/kubeconfig-example.yaml | 17 + 3 files changed, 606 insertions(+) create mode 100644 pkg/kubernetes/client/client.go create mode 100644 pkg/kubernetes/client/client_test.go create mode 100644 pkg/kubernetes/client/testdata/kubeconfig-example.yaml diff --git a/pkg/kubernetes/client/client.go b/pkg/kubernetes/client/client.go new file mode 100644 index 000000000..27c65f239 --- /dev/null +++ b/pkg/kubernetes/client/client.go @@ -0,0 +1,219 @@ +/* +Copyright The NRI Plugins Authors. + +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 client builds a *kubernetes.Clientset from a kubeconfig file or +// in-cluster credentials, and exposes the REST config and HTTP client it +// was built from so callers can share one client. +package client + +import ( + "errors" + "net/http" + "strings" + + "k8s.io/client-go/kubernetes" + "k8s.io/client-go/rest" + "k8s.io/client-go/tools/clientcmd" +) + +// Wire content types accepted by the Kubernetes API server. +const ( + ContentTypeJSON = "application/json" + ContentTypeProtobuf = "application/vnd.kubernetes.protobuf" +) + +// Client wraps a Kubernetes clientset together with the REST config and +// HTTP client it was built from. Use the embedded Clientset directly for +// API calls, or HttpClient()/RestConfig() to build other clients sharing +// the same transport. +type Client struct { + cfg *rest.Config + http *http.Client + *kubernetes.Clientset +} + +// Option configures a Client during construction via New. Options apply +// in order; config-dependent options (WithContentType, +// WithAcceptContentTypes) require a config-source option (WithKubeConfig, +// WithInClusterConfig, or WithRestConfig) earlier in the list. +type Option func(*Client) error + +// errNoConfigSet is returned by options that require the REST config +// to be present but are called before any config-source option. +var errNoConfigSet = errors.New("option requires REST config; pass a config-source option (WithKubeConfig, WithInClusterConfig, or WithRestConfig) before this option") + +// GetConfigForFile returns a REST configuration parsed from the given +// kubeconfig file path. Thin wrapper over clientcmd.BuildConfigFromFlags +// exposed for callers that need a config but not a full Client. +func GetConfigForFile(kubeConfig string) (*rest.Config, error) { + return clientcmd.BuildConfigFromFlags("", kubeConfig) +} + +// InClusterConfig returns the REST configuration for the pod's service +// account, if the process is running inside a Kubernetes cluster. +// Returns rest.ErrNotInCluster (wrapped) when not in a cluster. +func InClusterConfig() (*rest.Config, error) { + return rest.InClusterConfig() +} + +// New constructs a Client by applying the given options in order, +// defaulting to WithInClusterConfig() if none set a REST config. +func New(options ...Option) (*Client, error) { + c := &Client{} + + for _, o := range options { + if err := o(c); err != nil { + return nil, err + } + } + + if c.cfg == nil { + if err := WithInClusterConfig()(c); err != nil { + return nil, err + } + } + + if c.http == nil { + hc, err := rest.HTTPClientFor(c.cfg) + if err != nil { + return nil, err + } + c.http = hc + } + + cs, err := kubernetes.NewForConfigAndClient(c.cfg, c.http) + if err != nil { + return nil, err + } + c.Clientset = cs + + return c, nil +} + +// WithKubeConfig returns an Option that resolves the REST config from +// the given kubeconfig file. +func WithKubeConfig(file string) Option { + return func(c *Client) error { + cfg, err := GetConfigForFile(file) + if err != nil { + return err + } + return WithRestConfig(cfg)(c) + } +} + +// WithInClusterConfig returns an Option that resolves the REST config +// from the pod's service-account credentials. +func WithInClusterConfig() Option { + return func(c *Client) error { + cfg, err := InClusterConfig() + if err != nil { + return err + } + return WithRestConfig(cfg)(c) + } +} + +// WithKubeOrInClusterConfig resolves the REST config from the given +// kubeconfig file if non-empty, or from in-cluster credentials otherwise. +func WithKubeOrInClusterConfig(file string) Option { + if file == "" { + return WithInClusterConfig() + } + return WithKubeConfig(file) +} + +// WithRestConfig uses a deep copy (via rest.CopyConfig) of the given REST +// config, so the caller keeps ownership of the original. +func WithRestConfig(cfg *rest.Config) Option { + return func(c *Client) error { + if cfg == nil { + return errors.New("rest config must not be nil") + } + c.cfg = rest.CopyConfig(cfg) + return nil + } +} + +// WithHttpClient returns an Option that uses the given pre-built HTTP +// client. Useful when multiple components should share one client +// (and therefore its connection pool). +func WithHttpClient(hc *http.Client) Option { + return func(c *Client) error { + c.http = hc + return nil + } +} + +// WithAcceptContentTypes sets the Accept content types to negotiate with +// the API server, joined with commas. Requires a config-source option +// earlier in the list. +func WithAcceptContentTypes(contentTypes ...string) Option { + return func(c *Client) error { + if c.cfg == nil { + return errNoConfigSet + } + c.cfg.AcceptContentTypes = strings.Join(contentTypes, ",") + return nil + } +} + +// WithContentType sets the wire content type used for requests. Requires +// a config-source option earlier in the list. +func WithContentType(contentType string) Option { + return func(c *Client) error { + if c.cfg == nil { + return errNoConfigSet + } + c.cfg.ContentType = contentType + return nil + } +} + +// RestConfig returns a copy of the Client's REST config. Top-level and +// value-typed nested fields may be freely overwritten, but nested +// maps/slices (e.g. TLSClientConfig.CAData) share storage with the +// Client's internal config and must not be mutated. +func (c *Client) RestConfig() *rest.Config { + return rest.CopyConfig(c.cfg) +} + +// HttpClient returns the Client's underlying HTTP client, e.g. for +// constructing other clients that share the same transport. +func (c *Client) HttpClient() *http.Client { + return c.http +} + +// K8sClient returns the Client's underlying *kubernetes.Clientset. +// Callers may alternatively use the embedded Clientset directly on +// the Client value. +func (c *Client) K8sClient() *kubernetes.Clientset { + return c.Clientset +} + +// Close releases resources held by the Client. Safe to call on a nil or +// already-closed Client. +func (c *Client) Close() { + if c == nil { + return + } + if c.http != nil { + c.http.CloseIdleConnections() + } + c.cfg = nil + c.http = nil + c.Clientset = nil +} diff --git a/pkg/kubernetes/client/client_test.go b/pkg/kubernetes/client/client_test.go new file mode 100644 index 000000000..3990af7f3 --- /dev/null +++ b/pkg/kubernetes/client/client_test.go @@ -0,0 +1,370 @@ +/* +Copyright The NRI Plugins Authors. + +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 client + +import ( + "errors" + "net/http" + "os" + "path/filepath" + "testing" + + "k8s.io/client-go/rest" +) + +// fixtureKubeconfig is the path to a minimal valid kubeconfig used by tests +// that need a config source without contacting a real API server. +const fixtureKubeconfig = "testdata/kubeconfig-example.yaml" + +// skipIfInCluster fails-fast for tests that assert not-in-cluster behavior +// when they happen to run inside a Pod (e.g. e2e CI). +func skipIfInCluster(t *testing.T) { + t.Helper() + if os.Getenv("KUBERNETES_SERVICE_HOST") != "" { + t.Skip("running inside a Kubernetes Pod; skipping not-in-cluster test") + } +} + +func TestGetConfigForFile_Success(t *testing.T) { + cfg, err := GetConfigForFile(fixtureKubeconfig) + if err != nil { + t.Fatalf("GetConfigForFile(%q) returned error: %v", fixtureKubeconfig, err) + } + if cfg == nil { + t.Fatal("GetConfigForFile returned nil config with no error") + } + if cfg.Host == "" { + t.Errorf("returned config has empty Host; expected value from fixture") + } +} + +func TestGetConfigForFile_MissingFile(t *testing.T) { + missing := filepath.Join(t.TempDir(), "nonexistent-kubeconfig.yaml") + cfg, err := GetConfigForFile(missing) + if err == nil { + t.Fatalf("expected error for missing file, got config: %+v", cfg) + } + if cfg != nil { + t.Errorf("expected nil config on error, got %+v", cfg) + } +} + +func TestGetConfigForFile_MalformedFile(t *testing.T) { + malformed := filepath.Join(t.TempDir(), "malformed-kubeconfig.yaml") + if err := os.WriteFile(malformed, []byte("this: is: not: valid: yaml: {[}\n"), 0o600); err != nil { + t.Fatalf("failed to write malformed fixture: %v", err) + } + cfg, err := GetConfigForFile(malformed) + if err == nil { + t.Fatalf("expected error for malformed file, got config: %+v", cfg) + } + if cfg != nil { + t.Errorf("expected nil config on error, got %+v", cfg) + } +} + +func TestInClusterConfig_NotInCluster(t *testing.T) { + skipIfInCluster(t) + cfg, err := InClusterConfig() + if err == nil { + t.Fatalf("expected error outside a cluster, got config: %+v", cfg) + } + if !errors.Is(err, rest.ErrNotInCluster) { + t.Errorf("expected rest.ErrNotInCluster, got: %v", err) + } + if cfg != nil { + t.Errorf("expected nil config on error, got %+v", cfg) + } +} + +// TestNew_NoOptions verifies that New() with no options falls back to +// WithInClusterConfig — which fails when the test runs outside a Pod. +// Only exercises the fallback path; the success path requires an in- +// cluster environment which is not usable from a unit test. +func TestNew_NoOptions(t *testing.T) { + skipIfInCluster(t) + c, err := New() + if err == nil { + t.Fatalf("expected error from New() outside a cluster, got: %+v", c) + } + if !errors.Is(err, rest.ErrNotInCluster) { + t.Errorf("expected rest.ErrNotInCluster, got: %v", err) + } + if c != nil { + t.Errorf("expected nil client on error, got %+v", c) + } +} + +func TestNew_WithKubeConfig_Success(t *testing.T) { + c, err := New(WithKubeConfig(fixtureKubeconfig)) + if err != nil { + t.Fatalf("New(WithKubeConfig) returned error: %v", err) + } + if c == nil { + t.Fatal("New returned nil client with no error") + } + if c.K8sClient() == nil { + t.Error("Client.K8sClient() is nil") + } + if c.RestConfig() == nil { + t.Error("Client.RestConfig() is nil") + } + if c.HttpClient() == nil { + t.Error("Client.HttpClient() is nil") + } +} + +func TestNew_WithKubeConfig_MissingFile(t *testing.T) { + missing := filepath.Join(t.TempDir(), "nope.yaml") + c, err := New(WithKubeConfig(missing)) + if err == nil { + t.Fatalf("expected error for missing kubeconfig, got: %+v", c) + } + if c != nil { + t.Errorf("expected nil client on error, got %+v", c) + } +} + +func TestNew_WithInClusterConfig(t *testing.T) { + skipIfInCluster(t) + c, err := New(WithInClusterConfig()) + if err == nil { + t.Fatalf("expected error outside a cluster, got: %+v", c) + } + if !errors.Is(err, rest.ErrNotInCluster) { + t.Errorf("expected rest.ErrNotInCluster, got: %v", err) + } + if c != nil { + t.Errorf("expected nil client on error, got %+v", c) + } +} + +func TestNew_WithKubeOrInClusterConfig_EmptyFallsBack(t *testing.T) { + skipIfInCluster(t) + // Empty file path should fall back to in-cluster, which fails outside + // a cluster; that's how we know the fallback path was taken. + _, err := New(WithKubeOrInClusterConfig("")) + if !errors.Is(err, rest.ErrNotInCluster) { + t.Errorf("expected rest.ErrNotInCluster from empty-file fallback, got: %v", err) + } +} + +func TestNew_WithKubeOrInClusterConfig_FileWins(t *testing.T) { + c, err := New(WithKubeOrInClusterConfig(fixtureKubeconfig)) + if err != nil { + t.Fatalf("New(WithKubeOrInClusterConfig(file)) returned error: %v", err) + } + if c == nil || c.K8sClient() == nil { + t.Fatalf("expected non-nil client, got %+v", c) + } +} + +func TestNew_WithRestConfig(t *testing.T) { + // Build a config via the file helper first; use it as input to WithRestConfig + // to skip the file/in-cluster resolvers entirely. + cfg, err := GetConfigForFile(fixtureKubeconfig) + if err != nil { + t.Fatalf("GetConfigForFile fixture failed: %v", err) + } + c, err := New(WithRestConfig(cfg)) + if err != nil { + t.Fatalf("New(WithRestConfig) returned error: %v", err) + } + if c == nil || c.K8sClient() == nil { + t.Fatalf("expected non-nil client, got %+v", c) + } +} + +func TestNew_WithRestConfig_NilConfig(t *testing.T) { + // A nil *rest.Config must produce a normal error, not a panic inside + // rest.CopyConfig. + _, err := New(WithRestConfig(nil)) + if err == nil { + t.Fatal("New(WithRestConfig(nil)) returned nil error, want an error") + } +} + +func TestNew_WithHttpClient(t *testing.T) { + // Provide a pre-built HTTP client, verify HttpClient() returns the same pointer. + hc := &http.Client{} + c, err := New(WithHttpClient(hc), WithKubeConfig(fixtureKubeconfig)) + if err != nil { + t.Fatalf("New returned error: %v", err) + } + if c.HttpClient() != hc { + t.Errorf("HttpClient() returned different pointer than provided: got %p, want %p", c.HttpClient(), hc) + } +} + +// newTestConfig returns a fresh rest.Config. Built inline (not via the +// fixture) so tests fully control every field. Uses Insecure=true and no +// CAData so rest.HTTPClientFor inside New() does not try to parse CAData +// as a PEM block. +func newTestConfig() *rest.Config { + return &rest.Config{ + Host: "https://example.com:6443", + TLSClientConfig: rest.TLSClientConfig{ + Insecure: true, + }, + UserAgent: "test-user-agent", + } +} + +// TestClient_RestConfig_CopySemantics verifies RestConfig() returns a copy +// with rest.CopyConfig semantics: top-level and value-struct fields are +// safely overwritable on the returned value without affecting subsequent +// RestConfig() calls. Nested map/slice contents are NOT tested here — +// they share storage per rest.CopyConfig's contract. +func TestClient_RestConfig_CopySemantics(t *testing.T) { + c, err := New(WithRestConfig(newTestConfig())) + if err != nil { + t.Fatalf("New returned error: %v", err) + } + + first := c.RestConfig() + first.Host = "https://mutated.example.com" + first.UserAgent = "mutated-user-agent" + + second := c.RestConfig() + if second.Host == first.Host { + t.Errorf("RestConfig Host mutation leaked: got %q, want unchanged", second.Host) + } + if second.UserAgent == first.UserAgent { + t.Errorf("RestConfig UserAgent mutation leaked: got %q, want unchanged", second.UserAgent) + } +} + +// TestClient_WithRestConfig_CopySemanticsOnInput verifies that WithRestConfig +// takes a rest.CopyConfig copy of its input — post-New mutations of the +// original config's top-level fields do not leak into the client. +func TestClient_WithRestConfig_CopySemanticsOnInput(t *testing.T) { + cfg := newTestConfig() + + c, err := New(WithRestConfig(cfg)) + if err != nil { + t.Fatalf("New returned error: %v", err) + } + + cfg.Host = "https://mutated-input.example.com" + cfg.UserAgent = "mutated-input-user-agent" + + rc := c.RestConfig() + if rc.Host == cfg.Host { + t.Errorf("input-side Host mutation leaked to client: got %q", rc.Host) + } + if rc.UserAgent == cfg.UserAgent { + t.Errorf("input-side UserAgent mutation leaked to client: got %q", rc.UserAgent) + } +} + +func TestWithAcceptContentTypes(t *testing.T) { + c, err := New( + WithKubeConfig(fixtureKubeconfig), + WithAcceptContentTypes(ContentTypeProtobuf, ContentTypeJSON), + ) + if err != nil { + t.Fatalf("New returned error: %v", err) + } + got := c.RestConfig().AcceptContentTypes + want := ContentTypeProtobuf + "," + ContentTypeJSON + if got != want { + t.Errorf("AcceptContentTypes: got %q, want %q", got, want) + } +} + +func TestWithContentType(t *testing.T) { + c, err := New( + WithKubeConfig(fixtureKubeconfig), + WithContentType(ContentTypeProtobuf), + ) + if err != nil { + t.Fatalf("New returned error: %v", err) + } + if got := c.RestConfig().ContentType; got != ContentTypeProtobuf { + t.Errorf("ContentType: got %q, want %q", got, ContentTypeProtobuf) + } +} + +// TestContentType_OrderDependence verifies that config-dependent options +// (WithContentType, WithAcceptContentTypes) must appear after the config- +// source option; placing them before the config source returns an error. +func TestContentType_OrderDependence(t *testing.T) { + // Case A: config-source first, content-type second — works correctly. + a, err := New( + WithKubeConfig(fixtureKubeconfig), + WithContentType(ContentTypeProtobuf), + WithAcceptContentTypes(ContentTypeProtobuf, ContentTypeJSON), + ) + if err != nil { + t.Fatalf("case A New returned error: %v", err) + } + if a.RestConfig().ContentType != ContentTypeProtobuf { + t.Errorf("ContentType: got %q, want %q", a.RestConfig().ContentType, ContentTypeProtobuf) + } + + // Case B: content-type before config-source — must return an error. + _, err = New( + WithContentType(ContentTypeProtobuf), + WithKubeConfig(fixtureKubeconfig), + ) + if err == nil { + t.Fatal("case B New should return an error when content-type precedes config source, got nil") + } +} + +// TestContentType_OnlyNoConfigSource verifies that a content-type option +// alone (no config-source option) returns an error because the REST config +// is not yet set when the option is applied. +func TestContentType_OnlyNoConfigSource(t *testing.T) { + _, err := New(WithContentType(ContentTypeProtobuf)) + if err == nil { + t.Fatal("New(WithContentType) without a config source should return an error, got nil") + } +} + +// TestContentType_MultipleOptions verifies that when multiple content-type +// options are passed after the config source, all of them apply in the +// order they were passed (last write wins). +func TestContentType_MultipleOptions(t *testing.T) { + c, err := New( + WithKubeConfig(fixtureKubeconfig), // provides the config + WithAcceptContentTypes(ContentTypeProtobuf), // overridden below + WithAcceptContentTypes(ContentTypeJSON), // last one wins + WithContentType(ContentTypeProtobuf), + ) + if err != nil { + t.Fatalf("New returned error: %v", err) + } + // Last WithAcceptContentTypes wins. + if got := c.RestConfig().AcceptContentTypes; got != ContentTypeJSON { + t.Errorf("last-applied AcceptContentTypes should win: got %q, want %q", got, ContentTypeJSON) + } + if got := c.RestConfig().ContentType; got != ContentTypeProtobuf { + t.Errorf("ContentType: got %q, want %q", got, ContentTypeProtobuf) + } +} + +func TestClient_Close_Idempotent(t *testing.T) { + c, err := New(WithKubeConfig(fixtureKubeconfig)) + if err != nil { + t.Fatalf("New returned error: %v", err) + } + // First call — must not panic. + c.Close() + // Second call — must also not panic. + c.Close() +} diff --git a/pkg/kubernetes/client/testdata/kubeconfig-example.yaml b/pkg/kubernetes/client/testdata/kubeconfig-example.yaml new file mode 100644 index 000000000..350cfe160 --- /dev/null +++ b/pkg/kubernetes/client/testdata/kubeconfig-example.yaml @@ -0,0 +1,17 @@ +apiVersion: v1 +kind: Config +current-context: nri-plugins-test +clusters: +- name: nri-plugins-test-cluster + cluster: + server: https://example.com:6443 + insecure-skip-tls-verify: true +contexts: +- name: nri-plugins-test + context: + cluster: nri-plugins-test-cluster + user: nri-plugins-test-user +users: +- name: nri-plugins-test-user + user: + token: dummy-token From ba5f16eefa709a9157b8bf349d8263105411a1fb Mon Sep 17 00:00:00 2001 From: Ed Bartosh Date: Mon, 14 Sep 2026 10:38:13 +0300 Subject: [PATCH 3/3] agent: switch to pkg/kubernetes/client Replace the agent's hand-rolled REST config/HTTP client/clientset setup with the new pkg/kubernetes/client wrapper. This also drops the now-redundant getRESTConfig helper and exposes the wrapped client, kubeconfig path, and REST config to callers via new KubeClient/KubeConfig/RestConfig accessors. Signed-off-by: Ed Bartosh --- pkg/agent/agent.go | 88 ++++++++++++++++++++-------------------------- 1 file changed, 38 insertions(+), 50 deletions(-) diff --git a/pkg/agent/agent.go b/pkg/agent/agent.go index 8a3be0245..3c98c223d 100644 --- a/pkg/agent/agent.go +++ b/pkg/agent/agent.go @@ -29,13 +29,12 @@ import ( "k8s.io/apimachinery/pkg/runtime" "k8s.io/apimachinery/pkg/types" "k8s.io/client-go/rest" - "k8s.io/client-go/tools/clientcmd" nrtapi "github.com/containers/nri-plugins/pkg/agent/nrtapi" "github.com/containers/nri-plugins/pkg/agent/podresapi" cfgapi "github.com/containers/nri-plugins/pkg/apis/config/v1alpha1" + "github.com/containers/nri-plugins/pkg/kubernetes/client" "github.com/containers/nri-plugins/pkg/kubernetes/watch" - k8sclient "k8s.io/client-go/kubernetes" logger "github.com/containers/nri-plugins/pkg/log" ) @@ -129,12 +128,11 @@ type Agent struct { kubeConfig string // kubeconfig path configFile string // configuration file to use instead of custom resource - cfgIf ConfigInterface // custom resource access interface - httpCli *http.Client // shared HTTP client - k8sCli *k8sclient.Clientset // kubernetes client - nrtCli *nrtapi.Client // NRT custom resources client - nrtLock sync.Mutex // serialize NRT custom resource updates - podResCli *podresapi.Client // pod resources API client + cfgIf ConfigInterface // custom resource access interface + k8sCli *client.Client // wrapped kubernetes client + REST config + HTTP client + nrtCli *nrtapi.Client // NRT custom resources client + nrtLock sync.Mutex // serialize NRT custom resource updates + podResCli *podresapi.Client // pod resources API client notifyFn NotifyFn // config resource change notification callback nodeWatch watch.Interface // kubernetes node watch @@ -300,12 +298,11 @@ func (a *Agent) configure(newConfig metav1.Object) { switch { case cfg.NodeResourceTopology && a.nrtCli == nil: log.Infof("enabling NRT client") - cfg, err := a.getRESTConfig() - if err != nil { - log.Errorf("failed to setup NRT client: %v", err) + if a.k8sCli == nil { + log.Errorf("failed to setup NRT client: no kubernetes client") break } - cli, err := nrtapi.NewForConfigAndClient(cfg, a.httpCli) + cli, err := nrtapi.NewForConfigAndClient(a.k8sCli.RestConfig(), a.k8sCli.HttpClient()) if err != nil { log.Errorf("failed to setup NRT client: %v", err) break @@ -346,30 +343,17 @@ func (a *Agent) setupClients() error { return nil } - // Create HTTP/REST client and K8s client on initial startup. Any failure - // to create these is a failure start up. - if a.httpCli == nil { - log.Infof("setting up HTTP/REST client...") - restCfg, err := a.getRESTConfig() - if err != nil { - return err - } - - a.httpCli, err = rest.HTTPClientFor(restCfg) + // Create the kubernetes client on initial startup. Any failure is fatal. + if a.k8sCli == nil { + log.Infof("setting up kubernetes client...") + c, err := client.New(client.WithKubeOrInClusterConfig(a.kubeConfig)) if err != nil { - return fmt.Errorf("failed to setup kubernetes HTTP client: %w", err) - } - - log.Infof("setting up K8s client...") - a.k8sCli, err = k8sclient.NewForConfigAndClient(restCfg, a.httpCli) - if err != nil { - a.cleanupClients() return fmt.Errorf("failed to setup kubernetes client: %w", err) } + a.k8sCli = c - kubeCfg := *restCfg - err = a.cfgIf.SetKubeClient(a.httpCli, &kubeCfg) - if err != nil { + if err := a.cfgIf.SetKubeClient(a.k8sCli.HttpClient(), a.k8sCli.RestConfig()); err != nil { + a.cleanupClients() return fmt.Errorf("failed to setup kubernetes config resource client: %w", err) } } @@ -380,31 +364,35 @@ func (a *Agent) setupClients() error { } func (a *Agent) cleanupClients() { - if a.httpCli != nil { - a.httpCli.CloseIdleConnections() - } - a.httpCli = nil + a.k8sCli.Close() a.k8sCli = nil a.nrtCli = nil } -func (a *Agent) getRESTConfig() (*rest.Config, error) { - var ( - cfg *rest.Config - err error - ) +// NodeName returns the kubernetes node name this agent is running on. +func (a *Agent) NodeName() string { + return a.nodeName +} - if a.kubeConfig == "" { - cfg, err = rest.InClusterConfig() - } else { - cfg, err = clientcmd.BuildConfigFromFlags("", a.kubeConfig) - } +// KubeClient returns the shared kubernetes client wrapper. Returns nil +// before setupClients has run successfully. +func (a *Agent) KubeClient() *client.Client { + return a.k8sCli +} - if err != nil { - return nil, fmt.Errorf("failed to get kubernetes REST client config: %w", err) - } +// KubeConfig returns the kubeconfig file path this agent was configured +// with. Returns the empty string when running with in-cluster credentials. +func (a *Agent) KubeConfig() string { + return a.kubeConfig +} - return cfg, err +// RestConfig returns a copy of the REST config used by the shared +// kubernetes client, or nil before setupClients has run successfully. +func (a *Agent) RestConfig() *rest.Config { + if a.k8sCli == nil { + return nil + } + return a.k8sCli.RestConfig() } func (a *Agent) setupNodeWatch() error {