From 9dad1ee7cfb368ebb3b9f4e3609ecad99b015088 Mon Sep 17 00:00:00 2001 From: slasher Date: Wed, 6 Nov 2024 17:13:25 +0800 Subject: [PATCH] feat(access): punish volume disk and service with own value . #22762717 Signed-off-by: slasher --- blobstore/access/controller/cluster.go | 20 +++-- blobstore/access/controller/service.go | 81 ++++++++++++--------- blobstore/access/controller/service_test.go | 14 ++-- blobstore/access/controller/volume.go | 72 +++++++++--------- blobstore/access/controller/volume_test.go | 27 ++++--- blobstore/access/stream/stream.go | 17 ++++- blobstore/access/stream/stream_blob.go | 7 -- blobstore/access/stream/stream_mock_test.go | 11 ++- 8 files changed, 140 insertions(+), 109 deletions(-) diff --git a/blobstore/access/controller/cluster.go b/blobstore/access/controller/cluster.go index e1a4d03e4..03b36fae2 100644 --- a/blobstore/access/controller/cluster.go +++ b/blobstore/access/controller/cluster.go @@ -138,12 +138,11 @@ type ClusterConfig struct { Region string `json:"region"` RegionMagic string `json:"region_magic"` ClusterReloadSecs int `json:"cluster_reload_secs"` - ServiceReloadSecs int `json:"service_reload_secs"` ShardReloadSecs int `json:"shard_reload_secs"` CMClientConfig cmapi.Config `json:"clustermgr_client_config"` - ServicePunishThreshold uint32 `json:"service_punish_threshold"` - ServicePunishValidIntervalS int `json:"service_punish_valid_interval_s"` + ServiceConfig + VolumeConfig ConsulAgentAddr string `json:"consul_agent_addr"` ConsulToken string `json:"consul_token"` @@ -375,20 +374,19 @@ func (c *clusterControllerImpl) deal(ctx context.Context, } } - serviceController, err := NewServiceController(ServiceConfig{ - ClusterID: clusterID, - IDC: c.config.IDC, - ReloadSec: c.config.ServiceReloadSecs, - ServicePunishThreshold: c.config.ServicePunishThreshold, - ServicePunishValidIntervalS: c.config.ServicePunishValidIntervalS, - }, cmCli, c.proxy, c.stopCh) + serviceConfig := c.config.ServiceConfig + serviceConfig.ClusterID = clusterID + serviceConfig.IDC = c.config.IDC + serviceController, err := NewServiceController(serviceConfig, cmCli, c.proxy, c.stopCh) if err != nil { removeThisCluster() span.Warn("new service manager failed", clusterID, err) continue } - volumeGetter, err := NewVolumeGetter(clusterID, serviceController, c.proxy, -1, c.stopCh) + volumeConfig := c.config.VolumeConfig + volumeConfig.ClusterID = clusterID + volumeGetter, err := NewVolumeGetter(volumeConfig, serviceController, c.proxy, c.stopCh) if err != nil { removeThisCluster() span.Warn("new volume getter failed", clusterID, err) diff --git a/blobstore/access/controller/service.go b/blobstore/access/controller/service.go index 0a45f968a..d28f0684b 100644 --- a/blobstore/access/controller/service.go +++ b/blobstore/access/controller/service.go @@ -34,13 +34,8 @@ import ( ) const ( - _diskHostServicePrefix = "diskhost" - _shardnodeDiskServicePrefix = "sdDiskhost" - - // default service punish check valid interval - defaultServicePinishValidIntervalS int = 30 - // default service punish check threshold - defaultServicePinishThreshold uint32 = 3 + _primaryDisk = "_disk_" + _primaryShardnodeDisk = "_sddisk_" ) // HostIDC item of host with idc @@ -98,12 +93,15 @@ func (h *hostItem) isPunish() bool { // ServiceConfig service config type ServiceConfig struct { - ClusterID proto.ClusterID - IDC string - ReloadSec int - LoadDiskInterval int - ServicePunishThreshold uint32 - ServicePunishValidIntervalS int + ClusterID proto.ClusterID `json:"-"` + IDC string `json:"-"` + + ServiceReloadSecs int `json:"service_reload_secs"` + LoadDiskIntervalS int `json:"load_disk_interval_s"` + DiskPunishThreshold uint32 `json:"disk_punish_threshold"` + DiskPunishValidIntervalS int `json:"disk_punish_valid_interval_s"` + ServicePunishThreshold uint32 `json:"service_punish_threshold"` + ServicePunishValidIntervalS int `json:"service_punish_valid_interval_s"` } type serviceControllerImpl struct { @@ -125,10 +123,12 @@ type serviceControllerImpl struct { func NewServiceController(cfg ServiceConfig, cmCli clustermgr.APIAccess, proxy proxy.Cacher, stopCh <-chan struct{}, ) (ServiceController, error) { - defaulter.Equal(&cfg.ServicePunishThreshold, defaultServicePinishThreshold) - defaulter.LessOrEqual(&cfg.ServicePunishValidIntervalS, defaultServicePinishValidIntervalS) - defaulter.LessOrEqual(&cfg.LoadDiskInterval, int(300)) - defaulter.LessOrEqual(&cfg.ReloadSec, int(10)) + defaulter.IntegerLessOrEqual(&cfg.ServiceReloadSecs, 10) + defaulter.IntegerLessOrEqual(&cfg.LoadDiskIntervalS, 300) + defaulter.IntegerEqual(&cfg.DiskPunishThreshold, 3) + defaulter.IntegerLessOrEqual(&cfg.DiskPunishValidIntervalS, 30) + defaulter.IntegerEqual(&cfg.ServicePunishThreshold, 3) + defaulter.IntegerLessOrEqual(&cfg.ServicePunishValidIntervalS, 30) controller := &serviceControllerImpl{ serviceHosts: serviceMap{ @@ -151,7 +151,7 @@ func NewServiceController(cfg ServiceConfig, return controller, nil } go func() { - tick := time.NewTicker(time.Duration(cfg.ReloadSec) * time.Second) + tick := time.NewTicker(time.Duration(cfg.ServiceReloadSecs) * time.Second) defer tick.Stop() for { select { @@ -165,7 +165,7 @@ func NewServiceController(cfg ServiceConfig, } }() go func() { - tick := time.NewTicker(time.Duration(cfg.LoadDiskInterval) * time.Second) + tick := time.NewTicker(time.Duration(cfg.LoadDiskIntervalS) * time.Second) defer tick.Stop() for { controller.loadBrokenDisks() @@ -201,7 +201,7 @@ func (s *serviceControllerImpl) load(cid proto.ClusterID, idc string) error { } if len(hostItems) > 0 { for _, item := range hostItems { - s.allServices.Store(serviceName+item.host, item) + s.allServices.Store(s.getServiceKey(serviceName, item.host), item) span.Debugf("store node %+v", item) } s.serviceHosts[serviceName].Store(hostItems) @@ -372,7 +372,7 @@ func (s *serviceControllerImpl) GetDiskHost(ctx context.Context, diskID proto.Di _, broken := s.brokenDisks.Load(diskID) - v, ok := s.allServices.Load(_diskHostServicePrefix + (diskID.ToString())) + v, ok := s.allServices.Load(s.getServiceKey(_primaryDisk, diskID)) if ok { item := v.(*hostItem) return &HostIDC{ @@ -403,7 +403,7 @@ func (s *serviceControllerImpl) GetDiskHost(ctx context.Context, diskID proto.Di diskInfo := ret.(*clustermgr.BlobNodeDiskInfo) item := &hostItem{host: diskInfo.Host, idc: diskInfo.Idc} - s.allServices.Store(_diskHostServicePrefix+(diskInfo.DiskID.ToString()), item) + s.allServices.Store(s.getServiceKey(_primaryDisk, diskInfo.DiskID), item) return &HostIDC{ Host: item.host, IDC: item.idc, @@ -415,7 +415,7 @@ func (s *serviceControllerImpl) GetShardnodeHost(ctx context.Context, diskID pro span := trace.SpanFromContextSafe(ctx) _, broken := s.sdBrokenDisks.Load(diskID) - v, ok := s.allServices.Load(_shardnodeDiskServicePrefix + (diskID.ToString())) + v, ok := s.allServices.Load(s.getServiceKey(_primaryShardnodeDisk, diskID)) if ok { item := v.(*hostItem) return &HostIDC{ @@ -441,7 +441,7 @@ func (s *serviceControllerImpl) GetShardnodeHost(ctx context.Context, diskID pro diskInfo := ret.(*clustermgr.ShardNodeDiskInfo) item := &hostItem{host: diskInfo.Host, idc: diskInfo.Idc} - s.allServices.Store(_shardnodeDiskServicePrefix+(diskInfo.DiskID.ToString()), item) + s.allServices.Store(s.getServiceKey(_primaryShardnodeDisk, diskInfo.DiskID), item) return &HostIDC{ Host: item.host, IDC: item.idc, @@ -451,9 +451,9 @@ func (s *serviceControllerImpl) GetShardnodeHost(ctx context.Context, diskID pro // PunishService will punish an service host for an punishTimeSec interval func (s *serviceControllerImpl) PunishService(ctx context.Context, service, host string, punishTimeSec int) { - v, ok := s.allServices.Load(service + host) + v, ok := s.allServices.Load(s.getServiceKey(service, host)) if !ok { - panic(fmt.Sprintf("can't find host in all services map, %s-%s", service, host)) + panic(fmt.Sprintf("can't find host in all services map, %s", s.getServiceKey(service, host))) } item := v.(*hostItem) @@ -463,33 +463,42 @@ func (s *serviceControllerImpl) PunishService(ctx context.Context, service, host // PunishDisk will punish a disk host for an punishTimeSec interval func (s *serviceControllerImpl) PunishDisk(ctx context.Context, diskID proto.DiskID, punishTimeSec int) { - s.PunishService(ctx, _diskHostServicePrefix, diskID.ToString(), punishTimeSec) + s.PunishService(ctx, _primaryDisk, diskID.ToString(), punishTimeSec) } // PunishShardnode will punish a shardnode disk host for an punishTimeSec interval func (s *serviceControllerImpl) PunishShardnode(ctx context.Context, diskID proto.DiskID, punishTimeSec int) { - s.PunishService(ctx, _shardnodeDiskServicePrefix, diskID.ToString(), punishTimeSec) + s.PunishService(ctx, _primaryShardnodeDisk, diskID.ToString(), punishTimeSec) } // PunishDiskWithThreshold will punish a disk host for // an punishTimeSec interval if disk host failed times satisfied with threshold func (s *serviceControllerImpl) PunishDiskWithThreshold(ctx context.Context, diskID proto.DiskID, punishTimeSec int) { - s.PunishServiceWithThreshold(ctx, _diskHostServicePrefix, diskID.ToString(), punishTimeSec) + s.punishWith(ctx, _primaryDisk, diskID.ToString(), punishTimeSec, + s.config.DiskPunishThreshold, s.config.DiskPunishValidIntervalS) } // PunishServiceWithThreshold will punish an service host for // an punishTimeSec interval if service failed times satisfied with threshold func (s *serviceControllerImpl) PunishServiceWithThreshold(ctx context.Context, service, host string, punishTimeSec int) { - v, ok := s.allServices.Load(service + host) + s.punishWith(ctx, service, host, punishTimeSec, s.config.ServicePunishThreshold, s.config.ServicePunishValidIntervalS) +} + +func (s *serviceControllerImpl) punishWith(ctx context.Context, + primary, secondary string, punishTimeSec int, + threshold uint32, interval int, +) { + serviceKey := s.getServiceKey(primary, secondary) + v, ok := s.allServices.Load(serviceKey) if !ok { - panic(fmt.Sprintf("can't can host in all services map, %s-%s", service, host)) + panic(fmt.Sprintf("can't load host in all services map, %s", serviceKey)) } item := v.(*hostItem) new := atomic.AddUint32(&item.failedTimes, 1) // failedTimes larger than threshold, then check the lastModifyTime - if new >= s.config.ServicePunishThreshold { - if time.Since(time.Unix(atomic.LoadInt64(&item.lastModifyTime), 0)) < time.Duration(s.config.ServicePunishValidIntervalS)*time.Second { - s.PunishService(ctx, service, host, punishTimeSec) + if new >= threshold { + if time.Since(time.Unix(atomic.LoadInt64(&item.lastModifyTime), 0)) < time.Duration(interval)*time.Second { + s.PunishService(ctx, primary, secondary, punishTimeSec) return } atomic.AddUint32(&item.failedTimes, -(new - 1)) @@ -500,3 +509,7 @@ func (s *serviceControllerImpl) PunishServiceWithThreshold(ctx context.Context, func (s *serviceControllerImpl) getServiceLock(name string) *sync.RWMutex { return s.serviceLocks[name] } + +func (s *serviceControllerImpl) getServiceKey(primary string, secondary any) string { + return fmt.Sprintf("%s/%v", primary, secondary) +} diff --git a/blobstore/access/controller/service_test.go b/blobstore/access/controller/service_test.go index 7dedc55a2..a7edcef26 100644 --- a/blobstore/access/controller/service_test.go +++ b/blobstore/access/controller/service_test.go @@ -56,7 +56,7 @@ func TestAccessServiceNew(t *testing.T) { } { sc, err := controller.NewServiceController( - controller.ServiceConfig{IDC: idc + "x", ReloadSec: 1}, cmcli, proxycli, nil) + controller.ServiceConfig{IDC: idc + "x", ServiceReloadSecs: 1}, cmcli, proxycli, nil) require.NoError(t, err) _, err = sc.GetServiceHost(serviceCtx, serviceName) @@ -66,7 +66,7 @@ func TestAccessServiceNew(t *testing.T) { func TestAccessServiceGetServiceHost(t *testing.T) { sc, err := controller.NewServiceController( - controller.ServiceConfig{IDC: idc, ReloadSec: 1}, cmcli, proxycli, nil) + controller.ServiceConfig{IDC: idc, ServiceReloadSecs: 1}, cmcli, proxycli, nil) require.NoError(t, err) keys := make(hostSet) @@ -86,7 +86,7 @@ func TestAccessServicePunishService(t *testing.T) { stop := closer.New() defer stop.Close() sc, err := controller.NewServiceController( - controller.ServiceConfig{IDC: idc, ReloadSec: 1}, cmcli, proxycli, stop.Done()) + controller.ServiceConfig{IDC: idc, ServiceReloadSecs: 1}, cmcli, proxycli, stop.Done()) require.NoError(t, err) { @@ -137,7 +137,7 @@ func TestAccessServicePunishServiceWithThreshold(t *testing.T) { sc, err := controller.NewServiceController( controller.ServiceConfig{ IDC: idc, - ReloadSec: 1, + ServiceReloadSecs: 1, ServicePunishThreshold: threshold, ServicePunishValidIntervalS: 2, }, cmcli, proxycli, stop.Done()) @@ -189,7 +189,7 @@ func TestAccessServicePunishServiceWithThreshold(t *testing.T) { func TestAccessServiceGetDiskHost(t *testing.T) { sc, err := controller.NewServiceController( - controller.ServiceConfig{IDC: idc, ReloadSec: 1}, cmcli, proxycli, nil) + controller.ServiceConfig{IDC: idc, ServiceReloadSecs: 1}, cmcli, proxycli, nil) require.NoError(t, err) { @@ -232,7 +232,7 @@ func TestAccessServiceGetBrokenDiskHost(t *testing.T) { stop := closer.New() defer stop.Close() sc, err := controller.NewServiceController( - controller.ServiceConfig{IDC: idc, ReloadSec: 1, LoadDiskInterval: 1}, cli, pcli, stop.Done()) + controller.ServiceConfig{IDC: idc, ServiceReloadSecs: 1, LoadDiskIntervalS: 1}, cli, pcli, stop.Done()) require.NoError(t, err) { @@ -299,7 +299,7 @@ func TestAccessServicePunishDisk(t *testing.T) { stop := closer.New() defer stop.Close() sc, err := controller.NewServiceController( - controller.ServiceConfig{IDC: idc, ReloadSec: 1}, cmcli, proxycli, stop.Done()) + controller.ServiceConfig{IDC: idc, ServiceReloadSecs: 1}, cmcli, proxycli, stop.Done()) require.NoError(t, err) { diff --git a/blobstore/access/controller/volume.go b/blobstore/access/controller/volume.go index f35f828e3..944a7a987 100644 --- a/blobstore/access/controller/volume.go +++ b/blobstore/access/controller/volume.go @@ -30,15 +30,11 @@ import ( "github.com/cubefs/cubefs/blobstore/common/proto" "github.com/cubefs/cubefs/blobstore/common/rpc" "github.com/cubefs/cubefs/blobstore/common/trace" + "github.com/cubefs/cubefs/blobstore/util/defaulter" "github.com/cubefs/cubefs/blobstore/util/errors" "github.com/cubefs/cubefs/blobstore/util/retry" ) -const ( - _defaultCacheSize = 1 << 20 - _defaultCacheExpiration = int64(2 * time.Minute) -) - // Unit alias of clustermgr.Unit type Unit = clustermgr.Unit @@ -107,9 +103,6 @@ func (vc *volumeMemCache) Set(key cvid, value *VolumePhy) { } type volumeGetterImpl struct { - ctx context.Context - cid proto.ClusterID - volumeMemCache volumePhyCacher memExpiration int64 punishCache *memcache.MemCache @@ -120,33 +113,44 @@ type volumeGetterImpl struct { unusualLock sync.Mutex unusualVolume map[proto.Vid]int + + config VolumeConfig +} + +// VolumeConfig controller of volume's config +type VolumeConfig struct { + ClusterID proto.ClusterID `json:"-"` + + VolumeMemcacheSize int `json:"volume_memcache_size"` + VolumeMemcachePunishSize int `json:"volume_memcache_punish_size"` + VolumeMemcacheExpirationMs int64 `json:"volume_memcache_expiration_ms"` // -1 means no expiration + VolumePunishThreshold int `json:"volume_punish_threshold"` + VolumePunishIntervalS int `json:"volume_punish_interval_s"` } // NewVolumeGetter new a volume getter -// -// memExpiration expiration of memcache, 0 means no expiration -func NewVolumeGetter(clusterID proto.ClusterID, service ServiceController, - proxy proxy.Cacher, memExpiration time.Duration, stop <-chan struct{}, +func NewVolumeGetter(cfg VolumeConfig, service ServiceController, + proxy proxy.Cacher, stop <-chan struct{}, ) (VolumeGetter, error) { - _, ctx := trace.StartSpanFromContext(context.Background(), "") + expiration := cfg.VolumeMemcacheExpirationMs * int64(time.Millisecond) + defaulter.IntegerEqual(&expiration, int64(2*time.Minute)) + defaulter.IntegerLess(&expiration, 0) - expiration := int64(memExpiration) - if expiration < 0 { - expiration = _defaultCacheExpiration - } + defaulter.IntegerLessOrEqual(&cfg.VolumeMemcacheSize, 1<<20) + defaulter.IntegerLessOrEqual(&cfg.VolumeMemcachePunishSize, 1<<10) + defaulter.IntegerLessOrEqual(&cfg.VolumePunishThreshold, 10) + defaulter.IntegerLessOrEqual(&cfg.VolumePunishIntervalS, 600) - mc, err := memcache.NewMemCache(_defaultCacheSize) + mc, err := memcache.NewMemCache(cfg.VolumeMemcacheSize) if err != nil { return nil, err } - punishCache, err := memcache.NewMemCache(1024) + punishCache, err := memcache.NewMemCache(cfg.VolumeMemcachePunishSize) if err != nil { return nil, err } getter := &volumeGetterImpl{ - ctx: ctx, - cid: clusterID, volumeMemCache: &volumeMemCache{cache: mc}, memExpiration: expiration, punishCache: punishCache, @@ -154,10 +158,11 @@ func NewVolumeGetter(clusterID proto.ClusterID, service ServiceController, proxy: proxy, singleRun: new(singleflight.Group), unusualVolume: make(map[proto.Vid]int), + config: cfg, } go func() { - ticker := time.NewTicker(10 * time.Minute) + ticker := time.NewTicker(time.Duration(cfg.VolumePunishIntervalS) * time.Second) defer ticker.Stop() for { getter.tickerUpdate() @@ -179,8 +184,8 @@ func NewVolumeGetter(clusterID proto.ClusterID, service ServiceController, // 2.second level cache from proxy cluster func (v *volumeGetterImpl) Get(ctx context.Context, vid proto.Vid, isCache bool) (phy *VolumePhy) { span := trace.SpanFromContextSafe(ctx) - cid := v.cid.ToString() - id := addCVid(v.cid, vid) + cid := v.config.ClusterID.ToString() + id := addCVid(v.config.ClusterID, vid) // check if volume punish defer func() { @@ -259,11 +264,11 @@ func (v *volumeGetterImpl) Get(ctx context.Context, vid proto.Vid, isCache bool) return } -func (v *volumeGetterImpl) Punish(ctx context.Context, vid proto.Vid, punishIntervalS int) { - v.punishCache.Set(addCVid(v.cid, vid), time.Now().Add(time.Duration(punishIntervalS)*time.Second).Unix()) +func (v *volumeGetterImpl) Punish(_ context.Context, vid proto.Vid, punishIntervalS int) { + v.punishCache.Set(addCVid(v.config.ClusterID, vid), time.Now().Add(time.Duration(punishIntervalS)*time.Second).Unix()) } -func (v *volumeGetterImpl) Update(ctx context.Context, vid proto.Vid) { +func (v *volumeGetterImpl) Update(_ context.Context, vid proto.Vid) { v.unusualLock.Lock() v.unusualVolume[vid] += 1 v.unusualLock.Unlock() @@ -285,7 +290,8 @@ func (v *volumeGetterImpl) getFromProxy(ctx context.Context, vid proto.Vid, flus } var volume *proxy.VersionVolume - id := addCVid(v.cid, vid) + cid := v.config.ClusterID + id := addCVid(cid, vid) triedHosts := make(map[string]struct{}) if err = retry.ExponentialBackoff(3, 30).RuptOn(func() (bool, error) { for _, host := range hosts { @@ -308,14 +314,14 @@ func (v *volumeGetterImpl) getFromProxy(ctx context.Context, vid proto.Vid, flus Vid: vid, Timestamp: -time.Now().UnixNano(), } - span.Infof("to update memcache on not exist volume(%d-%d) %+v", v.cid, vid, phy) + span.Infof("to update memcache on not exist volume(%d-%d) %+v", cid, vid, phy) v.setToLocalCache(ctx, id, phy) } else if flush { - span.Warnf("to flush force on all proxy of volume(%d-%d)", v.cid, vid) + span.Warnf("to flush force on all proxy of volume(%d-%d)", cid, vid) v.setToLocalCache(ctx, id, nil) v.flush(ctx, vid, 0, hosts, map[string]struct{}{}) } - return nil, errors.Base(err, "get volume from proxy", v.cid, vid) + return nil, errors.Base(err, "get volume from proxy", cid, vid) } phy := &VolumePhy{ @@ -327,7 +333,7 @@ func (v *volumeGetterImpl) getFromProxy(ctx context.Context, vid proto.Vid, flus } copy(phy.Units, volume.Units[:]) - span.Debugf("to update memcache on volume(%d-%d) %+v", v.cid, vid, phy) + span.Debugf("to update memcache on volume(%d-%d) %+v", cid, vid, phy) v.setToLocalCache(ctx, id, phy) if flush { @@ -369,7 +375,7 @@ func (v *volumeGetterImpl) tickerUpdate() { var vids []proto.Vid v.unusualLock.Lock() for vid, n := range v.unusualVolume { - if n > 10 { + if n > v.config.VolumePunishThreshold { vids = append(vids, vid) if len(vids) >= 10 { break diff --git a/blobstore/access/controller/volume_test.go b/blobstore/access/controller/volume_test.go index 723e5ae32..0012544b2 100644 --- a/blobstore/access/controller/volume_test.go +++ b/blobstore/access/controller/volume_test.go @@ -40,7 +40,8 @@ func proxyService() controller.ServiceController { func TestAccessVolumeGetterNew(t *testing.T) { _, ctx := trace.StartSpanFromContext(context.Background(), "TestAccessVolumeGetterNew") - getter, err := controller.NewVolumeGetter(1, proxyService(), proxycli, time.Millisecond*200, closedCh()) + cfg := controller.VolumeConfig{ClusterID: 1, VolumeMemcacheExpirationMs: 200} + getter, err := controller.NewVolumeGetter(cfg, proxyService(), proxycli, closedCh()) require.Nil(t, err) require.Nil(t, getter.Get(ctx, proto.Vid(0), true)) @@ -68,7 +69,8 @@ func TestAccessVolumeGetterNew(t *testing.T) { func TestAccessVolumeGetterNotExistVolume(t *testing.T) { _, ctx := trace.StartSpanFromContext(context.Background(), "TestAccessVolumeGetterNotExistVolume") - getter, err := controller.NewVolumeGetter(0xfe, proxyService(), proxycli, time.Millisecond*200, closedCh()) + cfg := controller.VolumeConfig{ClusterID: 0xfe, VolumeMemcacheExpirationMs: 200} + getter, err := controller.NewVolumeGetter(cfg, proxyService(), proxycli, closedCh()) require.NoError(t, err) id := vid404 @@ -84,7 +86,8 @@ func TestAccessVolumeGetterNotExistVolume(t *testing.T) { getter.Get(ctx, id, true) require.Equal(t, 2, dataCalled[id]) - getter, err = controller.NewVolumeGetter(0xee, proxyService(), proxycli, 0, closedCh()) + cfg = controller.VolumeConfig{ClusterID: 0xee, VolumeMemcacheExpirationMs: -1} + getter, err = controller.NewVolumeGetter(cfg, proxyService(), proxycli, closedCh()) require.NoError(t, err) id = vid404 dataCalled[id] = 0 @@ -103,7 +106,8 @@ func TestAccessVolumeGetterNotExistVolume(t *testing.T) { func TestAccessVolumeGetterNotExistFlush(t *testing.T) { _, ctx := trace.StartSpanFromContext(context.Background(), "TestAccessVolumeGetterNotExistVolumeFlush") - getter, err := controller.NewVolumeGetter(0xfe, proxyService(), proxycli, time.Millisecond*200, closedCh()) + cfg := controller.VolumeConfig{ClusterID: 0xfe, VolumeMemcacheExpirationMs: 200} + getter, err := controller.NewVolumeGetter(cfg, proxyService(), proxycli, closedCh()) require.NoError(t, err) id := vid404 @@ -114,7 +118,8 @@ func TestAccessVolumeGetterNotExistFlush(t *testing.T) { time.Sleep(time.Millisecond * 210) - getter, err = controller.NewVolumeGetter(0xee, proxyService(), proxycli, 0, closedCh()) + cfg = controller.VolumeConfig{ClusterID: 0xee, VolumeMemcacheExpirationMs: -1} + getter, err = controller.NewVolumeGetter(cfg, proxyService(), proxycli, closedCh()) require.NoError(t, err) for range [10]struct{}{} { require.Nil(t, getter.Get(ctx, id, false)) @@ -124,7 +129,8 @@ func TestAccessVolumeGetterNotExistFlush(t *testing.T) { func TestAccessVolumeGetterExpiration(t *testing.T) { _, ctx := trace.StartSpanFromContext(context.Background(), "TestAccessVolumeGetterExpiration") - getter, err := controller.NewVolumeGetter(1, proxyService(), proxycli, time.Millisecond*200, closedCh()) + cfg := controller.VolumeConfig{ClusterID: 1, VolumeMemcacheExpirationMs: 200} + getter, err := controller.NewVolumeGetter(cfg, proxyService(), proxycli, closedCh()) require.Nil(t, err) id := proto.Vid(1) @@ -145,7 +151,8 @@ func TestAccessVolumeGetterExpiration(t *testing.T) { func TestAccessVolumePunish(t *testing.T) { _, ctx := trace.StartSpanFromContext(context.Background(), "TestAccessVolumePunish") - getter, err := controller.NewVolumeGetter(1, proxyService(), proxycli, 0, closedCh()) + cfg := controller.VolumeConfig{ClusterID: 1, VolumeMemcacheExpirationMs: -1} + getter, err := controller.NewVolumeGetter(cfg, proxyService(), proxycli, closedCh()) require.Nil(t, err) require.Nil(t, getter.Get(ctx, proto.Vid(0), true)) @@ -177,7 +184,8 @@ func TestAccessVolumeUpdate(t *testing.T) { time.Sleep(time.Second) close(ch) }() - getter, err := controller.NewVolumeGetter(1, proxyService(), proxycli, 0, ch) + cfg := controller.VolumeConfig{ClusterID: 1, VolumeMemcacheExpirationMs: -1} + getter, err := controller.NewVolumeGetter(cfg, proxyService(), proxycli, ch) require.NoError(t, err) getter.Update(ctx, 1) for range [11]struct{}{} { @@ -204,7 +212,8 @@ func TestAccessVolumeUpdateForce(t *testing.T) { time.Sleep(time.Second) close(ch) }() - getter, err := controller.NewVolumeGetter(1, proxyService(), proxycli, 0, ch) + cfg := controller.VolumeConfig{ClusterID: 1, VolumeMemcacheExpirationMs: -1} + getter, err := controller.NewVolumeGetter(cfg, proxyService(), proxycli, ch) require.NoError(t, err) id := proto.Vid(1) diff --git a/blobstore/access/stream/stream.go b/blobstore/access/stream/stream.go index 32ece7e1b..651abce7a 100644 --- a/blobstore/access/stream/stream.go +++ b/blobstore/access/stream/stream.go @@ -130,9 +130,11 @@ type StreamConfig struct { IDC string `json:"idc"` MaxBlobSize uint32 `json:"max_blob_size"` + VolumePunishIntervalS int `json:"volume_punish_interval_s"` DiskPunishIntervalS int `json:"disk_punish_interval_s"` DiskTimeoutPunishIntervalS int `json:"disk_timeout_punish_interval_s"` - ServicePunishIntervalS int `json:"service_punish_interval_s"` + ServicePunishIntervalS int `json:"service_punish_interval_s"` // just service of proxy + ShardnodePunishIntervalS int `json:"shardnode_punish_interval_s"` AllocRetryTimes int `json:"alloc_retry_times"` AllocRetryIntervalMS int `json:"alloc_retry_interval_ms"` EncoderEnableVerify bool `json:"encoder_enableverify"` @@ -219,9 +221,11 @@ func confCheck(cfg *StreamConfig) error { } defaulter.Equal(&cfg.MaxBlobSize, defaultMaxBlobSize) + defaulter.IntegerLessOrEqual(&cfg.VolumePunishIntervalS, 60) defaulter.LessOrEqual(&cfg.DiskPunishIntervalS, defaultDiskPunishIntervalS) defaulter.LessOrEqual(&cfg.DiskTimeoutPunishIntervalS, defaultDiskPunishIntervalS/10) defaulter.LessOrEqual(&cfg.ServicePunishIntervalS, defaultServicePunishIntervalS) + defaulter.IntegerLessOrEqual(&cfg.ShardnodePunishIntervalS, 60) defaulter.LessOrEqual(&cfg.AllocRetryTimes, defaultAllocRetryTimes) if cfg.AllocRetryIntervalMS <= 100 { cfg.AllocRetryIntervalMS = defaultAllocRetryIntervalMS @@ -468,12 +472,10 @@ func (h *Handler) getVolume(ctx context.Context, clusterID proto.ClusterID, vid if err != nil { return nil, err } - volume := volumeGetter.Get(ctx, vid, isCache) if volume == nil { return nil, errors.Newf("not found volume of (%d %d)", clusterID, vid) } - return volume, nil } @@ -488,7 +490,7 @@ func (h *Handler) updateVolume(ctx context.Context, clusterID proto.ClusterID, v func (h *Handler) punishVolume(ctx context.Context, clusterID proto.ClusterID, vid proto.Vid, host, reason string) { reportUnhealth(clusterID, "punish", "volume", host, reason) if volumeGetter, err := h.clusterController.GetVolumeGetter(clusterID); err == nil { - volumeGetter.Punish(ctx, vid, h.DiskPunishIntervalS) + volumeGetter.Punish(ctx, vid, h.VolumePunishIntervalS) } } @@ -506,6 +508,13 @@ func (h *Handler) punishDiskWith(ctx context.Context, clusterID proto.ClusterID, } } +func (h *Handler) punishShardnodeDisk(ctx context.Context, clusterID proto.ClusterID, diskID proto.DiskID, host, reason string) { + reportUnhealth(clusterID, "punish", "shardnode", host, reason) + if serviceController, err := h.clusterController.GetServiceController(clusterID); err == nil { + serviceController.PunishShardnode(ctx, diskID, h.ShardnodePunishIntervalS) + } +} + // blobCount blobSize > 0 is certain func blobCount(size uint64, blobSize uint32) uint64 { return (size + uint64(blobSize) - 1) / uint64(blobSize) diff --git a/blobstore/access/stream/stream_blob.go b/blobstore/access/stream/stream_blob.go index 82a092c41..100b0a59c 100644 --- a/blobstore/access/stream/stream_blob.go +++ b/blobstore/access/stream/stream_blob.go @@ -526,13 +526,6 @@ func (h *Handler) punishAndUpdate(ctx context.Context, args *punishArgs) bool { return true } -func (h *Handler) punishShardnodeDisk(ctx context.Context, clusterID proto.ClusterID, diskID proto.DiskID, host, reason string) { - reportUnhealth(clusterID, "punish", "shardnode", host, reason) - if serviceController, err := h.clusterController.GetServiceController(clusterID); err == nil { - serviceController.PunishShardnode(ctx, diskID, h.DiskPunishIntervalS) - } -} - func (h *Handler) updateShardRoute(ctx context.Context, clusterID proto.ClusterID) error { shardMgr, err := h.clusterController.GetShardController(clusterID) if err != nil { diff --git a/blobstore/access/stream/stream_mock_test.go b/blobstore/access/stream/stream_mock_test.go index d04e6e9df..67dc35468 100644 --- a/blobstore/access/stream/stream_mock_test.go +++ b/blobstore/access/stream/stream_mock_test.go @@ -409,11 +409,14 @@ func initMockData() { serviceController, _ = controller.NewServiceController( controller.ServiceConfig{ - ClusterID: clusterID, - IDC: idc, - ReloadSec: 1000, + ClusterID: clusterID, + IDC: idc, + ServiceReloadSecs: 1000, }, cmcli, proxycli, nil) - volumeGetter, _ = controller.NewVolumeGetter(clusterID, serviceController, proxycli, 0, nil) + volumeGetter, _ = controller.NewVolumeGetter(controller.VolumeConfig{ + ClusterID: clusterID, + VolumeMemcacheExpirationMs: -1, + }, serviceController, proxycli, nil) ctr = gomock.NewController(&testing.T{}) c := NewMockClusterController(ctr)