mirror of
https://github.com/cubefs/cubefs.git
synced 2026-08-02 02:00:56 +00:00
feat(access): punish volume disk and service with own value
. #22762717 Signed-off-by: slasher <shenjie1@oppo.com>
This commit is contained in:
slasher
parent
bed91e7765
commit
9dad1ee7cf
@ -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)
|
||||
|
||||
@ -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)
|
||||
}
|
||||
|
||||
@ -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)
|
||||
|
||||
{
|
||||
|
||||
@ -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
|
||||
|
||||
@ -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)
|
||||
|
||||
@ -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)
|
||||
|
||||
@ -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 {
|
||||
|
||||
@ -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)
|
||||
|
||||
Loading…
Reference in New Issue
Block a user
