blob: 294983f5669ef3ed8b7985486624d522f45678a9 [file] [log] [blame]
khenaidoo26721882021-08-11 17:42:52 -04001/*
2 * Copyright 2021-present Open Networking Foundation
3 *
4 * Licensed under the Apache License, Version 2.0 (the "License");
5 * you may not use this file except in compliance with the License.
6 * You may obtain a copy of the License at
7 *
8 * http://www.apache.org/licenses/LICENSE-2.0
9 *
10 * Unless required by applicable law or agreed to in writing, software
11 * distributed under the License is distributed on an "AS IS" BASIS,
12 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
13 * See the License for the specific language governing permissions and
14 * limitations under the License.
15 */
16package grpc
17
18import (
19 "context"
20 "fmt"
21 "reflect"
22 "strings"
23 "sync"
24 "time"
25
26 grpc_middleware "github.com/grpc-ecosystem/go-grpc-middleware"
27 grpc_opentracing "github.com/grpc-ecosystem/go-grpc-middleware/tracing/opentracing"
khenaidoo0927c722021-12-15 16:49:32 -050028 "github.com/jhump/protoreflect/dynamic/grpcdynamic"
29 "github.com/jhump/protoreflect/grpcreflect"
khenaidoo26721882021-08-11 17:42:52 -040030 "github.com/opencord/voltha-lib-go/v7/pkg/log"
31 "github.com/opencord/voltha-lib-go/v7/pkg/probe"
khenaidoo0927c722021-12-15 16:49:32 -050032 "github.com/opencord/voltha-protos/v5/go/adapter_service"
khenaidoob9503212021-12-08 14:22:21 -050033 "github.com/opencord/voltha-protos/v5/go/common"
khenaidooa5feb8e2021-10-19 17:29:22 -040034 "github.com/opencord/voltha-protos/v5/go/core_service"
35 "github.com/opencord/voltha-protos/v5/go/olt_inter_adapter_service"
36 "github.com/opencord/voltha-protos/v5/go/onu_inter_adapter_service"
khenaidoo26721882021-08-11 17:42:52 -040037 "google.golang.org/grpc"
khenaidoo0927c722021-12-15 16:49:32 -050038 "google.golang.org/grpc/codes"
39 rpb "google.golang.org/grpc/reflection/grpc_reflection_v1alpha"
40 "google.golang.org/grpc/status"
khenaidoo26721882021-08-11 17:42:52 -040041)
42
43type event byte
44type state byte
khenaidoo0927c722021-12-15 16:49:32 -050045type GetServiceClient func(context.Context, *grpc.ClientConn) interface{}
khenaidoo26721882021-08-11 17:42:52 -040046type RestartedHandler func(ctx context.Context, endPoint string) error
47
khenaidoo26721882021-08-11 17:42:52 -040048const (
49 grpcBackoffInitialInterval = "GRPC_BACKOFF_INITIAL_INTERVAL"
50 grpcBackoffMaxInterval = "GRPC_BACKOFF_MAX_INTERVAL"
51 grpcBackoffMaxElapsedTime = "GRPC_BACKOFF_MAX_ELAPSED_TIME"
52 grpcMonitorInterval = "GRPC_MONITOR_INTERVAL"
53)
54
55const (
56 DefaultBackoffInitialInterval = 100 * time.Millisecond
57 DefaultBackoffMaxInterval = 5 * time.Second
58 DefaultBackoffMaxElapsedTime = 0 * time.Second // No time limit
59 DefaultGRPCMonitorInterval = 5 * time.Second
60)
61
62const (
khenaidoo26721882021-08-11 17:42:52 -040063 eventConnecting = event(iota)
khenaidoo0927c722021-12-15 16:49:32 -050064 eventValidatingConnection
khenaidoo26721882021-08-11 17:42:52 -040065 eventConnected
66 eventDisconnected
67 eventStopped
68 eventError
69
70 stateConnected = state(iota)
khenaidoo0927c722021-12-15 16:49:32 -050071 stateValidatingConnection
khenaidoo26721882021-08-11 17:42:52 -040072 stateConnecting
73 stateDisconnected
74)
75
76type Client struct {
khenaidoob9503212021-12-08 14:22:21 -050077 clientEndpoint string
khenaidoo0927c722021-12-15 16:49:32 -050078 clientContextData string
khenaidoob9503212021-12-08 14:22:21 -050079 serverEndPoint string
khenaidoo0927c722021-12-15 16:49:32 -050080 remoteServiceName string
khenaidoo26721882021-08-11 17:42:52 -040081 connection *grpc.ClientConn
82 connectionLock sync.RWMutex
83 stateLock sync.RWMutex
84 state state
85 service interface{}
86 events chan event
87 onRestart RestartedHandler
88 backoffInitialInterval time.Duration
89 backoffMaxInterval time.Duration
90 backoffMaxElapsedTime time.Duration
khenaidoo26721882021-08-11 17:42:52 -040091 monitorInterval time.Duration
khenaidoo26721882021-08-11 17:42:52 -040092 done bool
khenaidoo0927c722021-12-15 16:49:32 -050093 livenessLock sync.RWMutex
khenaidoo26721882021-08-11 17:42:52 -040094 livenessCallback func(timestamp time.Time)
95}
96
97type ClientOption func(*Client)
98
khenaidoo0927c722021-12-15 16:49:32 -050099func ClientContextData(data string) ClientOption {
100 return func(args *Client) {
101 args.clientContextData = data
102 }
103}
104
105func NewClient(clientEndpoint, serverEndpoint, remoteServiceName string, onRestart RestartedHandler,
106 opts ...ClientOption) (*Client, error) {
khenaidoo26721882021-08-11 17:42:52 -0400107 c := &Client{
khenaidoob9503212021-12-08 14:22:21 -0500108 clientEndpoint: clientEndpoint,
109 serverEndPoint: serverEndpoint,
khenaidoo0927c722021-12-15 16:49:32 -0500110 remoteServiceName: remoteServiceName,
khenaidoo26721882021-08-11 17:42:52 -0400111 onRestart: onRestart,
khenaidoo0927c722021-12-15 16:49:32 -0500112 events: make(chan event, 5),
khenaidoo26721882021-08-11 17:42:52 -0400113 state: stateDisconnected,
114 backoffInitialInterval: DefaultBackoffInitialInterval,
115 backoffMaxInterval: DefaultBackoffMaxInterval,
116 backoffMaxElapsedTime: DefaultBackoffMaxElapsedTime,
117 monitorInterval: DefaultGRPCMonitorInterval,
118 }
119 for _, option := range opts {
120 option(c)
121 }
122
123 // Check for environment variables
124 if err := SetFromEnvVariable(grpcBackoffInitialInterval, &c.backoffInitialInterval); err != nil {
125 logger.Warnw(context.Background(), "failure-reading-env-variable", log.Fields{"error": err, "variable": grpcBackoffInitialInterval})
126 }
127
128 if err := SetFromEnvVariable(grpcBackoffMaxInterval, &c.backoffMaxInterval); err != nil {
129 logger.Warnw(context.Background(), "failure-reading-env-variable", log.Fields{"error": err, "variable": grpcBackoffMaxInterval})
130 }
131
132 if err := SetFromEnvVariable(grpcBackoffMaxElapsedTime, &c.backoffMaxElapsedTime); err != nil {
133 logger.Warnw(context.Background(), "failure-reading-env-variable", log.Fields{"error": err, "variable": grpcBackoffMaxElapsedTime})
134 }
135
136 if err := SetFromEnvVariable(grpcMonitorInterval, &c.monitorInterval); err != nil {
137 logger.Warnw(context.Background(), "failure-reading-env-variable", log.Fields{"error": err, "variable": grpcMonitorInterval})
138 }
139
140 logger.Infow(context.Background(), "initialized-client", log.Fields{"client": c})
141
142 // Sanity check
143 if c.backoffInitialInterval > c.backoffMaxInterval {
144 return nil, fmt.Errorf("initial retry delay %v is greater than maximum retry delay %v", c.backoffInitialInterval, c.backoffMaxInterval)
145 }
146
khenaidoo0927c722021-12-15 16:49:32 -0500147 grpc.EnableTracing = true
148
khenaidoo26721882021-08-11 17:42:52 -0400149 return c, nil
150}
151
152func (c *Client) GetClient() (interface{}, error) {
153 c.connectionLock.RLock()
154 defer c.connectionLock.RUnlock()
155 if c.service == nil {
khenaidoob9503212021-12-08 14:22:21 -0500156 return nil, fmt.Errorf("no connection to %s", c.serverEndPoint)
khenaidoo26721882021-08-11 17:42:52 -0400157 }
158 return c.service, nil
159}
160
161// GetCoreServiceClient is a helper function that returns a concrete service instead of the GetClient() API
162// which returns an interface
khenaidooa5feb8e2021-10-19 17:29:22 -0400163func (c *Client) GetCoreServiceClient() (core_service.CoreServiceClient, error) {
khenaidoo26721882021-08-11 17:42:52 -0400164 c.connectionLock.RLock()
165 defer c.connectionLock.RUnlock()
166 if c.service == nil {
khenaidoob9503212021-12-08 14:22:21 -0500167 return nil, fmt.Errorf("no core connection to %s", c.serverEndPoint)
khenaidoo26721882021-08-11 17:42:52 -0400168 }
khenaidooa5feb8e2021-10-19 17:29:22 -0400169 client, ok := c.service.(core_service.CoreServiceClient)
khenaidoo26721882021-08-11 17:42:52 -0400170 if ok {
171 return client, nil
172 }
173 return nil, fmt.Errorf("invalid-service-%s", reflect.TypeOf(c.service))
174}
175
176// GetOnuAdapterServiceClient is a helper function that returns a concrete service instead of the GetClient() API
177// which returns an interface
khenaidooa5feb8e2021-10-19 17:29:22 -0400178func (c *Client) GetOnuInterAdapterServiceClient() (onu_inter_adapter_service.OnuInterAdapterServiceClient, error) {
khenaidoo26721882021-08-11 17:42:52 -0400179 c.connectionLock.RLock()
180 defer c.connectionLock.RUnlock()
181 if c.service == nil {
khenaidoob9503212021-12-08 14:22:21 -0500182 return nil, fmt.Errorf("no child adapter connection to %s", c.serverEndPoint)
khenaidoo26721882021-08-11 17:42:52 -0400183 }
khenaidooa5feb8e2021-10-19 17:29:22 -0400184 client, ok := c.service.(onu_inter_adapter_service.OnuInterAdapterServiceClient)
khenaidoo26721882021-08-11 17:42:52 -0400185 if ok {
186 return client, nil
187 }
188 return nil, fmt.Errorf("invalid-service-%s", reflect.TypeOf(c.service))
189}
190
191// GetOltAdapterServiceClient is a helper function that returns a concrete service instead of the GetClient() API
192// which returns an interface
khenaidooa5feb8e2021-10-19 17:29:22 -0400193func (c *Client) GetOltInterAdapterServiceClient() (olt_inter_adapter_service.OltInterAdapterServiceClient, error) {
khenaidoo26721882021-08-11 17:42:52 -0400194 c.connectionLock.RLock()
195 defer c.connectionLock.RUnlock()
196 if c.service == nil {
khenaidoob9503212021-12-08 14:22:21 -0500197 return nil, fmt.Errorf("no parent adapter connection to %s", c.serverEndPoint)
khenaidoo26721882021-08-11 17:42:52 -0400198 }
khenaidooa5feb8e2021-10-19 17:29:22 -0400199 client, ok := c.service.(olt_inter_adapter_service.OltInterAdapterServiceClient)
khenaidoo26721882021-08-11 17:42:52 -0400200 if ok {
201 return client, nil
202 }
203 return nil, fmt.Errorf("invalid-service-%s", reflect.TypeOf(c.service))
204}
205
khenaidoo0927c722021-12-15 16:49:32 -0500206// GetAdapterServiceClient is a helper function that returns a concrete service instead of the GetClient() API
207// which returns an interface
208func (c *Client) GetAdapterServiceClient() (adapter_service.AdapterServiceClient, error) {
209 c.connectionLock.RLock()
210 defer c.connectionLock.RUnlock()
211 if c.service == nil {
212 return nil, fmt.Errorf("no adapter service connection to %s", c.serverEndPoint)
213 }
214 client, ok := c.service.(adapter_service.AdapterServiceClient)
215 if ok {
216 return client, nil
217 }
218 return nil, fmt.Errorf("invalid-service-%s", reflect.TypeOf(c.service))
219}
220
khenaidoo26721882021-08-11 17:42:52 -0400221func (c *Client) Reset(ctx context.Context) {
khenaidoo0927c722021-12-15 16:49:32 -0500222 logger.Debugw(ctx, "resetting-client-connection", log.Fields{"api-endpoint": c.serverEndPoint, "client": c.clientEndpoint})
khenaidoo26721882021-08-11 17:42:52 -0400223 c.stateLock.Lock()
224 defer c.stateLock.Unlock()
225 if c.state == stateConnected {
226 c.state = stateDisconnected
227 c.events <- eventDisconnected
228 }
229}
230
khenaidoo0927c722021-12-15 16:49:32 -0500231// executeWithTimeout runs a sending function (sf) along with a receiving one(rf) and returns an error, if any.
232// If the deadline d elapses first, it returns a grpc DeadlineExceeded error instead.
233func (c *Client) executeWithTimeout(sf func(*common.Connection) error, rf func() (interface{}, error), conn *common.Connection, d time.Duration) error {
234 errChan := make(chan error, 1)
235 go func() {
236 err := sf(conn)
237 logger.Debugw(context.Background(), "message-sent", log.Fields{"error": err, "qpi-endpoint": c.serverEndPoint, "client": c.clientEndpoint})
238 if err == nil {
239 response, err := rf()
240 logger.Debugw(context.Background(), "message-received", log.Fields{"error": err, "qpi-endpoint": c.serverEndPoint, "client": c.clientEndpoint, "health": response})
241 }
242 errChan <- err
243 close(errChan)
244 }()
245 t := time.NewTimer(d)
246 select {
247 case <-t.C:
248 return status.Errorf(codes.DeadlineExceeded, "timeout-on-sending-message")
249 case err := <-errChan:
250 if !t.Stop() {
251 <-t.C
khenaidoo26721882021-08-11 17:42:52 -0400252 }
253 return err
254 }
khenaidoo26721882021-08-11 17:42:52 -0400255}
256
khenaidoo0927c722021-12-15 16:49:32 -0500257func (c *Client) monitorConnection(ctx context.Context) {
258 logger.Debugw(ctx, "monitor-connection-started", log.Fields{"qpi-endpoint": c.serverEndPoint, "client": c.clientEndpoint})
khenaidoo26721882021-08-11 17:42:52 -0400259
khenaidoo0927c722021-12-15 16:49:32 -0500260 // If we exit, assume disconnected
261 defer func() {
262 c.stateLock.Lock()
263 if !c.done && (c.state == stateConnected || c.state == stateValidatingConnection) {
264 // Handle only connected state here. We need the validating state to know if we need to backoff before a retry
265 if c.state == stateConnected {
266 c.state = stateDisconnected
267 }
268 logger.Warnw(ctx, "sending-disconnect-event", log.Fields{"api-endpoint": c.serverEndPoint, "client": c.clientEndpoint, "curr-state": stateConnected, "new-state": c.state})
269 c.events <- eventDisconnected
270 } else {
271 logger.Debugw(ctx, "no-state-change-needed", log.Fields{"api-endpoint": c.serverEndPoint, "client": c.clientEndpoint, "state": c.state, "client-done": c.done})
khenaidoo26721882021-08-11 17:42:52 -0400272 }
khenaidoo0927c722021-12-15 16:49:32 -0500273 c.stateLock.Unlock()
274 logger.Debugw(ctx, "monitor-connection-ended", log.Fields{"api-endpoint": c.serverEndPoint, "client": c.clientEndpoint})
275 }()
276
277 c.connectionLock.RLock()
278 conn := c.connection
279 c.connectionLock.RUnlock()
280 if conn == nil {
281 logger.Errorw(ctx, "connection-nil", log.Fields{"api-endpoint": c.serverEndPoint, "client": c.clientEndpoint})
282 return
khenaidoo26721882021-08-11 17:42:52 -0400283 }
khenaidoo26721882021-08-11 17:42:52 -0400284
khenaidoo0927c722021-12-15 16:49:32 -0500285 // Get a new client using reflection. The server can implement any grpc service, but it
286 // needs to also implement the "StartKeepAliveStream" API
287 grpcReflectClient := grpcreflect.NewClient(ctx, rpb.NewServerReflectionClient(conn))
288 if grpcReflectClient == nil {
289 logger.Errorw(ctx, "grpc-reflect-client-nil", log.Fields{"api-endpoint": c.serverEndPoint, "client": c.clientEndpoint})
290 return
khenaidoo26721882021-08-11 17:42:52 -0400291 }
khenaidoo26721882021-08-11 17:42:52 -0400292
khenaidoo0927c722021-12-15 16:49:32 -0500293 // Get the list of services - there should be 2 services: a server reflection and the voltha service we are interested in
294 services, err := grpcReflectClient.ListServices()
295 if err != nil {
296 logger.Errorw(ctx, "list-services-error", log.Fields{"api-endpoint": c.serverEndPoint, "client": c.clientEndpoint, "error": err})
297 return
298 }
khenaidoo26721882021-08-11 17:42:52 -0400299
khenaidoo0927c722021-12-15 16:49:32 -0500300 // Filter out the service
301 logger.Debugw(ctx, "services", log.Fields{"services": services})
302 serviceOfInterest := ""
303 for _, service := range services {
304 if strings.EqualFold(service, c.remoteServiceName) {
305 serviceOfInterest = service
306 break
307 }
308 }
309 if serviceOfInterest == "" {
310 logger.Errorw(ctx, "no-service-found", log.Fields{"api-endpoint": c.serverEndPoint, "client": c.clientEndpoint, "services": services, "expected-remote-service": c.remoteServiceName})
311 return
312 }
khenaidooaa290962021-10-22 18:14:33 -0400313
khenaidoo0927c722021-12-15 16:49:32 -0500314 // Resolve the service
315 resolvedService, err := grpcReflectClient.ResolveService(serviceOfInterest)
316 if err != nil {
317 logger.Errorw(ctx, "service-error", log.Fields{"api-endpoint": c.serverEndPoint, "client": c.clientEndpoint, "service": resolvedService, "error": err})
318 return
319 }
320
321 // Find the method of interest
322 method := resolvedService.FindMethodByName("GetHealthStatus")
323 if method == nil {
324 logger.Errorw(ctx, "nil-method", log.Fields{"api-endpoint": c.serverEndPoint, "client": c.clientEndpoint, "service": resolvedService})
325 return
326 }
327 logger.Debugw(ctx, "resolved-to-method", log.Fields{"service": resolvedService.GetName(), "method": method.GetName()})
328
329 // Get a dynamic connection
330 dynamicConn := grpcdynamic.NewStub(conn)
331
332 // Get the stream and send this client information
333 streamCtx, streamDone := context.WithCancel(log.WithSpanFromContext(context.Background(), ctx))
334 defer streamDone()
335 stream, err := dynamicConn.InvokeRpcBidiStream(streamCtx, method)
336 if err != nil {
337 logger.Errorw(ctx, "stream-error", log.Fields{"api-endpoint": c.serverEndPoint, "client": c.clientEndpoint, "service": resolvedService, "error": err})
338 return
339 }
340
341 clientInfo := &common.Connection{
342 Endpoint: c.clientEndpoint,
343 ContextInfo: c.clientContextData,
344 KeepAliveInterval: int64(c.monitorInterval),
345 }
346
347 initialConnection := true
khenaidoo26721882021-08-11 17:42:52 -0400348loop:
349 for {
khenaidoo0927c722021-12-15 16:49:32 -0500350 // Let's send a keep alive message with our info
351 err := c.executeWithTimeout(
352 func(conn *common.Connection) error { return stream.SendMsg(conn) },
353 func() (interface{}, error) { return stream.RecvMsg() },
354 clientInfo,
355 c.monitorInterval)
khenaidoo26721882021-08-11 17:42:52 -0400356
khenaidoo0927c722021-12-15 16:49:32 -0500357 if err != nil {
358 // Any error means the far end is gone
359 logger.Errorw(ctx, "sending-stream-error", log.Fields{"error": err, "api-endpoint": c.serverEndPoint, "client": c.clientEndpoint, "context": stream.Context().Err()})
khenaidoo26721882021-08-11 17:42:52 -0400360 break loop
khenaidoo0927c722021-12-15 16:49:32 -0500361 }
362 // Send a connect event
363 if initialConnection {
364 logger.Debugw(ctx, "first-stream-data-sent", log.Fields{"api-endpoint": c.serverEndPoint, "client": c.clientEndpoint})
365 c.events <- eventConnected
366 initialConnection = false
367 }
368 logger.Debugw(ctx, "stream-data-sent", log.Fields{"api-endpoint": c.serverEndPoint, "client": c.clientEndpoint})
369 // Update liveness, if configured
370 c.livenessLock.RLock()
371 if c.livenessCallback != nil {
372 go c.livenessCallback(time.Now())
373 }
374 c.livenessLock.RUnlock()
khenaidoo26721882021-08-11 17:42:52 -0400375
khenaidoo0927c722021-12-15 16:49:32 -0500376 // Wait to send the next keep alive
377 keepAliveTimer := time.NewTimer(time.Duration(clientInfo.KeepAliveInterval))
378 select {
379 case <-ctx.Done():
380 logger.Warnw(ctx, "context-done", log.Fields{"api-endpont": c.serverEndPoint, "client": c.clientEndpoint})
381 break loop
382 case <-stream.Context().Done():
383 logger.Debugw(ctx, "stream-context-done", log.Fields{"api-endpoint": c.serverEndPoint, "stream-info": stream.Context(), "client": c.clientEndpoint})
384 break loop
385 case <-keepAliveTimer.C:
386 continue
khenaidoo26721882021-08-11 17:42:52 -0400387 }
388 }
khenaidoo0927c722021-12-15 16:49:32 -0500389 if stream != nil {
390 if err := stream.CloseSend(); err != nil {
391 logger.Warnw(ctx, "closing-stream-error", log.Fields{"error": err, "api-endpoint": c.serverEndPoint, "client": c.clientEndpoint})
392 }
393 }
khenaidoo26721882021-08-11 17:42:52 -0400394}
395
396// Start kicks off the adapter agent by trying to connect to the adapter
khenaidoo0927c722021-12-15 16:49:32 -0500397func (c *Client) Start(ctx context.Context, handler GetServiceClient) {
khenaidoob9503212021-12-08 14:22:21 -0500398 logger.Debugw(ctx, "Starting GRPC - Client", log.Fields{"api-endpoint": c.serverEndPoint})
khenaidoo26721882021-08-11 17:42:52 -0400399
400 // If the context contains a k8s probe then register services
401 p := probe.GetProbeFromContext(ctx)
402 if p != nil {
khenaidoob9503212021-12-08 14:22:21 -0500403 p.RegisterService(ctx, c.serverEndPoint)
khenaidoo26721882021-08-11 17:42:52 -0400404 }
405
khenaidoo0927c722021-12-15 16:49:32 -0500406 var monitorConnectionCtx context.Context
407 var monitorConnectionDone func()
khenaidoo26721882021-08-11 17:42:52 -0400408
409 initialConnection := true
410 c.events <- eventConnecting
411 backoff := NewBackoff(c.backoffInitialInterval, c.backoffMaxInterval, c.backoffMaxElapsedTime)
412 attempt := 1
413loop:
414 for {
415 select {
416 case <-ctx.Done():
khenaidoo0927c722021-12-15 16:49:32 -0500417 logger.Warnw(ctx, "context-closing", log.Fields{"api_endpoint": c.serverEndPoint, "client": c.clientEndpoint, "context": ctx})
418 c.connectionLock.Lock()
419 if !c.done {
420 c.done = true
421 c.events <- eventStopped
422 close(c.events)
423 }
424 c.connectionLock.Unlock()
425 // break loop
khenaidoo26721882021-08-11 17:42:52 -0400426 case event := <-c.events:
khenaidoo0927c722021-12-15 16:49:32 -0500427 logger.Debugw(ctx, "received-event", log.Fields{"event": event, "api-endpoint": c.serverEndPoint, "client": c.clientEndpoint})
khenaidoofe90ac32021-11-08 18:17:32 -0500428 c.connectionLock.RLock()
429 // On a client stopped, just allow the stop event to go through
430 if c.done && event != eventStopped {
431 c.connectionLock.RUnlock()
khenaidoo0927c722021-12-15 16:49:32 -0500432 logger.Debugw(ctx, "ignoring-event-on-client-stop", log.Fields{"event": event, "api-endpoint": c.serverEndPoint, "client": c.clientEndpoint})
khenaidoofe90ac32021-11-08 18:17:32 -0500433 continue
434 }
435 c.connectionLock.RUnlock()
khenaidoo26721882021-08-11 17:42:52 -0400436 switch event {
437 case eventConnecting:
khenaidoo26721882021-08-11 17:42:52 -0400438 c.stateLock.Lock()
khenaidoo0927c722021-12-15 16:49:32 -0500439 logger.Debugw(ctx, "connection-start", log.Fields{"api-endpoint": c.serverEndPoint, "attempts": attempt, "curr-state": c.state, "client": c.clientEndpoint})
khenaidoo26721882021-08-11 17:42:52 -0400440 if c.state == stateConnected {
441 c.state = stateDisconnected
442 }
443 if c.state != stateConnecting {
444 c.state = stateConnecting
445 go func() {
khenaidoo0927c722021-12-15 16:49:32 -0500446 if err := c.connectToEndpoint(ctx, p); err != nil {
khenaidoo26721882021-08-11 17:42:52 -0400447 c.stateLock.Lock()
448 c.state = stateDisconnected
449 c.stateLock.Unlock()
khenaidoo0927c722021-12-15 16:49:32 -0500450 logger.Errorw(ctx, "connection-failed", log.Fields{"api-endpoint": c.serverEndPoint, "attempt": attempt, "client": c.clientEndpoint, "error": err})
khenaidoo26721882021-08-11 17:42:52 -0400451
452 // Retry connection after a delay
453 if err = backoff.Backoff(ctx); err != nil {
454 // Context has closed or reached maximum elapsed time, if set
khenaidoo0927c722021-12-15 16:49:32 -0500455 logger.Errorw(ctx, "retry-aborted", log.Fields{"api-endpoint": c.serverEndPoint, "client": c.clientEndpoint, "error": err})
khenaidoo26721882021-08-11 17:42:52 -0400456 return
457 }
458 attempt += 1
khenaidoofe90ac32021-11-08 18:17:32 -0500459 c.connectionLock.RLock()
460 if !c.done {
461 c.events <- eventConnecting
462 }
463 c.connectionLock.RUnlock()
khenaidoo26721882021-08-11 17:42:52 -0400464 }
465 }()
466 }
467 c.stateLock.Unlock()
468
khenaidoo0927c722021-12-15 16:49:32 -0500469 case eventValidatingConnection:
470 logger.Debugw(ctx, "connection-validation", log.Fields{"api-endpoint": c.serverEndPoint, "client": c.clientEndpoint})
471 c.stateLock.Lock()
472 if c.state != stateConnected {
473 c.state = stateValidatingConnection
474 }
475 c.stateLock.Unlock()
476 monitorConnectionCtx, monitorConnectionDone = context.WithCancel(context.Background())
477 go c.monitorConnection(monitorConnectionCtx)
478
khenaidoo26721882021-08-11 17:42:52 -0400479 case eventConnected:
khenaidoo26721882021-08-11 17:42:52 -0400480 attempt = 1
khenaidoo0927c722021-12-15 16:49:32 -0500481 backoff.Reset()
khenaidoo26721882021-08-11 17:42:52 -0400482 c.stateLock.Lock()
khenaidoo0927c722021-12-15 16:49:32 -0500483 logger.Debugw(ctx, "endpoint-connected", log.Fields{"api-endpoint": c.serverEndPoint, "curr-state": c.state, "client": c.clientEndpoint})
khenaidoo26721882021-08-11 17:42:52 -0400484 if c.state != stateConnected {
khenaidoo0927c722021-12-15 16:49:32 -0500485 // Setup the service
486 c.connectionLock.RLock()
487 conn := c.connection
488 c.connectionLock.RUnlock()
489
490 subCtx, cancel := context.WithTimeout(ctx, c.backoffMaxInterval)
491 svc := handler(subCtx, conn)
492 if svc != nil {
493 c.service = svc
494 if p != nil {
495 p.UpdateStatus(ctx, c.serverEndPoint, probe.ServiceStatusRunning)
496 }
497 logger.Infow(ctx, "connected-to-endpoint", log.Fields{"api-endpoint": c.serverEndPoint, "client": c.clientEndpoint})
498 } else {
499 // Should never happen, but just in case
500 logger.Warnw(ctx, "service-is-nil", log.Fields{"api-endpoint": c.serverEndPoint, "client": c.clientEndpoint})
501 c.events <- eventDisconnected
502 }
503 cancel()
khenaidoo26721882021-08-11 17:42:52 -0400504 c.state = stateConnected
505 if initialConnection {
khenaidoo0927c722021-12-15 16:49:32 -0500506 logger.Debugw(ctx, "initial-endpoint-connection", log.Fields{"api-endpoint": c.serverEndPoint, "client": c.clientEndpoint})
khenaidoo26721882021-08-11 17:42:52 -0400507 initialConnection = false
508 } else {
khenaidoo0927c722021-12-15 16:49:32 -0500509 logger.Debugw(ctx, "endpoint-reconnection", log.Fields{"api-endpoint": c.serverEndPoint, "client": c.clientEndpoint})
khenaidoo26721882021-08-11 17:42:52 -0400510 // Trigger any callback on a restart
511 go func() {
khenaidoob9503212021-12-08 14:22:21 -0500512 err := c.onRestart(log.WithSpanFromContext(context.Background(), ctx), c.serverEndPoint)
khenaidoo26721882021-08-11 17:42:52 -0400513 if err != nil {
khenaidoo0927c722021-12-15 16:49:32 -0500514 logger.Errorw(ctx, "unable-to-restart-endpoint", log.Fields{"error": err, "api-endpoint": c.serverEndPoint, "client": c.clientEndpoint})
khenaidoo26721882021-08-11 17:42:52 -0400515 }
516 }()
517 }
518 }
519 c.stateLock.Unlock()
520
521 case eventDisconnected:
522 if p != nil {
khenaidoob9503212021-12-08 14:22:21 -0500523 p.UpdateStatus(ctx, c.serverEndPoint, probe.ServiceStatusNotReady)
khenaidoo26721882021-08-11 17:42:52 -0400524 }
khenaidoo0927c722021-12-15 16:49:32 -0500525 connectionValidationFail := false
526 c.stateLock.Lock()
527 logger.Debugw(ctx, "endpoint-disconnected", log.Fields{"api-endpoint": c.serverEndPoint, "curr-state": c.state, "client": c.clientEndpoint})
528 if c.state == stateValidatingConnection {
529 connectionValidationFail = true
530 c.state = stateDisconnected
531 }
532 c.stateLock.Unlock()
khenaidoo26721882021-08-11 17:42:52 -0400533
khenaidoo0927c722021-12-15 16:49:32 -0500534 // Stop the streaming connection
535 if monitorConnectionDone != nil {
536 monitorConnectionDone()
537 monitorConnectionDone = nil
538 }
539
540 if connectionValidationFail {
541 // Retry connection after a delay
542 if err := backoff.Backoff(ctx); err != nil {
543 // Context has closed or reached maximum elapsed time, if set
544 logger.Errorw(ctx, "retry-aborted", log.Fields{"api-endpoint": c.serverEndPoint, "client": c.clientEndpoint, "error": err})
545 return
546 }
547 }
548 c.connectionLock.RLock()
549 if !c.done {
550 c.events <- eventConnecting
551 }
552 c.connectionLock.RUnlock()
khenaidoo26721882021-08-11 17:42:52 -0400553
554 case eventStopped:
khenaidoo0927c722021-12-15 16:49:32 -0500555 logger.Debugw(ctx, "endpoint-stopped", log.Fields{"api-endpoint": c.serverEndPoint, "client": c.clientEndpoint})
556
557 if monitorConnectionDone != nil {
558 monitorConnectionDone()
559 monitorConnectionDone = nil
560 }
561 if err := c.closeConnection(ctx, p); err != nil {
562 logger.Errorw(ctx, "endpoint-closing-connection-failed", log.Fields{"api-endpoint": c.serverEndPoint, "client": c.clientEndpoint, "error": err})
563 }
khenaidoo26721882021-08-11 17:42:52 -0400564 break loop
565 case eventError:
khenaidoo0927c722021-12-15 16:49:32 -0500566 logger.Errorw(ctx, "endpoint-error-event", log.Fields{"api-endpoint": c.serverEndPoint, "client": c.clientEndpoint})
khenaidoo26721882021-08-11 17:42:52 -0400567 default:
khenaidoo0927c722021-12-15 16:49:32 -0500568 logger.Errorw(ctx, "endpoint-unknown-event", log.Fields{"api-endpoint": c.serverEndPoint, "client": c.clientEndpoint, "error": event})
khenaidoo26721882021-08-11 17:42:52 -0400569 }
570 }
571 }
khenaidoo0927c722021-12-15 16:49:32 -0500572
573 // Stop the streaming connection
574 if monitorConnectionDone != nil {
575 logger.Debugw(ctx, "closing-connection-monitoring", log.Fields{"api-endpoint": c.serverEndPoint, "client": c.clientEndpoint})
576 monitorConnectionDone()
577 }
578
579 logger.Infow(ctx, "client-stopped", log.Fields{"api-endpoint": c.serverEndPoint, "client": c.clientEndpoint})
khenaidoo26721882021-08-11 17:42:52 -0400580}
581
khenaidoo0927c722021-12-15 16:49:32 -0500582func (c *Client) connectToEndpoint(ctx context.Context, p *probe.Probe) error {
khenaidoo26721882021-08-11 17:42:52 -0400583 if p != nil {
khenaidoob9503212021-12-08 14:22:21 -0500584 p.UpdateStatus(ctx, c.serverEndPoint, probe.ServiceStatusPreparing)
khenaidoo26721882021-08-11 17:42:52 -0400585 }
586
587 c.connectionLock.Lock()
588 defer c.connectionLock.Unlock()
589
590 if c.connection != nil {
591 _ = c.connection.Close()
592 c.connection = nil
593 }
594
595 c.service = nil
596
597 // Use Interceptors to:
598 // 1. automatically inject
599 // 2. publish Open Tracing Spans by this GRPC Client
600 // 3. detect connection failure on client calls such that the reconnection process can begin
khenaidoob9503212021-12-08 14:22:21 -0500601 conn, err := grpc.Dial(c.serverEndPoint,
khenaidoo26721882021-08-11 17:42:52 -0400602 grpc.WithInsecure(),
603 grpc.WithStreamInterceptor(grpc_middleware.ChainStreamClient(
604 grpc_opentracing.StreamClientInterceptor(grpc_opentracing.WithTracer(log.ActiveTracerProxy{})),
605 )),
606 grpc.WithUnaryInterceptor(grpc_middleware.ChainUnaryClient(
607 grpc_opentracing.UnaryClientInterceptor(grpc_opentracing.WithTracer(log.ActiveTracerProxy{})),
608 )),
khenaidoo26721882021-08-11 17:42:52 -0400609 )
610
611 if err == nil {
khenaidoo0927c722021-12-15 16:49:32 -0500612 c.connection = conn
613 c.events <- eventValidatingConnection
614 return nil
615 } else {
616 logger.Warnw(ctx, "no-connection-to-endpoint", log.Fields{"api-endpoint": c.serverEndPoint, "client": c.clientEndpoint, "error": err})
khenaidoo26721882021-08-11 17:42:52 -0400617 }
khenaidoo26721882021-08-11 17:42:52 -0400618
619 if p != nil {
khenaidoob9503212021-12-08 14:22:21 -0500620 p.UpdateStatus(ctx, c.serverEndPoint, probe.ServiceStatusFailed)
khenaidoo26721882021-08-11 17:42:52 -0400621 }
khenaidoo0927c722021-12-15 16:49:32 -0500622 return fmt.Errorf("no connection to api endpoint %s", c.serverEndPoint)
khenaidoo26721882021-08-11 17:42:52 -0400623}
624
625func (c *Client) closeConnection(ctx context.Context, p *probe.Probe) error {
626 if p != nil {
khenaidoob9503212021-12-08 14:22:21 -0500627 p.UpdateStatus(ctx, c.serverEndPoint, probe.ServiceStatusStopped)
khenaidoo26721882021-08-11 17:42:52 -0400628 }
khenaidoo0927c722021-12-15 16:49:32 -0500629 logger.Infow(ctx, "client-closing-connection", log.Fields{"api-endpoint": c.serverEndPoint, "client": c.clientEndpoint})
khenaidoo26721882021-08-11 17:42:52 -0400630
631 c.connectionLock.Lock()
632 defer c.connectionLock.Unlock()
633
634 if c.connection != nil {
635 err := c.connection.Close()
khenaidoo0927c722021-12-15 16:49:32 -0500636 c.service = nil
khenaidoo26721882021-08-11 17:42:52 -0400637 c.connection = nil
638 return err
639 }
640
641 return nil
642}
643
644func (c *Client) Stop(ctx context.Context) {
khenaidoo0927c722021-12-15 16:49:32 -0500645 logger.Infow(ctx, "client-stop-request-event-received", log.Fields{"api-endpoint": c.serverEndPoint, "client": c.clientEndpoint})
khenaidoofe90ac32021-11-08 18:17:32 -0500646 c.connectionLock.Lock()
647 defer c.connectionLock.Unlock()
khenaidoo26721882021-08-11 17:42:52 -0400648 if !c.done {
khenaidoofe90ac32021-11-08 18:17:32 -0500649 c.done = true
khenaidoo26721882021-08-11 17:42:52 -0400650 c.events <- eventStopped
651 close(c.events)
khenaidoo26721882021-08-11 17:42:52 -0400652 }
khenaidoo0927c722021-12-15 16:49:32 -0500653 logger.Infow(ctx, "client-stop-request-event-sent", log.Fields{"api-endpoint": c.serverEndPoint, "client": c.clientEndpoint})
khenaidoo26721882021-08-11 17:42:52 -0400654}
655
656// SetService is used for testing only
657func (c *Client) SetService(srv interface{}) {
658 c.connectionLock.Lock()
659 defer c.connectionLock.Unlock()
660 c.service = srv
661}
662
663func (c *Client) SubscribeForLiveness(callback func(timestamp time.Time)) {
khenaidoo0927c722021-12-15 16:49:32 -0500664 c.livenessLock.Lock()
665 defer c.livenessLock.Unlock()
khenaidoo26721882021-08-11 17:42:52 -0400666 c.livenessCallback = callback
667}