diff --git a/weed/admin/plugin/plugin_scheduler.go b/weed/admin/plugin/plugin_scheduler.go index aa97e92737..58ad878d6a 100644 --- a/weed/admin/plugin/plugin_scheduler.go +++ b/weed/admin/plugin/plugin_scheduler.go @@ -12,6 +12,7 @@ import ( "github.com/seaweedfs/seaweedfs/weed/glog" "github.com/seaweedfs/seaweedfs/weed/pb/plugin_pb" + "google.golang.org/protobuf/proto" "google.golang.org/protobuf/types/known/timestamppb" ) @@ -614,7 +615,7 @@ func deriveSchedulerAdminRuntime( descriptor *plugin_pb.JobTypeDescriptor, ) *plugin_pb.AdminRuntimeConfig { if cfg != nil && cfg.AdminRuntime != nil { - adminConfig := *cfg.AdminRuntime + adminConfig := proto.Clone(cfg.AdminRuntime).(*plugin_pb.AdminRuntimeConfig) // Overlay descriptor defaults for any zero numeric fields. Persisted // configs from older versions have no execution_timeout_seconds, and // without this overlay the scheduler would fall back to the 90s @@ -646,7 +647,7 @@ func deriveSchedulerAdminRuntime( adminConfig.ExecutionTimeoutSeconds = defaults.ExecutionTimeoutSeconds } } - return &adminConfig + return adminConfig } if descriptor == nil || descriptor.AdminRuntimeDefaults == nil { diff --git a/weed/admin/plugin/registry.go b/weed/admin/plugin/registry.go index 2201e6a9c5..6e27d30411 100644 --- a/weed/admin/plugin/registry.go +++ b/weed/admin/plugin/registry.go @@ -8,6 +8,7 @@ import ( "time" "github.com/seaweedfs/seaweedfs/weed/pb/plugin_pb" + "google.golang.org/protobuf/proto" ) const defaultWorkerStaleTimeout = 2 * time.Minute @@ -495,36 +496,35 @@ func cloneJobTypeCapability(in *plugin_pb.JobTypeCapability) *plugin_pb.JobTypeC if in == nil { return nil } - out := *in - return &out + return proto.Clone(in).(*plugin_pb.JobTypeCapability) } func cloneWorkerHeartbeat(in *plugin_pb.WorkerHeartbeat) *plugin_pb.WorkerHeartbeat { if in == nil { return nil } - out := *in + out := proto.Clone(in).(*plugin_pb.WorkerHeartbeat) if in.RunningWork != nil { - out.RunningWork = make([]*plugin_pb.RunningWork, 0, len(in.RunningWork)) - for _, rw := range in.RunningWork { - if rw == nil { - continue + running := out.RunningWork[:0] + // Clone can turn nil messages into empty messages, so filter using + // their original positions while reusing the cloned entries. + for i, work := range in.RunningWork { + if work != nil { + running = append(running, out.RunningWork[i]) } - clone := *rw - out.RunningWork = append(out.RunningWork, &clone) } - } - if in.QueuedJobsByType != nil { - out.QueuedJobsByType = make(map[string]int32, len(in.QueuedJobsByType)) - for k, v := range in.QueuedJobsByType { - out.QueuedJobsByType[k] = v + clear(out.RunningWork[len(running):]) + if running == nil { + running = []*plugin_pb.RunningWork{} } + out.RunningWork = running } - if in.Metadata != nil { - out.Metadata = make(map[string]string, len(in.Metadata)) - for k, v := range in.Metadata { - out.Metadata[k] = v - } + // Preserve writable empty maps without rebuilding populated clones. + if in.QueuedJobsByType != nil && out.QueuedJobsByType == nil { + out.QueuedJobsByType = make(map[string]int32) } - return &out + if in.Metadata != nil && out.Metadata == nil { + out.Metadata = make(map[string]string) + } + return out } diff --git a/weed/admin/plugin/registry_snapshot_test.go b/weed/admin/plugin/registry_snapshot_test.go new file mode 100644 index 0000000000..1546096048 --- /dev/null +++ b/weed/admin/plugin/registry_snapshot_test.go @@ -0,0 +1,65 @@ +package plugin + +import ( + "testing" + + "github.com/seaweedfs/seaweedfs/weed/pb/plugin_pb" +) + +func TestCloneWorkerHeartbeatFiltersNilAndIsolatesState(t *testing.T) { + t.Parallel() + source := &plugin_pb.WorkerHeartbeat{ + RunningWork: []*plugin_pb.RunningWork{nil, {WorkId: "first"}, nil, {WorkId: "second"}, nil}, + QueuedJobsByType: map[string]int32{"vacuum": 2}, + Metadata: map[string]string{"rack": "rack-a"}, + } + got := cloneWorkerHeartbeat(source) + if len(got.RunningWork) != 2 || got.RunningWork[0].GetWorkId() != "first" || got.RunningWork[1].GetWorkId() != "second" { + t.Fatalf("unexpected filtered work: %v", got.RunningWork) + } + got.RunningWork[0].WorkId = "changed" + got.QueuedJobsByType["vacuum"] = 99 + got.Metadata["rack"] = "changed" + if len(source.RunningWork) != 5 || source.RunningWork[0] != nil || source.RunningWork[1].WorkId != "first" { + t.Fatalf("cloning mutated source work: %v", source.RunningWork) + } + if source.QueuedJobsByType["vacuum"] != 2 || source.Metadata["rack"] != "rack-a" { + t.Fatal("cloned maps alias the source") + } +} + +func TestCloneWorkerHeartbeatPreservesEmptyCollections(t *testing.T) { + t.Parallel() + if cloneWorkerHeartbeat(nil) != nil { + t.Fatal("nil heartbeat must stay nil") + } + for _, tc := range []struct { + name string + source *plugin_pb.WorkerHeartbeat + wantNilWork bool + }{ + {"nil", &plugin_pb.WorkerHeartbeat{}, true}, + {"empty", &plugin_pb.WorkerHeartbeat{ + RunningWork: []*plugin_pb.RunningWork{}, + QueuedJobsByType: map[string]int32{}, + Metadata: map[string]string{}, + }, false}, + {"all nil work", &plugin_pb.WorkerHeartbeat{RunningWork: []*plugin_pb.RunningWork{nil, nil}}, false}, + } { + t.Run(tc.name, func(t *testing.T) { + got := cloneWorkerHeartbeat(tc.source) + if len(got.RunningWork) != 0 || (got.RunningWork == nil) != tc.wantNilWork { + t.Fatalf("unexpected empty work slice: %#v", got.RunningWork) + } + if (got.QueuedJobsByType == nil) != (tc.source.QueuedJobsByType == nil) || (got.Metadata == nil) != (tc.source.Metadata == nil) { + t.Fatal("clone changed whether empty maps are writable") + } + if got.Metadata != nil { + got.Metadata["added"] = "value" + if len(tc.source.Metadata) != 0 { + t.Fatal("empty metadata map aliases source") + } + } + }) + } +} diff --git a/weed/cluster/lock_ring_manager.go b/weed/cluster/lock_ring_manager.go index 036db3c4f9..04577ac18b 100644 --- a/weed/cluster/lock_ring_manager.go +++ b/weed/cluster/lock_ring_manager.go @@ -7,6 +7,7 @@ import ( "github.com/seaweedfs/seaweedfs/weed/glog" "github.com/seaweedfs/seaweedfs/weed/pb" "github.com/seaweedfs/seaweedfs/weed/pb/master_pb" + "google.golang.org/protobuf/proto" ) const LockRingStabilizationInterval = 1 * time.Second @@ -102,9 +103,7 @@ func (lrm *LockRingManager) GetLastUpdate(filerGroup FilerGroupName) *master_pb. if !ok || update == nil { return nil } - cp := *update - cp.Servers = append([]string(nil), update.Servers...) - return &cp + return proto.Clone(update).(*master_pb.LockRingUpdate) } // scheduleBroadcast resets the stabilization timer. If another change arrives diff --git a/weed/credential/credential_manager.go b/weed/credential/credential_manager.go index ec37b2a8e0..7bf84c9670 100644 --- a/weed/credential/credential_manager.go +++ b/weed/credential/credential_manager.go @@ -15,6 +15,7 @@ import ( "github.com/seaweedfs/seaweedfs/weed/wdclient" "google.golang.org/grpc" "google.golang.org/protobuf/encoding/protojson" + "google.golang.org/protobuf/proto" ) // FilerAddressSetter is an interface for credential stores that need a dynamic filer address @@ -168,15 +169,16 @@ func (cm *CredentialManager) SaveConfiguration(ctx context.Context, config *iam_ staticNames := cm.staticNames cm.staticMu.RUnlock() if len(staticNames) > 0 { - var dynamicOnly []*iam_pb.Identity - for _, ident := range config.Identities { + configCopy := proto.Clone(config).(*iam_pb.S3ApiConfiguration) + dynamicOnly := configCopy.Identities[:0] + for _, ident := range configCopy.Identities { if !staticNames[ident.Name] { dynamicOnly = append(dynamicOnly, ident) } } - configCopy := *config + clear(configCopy.Identities[len(dynamicOnly):]) configCopy.Identities = dynamicOnly - return cm.Store.SaveConfiguration(ctx, &configCopy) + return cm.Store.SaveConfiguration(ctx, configCopy) } return cm.Store.SaveConfiguration(ctx, config) } diff --git a/weed/credential/credential_manager_snapshot_test.go b/weed/credential/credential_manager_snapshot_test.go new file mode 100644 index 0000000000..41c6d4628d --- /dev/null +++ b/weed/credential/credential_manager_snapshot_test.go @@ -0,0 +1,46 @@ +package credential + +import ( + "context" + "testing" + + "github.com/seaweedfs/seaweedfs/weed/pb/iam_pb" + "google.golang.org/protobuf/proto" +) + +type snapshotConfigurationStore struct { + CredentialStore + saved *iam_pb.S3ApiConfiguration +} + +func (s *snapshotConfigurationStore) SaveConfiguration(_ context.Context, config *iam_pb.S3ApiConfiguration) error { + s.saved = config + return nil +} + +func TestSaveConfigurationFiltersStaticIdentitySnapshot(t *testing.T) { + t.Parallel() + store := &snapshotConfigurationStore{} + manager := &CredentialManager{Store: store} + manager.SetStaticIdentities([]*iam_pb.Identity{{Name: "static"}}) + source := &iam_pb.S3ApiConfiguration{Identities: []*iam_pb.Identity{ + {Name: "alice", Actions: []string{"Read"}}, + {Name: "static"}, + {Name: "bob", Actions: []string{"Write"}}, + }} + before := proto.Clone(source) + if err := manager.SaveConfiguration(context.Background(), source); err != nil { + t.Fatal(err) + } + if store.saved == nil || len(store.saved.Identities) != 2 || store.saved.Identities[0].Name != "alice" || store.saved.Identities[1].Name != "bob" { + t.Fatalf("unexpected saved configuration: %v", store.saved) + } + if !proto.Equal(source, before) { + t.Fatal("filtering mutated the caller's configuration") + } + store.saved.Identities[0].Actions[0] = "Admin" + store.saved.Identities[1].Name = "changed" + if !proto.Equal(source, before) { + t.Fatal("saved identities alias the caller's configuration") + } +} diff --git a/weed/mount/peer_registrar_test.go b/weed/mount/peer_registrar_test.go index 7fee73d9bd..21cb7ac7f2 100644 --- a/weed/mount/peer_registrar_test.go +++ b/weed/mount/peer_registrar_test.go @@ -8,6 +8,7 @@ import ( "github.com/seaweedfs/seaweedfs/weed/pb" "github.com/seaweedfs/seaweedfs/weed/pb/filer_pb" "google.golang.org/grpc" + "google.golang.org/protobuf/proto" ) // fakeFilerClient captures MountRegister/MountList calls and lets the test @@ -17,22 +18,21 @@ type fakeFilerClient struct { filer_pb.SeaweedFilerClient // embed for methods we don't need mu sync.Mutex - registerCalls []filer_pb.MountRegisterRequest + registerCalls []*filer_pb.MountRegisterRequest listResponse filer_pb.MountListResponse } func (f *fakeFilerClient) MountRegister(ctx context.Context, req *filer_pb.MountRegisterRequest, opts ...grpc.CallOption) (*filer_pb.MountRegisterResponse, error) { f.mu.Lock() defer f.mu.Unlock() - f.registerCalls = append(f.registerCalls, *req) + f.registerCalls = append(f.registerCalls, proto.Clone(req).(*filer_pb.MountRegisterRequest)) return &filer_pb.MountRegisterResponse{}, nil } func (f *fakeFilerClient) MountList(ctx context.Context, req *filer_pb.MountListRequest, opts ...grpc.CallOption) (*filer_pb.MountListResponse, error) { f.mu.Lock() defer f.mu.Unlock() - resp := f.listResponse // value copy - return &resp, nil + return proto.Clone(&f.listResponse).(*filer_pb.MountListResponse), nil } // fakeFilerFleet maps each addr to its own fakeFilerClient so a test can diff --git a/weed/plugin/worker/worker.go b/weed/plugin/worker/worker.go index 5c270bfa7e..28a04c2f2c 100644 --- a/weed/plugin/worker/worker.go +++ b/weed/plugin/worker/worker.go @@ -660,8 +660,7 @@ func (w *Worker) buildHeartbeat() *plugin_pb.WorkerHeartbeat { if work == nil { continue } - cloned := *work - running = append(running, &cloned) + running = append(running, proto.Clone(work).(*plugin_pb.RunningWork)) } w.runningMu.RUnlock() diff --git a/weed/replication/sink/filersink/fetch_write.go b/weed/replication/sink/filersink/fetch_write.go index bffce98b9e..2675975dd8 100644 --- a/weed/replication/sink/filersink/fetch_write.go +++ b/weed/replication/sink/filersink/fetch_write.go @@ -229,8 +229,7 @@ func (fs *FilerSink) replicateOneManifestChunk(ctx context.Context, sourceChunk manifestDataChunks := make([]*filer_pb.FileChunk, len(replicatedResolvedChunks)) for i, chunk := range replicatedResolvedChunks { - copied := *chunk - manifestDataChunks[i] = &copied + manifestDataChunks[i] = proto.Clone(chunk).(*filer_pb.FileChunk) } filer_pb.BeforeEntrySerialization(manifestDataChunks) manifestData, err := proto.Marshal(&filer_pb.FileChunkManifest{ diff --git a/weed/s3api/chunked_reader_v4_test.go b/weed/s3api/chunked_reader_v4_test.go index e946dbb306..03b37215a1 100644 --- a/weed/s3api/chunked_reader_v4_test.go +++ b/weed/s3api/chunked_reader_v4_test.go @@ -111,7 +111,7 @@ func TestNewSignV4ChunkedReaderStreamingUnsignedPayloadTrailer(t *testing.T) { runWithRequest(iam, req, t, expectedPayload) } -func runWithRequest(iam IdentityAccessManagement, req *http.Request, t *testing.T, expectedPayload string) { +func runWithRequest(iam *IdentityAccessManagement, req *http.Request, t *testing.T, expectedPayload string) { reader, errCode := iam.newChunkedReader(req) assert.NotNil(t, reader) assert.Equal(t, s3err.ErrNone, errCode) @@ -124,11 +124,11 @@ func runWithRequest(iam IdentityAccessManagement, req *http.Request, t *testing. assert.Equal(t, expectedPayload, string(data)) } -func setupIam() IdentityAccessManagement { +func setupIam() *IdentityAccessManagement { // Create an IdentityAccessManagement instance // Add default access keys and secrets - iam := IdentityAccessManagement{ + iam := &IdentityAccessManagement{ identities: []*Identity{}, accessKeyIdent: map[string]*Identity{}, accounts: map[string]*Account{}, diff --git a/weed/stats/disk.go b/weed/stats/disk.go index a75cbbab41..67bc06fb86 100644 --- a/weed/stats/disk.go +++ b/weed/stats/disk.go @@ -12,6 +12,7 @@ import ( "github.com/seaweedfs/seaweedfs/weed/glog" "github.com/seaweedfs/seaweedfs/weed/pb/volume_server_pb" + "google.golang.org/protobuf/proto" ) type DiskIOProbeConfig struct { @@ -91,7 +92,7 @@ type diskState struct { statFailureCount int statSuccessCount int lastStatErr error - lastGoodStatus volume_server_pb.DiskStatus + lastGoodStatus *volume_server_pb.DiskStatus hasLastGood bool samples []ioSample @@ -497,7 +498,7 @@ func (s *diskState) updateErrorLocked(disk *volume_server_pb.DiskStatus, config } func (s *diskState) rememberGoodStatusLocked(disk *volume_server_pb.DiskStatus) { - s.lastGoodStatus = *disk + s.lastGoodStatus = proto.Clone(disk).(*volume_server_pb.DiskStatus) s.hasLastGood = true } @@ -508,7 +509,8 @@ func (s *diskState) applyLastGoodStatusLocked(disk *volume_server_pb.DiskStatus) dir := disk.Dir errorMessage := disk.Error - *disk = s.lastGoodStatus + proto.Reset(disk) + proto.Merge(disk, s.lastGoodStatus) disk.Dir = dir disk.Error = errorMessage }