feat: add backpressure, body size limit and metrics to pushgw proxy remote write (#3141)

Author: 710leo

pushgw/pconf/conf.go

@@ -22,18 +22,27 @@ type Pushgw struct {
 	UpdateTargetByUrlConcurrency   int
 
 	GetHeartbeatFromMetric bool // 是否从时序数据中提取机器心跳时间，默认 false
-	BusiGroupLabelKey   string
-	IdentMetrics        []string
-	IdentStatsThreshold int
-	IdentDropThreshold  int // 每分钟单个 ident 的样本数超过该阈值，则丢弃
-	WriteConcurrency    int
-	LabelRewrite        bool
-	ForceUseServerTS    bool
-	DebugSample         map[string]string
-	DropSample          []map[string]string
-	WriterOpt           WriterGlobalOpt
-	Writers             []WriterOptions
-	KafkaWriters        []KafkaWriterOptions
+	BusiGroupLabelKey      string
+	IdentMetrics           []string
+	IdentStatsThreshold    int
+	IdentDropThreshold     int // 每分钟单个 ident 的样本数超过该阈值，则丢弃
+	WriteConcurrency       int
+
+	// ProxyInflightMax 控制 /proxy/v1/write 的并发上限。超过阈值直接返回 429，
+	// 把背压交给客户端 WAL（remote_write 协议原生支持）。<=0 使用默认值。
+	ProxyInflightMax int
+
+	// ProxyMaxBodyBytes 限制 /proxy/v1/write 单个请求 body 的最大字节数，超过返回 413。
+	// 和 ProxyInflightMax 配套：并发 × 单请求大小 = pushgw 内存占用上限。<=0 使用默认值。
+	ProxyMaxBodyBytes int64
+
+	LabelRewrite     bool
+	ForceUseServerTS bool
+	DebugSample      map[string]string
+	DropSample       []map[string]string
+	WriterOpt        WriterGlobalOpt
+	Writers          []WriterOptions
+	KafkaWriters     []KafkaWriterOptions
 }
 
 type WriterGlobalOpt struct {
@@ -190,6 +199,14 @@ func (p *Pushgw) PreCheck() {
 		p.IdentDropThreshold = 5000000
 	}
 
+	if p.ProxyInflightMax <= 0 {
+		p.ProxyInflightMax = 1000
+	}
+
+	if p.ProxyMaxBodyBytes <= 0 {
+		p.ProxyMaxBodyBytes = 32 * 1024 * 1024
+	}
+
 	for index := range p.Writers {
 		for _, relabel := range p.Writers[index].WriteRelabels {
 			if relabel.Regex == "" {

pushgw/pstat/pstat.go

@@ -105,6 +105,58 @@ var (
 		},
 		[]string{"operation", "status"},
 	)
+
+	GaugeProxyRemoteWriteInflight = prometheus.NewGauge(prometheus.GaugeOpts{
+		Namespace: namespace,
+		Subsystem: subsystem,
+		Name:      "proxy_remote_write_inflight",
+		Help:      "Current number of in-flight requests on /proxy/v1/write.",
+	})
+
+	CounterProxyRemoteWriteOverLimitTotal = prometheus.NewCounter(prometheus.CounterOpts{
+		Namespace: namespace,
+		Subsystem: subsystem,
+		Name:      "proxy_remote_write_over_limit_total",
+		Help:      "Number of /proxy/v1/write requests rejected with 429 due to in-flight over limit.",
+	})
+
+	CounterProxyRemoteWriteBodyTooLargeTotal = prometheus.NewCounter(prometheus.CounterOpts{
+		Namespace: namespace,
+		Subsystem: subsystem,
+		Name:      "proxy_remote_write_body_too_large_total",
+		Help:      "Number of /proxy/v1/write requests rejected with 413 due to body size over limit.",
+	})
+
+	CounterProxyRemoteWriteTotal = prometheus.NewCounter(prometheus.CounterOpts{
+		Namespace: namespace,
+		Subsystem: subsystem,
+		Name:      "proxy_remote_write_total",
+		Help:      "Number of /proxy/v1/write requests received.",
+	})
+
+	CounterProxyForwardTotal = prometheus.NewCounterVec(prometheus.CounterOpts{
+		Namespace: namespace,
+		Subsystem: subsystem,
+		Name:      "proxy_forward_total",
+		Help:      "Number of forwards performed by /proxy/v1/write.",
+	}, []string{"url"})
+
+	CounterProxyForwardErrorTotal = prometheus.NewCounterVec(prometheus.CounterOpts{
+		Namespace: namespace,
+		Subsystem: subsystem,
+		Name:      "proxy_forward_error_total",
+		Help:      "Number of forward errors on /proxy/v1/write.",
+	}, []string{"url", "reason"})
+
+	ProxyForwardDuration = prometheus.NewHistogramVec(
+		prometheus.HistogramOpts{
+			Namespace: namespace,
+			Subsystem: subsystem,
+			Buckets:   []float64{.001, .01, .1, 1, 5, 10},
+			Name:      "proxy_forward_duration_seconds",
+			Help:      "Forward latencies on /proxy/v1/write in seconds.",
+		}, []string{"url"},
+	)
 )
 
 func init() {
@@ -121,5 +173,12 @@ func init() {
 		GaugeSampleQueueSize,
 		CounterPushQueueOverLimitTotal,
 		RedisOperationLatency,
+		GaugeProxyRemoteWriteInflight,
+		CounterProxyRemoteWriteOverLimitTotal,
+		CounterProxyRemoteWriteBodyTooLargeTotal,
+		CounterProxyRemoteWriteTotal,
+		CounterProxyForwardTotal,
+		CounterProxyForwardErrorTotal,
+		ProxyForwardDuration,
 	)
 }