refactor

Author: tianxiaoliang

cse_collector.go

@@ -0,0 +1,76 @@
+package metricsink
+
+// Forked from github.com/afex/hystrix-go
+// Some parts of this file have been modified to make it functional in this package
+
+import (
+	"crypto/tls"
+	"github.com/go-chassis/go-chassis/core/common"
+	"github.com/go-chassis/go-chassis/third_party/forked/afex/hystrix-go/hystrix"
+	"github.com/go-mesh/openlogging"
+	"net/http"
+	"sync"
+	"time"
+)
+
+var initOnce = sync.Once{}
+
+// CseCollectorConfig is a struct to keep monitoring information
+type CseCollectorConfig struct {
+	// CseMonitorAddr is the http address of the csemonitor server
+	CseMonitorAddr string
+	// Headers for csemonitor server
+	Header http.Header
+	// TickInterval specifies the period that this collector will send metrics to the server.
+	TimeInterval time.Duration
+	// Config structure to configure a TLS client for sending Metric data
+	TLSConfig *tls.Config
+
+	Env string
+}
+
+func init() {
+	hystrix.InstallReporter("CSE Monitoring", reportMetricsToCSEDashboard)
+}
+
+var reporter *Reporter
+
+//GetReporter get reporter singleton
+func GetReporter() (*Reporter, error) {
+	var errResult error
+	initOnce.Do(func() {
+		monitorServerURL, err := getMonitorEndpoint()
+		if err != nil {
+			openlogging.GetLogger().Warnf("Get Monitoring URL failed, CSE monitoring function disabled, err: %v", err)
+			errResult = err
+		}
+		openlogging.GetLogger().Infof("send metric to monitoring service : %s", monitorServerURL)
+		tlsConfig, err := getTLSForClient(monitorServerURL)
+		if err != nil {
+			openlogging.GetLogger().Errorf("Get %s.%s TLS config failed,error : %s", monitorServerURL, common.Consumer, err)
+			errResult = err
+		}
+		reporter, err = NewReporter(&CseCollectorConfig{
+			CseMonitorAddr: monitorServerURL,
+			Header:         getAuthHeaders(),
+			TLSConfig:      tlsConfig,
+		})
+		if err != nil {
+			openlogging.Error("new reporter failed", openlogging.WithTags(openlogging.Tags{
+				"err": err.Error(),
+			}))
+			errResult = err
+		}
+	})
+	return reporter, errResult
+}
+
+//reportMetricsToCSEDashboard use send metrics to cse dashboard
+func reportMetricsToCSEDashboard(cb *hystrix.CircuitBreaker) error {
+	r, err := GetReporter()
+	if err != nil {
+		return err
+	}
+	r.Send(cb)
+	return nil
+}

cse_reporter.go

@@ -1,15 +1,12 @@
 package metricsink
 
 import (
-	"crypto/tls"
-	"net/http"
-	"os"
-	"time"
-
 	"github.com/go-chassis/go-archaius"
 	chassisRuntime "github.com/go-chassis/go-chassis/pkg/runtime"
 	"github.com/go-chassis/go-chassis/third_party/forked/afex/hystrix-go/hystrix"
 	"github.com/go-mesh/openlogging"
+	"github.com/huaweicse/cse-collector/pkg/monitoring"
+	"os"
 	"runtime"
 )
 
@@ -18,71 +15,51 @@ var IsMonitoringConnected bool
 
 // Reporter is a struct to store the registry address and different monitoring information
 type Reporter struct {
-	CseMonitorAddr string
-	Header         http.Header
-	Interval       time.Duration
-	Percentiles    []float64
-	TLSConfig      *tls.Config
-	app            string
-	version        string
-	service        string
-	environment    string
-	serviceID      string
-	metricsAPI     *CseMonitorClient
+	environment string
+	c           *monitoring.CseMonitorClient
 }
 
 // NewReporter creates a new monitoring object for CSE type collections
-func NewReporter(addr string, header http.Header, interval time.Duration, tls *tls.Config, app, version, service, env string) *Reporter {
-	reporter := &Reporter{
-		CseMonitorAddr: addr,
-		Header:         header,
-		Interval:       interval,
-		Percentiles:    []float64{0.5, 0.75, 0.95, 0.99, 0.999},
-		TLSConfig:      tls,
-		app:            app,
-		version:        version,
-		service:        service,
-		environment:    env,
-	}
-	metricsAPI, err := NewCseMonitorClient(reporter.Header, reporter.CseMonitorAddr, reporter.TLSConfig, "v2")
+func NewReporter(config *CseCollectorConfig) (*Reporter, error) {
+	c, err := monitoring.NewCseMonitorClient(config.Header, config.CseMonitorAddr, config.TLSConfig)
 	if err != nil {
 		openlogging.GetLogger().Errorf("Get cse monitor client failed:%s", err)
+		return nil, err
+	}
+	reporter := &Reporter{
+		environment: config.Env,
 	}
-	reporter.metricsAPI = metricsAPI
 	IsMonitoringConnected = true
-	return reporter
+	return &Reporter{
+		environment: config.Env,
+		c:           c,
+	}, nil
 }
 
-// Run creates a go_routine which runs continuously and capture the monitoring data
-func (reporter *Reporter) Run(cb *hystrix.CircuitBreaker) {
-	ticker := time.Tick(reporter.Interval)
-
-	for range ticker {
-		if archaius.GetBool("cse.monitor.client.enable", true) {
-			reporter.serviceID = chassisRuntime.ServiceID
-			monitorData := reporter.getData(cb, reporter.app, reporter.version,
-				reporter.service, reporter.environment, reporter.serviceID, chassisRuntime.InstanceID)
-			err := reporter.metricsAPI.PostMetrics(monitorData)
-			if err != nil {
-				openlogging.GetLogger().Warnf("Unable to report to monitoring server, err: %v", err)
-			}
+//Send send metrics to monitoring service
+func (reporter *Reporter) Send(cb *hystrix.CircuitBreaker) {
+	if archaius.GetBool("cse.monitor.client.enable", true) {
+		monitorData := reporter.getData(cb)
+		err := reporter.c.PostMetrics(monitorData)
+		if err != nil {
+			openlogging.GetLogger().Warnf("unable to report to monitoring server, err: %v", err)
 		}
 	}
 }
 
-func (reporter *Reporter) getData(cb *hystrix.CircuitBreaker,
-	app, version, service, env, serviceID, instanceID string) MonitorData {
-	var monitorData = NewMonitorData()
-	monitorData.AppID = app
-	monitorData.Version = version
-	monitorData.Name = service
-	monitorData.ServiceID = serviceID
-	monitorData.InstanceID = instanceID
-	monitorData.Environment = env
+func (reporter *Reporter) getData(cb *hystrix.CircuitBreaker) monitoring.MonitorData {
+	var monitorData = monitoring.NewMonitorData()
+	monitorData.AppID = chassisRuntime.App
+	monitorData.Version = chassisRuntime.Version
+	monitorData.Name = chassisRuntime.ServiceName
+	monitorData.ServiceID = chassisRuntime.ServiceID
+	monitorData.InstanceID = chassisRuntime.InstanceID
+
+	monitorData.Environment = reporter.environment
 	monitorData.Instance, _ = os.Hostname()
 	monitorData.Memory = getProcessInfo()
 	monitorData.Thread = threadCreateProfile.Count()
 	monitorData.CPU = float64(runtime.NumCPU())
-	monitorData.appendInterfaceInfo(cb.Name, cb.Metrics.DefaultCollector())
+	monitorData.AppendInterfaceInfo(cb.Name, cb.Metrics.DefaultCollector())
 	return *monitorData
 }

csemonitor.go

@@ -3,6 +3,6 @@ module github.com/huaweicse/cse-collector
 require (
 	github.com/go-chassis/foundation v0.0.0-20190621030543-c3b63f787f4c
 	github.com/go-chassis/go-archaius v0.19.0
-	github.com/go-chassis/go-chassis v1.5.2-0.20190713094014-10193b3a09b8
+	github.com/go-chassis/go-chassis v1.6.0
 	github.com/go-mesh/openlogging v1.0.0
 )

go.mod

@@ -1,145 +1,13 @@
 package metricsink
 
 import (
-	"fmt"
-	"math"
 	"runtime"
 	"runtime/pprof"
-	"strconv"
 	"strings"
-	"time"
-
-	"github.com/go-chassis/go-chassis/third_party/forked/afex/hystrix-go/hystrix"
-	"github.com/go-chassis/go-chassis/third_party/forked/afex/hystrix-go/hystrix/metric_collector"
-	"github.com/go-mesh/openlogging"
 )
 
 var threadCreateProfile = pprof.Lookup("threadcreate")
 
-// MonitorData is an object which stores the monitoring information for an application
-type MonitorData struct {
-	AppID       string                 `json:"appId"`
-	Version     string                 `json:"version"`
-	Name        string                 `json:"name"`
-	Environment string                 `json:"environment"`
-	Instance    string                 `json:"instance"`
-	Thread      int                    `json:"thread"`
-	Customs     map[string]interface{} `json:"customs"` // ?
-	Interfaces  []*InterfaceInfo       `json:"interfaces"`
-	CPU         float64                `json:"cpu"`
-	Memory      map[string]interface{} `json:"memory"`
-	ServiceID   string                 `json:"serviceId"`
-	InstanceID  string                 `json:"instanceId"`
-}
-
-// InterfaceInfo is an object which store the monitoring information of a particular interface
-type InterfaceInfo struct {
-	Name                 string  `json:"name"`
-	Desc                 string  `json:"desc"`
-	QPS                  float64 `json:"qps"`
-	Latency              int     `json:"latency"`
-	L995                 int     `json:"l995"`
-	L99                  int     `json:"l99"`
-	L90                  int     `json:"l90"`
-	L75                  int     `json:"l75"`
-	L50                  int     `json:"l50"`
-	L25                  int     `json:"l25"`
-	L5                   int     `json:"l5"`
-	Rate                 float64 `json:"rate"`
-	Total                int64   `json:"total"`
-	Failure              int64   `json:"failure"`
-	ShortCircuited       int64   `json:"shortCircuited"`
-	IsCircuitBreakerOpen bool    `json:"circuitBreakerOpen"`
-	SemaphoreRejected    int64   `json:"semaphoreRejected"`
-	ThreadPoolRejected   int64   `json:"threadPoolRejected"`
-	CountTimeout         int64   `json:"countTimeout"`
-	FailureRate          float64 `json:"failureRate"`
-	successCount         int64
-}
-
-// NewMonitorData creates a new monitoring object
-func NewMonitorData() *MonitorData {
-	monitorData := new(MonitorData)
-	monitorData.Interfaces = make([]*InterfaceInfo, 0)
-	return monitorData
-}
-
-func (monitorData *MonitorData) getOrCreateInterfaceInfo(name string) *InterfaceInfo {
-	for _, interfaceInfo := range monitorData.Interfaces {
-		if interfaceInfo.Name == name {
-			return interfaceInfo
-		}
-	}
-	interfaceInfo := new(InterfaceInfo)
-	interfaceInfo.Name = name
-	interfaceInfo.Desc = name
-	monitorData.Interfaces = append(monitorData.Interfaces, interfaceInfo)
-	return interfaceInfo
-}
-
-func (monitorData *MonitorData) appendInterfaceInfo(name string, c *metricCollector.DefaultMetricCollector) {
-	var interfaceInfo = monitorData.getOrCreateInterfaceInfo(name)
-	now := time.Now()
-	//attempts:
-	interfaceInfo.Total = int64(c.NumRequests().Sum(now))
-	//errors
-	interfaceInfo.Failure = int64(c.Failures().Sum(now))
-	//shortCircuits
-	interfaceInfo.ShortCircuited = int64(c.ShortCircuits().Sum(now))
-	//successes
-	interfaceInfo.successCount = int64(c.Successes().Sum(now))
-
-	if isCBOpen, err := hystrix.IsCircuitBreakerOpen(name); err != nil {
-		interfaceInfo.IsCircuitBreakerOpen = false
-		openlogging.Error("can't get circuit status", openlogging.WithTags(openlogging.Tags{
-			"err":  err.Error(),
-			"name": name,
-		}))
-	} else {
-		interfaceInfo.IsCircuitBreakerOpen = isCBOpen
-	}
-
-	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
-	}
-	runDuration := c.RunDuration()
-	interfaceInfo.L5 = int(runDuration.Percentile(0.05))
-	interfaceInfo.L25 = int(runDuration.Percentile(0.25))
-	interfaceInfo.L50 = int(float64(runDuration.Percentile(0.5)))
-	interfaceInfo.L75 = int(runDuration.Percentile(0.75))
-	interfaceInfo.L90 = int(runDuration.Percentile(0.90))
-	interfaceInfo.L99 = int(runDuration.Percentile(0.99))
-	interfaceInfo.L995 = int(runDuration.Percentile(0.995))
-	interfaceInfo.Latency = int(runDuration.Mean())
-	interfaceInfo.Rate = 1 //rate is no use any more and must be set to 1
-	if interfaceInfo.Total == 0 {
-		interfaceInfo.FailureRate = 0
-	} else {
-		totalErrorCount := interfaceInfo.Failure + interfaceInfo.SemaphoreRejected + interfaceInfo.ThreadPoolRejected + interfaceInfo.CountTimeout
-		if totalErrorCount == 0 {
-			interfaceInfo.FailureRate = 0
-		} else {
-			failureRate, err := strconv.ParseFloat(fmt.Sprintf("%.3f", float64(totalErrorCount)/float64(interfaceInfo.Total)), 64)
-			if err == nil && failureRate > 0 {
-				interfaceInfo.FailureRate = failureRate
-			} else {
-				openlogging.GetLogger().Warnf("Error in calculating the failureRate %v, default value(0) is assigned to failureRate", failureRate)
-				interfaceInfo.FailureRate = 0
-			}
-		}
-	}
-}
-
-func GetInterfaceName(metricName string) string {
-	command := strings.Split(metricName, ".")
-	return strings.Join(command[:len(command)-1], ".")
-
-}
-
 func getProcessInfo() map[string]interface{} {
 	var memoryInfo = make(map[string]interface{})
 	var memStats = runtime.MemStats{}