fix(access): report punished service with threshold

@formatter:off

Signed-off-by: slasher <shenjie1@oppo.com>

blobstore/access/controller/volume.go

@@ -269,11 +269,11 @@ func (v *volumeGetterImpl) Update(ctx context.Context, vid proto.Vid) {
 	v.unusualLock.Unlock()
 }
 
-func (v *volumeGetterImpl) setToLocalCache(ctx context.Context, id cvid, phy *VolumePhy) {
+func (v *volumeGetterImpl) setToLocalCache(_ context.Context, id cvid, phy *VolumePhy) {
 	v.volumeMemCache.Set(id, phy)
 }
 
-func (v *volumeGetterImpl) getFromLocalCache(ctx context.Context, id cvid) *VolumePhy {
+func (v *volumeGetterImpl) getFromLocalCache(_ context.Context, id cvid) *VolumePhy {
 	return v.volumeMemCache.Get(id)
 }
 
@@ -292,7 +292,7 @@ func (v *volumeGetterImpl) getFromProxy(ctx context.Context, vid proto.Vid, flus
 			triedHosts[host] = struct{}{}
 			if volume, err = v.proxy.GetCacheVolume(ctx, host,
 				&proxy.CacheVolumeArgs{Vid: vid, Flush: flush, Version: ver}); err != nil {
-				if rpc.DetectStatusCode(err) == errcode.CodeVolumeNotExist {
+				if err == context.Canceled || rpc.DetectStatusCode(err) == errcode.CodeVolumeNotExist {
 					return true, err
 				}
 				span.Warnf("get from proxy(%s) volume(%d) error(%s)", host, vid, err.Error())

blobstore/access/stream/stream.go

@@ -398,6 +398,7 @@ func (h *Handler) sendRepairMsg(ctx context.Context, blob blobIdent, badIdxes []
 	if err := retry.Timed(3, 200).On(func() error {
 		host, err := serviceController.GetServiceHost(ctx, serviceProxy)
 		if err != nil {
+			reportUnhealth(clusterID, "repair.msg", serviceProxy, "-", "failed")
 			span.Warn(err)
 			return err
 		}
@@ -405,14 +406,15 @@ func (h *Handler) sendRepairMsg(ctx context.Context, blob blobIdent, badIdxes []
 		if err != nil {
 			if errorTimeout(err) || errorConnectionRefused(err) {
 				serviceController.PunishServiceWithThreshold(ctx, serviceProxy, host, h.ServicePunishIntervalS)
+				reportUnhealth(clusterID, "punish", serviceProxy, host, "failed")
+			} else {
+				reportUnhealth(clusterID, "repair.msg", serviceProxy, host, "failed")
 			}
 			span.Warnf("send to %s repair message(%+v) %s", host, repairArgs, err.Error())
-			reportUnhealth(clusterID, "punish", serviceProxy, host, "failed")
 			err = errors.Base(err, host)
 		}
 		return err
 	}); err != nil {
-		reportUnhealth(clusterID, "repair.msg", serviceProxy, "-", "failed")
 		span.Errorf("send repair message(%+v) failed %s", repairArgs, errors.Detail(err))
 		return
 	}
@@ -448,6 +450,7 @@ func (h *Handler) clearGarbage(ctx context.Context, location *proto.Location) er
 	if err := retry.Timed(3, 200).On(func() error {
 		host, err := serviceController.GetServiceHost(ctx, serviceProxy)
 		if err != nil {
+			reportUnhealth(location.ClusterID, "delete.msg", serviceProxy, "-", "failed")
 			span.Warn(err)
 			return err
 		}
@@ -455,14 +458,15 @@ func (h *Handler) clearGarbage(ctx context.Context, location *proto.Location) er
 		if err != nil {
 			if errorTimeout(err) || errorConnectionRefused(err) {
 				serviceController.PunishServiceWithThreshold(ctx, serviceProxy, host, h.ServicePunishIntervalS)
+				reportUnhealth(location.ClusterID, "punish", serviceProxy, host, "failed")
+			} else {
+				reportUnhealth(location.ClusterID, "delete.msg", serviceProxy, host, "failed")
 			}
 			span.Warnf("send to %s delete message(%+v) %s", host, logMsg, err.Error())
-			reportUnhealth(location.ClusterID, "punish", serviceProxy, host, "failed")
 			err = errors.Base(err, host)
 		}
 		return err
 	}); err != nil {
-		reportUnhealth(location.ClusterID, "delete.msg", serviceProxy, "-", "failed")
 		span.Errorf("send delete message(%+v) failed %s", logMsg, errors.Detail(err))
 		return errors.Base(err, "send delete message:", logMsg)
 	}

blobstore/api/access/client.go

@@ -88,8 +88,6 @@ func (mode RPCConnectMode) getConfig(speed float64, timeout, baseTimeout int64)
 		Tc: rpc.TransportConfig{
 			// dial timeout
 			DialTimeoutMs: 200,
-			// response header timeout after send the request
-			ResponseHeaderTimeoutMs: 5 * 1000,
 			// IdleConnTimeout is the maximum amount of time an idle
 			// (keep-alive) connection will remain idle before closing
 			// itself.Zero means no limit.
@@ -107,29 +105,21 @@ func (mode RPCConnectMode) getConfig(speed float64, timeout, baseTimeout int64)
 		config.ClientTimeoutMs = getTimeout(getSpeed(40))
 		config.BodyBandwidthMBPs = getSpeed(40)
 		config.BodyBaseTimeoutMs = getBaseTimeout(3 * 1000)
-		config.Tc.DialTimeoutMs = 100
-		config.Tc.ResponseHeaderTimeoutMs = 2 * 1000
 		config.Tc.IdleConnTimeoutMs = 10 * 1000
 	case GeneralConnMode:
 		config.ClientTimeoutMs = getTimeout(getSpeed(20))
 		config.BodyBandwidthMBPs = getSpeed(20)
 		config.BodyBaseTimeoutMs = getBaseTimeout(10 * 1000)
-		config.Tc.DialTimeoutMs = 100
-		config.Tc.ResponseHeaderTimeoutMs = 3 * 1000
 		config.Tc.IdleConnTimeoutMs = 30 * 1000
 	case SlowConnMode:
 		config.ClientTimeoutMs = getTimeout(getSpeed(4))
 		config.BodyBandwidthMBPs = getSpeed(4)
 		config.BodyBaseTimeoutMs = getBaseTimeout(120 * 1000)
-		config.Tc.DialTimeoutMs = 200
-		config.Tc.ResponseHeaderTimeoutMs = 10 * 1000
 		config.Tc.IdleConnTimeoutMs = 60 * 1000
 	case NoLimitConnMode:
 		config.ClientTimeoutMs = 0
 		config.BodyBandwidthMBPs = getSpeed(0)
 		config.BodyBaseTimeoutMs = getBaseTimeout(0)
-		config.Tc.DialTimeoutMs = 0
-		config.Tc.ResponseHeaderTimeoutMs = 0
 		config.Tc.IdleConnTimeoutMs = 600 * 1000
 	default:
 	}
@@ -140,53 +130,53 @@ func (mode RPCConnectMode) getConfig(speed float64, timeout, baseTimeout int64)
 // Config access client config
 type Config struct {
 	// ConnMode rpc connection timeout setting
-	ConnMode RPCConnectMode
+	ConnMode RPCConnectMode `json:"connection_mode"`
 	// ClientTimeoutMs the whole request and response timeout
-	ClientTimeoutMs int64
+	ClientTimeoutMs int64 `json:"client_timeout_ms"`
 	// BodyBandwidthMBPs reading body timeout, request or response
 	//   timeout = ContentLength/BodyBandwidthMBPs + BodyBaseTimeoutMs
-	BodyBandwidthMBPs float64
+	BodyBandwidthMBPs float64 `json:"body_bandwidth_mbps"`
 	// BodyBaseTimeoutMs base timeout for read body
-	BodyBaseTimeoutMs int64
+	BodyBaseTimeoutMs int64 `json:"body_base_timeout_ms"`
 
 	// Consul is consul config for discovering service
-	Consul ConsulConfig
+	Consul ConsulConfig `json:"consul"`
 	// ServiceIntervalS is interval seconds for discovering service hosts,
 	//   at least 5 seconds and default is 5 minutes.
-	ServiceIntervalS int
+	ServiceIntervalS int `json:"service_interval_s"`
 	// PriorityAddrs priority addrs of access service when retry
-	PriorityAddrs []string
+	PriorityAddrs []string `json:"priority_addrs"`
 	// MaxSizePutOnce max size using once-put object interface, default is 256MB.
-	MaxSizePutOnce int64
+	MaxSizePutOnce int64 `json:"max_size_put_once"`
 	// MaxPartRetry max retry times when putting one part, 0 means forever
-	MaxPartRetry int
+	MaxPartRetry int `json:"max_part_retry"`
 	// MaxHostRetry max retry hosts of access, default all hosts.
-	MaxHostRetry int
+	MaxHostRetry int `json:"max_host_retry"`
 	// PartConcurrence concurrence of put parts
-	PartConcurrence int
+	PartConcurrence int `json:"part_concurrence"`
 
 	// rpc selector config
 	// Failure retry interval, default value is 300s,
 	// if FailRetryIntervalS < 0, remove failed hosts will not work.
-	FailRetryIntervalS int
+	FailRetryIntervalS int `json:"fail_retry_interval_s"`
 	// Within MaxFailsPeriodS, if the number of failures is greater than or equal to MaxFails,
 	// the host is considered disconnected.
-	MaxFailsPeriodS int
+	MaxFailsPeriodS int `json:"max_fails_period_s"`
 	// HostTryTimes Number of host failure retries
-	HostTryTimes int
+	HostTryTimes int `json:"host_try_times"`
 
 	// RPCConfig user-defined rpc config
 	// All connections will use the config if it's not nil
 	// ConnMode will be ignored if rpc config is setting
-	RPCConfig *rpc.Config
+	RPCConfig *rpc.Config `json:"rpc_config"`
 
 	// LogLevel client output logging level.
-	LogLevel log.Level
+	LogLevel log.Level `json:"log_level"`
 
 	// Logger trace all logging to the logger if setting.
 	// It is an io.WriteCloser that writes to the specified filename.
 	// YOU should CLOSE it after you do not use the client anymore.
-	Logger *Logger
+	Logger *Logger `json:"logger"`
 }
 
 // ConsulConfig alias of consul api.Config

blobstore/api/access/client_test.go

@@ -723,17 +723,16 @@ func TestAccessClientPutTimeout(t *testing.T) {
 		maxMs time.Duration
 	}
 	cases := []caseT{
-		// QuickConnMode 3s + size / 40, dial and response 2s
+		// QuickConnMode 3s + size / 40
 		{access.QuickConnMode, mb * -119, ms * 0, ms * 600},
 		{access.QuickConnMode, mb * -100, ms * 500, ms * 1000},
 		{access.QuickConnMode, mb * -80, ms * 1000, ms * 1500},
-		{access.QuickConnMode, mb * -1, ms * 2000, ms * 2500},
+		{access.QuickConnMode, mb * -1, ms * 2500, ms * 3500},
 
-		// DefaultConnMode 30s + size / 10, dial and response 5s
+		// DefaultConnMode 30s + size / 10
 		{access.DefaultConnMode, mb * -299, ms * 0, ms * 600},
 		{access.DefaultConnMode, mb * -280, ms * 2000, ms * 2500},
 		{access.DefaultConnMode, mb * -270, ms * 3000, ms * 3500},
-		{access.DefaultConnMode, mb * -1, ms * 5000, ms * 5500},
 	}
 
 	var wg sync.WaitGroup