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

feat(proxy): add volume cache in proxy

Browse Source 
#1594

goos: linux
goarch: amd64
pkg: github.com/cubefs/cubefs/blobstore/proxy/cacher
cpu: Intel(R) Core(TM) i7-10700 CPU @ 2.90GHz
BenchmarkProxyMemoryHit
BenchmarkProxyMemoryHit-4         561333              2241 ns/op             752 B/op         27 allocs/op
BenchmarkProxyDiskvHit
BenchmarkProxyDiskvHit-4           16154             80132 ns/op            4035 B/op        102 allocs/op
BenchmarkProxyDiskvMiss
BenchmarkProxyDiskvMiss-4          10000            122750 ns/op            4067 B/op        104 allocs/op

Signed-off-by: slasher <shenjie1@oppo.com>
This commit is contained in:
slasher 2022-11-02 17:50:08 +08:00
parent 40dfa184ba
commit 0b03bdaa44
16 changed files with 890 additions and 4 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

1
blobstore/api/clustermgr/proto.go
View File

@ -68,6 +68,7 @@ type APIAccess interface {
// APIProxy sub of cluster manager api for allocator
type APIProxy interface {
GetConfig(ctx context.Context, key string) (string, error)
GetVolumeInfo(ctx context.Context, args *GetVolumeArgs) (*VolumeInfo, error)
AllocVolume(ctx context.Context, args *AllocVolumeArgs) (AllocatedVolumeInfos, error)
AllocBid(ctx context.Context, args *BidScopeArgs) (*BidScopeRet, error)
RetainVolume(ctx context.Context, args *RetainVolumeArgs) (RetainVolumes, error)

51
blobstore/api/proxy/cacher.go Normal file
View File

@ -0,0 +1,51 @@
// Copyright 2022 The CubeFS Authors.
//
// 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 proxy
import (
"context"
"encoding/binary"
"hash/crc32"
"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"`
Flush bool `json:"flush,omitempty"`
}
// Cacher interface of proxy cache.
type Cacher interface {
GetCacheVolume(ctx context.Context, host string, args *CacheVolumeArgs) (*VersionVolume, error)
}

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

@ -16,6 +16,7 @@ package proxy
import (
"context"
"fmt"
"github.com/cubefs/cubefs/blobstore/common/rpc"
)
@ -31,6 +32,7 @@ type client struct {
type Client interface {
MsgSender
Allocator
Cacher
}
func New(cfg *Config) Client {
@ -50,3 +52,10 @@ func (c *client) SendShardRepairMsg(ctx context.Context, host string, args *Shar
func (c *client) SendDeleteMsg(ctx context.Context, host string, args *DeleteArgs) error {
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)
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
}

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

@ -24,6 +24,7 @@ import (
"github.com/stretchr/testify/require"
"github.com/cubefs/cubefs/blobstore/api/clustermgr"
"github.com/cubefs/cubefs/blobstore/common/proto"
"github.com/cubefs/cubefs/blobstore/common/rpc"
_ "github.com/cubefs/cubefs/blobstore/testing/nolog"
)
@ -40,6 +41,27 @@ func TestClient_VolumeAlloc(t *testing.T) {
require.Equal(t, make([]AllocRet, 0), ret)
}
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}]}`))
}))
defer mockServer.Close()
for _, args := range []CacheVolumeArgs{
{Vid: 1, Version: 0, Flush: false},
{Vid: 1, Version: 0xb06bccdb, Flush: false},
{Vid: 2, Version: 0, Flush: true},
{Vid: 2, Version: 0xb06bccdb, Flush: true},
} {
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())
}
}
func TestLbClient_SendShardRepairMsg(t *testing.T) {
mqproxyServer := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, req *http.Request) {
w.WriteHeader(200)

1
blobstore/cmd/proxy/proxy.conf
View File

@ -5,6 +5,7 @@
"cluster_id": 1,
"default_alloc_vols_num" : 2,
"heartbeat_interval_s": 3,
"diskv_base_path": "./run/proxycache",
"clustermgr": {
"hosts": [
"http://127.0.0.1:9998",

41
blobstore/proxy/cacher.go Normal file
View File

@ -0,0 +1,41 @@
// Copyright 2022 The CubeFS Authors.
//
// 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 proxy
import (
"github.com/cubefs/cubefs/blobstore/api/proxy"
"github.com/cubefs/cubefs/blobstore/common/rpc"
"github.com/cubefs/cubefs/blobstore/common/trace"
)
// GetCacheVolume returns volume in cacher.
func (s *Service) GetCacheVolume(c *rpc.Context) {
args := new(proxy.CacheVolumeArgs)
if err := c.ParseArgs(args); err != nil {
c.RespondError(err)
return
}
ctx := c.Request.Context()
volume, err := s.cacher.GetVolume(ctx, args)
if err != nil {
span := trace.SpanFromContextSafe(ctx)
span.Warnf("get volume args:%v error:%s", args, err.Error())
c.RespondError(err)
return
}
c.RespondJSON(volume)
}

130
blobstore/proxy/cacher/cacher.go Normal file
View File

@ -0,0 +1,130 @@
// Copyright 2022 The CubeFS Authors.
//
// 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 cacher
import (
"context"
"crypto/sha1"
"encoding/hex"
"strings"
"github.com/peterbourgon/diskv/v3"
"golang.org/x/sync/singleflight"
"github.com/cubefs/cubefs/blobstore/api/clustermgr"
"github.com/cubefs/cubefs/blobstore/api/proxy"
"github.com/cubefs/cubefs/blobstore/common/memcache"
"github.com/cubefs/cubefs/blobstore/common/proto"
"github.com/cubefs/cubefs/blobstore/util/defaulter"
"github.com/cubefs/cubefs/blobstore/util/limit"
"github.com/cubefs/cubefs/blobstore/util/limit/keycount"
)
// Cacher is 2-level cache of clustermgr data.
// Data flow structure
// - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - -
// L1 | request | ---> | LRU cache | -- hit --> | response |
// | ^
// | miss | expired <---------------+
// | |
// L2 | | disk kv | ----- hit ------->
// | ^
// | miss | expired <---------------+
// | |
// Data| | clustermgr | ----- hit ------->
//
// - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - -
const (
_defaultCapacity = 1 << 20
_defaultExpirationS = 0 // 0 means no expiration
_defaultClustermgrConcurrency = 32
)
// ConfigCache is setting of cache.
type ConfigCache struct {
DiskvBasePath string `json:"diskv_base_path"`
DiskvTempDir string `json:"diskv_temp_dir"`
VolumeCapacity int `json:"volume_capacity"`
VolumeExpirationS int `json:"volume_expiration_seconds"`
}
func diskvKeyVolume(vid proto.Vid) string {
return "volume-" + vid.ToString()
}
func diskvKeyDisk(diskID proto.DiskID) string {
return "disk-" + diskID.ToString()
}
// diskvPathTransform transform key to multi-level path.
// eg: key(with '-') --> ~/hash(key)[0:2]/hash(key)[2:4]/key
func diskvPathTransform(key string) []string {
paths := strings.SplitN(key, "-", 2)
if len(paths) < 2 {
return []string{}
}
sha := sha1.New()
sha.Write([]byte(key))
h := hex.EncodeToString(sha.Sum(nil))
return []string{h[0:2], h[2:4]}
}
// Cacher memory cache handlers.
type Cacher interface {
GetVolume(ctx context.Context, args *proxy.CacheVolumeArgs) (*proxy.VersionVolume, error)
}
// New returns a Cacher.
func New(clusterID proto.ClusterID, config ConfigCache, cmClient clustermgr.APIProxy) (Cacher, error) {
defaulter.LessOrEqual(&config.VolumeCapacity, _defaultCapacity)
defaulter.LessOrEqual(&config.VolumeExpirationS, _defaultExpirationS)
vc, err := memcache.NewMemCache(context.Background(), config.VolumeCapacity)
if err != nil {
return nil, err
}
dv := diskv.New(diskv.Options{
CacheSizeMax: 1 << 20,
BasePath: config.DiskvBasePath,
TempDir: config.DiskvTempDir,
Transform: diskvPathTransform,
})
return &cacher{
config: config,
clusterID: clusterID,
cmClient: cmClient,
cmConcurrency: keycount.NewBlockingKeyCountLimit(_defaultClustermgrConcurrency),
singleRun: new(singleflight.Group),
volumeCache: vc,
diskv: dv,
}, nil
}
type cacher struct {
config ConfigCache
clusterID proto.ClusterID
cmClient clustermgr.APIProxy
cmConcurrency limit.Limiter
singleRun *singleflight.Group
volumeCache *memcache.MemCache
diskv *diskv.Diskv
}

69
blobstore/proxy/cacher/cacher_test.go Normal file
View File

@ -0,0 +1,69 @@
// Copyright 2022 The CubeFS Authors.
//
// 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 cacher
import (
"testing"
"github.com/stretchr/testify/require"
_ "github.com/cubefs/cubefs/blobstore/testing/nolog"
)
func TestProxyCacherConfigVolume(t *testing.T) {
config := ConfigCache{}
getCacher := func() *cacher {
c, err := New(1, config, nil)
require.NoError(t, err)
return c.(*cacher)
}
for _, cs := range []struct {
capacity, expCapacity int
expiration, expExpiration int
}{
{0, _defaultCapacity, 0, _defaultExpirationS},
{-100, _defaultCapacity, 0, _defaultExpirationS},
{-100, _defaultCapacity, -1, _defaultExpirationS},
{1 << 11, 1 << 11, 600, 600},
} {
config.VolumeCapacity = cs.capacity
config.VolumeExpirationS = cs.expiration
c := getCacher()
require.Equal(t, cs.expCapacity, c.config.VolumeCapacity)
require.Equal(t, cs.expExpiration, c.config.VolumeExpirationS)
}
}
func TestProxyCacherConfigPath(t *testing.T) {
for _, cs := range []struct {
key string
paths []string
}{
{"", []string{}},
{"akey", []string{}},
{"-id", []string{"8b", "d5"}},
{"volume-", []string{"fc", "08"}},
{"volume-111", []string{"59", "90"}},
{"volume-111-", []string{"cb", "dc"}},
{"volume-111-10", []string{"cf", "77"}},
{"disk-111", []string{"a6", "51"}},
{"disk-111-10", []string{"17", "3a"}},
{diskvKeyVolume(111), []string{"59", "90"}},
{diskvKeyDisk(111), []string{"a6", "51"}},
} {
require.Equal(t, cs.paths, diskvPathTransform(cs.key))
}
}

41
blobstore/proxy/cacher/metric.go Normal file
View File

@ -0,0 +1,41 @@
// Copyright 2022 The CubeFS Authors.
//
// 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 cacher
import (
"github.com/prometheus/client_golang/prometheus"
)
var cacheMetric = prometheus.NewCounterVec(
prometheus.CounterOpts{
Namespace: "blobstore",
Subsystem: "proxy",
Name: "cache",
Help: "cache statuts on proxy",
},
[]string{"cluster", "service", "name", "action"},
)
func init() {
prometheus.MustRegister(cacheMetric)
}
func (c *cacher) metricReport(service, name, action string) {
cacheMetric.WithLabelValues(c.clusterID.ToString(), service, name, action).Inc()
}
func (c *cacher) volumeReport(name, action string) {
c.metricReport("volume", name, action)
}

148
blobstore/proxy/cacher/volume.go Normal file
View File

@ -0,0 +1,148 @@
// Copyright 2022 The CubeFS Authors.
//
// 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 cacher
import (
"context"
"encoding/json"
"math/rand"
"time"
"github.com/cubefs/cubefs/blobstore/api/clustermgr"
"github.com/cubefs/cubefs/blobstore/api/proxy"
"github.com/cubefs/cubefs/blobstore/common/proto"
"github.com/cubefs/cubefs/blobstore/common/trace"
"github.com/cubefs/cubefs/blobstore/util/errors"
)
const keyVolumeConcurrency = "volume"
type expiryVolume struct {
proxy.VersionVolume
ExpiryAt int64 `json:"expiry,omitempty"` // seconds
}
func (v *expiryVolume) Expired() bool {
return v.ExpiryAt > 0 && time.Now().Unix() >= v.ExpiryAt
}
func encodeVolume(v *expiryVolume) ([]byte, error) {
return json.Marshal(v)
}
func decodeVolume(data []byte) (*expiryVolume, error) {
volume := new(expiryVolume)
err := json.Unmarshal(data, &volume)
return volume, err
}
func (c *cacher) GetVolume(ctx context.Context, args *proxy.CacheVolumeArgs) (*proxy.VersionVolume, 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
}
if args.Version > 0 && args.Version != vol.Version {
span.Infof("request to flush, but version mismatch request(%d) != cache(%d)",
args.Version, vol.Version)
return &vol.VersionVolume, nil
}
}
err := c.cmConcurrency.Acquire(keyVolumeConcurrency)
if err != nil {
return nil, err
}
defer c.cmConcurrency.Release(keyVolumeConcurrency)
val, err, _ := c.singleRun.Do("volume-"+vid.ToString(), func() (interface{}, error) {
return c.cmClient.GetVolumeInfo(ctx, &clustermgr.GetVolumeArgs{Vid: vid})
})
if err != nil {
c.volumeReport("clustermgr", "miss")
span.Error("get volume from clustermgr failed", errors.Detail(err))
return nil, err
}
volume, ok := val.(*clustermgr.VolumeInfo)
if !ok {
return nil, errors.New("error convert to volume struct after singleflight")
}
c.volumeReport("clustermgr", "hit")
vol := new(expiryVolume)
vol.VersionVolume.VolumeInfo = *volume
vol.VersionVolume.Version = vol.GetVersion()
if expire := c.config.VolumeExpirationS; expire > 0 {
// random expiration to reduce intensive clustermgr requests.
expiration := rand.Intn(expire) + expire
vol.ExpiryAt = time.Now().Add(time.Second * time.Duration(expiration)).Unix()
}
c.volumeCache.Set(vid, vol)
go func() {
if data, err := encodeVolume(vol); err == nil {
if err := c.diskv.Write(diskvKeyVolume(vid), data); err != nil {
span.Warnf("write to diskv vid:%d data:%s error:%s", vid, string(data), err.Error())
}
} else {
span.Warnf("encode vid:%d volume:%v error:%s", vid, vol, err.Error())
}
}()
return &vol.VersionVolume, nil
}
func (c *cacher) getVolume(span trace.Span, vid proto.Vid) *expiryVolume {
if val := c.volumeCache.Get(vid); val != nil {
c.volumeReport("memcache", "hit")
if vol, ok := val.(*expiryVolume); ok {
if !vol.Expired() {
span.Debug("hits at memory cache", vid)
return vol
}
c.volumeReport("memcache", "expired")
}
} else {
c.volumeReport("memcache", "miss")
}
data, err := c.diskv.Read(diskvKeyVolume(vid))
if err != nil {
c.volumeReport("diskv", "miss")
span.Warnf("read from diskv vid:%d error:%s", vid, err.Error())
return nil
}
vol, err := decodeVolume(data)
if err != nil {
c.volumeReport("diskv", "error")
span.Warnf("decode diskv vid:%d data:%s error:%s", vid, string(data), err.Error())
return nil
}
if !vol.Expired() {
c.volumeReport("diskv", "hit")
c.volumeCache.Set(vid, vol)
span.Debug("hits at diskv cache, set back to memory cache", vid)
return vol
}
c.volumeReport("diskv", "expired")
span.Debugf("expired at diskv vid:%d expiryat:%d", vid, vol.ExpiryAt)
return nil
}

246
blobstore/proxy/cacher/volume_test.go Normal file
View File

@ -0,0 +1,246 @@
// Copyright 2022 The CubeFS Authors.
//
// 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 cacher
import (
"context"
"fmt"
"math/rand"
"os"
"path"
"sync"
"testing"
"time"
"github.com/golang/mock/gomock"
"github.com/peterbourgon/diskv/v3"
"github.com/stretchr/testify/require"
"github.com/cubefs/cubefs/blobstore/api/clustermgr"
"github.com/cubefs/cubefs/blobstore/api/proxy"
errcode "github.com/cubefs/cubefs/blobstore/common/errors"
"github.com/cubefs/cubefs/blobstore/common/proto"
"github.com/cubefs/cubefs/blobstore/testing/mocks"
"github.com/cubefs/cubefs/blobstore/util/errors"
)
var (
A = gomock.Any()
C = gomock.NewController
)
func newCacher(t gomock.TestReporter, expiration int) (Cacher, *mocks.MockClientAPI, func()) {
cmCli := mocks.NewMockClientAPI(C(t))
basePath := path.Join(os.TempDir(), fmt.Sprintf("proxy-cacher-%d", rand.Intn(1000)+1000))
cacher, _ := New(1, ConfigCache{DiskvBasePath: basePath, VolumeExpirationS: expiration}, cmCli)
return cacher, cmCli, func() { os.RemoveAll(basePath) }
}
func TestProxyCacherVolumeUpdate(t *testing.T) {
c, cmCli, clean := newCacher(t, 2)
defer clean()
cmCli.EXPECT().GetVolumeInfo(A, A).Return(&clustermgr.VolumeInfo{}, nil).Times(4)
for range [100]struct{}{} {
_, err := c.GetVolume(context.Background(), &proxy.CacheVolumeArgs{Vid: 1})
require.NoError(t, err)
}
for range [100]struct{}{} {
_, err := c.GetVolume(context.Background(), &proxy.CacheVolumeArgs{Vid: 2})
require.NoError(t, err)
}
time.Sleep(time.Second * 4) // expired
for range [100]struct{}{} {
_, err := c.GetVolume(context.Background(), &proxy.CacheVolumeArgs{Vid: 1})
require.NoError(t, err)
}
for range [100]struct{}{} {
_, err := c.GetVolume(context.Background(), &proxy.CacheVolumeArgs{Vid: 2})
require.NoError(t, err)
}
}
func TestProxyCacherVolumeFlush(t *testing.T) {
c, cmCli, clean := newCacher(t, 0)
defer clean()
volume := new(clustermgr.VolumeInfo)
volume.Units = []clustermgr.Unit{{Vuid: 1234}, {Vuid: 5678}}
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)
}
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)
}
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)
}
cmCli.EXPECT().GetVolumeInfo(A, A).Return(volume, nil).Times(100)
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)
}
}
func TestProxyCacherVolumeSingle(t *testing.T) {
c, cmCli, clean := newCacher(t, 0)
defer clean()
cmCli.EXPECT().GetVolumeInfo(A, A).DoAndReturn(
func(_ context.Context, _ *clustermgr.GetVolumeArgs) (*clustermgr.VolumeInfo, error) {
time.Sleep(time.Second)
return &clustermgr.VolumeInfo{}, nil
}).Times(3)
var wg sync.WaitGroup
wg.Add(_defaultClustermgrConcurrency)
for range [_defaultClustermgrConcurrency]struct{}{} {
go func() {
c.GetVolume(context.Background(), &proxy.CacheVolumeArgs{Vid: 1})
wg.Done()
}()
}
wg.Wait()
wg.Add(_defaultClustermgrConcurrency + 1)
for range [_defaultClustermgrConcurrency + 1]struct{}{} {
go func() {
c.GetVolume(context.Background(), &proxy.CacheVolumeArgs{Vid: 2})
wg.Done()
}()
}
wg.Wait()
}
func TestProxyCacherVolumeError(t *testing.T) {
c, cmCli, clean := newCacher(t, 0)
defer clean()
cmCli.EXPECT().GetVolumeInfo(A, A).Return(nil, errors.New("mock error")).Times(1)
cmCli.EXPECT().GetVolumeInfo(A, A).Return(nil, errcode.ErrVolumeNotExist).Times(2)
_, err := c.GetVolume(context.Background(), &proxy.CacheVolumeArgs{Vid: 1})
require.Error(t, err)
_, err = c.GetVolume(context.Background(), &proxy.CacheVolumeArgs{Vid: 2, Flush: true})
require.ErrorIs(t, errcode.ErrVolumeNotExist, err)
_, err = c.GetVolume(context.Background(), &proxy.CacheVolumeArgs{Vid: 1, Flush: false})
require.ErrorIs(t, errcode.ErrVolumeNotExist, err)
}
func TestProxyCacherVolumeCacheMiss(t *testing.T) {
c, cmCli, clean := newCacher(t, 2)
defer clean()
cmCli.EXPECT().GetVolumeInfo(A, A).Return(&clustermgr.VolumeInfo{}, nil).Times(3)
_, err := c.GetVolume(context.Background(), &proxy.CacheVolumeArgs{Vid: 1})
require.NoError(t, err)
time.Sleep(time.Microsecond * 200) // waiting diskv write to disk
basePath := c.(*cacher).config.DiskvBasePath
{ // memory cache miss, load from diskv
c, _ = New(1, ConfigCache{DiskvBasePath: basePath, VolumeExpirationS: 2}, cmCli)
_, err = c.GetVolume(context.Background(), &proxy.CacheVolumeArgs{Vid: 1})
require.NoError(t, err)
}
{ // cannot decode diskv value
file, err := os.OpenFile(path.Join(basePath, "a7", "13", "volume-1"), os.O_RDWR, 0o644)
require.NoError(t, err)
file.Write([]byte("}}}}}"))
file.Close()
c, _ = New(1, ConfigCache{DiskvBasePath: basePath, VolumeExpirationS: 2}, cmCli)
_, err = c.GetVolume(context.Background(), &proxy.CacheVolumeArgs{Vid: 1})
require.NoError(t, err)
}
{ // load diskv expired
c, _ = New(1, ConfigCache{DiskvBasePath: basePath, VolumeExpirationS: 2}, cmCli)
time.Sleep(time.Second * 3)
_, err = c.GetVolume(context.Background(), &proxy.CacheVolumeArgs{Vid: 1})
require.NoError(t, err)
}
}
func BenchmarkProxyMemoryHit(b *testing.B) {
c, cmCli, clean := newCacher(b, 0)
defer clean()
cmCli.EXPECT().GetVolumeInfo(A, A).Return(&clustermgr.VolumeInfo{}, nil).AnyTimes()
ctx := context.Background()
args := &proxy.CacheVolumeArgs{Vid: 1}
_, err := c.GetVolume(ctx, args)
require.NoError(b, err)
b.ResetTimer()
for ii := 0; ii < b.N; ii++ {
c.GetVolume(ctx, args)
}
}
func BenchmarkProxyDiskvHit(b *testing.B) {
c, cmCli, clean := newCacher(b, 0)
defer clean()
c.(*cacher).diskv.AdvancedTransform = func(s string) *diskv.PathKey {
return &diskv.PathKey{Path: diskvPathTransform(s), FileName: diskvKeyVolume(1)}
}
cmCli.EXPECT().GetVolumeInfo(A, A).Return(&clustermgr.VolumeInfo{}, nil).AnyTimes()
ctx := context.Background()
args := &proxy.CacheVolumeArgs{}
b.ResetTimer()
for ii := 0; ii < b.N; ii++ {
args.Vid = proto.Vid(ii)
c.GetVolume(ctx, args)
}
}
func BenchmarkProxyDiskvMiss(b *testing.B) {
c, cmCli, clean := newCacher(b, 0)
defer clean()
cmCli.EXPECT().GetVolumeInfo(A, A).Return(&clustermgr.VolumeInfo{}, nil).AnyTimes()
ctx := context.Background()
args := &proxy.CacheVolumeArgs{}
b.ResetTimer()
for ii := 0; ii < b.N; ii++ {
args.Vid = proto.Vid(ii)
c.GetVolume(ctx, args)
}
}
func BenchmarkProxyClusterMiss(b *testing.B) {
c, cmCli, clean := newCacher(b, 0)
defer clean()
cmCli.EXPECT().GetVolumeInfo(A, A).Return(nil, errcode.ErrVolumeNotExist).AnyTimes()
ctx := context.Background()
args := &proxy.CacheVolumeArgs{Vid: 1}
b.ResetTimer()
for ii := 0; ii < b.N; ii++ {
args.Vid = proto.Vid(ii)
c.GetVolume(ctx, args)
}
}

51
blobstore/proxy/mock/cacher_mock.go Normal file
View File

@ -0,0 +1,51 @@
// Code generated by MockGen. DO NOT EDIT.
// Source: github.com/cubefs/cubefs/blobstore/proxy/cacher (interfaces: Cacher)
// Package mock is a generated GoMock package.
package mock
import (
context "context"
reflect "reflect"
proxy "github.com/cubefs/cubefs/blobstore/api/proxy"
gomock "github.com/golang/mock/gomock"
)
// MockCacher is a mock of Cacher interface.
type MockCacher struct {
ctrl *gomock.Controller
recorder *MockCacherMockRecorder
}
// MockCacherMockRecorder is the mock recorder for MockCacher.
type MockCacherMockRecorder struct {
mock *MockCacher
}
// NewMockCacher creates a new mock instance.
func NewMockCacher(ctrl *gomock.Controller) *MockCacher {
mock := &MockCacher{ctrl: ctrl}
mock.recorder = &MockCacherMockRecorder{mock}
return mock
}
// EXPECT returns an object that allows the caller to indicate expected use.
func (m *MockCacher) EXPECT() *MockCacherMockRecorder {
return m.recorder
}
// GetVolume mocks base method.
func (m *MockCacher) GetVolume(arg0 context.Context, arg1 *proxy.CacheVolumeArgs) (*proxy.VersionVolume, error) {
m.ctrl.T.Helper()
ret := m.ctrl.Call(m, "GetVolume", arg0, arg1)
ret0, _ := ret[0].(*proxy.VersionVolume)
ret1, _ := ret[1].(error)
return ret0, ret1
}
// GetVolume indicates an expected call of GetVolume.
func (mr *MockCacherMockRecorder) GetVolume(arg0, arg1 interface{}) *gomock.Call {
mr.mock.ctrl.T.Helper()
return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "GetVolume", reflect.TypeOf((*MockCacher)(nil).GetVolume), arg0, arg1)
}

1
blobstore/proxy/mock/mock_gen.go
View File

@ -18,3 +18,4 @@ package mock
//go:generate mockgen -destination=./mq_mock.go -package=mock -mock_names BlobDeleteHandler=MockBlobDeleteHandler,ShardRepairHandler=MockShardRepairHandler,Producer=MockProducer github.com/cubefs/cubefs/blobstore/proxy/mq BlobDeleteHandler,ShardRepairHandler,Producer
//go:generate mockgen -destination=./allocator_mock.go -package=mock -mock_names BlobDeleteHandler=MockBlobDeleteHandler,ShardRepairHandler=MockShardRepairHandler,Producer=MockProducer github.com/cubefs/cubefs/blobstore/proxy/allocator VolumeMgr
//go:generate mockgen -destination=./cacher_mock.go -package=mock -mock_names Cacher=MockCacher github.com/cubefs/cubefs/blobstore/proxy/cacher Cacher

20
blobstore/proxy/service.go
View File

@ -19,13 +19,14 @@ import (
"net/http"
"github.com/cubefs/cubefs/blobstore/api/clustermgr"
api "github.com/cubefs/cubefs/blobstore/api/proxy"
"github.com/cubefs/cubefs/blobstore/api/proxy"
"github.com/cubefs/cubefs/blobstore/cmd"
"github.com/cubefs/cubefs/blobstore/common/config"
"github.com/cubefs/cubefs/blobstore/common/kafka"
"github.com/cubefs/cubefs/blobstore/common/proto"
"github.com/cubefs/cubefs/blobstore/common/rpc"
alloc "github.com/cubefs/cubefs/blobstore/proxy/allocator"
"github.com/cubefs/cubefs/blobstore/proxy/cacher"
"github.com/cubefs/cubefs/blobstore/proxy/mq"
"github.com/cubefs/cubefs/blobstore/util/defaulter"
"github.com/cubefs/cubefs/blobstore/util/errors"
@ -61,6 +62,7 @@ type Config struct {
alloc.BlobConfig
alloc.VolConfig
cacher.ConfigCache
HeartbeatIntervalS uint32 `json:"heartbeat_interval_s"` // proxy heartbeat interval to ClusterManager
HeartbeatTicks uint32 `json:"heartbeat_ticks"`
@ -90,9 +92,10 @@ type Service struct {
// mq
shardRepairMgr mq.ShardRepairHandler
blobDeleteMgr mq.BlobDeleteHandler
// allocator
volumeMgr alloc.VolumeMgr
// cacher
cacher cacher.Cacher
}
func init() {
@ -144,6 +147,11 @@ func New(cfg Config, cmcli clustermgr.APIProxy) *Service {
log.Fatalf("fail to new volumeMgr, error: %s", err.Error())
}
cacher, err := cacher.New(cfg.ClusterID, cfg.ConfigCache, cmcli)
if err != nil {
log.Fatalf("fail to new cacher, error: %s", err.Error())
}
// register to clustermgr
node := clustermgr.ServiceNode{
ClusterID: uint64(cfg.ClusterID),
@ -159,6 +167,7 @@ func New(cfg Config, cmcli clustermgr.APIProxy) *Service {
return &Service{
Config: cfg,
volumeMgr: volumeMgr,
cacher: cacher,
shardRepairMgr: shardRepairMgr,
blobDeleteMgr: blobDeleteMgr,
}
@ -166,7 +175,8 @@ func New(cfg Config, cmcli clustermgr.APIProxy) *Service {
func NewHandler(service *Service) *rpc.Router {
router := rpc.New()
rpc.RegisterArgsParser(&api.ListVolsArgs{}, "json")
rpc.RegisterArgsParser(&proxy.ListVolsArgs{}, "json")
rpc.RegisterArgsParser(&proxy.CacheVolumeArgs{}, "json")
// POST /volume/alloc
// request body: json
@ -184,6 +194,10 @@ func NewHandler(service *Service) *rpc.Router {
// request body: json
router.Handle(http.MethodPost, "/deletemsg", service.SendDeleteMessage, rpc.OptArgsBody())
// GET /cache/volume/{vid}?flush={flush}&version={version}
// response body: json
router.Handle(http.MethodGet, "/cache/volume/:vid", service.GetCacheVolume, rpc.OptArgsURI(), rpc.OptArgsQuery())
return router
}

48
blobstore/proxy/service_test.go
View File

@ -22,7 +22,6 @@ import (
"github.com/Shopify/sarama"
"github.com/golang/mock/gomock"
_ "github.com/peterbourgon/diskv/v3"
"github.com/stretchr/testify/require"
"github.com/cubefs/cubefs/blobstore/api/clustermgr"
@ -39,6 +38,8 @@ import (
)
var (
A = gomock.Any()
errCodeMode = errors.New("codeMode not exist")
errBidCount = errors.New("count too large")
ctx = context.Background()
@ -95,6 +96,20 @@ func newMockService(t *testing.T) *Service {
return nil, nil, nil
})
cacher := mock.NewMockCacher(ctr)
cacher.EXPECT().GetVolume(A, A).AnyTimes().DoAndReturn(
func(_ context.Context, args *proxy.CacheVolumeArgs) (*proxy.VersionVolume, error) {
volume := new(proxy.VersionVolume)
if args.Vid%2 == 0 {
volume.Vid = args.Vid
return volume, nil
}
if args.Flush {
return nil, errcode.ErrVolumeNotExist
}
return nil, errors.New("internal error")
})
return &Service{
Config: Config{
VolConfig: allocator.VolConfig{
@ -104,6 +119,7 @@ func newMockService(t *testing.T) *Service {
shardRepairMgr: shardRepairMgr,
blobDeleteMgr: blobDeleteMgr,
volumeMgr: volumeMgr,
cacher: cacher,
}
}
@ -282,6 +298,36 @@ func TestService_Allocator(t *testing.T) {
}
}
func TestService_Cacher(t *testing.T) {
url := runMockService(newMockService(t)) + "/cache/volume/"
cli := newClient()
var volume clustermgr.VolumeInfo
{
err := cli.GetWith(ctx, url+"1024", &volume)
require.NoError(t, err)
require.Equal(t, proto.Vid(1024), volume.Vid)
}
{
err := cli.GetWith(ctx, url+"111", &volume)
require.Error(t, err)
}
{
err := cli.GetWith(ctx, url+"111?flush=0", &volume)
require.Error(t, err)
require.Equal(t, 500, rpc.DetectStatusCode(err))
}
{
err := cli.GetWith(ctx, url+"111?flush=1", &volume)
require.Error(t, err)
require.Equal(t, errcode.CodeVolumeNotExist, rpc.DetectStatusCode(err))
}
{
err := cli.GetWith(ctx, url+"111?flush=true", &volume)
require.Error(t, err)
require.Equal(t, errcode.CodeVolumeNotExist, rpc.DetectStatusCode(err))
}
}
func TestConfigFix(t *testing.T) {
testCases := []struct {
cfg *Config

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

@ -35,6 +35,21 @@ func (m *MockProxyClient) EXPECT() *MockProxyClientMockRecorder {
return m.recorder
}
// GetCacheVolume mocks base method.
func (m *MockProxyClient) GetCacheVolume(arg0 context.Context, arg1 string, arg2 *proxy.CacheVolumeArgs) (*proxy.VersionVolume, error) {
m.ctrl.T.Helper()
ret := m.ctrl.Call(m, "GetCacheVolume", arg0, arg1, arg2)
ret0, _ := ret[0].(*proxy.VersionVolume)
ret1, _ := ret[1].(error)
return ret0, ret1
}
// GetCacheVolume indicates an expected call of GetCacheVolume.
func (mr *MockProxyClientMockRecorder) GetCacheVolume(arg0, arg1, arg2 interface{}) *gomock.Call {
mr.mock.ctrl.T.Helper()
return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "GetCacheVolume", reflect.TypeOf((*MockProxyClient)(nil).GetCacheVolume), arg0, arg1, arg2)
}
// SendDeleteMsg mocks base method.
func (m *MockProxyClient) SendDeleteMsg(arg0 context.Context, arg1 string, arg2 *proxy.DeleteArgs) error {
m.ctrl.T.Helper()