@@ -8,29 +8,55 @@ import (
88 "sync/atomic"
99 "time"
1010
11+ "go.uber.org/fx"
1112 "go.uber.org/zap"
1213
13- "github.com/go-kit/kit/metrics "
14+ "github.com/prometheus/client_golang/prometheus "
1415 uuid "github.com/satori/go.uuid"
1516
1617 "github.com/xmidt-org/sallust"
1718 "github.com/xmidt-org/webpa-common/v2/adapter"
1819 "github.com/xmidt-org/wrp-go/v3"
1920)
2021
22+ type ServerHandlerIn struct {
23+ fx.In
24+ Logger * zap.Logger
25+ Telemetry * HandlerTelemetry
26+ }
27+
28+ type ServerHandlerOut struct {
29+ fx.Out
30+ Handler * ServerHandler
31+ }
32+
2133// Below is the struct that will implement our ServeHTTP method
2234type ServerHandler struct {
23- * zap.Logger
24- caduceusHandler RequestHandler
25- errorRequests metrics.Counter
26- emptyRequests metrics.Counter
27- invalidCount metrics.Counter
28- incomingQueueDepthMetric metrics.Gauge
29- modifiedWRPCount metrics.Counter
30- incomingQueueDepth int64
31- maxOutstanding int64
32- incomingQueueLatency metrics.Histogram
33- now func () time.Time
35+ log * zap.Logger
36+ // caduceusHandler RequestHandler
37+ telemetry * HandlerTelemetry
38+ incomingQueueDepth int64
39+ maxOutstanding int64
40+
41+ now func () time.Time
42+ }
43+
44+ type HandlerTelemetryIn struct {
45+ fx.In
46+ ErrorRequests prometheus.Counter `name:"error_request_body_counter"`
47+ EmptyRequests prometheus.Counter `name:"empty_request_boyd_counter"`
48+ InvalidCount prometheus.Counter `name:"drops_due_to_invalid_payload"`
49+ IncomingQueueDepthMetric prometheus.Gauge `name:"incoming_queue_depth"`
50+ ModifiedWRPCount prometheus.CounterVec `name:"modified_wrp_count"`
51+ IncomingQueueLatency prometheus.HistogramVec `name:"incoming_queue_latency_histogram_seconds"`
52+ }
53+ type HandlerTelemetry struct {
54+ errorRequests prometheus.Counter
55+ emptyRequests prometheus.Counter
56+ invalidCount prometheus.Counter
57+ incomingQueueDepthMetric prometheus.Gauge
58+ modifiedWRPCount prometheus.CounterVec
59+ incomingQueueLatency prometheus.HistogramVec
3460}
3561
3662func (sh * ServerHandler ) ServeHTTP (response http.ResponseWriter , request * http.Request ) {
@@ -42,7 +68,7 @@ func (sh *ServerHandler) ServeHTTP(response http.ResponseWriter, request *http.R
4268
4369 logger := sallust .Get (request .Context ())
4470 if logger == adapter .DefaultLogger ().Logger {
45- logger = sh .Logger
71+ logger = sh .log
4672 }
4773
4874 logger .Info ("Receiving incoming request..." )
@@ -66,19 +92,19 @@ func (sh *ServerHandler) ServeHTTP(response http.ResponseWriter, request *http.R
6692 return
6793 }
6894
69- sh .incomingQueueDepthMetric .Add (1.0 )
70- defer sh .incomingQueueDepthMetric .Add (- 1.0 )
95+ sh .telemetry . incomingQueueDepthMetric .Add (1.0 )
96+ defer sh .telemetry . incomingQueueDepthMetric .Add (- 1.0 )
7197
7298 payload , err := io .ReadAll (request .Body )
7399 if err != nil {
74- sh .errorRequests .Add (1.0 )
100+ sh .telemetry . errorRequests .Add (1.0 )
75101 logger .Error ("Unable to retrieve the request body." , zap .Error (err ))
76102 response .WriteHeader (http .StatusBadRequest )
77103 return
78104 }
79105
80106 if len (payload ) == 0 {
81- sh .emptyRequests .Add (1.0 )
107+ sh .telemetry . emptyRequests .Add (1.0 )
82108 logger .Error ("Empty payload." )
83109 response .WriteHeader (http .StatusBadRequest )
84110 response .Write ([]byte ("Empty payload.\n " ))
@@ -91,7 +117,7 @@ func (sh *ServerHandler) ServeHTTP(response http.ResponseWriter, request *http.R
91117 err = decoder .Decode (msg )
92118 if err != nil || msg .MessageType () != 4 {
93119 // return a 400
94- sh .invalidCount .Add (1.0 )
120+ sh .telemetry . invalidCount .Add (1.0 )
95121 response .WriteHeader (http .StatusBadRequest )
96122 if err != nil {
97123 response .Write ([]byte ("Invalid payload format.\n " ))
@@ -106,15 +132,15 @@ func (sh *ServerHandler) ServeHTTP(response http.ResponseWriter, request *http.R
106132 err = wrp .UTF8 (msg )
107133 if err != nil {
108134 // return a 400
109- sh .invalidCount .Add (1.0 )
135+ sh .telemetry . invalidCount .Add (1.0 )
110136 response .WriteHeader (http .StatusBadRequest )
111137 response .Write ([]byte ("Strings must be UTF-8.\n " ))
112138 logger .Debug ("Strings must be UTF-8." )
113139 return
114140 }
115141 eventType = msg .FindEventStringSubMatch ()
116142
117- sh .caduceusHandler .HandleRequest (0 , sh .fixWrp (msg ))
143+ // sh.caduceusHandler.HandleRequest(0, sh.fixWrp(msg))
118144
119145 // return a 202
120146 response .WriteHeader (http .StatusAccepted )
@@ -125,7 +151,7 @@ func (sh *ServerHandler) ServeHTTP(response http.ResponseWriter, request *http.R
125151
126152func (sh * ServerHandler ) recordQueueLatencyToHistogram (startTime time.Time , eventType string ) {
127153 endTime := sh .now ()
128- sh .incomingQueueLatency .With ("event" , eventType ).Observe (endTime .Sub (startTime ).Seconds ())
154+ sh .telemetry . incomingQueueLatency .With (prometheus. Labels { "event" : eventType } ).Observe (float64 ( endTime .Sub (startTime ).Seconds () ))
129155}
130156
131157func (sh * ServerHandler ) fixWrp (msg * wrp.Message ) * wrp.Message {
@@ -134,13 +160,13 @@ func (sh *ServerHandler) fixWrp(msg *wrp.Message) *wrp.Message {
134160
135161 // Default to "application/json" if there is no content type, otherwise
136162 // use the one the source specified.
137- if "" == msg . ContentType {
163+ if msg . ContentType == "" {
138164 msg .ContentType = wrp .MimeTypeJson
139165 reason = emptyContentTypeReason
140166 }
141167
142168 // Ensure there is a transaction id even if we make one up
143- if "" == msg . TransactionUUID {
169+ if msg . TransactionUUID == "" {
144170 msg .TransactionUUID = uuid .NewV4 ().String ()
145171 if reason == "" {
146172 reason = emptyUUIDReason
@@ -150,8 +176,40 @@ func (sh *ServerHandler) fixWrp(msg *wrp.Message) *wrp.Message {
150176 }
151177
152178 if reason != "" {
153- sh .modifiedWRPCount .With ("reason" , reason ).Add (1.0 )
179+ sh .telemetry . modifiedWRPCount .With (prometheus. Labels { "reason" : reason } ).Add (1.0 )
154180 }
155181
156182 return msg
157183}
184+
185+ var HandlerModule = fx .Module ("server" ,
186+ fx .Provide (
187+ func (in HandlerTelemetryIn ) * HandlerTelemetry {
188+ return & HandlerTelemetry {
189+ errorRequests : in .ErrorRequests ,
190+ emptyRequests : in .EmptyRequests ,
191+ invalidCount : in .InvalidCount ,
192+ incomingQueueDepthMetric : in .IncomingQueueDepthMetric ,
193+ modifiedWRPCount : in .ModifiedWRPCount ,
194+ incomingQueueLatency : in .IncomingQueueLatency ,
195+ }
196+ }),
197+ fx .Provide (
198+ func (in ServerHandlerIn ) (ServerHandlerOut , error ) {
199+ //Hard coding maxOutstanding and incomingQueueDepth for now
200+ handler , err := New (in .Logger , in .Telemetry , 0.0 , 0.0 )
201+ return ServerHandlerOut {
202+ Handler : handler ,
203+ }, err
204+ },
205+ ),
206+ )
207+
208+ func New (log * zap.Logger , t * HandlerTelemetry , maxOutstanding , incomingQueueDepth int64 ) (* ServerHandler , error ) {
209+ return & ServerHandler {
210+ log : log ,
211+ telemetry : t ,
212+ maxOutstanding : maxOutstanding ,
213+ incomingQueueDepth : incomingQueueDepth ,
214+ }, nil
215+ }
0 commit comments