gian/cubefs
1
0
Fork 0
mirror of https://github.com/cubefs/cubefs.git synced 2026-08-02 02:00:56 +00:00
Code Issues Actions Packages Projects Releases Wiki Activity

refactor(proxy): replace VersionVolume with VolumeInfo

Browse Source 
with: #1000419929

Signed-off-by: xiejian <xiejian3@oppo.com>
This commit is contained in:
xiejian 2025-12-16 17:04:58 +08:00 committed by 梁曟風
parent 00919b446f
commit e93202c7c9
13 changed files with 41 additions and 63 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

7
blobstore/access/controller/mock_test.go
View File

@ -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
})

12
blobstore/access/controller/volume.go
View File

@ -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)

6
blobstore/access/stream/stream_mock_test.go
View File

@ -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 {

21
blobstore/api/proxy/cacher.go
View File

@ -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}".

4
blobstore/api/proxy/client.go
View File

@ -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

5
blobstore/api/proxy/client_test.go
View File

@ -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))
}
}

2
blobstore/cli/proxy/cacher.go
View File

@ -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 {

2
blobstore/proxy/cacher/cacher.go
View File

@ -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

21
blobstore/proxy/cacher/volume.go
View File

@ -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
}

12
blobstore/proxy/cacher/volume_test.go
View File

@ -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)
}
}

4
blobstore/proxy/service_test.go
View File

@ -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

4
blobstore/testing/mockproxy/cacher.go
View File

@ -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
}

4
blobstore/testing/mocks/api_proxy.go
View File

@ -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
}