From 0aab1d7f305368576ca96f6e6fabebb8bebf8fc6 Mon Sep 17 00:00:00 2001 From: Nathaniel Cook Date: Fri, 19 Feb 2016 16:46:26 -0700 Subject: [PATCH] expvar refactored to allow getting and deleting --- CHANGELOG.md | 6 + batch.go | 2 +- cmd/kapacitord/run/server.go | 32 +---- edge.go | 7 +- expvar/expvar.go | 245 +++++++++++++++++++++++++++++++++ global_stats.go | 199 ++++++++++++-------------- join.go | 18 +-- node.go | 34 +++-- services/influxdb/service.go | 2 +- services/reporting/service.go | 14 +- services/stats/service.go | 5 +- services/task_store/service.go | 12 +- task.go | 2 +- task_master.go | 5 +- timer/noop.go | 11 +- timer/timer.go | 45 ++---- timer/timer_test.go | 25 ++++ union.go | 17 +-- 18 files changed, 439 insertions(+), 242 deletions(-) create mode 100644 expvar/expvar.go create mode 100644 timer/timer_test.go diff --git a/CHANGELOG.md b/CHANGELOG.md index 67c439c07..80cef351e 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -4,12 +4,18 @@ ### Release Notes +Kapacitor now exposes more internal metrics for determining the performance of a given task. +The internal stats now includes a measurement `node` that contains an averaged execution time for the node, tagged by the task, node, task type and kind of node (i.e. window vs union). +These stats are also available in the DOT output of the Kapacitor show command. + + ### Features - [#236](https://github.com/influxdata/kapacitor/issues/236): Implement batched group by - [#231](https://github.com/influxdata/kapacitor/pull/231): Add ShiftNode so values can be shifted in time for joining/comparisons. - [#190](https://github.com/influxdata/kapacitor/issues/190): BREAKING: Deadman's switch now triggers off emitted counts and is grouped by to original grouping of the data. The breaking change is that the 'collected' stat is no longer output for `.stats` and has been replaced by `emitted`. - [#145](https://github.com/influxdata/kapacitor/issues/145): The InfluxDB Out Node now writes data to InfluxDB in buffers. +- [#215](https://github.com/influxdata/kapacitor/issues/215): Add performance metrics to nodes for average execution times and node throughput values. ### Bugfixes diff --git a/batch.go b/batch.go index bfcf50587..eee35f23b 100644 --- a/batch.go +++ b/batch.go @@ -96,7 +96,7 @@ func (s *SourceBatchNode) Queries(start, stop time.Time) [][]string { // Do not add the source batch node to the dot output // since its not really an edge. -func (s *SourceBatchNode) edot(*bytes.Buffer, time.Duration) {} +func (s *SourceBatchNode) edot(*bytes.Buffer) {} func (s *SourceBatchNode) collectedCount() (count int64) { for _, child := range s.children { diff --git a/cmd/kapacitord/run/server.go b/cmd/kapacitord/run/server.go index ddefbd3c5..953b24879 100644 --- a/cmd/kapacitord/run/server.go +++ b/cmd/kapacitord/run/server.go @@ -2,7 +2,6 @@ package run import ( "errors" - "expvar" "fmt" "io/ioutil" "log" @@ -45,27 +44,6 @@ import ( const clusterIDFilename = "cluster.id" const serverIDFilename = "server.id" -var ( - //Published vars - cidVar = &expvar.String{} - - sidVar = &expvar.String{} - - hostVar = &expvar.String{} - - productVar = &expvar.String{} - - versionVar = &expvar.String{} -) - -func init() { - expvar.Publish(kapacitor.ClusterIDVarName, cidVar) - expvar.Publish(kapacitor.ServerIDVarName, sidVar) - expvar.Publish(kapacitor.HostVarName, hostVar) - expvar.Publish(kapacitor.ProductVarName, productVar) - expvar.Publish(kapacitor.VersionVarName, versionVar) -} - // BuildInfo represents the build details for the server code. type BuildInfo struct { Version string @@ -414,11 +392,11 @@ func (s *Server) Open() error { } // Set published vars - cidVar.Set(s.ClusterID) - sidVar.Set(s.ServerID) - hostVar.Set(s.hostname) - productVar.Set(kapacitor.Product) - versionVar.Set(s.buildInfo.Version) + kapacitor.ClusterIDVar.Set(s.ClusterID) + kapacitor.ServerIDVar.Set(s.ServerID) + kapacitor.HostVar.Set(s.hostname) + kapacitor.ProductVar.Set(kapacitor.Product) + kapacitor.VersionVar.Set(s.buildInfo.Version) s.Logger.Printf("I! ClusterID: %s ServerID: %s", s.ClusterID, s.ServerID) // Start profiling, if set. diff --git a/edge.go b/edge.go index 44ba49029..088338716 100644 --- a/edge.go +++ b/edge.go @@ -2,12 +2,12 @@ package kapacitor import ( "errors" - "expvar" "fmt" "log" "strconv" "sync" + "github.com/influxdata/kapacitor/expvar" "github.com/influxdata/kapacitor/models" "github.com/influxdata/kapacitor/pipeline" ) @@ -36,6 +36,7 @@ type Edge struct { logger *log.Logger aborted chan struct{} + statsKey string statMap *expvar.Map groupMu sync.RWMutex groupStats map[models.GroupID]*edgeStat @@ -48,10 +49,11 @@ func newEdge(taskName, parentName, childName string, t pipeline.EdgeType, logSer "child": childName, "type": t.String(), } - sm := NewStatistics("edges", tags) + key, sm := NewStatistics("edges", tags) sm.Add(statCollected, 0) sm.Add(statEmitted, 0) e := &Edge{ + statsKey: key, statMap: sm, aborted: make(chan struct{}), groupStats: make(map[models.GroupID]*edgeStat), @@ -125,6 +127,7 @@ func (e *Edge) Close() { if e.reduce != nil { close(e.reduce) } + DeleteStatistics(e.statsKey) } // Abort all next and collect calls. diff --git a/expvar/expvar.go b/expvar/expvar.go new file mode 100644 index 000000000..230448ce0 --- /dev/null +++ b/expvar/expvar.go @@ -0,0 +1,245 @@ +// This package is a fork of the golang expvar expvar.Var types. +// Adding extra support for deleting and accessing raw typed values. +package expvar + +import ( + "bytes" + "expvar" + "fmt" + "math" + "sort" + "strconv" + "sync" + "sync/atomic" +) + +// Int is a 64-bit integer variable that satisfies the expvar.Var interface. +type Int struct { + i int64 +} + +func (v *Int) String() string { + return strconv.FormatInt(v.Get(), 10) +} + +func (v *Int) Add(delta int64) { + atomic.AddInt64(&v.i, delta) +} + +func (v *Int) Set(value int64) { + atomic.StoreInt64(&v.i, value) +} + +func (v *Int) Get() int64 { + return atomic.LoadInt64(&v.i) +} + +// Float is a 64-bit float variable that satisfies the expvar.Var interface. +type Float struct { + f uint64 +} + +func (v *Float) String() string { + return strconv.FormatFloat(v.Get(), 'g', -1, 64) +} + +func (v *Float) Get() float64 { + return math.Float64frombits(atomic.LoadUint64(&v.f)) +} + +// Add adds delta to v. +func (v *Float) Add(delta float64) { + for { + cur := atomic.LoadUint64(&v.f) + curVal := math.Float64frombits(cur) + nxtVal := curVal + delta + nxt := math.Float64bits(nxtVal) + if atomic.CompareAndSwapUint64(&v.f, cur, nxt) { + return + } + } +} + +// Set sets v to value. +func (v *Float) Set(value float64) { + atomic.StoreUint64(&v.f, math.Float64bits(value)) +} + +// MaxFloat is a 64-bit float variable that satisfies the expvar.Var interface. +// When setting a value it will only be set if it is greater than the current value. +type MaxFloat struct { + f uint64 +} + +func (v *MaxFloat) String() string { + return strconv.FormatFloat(v.Get(), 'g', -1, 64) +} + +func (v *MaxFloat) Get() float64 { + return math.Float64frombits(atomic.LoadUint64(&v.f)) +} + +// Set sets v to value. +func (v *MaxFloat) Set(value float64) { + nxtBits := math.Float64bits(value) + for { + curBits := atomic.LoadUint64(&v.f) + cur := math.Float64frombits(curBits) + if value > cur { + if atomic.CompareAndSwapUint64(&v.f, curBits, nxtBits) { + return + } + } else { + return + } + } +} + +// Map is a string-to-expvar.Var map variable that satisfies the expvar.Var interface. +type Map struct { + mu sync.RWMutex + m map[string]expvar.Var + keys []string // sorted +} + +func (v *Map) String() string { + v.mu.RLock() + defer v.mu.RUnlock() + var b bytes.Buffer + fmt.Fprintf(&b, "{") + first := true + v.doLocked(func(kv expvar.KeyValue) { + if !first { + fmt.Fprintf(&b, ", ") + } + fmt.Fprintf(&b, "%q: %v", kv.Key, kv.Value) + first = false + }) + fmt.Fprintf(&b, "}") + return b.String() +} + +func (v *Map) Init() *Map { + v.m = make(map[string]expvar.Var) + return v +} + +// updateKeys updates the sorted list of keys in v.keys. +// must be called with v.mu held. +func (v *Map) updateKeys() { + if len(v.m) == len(v.keys) { + // No new key. + return + } + v.keys = v.keys[:0] + for k := range v.m { + v.keys = append(v.keys, k) + } + sort.Strings(v.keys) +} + +func (v *Map) Get(key string) expvar.Var { + v.mu.RLock() + defer v.mu.RUnlock() + return v.m[key] +} + +func (v *Map) Set(key string, av expvar.Var) { + v.mu.Lock() + defer v.mu.Unlock() + v.m[key] = av + v.updateKeys() +} + +func (v *Map) Delete(key string) { + v.mu.Lock() + defer v.mu.Unlock() + delete(v.m, key) + v.updateKeys() +} + +func (v *Map) Add(key string, delta int64) { + v.mu.RLock() + av, ok := v.m[key] + v.mu.RUnlock() + if !ok { + // check again under the write lock + v.mu.Lock() + av, ok = v.m[key] + if !ok { + av = new(Int) + v.m[key] = av + v.updateKeys() + } + v.mu.Unlock() + } + + // Add to Int; ignore otherwise. + if iv, ok := av.(*Int); ok { + iv.Add(delta) + } +} + +// AddFloat adds delta to the *Float value stored under the given map key. +func (v *Map) AddFloat(key string, delta float64) { + v.mu.RLock() + av, ok := v.m[key] + v.mu.RUnlock() + if !ok { + // check again under the write lock + v.mu.Lock() + av, ok = v.m[key] + if !ok { + av = new(Float) + v.m[key] = av + v.updateKeys() + } + v.mu.Unlock() + } + + // Add to Float; ignore otherwise. + if iv, ok := av.(*Float); ok { + iv.Add(delta) + } +} + +// Do calls f for each entry in the map. +// The map is locked during the iteration, +// but existing entries may be concurrently updated. +func (v *Map) Do(f func(expvar.KeyValue)) { + v.mu.RLock() + defer v.mu.RUnlock() + v.doLocked(f) +} + +// doLocked calls f for each entry in the map. +// v.mu must be held for reads. +func (v *Map) doLocked(f func(expvar.KeyValue)) { + for _, k := range v.keys { + f(expvar.KeyValue{k, v.m[k]}) + } +} + +// String is a string variable, and satisfies the expvar.Var interface. +type String struct { + mu sync.RWMutex + s string +} + +func (v *String) String() string { + v.mu.RLock() + defer v.mu.RUnlock() + return strconv.Quote(v.s) +} + +func (v *String) Set(value string) { + v.mu.Lock() + defer v.mu.Unlock() + v.s = value +} + +func (v *String) Get() string { + v.mu.RLock() + defer v.mu.RUnlock() + return v.s +} diff --git a/global_stats.go b/global_stats.go index 3927843e1..9e1c22cfb 100644 --- a/global_stats.go +++ b/global_stats.go @@ -2,11 +2,11 @@ package kapacitor import ( "expvar" + "fmt" "runtime" - "strconv" - "sync" "time" + kexpvar "github.com/influxdata/kapacitor/expvar" "github.com/twinj/uuid" ) @@ -30,9 +30,18 @@ const ( var ( // Global expvars - NumTasks = &expvar.Int{} - NumEnabledTasks = &expvar.Int{} - NumSubscriptions = &expvar.Int{} + NumTasksVar = &kexpvar.Int{} + NumEnabledTasksVar = &kexpvar.Int{} + NumSubscriptionsVar = &kexpvar.Int{} + + ClusterIDVar = &kexpvar.String{} + ServerIDVar = &kexpvar.String{} + HostVar = &kexpvar.String{} + ProductVar = &kexpvar.String{} + VersionVar = &kexpvar.String{} + + // All internal stats are added as sub-maps to this top level map. + stats *kexpvar.Map ) var ( @@ -41,78 +50,65 @@ var ( func init() { startTime = time.Now().UTC() - expvar.Publish(NumTasksVarName, NumTasks) - expvar.Publish(NumEnabledTasksVarName, NumEnabledTasks) - expvar.Publish(NumSubscriptionsVarName, NumSubscriptions) -} -// Gets an exported var and returns its unquoted string contents -func GetStringVar(name string) string { - s, err := strconv.Unquote(expvar.Get(name).String()) - if err != nil { - panic(err) - } - return s -} + expvar.Publish(NumTasksVarName, NumTasksVar) + expvar.Publish(NumEnabledTasksVarName, NumEnabledTasksVar) + expvar.Publish(NumSubscriptionsVarName, NumSubscriptionsVar) -// Gets an exported var and returns its int value -func GetIntVar(name string) int64 { - i, err := strconv.ParseInt(expvar.Get(name).String(), 10, 64) - if err != nil { - panic(err) - } - return i -} + expvar.Publish(ClusterIDVarName, ClusterIDVar) + expvar.Publish(ServerIDVarName, ServerIDVar) + expvar.Publish(HostVarName, HostVar) + expvar.Publish(ProductVarName, ProductVar) + expvar.Publish(VersionVarName, VersionVar) -// Gets an exported var and returns its float value -func GetFloatVar(name string) float64 { - f, err := strconv.ParseFloat(expvar.Get(name).String(), 64) - if err != nil { - panic(err) - } - return f + // Initialze the global stats map + stats = &kexpvar.Map{} + stats.Init() + expvar.Publish(Product, stats) } func Uptime() time.Duration { return time.Now().Sub(startTime) } -var expvarMu sync.Mutex - // NewStatistics creates an expvar-based map. Within there "name" is the Measurement name, "tags" are the tags, // and values are placed at the key "values". // The "values" map is returned so that statistics can be set. -func NewStatistics(name string, tags map[string]string) *expvar.Map { - expvarMu.Lock() - defer expvarMu.Unlock() - +func NewStatistics(name string, tags map[string]string) (string, *kexpvar.Map) { key := uuid.NewV4().String() - m := &expvar.Map{} + m := &kexpvar.Map{} m.Init() - expvar.Publish(key, m) // Set the name - nameVar := &expvar.String{} + nameVar := &kexpvar.String{} nameVar.Set(name) m.Set("name", nameVar) // Set the tags - tagsVar := &expvar.Map{} + tagsVar := &kexpvar.Map{} tagsVar.Init() for k, v := range tags { - value := &expvar.String{} + value := &kexpvar.String{} value.Set(v) tagsVar.Set(k, value) } m.Set("tags", tagsVar) // Create and set the values entry used for actual stats. - statMap := &expvar.Map{} + statMap := &kexpvar.Map{} statMap.Init() m.Set("values", statMap) - return statMap + // Set new statsMap on the top level map. + stats.Set(key, m) + + return key, statMap +} + +// Remove a statistics map. +func DeleteStatistics(key string) { + stats.Delete(key) } type StatsData struct { @@ -132,83 +128,68 @@ func GetStatsData() ([]StatsData, error) { allData = append(allData, globalData) + // Get all global statistics expvar.Do(func(kv expvar.KeyValue) { - var f interface{} - var err error switch v := kv.Value.(type) { - case *expvar.Float: - f, err = strconv.ParseFloat(v.String(), 64) - if err == nil { - globalData.Values[kv.Key] = f - } - case *expvar.Int: - f, err = strconv.ParseInt(v.String(), 10, 64) - if err == nil { - globalData.Values[kv.Key] = f - } - case *expvar.Map: - data := StatsData{ - Tags: make(map[string]string), - Values: make(map[string]interface{}), + case *kexpvar.Int: + globalData.Values[kv.Key] = v.Get() + case *kexpvar.Float: + globalData.Values[kv.Key] = v.Get() + case *kexpvar.Map: + if kv.Key != Product { + panic("unexpected published top level expvar.Map with key " + kv.Key) } + } + }) + // Get all other specific statistics + stats.Do(func(kv expvar.KeyValue) { + v := kv.Value.(*kexpvar.Map) + + data := StatsData{ + Tags: make(map[string]string), + Values: make(map[string]interface{}), + } - v.Do(func(subKV expvar.KeyValue) { - switch subKV.Key { - case "name": - // straight to string name. - u, err := strconv.Unquote(subKV.Value.String()) - if err != nil { - return + v.Do(func(subKV expvar.KeyValue) { + switch subKV.Key { + case "name": + data.Name = subKV.Value.(*kexpvar.String).Get() + case "tags": + // string-string tags map. + n := subKV.Value.(*kexpvar.Map) + n.Do(func(t expvar.KeyValue) { + data.Tags[t.Key] = t.Value.(*kexpvar.String).Get() + }) + case "values": + // string-interface map. + n := subKV.Value.(*kexpvar.Map) + n.Do(func(kv expvar.KeyValue) { + switch v := kv.Value.(type) { + case *kexpvar.Int: + data.Values[kv.Key] = v.Get() + case *kexpvar.Float: + data.Values[kv.Key] = v.Get() + case *kexpvar.MaxFloat: + data.Values[kv.Key] = v.Get() + default: + panic(fmt.Sprintf("unknown expvar.Var type for stats %T", kv.Value)) } - data.Name = u - case "tags": - // string-string tags map. - n := subKV.Value.(*expvar.Map) - n.Do(func(t expvar.KeyValue) { - u, err := strconv.Unquote(t.Value.String()) - if err != nil { - return - } - data.Tags[t.Key] = u - }) - case "values": - // string-interface map. - n := subKV.Value.(*expvar.Map) - n.Do(func(kv expvar.KeyValue) { - var f interface{} - var err error - switch v := kv.Value.(type) { - case *expvar.Float: - f, err = strconv.ParseFloat(v.String(), 64) - if err != nil { - return - } - case *expvar.Int: - f, err = strconv.ParseInt(v.String(), 10, 64) - if err != nil { - return - } - default: - return - } - data.Values[kv.Key] = f - }) - } - }) - - // If no field data, don't include it in the results - if len(data.Values) == 0 { - return + }) } + }) - allData = append(allData, data) + // If no field data, don't include it in the results + if len(data.Values) == 0 { + return } + + allData = append(allData, data) }) // Add uptime to globalData globalData.Values[UptimeVarName] = Uptime().Seconds() - // Add Go memstats. + // Add Go runtime stats. data := StatsData{ Name: "runtime", } diff --git a/join.go b/join.go index 67de8605b..cb8551fe9 100644 --- a/join.go +++ b/join.go @@ -99,22 +99,6 @@ func (j *JoinNode) runJoin([]byte) error { return nil } -func (j *JoinNode) nodeExecTime() time.Duration { - j.mu.RLock() - defer j.mu.RUnlock() - sum := 0.0 - total := 0.0 - for _, group := range j.groups { - avg, count := group.timer.AverageTime() - sum += float64(avg) * float64(count) - total += float64(count) - } - if total == 0 { - return 0 - } - return time.Duration(sum / total) -} - // safely get the group for the point or create one if it doesn't exist. func (j *JoinNode) getGroup(p models.PointInterface) *group { j.mu.Lock() @@ -151,7 +135,7 @@ func newGroup(i int, j *JoinNode) *group { head: make([]time.Time, i), j: j, points: make(chan srcPoint), - timer: j.et.tm.TimingService.NewTimer(), + timer: j.et.tm.TimingService.NewTimer(j.avgExecVar), } } diff --git a/node.go b/node.go index db9e0acd1..caac4f5f4 100644 --- a/node.go +++ b/node.go @@ -8,11 +8,16 @@ import ( "sync" "time" + "github.com/influxdata/kapacitor/expvar" "github.com/influxdata/kapacitor/models" "github.com/influxdata/kapacitor/pipeline" "github.com/influxdata/kapacitor/timer" ) +const ( + averageExecTimeVarName = "avg_execution_time" +) + // A node that can be in an executor. type Node interface { pipeline.Node @@ -40,12 +45,10 @@ type Node interface { abortParentEdges() // executing dot - edot(buf *bytes.Buffer, execTime time.Duration) + edot(buf *bytes.Buffer) nodeStatsByGroup() map[models.GroupID]nodeStats - nodeExecTime() time.Duration - collectedCount() int64 } @@ -65,6 +68,9 @@ type node struct { outs []*Edge logger *log.Logger timer timer.Timer + statsKey string + statMap *expvar.Map + avgExecVar *expvar.MaxFloat } func (n *node) addParentEdge(e *Edge) { @@ -78,7 +84,16 @@ func (n *node) abortParentEdges() { } func (n *node) start(snapshot []byte) { - n.timer = n.et.tm.TimingService.NewTimer() + tags := map[string]string{ + "task": n.et.Task.Name, + "node": n.Name(), + "type": n.et.Task.Type.String(), + "kind": n.Desc(), + } + n.statsKey, n.statMap = NewStatistics("nodes", tags) + n.avgExecVar = &expvar.MaxFloat{} + n.statMap.Set(averageExecTimeVarName, n.avgExecVar) + n.timer = n.et.tm.TimingService.NewTimer(n.avgExecVar) n.errCh = make(chan error, 1) go func() { var err error @@ -108,7 +123,7 @@ func (n *node) stop() { if n.stopF != nil { n.stopF() } - + DeleteStatistics(n.statsKey) } // no-op snapshot @@ -167,17 +182,12 @@ func (n *node) closeChildEdges() { } } -func (n *node) nodeExecTime() time.Duration { - execTime, _ := n.timer.AverageTime() - return execTime -} - -func (n *node) edot(buf *bytes.Buffer, execTime time.Duration) { +func (n *node) edot(buf *bytes.Buffer) { buf.Write([]byte( fmt.Sprintf("\n%s [label=\"%s %v\"];\n", n.Name(), n.Name(), - execTime, + time.Duration(n.avgExecVar.Get()), ), )) for i, c := range n.children { diff --git a/services/influxdb/service.go b/services/influxdb/service.go index 8069b966c..682828bd5 100644 --- a/services/influxdb/service.go +++ b/services/influxdb/service.go @@ -284,7 +284,7 @@ func (s *Service) linkSubscriptions() error { } } - kapacitor.NumSubscriptions.Set(numSubscriptions) + kapacitor.NumSubscriptionsVar.Set(numSubscriptions) return nil } diff --git a/services/reporting/service.go b/services/reporting/service.go index dcdb72768..45810c022 100644 --- a/services/reporting/service.go +++ b/services/reporting/service.go @@ -47,10 +47,10 @@ func (s *Service) Open() error { } // Populate published vars - s.clusterID = kapacitor.GetStringVar(kapacitor.ClusterIDVarName) - s.serverID = kapacitor.GetStringVar(kapacitor.ServerIDVarName) - s.hostname = kapacitor.GetStringVar(kapacitor.HostVarName) - s.version = kapacitor.GetStringVar(kapacitor.VersionVarName) + s.clusterID = kapacitor.ClusterIDVar.Get() + s.serverID = kapacitor.ServerIDVar.Get() + s.hostname = kapacitor.HostVar.Get() + s.version = kapacitor.VersionVar.Get() s.product = kapacitor.Product // Populate anonymous tags @@ -104,9 +104,9 @@ func (s *Service) sendUsageReport() error { // Add values data.Values[kapacitor.ClusterIDVarName] = s.clusterID data.Values[kapacitor.ServerIDVarName] = s.serverID - data.Values[kapacitor.NumTasksVarName] = kapacitor.GetIntVar(kapacitor.NumTasksVarName) - data.Values[kapacitor.NumEnabledTasksVarName] = kapacitor.GetIntVar(kapacitor.NumEnabledTasksVarName) - data.Values[kapacitor.NumSubscriptionsVarName] = kapacitor.GetIntVar(kapacitor.NumSubscriptionsVarName) + data.Values[kapacitor.NumTasksVarName] = kapacitor.NumTasksVar.Get() + data.Values[kapacitor.NumEnabledTasksVarName] = kapacitor.NumEnabledTasksVar.Get() + data.Values[kapacitor.NumSubscriptionsVarName] = kapacitor.NumSubscriptionsVar.Get() data.Values[kapacitor.UptimeVarName] = kapacitor.Uptime().Seconds() usage := client.Usage{ diff --git a/services/stats/service.go b/services/stats/service.go index 0f04ed120..44bb86338 100644 --- a/services/stats/service.go +++ b/services/stats/service.go @@ -29,6 +29,7 @@ import ( "time" "github.com/influxdata/kapacitor" + "github.com/influxdata/kapacitor/expvar" "github.com/influxdata/kapacitor/models" "github.com/influxdata/kapacitor/timer" ) @@ -133,6 +134,6 @@ func (s *Service) reportStats() { } } -func (s *Service) NewTimer() timer.Timer { - return timer.New(s.timingSampleRate, s.timingMovingAvgSize) +func (s *Service) NewTimer(avgVar *expvar.MaxFloat) timer.Timer { + return timer.New(s.timingSampleRate, s.timingMovingAvgSize, avgVar) } diff --git a/services/task_store/service.go b/services/task_store/service.go index eb60ec547..e4acf9a76 100644 --- a/services/task_store/service.go +++ b/services/task_store/service.go @@ -177,8 +177,8 @@ func (ts *Service) Open() error { } // Set expvars - kapacitor.NumTasks.Set(numTasks) - kapacitor.NumEnabledTasks.Set(numEnabledTasks) + kapacitor.NumTasksVar.Set(numTasks) + kapacitor.NumEnabledTasksVar.Set(numEnabledTasks) return nil } @@ -484,7 +484,7 @@ func (ts *Service) Save(task *rawTask) error { return err } if !exists { - kapacitor.NumTasks.Add(1) + kapacitor.NumTasksVar.Add(1) } return nil }) @@ -500,7 +500,7 @@ func (ts *Service) Delete(name string) error { exists := tb.Get([]byte(name)) != nil if exists { tb.Delete([]byte(name)) - kapacitor.NumTasks.Add(-1) + kapacitor.NumTasksVar.Add(-1) } } eb := tx.Bucket(enabledBucket) @@ -573,7 +573,7 @@ func (ts *Service) Enable(name string) error { return err } if !enabled { - kapacitor.NumEnabledTasks.Add(1) + kapacitor.NumEnabledTasksVar.Add(1) } return nil }) @@ -654,7 +654,7 @@ func (ts *Service) Disable(name string) error { if err != nil { return err } - kapacitor.NumEnabledTasks.Add(-1) + kapacitor.NumEnabledTasksVar.Add(-1) } return nil }) diff --git a/task.go b/task.go index 0496c6b2f..15fad599d 100644 --- a/task.go +++ b/task.go @@ -310,7 +310,7 @@ func (et *ExecutingTask) EDot() []byte { )) et.walk(func(n Node) error { - n.edot(&buf, n.nodeExecTime()) + n.edot(&buf) return nil }) buf.Write([]byte("}")) diff --git a/task_master.go b/task_master.go index e210b1f10..c32a67b69 100644 --- a/task_master.go +++ b/task_master.go @@ -7,6 +7,7 @@ import ( "sync" "time" + "github.com/influxdata/kapacitor/expvar" "github.com/influxdata/kapacitor/models" "github.com/influxdata/kapacitor/pipeline" "github.com/influxdata/kapacitor/services/httpd" @@ -91,7 +92,7 @@ type TaskMaster struct { Alert(title, text string) error } TimingService interface { - NewTimer() timer.Timer + NewTimer(*expvar.MaxFloat) timer.Timer } LogService LogService @@ -482,6 +483,6 @@ func (tm *TaskMaster) SnapshotTask(name string) (*TaskSnapshot, error) { type noOpTimingService struct{} -func (noOpTimingService) NewTimer() timer.Timer { +func (noOpTimingService) NewTimer(*expvar.MaxFloat) timer.Timer { return timer.NewNoOp() } diff --git a/timer/noop.go b/timer/noop.go index a8697cae2..d6f0935e6 100644 --- a/timer/noop.go +++ b/timer/noop.go @@ -1,15 +1,12 @@ package timer -import "time" - type noop struct { } -func (noop) Start() {} -func (noop) Pause() {} -func (noop) Resume() {} -func (noop) Stop() {} -func (noop) AverageTime() (time.Duration, int) { return 0, 1 } +func (noop) Start() {} +func (noop) Pause() {} +func (noop) Resume() {} +func (noop) Stop() {} func NewNoOp() Timer { return noop{} diff --git a/timer/timer.go b/timer/timer.go index 3cbe6d2d7..0e4ef5b13 100644 --- a/timer/timer.go +++ b/timer/timer.go @@ -2,8 +2,9 @@ package timer import ( "math/rand" - "sync" "time" + + "github.com/influxdata/kapacitor/expvar" ) type Timer interface { @@ -19,9 +20,6 @@ type Timer interface { // Stop the timer. // Timer must be started. Stop() - // Return average of timings and number of - // timings used to compute the average. - AverageTime() (time.Duration, int) } type timerState int @@ -40,12 +38,15 @@ type timer struct { current time.Duration avg *movavg state timerState + + avgVar *expvar.MaxFloat } -func New(sampleRate float64, movingAverageSize int) Timer { +func New(sampleRate float64, movingAverageSize int, avgVar *expvar.MaxFloat) Timer { return &timer{ sampleRate: sampleRate, avg: newMovAvg(movingAverageSize), + avgVar: avgVar, } } @@ -85,16 +86,10 @@ func (t *timer) Stop() { return } t.current += time.Now().Sub(t.start) - t.avg.update(float64(t.current)) + avg := t.avg.update(float64(t.current)) t.current = 0 t.state = Stopped -} - -// Return the average time in nanoseconds and -// the number of timings used to compute the average. -func (t *timer) AverageTime() (time.Duration, int) { - avg, count := t.avg.average() - return time.Duration(avg), count + t.avgVar.Set(avg) } // Maintains a moving average of values @@ -104,7 +99,6 @@ type movavg struct { idx int count int avg float64 - mu sync.RWMutex } func newMovAvg(size int) *movavg { @@ -115,30 +109,17 @@ func newMovAvg(size int) *movavg { } } -func (m *movavg) update(value float64) { - m.mu.Lock() - defer m.mu.Unlock() +func (m *movavg) update(value float64) float64 { + m.count++ n := float64(m.count) - if n == 0 { - m.count = 1 - m.avg = value - return - } - m.avg += (value - m.avg) / n m.idx = (m.idx + 1) % m.size - if m.count == m.size { + if m.count == m.size+1 { old := m.history[m.idx] m.avg = (n*m.avg - old) / (n - 1) - } else { - m.count++ + m.count-- } m.history[m.idx] = value -} - -func (m *movavg) average() (float64, int) { - m.mu.RLock() - defer m.mu.RUnlock() - return m.avg, m.count + return m.avg } diff --git a/timer/timer_test.go b/timer/timer_test.go new file mode 100644 index 000000000..95241b4f5 --- /dev/null +++ b/timer/timer_test.go @@ -0,0 +1,25 @@ +package timer + +import ( + "math" + "testing" +) + +func TestMovAvg(t *testing.T) { + count := 100 + ma := newMovAvg(count) + for i := 0; i < count; i++ { + avg := ma.update(1) + if got, exp := avg, 1.0; got != exp { + t.Fatalf("unexpected movavg: got %f, exp %f", got, exp) + } + } + c := float64(count) + for i := 0; i < count; i++ { + avg := ma.update(2) + f := float64(i + 1) + if got, exp := avg, ((c-f)+2.0*f)/c; math.Abs(got-exp) > 1e-8 { + t.Fatalf("unexpected movavg i: %d got %f, exp %f", i, got, exp) + } + } +} diff --git a/union.go b/union.go index 05cabffca..8fe46754c 100644 --- a/union.go +++ b/union.go @@ -2,7 +2,6 @@ package kapacitor import ( "log" - "time" "github.com/influxdata/kapacitor/pipeline" "github.com/influxdata/kapacitor/timer" @@ -33,7 +32,7 @@ func (u *UnionNode) runUnion([]byte) error { } errors := make(chan error, len(u.ins)) for _, in := range u.ins { - t := u.et.tm.TimingService.NewTimer() + t := u.et.tm.TimingService.NewTimer(u.avgExecVar) u.timers = append(u.timers, t) go func(e *Edge, t timer.Timer) { switch u.Wants() { @@ -76,17 +75,3 @@ func (u *UnionNode) runUnion([]byte) error { } return nil } - -func (u *UnionNode) nodeExecTime() time.Duration { - sum := 0.0 - total := 0.0 - for _, t := range u.timers { - avg, count := t.AverageTime() - sum += float64(avg) * float64(count) - total += float64(count) - } - if total == 0 { - return 0 - } - return time.Duration(sum / total) -}