report sample queue size (#1027)

* report sample queue size

* report sample channel size

src/server/stat/stat.go

@@ -49,6 +49,14 @@ var (
 		Name:      "alert_queue_size",
 		Help:      "The size of alert queue.",
 	}, []string{"cluster"})
+
+	// 数据转发队列，各个队列的长度
+	GaugeSampleQueueSize = prometheus.NewGaugeVec(prometheus.GaugeOpts{
+		Namespace: namespace,
+		Subsystem: subsystem,
+		Name:      "sample_queue_size",
+		Help:      "The size of sample queue.",
+	}, []string{"cluster", "channel_number"})
 )
 
 func Init() {

src/server/writer/writer.go

@@ -16,6 +16,8 @@ import (
 	"github.com/prometheus/client_golang/api"
 	"github.com/prometheus/prometheus/prompb"
 	"github.com/toolkits/pkg/logger"
+
+	promstat "github.com/didi/nightingale/v5/src/server/stat"
 )
 
 type WriterType struct {
@@ -192,6 +194,8 @@ func Init(opts []config.WriterOptions, globalOpt config.WriterGlobalOpt) error {
 		go Writers.StartConsumer(i, Writers.chans[i])
 	}
 
+	go reportChanSize()
+
 	for i := 0; i < len(opts); i++ {
 		cli, err := api.NewClient(api.Config{
 			Address: opts[i].Url,
@@ -226,3 +230,13 @@ func Init(opts []config.WriterOptions, globalOpt config.WriterGlobalOpt) error {
 
 	return nil
 }
+
+func reportChanSize() {
+	for {
+		time.Sleep(time.Second * 3)
+		for i, c := range Writers.chans {
+			size := len(c)
+			promstat.GaugeSampleQueueSize.WithLabelValues(config.C.ClusterName, fmt.Sprint(i)).Set(float64(size))
+		}
+	}
+}