From e93202c7c9437b61b76b4719a0c8d16c95299cd0 Mon Sep 17 00:00:00 2001 From: xiejian Date: Tue, 16 Dec 2025 17:04:58 +0800 Subject: [PATCH] refactor(proxy): replace VersionVolume with VolumeInfo with: #1000419929 Signed-off-by: xiejian --- blobstore/access/controller/mock_test.go | 7 ++----- blobstore/access/controller/volume.go | 12 ++++++------ blobstore/access/stream/stream_mock_test.go | 6 +++--- blobstore/api/proxy/cacher.go | 21 ++------------------- blobstore/api/proxy/client.go | 4 ++-- blobstore/api/proxy/client_test.go | 5 ++--- blobstore/cli/proxy/cacher.go | 2 +- blobstore/proxy/cacher/cacher.go | 2 +- blobstore/proxy/cacher/volume.go | 21 +++++++++------------ blobstore/proxy/cacher/volume_test.go | 12 +++++++----- blobstore/proxy/service_test.go | 4 ++-- blobstore/testing/mockproxy/cacher.go | 4 ++-- blobstore/testing/mocks/api_proxy.go | 4 ++-- 13 files changed, 41 insertions(+), 63 deletions(-) diff --git a/blobstore/access/controller/mock_test.go b/blobstore/access/controller/mock_test.go index 2be2c111f..ef9621b89 100644 --- a/blobstore/access/controller/mock_test.go +++ b/blobstore/access/controller/mock_test.go @@ -122,13 +122,12 @@ func init() { pcli := mocks.NewMockProxyClient(C(&testing.T{})) pcli.EXPECT().GetCacheVolume(A, A, A).AnyTimes().DoAndReturn( - func(ctx context.Context, _ string, args *proxy.CacheVolumeArgs) (*proxy.VersionVolume, error) { + func(ctx context.Context, _ string, args *proxy.CacheVolumeArgs) (*cmapi.VolumeInfo, error) { select { case <-ctx.Done(): return nil, ctx.Err() default: } - volume := new(proxy.VersionVolume) vid := args.Vid dataMu.Lock() dataCalled[vid]++ @@ -137,9 +136,7 @@ func init() { if vid == vid404 { return nil, errcode.ErrVolumeNotExist } - volume.VolumeInfo = val - volume.Version = volume.GetVersion() - return volume, nil + return &val, nil } return nil, errNotFound }) diff --git a/blobstore/access/controller/volume.go b/blobstore/access/controller/volume.go index 944a7a987..f9d67aec7 100644 --- a/blobstore/access/controller/volume.go +++ b/blobstore/access/controller/volume.go @@ -48,7 +48,7 @@ type VolumePhy struct { Vid proto.Vid CodeMode codemode.CodeMode IsPunish bool - Version uint32 + Version uint64 Timestamp int64 Units []Unit } @@ -240,7 +240,7 @@ func (v *volumeGetterImpl) Get(ctx context.Context, vid proto.Vid, isCache bool) } singleID := fmt.Sprintf("get-volume-%d", id) - ver := uint32(0) + ver := uint64(0) if phy != nil { ver = phy.Version } @@ -282,14 +282,14 @@ func (v *volumeGetterImpl) getFromLocalCache(_ context.Context, id cvid) *Volume return v.volumeMemCache.Get(id) } -func (v *volumeGetterImpl) getFromProxy(ctx context.Context, vid proto.Vid, flush bool, ver uint32) (*VolumePhy, error) { +func (v *volumeGetterImpl) getFromProxy(ctx context.Context, vid proto.Vid, flush bool, ver uint64) (*VolumePhy, error) { span := trace.SpanFromContextSafe(ctx) hosts, err := v.service.GetServiceHosts(ctx, proto.ServiceNameProxy) if err != nil { return nil, err } - var volume *proxy.VersionVolume + var volume *clustermgr.VolumeInfo cid := v.config.ClusterID id := addCVid(cid, vid) triedHosts := make(map[string]struct{}) @@ -327,7 +327,7 @@ func (v *volumeGetterImpl) getFromProxy(ctx context.Context, vid proto.Vid, flus phy := &VolumePhy{ Vid: volume.Vid, CodeMode: volume.CodeMode, - Version: volume.Version, + Version: uint64(volume.RouteVersion), Timestamp: time.Now().UnixNano(), Units: make([]Unit, len(volume.Units)), } @@ -344,7 +344,7 @@ func (v *volumeGetterImpl) getFromProxy(ctx context.Context, vid proto.Vid, flus } // flush update all proxy cache of this idc -func (v *volumeGetterImpl) flush(ctx context.Context, vid proto.Vid, ver uint32, hosts []string, except map[string]struct{}) { +func (v *volumeGetterImpl) flush(ctx context.Context, vid proto.Vid, ver uint64, hosts []string, except map[string]struct{}) { span := trace.SpanFromContextSafe(ctx) span.Infof("to flush volume cache %d on proxy:%v version:%d except:%v", vid, hosts, ver, except) diff --git a/blobstore/access/stream/stream_mock_test.go b/blobstore/access/stream/stream_mock_test.go index 943108675..38fe497b1 100644 --- a/blobstore/access/stream/stream_mock_test.go +++ b/blobstore/access/stream/stream_mock_test.go @@ -78,7 +78,7 @@ var ( cc controller.ClusterController clusterInfo *clustermgr.ClusterInfo - dataVolume *proxy.VersionVolume + dataVolume *clustermgr.VolumeInfo dataAllocs []proxy.AllocRet dataNodes map[string]clustermgr.ServiceInfo dataDisks map[proto.DiskID]clustermgr.BlobNodeDiskInfo @@ -329,7 +329,7 @@ func initMockData() { Vid: volumeID, } - dataVolume = &proxy.VersionVolume{VolumeInfo: clustermgr.VolumeInfo{ + dataVolume = &clustermgr.VolumeInfo{ VolumeInfoBase: clustermgr.VolumeInfoBase{ Vid: volumeID, CodeMode: codemode.EC6P6, @@ -343,7 +343,7 @@ func initMockData() { } return }(), - }} + } proxyNodes := make([]clustermgr.ServiceNode, 32) for idx := range proxyNodes { diff --git a/blobstore/api/proxy/cacher.go b/blobstore/api/proxy/cacher.go index 4260cd7bb..decaaedc8 100644 --- a/blobstore/api/proxy/cacher.go +++ b/blobstore/api/proxy/cacher.go @@ -17,34 +17,17 @@ package proxy import ( "context" "crypto/sha1" - "encoding/binary" "encoding/hex" - "hash/crc32" "strings" "github.com/cubefs/cubefs/blobstore/api/clustermgr" "github.com/cubefs/cubefs/blobstore/common/proto" ) -// VersionVolume volume with version. -type VersionVolume struct { - clustermgr.VolumeInfo - Version uint32 `json:"version,omitempty"` -} - -// GetVersion calculate version with volume's units. -func (v *VersionVolume) GetVersion() uint32 { - crcWriter := crc32.NewIEEE() - for _, unit := range v.Units { - binary.Write(crcWriter, binary.LittleEndian, uint64(unit.Vuid)) - } - return crcWriter.Sum32() -} - // CacheVolumeArgs volume arguments. type CacheVolumeArgs struct { Vid proto.Vid `json:"vid"` - Version uint32 `json:"version,omitempty"` + Version uint64 `json:"version,omitempty"` Flush bool `json:"flush,omitempty"` } @@ -56,7 +39,7 @@ type CacheDiskArgs struct { // Cacher interface of proxy cache. type Cacher interface { - GetCacheVolume(ctx context.Context, host string, args *CacheVolumeArgs) (*VersionVolume, error) + GetCacheVolume(ctx context.Context, host string, args *CacheVolumeArgs) (*clustermgr.VolumeInfo, error) GetCacheDisk(ctx context.Context, host string, args *CacheDiskArgs) (*clustermgr.BlobNodeDiskInfo, error) // Erase cache in proxy memory and diskv. // Volume key is "volume-{vid}", and disk key is "disk-{disk_id}". diff --git a/blobstore/api/proxy/client.go b/blobstore/api/proxy/client.go index 3906e09f2..b9e82cf3c 100644 --- a/blobstore/api/proxy/client.go +++ b/blobstore/api/proxy/client.go @@ -59,8 +59,8 @@ func (c *client) SendDeleteMsg(ctx context.Context, host string, args *DeleteArg return c.PostWith(ctx, host+"/deletemsg", nil, args) } -func (c *client) GetCacheVolume(ctx context.Context, host string, args *CacheVolumeArgs) (volume *VersionVolume, err error) { - volume = new(VersionVolume) +func (c *client) GetCacheVolume(ctx context.Context, host string, args *CacheVolumeArgs) (volume *clustermgr.VolumeInfo, err error) { + volume = new(clustermgr.VolumeInfo) url := fmt.Sprintf("%s/cache/volume/%d?flush=%v&version=%d", host, args.Vid, args.Flush, args.Version) err = c.GetWith(ctx, url, &volume) return diff --git a/blobstore/api/proxy/client_test.go b/blobstore/api/proxy/client_test.go index 7a581ad08..698828caa 100644 --- a/blobstore/api/proxy/client_test.go +++ b/blobstore/api/proxy/client_test.go @@ -58,7 +58,7 @@ func TestClient_GetCacheVolume(t *testing.T) { cli := New(&Config{}) mockServer := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) { w.WriteHeader(http.StatusOK) - w.Write([]byte(`{"vid": 111, "units": [{"vuid": 425335980033}]}`)) + w.Write([]byte(`{"vid": 111, "route_version": 2959854811, "units": [{"vuid": 425335980033}]}`)) })) defer mockServer.Close() for _, args := range []CacheVolumeArgs{ @@ -70,8 +70,7 @@ func TestClient_GetCacheVolume(t *testing.T) { volume, err := cli.GetCacheVolume(context.Background(), mockServer.URL, &args) require.NoError(t, err) require.Equal(t, proto.Vid(111), volume.Vid) - require.Equal(t, uint32(0), volume.Version) - require.Equal(t, uint32(0xb06bccdb), volume.GetVersion()) + require.Equal(t, uint64(0xb06bccdb), uint64(volume.RouteVersion)) } } diff --git a/blobstore/cli/proxy/cacher.go b/blobstore/cli/proxy/cacher.go index b79828f28..fff9cd27e 100644 --- a/blobstore/cli/proxy/cacher.go +++ b/blobstore/cli/proxy/cacher.go @@ -48,7 +48,7 @@ func addCmdCacher(cmd *grumble.Command) { volume, err := proxyCli.GetCacheVolume(common.CmdContext(), c.Flags.String(_host), &proxy.CacheVolumeArgs{ Vid: proto.Vid(c.Flags.Uint64("vid")), - Version: uint32(c.Flags.Uint64("version")), + Version: c.Flags.Uint64("version"), Flush: c.Flags.Bool("flush"), }) if err != nil { diff --git a/blobstore/proxy/cacher/cacher.go b/blobstore/proxy/cacher/cacher.go index 1fa31e5e1..bde24d141 100644 --- a/blobstore/proxy/cacher/cacher.go +++ b/blobstore/proxy/cacher/cacher.go @@ -118,7 +118,7 @@ func parseID(key string) (uint32, error) { // Cacher memory cache handlers. type Cacher interface { - GetVolume(ctx context.Context, args *proxy.CacheVolumeArgs) (*proxy.VersionVolume, error) + GetVolume(ctx context.Context, args *proxy.CacheVolumeArgs) (*clustermgr.VolumeInfo, error) GetDisk(ctx context.Context, args *proxy.CacheDiskArgs) (*clustermgr.BlobNodeDiskInfo, error) // Erase remove all if key is "ALL". Erase(ctx context.Context, key string) error diff --git a/blobstore/proxy/cacher/volume.go b/blobstore/proxy/cacher/volume.go index 872c3827f..e888b1a99 100644 --- a/blobstore/proxy/cacher/volume.go +++ b/blobstore/proxy/cacher/volume.go @@ -33,7 +33,7 @@ import ( const keyVolumeConcurrency = "volume" type expiryVolume struct { - proxy.VersionVolume + clustermgr.VolumeInfo ExpiryAt int64 `json:"expiry,omitempty"` // seconds } @@ -51,20 +51,20 @@ func decodeVolume(data []byte) (valueExpired, error) { return volume, err } -func (c *cacher) GetVolume(ctx context.Context, args *proxy.CacheVolumeArgs) (*proxy.VersionVolume, error) { +func (c *cacher) GetVolume(ctx context.Context, args *proxy.CacheVolumeArgs) (*clustermgr.VolumeInfo, error) { span := trace.SpanFromContextSafe(ctx) span.Debugf("try to get volume %+v", args) vid := args.Vid if vol := c.getVolume(span, vid); vol != nil { if !args.Flush { // read cache - return &vol.VersionVolume, nil + return &vol.VolumeInfo, nil } - if args.Version > 0 && args.Version != vol.Version { + if args.Version > 0 && args.Version != uint64(vol.RouteVersion) { span.Infof("request to flush, but version mismatch request(%d) != cache(%d)", - args.Version, vol.Version) - return &vol.VersionVolume, nil + args.Version, vol.RouteVersion) + return &vol.VolumeInfo, nil } } @@ -92,11 +92,11 @@ func (c *cacher) GetVolume(ctx context.Context, args *proxy.CacheVolumeArgs) (*p } c.volumeReport("clustermgr", "hit") - var result *proxy.VersionVolume + var result *clustermgr.VolumeInfo if err := c.withVolumeLock(vid, func() error { vol := c.newExpiryVolume(*volume) c.storeVolume(ctx, vid, vol) - result = &vol.VersionVolume + result = &vol.VolumeInfo return nil }); err != nil { return nil, err @@ -255,11 +255,8 @@ func (c *cacher) applyVolumeUpdate(ctx context.Context, payload *clustermgr.Rout func (c *cacher) newExpiryVolume(info clustermgr.VolumeInfo) *expiryVolume { vol := &expiryVolume{ - VersionVolume: proxy.VersionVolume{ - VolumeInfo: info, - }, + VolumeInfo: info, } - vol.VersionVolume.Version = vol.GetVersion() c.fillVolumeExpiry(vol) return vol } diff --git a/blobstore/proxy/cacher/volume_test.go b/blobstore/proxy/cacher/volume_test.go index 50151467d..80421d46c 100644 --- a/blobstore/proxy/cacher/volume_test.go +++ b/blobstore/proxy/cacher/volume_test.go @@ -72,32 +72,34 @@ func TestProxyCacherVolumeFlush(t *testing.T) { volume := new(clustermgr.VolumeInfo) volume.Units = []clustermgr.Unit{{Vuid: 1234}, {Vuid: 5678}} + version := proto.RouteVersion(12345678) + volume.RouteVersion = proto.RouteVersion(version) cmCli.EXPECT().GetVolumeInfo(A, A).Return(volume, nil).Times(1) for range [100]struct{}{} { vol, err := c.GetVolume(context.Background(), &proxy.CacheVolumeArgs{Vid: 1}) require.NoError(t, err) - require.Equal(t, uint32(0x9d31f755), vol.Version) + require.Equal(t, version, vol.RouteVersion) } cmCli.EXPECT().GetVolumeInfo(A, A).Return(volume, nil).Times(100) for range [100]struct{}{} { vol, err := c.GetVolume(context.Background(), &proxy.CacheVolumeArgs{Vid: 1, Flush: true}) require.NoError(t, err) - require.Equal(t, uint32(0x9d31f755), vol.Version) + require.Equal(t, version, vol.RouteVersion) } cmCli.EXPECT().GetVolumeInfo(A, A).Return(volume, nil).Times(1) for range [100]struct{}{} { vol, err := c.GetVolume(context.Background(), &proxy.CacheVolumeArgs{Vid: 3, Flush: true, Version: 0x01}) require.NoError(t, err) - require.Equal(t, uint32(0x9d31f755), vol.Version) + require.Equal(t, version, vol.RouteVersion) } - cmCli.EXPECT().GetVolumeInfo(A, A).Return(volume, nil).Times(100) + cmCli.EXPECT().GetVolumeInfo(A, A).Return(volume, nil).Times(1) for range [100]struct{}{} { vol, err := c.GetVolume(context.Background(), &proxy.CacheVolumeArgs{Vid: 4, Flush: true, Version: 0x9d31f755}) require.NoError(t, err) - require.Equal(t, uint32(0x9d31f755), vol.Version) + require.Equal(t, version, vol.RouteVersion) } } diff --git a/blobstore/proxy/service_test.go b/blobstore/proxy/service_test.go index 2311a10ec..024cc1834 100644 --- a/blobstore/proxy/service_test.go +++ b/blobstore/proxy/service_test.go @@ -105,8 +105,8 @@ func newMockService(t *testing.T) *Service { cacher := mock.NewMockCacher(ctr) cacher.EXPECT().GetVolume(A, A).AnyTimes().DoAndReturn( - func(_ context.Context, args *proxy.CacheVolumeArgs) (*proxy.VersionVolume, error) { - volume := new(proxy.VersionVolume) + func(_ context.Context, args *proxy.CacheVolumeArgs) (*clustermgr.VolumeInfo, error) { + volume := new(clustermgr.VolumeInfo) if args.Vid%2 == 0 { volume.Vid = args.Vid return volume, nil diff --git a/blobstore/testing/mockproxy/cacher.go b/blobstore/testing/mockproxy/cacher.go index 11170c787..4cde568b2 100644 --- a/blobstore/testing/mockproxy/cacher.go +++ b/blobstore/testing/mockproxy/cacher.go @@ -66,10 +66,10 @@ func (mr *MockCacherMockRecorder) GetDisk(arg0, arg1 interface{}) *gomock.Call { } // GetVolume mocks base method. -func (m *MockCacher) GetVolume(arg0 context.Context, arg1 *proxy.CacheVolumeArgs) (*proxy.VersionVolume, error) { +func (m *MockCacher) GetVolume(arg0 context.Context, arg1 *proxy.CacheVolumeArgs) (*clustermgr.VolumeInfo, error) { m.ctrl.T.Helper() ret := m.ctrl.Call(m, "GetVolume", arg0, arg1) - ret0, _ := ret[0].(*proxy.VersionVolume) + ret0, _ := ret[0].(*clustermgr.VolumeInfo) ret1, _ := ret[1].(error) return ret0, ret1 } diff --git a/blobstore/testing/mocks/api_proxy.go b/blobstore/testing/mocks/api_proxy.go index ff2d433d8..ff2dccf10 100644 --- a/blobstore/testing/mocks/api_proxy.go +++ b/blobstore/testing/mocks/api_proxy.go @@ -66,10 +66,10 @@ func (mr *MockProxyClientMockRecorder) GetCacheDisk(arg0, arg1, arg2 interface{} } // GetCacheVolume mocks base method. -func (m *MockProxyClient) GetCacheVolume(arg0 context.Context, arg1 string, arg2 *proxy.CacheVolumeArgs) (*proxy.VersionVolume, error) { +func (m *MockProxyClient) GetCacheVolume(arg0 context.Context, arg1 string, arg2 *proxy.CacheVolumeArgs) (*clustermgr.VolumeInfo, error) { m.ctrl.T.Helper() ret := m.ctrl.Call(m, "GetCacheVolume", arg0, arg1, arg2) - ret0, _ := ret[0].(*proxy.VersionVolume) + ret0, _ := ret[0].(*clustermgr.VolumeInfo) ret1, _ := ret[1].(error) return ret0, ret1 }