upstream/kubernetes
1
0
Fork 0
mirror of https://github.com/kubernetes/kubernetes.git synced 2026-05-22 10:00:21 -04:00
Code Issues Projects Releases Packages Wiki Activity Actions

Merge pull request #137949 from bitoku/use-atomic

Browse source 
cri-client: use atomic.Bool for useStreaming to fix data race
This commit is contained in:
Kubernetes Prow Robot 2026-03-24 07:28:19 +05:30 committed by GitHub
commit 32d6fb1759
No known key found for this signature in database
GPG key ID: B5690EEEBB952194
2 changed files with 27 additions and 22 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

19
staging/src/k8s.io/cri-client/pkg/remote_image.go
View file

@ -21,6 +21,7 @@ import (
"errors"
"fmt"
"io"
"sync/atomic"
"time"
"go.opentelemetry.io/contrib/instrumentation/google.golang.org/grpc/otelgrpc"
@ -46,9 +47,10 @@ type remoteImageService struct {
conn *grpc.ClientConn
// useStreaming indicates whether to use streaming RPCs for list operations
// when the CRIListStreaming feature gate is enabled. It falls back to false
// if a streaming RPC returns Unimplemented. The fallback is racy but kept
// simple since the worst case is a few extra fallback attempts.
useStreaming bool
// if a streaming RPC returns Unimplemented. Multiple goroutines may
// concurrently observe Unimplemented and store false, but that is harmless
// because the store is idempotent.
useStreaming atomic.Bool
}
// NewRemoteImageService creates a new internalapi.ImageManagerService.
@ -104,10 +106,11 @@ func NewRemoteImageService(ctx context.Context, endpoint string, connectionTimeo
}
service := &remoteImageService{
timeout: connectionTimeout,
conn: conn,
useStreaming: useStreaming,
timeout: connectionTimeout,
conn: conn,
}
service.useStreaming.Store(useStreaming)
if err := service.validateServiceConnection(ctx, conn, endpoint); err != nil {
return nil, fmt.Errorf("validate service connection: %w", err)
}
@ -143,7 +146,7 @@ func (r *remoteImageService) ListImages(ctx context.Context, filter *runtimeapi.
ctx, cancel := context.WithTimeout(ctx, r.timeout)
defer cancel()
if r.useStreaming {
if r.useStreaming.Load() {
return r.streamImagesV1(ctx, filter)
}
return r.listImagesV1(ctx, filter)
@ -184,7 +187,7 @@ func (r *remoteImageService) streamImagesV1(ctx context.Context, filter *runtime
// but when calling Recv() on the stream.
if status.Code(err) == codes.Unimplemented {
logger.Info("StreamImages not implemented, falling back to ListImages", "filter", filter)
r.useStreaming = false
r.useStreaming.Store(false)
return r.listImagesV1(ctx, filter)
}
logger.Error(err, "StreamImages recv failed", "filter", filter, "itemsReceived", len(images))

30
staging/src/k8s.io/cri-client/pkg/remote_runtime.go
View file

@ -22,6 +22,7 @@ import (
"fmt"
"io"
"strings"
"sync/atomic"
"time"
"go.opentelemetry.io/contrib/instrumentation/google.golang.org/grpc/otelgrpc"
@ -51,9 +52,10 @@ type remoteRuntimeService struct {
conn *grpc.ClientConn
// useStreaming indicates whether to use streaming RPCs for list operations
// when the CRIListStreaming feature gate is enabled. It falls back to false
// if a streaming RPC returns Unimplemented. The fallback is racy but kept
// simple since the worst case is a few extra fallback attempts.
useStreaming bool
// if a streaming RPC returns Unimplemented. Multiple goroutines may
// concurrently observe Unimplemented and store false, but that is harmless
// because the store is idempotent.
useStreaming atomic.Bool
}
const (
@ -138,8 +140,8 @@ func NewRemoteRuntimeService(ctx context.Context, endpoint string, connectionTim
timeout: connectionTimeout,
logReduction: logreduction.NewLogReduction(identicalErrorDelay),
conn: conn,
useStreaming: useStreaming,
}
service.useStreaming.Store(useStreaming)
if err := service.validateServiceConnection(ctx, conn, endpoint); err != nil {
return nil, fmt.Errorf("validate service connection: %w", err)
@ -330,7 +332,7 @@ func (r *remoteRuntimeService) ListPodSandbox(ctx context.Context, filter *runti
ctx, cancel := context.WithTimeout(ctx, r.timeout)
defer cancel()
if r.useStreaming {
if r.useStreaming.Load() {
return r.streamPodSandboxesV1(ctx, filter)
}
return r.listPodSandboxV1(ctx, filter)
@ -373,7 +375,7 @@ func (r *remoteRuntimeService) streamPodSandboxesV1(ctx context.Context, filter
// but when calling Recv() on the stream.
if status.Code(err) == codes.Unimplemented {
logger.Info("StreamPodSandboxes not implemented, falling back to ListPodSandbox", "filter", filter)
r.useStreaming = false
r.useStreaming.Store(false)
return r.listPodSandboxV1(ctx, filter)
}
logger.Error(err, "StreamPodSandboxes recv failed", "filter", filter, "itemsReceived", len(items))
@ -489,7 +491,7 @@ func (r *remoteRuntimeService) ListContainers(ctx context.Context, filter *runti
ctx, cancel := context.WithTimeout(ctx, r.timeout)
defer cancel()
if r.useStreaming {
if r.useStreaming.Load() {
return r.streamContainersV1(ctx, filter)
}
return r.listContainersV1(ctx, filter)
@ -531,7 +533,7 @@ func (r *remoteRuntimeService) streamContainersV1(ctx context.Context, filter *r
// but when calling Recv() on the stream.
if status.Code(err) == codes.Unimplemented {
logger.Info("StreamContainers not implemented, falling back to ListContainers", "filter", filter)
r.useStreaming = false
r.useStreaming.Store(false)
return r.listContainersV1(ctx, filter)
}
logger.Error(err, "StreamContainers recv failed", "filter", filter, "itemsReceived", len(containers))
@ -846,7 +848,7 @@ func (r *remoteRuntimeService) ListContainerStats(ctx context.Context, filter *r
ctx, cancel := context.WithTimeout(ctx, r.timeout)
defer cancel()
if r.useStreaming {
if r.useStreaming.Load() {
return r.streamContainerStatsV1(ctx, filter)
}
return r.listContainerStatsV1(ctx, filter)
@ -888,7 +890,7 @@ func (r *remoteRuntimeService) streamContainerStatsV1(ctx context.Context, filte
// but when calling Recv() on the stream.
if status.Code(err) == codes.Unimplemented {
logger.Info("StreamContainerStats not implemented, falling back to ListContainerStats", "filter", filter)
r.useStreaming = false
r.useStreaming.Store(false)
return r.listContainerStatsV1(ctx, filter)
}
logger.Error(err, "StreamContainerStats recv failed", "filter", filter, "itemsReceived", len(stats))
@ -937,7 +939,7 @@ func (r *remoteRuntimeService) ListPodSandboxStats(ctx context.Context, filter *
ctx, cancel := context.WithTimeout(ctx, r.timeout)
defer cancel()
if r.useStreaming {
if r.useStreaming.Load() {
return r.streamPodSandboxStatsV1(ctx, filter)
}
return r.listPodSandboxStatsV1(ctx, filter)
@ -979,7 +981,7 @@ func (r *remoteRuntimeService) streamPodSandboxStatsV1(ctx context.Context, filt
// but when calling Recv() on the stream.
if status.Code(err) == codes.Unimplemented {
logger.Info("StreamPodSandboxStats not implemented, falling back to ListPodSandboxStats", "filter", filter)
r.useStreaming = false
r.useStreaming.Store(false)
return r.listPodSandboxStatsV1(ctx, filter)
}
logger.Error(err, "StreamPodSandboxStats recv failed", "filter", filter, "itemsReceived", len(stats))
@ -1112,7 +1114,7 @@ func (r *remoteRuntimeService) ListPodSandboxMetrics(ctx context.Context) ([]*ru
ctx, cancel := context.WithTimeout(ctx, r.timeout)
defer cancel()
if r.useStreaming {
if r.useStreaming.Load() {
return r.streamPodSandboxMetricsV1(ctx)
}
return r.listPodSandboxMetricsV1(ctx)
@ -1150,7 +1152,7 @@ func (r *remoteRuntimeService) streamPodSandboxMetricsV1(ctx context.Context) ([
// but when calling Recv() on the stream.
if status.Code(err) == codes.Unimplemented {
logger.Info("StreamPodSandboxMetrics not implemented, falling back to ListPodSandboxMetrics")
r.useStreaming = false
r.useStreaming.Store(false)
return r.listPodSandboxMetricsV1(ctx)
}
logger.Error(err, "StreamPodSandboxMetrics recv failed", "itemsReceived", len(metrics))