Adding host queue size metric

bookingcom · Jun 11, 2024 · e2cde4e · e2cde4e
1 parent e1f9f58
commit e2cde4e
Showing 3 changed files with 17 additions and 0 deletions.
diff --git a/pkg/metrics/metrics.go b/pkg/metrics/metrics.go
@@ -36,6 +36,7 @@ type Prom struct {
 	TargetStates             *prometheus.GaugeVec
 	NumberOfAvailableTargets *prometheus.GaugeVec
 	NumberOfTargets          *prometheus.GaugeVec
+	HostQueueSize            *prometheus.GaugeVec
 
 	GlobalRateLimiterBlockedReaders    prometheus.Counter
 	ContainerRateLimiterBlockedReaders *prometheus.CounterVec
@@ -144,6 +145,11 @@ func New() *Prom {
 			Name:      "target_states",
 			Help:      "The current states of target hosts.",
 		}, []string{"upstream_host", "cluster"}),
+		HostQueueSize: prometheus.NewGaugeVec(prometheus.GaugeOpts{
+			Namespace: "nanotube",
+			Name:      "host_queue_size",
+			Help:      "The current size of host queue.",
+		}, []string{"upstream_host", "cluster"}),
 		NumberOfAvailableTargets: prometheus.NewGaugeVec(prometheus.GaugeOpts{
 			Namespace: "nanotube",
 			Name:      "number_of_available_targets",
@@ -311,6 +317,11 @@ func Register(m *Prom, cfg *conf.Main) {
 			log.Fatalf("error while registering target_states metric: %v", err)
 		}
 
+		err = prometheus.Register(m.HostQueueSize)
+		if err != nil {
+			log.Fatalf("error while registering target_states metric: %v", err)
+		}
+
 		err = prometheus.Register(m.NumberOfAvailableTargets)
 		if err != nil {
 			log.Fatalf("error while registering number_of_available_targets metric: %v", err)

diff --git a/pkg/target/host.go b/pkg/target/host.go
@@ -45,6 +45,7 @@ type Host struct {
 	stateChangesTotal         prometheus.Counter
 	oldConnectionRefresh      prometheus.Counter
 	oldConnectionRefreshTotal prometheus.Counter
+	hostQueueSize             prometheus.Gauge
 	targetState               prometheus.Gauge
 }
 
@@ -120,6 +121,7 @@ func ConstructHost(clusterName string, mainCfg conf.Main, hostCfg conf.Host, lg
 		stateChangesTotal:         ms.StateChangeHostsTotal,
 		oldConnectionRefresh:      ms.OldConnectionRefresh.With(promLabels),
 		oldConnectionRefreshTotal: ms.OldConnectionRefreshTotal,
+		hostQueueSize:             ms.HostQueueSize.With(promLabels),
 		targetState:               ms.TargetStates.With(promLabels),
 	}
 	h.rnd = rand.New(rand.NewSource(time.Now().UnixNano()))

diff --git a/pkg/target/hostMTCP.go b/pkg/target/hostMTCP.go
@@ -46,6 +46,7 @@ type HostMTCP struct {
 	stateChangesTotal         prometheus.Counter
 	oldConnectionRefresh      prometheus.Counter
 	oldConnectionRefreshTotal prometheus.Counter
+	hostQueueSize             prometheus.Gauge
 	targetState               prometheus.Gauge
 }
 
@@ -119,6 +120,7 @@ func ConstructHostMTCP(clusterName string, mainCfg conf.Main, hostCfg conf.Host,
 		stateChangesTotal:         ms.StateChangeHostsTotal,
 		oldConnectionRefresh:      ms.OldConnectionRefresh.With(promLabels),
 		oldConnectionRefreshTotal: ms.OldConnectionRefreshTotal,
+		hostQueueSize:             ms.HostQueueSize.With(promLabels),
 		targetState:               ms.TargetStates.With(promLabels),
 	}
 	h.NumMTCP = hostCfg.MTCP
@@ -260,6 +262,7 @@ func (h *HostMTCP) flush(d time.Duration) {
 }
 
 // Requires h.Conn.Mutex lock.
+// also update h.Conn.LastConnUse and h.hostQueueSize metric
 func (h *HostMTCP) tryToFlushIfNecessary() {
 	for c := range h.MTCPs {
 		if h.MTCPs[c].W != nil && h.MTCPs[c].W.Buffered() != 0 {
@@ -277,6 +280,7 @@ func (h *HostMTCP) tryToFlushIfNecessary() {
 			h.MTCPs[c].LastConnUse = time.Now()
 		}
 	}
+	h.hostQueueSize.Set(float64(len(h.Ch)))
 }
 
 // Requires h.Conn.Mutex lock.