Merge branch 'unstable' into fixAcl

Author: luky116

.github/workflows/pika.yml

@@ -26,25 +26,20 @@ jobs:
         with:
           go-version: 1.19
 
+      - name: ccache
+        uses: hendrikmuhs/ccache-action@v1.2.13
+        with:
+          key: ubuntu-latest
+
       - name: Install Deps
-        if: ${{ steps.cache.output.cache-hit != 'true' }}
         run: |
           sudo apt-get install -y autoconf libprotobuf-dev protobuf-compiler
           sudo apt-get install -y clang-tidy-12
 
       - name: Configure CMake
         # Configure CMake in a 'build' subdirectory. `CMAKE_BUILD_TYPE` is only required if you are using a single-configuration generator such as make.
         # See https://cmake.org/cmake/help/latest/variable/CMAKE_BUILD_TYPE.html?highlight=cmake_build_type
-        run: cmake -B build -DCMAKE_BUILD_TYPE=${{ env.BUILD_TYPE }} -DUSE_PIKA_TOOLS=ON -DCMAKE_CXX_FLAGS_DEBUG=-fsanitize=address
-
-      - name: Cache Build
-        uses: actions/cache@v3
-        id: cache-ubuntu
-        with:
-          key: ${{ runner.os }}-build-ubuntu-${{ hashFiles('**/CMakeLists.txt') }}
-          path: |
-            ${{ github.workspace }}/buildtrees
-            ${{ github.workspace }}/deps
+        run: cmake -B build -DCMAKE_BUILD_TYPE=${{ env.BUILD_TYPE }} -DUSE_PIKA_TOOLS=ON -DCMAKE_CXX_FLAGS_DEBUG=-fsanitize=address -D CMAKE_C_COMPILER_LAUNCHER=ccache -D CMAKE_CXX_COMPILER_LAUNCHER=ccache
 
       - name: Build
         # Build your program with the given configuration
@@ -166,25 +161,21 @@ jobs:
         with:
           go-version: 1.19
 
+      - name: ccache
+        uses: hendrikmuhs/ccache-action@v1.2.13
+        with:
+          key: macos-12
+
       - name: Install Deps
         run: |
           brew update
           brew install --overwrite python@3.12 autoconf protobuf llvm wget git
           brew install gcc@10 automake cmake make binutils
+
       - name: Configure CMake
         run: |
           export CC=/usr/local/opt/gcc@10/bin/gcc-10 
-          cmake -B build -DCMAKE_C_COMPILER=/usr/local/opt/gcc@10/bin/gcc-10 -DUSE_PIKA_TOOLS=ON -DCMAKE_BUILD_TYPE=${{ env.BUILD_TYPE }} -DCMAKE_CXX_FLAGS_DEBUG=-fsanitize=address
-
-      - uses: actions/cache@v3
-        with:
-          path: ${{ github.workspace }}/deps
-          key: ${{ runner.os }}-deps-${{ hashFiles('**/CMakeLists.txt') }}
-
-      - uses: actions/cache@v3
-        with:
-          path: ${{ github.workspace }}/buildtrees
-          key: ${{ runner.os }}-buildtrees-${{ hashFiles('**/CMakeLists.txt') }}
+          cmake -B build -DCMAKE_C_COMPILER=/usr/local/opt/gcc@10/bin/gcc-10 -DUSE_PIKA_TOOLS=ON -DCMAKE_BUILD_TYPE=${{ env.BUILD_TYPE }} -DCMAKE_CXX_FLAGS_DEBUG=-fsanitize=address -D CMAKE_C_COMPILER_LAUNCHER=ccache -D CMAKE_CXX_COMPILER_LAUNCHER=ccache
 
       - name: Build
         run: |

codis/pkg/proxy/config.go

@@ -194,8 +194,9 @@ type Config struct {
 
 	SlowlogLogSlowerThan int64 `toml:"slowlog_log_slower_than" json:"slowlog_log_slower_than"`
 
-	QuickCmdList string `toml:"quick_cmd_list" json:"quick_cmd_list"`
-	SlowCmdList  string `toml:"slow_cmd_list" json:"slow_cmd_list"`
+	QuickCmdList    string `toml:"quick_cmd_list" json:"quick_cmd_list"`
+	SlowCmdList     string `toml:"slow_cmd_list" json:"slow_cmd_list"`
+	AutoSetSlowFlag bool   `toml:"auto_set_slow_flag" json:"auto_set_slow_flag"`
 
 	MetricsReportServer           string            `toml:"metrics_report_server" json:"metrics_report_server"`
 	MetricsReportPeriod           timesize.Duration `toml:"metrics_report_period" json:"metrics_report_period"`

codis/pkg/proxy/proxy.go

@@ -555,6 +555,8 @@ func (p *Proxy) serveProxy() {
 		log.PanicErrorf(err, "setSlowCmdList [%s] failed", p.config.SlowCmdList)
 	}
 
+	StatsSetLogSlowerThan(p.config.SlowlogLogSlowerThan)
+
 	select {
 	case <-p.exit.C:
 		log.Warnf("[%p] proxy shutdown", p)
@@ -603,6 +605,16 @@ type Overview struct {
 	Slots   []*models.Slot `json:"slots,omitempty"`
 }
 
+type CmdInfo struct {
+	Total int64 `json:"total"`
+	Fails int64 `json:"fails"`
+	Redis struct {
+		Errors int64 `json:"errors"`
+	} `json:"redis"`
+	QPS int64      `json:"qps"`
+	Cmd []*OpStats `json:"cmd,omitempty"`
+}
+
 type Stats struct {
 	Online bool `json:"online"`
 	Closed bool `json:"closed"`
@@ -709,7 +721,7 @@ func (p *Proxy) Stats(flags StatsFlags) *Stats {
 	stats.Ops.Fails = OpFails()
 	stats.Ops.Redis.Errors = OpRedisErrors()
 	stats.Ops.QPS = OpQPS()
-
+	stats.Ops.Cmd = GetOpStatsByInterval(1)
 	if flags.HasBit(StatsCmds) {
 		stats.Ops.Cmd = GetOpStatsAll()
 	}
@@ -752,3 +764,21 @@ func (p *Proxy) Stats(flags StatsFlags) *Stats {
 	stats.SlowCmdCount = SlowCmdCount.Int64()
 	return stats
 }
+
+func (s *Proxy) CmdInfo(interval int64) *CmdInfo {
+	info := &CmdInfo{
+		Total: OpTotal(),
+		Fails: OpFails(),
+		QPS:   OpQPS(),
+		Cmd:   GetOpStatsByInterval(interval),
+	}
+	info.Redis.Errors = OpRedisErrors()
+	return info
+}
+
+func StatsSetLogSlowerThan(ms int64) {
+	if ms < 0 {
+		return
+	}
+	cmdstats.logSlowerThan.Set(ms)
+}

codis/pkg/proxy/proxy_api.go

@@ -65,12 +65,14 @@ func newApiServer(p *Proxy) http.Handler {
 		r.Get("/model", api.Model)
 		r.Get("/stats", api.StatsNoXAuth)
 		r.Get("/slots", api.SlotsNoXAuth)
+		r.Get("/cmdinfo/:interval", api.CmdInfoNoXAuth)
 	})
 	r.Group("/api/proxy", func(r martini.Router) {
 		r.Get("/model", api.Model)
 		r.Get("/xping/:xauth", api.XPing)
 		r.Get("/stats/:xauth", api.Stats)
 		r.Get("/stats/:xauth/:flags", api.Stats)
+		r.Get("/cmdinfo/:xauth/:interval", api.CmdInfo)
 		r.Get("/slots/:xauth", api.Slots)
 		r.Put("/start/:xauth", api.Start)
 		r.Put("/stats/reset/:xauth", api.ResetStats)
@@ -115,6 +117,10 @@ func (s *apiServer) SlotsNoXAuth() (int, string) {
 	return rpc.ApiResponseJson(s.proxy.Slots())
 }
 
+func (s *apiServer) CmdInfoNoXAuth() (int, string) {
+	return rpc.ApiResponseJson(s.proxy.CmdInfo(2))
+}
+
 func (s *apiServer) XPing(params martini.Params) (int, string) {
 	if err := s.verifyXAuth(params); err != nil {
 		return rpc.ApiResponseError(err)
@@ -123,6 +129,21 @@ func (s *apiServer) XPing(params martini.Params) (int, string) {
 	}
 }
 
+func (s *apiServer) CmdInfo(params martini.Params) (int, string) {
+	if err := s.verifyXAuth(params); err != nil {
+		return rpc.ApiResponseError(err)
+	}
+	var interval int64
+	if i := params["interval"]; i != "" {
+		n, err := strconv.Atoi(i)
+		if err != nil {
+			return rpc.ApiResponseError(err)
+		}
+		interval = int64(n)
+	}
+	return rpc.ApiResponseJson(s.proxy.CmdInfo(interval))
+}
+
 func (s *apiServer) Stats(params martini.Params) (int, string) {
 	if err := s.verifyXAuth(params); err != nil {
 		return rpc.ApiResponseError(err)
@@ -271,6 +292,15 @@ func (c *ApiClient) Stats(flags StatsFlags) (*Stats, error) {
 	return stats, nil
 }
 
+func (c *ApiClient) CmdInfo(interval int64) (*CmdInfo, error) {
+	url := c.encodeURL("/api/proxy/cmdinfo/%s/%d", c.xauth, interval)
+	cmdInfo := &CmdInfo{}
+	if err := rpc.ApiGetJson(url, cmdInfo); err != nil {
+		return nil, err
+	}
+	return cmdInfo, nil
+}
+
 func (c *ApiClient) Slots() ([]*models.Slot, error) {
 	url := c.encodeURL("/api/proxy/slots/%s", c.xauth)
 	slots := []*models.Slot{}

codis/pkg/proxy/session.go

@@ -6,15 +6,16 @@ package proxy
 import (
 	"encoding/json"
 	"fmt"
+	"math/rand"
 	"net"
+	"pika/codis/v2/pkg/utils"
 	"strconv"
 	"strings"
 	"sync"
 	"time"
 
 	"pika/codis/v2/pkg/models"
 	"pika/codis/v2/pkg/proxy/redis"
-	"pika/codis/v2/pkg/utils"
 	"pika/codis/v2/pkg/utils/errors"
 	"pika/codis/v2/pkg/utils/log"
 	"pika/codis/v2/pkg/utils/sync2/atomic2"
@@ -48,6 +49,8 @@ type Session struct {
 	config *Config
 	proxy  *Proxy
 
+	rand *rand.Rand
+
 	authorized bool
 }
 
@@ -79,6 +82,7 @@ func NewSession(sock net.Conn, config *Config, proxy *Proxy) *Session {
 		CreateUnix: time.Now().Unix(),
 	}
 	s.stats.opmap = make(map[string]*opStats, 16)
+	s.rand = rand.New(rand.NewSource(time.Now().UnixNano()))
 	log.Infof("session [%p] create: %s", s, s)
 	return s
 }
@@ -237,31 +241,35 @@ func (s *Session) loopWriter(tasks *RequestChan) (err error) {
 		} else {
 			s.incrOpStats(r, resp.Type)
 		}
+
 		nowTime := time.Now().UnixNano()
 		duration := int64((nowTime - r.ReceiveTime) / 1e3)
 		s.updateMaxDelay(duration, r)
 		if fflush {
 			s.flushOpStats(false)
 		}
-		if duration >= s.config.SlowlogLogSlowerThan {
-			SlowCmdCount.Incr() // Atomic global variable, increment by 1 when slow log occurs.
-			//client -> proxy -> server -> porxy -> client
-			//Record the waiting time from receiving the request from the client to sending it to the backend server
-			//the waiting time from sending the request to the backend server to receiving the response from the server
-			//the waiting time from receiving the server response to sending it to the client
-			var d0, d1, d2 int64 = -1, -1, -1
-			if r.SendToServerTime > 0 {
-				d0 = int64((r.SendToServerTime - r.ReceiveTime) / 1e3)
-			}
-			if r.SendToServerTime > 0 && r.ReceiveFromServerTime > 0 {
-				d1 = int64((r.ReceiveFromServerTime - r.SendToServerTime) / 1e3)
-			}
-			if r.ReceiveFromServerTime > 0 {
-				d2 = int64((nowTime - r.ReceiveFromServerTime) / 1e3)
+		if s.config.SlowlogLogSlowerThan >= 0 {
+			if duration >= s.config.SlowlogLogSlowerThan {
+				SlowCmdCount.Incr()
+				// Atomic global variable, increment by 1 when slow log occurs.
+				//client -> proxy -> server -> porxy -> client
+				//Record the waiting time from receiving the request from the client to sending it to the backend server
+				//the waiting time from sending the request to the backend server to receiving the response from the server
+				//the waiting time from receiving the server response to sending it to the client
+				var d0, d1, d2 int64 = -1, -1, -1
+				if r.SendToServerTime > 0 {
+					d0 = int64((r.SendToServerTime - r.ReceiveTime) / 1e3)
+				}
+				if r.SendToServerTime > 0 && r.ReceiveFromServerTime > 0 {
+					d1 = int64((r.ReceiveFromServerTime - r.SendToServerTime) / 1e3)
+				}
+				if r.ReceiveFromServerTime > 0 {
+					d2 = int64((nowTime - r.ReceiveFromServerTime) / 1e3)
+				}
+				index := getWholeCmd(r.Multi, cmd)
+				log.Errorf("%s remote:%s, start_time(us):%d, duration(us): [%d, %d, %d], %d, tasksLen:%d, command:[%s].",
+					time.Unix(r.ReceiveTime/1e9, 0).Format("2006-01-02 15:04:05"), s.Conn.RemoteAddr(), r.ReceiveTime/1e3, d0, d1, d2, duration, r.TasksLen, string(cmd[:index]))
 			}
-			index := getWholeCmd(r.Multi, cmd)
-			log.Errorf("%s remote:%s, start_time(us):%d, duration(us): [%d, %d, %d], %d, tasksLen:%d, command:[%s].",
-				time.Unix(r.ReceiveTime/1e9, 0).Format("2006-01-02 15:04:05"), s.Conn.RemoteAddr(), r.ReceiveTime/1e3, d0, d1, d2, duration, r.TasksLen, string(cmd[:index]))
 		}
 		return nil
 	})
@@ -681,32 +689,67 @@ func (s *Session) handleRequestSlotsMapping(r *Request, d *Router) error {
 	}
 }
 
-func (s *Session) incrOpTotal() {
-	s.stats.total.Incr()
-}
+func (s *Session) getOpStats(opstr string, create bool) *opStats {
+	var (
+		ok   bool
+		stat *opStats
+	)
 
-func (s *Session) getOpStats(opstr string) *opStats {
-	e := s.stats.opmap[opstr]
-	if e == nil {
-		e = &opStats{opstr: opstr}
-		s.stats.opmap[opstr] = e
+	func() {
+		cmdstats.opmapLock.RLock()
+		defer cmdstats.opmapLock.RUnlock()
+		stat, ok = s.stats.opmap[opstr]
+	}()
+	if (ok && stat != nil) || !create {
+		return stat
+	}
+	cmdstats.opmapLock.Lock()
+	defer cmdstats.opmapLock.Unlock()
+	stat, ok = cmdstats.opmap[opstr]
+	if ok && stat != nil {
+		return stat
+	}
+	stat = &opStats{opstr: opstr}
+	for i := 0; i < IntervalNum; i++ {
+		stat.delayInfo[i] = &delayInfo{interval: IntervalMark[i]}
 	}
-	return e
+	s.stats.opmap[opstr] = stat
+
+	return stat
 }
 
 func (s *Session) incrOpStats(r *Request, t redis.RespType) {
-	e := s.getOpStats(r.OpStr)
-	e.calls.Incr()
-	e.nsecs.Add(time.Now().UnixNano() - r.ReceiveTime)
+	if r == nil {
+		return
+	}
+	responseTime := time.Now().UnixNano() - r.ReceiveTime
+	var (
+		ok   bool
+		stat *opStats
+	)
+	stat, ok = s.stats.opmap[r.OpStr]
+	if !ok || stat == nil {
+		stat = getOpStats(r.OpStr, true)
+		s.stats.opmap[r.OpStr] = stat
+	}
+	stat.incrOpStats(responseTime, redis.RespType(t))
+	stat, ok = s.stats.opmap["ALL"]
+	if !ok || stat == nil {
+		stat = getOpStats("ALL", true)
+		s.stats.opmap["ALL"] = stat
+	}
+	stat.incrOpStats(responseTime, redis.RespType(t))
+	stat.calls.Incr()
+	stat.nsecs.Add(time.Now().UnixNano() - r.ReceiveTime)
 	switch t {
 	case redis.TypeError:
-		e.redis.errors.Incr()
+		incrOpRedisErrors()
 	}
 }
 
 func (s *Session) incrOpFails(r *Request, err error) error {
 	if r != nil {
-		e := s.getOpStats(r.OpStr)
+		e := s.getOpStats(r.OpStr, true)
 		e.fails.Incr()
 	} else {
 		s.stats.fails.Incr()
@@ -781,7 +824,7 @@ func (s *Session) handlePConfig(r *Request) error {
 }
 
 func (s *Session) updateMaxDelay(duration int64, r *Request) {
-	e := s.getOpStats(r.OpStr) // There is no race condition in the session
+	e := s.getOpStats(r.OpStr, true) // There is no race condition in the session
 	if duration > e.maxDelay.Int64() {
 		e.maxDelay.Set(duration)
 	}