Merge pull request #18 from savitaashture/flush

Added changes to flush the metrics data after a period of interval

Author: tianxiaoliang

cse_collector.go

@@ -77,6 +77,14 @@ func (c *CseCollector) updateTimerMetric(prefix string, dur time.Duration) {
 	count.Update(dur)
 }
 
+func (c *CseCollector) cleanMetric(prefix string) {
+	count, ok := metrics.GetOrRegister(prefix, metrics.NewCounter).(metrics.Counter)
+	if !ok {
+		return
+	}
+	count.Clear()
+}
+
 // IncrementAttempts function increments the number of calls to this circuit.
 // This registers as a counter
 func (c *CseCollector) IncrementAttempts() {
@@ -148,4 +156,14 @@ func (c *CseCollector) UpdateRunDuration(runDuration time.Duration) {
 }
 
 // Reset function is a noop operation in this collector.
-func (c *CseCollector) Reset() {}
+func (c *CseCollector) Reset() {
+	c.cleanMetric(c.attempts)
+	c.cleanMetric(c.failures)
+	c.cleanMetric(c.successes)
+	c.cleanMetric(c.shortCircuits)
+	c.cleanMetric(c.errors)
+	c.cleanMetric(c.rejects)
+	c.cleanMetric(c.timeouts)
+	c.cleanMetric(c.fallbackSuccesses)
+	c.cleanMetric(c.fallbackFailures)
+}

csemonitor.go

@@ -11,6 +11,7 @@ import (
 	"github.com/ServiceComb/go-chassis/core/config"
 	"github.com/ServiceComb/go-chassis/core/lager"
 	"github.com/ServiceComb/go-chassis/core/registry"
+	"github.com/ServiceComb/go-chassis/third_party/forked/afex/hystrix-go/hystrix"
 	"github.com/rcrowley/go-metrics"
 )
 
@@ -51,6 +52,20 @@ func NewReporter(r metrics.Registry, addr string, header http.Header, interval t
 
 // Run creates a go_routine which runs continuously and capture the monitoring data
 func (reporter *Reporter) Run() {
+
+	go reporter.postData()
+	go reporter.cleanData()
+
+}
+
+func (reporter *Reporter) cleanData() {
+	ticker := time.Tick(20 * time.Second)
+	for range ticker {
+		hystrix.Flush()
+	}
+}
+
+func (reporter *Reporter) postData() {
 	ticker := time.Tick(reporter.Interval)
 	metricsAPI := NewCseMonitorClient(reporter.Header, reporter.CseMonitorAddr, reporter.TLSConfig, "v2")
 	IsMonitoringConnected = true

monitorData.go

@@ -1,14 +1,15 @@
 package metricsink
 
 import (
+	"fmt"
+	"math"
 	"runtime"
 	"runtime/pprof"
+	"strconv"
 	"strings"
 	"time"
 
-	"fmt"
 	"github.com/rcrowley/go-metrics"
-	"strconv"
 )
 
 var threadCreateProfile = pprof.Lookup("threadcreate")
@@ -73,6 +74,7 @@ func (monitorData *MonitorData) getOrCreateInterfaceInfo(name string) *Interface
 	monitorData.Interfaces = append(monitorData.Interfaces, interfaceInfo)
 	return interfaceInfo
 }
+
 func (monitorData *MonitorData) appendInterfaceInfo(name string, i interface{}) {
 	var interfaceInfo = monitorData.getOrCreateInterfaceInfo(getInterfaceName(name))
 	switch metric := i.(type) {
@@ -87,6 +89,15 @@ func (monitorData *MonitorData) appendInterfaceInfo(name string, i interface{})
 		case "successes":
 			interfaceInfo.successCount = metric.Count()
 		}
+
+		qps := (float64(interfaceInfo.Total) * (1 - math.Exp(-5.0/60.0/1)))
+		movingAverageFor3Precision, err := strconv.ParseFloat(fmt.Sprintf("%.3f", qps), 64)
+		if err == nil {
+			interfaceInfo.QPS = movingAverageFor3Precision
+		} else {
+			interfaceInfo.QPS = 0
+		}
+
 	case metrics.Timer:
 		t := metric.Snapshot()
 		ps := t.Percentiles([]float64{0.05, 0.25, 0.5, 0.75, 0.90, 0.99, 0.995})
@@ -100,12 +111,6 @@ func (monitorData *MonitorData) appendInterfaceInfo(name string, i interface{})
 			interfaceInfo.L99 = int(ps[5] / float64(time.Millisecond))
 			interfaceInfo.L995 = int(ps[6] / float64(time.Millisecond))
 			interfaceInfo.Latency = int(t.Mean() / float64(time.Millisecond))
-			movingAverageFor3Precision, err := strconv.ParseFloat(fmt.Sprintf("%.3f", t.Rate1()), 64)
-			if err == nil {
-				interfaceInfo.QPS = movingAverageFor3Precision
-			} else {
-				interfaceInfo.QPS = 0
-			}
 		}
 	}
 	if interfaceInfo.Total == 0 {