Add metric exporter abstraction and add Prometheus support (#382)

* Add metric exporter abstraction
* Support Prometheus metrics

api/init.go

@@ -16,10 +16,13 @@ package api
 
 import (
  "fmt"
+ "net"
+ "net/http"
 
  "github.com/alibaba/sentinel-golang/core/config"
  "github.com/alibaba/sentinel-golang/core/log/metric"
  "github.com/alibaba/sentinel-golang/core/system_metric"
+ metric_exporter "github.com/alibaba/sentinel-golang/exporter/metric"
  "github.com/alibaba/sentinel-golang/util"
  "github.com/pkg/errors"
 )
@@ -113,6 +116,23 @@ func initCoreComponents() error {
  util.StartTimeTicker()
  }
 
+ if config.MetricExportHTTPAddr() != "" {
+ httpAddr := config.MetricExportHTTPAddr()
+ httpPath := config.MetricExportHTTPPath()
+
+ l, err := net.Listen("tcp", httpAddr)
+ if err != nil {
+ return fmt.Errorf("init metric exporter http server err: %s", err.Error())
+ }
+
+ http.Handle(httpPath, metric_exporter.HTTPHandler())
+ go func() {
+ _ = http.Serve(l, nil)
+ }()
+
+ return nil
+ }
+
  return nil
 }
 

core/circuitbreaker/circuit_breaker.go

@@ -20,6 +20,7 @@ import (
 
  "github.com/alibaba/sentinel-golang/core/base"
  sbase "github.com/alibaba/sentinel-golang/core/stat/base"
+ metric_exporter "github.com/alibaba/sentinel-golang/exporter/metric"
  "github.com/alibaba/sentinel-golang/logging"
  "github.com/alibaba/sentinel-golang/util"
  "github.com/pkg/errors"
@@ -47,6 +48,17 @@ const (
  Open
 )
 
+var (
+ stateChangedCounter = metric_exporter.NewCounter(
+ "circuit_breaker_state_changed_total",
+ "Circuit breaker total state change count",
+ []string{"resource", "from_state", "to_state"})
+)
+
+func init() {
+ metric_exporter.Register(stateChangedCounter)
+}
+
 func newState() *State {
  var state State
  state = Closed
@@ -152,6 +164,8 @@ func (b *circuitBreakerBase) fromClosedToOpen(snapshot interface{}) bool {
  for _, listener := range stateChangeListeners {
  listener.OnTransformToOpen(Closed, *b.rule, snapshot)
  }
+
+ stateChangedCounter.Add(float64(1), b.BoundRule().Resource, "Closed", "Open")
  return true
  }
  return false
@@ -181,6 +195,8 @@ func (b *circuitBreakerBase) fromOpenToHalfOpen(ctx *base.EntryContext) bool {
  return nil
  })
  }
+
+ stateChangedCounter.Add(float64(1), b.BoundRule().Resource, "Open", "HalfOpen")
  return true
  }
  return false
@@ -194,6 +210,8 @@ func (b *circuitBreakerBase) fromHalfOpenToOpen(snapshot interface{}) bool {
  for _, listener := range stateChangeListeners {
  listener.OnTransformToOpen(HalfOpen, *b.rule, snapshot)
  }
+
+ stateChangedCounter.Add(float64(1), b.BoundRule().Resource, "HalfOpen", "Open")
  return true
  }
  return false
@@ -206,6 +224,8 @@ func (b *circuitBreakerBase) fromHalfOpenToClosed() bool {
  for _, listener := range stateChangeListeners {
  listener.OnTransformToClosed(HalfOpen, *b.rule)
  }
+
+ stateChangedCounter.Add(float64(1), b.BoundRule().Resource, "HalfOpen", "Closed")
  return true
  }
  return false

core/config/config.go

@@ -204,6 +204,14 @@ func LogUsePid() bool {
  return globalCfg.LogUsePid()
 }
 
+func MetricExportHTTPAddr() string {
+ return globalCfg.MetricExportHTTPAddr()
+}
+
+func MetricExportHTTPPath() string {
+ return globalCfg.MetricExportHTTPPath()
+}
+
 func MetricLogFlushIntervalSec() uint32 {
  return globalCfg.MetricLogFlushIntervalSec()
 }

core/config/entity.go

@@ -38,6 +38,8 @@ type SentinelConfig struct {
  // Type indicates the classification of the service (e.g. web service, API gateway).
  Type int32
  }
+ // Exporter represents configuration items related to exporter, like metric exporter.
+ Exporter ExporterConfig
  // Log represents configuration items related to logging.
  Log LogConfig
  // Stat represents configuration items related to statistics.
@@ -46,6 +48,19 @@ type SentinelConfig struct {
  UseCacheTime bool `yaml:"useCacheTime"`
 }
 
+// ExporterConfig represents configuration items related to exporter, like metric exporter.
+type ExporterConfig struct {
+ Metric MetricExporterConfig
+}
+
+// MetricExporterConfig represents configuration of metric exporter.
+type MetricExporterConfig struct {
+ // HttpAddr is the http server listen address, like ":8080".
+ HttpAddr string `yaml:"http_addr"`
+ // HttpPath is the http request path of access metrics, like "/metrics".
+ HttpPath string `yaml:"http_path"`
+}
+
 // LogConfig represent the configuration of logging in Sentinel.
 type LogConfig struct {
  // Logger indicates that using logger to replace default logging.
@@ -190,6 +205,14 @@ func (entity *Entity) LogUsePid() bool {
  return entity.Sentinel.Log.UsePid
 }
 
+func (entity *Entity) MetricExportHTTPAddr() string {
+ return entity.Sentinel.Exporter.Metric.HttpAddr
+}
+
+func (entity *Entity) MetricExportHTTPPath() string {
+ return entity.Sentinel.Exporter.Metric.HttpPath
+}
+
 func (entity *Entity) MetricLogFlushIntervalSec() uint32 {
  return entity.Sentinel.Log.Metric.FlushIntervalSec
 }

core/flow/traffic_shaping.go

@@ -16,9 +16,20 @@ package flow
 
 import (
  "github.com/alibaba/sentinel-golang/core/base"
- "github.com/alibaba/sentinel-golang/metrics"
+ metric_exporter "github.com/alibaba/sentinel-golang/exporter/metric"
 )
 
+var (
+ resourceFlowThresholdGauge = metric_exporter.NewGauge(
+ "resource_flow_threshold",
+ "Resource flow threshold",
+ []string{"resource"})
+)
+
+func init() {
+ metric_exporter.Register(resourceFlowThresholdGauge)
+}
+
 // TrafficShapingCalculator calculates the actual traffic shaping threshold
 // based on the threshold of rule and the traffic shaping strategy.
 type TrafficShapingCalculator interface {
@@ -74,6 +85,8 @@ func (t *TrafficShapingController) FlowCalculator() TrafficShapingCalculator {
 
 func (t *TrafficShapingController) PerformChecking(resStat base.StatNode, batchCount uint32, flag int32) *base.TokenResult {
  allowedTokens := t.flowCalculator.CalculateAllowedTokens(batchCount, flag)
- metrics.SetResourceFlowThreshold(t.rule.Resource, allowedTokens)
+
+ resourceFlowThresholdGauge.Set(float64(allowedTokens), t.rule.Resource)
+
  return t.flowChecker.DoCheck(resStat, batchCount, allowedTokens)
 }

core/stat/stat_slot.go

@@ -16,17 +16,29 @@ package stat
 
 import (
  "github.com/alibaba/sentinel-golang/core/base"
+ metric_exporter "github.com/alibaba/sentinel-golang/exporter/metric"
  "github.com/alibaba/sentinel-golang/util"
 )
 
 const (
  StatSlotOrder = 1000
+ ResultPass = "pass"
+ ResultBlock = "block"
 )
 
 var (
  DefaultSlot = &Slot{}
+
+ handledCounter = metric_exporter.NewCounter(
+ "handled_total",
+ "Total handled count",
+ []string{"resource", "result", "block_type"})
 )
 
+func init() {
+ metric_exporter.Register(handledCounter)
+}
+
 type Slot struct {
 }
 
@@ -39,13 +51,17 @@ func (s *Slot) OnEntryPassed(ctx *base.EntryContext) {
  if ctx.Resource.FlowType() == base.Inbound {
  s.recordPassFor(InboundNode(), ctx.Input.BatchCount)
  }
+
+ handledCounter.Add(float64(ctx.Input.BatchCount), ctx.Resource.Name(), ResultPass, "")
 }
 
 func (s *Slot) OnEntryBlocked(ctx *base.EntryContext, blockError *base.BlockError) {
  s.recordBlockFor(ctx.StatNode, ctx.Input.BatchCount)
  if ctx.Resource.FlowType() == base.Inbound {
  s.recordBlockFor(InboundNode(), ctx.Input.BatchCount)
  }
+
+ handledCounter.Add(float64(ctx.Input.BatchCount), ctx.Resource.Name(), ResultBlock, blockError.BlockType().String())
 }
 
 func (s *Slot) OnCompleted(ctx *base.EntryContext) {

core/system_metric/sys_metric_stat.go

@@ -20,8 +20,8 @@ import (
  "sync/atomic"
  "time"
 
+ metric_exporter "github.com/alibaba/sentinel-golang/exporter/metric"
  "github.com/alibaba/sentinel-golang/logging"
- "github.com/alibaba/sentinel-golang/metrics"
  "github.com/alibaba/sentinel-golang/util"
  "github.com/shirou/gopsutil/v3/load"
  "github.com/shirou/gopsutil/v3/mem"
@@ -49,6 +49,15 @@ var (
  TotalMemorySize = getTotalMemorySize()
 
  ssStopChan = make(chan struct{})
+
+ cpuRatioGauge = metric_exporter.NewGauge(
+ "cpu_ratio",
+ "Process cpu ratio",
+ []string{})
+ processMemoryGauge = metric_exporter.NewGauge(
+ "process_memory_bytes",
+ "Process memory in bytes",
+ []string{})
 )
 
 func init() {
@@ -64,6 +73,9 @@ func init() {
  currentProcessOnce.Do(func() {
  currentProcess.Store(p)
  })
+
+ metric_exporter.Register(cpuRatioGauge)
+ metric_exporter.Register(processMemoryGauge)
 }
 
 // getMemoryStat returns the current machine's memory statistic
@@ -105,7 +117,9 @@ func retrieveAndUpdateMemoryStat() {
  logging.Error(err, "Fail to retrieve and update cpu statistic")
  return
  }
- metrics.SetProcessMemorySize(memoryUsedBytes)
+
+ processMemoryGauge.Set(float64(memoryUsedBytes))
+
  currentMemoryUsage.Store(memoryUsedBytes)
 }
 
@@ -161,7 +175,9 @@ func retrieveAndUpdateCpuStat() {
  logging.Error(err, "Fail to retrieve and update cpu statistic")
  return
  }
- metrics.SetCPURatio(cpuPercent)
+
+ cpuRatioGauge.Set(cpuPercent)
+
  currentCpuUsage.Store(cpuPercent)
 }