Dynamically adjust refresh time.

codis/cmd/proxy/main.go

@@ -193,6 +193,8 @@ Options:
 	}
 	defer s.Close()
 
+	proxy.RefreshPeriod.Set(config.MaxDelayRefreshTimeInterval.Int64())
+
 	log.Warnf("create proxy with config\n%s", config)
 
 	if s, ok := utils.Argument(d, "--pidfile"); ok {

codis/config/proxy.toml

@@ -118,3 +118,5 @@ metrics_report_statsd_server = ""
 metrics_report_statsd_period = "1s"
 metrics_report_statsd_prefix = ""
 
+# Maximum delay statistical time interval.(This value must be greater than 0.)
+max_delay_refresh_time_interval = "15s"

codis/pkg/proxy/config.go

@@ -133,6 +133,9 @@ metrics_report_influxdb_database = ""
 metrics_report_statsd_server = ""
 metrics_report_statsd_period = "1s"
 metrics_report_statsd_prefix = ""
+
+# Maximum delay statistical time interval.(This value must be greater than 0.)
+max_delay_refresh_time_interval = "15s"
 `
 
 type Config struct {
@@ -192,6 +195,8 @@ type Config struct {
 	MetricsReportStatsdPeriod     timesize.Duration `toml:"metrics_report_statsd_period" json:"metrics_report_statsd_period"`
 	MetricsReportStatsdPrefix     string            `toml:"metrics_report_statsd_prefix" json:"metrics_report_statsd_prefix"`
 
+	MaxDelayRefreshTimeInterval timesize.Duration `toml:"max_delay_refresh_time_interval" json:"max_delay_refresh_time_interval"`
+
 	ConfigFileName string `toml:"-" json:"config_file_name"`
 }
 
@@ -323,5 +328,9 @@ func (c *Config) Validate() error {
 		return errors.New("invalid metrics_report_statsd_period")
 	}
 
+	if c.MaxDelayRefreshTimeInterval <= 0 {
+		return errors.New("max_delay_refresh_time_interval must be greater than 0")
+	}
+
 	return nil
 }

codis/pkg/proxy/proxy.go

@@ -308,6 +308,12 @@ func (p *Proxy) ConfigGet(key string) *redis.Resp {
 			redis.NewBulkBytes([]byte("metrics_report_statsd_prefix")),
 			redis.NewBulkBytes([]byte(p.config.MetricsReportStatsdPrefix)),
 		})
+	case "max_delay_refresh_time_interval":
+		if text, err := p.config.MaxDelayRefreshTimeInterval.MarshalText(); err != nil {
+			return redis.NewErrorf("cant get max_delay_refresh_time_interval value.")
+		} else {
+			return redis.NewBulkBytes(text)
+		}
 	default:
 		return redis.NewErrorf("unsupported key: %s", key)
 	}
@@ -342,6 +348,18 @@ func (p *Proxy) ConfigSet(key, value string) *redis.Resp {
 		}
 		p.config.SlowlogLogSlowerThan = n
 		return redis.NewString([]byte("OK"))
+	case "max_delay_refresh_time_interval":
+		s := &(p.config.MaxDelayRefreshTimeInterval)
+		err := s.UnmarshalText([]byte(value))
+		if err != nil {
+			return redis.NewErrorf("err：%s.", err)
+		}
+		if d := p.config.MaxDelayRefreshTimeInterval.Duration(); d <= 0 {
+			return redis.NewErrorf("max_delay_refresh_time_interval must be greater than 0")
+		} else {
+			RefreshPeriod.Set(int64(d))
+			return redis.NewString([]byte("OK"))
+		}
 	default:
 		return redis.NewErrorf("unsupported key: %s", key)
 	}

codis/pkg/proxy/stats.go

@@ -14,7 +14,10 @@ import (
 	"pika/codis/v2/pkg/utils/sync2/atomic2"
 )
 
-var SlowCmdCount atomic2.Int64 // Cumulative count of slow log
+var (
+	SlowCmdCount  atomic2.Int64 // Cumulative count of slow log
+	RefreshPeriod atomic2.Int64
+)
 
 type opStats struct {
 	opstr string
@@ -79,10 +82,10 @@ func init() {
 		}
 	}()
 
-	// Clear the accumulated maximum delay to 0 every 15 seconds.
+	// Clear the accumulated maximum delay to 0
 	go func() {
 		for {
-			time.Sleep(15 * time.Second)
+			time.Sleep(time.Duration(RefreshPeriod.Int64()))
 			for _, s := range cmdstats.opmap {
 				s.maxDelay.Set(0)
 			}

tools/pika_exporter/discovery/discovery.go

@@ -206,8 +206,11 @@ func (d *codisDiscovery) CheckUpdate(updatechan chan int, codisaddr string) {
 }
 
 func (d *codisDiscovery) comparedis(new_instance *codisDiscovery) bool {
-	var addrs, addrsProxy []string
-	var diff bool = false
+	var (
+		addrs      []string
+		addrsProxy []string
+		diff       bool
+	)
 	for _, instance := range new_instance.instances {
 		addrs = append(addrs, instance.Addr)
 	}

tools/pika_exporter/exporter/future.go

@@ -3,7 +3,8 @@ package exporter
 import "sync"
 
 type futureKey struct {
-	addr, alias string
+	addr  string
+	alias string
 }
 
 type future struct {
@@ -38,7 +39,9 @@ func (f *future) Wait() map[futureKey]error {
 }
 
 type futureKeyForProxy struct {
-	addr, ID, productName string
+	addr        string
+	ID          string
+	productName string
 }
 
 type futureForProxy struct {
@@ -60,9 +63,11 @@ func (f *futureForProxy) Add() {
 
 func (f *futureForProxy) Done(key futureKeyForProxy, val error) {
 	f.Lock()
-	defer f.Unlock()
+	defer func() {
+		f.Unlock()
+		f.wait.Done()
+	}()
 	f.m[key] = val
-	f.wait.Done()
 }
 
 func (f *futureForProxy) Wait() map[futureKeyForProxy]error {

tools/pika_exporter/exporter/metrics/parser.go

@@ -348,14 +348,10 @@ func StructToMap(obj interface{}) (map[string]string, map[string][]int64, error)
 			for j := 0; j < field.Len(); j++ {
 				elemType := field.Index(j).Type()
 				elemValue := field.Index(j)
-				if elemType.Kind() == reflect.Struct {
-					var key string
-					for p := 0; p < elemValue.NumField(); p++ {
-						if p == 0 {
-							key = elemValue.Field(p).String()
-						} else {
-							cmdResult[key] = append(cmdResult[key], elemValue.Field(p).Int())
-						}
+				if elemType.Kind() == reflect.Struct && elemValue.NumField() > 0 {
+					var key string = elemValue.Field(0).String()
+					for p := 1; p < elemValue.NumField(); p++ {
+						cmdResult[key] = append(cmdResult[key], elemValue.Field(p).Int())
 					}
 				} else {
 					result[strings.ToLower(jsonName)] = fmt.Sprintf("%v", value)