feat: A new port is added for prometheus collection, and the indicator init…

collector/go.mod

@@ -15,7 +15,7 @@
 // specific language governing permissions and limitations
 // under the License.
 
-module github.com/pegasus-kv/collector
+module github.com/limowang/incubator-pegasus/collector
 
 go 1.18
 
@@ -98,4 +98,4 @@ require (
 	gopkg.in/ini.v1 v1.67.0 // indirect
 	gopkg.in/yaml.v2 v2.4.0 // indirect
 	gopkg.in/yaml.v3 v3.0.1 // indirect
-)
+)

collector/main.go

@@ -26,9 +26,12 @@ import (
 	"strings"
 	"syscall"
 
-	"github.com/pegasus-kv/collector/avail"
-	"github.com/pegasus-kv/collector/metrics"
-	"github.com/pegasus-kv/collector/webui"
+	// "github.com/pegasus-kv/collector/avail"
+	// "github.com/pegasus-kv/collector/metrics"
+	// "github.com/pegasus-kv/collector/webui"
+	"github.com/limowang/incubator-pegasus/collector/avail"
+	"github.com/limowang/incubator-pegasus/collector/metrics"
+	"github.com/limowang/incubator-pegasus/collector/webui"
 	log "github.com/sirupsen/logrus"
 	"github.com/spf13/viper"
 	"gopkg.in/natefinch/lumberjack.v2"
@@ -81,7 +84,8 @@ func main() {
 		return
 	}
 
-	webui.StartWebServer()
+	metrics.InitMetrics()
+	//webui.StartWebServer()
 
 	tom := &tomb.Tomb{}
 	setupSignalHandler(func() {
@@ -97,5 +101,7 @@ func main() {
 	tom.Go(func() error {
 		return metrics.NewReplicaServerMetricCollector().Start(tom)
 	})
+
+	webui.StartWebServer()
 	<-tom.Dead() // gracefully wait until all goroutines dead
 }

collector/metrics/metric_collector.go

@@ -67,22 +67,13 @@ func NewMetricCollector(
 	dataSource int,
 	detectInterval time.Duration,
 	detectTimeout time.Duration) MetricCollector {
-	DataSource = dataSource
-	GaugeMetricsMap = make(map[string]prometheus.GaugeVec, 128)
-	CounterMetricsMap = make(map[string]prometheus.CounterVec, 128)
-	SummaryMetricsMap = make(map[string]prometheus.Summary, 128)
-	RoleByDataSource = make(map[int]string, 128)
-	TableNameByID = make(map[string]string, 128)
-	RoleByDataSource[0] = "meta_server"
-	RoleByDataSource[1] = "replica_server"
-	initMetrics()
-
-	return &Collector{detectInterval: detectInterval, detectTimeout: detectTimeout}
+	return &Collector{detectInterval: detectInterval, detectTimeout: detectTimeout, dataSource: dataSource}
 }
 
 type Collector struct {
 	detectInterval time.Duration
 	detectTimeout  time.Duration
+	dataSource     int
 }
 
 func (collector *Collector) Start(tom *tomb.Tomb) error {
@@ -123,19 +114,10 @@ func getReplicaAddrs() ([]string, error) {
 	return rserverAddrs, nil
 }
 
-// Register all metrics.
-func initMetrics() {
-	var addrs []string
-	var err error
-	if DataSource == MetaServer {
-		addrs = viper.GetStringSlice("meta_servers")
-	} else {
-		addrs, err = getReplicaAddrs()
-		if err != nil {
-			log.Errorf("Get replica server address failed, err: %s", err)
-			return
-		}
-	}
+//Get metrics with new labels
+
+// Get all metrics of meta-server and replica-server by their addrs
+func getAllMetricsByAddrs(addrs []string) {
 	for _, addr := range addrs {
 		data, err := getOneServerMetrics(addr)
 		if err != nil {
@@ -167,17 +149,15 @@ func initMetrics() {
 						Help: desc,
 					}, []string{"endpoint", "role", "level", "title"})
 					GaugeMetricsMap[name] = *gaugeMetric
-				case "Percentile":
-					if _, ok := SummaryMetricsMap[name]; ok {
+				case "Percentile": //这个需要改动不能用这个表示,用gauge来表示分位数  --level(p50,p99),title(task_name)来替代区分
+					if _, ok := GaugeMetricsMap[name]; ok {
 						continue
 					}
-					summaryMetric := promauto.NewSummary(prometheus.SummaryOpts{
+					gaugeMetric := promauto.NewGaugeVec(prometheus.GaugeOpts{
 						Name: name,
 						Help: desc,
-						Objectives: map[float64]float64{
-							0.5: 0.05, 0.9: 0.01, 0.95: 0.005, 0.99: 0.001, 0.999: 0.0001},
-					})
-					SummaryMetricsMap[name] = summaryMetric
+					}, []string{"endpoint", "role", "level", "title"})
+					GaugeMetricsMap[name] = *gaugeMetric
 				case "Histogram":
 				default:
 					log.Errorf("Unsupport metric type %s", mtype)
@@ -187,6 +167,27 @@ func initMetrics() {
 	}
 }
 
+// Register all metrics.
+func InitMetrics() {
+	GaugeMetricsMap = make(map[string]prometheus.GaugeVec, 256)
+	CounterMetricsMap = make(map[string]prometheus.CounterVec, 256)
+	SummaryMetricsMap = make(map[string]prometheus.Summary, 256)
+	RoleByDataSource = make(map[int]string, 128)
+	TableNameByID = make(map[string]string, 256)
+	RoleByDataSource[0] = "meta_server"
+	RoleByDataSource[1] = "replica_server"
+
+	var addrs []string
+	addrs = viper.GetStringSlice("meta_servers")
+	replicAddrs, err := getReplicaAddrs()
+	if err != nil {
+		log.Errorf("Get raw metrics from %s failed, err: %s", replicAddrs, err)
+		return
+	}
+	addrs = append(addrs, replicAddrs...)
+	getAllMetricsByAddrs(addrs)
+}
+
 // Parse metric data and update metrics.
 func processAllServerMetrics() {
 	var addrs []string

collector/sink/falcon_sink.go

@@ -25,7 +25,7 @@ import (
 	"net/http"
 	"time"
 
-	"github.com/pegasus-kv/collector/aggregate"
+	"github.com/limowang/incubator-pegasus/collector/aggregate"
 	log "github.com/sirupsen/logrus"
 	"github.com/spf13/viper"
 )

collector/sink/prometheus_sink.go

@@ -20,7 +20,7 @@ package sink
 import (
 	"sync"
 
-	"github.com/pegasus-kv/collector/aggregate"
+	"github.com/limowang/incubator-pegasus/collector/aggregate"
 	"github.com/prometheus/client_golang/prometheus"
 )
 

collector/sink/sink.go

@@ -18,7 +18,7 @@
 package sink
 
 import (
-	"github.com/pegasus-kv/collector/aggregate"
+	"github.com/limowang/incubator-pegasus/collector/aggregate"
 	log "github.com/sirupsen/logrus"
 	"github.com/spf13/viper"
 )

collector/usage/usage_recorder.go

@@ -23,7 +23,7 @@ import (
 	"time"
 
 	"github.com/apache/incubator-pegasus/go-client/pegasus"
-	"github.com/pegasus-kv/collector/aggregate"
+	"github.com/limowang/incubator-pegasus/collector/aggregate"
 	log "github.com/sirupsen/logrus"
 	"github.com/spf13/viper"
 	"gopkg.in/tomb.v2"

collector/webui/index.go

@@ -19,7 +19,7 @@ package webui
 
 import (
 	"github.com/kataras/iris/v12"
-	"github.com/pegasus-kv/collector/aggregate"
+	"github.com/limowang/incubator-pegasus/collector/aggregate"
 )
 
 var indexPageClusterStats = []string{

collector/webui/webserver.go

@@ -19,9 +19,12 @@ package webui
 
 import (
 	"context"
+	"net/http"
 	"time"
 
 	"github.com/kataras/iris/v12"
+	"github.com/limowang/incubator-pegasus/collector/metrics"
+	"github.com/prometheus/client_golang/prometheus"
 	"github.com/prometheus/client_golang/prometheus/promhttp"
 )
 
@@ -53,9 +56,22 @@ func StartWebServer() {
 	app.RegisterView(tmpl)
 
 	go func() {
-		err := app.Listen(":8080")
+		err := app.Listen(":8081")
 		if err != nil {
 			return
 		}
 	}()
+
+	//Provide metrics for prometheus
+	registry := prometheus.NewRegistry()
+	for _, cV := range metrics.CounterMetricsMap {
+		registry.MustRegister(cV)
+	}
+	for _, gV := range metrics.GaugeMetricsMap {
+		registry.MustRegister(gV)
+	}
+
+	http.Handle("/metrics", promhttp.Handler())
+
+	_ = http.ListenAndServe(":8080", nil)
 }