Dynamic modification of configuration files&&Slow log

codis/cmd/proxy/main.go

@@ -112,6 +112,7 @@ Options:
 		if err := config.LoadFromFile(s); err != nil {
 			log.PanicErrorf(err, "load config %s failed", s)
 		}
+		config.ConfigFileName = s
 	}
 	models.SetMaxSlotNum(config.MaxSlotNum)
 	if s, ok := utils.Argument(d, "--host-admin"); ok {

codis/config/proxy.toml

@@ -45,7 +45,7 @@ proxy_datacenter = ""
 proxy_max_clients = 1000
 
 # Set max offheap memory size. (0 to disable)
-proxy_max_offheap_size = "1024mb"
+proxy_max_offheap_size = "1gb"
 
 # Set heap placeholder to reduce GC frequency.
 proxy_heap_placeholder = "256mb"
@@ -115,3 +115,6 @@ metrics_report_statsd_server = ""
 metrics_report_statsd_period = "1s"
 metrics_report_statsd_prefix = ""
 
+ncpu = 0
+slowlog_log_slower_than = 0
+slowlog_max_len = 0

codis/pkg/proxy/config.go

@@ -133,6 +133,11 @@ metrics_report_statsd_prefix = ""
 `
 
 type Config struct {
+	ConfigFileName string `toml:"-" json:"config_file_name"`
+	Log            string `toml:"log"`
+	LogLevel       string `toml:"log_level"`
+	Ncpu           int    `toml:"ncpu"`
+
 	ProtoType string `toml:"proto_type" json:"proto_type"`
 	ProxyAddr string `toml:"proxy_addr" json:"proxy_addr"`
 	AdminAddr string `toml:"admin_addr" json:"admin_addr"`
@@ -176,6 +181,9 @@ type Config struct {
 	SessionKeepAlivePeriod timesize.Duration `toml:"session_keepalive_period" json:"session_keepalive_period"`
 	SessionBreakOnFailure  bool              `toml:"session_break_on_failure" json:"session_break_on_failure"`
 
+	//SlowlogLogSlowerThan int64 `toml:"slowlog_log_slower_than" json:"slowlog_log_slower_than"`
+	//SlowlogMaxLen        int64 `toml:"slowlog_max_len" json:"slowlog_max_len"`
+
 	MetricsReportServer           string            `toml:"metrics_report_server" json:"metrics_report_server"`
 	MetricsReportPeriod           timesize.Duration `toml:"metrics_report_period" json:"metrics_report_period"`
 	MetricsReportInfluxdbServer   string            `toml:"metrics_report_influxdb_server" json:"metrics_report_influxdb_server"`
@@ -302,6 +310,22 @@ func (c *Config) Validate() error {
 		return errors.New("invalid session_keepalive_period")
 	}
 
+	//if c.SlowlogLogSlowerThan < 0 {
+	//	return errors.New("invalid slowlog_log_slower_than")
+	//}
+	//if c.SlowlogMaxLen < 0 {
+	//	return errors.New("invalid slowlog_max_len")
+	//}
+	//if c.Log == "" {
+	//	return errors.New("invalid log")
+	//}
+	//if c.LogLevel == "" {
+	//	return errors.New("invalid log_level")
+	//}
+	//if c.Ncpu <= 0 {
+	//	return errors.New("invalid ncpu")
+	//}
+
 	if c.MetricsReportPeriod < 0 {
 		return errors.New("invalid metrics_report_period")
 	}
@@ -311,5 +335,6 @@ func (c *Config) Validate() error {
 	if c.MetricsReportStatsdPeriod < 0 {
 		return errors.New("invalid metrics_report_statsd_period")
 	}
+
 	return nil
 }

codis/pkg/proxy/mapper.go

@@ -6,6 +6,7 @@ package proxy
 import (
 	"bytes"
 	"hash/crc32"
+	"strconv"
 	"strings"
 
 	"pika/codis/v2/pkg/proxy/redis"
@@ -72,6 +73,7 @@ func init() {
 		{"BLPOP", FlagWrite | FlagNotAllow},
 		{"BRPOP", FlagWrite | FlagNotAllow},
 		{"BRPOPLPUSH", FlagWrite | FlagNotAllow},
+		{"CCONFIG", FlagWrite},
 		{"CLIENT", FlagNotAllow},
 		{"CLUSTER", FlagNotAllow},
 		{"COMMAND", 0},
@@ -214,7 +216,7 @@ func init() {
 		{"SLOTSRESTORE-ASYNC-AUTH", FlagWrite | FlagNotAllow},
 		{"SLOTSRESTORE-ASYNC-ACK", FlagWrite | FlagNotAllow},
 		{"SLOTSSCAN", FlagMasterOnly},
-		{"SLOWLOG", FlagNotAllow},
+		{"SLOWLOG", 0},
 		{"SMEMBERS", 0},
 		{"SMOVE", FlagWrite},
 		{"SORT", FlagWrite},
@@ -318,3 +320,30 @@ func getHashKey(multi []*redis.Resp, opstr string) []byte {
 	}
 	return nil
 }
+
+func getWholeCmd(multi []*redis.Resp, cmd []byte) int {
+	var index = 0
+	var bytes = 0
+
+	for i := 0; i < len(multi); i++ {
+		if index < len(cmd) {
+			index += copy(cmd[index:], multi[i].Value)
+			if i < len(multi)-i {
+				index += copy(cmd[index:], []byte(" "))
+			}
+		}
+		bytes += len(multi[i].Value)
+
+		//  如果cmd已经满了，那么最后腾出来一个more长度的位置添加more信息
+		if i == len(multi)-1 && index == len(cmd) {
+			more := []byte("... " + strconv.Itoa(len(multi)) + " elements " + strconv.Itoa(bytes) + " bytes.")
+			index = len(cmd) - len(more)
+			if index < 0 {
+				index = 0
+			}
+			index += copy(cmd[index:], more)
+			break
+		}
+	}
+	return index
+}

codis/pkg/proxy/proxy.go

@@ -10,7 +10,9 @@ import (
 	"os"
 	"os/exec"
 	"path/filepath"
+	"pika/codis/v2/pkg/proxy/redis"
 	"runtime"
+	"strconv"
 	"strings"
 	"sync"
 	"time"
@@ -205,9 +207,50 @@ func (s *Proxy) Model() *models.Proxy {
 }
 
 func (s *Proxy) Config() *Config {
+	s.mu.Lock()
+	defer s.mu.Unlock()
 	return s.config
 }
 
+func (s *Proxy) ConfigGet(key string) *redis.Resp {
+	s.mu.Lock()
+	defer s.mu.Unlock()
+	switch key {
+	case "proxy_max_clients":
+		return redis.NewBulkBytes([]byte(strconv.Itoa(s.config.ProxyMaxClients)))
+	default:
+		return redis.NewErrorf("unsurport key[%s].", key)
+	}
+}
+
+func (s *Proxy) ConfigSet(key, value string) *redis.Resp {
+	s.mu.Lock()
+	defer s.mu.Unlock()
+	switch key {
+	case "proxy_max_clients":
+		n, err := strconv.Atoi(value)
+		if err != nil {
+			return redis.NewErrorf("err：%s.", err)
+		}
+
+		if n <= 0 {
+			return redis.NewErrorf("invalid proxy_max_clients")
+		} else {
+			s.config.ProxyMaxClients = n
+			return redis.NewString([]byte("OK"))
+		}
+	default:
+		return redis.NewErrorf("unsurport key.")
+	}
+}
+
+func (s *Proxy) ConfigRewrite() *redis.Resp {
+	s.mu.Lock()
+	defer s.mu.Unlock()
+	utils.RewriteConfig(*(s.config), s.config.ConfigFileName, "=", true)
+	return redis.NewString([]byte("OK"))
+}
+
 func (s *Proxy) IsOnline() bool {
 	s.mu.Lock()
 	defer s.mu.Unlock()
@@ -320,7 +363,7 @@ func (s *Proxy) serveProxy() {
 			if err != nil {
 				return err
 			}
-			NewSession(c, s.config).Start(s.router)
+			NewSession(c, s.config, s).Start(s.router)
 		}
 	}(s.lproxy)