forked from moby/libnetwork
/
query.go
210 lines (181 loc) · 5.36 KB
/
query.go
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
package serf
import (
"math"
"regexp"
"sync"
"time"
)
// QueryParam is provided to Query() to configure the parameters of the
// query. If not provided, sane defaults will be used.
type QueryParam struct {
// If provided, we restrict the nodes that should respond to those
// with names in this list
FilterNodes []string
// FilterTags maps a tag name to a regular expression that is applied
// to restrict the nodes that should respond
FilterTags map[string]string
// If true, we are requesting an delivery acknowledgement from
// every node that meets the filter requirement. This means nodes
// the receive the message but do not pass the filters, will not
// send an ack.
RequestAck bool
// The timeout limits how long the query is left open. If not provided,
// then a default timeout is used based on the configuration of Serf
Timeout time.Duration
}
// DefaultQueryTimeout returns the default timeout value for a query
// Computed as GossipInterval * QueryTimeoutMult * log(N+1)
func (s *Serf) DefaultQueryTimeout() time.Duration {
n := s.memberlist.NumMembers()
timeout := s.config.MemberlistConfig.GossipInterval
timeout *= time.Duration(s.config.QueryTimeoutMult)
timeout *= time.Duration(math.Ceil(math.Log10(float64(n + 1))))
return timeout
}
// DefaultQueryParam is used to return the default query parameters
func (s *Serf) DefaultQueryParams() *QueryParam {
return &QueryParam{
FilterNodes: nil,
FilterTags: nil,
RequestAck: false,
Timeout: s.DefaultQueryTimeout(),
}
}
// encodeFilters is used to convert the filters into the wire format
func (q *QueryParam) encodeFilters() ([][]byte, error) {
var filters [][]byte
// Add the node filter
if len(q.FilterNodes) > 0 {
if buf, err := encodeFilter(filterNodeType, q.FilterNodes); err != nil {
return nil, err
} else {
filters = append(filters, buf)
}
}
// Add the tag filters
for tag, expr := range q.FilterTags {
filt := filterTag{tag, expr}
if buf, err := encodeFilter(filterTagType, &filt); err != nil {
return nil, err
} else {
filters = append(filters, buf)
}
}
return filters, nil
}
// QueryResponse is returned for each new Query. It is used to collect
// Ack's as well as responses and to provide those back to a client.
type QueryResponse struct {
// ackCh is used to send the name of a node for which we've received an ack
ackCh chan string
// deadline is the query end time (start + query timeout)
deadline time.Time
// Query ID
id uint32
// Stores the LTime of the query
lTime LamportTime
// respCh is used to send a response from a node
respCh chan NodeResponse
closed bool
closeLock sync.Mutex
}
// newQueryResponse is used to construct a new query response
func newQueryResponse(n int, q *messageQuery) *QueryResponse {
resp := &QueryResponse{
deadline: time.Now().Add(q.Timeout),
id: q.ID,
lTime: q.LTime,
respCh: make(chan NodeResponse, n),
}
if q.Ack() {
resp.ackCh = make(chan string, n)
}
return resp
}
// Close is used to close the query, which will close the underlying
// channels and prevent further deliveries
func (r *QueryResponse) Close() {
r.closeLock.Lock()
defer r.closeLock.Unlock()
if r.closed {
return
}
r.closed = true
if r.ackCh != nil {
close(r.ackCh)
}
if r.respCh != nil {
close(r.respCh)
}
}
// Deadline returns the ending deadline of the query
func (r *QueryResponse) Deadline() time.Time {
return r.deadline
}
// Finished returns if the query is finished running
func (r *QueryResponse) Finished() bool {
return r.closed || time.Now().After(r.deadline)
}
// AckCh returns a channel that can be used to listen for acks
// Channel will be closed when the query is finished. This is nil,
// if the query did not specify RequestAck.
func (r *QueryResponse) AckCh() <-chan string {
return r.ackCh
}
// ResponseCh returns a channel that can be used to listen for responses.
// Channel will be closed when the query is finished.
func (r *QueryResponse) ResponseCh() <-chan NodeResponse {
return r.respCh
}
// NodeResponse is used to represent a single response from a node
type NodeResponse struct {
From string
Payload []byte
}
// shouldProcessQuery checks if a query should be proceeded given
// a set of filers.
func (s *Serf) shouldProcessQuery(filters [][]byte) bool {
for _, filter := range filters {
switch filterType(filter[0]) {
case filterNodeType:
// Decode the filter
var nodes filterNode
if err := decodeMessage(filter[1:], &nodes); err != nil {
s.logger.Printf("[WARN] serf: failed to decode filterNodeType: %v", err)
return false
}
// Check if we are being targeted
found := false
for _, n := range nodes {
if n == s.config.NodeName {
found = true
break
}
}
if !found {
return false
}
case filterTagType:
// Decode the filter
var filt filterTag
if err := decodeMessage(filter[1:], &filt); err != nil {
s.logger.Printf("[WARN] serf: failed to decode filterTagType: %v", err)
return false
}
// Check if we match this regex
tags := s.config.Tags
matched, err := regexp.MatchString(filt.Expr, tags[filt.Tag])
if err != nil {
s.logger.Printf("[WARN] serf: failed to compile filter regex (%s): %v", filt.Expr, err)
return false
}
if !matched {
return false
}
default:
s.logger.Printf("[WARN] serf: query has unrecognized filter type: %d", filter[0])
return false
}
}
return true
}