@@ -2,22 +2,44 @@ package push
22
33import (
44 "context"
5+ "fmt"
56 "net/http"
7+ "strconv"
8+ "strings"
69
710 "github.com/go-kit/log/level"
11+ "github.com/prometheus/prometheus/config"
12+ "github.com/prometheus/prometheus/storage/remote"
813 "github.com/weaveworks/common/httpgrpc"
914 "github.com/weaveworks/common/middleware"
1015
1116 "github.com/cortexproject/cortex/pkg/cortexpb"
17+ "github.com/cortexproject/cortex/pkg/cortexpbv2"
1218 "github.com/cortexproject/cortex/pkg/util"
1319 "github.com/cortexproject/cortex/pkg/util/log"
1420)
1521
22+ const (
23+ remoteWriteVersionHeader = "X-Prometheus-Remote-Write-Version"
24+ remoteWriteVersion1HeaderValue = "0.1.0"
25+ remoteWriteVersion20HeaderValue = "2.0.0"
26+ appProtoContentType = "application/x-protobuf"
27+ appProtoV1ContentType = "application/x-protobuf;proto=prometheus.WriteRequest"
28+ appProtoV2ContentType = "application/x-protobuf;proto=io.prometheus.write.v2.Request"
29+
30+ rw20WrittenSamplesHeader = "X-Prometheus-Remote-Write-Samples-Written"
31+ rw20WrittenHistogramsHeader = "X-Prometheus-Remote-Write-Histograms-Written"
32+ rw20WrittenExemplarsHeader = "X-Prometheus-Remote-Write-Exemplars-Written"
33+ )
34+
1635// Func defines the type of the push. It is similar to http.HandlerFunc.
1736type Func func (context.Context , * cortexpb.WriteRequest ) (* cortexpb.WriteResponse , error )
1837
38+ // FuncV2 defines the type of the pushV2. It is similar to http.HandlerFunc.
39+ type FuncV2 func (ctx context.Context , request * cortexpbv2.WriteRequest ) (* cortexpbv2.WriteResponse , error )
40+
1941// Handler is a http.Handler which accepts WriteRequests.
20- func Handler (maxRecvMsgSize int , sourceIPs * middleware.SourceIPExtractor , push Func ) http.Handler {
42+ func Handler (maxRecvMsgSize int , sourceIPs * middleware.SourceIPExtractor , push Func , pushV2 FuncV2 ) http.Handler {
2143 return http .HandlerFunc (func (w http.ResponseWriter , r * http.Request ) {
2244 ctx := r .Context ()
2345 logger := log .WithContext (ctx , log .Logger )
@@ -28,31 +50,123 @@ func Handler(maxRecvMsgSize int, sourceIPs *middleware.SourceIPExtractor, push F
2850 logger = log .WithSourceIPs (source , logger )
2951 }
3052 }
31- var req cortexpb.PreallocWriteRequest
32- err := util .ParseProtoReader (ctx , r .Body , int (r .ContentLength ), maxRecvMsgSize , & req , util .RawSnappy )
53+
54+ contentType := r .Header .Get ("Content-Type" )
55+ if contentType == "" {
56+ contentType = appProtoContentType
57+ }
58+
59+ msgType , err := parseProtoMsg (contentType )
3360 if err != nil {
34- level .Error (logger ).Log ("err" , err . Error () )
35- http .Error (w , err .Error (), http .StatusBadRequest )
61+ level .Error (logger ).Log ("Error decoding remote write request" , " err"err )
62+ http .Error (w , err .Error (), http .StatusUnsupportedMediaType )
3663 return
3764 }
3865
39- req .SkipLabelNameValidation = false
40- if req .Source == 0 {
41- req .Source = cortexpb .API
66+ if msgType != config .RemoteWriteProtoMsgV1 && msgType != config .RemoteWriteProtoMsgV2 {
67+ level .Error (logger ).Log ("Error decoding remote write request" , "err" , err )
68+ http .Error (w , err .Error (), http .StatusUnsupportedMediaType )
69+ return
70+ }
71+
72+ enc := r .Header .Get ("Content-Encoding" )
73+ if enc == "" {
74+ } else if enc != string (remote .SnappyBlockCompression ) {
75+ err := fmt .Errorf ("%v encoding (compression) is not accepted by this server; only %v is acceptable" , enc , remote .SnappyBlockCompression )
76+ level .Error (logger ).Log ("Error decoding remote write request" , "err" , err )
77+ http .Error (w , err .Error (), http .StatusUnsupportedMediaType )
78+ return
4279 }
4380
44- if _ , err := push (ctx , & req .WriteRequest ); err != nil {
45- resp , ok := httpgrpc .HTTPResponseFromError (err )
46- if ! ok {
47- http .Error (w , err .Error (), http .StatusInternalServerError )
81+ switch msgType {
82+ case config .RemoteWriteProtoMsgV1 :
83+ var req cortexpb.PreallocWriteRequest
84+ err := util .ParseProtoReader (ctx , r .Body , int (r .ContentLength ), maxRecvMsgSize , & req , util .RawSnappy )
85+ if err != nil {
86+ level .Error (logger ).Log ("err" , err .Error ())
87+ http .Error (w , err .Error (), http .StatusBadRequest )
88+ return
89+ }
90+
91+ req .SkipLabelNameValidation = false
92+ if req .Source == 0 {
93+ req .Source = cortexpb .API
94+ }
95+
96+ if _ , err := push (ctx , & req .WriteRequest ); err != nil {
97+ resp , ok := httpgrpc .HTTPResponseFromError (err )
98+ if ! ok {
99+ http .Error (w , err .Error (), http .StatusInternalServerError )
100+ return
101+ }
102+ if resp .GetCode ()/ 100 == 5 {
103+ level .Error (logger ).Log ("msg" , "push error" , "err" , err )
104+ } else if resp .GetCode () != http .StatusAccepted && resp .GetCode () != http .StatusTooManyRequests {
105+ level .Warn (logger ).Log ("msg" , "push refused" , "err" , err )
106+ }
107+ http .Error (w , string (resp .Body ), int (resp .Code ))
108+ }
109+ case config .RemoteWriteProtoMsgV2 :
110+ var req cortexpbv2.WriteRequest
111+ err := util .ParseProtoReader (ctx , r .Body , int (r .ContentLength ), maxRecvMsgSize , & req , util .RawSnappy )
112+ if err != nil {
113+ fmt .Println ("err" , err )
114+ level .Error (logger ).Log ("err" , err .Error ())
115+ http .Error (w , err .Error (), http .StatusBadRequest )
48116 return
49117 }
50- if resp .GetCode ()/ 100 == 5 {
51- level .Error (logger ).Log ("msg" , "push error" , "err" , err )
52- } else if resp .GetCode () != http .StatusAccepted && resp .GetCode () != http .StatusTooManyRequests {
53- level .Warn (logger ).Log ("msg" , "push refused" , "err" , err )
118+
119+ req .SkipLabelNameValidation = false
120+ if req .Source == 0 {
121+ req .Source = cortexpbv2 .API
122+ }
123+
124+ if resp , err := pushV2 (ctx , & req ); err != nil {
125+ resp , ok := httpgrpc .HTTPResponseFromError (err )
126+ w .Header ().Set (rw20WrittenSamplesHeader , "0" )
127+ w .Header ().Set (rw20WrittenHistogramsHeader , "0" )
128+ w .Header ().Set (rw20WrittenExemplarsHeader , "0" )
129+ if ! ok {
130+ http .Error (w , err .Error (), http .StatusInternalServerError )
131+ return
132+ }
133+ if resp .GetCode ()/ 100 == 5 {
134+ level .Error (logger ).Log ("msg" , "push error" , "err" , err )
135+ } else if resp .GetCode () != http .StatusAccepted && resp .GetCode () != http .StatusTooManyRequests {
136+ level .Warn (logger ).Log ("msg" , "push refused" , "err" , err )
137+ }
138+ http .Error (w , string (resp .Body ), int (resp .Code ))
139+ } else {
140+ w .Header ().Set (rw20WrittenSamplesHeader , strconv .FormatInt (resp .Samples , 10 ))
141+ w .Header ().Set (rw20WrittenHistogramsHeader , strconv .FormatInt (resp .Histograms , 10 ))
142+ w .Header ().Set (rw20WrittenExemplarsHeader , strconv .FormatInt (resp .Exemplars , 10 ))
54143 }
55- http .Error (w , string (resp .Body ), int (resp .Code ))
56144 }
57145 })
58146}
147+
148+ // Refer to parseProtoMsg in https://github.com/prometheus/prometheus/blob/main/storage/remote/write_handler.go
149+ func parseProtoMsg (contentType string ) (config.RemoteWriteProtoMsg , error ) {
150+ contentType = strings .TrimSpace (contentType )
151+
152+ parts := strings .Split (contentType , ";" )
153+ if parts [0 ] != appProtoContentType {
154+ return "" , fmt .Errorf ("expected %v as the first (media) part, got %v content-type" , appProtoContentType , contentType )
155+ }
156+ // Parse potential https://www.rfc-editor.org/rfc/rfc9110#parameter
157+ for _ , p := range parts [1 :] {
158+ pair := strings .Split (p , "=" )
159+ if len (pair ) != 2 {
160+ return "" , fmt .Errorf ("as per https://www.rfc-editor.org/rfc/rfc9110#parameter expected parameters to be key-values, got %v in %v content-type" , p , contentType )
161+ }
162+ if pair [0 ] == "proto" {
163+ ret := config .RemoteWriteProtoMsg (pair [1 ])
164+ if err := ret .Validate (); err != nil {
165+ return "" , fmt .Errorf ("got %v content type; %w" , contentType , err )
166+ }
167+ return ret , nil
168+ }
169+ }
170+ // No "proto=" parameter, assuming v1.
171+ return config .RemoteWriteProtoMsgV1 , nil
172+ }
0 commit comments