|
|
@ -2,6 +2,7 @@ package serf
|
|
|
|
|
|
|
|
|
|
|
|
import (
|
|
|
|
import (
|
|
|
|
"encoding/base64"
|
|
|
|
"encoding/base64"
|
|
|
|
|
|
|
|
"fmt"
|
|
|
|
"log"
|
|
|
|
"log"
|
|
|
|
"strings"
|
|
|
|
"strings"
|
|
|
|
)
|
|
|
|
)
|
|
|
@ -28,6 +29,13 @@ const (
|
|
|
|
|
|
|
|
|
|
|
|
// listKeysQuery is used to list all known keys in the cluster
|
|
|
|
// listKeysQuery is used to list all known keys in the cluster
|
|
|
|
listKeysQuery = "list-keys"
|
|
|
|
listKeysQuery = "list-keys"
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
// minEncodedKeyLength is used to compute the max number of keys in a list key
|
|
|
|
|
|
|
|
// response. eg 1024/25 = 40. a message with max size of 1024 bytes cannot
|
|
|
|
|
|
|
|
// contain more than 40 keys. There is a test
|
|
|
|
|
|
|
|
// (TestSerfQueries_estimateMaxKeysInListKeyResponse) which does the
|
|
|
|
|
|
|
|
// computation and in case of changes, the value can be adjusted.
|
|
|
|
|
|
|
|
minEncodedKeyLength = 25
|
|
|
|
)
|
|
|
|
)
|
|
|
|
|
|
|
|
|
|
|
|
// internalQueryName is used to generate a query name for an internal query
|
|
|
|
// internalQueryName is used to generate a query name for an internal query
|
|
|
@ -149,17 +157,62 @@ func (s *serfQueries) handleConflict(q *Query) {
|
|
|
|
}
|
|
|
|
}
|
|
|
|
}
|
|
|
|
}
|
|
|
|
|
|
|
|
|
|
|
|
// sendKeyResponse handles responding to key-related queries.
|
|
|
|
func (s *serfQueries) keyListResponseWithCorrectSize(q *Query, resp *nodeKeyResponse) ([]byte, messageQueryResponse, error) {
|
|
|
|
func (s *serfQueries) sendKeyResponse(q *Query, resp *nodeKeyResponse) {
|
|
|
|
maxListKeys := q.serf.config.QueryResponseSizeLimit / minEncodedKeyLength
|
|
|
|
buf, err := encodeMessage(messageKeyResponseType, resp)
|
|
|
|
actual := len(resp.Keys)
|
|
|
|
if err != nil {
|
|
|
|
for i := maxListKeys; i >= 0; i-- {
|
|
|
|
s.logger.Printf("[ERR] serf: Failed to encode key response: %v", err)
|
|
|
|
buf, err := encodeMessage(messageKeyResponseType, resp)
|
|
|
|
return
|
|
|
|
if err != nil {
|
|
|
|
|
|
|
|
return nil, messageQueryResponse{}, err
|
|
|
|
|
|
|
|
}
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
// Create response
|
|
|
|
|
|
|
|
qresp := q.createResponse(buf)
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
// Encode response
|
|
|
|
|
|
|
|
raw, err := encodeMessage(messageQueryResponseType, qresp)
|
|
|
|
|
|
|
|
if err != nil {
|
|
|
|
|
|
|
|
return nil, messageQueryResponse{}, err
|
|
|
|
|
|
|
|
}
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
// Check the size limit
|
|
|
|
|
|
|
|
if err = q.checkResponseSize(raw); err != nil {
|
|
|
|
|
|
|
|
resp.Keys = resp.Keys[0:i]
|
|
|
|
|
|
|
|
resp.Message = fmt.Sprintf("truncated key list response, showing first %d of %d keys", i, actual)
|
|
|
|
|
|
|
|
continue
|
|
|
|
|
|
|
|
}
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
if actual > i {
|
|
|
|
|
|
|
|
s.logger.Printf("[WARN] serf: %s", resp.Message)
|
|
|
|
|
|
|
|
}
|
|
|
|
|
|
|
|
return raw, qresp, nil
|
|
|
|
}
|
|
|
|
}
|
|
|
|
|
|
|
|
return nil, messageQueryResponse{}, fmt.Errorf("Failed to truncate response so that it fits into message")
|
|
|
|
|
|
|
|
}
|
|
|
|
|
|
|
|
|
|
|
|
if err := q.Respond(buf); err != nil {
|
|
|
|
// sendKeyResponse handles responding to key-related queries.
|
|
|
|
s.logger.Printf("[ERR] serf: Failed to respond to key query: %v", err)
|
|
|
|
func (s *serfQueries) sendKeyResponse(q *Query, resp *nodeKeyResponse) {
|
|
|
|
return
|
|
|
|
switch q.Name {
|
|
|
|
|
|
|
|
case internalQueryName(listKeysQuery):
|
|
|
|
|
|
|
|
raw, qresp, err := s.keyListResponseWithCorrectSize(q, resp)
|
|
|
|
|
|
|
|
if err != nil {
|
|
|
|
|
|
|
|
s.logger.Printf("[ERR] serf: %v", err)
|
|
|
|
|
|
|
|
return
|
|
|
|
|
|
|
|
}
|
|
|
|
|
|
|
|
if err := q.respondWithMessageAndResponse(raw, qresp); err != nil {
|
|
|
|
|
|
|
|
s.logger.Printf("[ERR] serf: Failed to respond to key query: %v", err)
|
|
|
|
|
|
|
|
return
|
|
|
|
|
|
|
|
}
|
|
|
|
|
|
|
|
default:
|
|
|
|
|
|
|
|
buf, err := encodeMessage(messageKeyResponseType, resp)
|
|
|
|
|
|
|
|
if err != nil {
|
|
|
|
|
|
|
|
s.logger.Printf("[ERR] serf: Failed to encode key response: %v", err)
|
|
|
|
|
|
|
|
return
|
|
|
|
|
|
|
|
}
|
|
|
|
|
|
|
|
if err := q.Respond(buf); err != nil {
|
|
|
|
|
|
|
|
s.logger.Printf("[ERR] serf: Failed to respond to key query: %v", err)
|
|
|
|
|
|
|
|
return
|
|
|
|
|
|
|
|
}
|
|
|
|
}
|
|
|
|
}
|
|
|
|
}
|
|
|
|
}
|
|
|
|
|
|
|
|
|
|
|
|