From 17fa42f3ed0c53b3cb5d711f0fab645d6438c06b Mon Sep 17 00:00:00 2001 From: Haihan Jiang Date: Sat, 30 May 2026 05:26:17 -0700 Subject: [PATCH] Cache containerd clients by namespace --- container/containerd/client.go | 113 +++++++++++------- container/containerd/client_cache_test.go | 132 ++++++++++++++++++++++ 2 files changed, 202 insertions(+), 43 deletions(-) create mode 100644 container/containerd/client_cache_test.go diff --git a/container/containerd/client.go b/container/containerd/client.go index c538e102fb..595024d603 100644 --- a/container/containerd/client.go +++ b/container/containerd/client.go @@ -54,9 +54,21 @@ var ( ErrTaskIsInUnknownState = errors.New("containerd task is in unknown state") // used when process reported in containerd task is in Unknown State ) -var once sync.Once -var ctrdClient ContainerdClient = nil -var ctrdClientErr error = nil +type clientCacheKey struct { + address string + namespace string +} + +type cachedClient struct { + once sync.Once + client ContainerdClient + err error +} + +var ( + ctrdClientsMu sync.Mutex + ctrdClients = map[clientCacheKey]*cachedClient{} +) const ( maxBackoffDelay = 3 * time.Second @@ -67,48 +79,63 @@ const ( // Client creates a containerd client func Client(address, namespace string) (ContainerdClient, error) { - once.Do(func() { - tryConn, err := net.DialTimeout("unix", address, connectionTimeout) - if err != nil { - ctrdClientErr = fmt.Errorf("containerd: cannot unix dial containerd api service: %v", err) - return - } - tryConn.Close() + key := clientCacheKey{ + address: address, + namespace: namespace, + } - connParams := grpc.ConnectParams{ - Backoff: backoff.DefaultConfig, - } - connParams.Backoff.BaseDelay = baseBackoffDelay - connParams.Backoff.MaxDelay = maxBackoffDelay - //nolint:staticcheck // SA1019 - gopts := []grpc.DialOption{ - grpc.WithTransportCredentials(insecure.NewCredentials()), - grpc.WithContextDialer(dialer.ContextDialer), - grpc.WithBlock(), - grpc.WithConnectParams(connParams), - grpc.WithDefaultCallOptions(grpc.MaxCallRecvMsgSize(maxMsgSize)), - } - unary, stream := newNSInterceptors(namespace) - gopts = append(gopts, - grpc.WithUnaryInterceptor(unary), - grpc.WithStreamInterceptor(stream), - ) - - ctx, cancel := context.WithTimeout(context.Background(), connectionTimeout) - defer cancel() - //nolint:staticcheck // SA1019 - conn, err := grpc.DialContext(ctx, dialer.DialAddress(address), gopts...) - if err != nil { - ctrdClientErr = err - return - } - ctrdClient = &client{ - containerService: containersapi.NewContainersClient(conn), - taskService: tasksapi.NewTasksClient(conn), - versionService: versionapi.NewVersionClient(conn), - } + ctrdClientsMu.Lock() + cached := ctrdClients[key] + if cached == nil { + cached = &cachedClient{} + ctrdClients[key] = cached + } + ctrdClientsMu.Unlock() + + cached.once.Do(func() { + cached.client, cached.err = newClient(address, namespace) }) - return ctrdClient, ctrdClientErr + return cached.client, cached.err +} + +func newClient(address, namespace string) (ContainerdClient, error) { + tryConn, err := net.DialTimeout("unix", address, connectionTimeout) + if err != nil { + return nil, fmt.Errorf("containerd: cannot unix dial containerd api service: %v", err) + } + tryConn.Close() + + connParams := grpc.ConnectParams{ + Backoff: backoff.DefaultConfig, + } + connParams.Backoff.BaseDelay = baseBackoffDelay + connParams.Backoff.MaxDelay = maxBackoffDelay + //nolint:staticcheck // SA1019 + gopts := []grpc.DialOption{ + grpc.WithTransportCredentials(insecure.NewCredentials()), + grpc.WithContextDialer(dialer.ContextDialer), + grpc.WithBlock(), + grpc.WithConnectParams(connParams), + grpc.WithDefaultCallOptions(grpc.MaxCallRecvMsgSize(maxMsgSize)), + } + unary, stream := newNSInterceptors(namespace) + gopts = append(gopts, + grpc.WithUnaryInterceptor(unary), + grpc.WithStreamInterceptor(stream), + ) + + ctx, cancel := context.WithTimeout(context.Background(), connectionTimeout) + defer cancel() + //nolint:staticcheck // SA1019 + conn, err := grpc.DialContext(ctx, dialer.DialAddress(address), gopts...) + if err != nil { + return nil, err + } + return &client{ + containerService: containersapi.NewContainersClient(conn), + taskService: tasksapi.NewTasksClient(conn), + versionService: versionapi.NewVersionClient(conn), + }, nil } func (c *client) LoadContainer(ctx context.Context, id string) (*containers.Container, error) { diff --git a/container/containerd/client_cache_test.go b/container/containerd/client_cache_test.go new file mode 100644 index 0000000000..8b3f17d8f7 --- /dev/null +++ b/container/containerd/client_cache_test.go @@ -0,0 +1,132 @@ +// Copyright 2026 Google Inc. 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 containerd + +import ( + "context" + "net" + "os" + "path/filepath" + "reflect" + "sync" + "testing" + + versionapi "github.com/containerd/containerd/api/services/version/v1" + "google.golang.org/grpc" + emptypb "google.golang.org/protobuf/types/known/emptypb" + + "github.com/google/cadvisor/container/containerd/namespaces" +) + +type testVersionServer struct { + versionapi.UnimplementedVersionServer + + mu sync.Mutex + namespaces []string +} + +func (s *testVersionServer) Version(ctx context.Context, _ *emptypb.Empty) (*versionapi.VersionResponse, error) { + namespace, _ := namespaces.Namespace(ctx) + + s.mu.Lock() + s.namespaces = append(s.namespaces, namespace) + s.mu.Unlock() + + return &versionapi.VersionResponse{Version: namespace}, nil +} + +func (s *testVersionServer) recordedNamespaces() []string { + s.mu.Lock() + defer s.mu.Unlock() + + out := make([]string, len(s.namespaces)) + copy(out, s.namespaces) + return out +} + +func TestClientCacheSeparatesNamespaces(t *testing.T) { + resetClientCacheForTest(t) + + socketDir, err := os.MkdirTemp("/tmp", "cadvisor-containerd-") + if err != nil { + t.Fatalf("os.MkdirTemp failed: %v", err) + } + t.Cleanup(func() { + if err := os.RemoveAll(socketDir); err != nil { + t.Logf("failed to clean up socket dir %q: %v", socketDir, err) + } + }) + socketPath := filepath.Join(socketDir, "containerd.sock") + listener, err := net.Listen("unix", socketPath) + if err != nil { + t.Fatalf("net.Listen(%q) failed: %v", socketPath, err) + } + server := grpc.NewServer() + versionServer := &testVersionServer{} + versionapi.RegisterVersionServer(server, versionServer) + go func() { + if err := server.Serve(listener); err != nil { + t.Logf("test containerd server stopped: %v", err) + } + }() + t.Cleanup(server.Stop) + + mobyClient, err := Client(socketPath, "moby") + if err != nil { + t.Fatalf("Client(%q, moby) failed: %v", socketPath, err) + } + mobyVersion, err := mobyClient.Version(context.Background()) + if err != nil { + t.Fatalf("moby Version() failed: %v", err) + } + if mobyVersion != "moby" { + t.Fatalf("moby Version() = %q, want moby", mobyVersion) + } + + k8sClient, err := Client(socketPath, "k8s.io") + if err != nil { + t.Fatalf("Client(%q, k8s.io) failed: %v", socketPath, err) + } + if mobyClient == k8sClient { + t.Fatalf("Client returned the same cached client for distinct namespaces") + } + k8sVersion, err := k8sClient.Version(context.Background()) + if err != nil { + t.Fatalf("k8s Version() failed: %v", err) + } + if k8sVersion != "k8s.io" { + t.Fatalf("k8s Version() = %q, want k8s.io", k8sVersion) + } + + wantNamespaces := []string{"moby", "k8s.io"} + if gotNamespaces := versionServer.recordedNamespaces(); !reflect.DeepEqual(gotNamespaces, wantNamespaces) { + t.Fatalf("recorded namespaces = %v, want %v", gotNamespaces, wantNamespaces) + } +} + +func resetClientCacheForTest(t *testing.T) { + t.Helper() + + ctrdClientsMu.Lock() + previous := ctrdClients + ctrdClients = map[clientCacheKey]*cachedClient{} + ctrdClientsMu.Unlock() + + t.Cleanup(func() { + ctrdClientsMu.Lock() + ctrdClients = previous + ctrdClientsMu.Unlock() + }) +}