tile38/internal/server/stats.go

541 lines
16 KiB
Go
Raw Normal View History

package server
2016-03-05 02:08:16 +03:00
import (
2016-08-26 22:54:19 +03:00
"bytes"
2016-03-05 02:08:16 +03:00
"encoding/json"
"fmt"
2016-12-23 00:52:37 +03:00
"os"
2016-03-05 02:08:16 +03:00
"runtime"
2016-03-29 01:50:18 +03:00
"sort"
2019-01-15 01:29:29 +03:00
"strconv"
2016-08-26 22:54:19 +03:00
"strings"
2019-09-04 02:35:42 +03:00
"sync"
2016-03-05 02:08:16 +03:00
"time"
2016-03-29 01:50:18 +03:00
"github.com/tidwall/resp"
"github.com/tidwall/tile38/core"
"github.com/tidwall/tile38/internal/collection"
2016-03-05 02:08:16 +03:00
)
2019-09-04 02:35:42 +03:00
var memStats runtime.MemStats
var memStatsMu sync.Mutex
var memStatsBG bool
// ReadMemStats returns the latest memstats. It provides an instant response.
func readMemStats() runtime.MemStats {
memStatsMu.Lock()
if !memStatsBG {
runtime.ReadMemStats(&memStats)
go func() {
for {
memStatsMu.Lock()
runtime.ReadMemStats(&memStats)
memStatsMu.Unlock()
time.Sleep(time.Second / 5)
}
}()
memStatsBG = true
}
ms := memStats
memStatsMu.Unlock()
return ms
}
func (c *Server) cmdStats(msg *Message) (res resp.Value, err error) {
2016-03-05 02:08:16 +03:00
start := time.Now()
vs := msg.Args[1:]
2016-03-05 02:08:16 +03:00
var ms = []map[string]interface{}{}
2017-10-05 18:20:40 +03:00
2016-03-29 02:11:29 +03:00
if len(vs) == 0 {
return NOMessage, errInvalidNumberOfArguments
2016-03-05 02:08:16 +03:00
}
2016-03-29 02:11:29 +03:00
var vals []resp.Value
var key string
var ok bool
for {
vs, key, ok = tokenval(vs)
if !ok {
break
}
2016-03-05 02:08:16 +03:00
col := c.getCol(key)
if col != nil {
m := make(map[string]interface{})
m["num_points"] = col.PointCount()
2016-03-05 02:08:16 +03:00
m["in_memory_size"] = col.TotalWeight()
2016-07-13 07:59:36 +03:00
m["num_objects"] = col.Count()
m["num_strings"] = col.StringCount()
2016-03-29 02:11:29 +03:00
switch msg.OutputType {
case JSON:
2016-03-29 02:11:29 +03:00
ms = append(ms, m)
case RESP:
2016-03-29 02:11:29 +03:00
vals = append(vals, resp.ArrayValue(respValuesSimpleMap(m)))
}
2016-03-05 02:08:16 +03:00
} else {
2016-03-29 02:11:29 +03:00
switch msg.OutputType {
case JSON:
2016-03-29 02:11:29 +03:00
ms = append(ms, nil)
case RESP:
2016-03-29 02:11:29 +03:00
vals = append(vals, resp.NullValue())
}
2016-03-05 02:08:16 +03:00
}
}
2016-03-29 02:11:29 +03:00
switch msg.OutputType {
case JSON:
2016-03-29 03:38:21 +03:00
2016-03-29 02:11:29 +03:00
data, err := json.Marshal(ms)
if err != nil {
return NOMessage, err
2016-03-29 02:11:29 +03:00
}
2017-10-05 18:20:40 +03:00
res = resp.StringValue(`{"ok":true,"stats":` + string(data) + `,"elapsed":"` + time.Now().Sub(start).String() + "\"}")
case RESP:
2017-10-05 18:20:40 +03:00
res = resp.ArrayValue(vals)
2016-03-05 02:08:16 +03:00
}
2016-03-29 02:11:29 +03:00
return res, nil
2016-03-05 02:08:16 +03:00
}
func (c *Server) cmdServer(msg *Message) (res resp.Value, err error) {
2016-03-05 02:08:16 +03:00
start := time.Now()
m := make(map[string]interface{})
args := msg.Args[1:]
2017-10-05 18:20:40 +03:00
// Switch on the type of stats requested
switch len(args) {
case 0:
c.basicStats(m)
case 1:
if strings.ToLower(args[0]) == "ext" {
c.extStats(m)
}
default:
return NOMessage, errInvalidNumberOfArguments
2016-03-05 02:08:16 +03:00
}
switch msg.OutputType {
case JSON:
data, err := json.Marshal(m)
if err != nil {
return NOMessage, err
}
res = resp.StringValue(`{"ok":true,"stats":` + string(data) + `,"elapsed":"` + time.Since(start).String() + "\"}")
case RESP:
vals := respValuesSimpleMap(m)
res = resp.ArrayValue(vals)
}
return res, nil
}
// basicStats populates the passed map with basic system/go/tile38 statistics
func (c *Server) basicStats(m map[string]interface{}) {
2017-09-30 04:11:05 +03:00
m["id"] = c.config.serverID()
if c.config.followHost() != "" {
m["following"] = fmt.Sprintf("%s:%d", c.config.followHost(),
c.config.followPort())
2016-03-05 02:08:16 +03:00
m["caught_up"] = c.fcup
m["caught_up_once"] = c.fcuponce
2016-03-05 02:08:16 +03:00
}
m["http_transport"] = c.http
2016-12-23 00:52:37 +03:00
m["pid"] = os.Getpid()
2016-03-05 02:08:16 +03:00
m["aof_size"] = c.aofsz
m["num_collections"] = c.cols.Len()
2016-03-30 19:32:38 +03:00
m["num_hooks"] = len(c.hooks)
2016-03-05 02:08:16 +03:00
sz := 0
c.cols.Scan(func(key string, value interface{}) bool {
col := value.(*collection.Collection)
2016-03-05 02:08:16 +03:00
sz += col.TotalWeight()
return true
})
m["in_memory_size"] = sz
points := 0
objects := 0
strings := 0
c.cols.Scan(func(key string, value interface{}) bool {
col := value.(*collection.Collection)
2016-03-05 02:08:16 +03:00
points += col.PointCount()
2016-07-13 07:59:36 +03:00
objects += col.Count()
strings += col.StringCount()
2016-03-05 02:08:16 +03:00
return true
})
m["num_points"] = points
m["num_objects"] = objects
m["num_strings"] = strings
2019-09-04 02:35:42 +03:00
mem := readMemStats()
2016-03-05 02:08:16 +03:00
avgsz := 0
if points != 0 {
avgsz = int(mem.HeapAlloc) / points
}
2016-10-25 01:58:30 +03:00
m["mem_alloc"] = mem.Alloc
2016-03-05 02:08:16 +03:00
m["heap_size"] = mem.HeapAlloc
2016-12-23 00:52:37 +03:00
m["heap_released"] = mem.HeapReleased
2017-09-30 04:11:05 +03:00
m["max_heap_size"] = c.config.maxMemory()
2016-03-05 02:08:16 +03:00
m["avg_item_size"] = avgsz
m["version"] = core.Version
2016-03-05 02:08:16 +03:00
m["pointer_size"] = (32 << uintptr(uint64(^uintptr(0))>>63)) / 8
2017-09-30 04:11:05 +03:00
m["read_only"] = c.config.readOnly()
2018-03-10 03:24:23 +03:00
m["cpus"] = runtime.NumCPU()
n, _ := runtime.ThreadCreateProfile(nil)
m["threads"] = float64(n)
}
2016-03-05 02:08:16 +03:00
// extStats populates the passed map with extended system/go/tile38 statistics
func (c *Server) extStats(m map[string]interface{}) {
n, _ := runtime.ThreadCreateProfile(nil)
2019-09-04 02:35:42 +03:00
mem := readMemStats()
// Go/Memory Stats
// Number of goroutines that currently exist
m["go_goroutines"] = runtime.NumGoroutine()
// Number of OS threads created
m["go_threads"] = float64(n)
// A summary of the GC invocation durations
m["go_version"] = runtime.Version()
// Number of bytes allocated and still in use
m["alloc_bytes"] = mem.Alloc
// Total number of bytes allocated, even if freed
m["alloc_bytes_total"] = mem.TotalAlloc
// Number of CPUS available on the system
m["sys_cpus"] = runtime.NumCPU()
// Number of bytes obtained from system
m["sys_bytes"] = mem.Sys
// Total number of pointer lookups
m["lookups_total"] = mem.Lookups
// Total number of mallocs
m["mallocs_total"] = mem.Mallocs
// Total number of frees
m["frees_total"] = mem.Frees
// Number of heap bytes allocated and still in use
m["heap_alloc_bytes"] = mem.HeapAlloc
// Number of heap bytes obtained from system
m["heap_sys_bytes"] = mem.HeapSys
// Number of heap bytes waiting to be used
m["heap_idle_bytes"] = mem.HeapIdle
// Number of heap bytes that are in use
m["heap_inuse_bytes"] = mem.HeapInuse
// Number of heap bytes released to OS
m["heap_released_bytes"] = mem.HeapReleased
// Number of allocated objects
m["heap_objects"] = mem.HeapObjects
// Number of bytes in use by the stack allocator
m["stack_inuse_bytes"] = mem.StackInuse
// Number of bytes obtained from system for stack allocator
m["stack_sys_bytes"] = mem.StackSys
// Number of bytes in use by mspan structures
m["mspan_inuse_bytes"] = mem.MSpanInuse
// Number of bytes used for mspan structures obtained from system
m["mspan_sys_bytes"] = mem.MSpanSys
// Number of bytes in use by mcache structures
m["mcache_inuse_bytes"] = mem.MCacheInuse
// Number of bytes used for mcache structures obtained from system
m["mcache_sys_bytes"] = mem.MCacheSys
// Number of bytes used by the profiling bucket hash table
m["buck_hash_sys_bytes"] = mem.BuckHashSys
// Number of bytes used for garbage collection system metadata
m["gc_sys_bytes"] = mem.GCSys
// Number of bytes used for other system allocations
m["other_sys_bytes"] = mem.OtherSys
// Number of heap bytes when next garbage collection will take place
m["next_gc_bytes"] = mem.NextGC
// Number of seconds since 1970 of last garbage collection
m["last_gc_time_seconds"] = float64(mem.LastGC) / 1e9
// The fraction of this program's available CPU time used by the GC since
// the program started
m["gc_cpu_fraction"] = mem.GCCPUFraction
// Tile38 Stats
// ID of the server
m["tile38_id"] = c.config.serverID()
// The process ID of the server
m["tile38_pid"] = os.Getpid()
// Version of Tile38 running
m["tile38_version"] = core.Version
// Maximum heap size allowed
m["tile38_max_heap_size"] = c.config.maxMemory()
// Type of instance running
if c.config.followHost() == "" {
m["tile38_type"] = "leader"
} else {
m["tile38_type"] = "follower"
2016-03-05 02:08:16 +03:00
}
// Whether or not the server is read-only
m["tile38_read_only"] = c.config.readOnly()
// Size of pointer
m["tile38_pointer_size"] = (32 << uintptr(uint64(^uintptr(0))>>63)) / 8
// Uptime of the Tile38 server in seconds
m["tile38_uptime_in_seconds"] = time.Since(c.started).Seconds()
// Number of currently connected Tile38 clients
c.connsmu.RLock()
m["tile38_connected_clients"] = len(c.conns)
c.connsmu.RUnlock()
// Whether or not a cluster is enabled
m["tile38_cluster_enabled"] = false
// Whether or not the Tile38 AOF is enabled
m["tile38_aof_enabled"] = core.AppendOnly
// Whether or not an AOF shrink is currently in progress
m["tile38_aof_rewrite_in_progress"] = c.shrinking
// Length of time the last AOF shrink took
m["tile38_aof_last_rewrite_time_sec"] = c.lastShrinkDuration.get() / int(time.Second)
// Duration of the on-going AOF rewrite operation if any
var currentShrinkStart time.Time
if currentShrinkStart.IsZero() {
m["tile38_aof_current_rewrite_time_sec"] = 0
} else {
m["tile38_aof_current_rewrite_time_sec"] = time.Since(currentShrinkStart).Seconds()
}
// Total size of the AOF in bytes
m["tile38_aof_size"] = c.aofsz
// Whether or no the HTTP transport is being served
m["tile38_http_transport"] = c.http
// Number of connections accepted by the server
m["tile38_total_connections_received"] = c.statsTotalConns.get()
// Number of commands processed by the server
m["tile38_total_commands_processed"] = c.statsTotalCommands.get()
2019-03-14 21:23:23 +03:00
// Number of webhook messages sent by server
m["tile38_total_messages_sent"] = c.statsTotalMsgsSent.get()
// Number of key expiration events
m["tile38_expired_keys"] = c.statsExpired.get()
// Number of connected slaves
m["tile38_connected_slaves"] = len(c.aofconnM)
points := 0
objects := 0
strings := 0
c.cols.Scan(func(key string, value interface{}) bool {
col := value.(*collection.Collection)
points += col.PointCount()
objects += col.Count()
strings += col.StringCount()
return true
})
// Number of points in the database
m["tile38_num_points"] = points
// Number of objects in the database
m["tile38_num_objects"] = objects
// Number of string in the database
m["tile38_num_strings"] = strings
// Number of collections in the database
m["tile38_num_collections"] = c.cols.Len()
// Number of hooks in the database
m["tile38_num_hooks"] = len(c.hooks)
avgsz := 0
if points != 0 {
avgsz = int(mem.HeapAlloc) / points
}
// Average point size in bytes
m["tile38_avg_point_size"] = avgsz
sz := 0
c.cols.Scan(func(key string, value interface{}) bool {
col := value.(*collection.Collection)
sz += col.TotalWeight()
return true
})
// Total in memory size of all collections
m["tile38_in_memory_size"] = sz
2016-03-05 02:08:16 +03:00
}
2016-08-26 23:42:52 +03:00
func (c *Server) writeInfoServer(w *bytes.Buffer) {
2016-08-26 23:42:52 +03:00
fmt.Fprintf(w, "tile38_version:%s\r\n", core.Version)
fmt.Fprintf(w, "redis_version:%s\r\n", core.Version) // Version of the Redis server
fmt.Fprintf(w, "uptime_in_seconds:%d\r\n", int(time.Since(c.started).Seconds())) // Number of seconds since Redis server start
2016-08-26 22:54:19 +03:00
}
func (c *Server) writeInfoClients(w *bytes.Buffer) {
2017-09-30 18:00:29 +03:00
c.connsmu.RLock()
2017-09-30 21:06:10 +03:00
fmt.Fprintf(w, "connected_clients:%d\r\n", len(c.conns)) // Number of client connections (excluding connections from slaves)
2017-09-30 18:00:29 +03:00
c.connsmu.RUnlock()
2016-08-26 22:54:19 +03:00
}
func (c *Server) writeInfoMemory(w *bytes.Buffer) {
2019-09-04 02:35:42 +03:00
mem := readMemStats()
2016-08-26 22:54:19 +03:00
fmt.Fprintf(w, "used_memory:%d\r\n", mem.Alloc) // total number of bytes allocated by Redis using its allocator (either standard libc, jemalloc, or an alternative allocator such as tcmalloc
}
func boolInt(t bool) int {
if t {
return 1
}
return 0
}
func (c *Server) writeInfoPersistence(w *bytes.Buffer) {
fmt.Fprintf(w, "aof_enabled:%d\r\n", boolInt(core.AppendOnly))
2017-09-30 18:00:29 +03:00
fmt.Fprintf(w, "aof_rewrite_in_progress:%d\r\n", boolInt(c.shrinking)) // Flag indicating a AOF rewrite operation is on-going
fmt.Fprintf(w, "aof_last_rewrite_time_sec:%d\r\n", c.lastShrinkDuration.get()/int(time.Second)) // Duration of the last AOF rewrite operation in seconds
2018-11-23 11:39:04 +03:00
var currentShrinkStart time.Time // c.currentShrinkStart.get()
2017-09-30 18:00:29 +03:00
if currentShrinkStart.IsZero() {
2016-08-26 22:54:19 +03:00
fmt.Fprintf(w, "aof_current_rewrite_time_sec:0\r\n") // Duration of the on-going AOF rewrite operation if any
} else {
2017-09-30 18:00:29 +03:00
fmt.Fprintf(w, "aof_current_rewrite_time_sec:%d\r\n", time.Now().Sub(currentShrinkStart)/time.Second) // Duration of the on-going AOF rewrite operation if any
2016-08-26 22:54:19 +03:00
}
}
func (c *Server) writeInfoStats(w *bytes.Buffer) {
2017-09-30 17:29:03 +03:00
fmt.Fprintf(w, "total_connections_received:%d\r\n", c.statsTotalConns.get()) // Total number of connections accepted by the server
fmt.Fprintf(w, "total_commands_processed:%d\r\n", c.statsTotalCommands.get()) // Total number of commands processed by the server
2019-03-14 21:23:23 +03:00
fmt.Fprintf(w, "total_messages_sent:%d\r\n", c.statsTotalMsgsSent.get()) // Total number of commands processed by the server
2017-09-30 17:29:03 +03:00
fmt.Fprintf(w, "expired_keys:%d\r\n", c.statsExpired.get()) // Total number of key expiration events
2016-08-26 22:54:19 +03:00
}
// writeInfoReplication writes all replication data to the 'info' response
func (c *Server) writeInfoReplication(w *bytes.Buffer) {
if c.config.followHost() != "" {
fmt.Fprintf(w, "role:slave\r\n")
fmt.Fprintf(w, "master_host:%s\r\n", c.config.followHost())
fmt.Fprintf(w, "master_port:%v\r\n", c.config.followPort())
} else {
fmt.Fprintf(w, "role:master\r\n")
var i int
2019-08-09 18:48:23 +03:00
c.connsmu.RLock()
for _, cc := range c.conns {
if cc.replPort != 0 {
fmt.Fprintf(w, "slave%v:ip=%s,port=%v,state=online\r\n", i,
strings.Split(cc.remoteAddr, ":")[0], cc.replPort)
i++
}
}
2019-08-09 18:48:23 +03:00
c.connsmu.RUnlock()
}
2016-08-26 22:54:19 +03:00
fmt.Fprintf(w, "connected_slaves:%d\r\n", len(c.aofconnM)) // Number of connected slaves
}
func (c *Server) writeInfoCluster(w *bytes.Buffer) {
2016-08-26 22:54:19 +03:00
fmt.Fprintf(w, "cluster_enabled:0\r\n")
}
func (c *Server) cmdInfo(msg *Message) (res resp.Value, err error) {
2016-08-26 22:54:19 +03:00
start := time.Now()
2017-10-05 18:20:40 +03:00
2016-08-26 22:54:19 +03:00
sections := []string{"server", "clients", "memory", "persistence", "stats", "replication", "cpu", "cluster", "keyspace"}
switch len(msg.Args) {
2016-08-26 22:54:19 +03:00
default:
return NOMessage, errInvalidNumberOfArguments
2016-08-26 22:54:19 +03:00
case 1:
case 2:
section := strings.ToLower(msg.Args[1])
2016-08-26 22:54:19 +03:00
switch section {
default:
sections = []string{section}
case "all":
sections = []string{"server", "clients", "memory", "persistence", "stats", "replication", "cpu", "commandstats", "cluster", "keyspace"}
case "default":
}
}
2016-03-29 02:11:29 +03:00
2016-08-26 22:54:19 +03:00
w := &bytes.Buffer{}
for i, section := range sections {
if i > 0 {
w.WriteString("\r\n")
}
switch strings.ToLower(section) {
default:
continue
case "server":
w.WriteString("# Server\r\n")
c.writeInfoServer(w)
case "clients":
w.WriteString("# Clients\r\n")
c.writeInfoClients(w)
case "memory":
w.WriteString("# Memory\r\n")
c.writeInfoMemory(w)
case "persistence":
w.WriteString("# Persistence\r\n")
c.writeInfoPersistence(w)
case "stats":
w.WriteString("# Stats\r\n")
c.writeInfoStats(w)
case "replication":
w.WriteString("# Replication\r\n")
c.writeInfoReplication(w)
case "cpu":
w.WriteString("# CPU\r\n")
c.writeInfoCPU(w)
case "cluster":
w.WriteString("# Cluster\r\n")
c.writeInfoCluster(w)
}
}
switch msg.OutputType {
case JSON:
2019-01-15 01:29:29 +03:00
// Create a map of all key/value info fields
m := make(map[string]interface{})
for _, kv := range strings.Split(w.String(), "\r\n") {
kv = strings.TrimSpace(kv)
if !strings.HasPrefix(kv, "#") {
if split := strings.SplitN(kv, ":", 2); len(split) == 2 {
m[split[0]] = tryParseType(split[1])
}
}
}
// Marshal the map and use the output in the JSON response
data, err := json.Marshal(m)
2016-08-26 22:54:19 +03:00
if err != nil {
return NOMessage, err
2016-08-26 22:54:19 +03:00
}
2017-10-05 18:20:40 +03:00
res = resp.StringValue(`{"ok":true,"info":` + string(data) + `,"elapsed":"` + time.Now().Sub(start).String() + "\"}")
case RESP:
2017-10-05 18:20:40 +03:00
res = resp.BytesValue(w.Bytes())
2016-08-26 22:54:19 +03:00
}
return res, nil
}
2019-01-15 01:29:29 +03:00
// tryParseType attempts to parse the passed string as an integer, float64 and
// a bool returning any successful parsed values. It returns the passed string
// if all tries fail
func tryParseType(str string) interface{} {
if v, err := strconv.ParseInt(str, 10, 64); err == nil {
return v
}
if v, err := strconv.ParseFloat(str, 64); err == nil {
return v
}
if v, err := strconv.ParseBool(str); err == nil {
return v
}
return str
}
2016-03-29 02:11:29 +03:00
func respValuesSimpleMap(m map[string]interface{}) []resp.Value {
var keys []string
2016-04-03 05:16:36 +03:00
for key := range m {
2016-03-29 02:11:29 +03:00
keys = append(keys, key)
}
sort.Strings(keys)
var vals []resp.Value
for _, key := range keys {
val := m[key]
vals = append(vals, resp.StringValue(key))
vals = append(vals, resp.StringValue(fmt.Sprintf("%v", val)))
}
return vals
}
func (c *Server) statsCollections(line string) (string, error) {
2016-03-05 02:08:16 +03:00
start := time.Now()
var key string
var ms = []map[string]interface{}{}
for len(line) > 0 {
line, key = token(line)
col := c.getCol(key)
if col != nil {
m := make(map[string]interface{})
points := col.PointCount()
m["num_points"] = points
m["in_memory_size"] = col.TotalWeight()
2016-07-13 07:59:36 +03:00
m["num_objects"] = col.Count()
2016-03-05 02:08:16 +03:00
ms = append(ms, m)
} else {
ms = append(ms, nil)
}
}
data, err := json.Marshal(ms)
if err != nil {
return "", err
}
return `{"ok":true,"stats":` + string(data) + `,"elapsed":"` + time.Now().Sub(start).String() + "\"}", nil
}