CCQ: Query server logging (#3508)
* CCQ: Add additional logging to query server * Code review rework
This commit is contained in:
parent
3ba2a46671
commit
140e8a08d7
|
@ -62,7 +62,7 @@ func (s *httpServer) handleQuery(w http.ResponseWriter, r *http.Request) {
|
||||||
var q queryRequest
|
var q queryRequest
|
||||||
err := json.NewDecoder(http.MaxBytesReader(w, r.Body, MAX_BODY_SIZE)).Decode(&q)
|
err := json.NewDecoder(http.MaxBytesReader(w, r.Body, MAX_BODY_SIZE)).Decode(&q)
|
||||||
if err != nil {
|
if err != nil {
|
||||||
s.logger.Debug("failed to decode body", zap.Error(err))
|
s.logger.Error("failed to decode body", zap.Error(err))
|
||||||
http.Error(w, err.Error(), http.StatusBadRequest)
|
http.Error(w, err.Error(), http.StatusBadRequest)
|
||||||
invalidQueryRequestReceived.WithLabelValues("failed_to_decode_body").Inc()
|
invalidQueryRequestReceived.WithLabelValues("failed_to_decode_body").Inc()
|
||||||
return
|
return
|
||||||
|
@ -71,7 +71,7 @@ func (s *httpServer) handleQuery(w http.ResponseWriter, r *http.Request) {
|
||||||
// There should be one and only one API key in the header.
|
// There should be one and only one API key in the header.
|
||||||
apiKeys, exists := r.Header["X-Api-Key"]
|
apiKeys, exists := r.Header["X-Api-Key"]
|
||||||
if !exists || len(apiKeys) != 1 {
|
if !exists || len(apiKeys) != 1 {
|
||||||
s.logger.Debug("received a request with the wrong number of api keys", zap.Stringer("url", r.URL), zap.Int("numApiKeys", len(apiKeys)))
|
s.logger.Error("received a request with the wrong number of api keys", zap.Stringer("url", r.URL), zap.Int("numApiKeys", len(apiKeys)))
|
||||||
http.Error(w, "api key is missing", http.StatusUnauthorized)
|
http.Error(w, "api key is missing", http.StatusUnauthorized)
|
||||||
invalidQueryRequestReceived.WithLabelValues("missing_api_key").Inc()
|
invalidQueryRequestReceived.WithLabelValues("missing_api_key").Inc()
|
||||||
return
|
return
|
||||||
|
@ -79,9 +79,9 @@ func (s *httpServer) handleQuery(w http.ResponseWriter, r *http.Request) {
|
||||||
apiKey := strings.ToLower(apiKeys[0])
|
apiKey := strings.ToLower(apiKeys[0])
|
||||||
|
|
||||||
// Make sure the user is authorized before we go any farther.
|
// Make sure the user is authorized before we go any farther.
|
||||||
_, exists = s.permissions[apiKey]
|
permEntry, exists := s.permissions[apiKey]
|
||||||
if !exists {
|
if !exists {
|
||||||
s.logger.Debug("invalid api key", zap.String("apiKey", apiKey))
|
s.logger.Error("invalid api key", zap.String("apiKey", apiKey))
|
||||||
http.Error(w, "invalid api key", http.StatusForbidden)
|
http.Error(w, "invalid api key", http.StatusForbidden)
|
||||||
invalidQueryRequestReceived.WithLabelValues("invalid_api_key").Inc()
|
invalidQueryRequestReceived.WithLabelValues("invalid_api_key").Inc()
|
||||||
return
|
return
|
||||||
|
@ -89,7 +89,7 @@ func (s *httpServer) handleQuery(w http.ResponseWriter, r *http.Request) {
|
||||||
|
|
||||||
queryRequestBytes, err := hex.DecodeString(q.Bytes)
|
queryRequestBytes, err := hex.DecodeString(q.Bytes)
|
||||||
if err != nil {
|
if err != nil {
|
||||||
s.logger.Debug("failed to decode request bytes", zap.Error(err))
|
s.logger.Error("failed to decode request bytes", zap.String("userId", permEntry.userName), zap.Error(err))
|
||||||
http.Error(w, err.Error(), http.StatusBadRequest)
|
http.Error(w, err.Error(), http.StatusBadRequest)
|
||||||
invalidQueryRequestReceived.WithLabelValues("failed_to_decode_request").Inc()
|
invalidQueryRequestReceived.WithLabelValues("failed_to_decode_request").Inc()
|
||||||
return
|
return
|
||||||
|
@ -97,7 +97,7 @@ func (s *httpServer) handleQuery(w http.ResponseWriter, r *http.Request) {
|
||||||
|
|
||||||
signature, err := hex.DecodeString(q.Signature)
|
signature, err := hex.DecodeString(q.Signature)
|
||||||
if err != nil {
|
if err != nil {
|
||||||
s.logger.Debug("failed to decode signature bytes", zap.Error(err))
|
s.logger.Error("failed to decode signature bytes", zap.String("userId", permEntry.userName), zap.Error(err))
|
||||||
http.Error(w, err.Error(), http.StatusBadRequest)
|
http.Error(w, err.Error(), http.StatusBadRequest)
|
||||||
invalidQueryRequestReceived.WithLabelValues("failed_to_decode_signature").Inc()
|
invalidQueryRequestReceived.WithLabelValues("failed_to_decode_signature").Inc()
|
||||||
return
|
return
|
||||||
|
@ -108,8 +108,11 @@ func (s *httpServer) handleQuery(w http.ResponseWriter, r *http.Request) {
|
||||||
Signature: signature,
|
Signature: signature,
|
||||||
}
|
}
|
||||||
|
|
||||||
|
requestId := hex.EncodeToString(signedQueryRequest.Signature)
|
||||||
|
s.logger.Info("received request from client", zap.String("userId", permEntry.userName), zap.String("requestId", requestId))
|
||||||
|
|
||||||
if status, err := validateRequest(s.logger, s.env, s.permissions, s.signerKey, apiKey, signedQueryRequest); err != nil {
|
if status, err := validateRequest(s.logger, s.env, s.permissions, s.signerKey, apiKey, signedQueryRequest); err != nil {
|
||||||
// Don't need to log here because the details were logged in the function.
|
s.logger.Error("failed to validate request", zap.String("userId", permEntry.userName), zap.String("requestId", requestId), zap.Int("status", status), zap.Error(err))
|
||||||
http.Error(w, err.Error(), status)
|
http.Error(w, err.Error(), status)
|
||||||
// Metric has already been pegged.
|
// Metric has already been pegged.
|
||||||
return
|
return
|
||||||
|
@ -123,7 +126,7 @@ func (s *httpServer) handleQuery(w http.ResponseWriter, r *http.Request) {
|
||||||
|
|
||||||
b, err := proto.Marshal(&m)
|
b, err := proto.Marshal(&m)
|
||||||
if err != nil {
|
if err != nil {
|
||||||
s.logger.Error("failed to marshal gossip message", zap.Error(err))
|
s.logger.Error("failed to marshal gossip message", zap.String("userId", permEntry.userName), zap.String("requestId", requestId), zap.Error(err))
|
||||||
http.Error(w, err.Error(), http.StatusInternalServerError)
|
http.Error(w, err.Error(), http.StatusInternalServerError)
|
||||||
invalidQueryRequestReceived.WithLabelValues("failed_to_marshal_gossip_msg").Inc()
|
invalidQueryRequestReceived.WithLabelValues("failed_to_marshal_gossip_msg").Inc()
|
||||||
return
|
return
|
||||||
|
@ -132,14 +135,16 @@ func (s *httpServer) handleQuery(w http.ResponseWriter, r *http.Request) {
|
||||||
pendingResponse := NewPendingResponse(signedQueryRequest)
|
pendingResponse := NewPendingResponse(signedQueryRequest)
|
||||||
added := s.pendingResponses.Add(pendingResponse)
|
added := s.pendingResponses.Add(pendingResponse)
|
||||||
if !added {
|
if !added {
|
||||||
|
s.logger.Info("duplicate request", zap.String("userId", permEntry.userName), zap.String("requestId", requestId))
|
||||||
http.Error(w, "Duplicate request", http.StatusBadRequest)
|
http.Error(w, "Duplicate request", http.StatusBadRequest)
|
||||||
invalidQueryRequestReceived.WithLabelValues("duplicate_request").Inc()
|
invalidQueryRequestReceived.WithLabelValues("duplicate_request").Inc()
|
||||||
return
|
return
|
||||||
}
|
}
|
||||||
|
|
||||||
|
s.logger.Info("posting request to gossip", zap.String("userId", permEntry.userName), zap.String("requestId", requestId))
|
||||||
err = s.topic.Publish(r.Context(), b)
|
err = s.topic.Publish(r.Context(), b)
|
||||||
if err != nil {
|
if err != nil {
|
||||||
s.logger.Error("failed to publish gossip message", zap.Error(err))
|
s.logger.Error("failed to publish gossip message", zap.String("userId", permEntry.userName), zap.String("requestId", requestId), zap.Error(err))
|
||||||
http.Error(w, err.Error(), http.StatusInternalServerError)
|
http.Error(w, err.Error(), http.StatusInternalServerError)
|
||||||
invalidQueryRequestReceived.WithLabelValues("failed_to_publish_gossip_msg").Inc()
|
invalidQueryRequestReceived.WithLabelValues("failed_to_publish_gossip_msg").Inc()
|
||||||
s.pendingResponses.Remove(pendingResponse)
|
s.pendingResponses.Remove(pendingResponse)
|
||||||
|
@ -149,11 +154,13 @@ func (s *httpServer) handleQuery(w http.ResponseWriter, r *http.Request) {
|
||||||
// Wait for the response or timeout
|
// Wait for the response or timeout
|
||||||
select {
|
select {
|
||||||
case <-time.After(query.RequestTimeout + 5*time.Second):
|
case <-time.After(query.RequestTimeout + 5*time.Second):
|
||||||
|
s.logger.Info("publishing time out to client", zap.String("userId", permEntry.userName), zap.String("requestId", requestId))
|
||||||
http.Error(w, "Timed out waiting for response", http.StatusGatewayTimeout)
|
http.Error(w, "Timed out waiting for response", http.StatusGatewayTimeout)
|
||||||
case res := <-pendingResponse.ch:
|
case res := <-pendingResponse.ch:
|
||||||
|
s.logger.Info("publishing response to client", zap.String("userId", permEntry.userName), zap.String("requestId", requestId))
|
||||||
resBytes, err := res.Response.Marshal()
|
resBytes, err := res.Response.Marshal()
|
||||||
if err != nil {
|
if err != nil {
|
||||||
s.logger.Error("failed to marshal response", zap.Error(err))
|
s.logger.Error("failed to marshal response", zap.String("userId", permEntry.userName), zap.String("requestId", requestId), zap.Error(err))
|
||||||
http.Error(w, err.Error(), http.StatusInternalServerError)
|
http.Error(w, err.Error(), http.StatusInternalServerError)
|
||||||
invalidQueryRequestReceived.WithLabelValues("failed_to_marshal_response").Inc()
|
invalidQueryRequestReceived.WithLabelValues("failed_to_marshal_response").Inc()
|
||||||
break
|
break
|
||||||
|
@ -174,9 +181,10 @@ func (s *httpServer) handleQuery(w http.ResponseWriter, r *http.Request) {
|
||||||
Bytes: hex.EncodeToString(resBytes),
|
Bytes: hex.EncodeToString(resBytes),
|
||||||
})
|
})
|
||||||
if err != nil {
|
if err != nil {
|
||||||
s.logger.Error("failed to encode response", zap.Error(err))
|
s.logger.Error("failed to encode response", zap.String("userId", permEntry.userName), zap.String("requestId", requestId), zap.Error(err))
|
||||||
http.Error(w, err.Error(), http.StatusInternalServerError)
|
http.Error(w, err.Error(), http.StatusInternalServerError)
|
||||||
invalidQueryRequestReceived.WithLabelValues("failed_to_encode_response").Inc()
|
invalidQueryRequestReceived.WithLabelValues("failed_to_encode_response").Inc()
|
||||||
|
break
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
|
|
|
@ -119,7 +119,8 @@ func runP2P(ctx context.Context, priv crypto.PrivKey, port uint, networkID, boot
|
||||||
switch m := msg.Message.(type) {
|
switch m := msg.Message.(type) {
|
||||||
case *gossipv1.GossipMessage_SignedQueryResponse:
|
case *gossipv1.GossipMessage_SignedQueryResponse:
|
||||||
logger.Debug("query response received", zap.Any("response", m.SignedQueryResponse))
|
logger.Debug("query response received", zap.Any("response", m.SignedQueryResponse))
|
||||||
queryResponsesReceived.WithLabelValues(envelope.GetFrom().String()).Inc()
|
peerId := envelope.GetFrom().String()
|
||||||
|
queryResponsesReceived.WithLabelValues(peerId).Inc()
|
||||||
var queryResponse query.QueryResponsePublication
|
var queryResponse query.QueryResponsePublication
|
||||||
err := queryResponse.Unmarshal(m.SignedQueryResponse.QueryResponse)
|
err := queryResponse.Unmarshal(m.SignedQueryResponse.QueryResponse)
|
||||||
if err != nil {
|
if err != nil {
|
||||||
|
@ -128,6 +129,7 @@ func runP2P(ctx context.Context, priv crypto.PrivKey, port uint, networkID, boot
|
||||||
continue
|
continue
|
||||||
}
|
}
|
||||||
requestSignature := hex.EncodeToString(queryResponse.Request.Signature)
|
requestSignature := hex.EncodeToString(queryResponse.Request.Signature)
|
||||||
|
logger.Info("query response received from gossip", zap.String("peerId", peerId), zap.Any("requestId", requestSignature))
|
||||||
// Check that we're handling the request for this response
|
// Check that we're handling the request for this response
|
||||||
pendingResponse := pendingResponses.Get(requestSignature)
|
pendingResponse := pendingResponses.Get(requestSignature)
|
||||||
if pendingResponse == nil {
|
if pendingResponse == nil {
|
||||||
|
@ -181,10 +183,22 @@ func runP2P(ctx context.Context, priv crypto.PrivKey, port uint, networkID, boot
|
||||||
delete(responses, requestSignature)
|
delete(responses, requestSignature)
|
||||||
select {
|
select {
|
||||||
case pendingResponse.ch <- s:
|
case pendingResponse.ch <- s:
|
||||||
logger.Debug("forwarded query response")
|
logger.Info("forwarded query response",
|
||||||
|
zap.String("peerId", peerId),
|
||||||
|
zap.Any("requestId", requestSignature),
|
||||||
|
zap.Int("numSigners", len(responses[requestSignature][digest])),
|
||||||
|
zap.Int("quorum", quorum),
|
||||||
|
)
|
||||||
default:
|
default:
|
||||||
logger.Error("failed to write query response to channel, dropping it")
|
logger.Error("failed to write query response to channel, dropping it", zap.String("peerId", peerId), zap.Any("requestId", requestSignature))
|
||||||
}
|
}
|
||||||
|
} else {
|
||||||
|
logger.Info("waiting for more query responses",
|
||||||
|
zap.String("peerId", peerId),
|
||||||
|
zap.Any("requestId", requestSignature),
|
||||||
|
zap.Int("numSigners", len(responses[requestSignature][digest])),
|
||||||
|
zap.Int("quorum", quorum),
|
||||||
|
)
|
||||||
}
|
}
|
||||||
} else {
|
} else {
|
||||||
logger.Warn("received observation by unknown guardian - is our guardian set outdated?",
|
logger.Warn("received observation by unknown guardian - is our guardian set outdated?",
|
||||||
|
|
Loading…
Reference in New Issue