-
Notifications
You must be signed in to change notification settings - Fork 5
/
mcpServer.go
394 lines (338 loc) · 11.4 KB
/
mcpServer.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
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
// Copyright Istio Authors
//
// Licensed under the Apache License, Version 2.0 (the "License");
// you may not use this file except in compliance with the License.
// You may obtain a copy of the License at
//
// http://www.apache.org/licenses/LICENSE-2.0
//
// Unless required by applicable law or agreed to in writing, software
// distributed under the License is distributed on an "AS IS" BASIS,
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
// See the License for the specific language governing permissions and
// limitations under the License.
package server
import (
"io"
"net"
"strconv"
"sync"
"sync/atomic"
"time"
"github.com/gogo/protobuf/types"
middleware "github.com/grpc-ecosystem/go-grpc-middleware"
prometheus "github.com/grpc-ecosystem/go-grpc-prometheus"
"github.com/istio-ecosystem/consul-mcp/pkg/serviceregistry"
"github.com/istio-ecosystem/consul-mcp/pkg/serviceregistry/consul"
"google.golang.org/grpc"
"google.golang.org/grpc/codes"
"google.golang.org/grpc/peer"
"google.golang.org/grpc/status"
"istio.io/api/mcp/v1alpha1"
mcp "istio.io/api/mcp/v1alpha1"
istio "istio.io/api/networking/v1alpha3"
"istio.io/pkg/log"
)
const (
serviceEntriesType = "istio/networking/v1alpha3/serviceentries"
// debounceAfter is the delay added to events to wait after a registry event for debouncing.
// This will delay the push by at least this interval, plus the time getting subsequent events.
// If no change is detected the push will happen, otherwise we'll keep delaying until things settle.
debounceAfter = 100 * time.Millisecond
// debounceMax is the maximum time to wait for events while debouncing.
// Defaults to 10 seconds. If events keep showing up with no break for this time, we'll trigger a push.
debounceMax = 10 * time.Second
)
var (
// Tracks connections, increment on each new connection.
connectionNumber = int64(0)
)
// Connection holds information about connected client.
type Connection struct {
sync.RWMutex
// ConID is the connection identifier, used as a key in the connection table.
// Currently based on the sink id, peer addr and a counter.
ConID string
// PeerAddr is the address of the client, from network layer.
PeerAddr string
// Time of connection, for debugging
Connect time.Time
// Sending on this channel results in a push.
//pushChannel chan *Event
pushChannel chan *ChangeEvent
// MCP stream implement this interface
stream mcp.ResourceSource_EstablishResourceStreamServer
// LastResponse stores the last response nonce to each sink
LastResponse map[string]string
}
type ChangeEvent struct {
}
type ConsulMcpServer struct {
listeningAddress string
consulAddress string
grpcServer *grpc.Server
GRPCListener net.Listener
registry serviceregistry.Registry
// mcpClients are the connected MCP sink.
mcpClients map[string]*Connection
mcpClientsMutex sync.RWMutex
stopchan chan struct{}
}
func NewServer(listeningAddress string, consulAddress string) *ConsulMcpServer {
consulMcpServer := &ConsulMcpServer{
listeningAddress: listeningAddress,
consulAddress: consulAddress,
mcpClients: make(map[string]*Connection),
stopchan: make(chan struct{}),
}
return consulMcpServer
}
// Start the gRPC MCP server
func (s *ConsulMcpServer) Start() error {
log.Info("Start MCP server")
log.Infof("Watch Consul at %s", s.consulAddress)
if err := s.watchRegistry(s.stopchan); err != nil {
log.Fatala(err)
return err
}
log.Infof("Listen on %v", s.listeningAddress)
if err := s.startGrpcServer(); err != nil {
log.Fatala(err)
return err
}
return nil
}
func (s *ConsulMcpServer) Stop() {
s.stopchan <- struct{}{}
s.grpcServer.Stop()
}
func (s *ConsulMcpServer) watchRegistry(stop <-chan struct{}) error {
var err error
s.registry, err = consul.NewController(s.consulAddress)
if err != nil {
return err
}
s.registry.AppendServiceChangeHandler(func() {
s.mcpClientsMutex.Lock()
defer s.mcpClientsMutex.Unlock()
for _, con := range s.mcpClients {
con.pushChannel <- &ChangeEvent{}
}
})
//@todo gracefully close the registry controller
s.registry.Run(stop)
return nil
}
func (s *ConsulMcpServer) startGrpcServer() error {
grpcOptions := s.grpcServerOptions()
s.grpcServer = grpc.NewServer(grpcOptions...)
mcp.RegisterResourceSourceServer(s.grpcServer, s)
listener, err := net.Listen("tcp", s.listeningAddress)
if err != nil {
return err
}
if err := s.grpcServer.Serve(listener); err != nil {
log.Fatala(err)
return err
}
return nil
}
func (s *ConsulMcpServer) EstablishResourceStream(stream mcp.ResourceSource_EstablishResourceStreamServer) error {
var timeChan <-chan time.Time
var startDebounce time.Time
var event *ChangeEvent
var lastResourceUpdateTime time.Time
pushCounter := 0
debouncedEvents := 0
con := s.newConnection(stream)
s.addConnection(con)
defer s.removeConnection(con)
go con.receive()
for {
select {
case e := <-con.pushChannel:
// Incremental push is not supported, so just use the latest event to represent the merged event
event = e
log.Debugf("Receive event from push chanel : %v", e)
lastResourceUpdateTime = time.Now()
if debouncedEvents == 0 {
log.Debugf("This is the first debounced event")
startDebounce = lastResourceUpdateTime
}
timeChan = time.After(debounceAfter)
debouncedEvents++
case <-timeChan:
log.Debugf("Receive event from time chanel")
eventDelay := time.Since(startDebounce)
quietTime := time.Since(lastResourceUpdateTime)
// it has been too long since the first debounced event or quiet enough since the last debounced event
if eventDelay >= debounceMax || quietTime >= debounceAfter {
if event != nil {
pushCounter++
log.Infof("Push debounce stable[%d] %d: %v since last change, %v since last push",
pushCounter, debouncedEvents,
quietTime, eventDelay)
serviceEntries, err := s.registry.ServiceEntries()
if err != nil {
log.Errorf("RESOURCE:%s: RESPONSE ERROR %s %v", serviceEntriesType, con.ConID, err)
}
resources, err := constructResoures(serviceEntries)
if err != nil {
log.Errorf("RESOURCE:%s: RESPONSE ERROR %s %v", serviceEntriesType, con.ConID, err)
}
response := &mcp.Resources{
Collection: serviceEntriesType,
Resources: resources,
Nonce: time.Now().String(),
Incremental: false,
}
con.send(response)
event = nil
debouncedEvents = 0
}
} else {
timeChan = time.After(debounceAfter - quietTime)
}
}
}
}
func constructResoures(serviceEntries []*istio.ServiceEntry) ([]mcp.Resource, error) {
resources := make([]mcp.Resource, 0)
for _, serviceEntry := range serviceEntries {
seAny, err := types.MarshalAny(serviceEntry)
if err != nil {
return resources, err
}
resources = append(resources, mcp.Resource{
Body: seAny,
Metadata: &v1alpha1.Metadata{
Name: serviceEntry.Hosts[0],
Version: "v1",
},
})
}
return resources, nil
}
func (con *Connection) receive() {
for {
req, err := con.stream.Recv()
if err != nil {
if isExpectedGRPCError(err) {
log.Infof("%s terminated %v", con.ConID, err)
}
log.Errorf("%s terminated with error: %v", con.ConID, err)
}
if con.shouldResponse(req) {
// This MCP server only supports ServiceEntry
if req.Collection != serviceEntriesType {
response := &mcp.Resources{
Incremental: false,
Collection: req.GetCollection(),
Nonce: time.Now().String(),
}
con.send(response)
} else {
// Send a change event to the connection channel to trigger a push to the client
con.pushChannel <- &ChangeEvent{}
}
}
}
}
// isExpectedGRPCError checks a gRPC error code and determines whether it is an expected error when
// things are operating normally. This is basically capturing when the client disconnects.
func isExpectedGRPCError(err error) bool {
if err == io.EOF {
return true
}
s := status.Convert(err)
if s.Code() == codes.Canceled || s.Code() == codes.DeadlineExceeded {
return true
}
if s.Code() == codes.Unavailable && s.Message() == "client disconnected" {
return true
}
return false
}
func (con *Connection) send(response *mcp.Resources) {
con.Lock()
err := con.stream.Send(response)
con.LastResponse[response.Collection] = response.Nonce
con.Unlock()
if err != nil {
log.Errorf("RESOURCE:%s: RESPONSE ERROR %s %v", response.Collection, con.ConID, err)
} else {
log.Infof("RESOURCE:%s: RESPONSE SUCCESS", response.Collection)
}
}
func (con *Connection) shouldResponse(req *mcp.RequestResources) bool {
// This is the first request, we should response.
if req.ResponseNonce == "" {
log.Infof("RESOURCE:%s: REQ %s initial request", req.Collection, con.ConID)
return true
}
// The presence of ErrorDetail means that this is a NACK and the previous response is invalid. We can't resent the
// same resources. It's perhaps caused by an error in the source code, we should check the error log and fix the code
// in that case.
if req.ErrorDetail != nil {
errCode := codes.Code(req.ErrorDetail.Code)
log.Errorf("RESOURCE:%s: ACK ERROR %s %s:%s", req.Collection, con.ConID, errCode.String(), req.ErrorDetail.GetMessage())
return false
}
previousRespone, ok := con.LastResponse[req.Collection]
// MCP Server does not have information about this collection, but MCP Sink client sends response nonce - either
// because MCP Server is restarted or MCP Sink client disconnects and reconnects.
// We should always respond with the current resource.
if !ok {
log.Warnf("RESOURCE:%s: RECONNECT %s %s", req.Collection, con.ConID, req.ResponseNonce)
return true
}
// If there is mismatch in the nonce, that is a case of expired/stale nonce.
// A nonce becomes stale following a newer nonce being sent to Envoy.
if req.ResponseNonce != previousRespone {
log.Warnf("RESOURCE:%s: REQ %s Expired nonce received %s, sent %s", req.Collection,
con.ConID, req.ResponseNonce, previousRespone)
return false
}
// If it comes here, that means nonce match. This an ACK. we should not response unless there is a change in MCP Server side.
log.Infof("RESOURCE:%s: ACK %s %s", req.Collection, con.ConID, req.ResponseNonce)
return false
}
func (s *ConsulMcpServer) grpcServerOptions() []grpc.ServerOption {
interceptors := []grpc.UnaryServerInterceptor{
// setup server prometheus monitoring (as final interceptor in chain)
prometheus.UnaryServerInterceptor,
}
grpcOptions := []grpc.ServerOption{
grpc.UnaryInterceptor(middleware.ChainUnaryServer(interceptors...)),
}
return grpcOptions
}
func (s *ConsulMcpServer) newConnection(stream mcp.ResourceSource_EstablishResourceStreamServer) *Connection {
ctx := stream.Context()
peerAddr := "0.0.0.0"
if peerInfo, ok := peer.FromContext(ctx); ok {
peerAddr = peerInfo.Addr.String()
}
id := atomic.AddInt64(&connectionNumber, 1)
conId := peerAddr + "-" + strconv.FormatInt(id, 10)
return &Connection{
PeerAddr: peerAddr,
Connect: time.Now(),
ConID: conId,
pushChannel: make(chan *ChangeEvent, 100),
stream: stream,
LastResponse: make(map[string]string),
}
}
func (s *ConsulMcpServer) addConnection(con *Connection) {
s.mcpClientsMutex.Lock()
defer s.mcpClientsMutex.Unlock()
s.mcpClients[con.ConID] = con
log.Infof("Receive connection from client: %s", con.ConID)
}
func (s *ConsulMcpServer) removeConnection(con *Connection) {
s.mcpClientsMutex.Lock()
defer s.mcpClientsMutex.Unlock()
delete(s.mcpClients, con.ConID)
log.Infof("Remove connection from client: %s", con.ConID)
}