blob: 307da44f922e6dc4306b3016563d0019c962f4ee [file] [log] [blame]
khenaidoo26721882021-08-11 17:42:52 -04001/*
Joey Armstrong26245a32022-12-17 21:49:06 -05002 * Copyright 2021-2022 Open Networking Foundation (ONF) and the ONF Contributors
khenaidoo26721882021-08-11 17:42:52 -04003 *
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.
nikesh.krishnanb547c1a2023-03-11 03:05:16 +0530232// If the deadline elapses first, it returns a grpc DeadlineExceeded error instead.
khenaidoo0927c722021-12-15 16:49:32 -0500233func (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
nikesh.krishnanb547c1a2023-03-11 03:05:16 +0530397func (c *Client) Start(ctx context.Context, handler GetServiceClient, retry_interceptor ...grpc.UnaryClientInterceptor) {
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() {
nikesh.krishnanb547c1a2023-03-11 03:05:16 +0530446 var err error
447 if len(retry_interceptor) > 0 {
448 err = c.connectToEndpoint(ctx, p, retry_interceptor...)
449 } else {
450 err = c.connectToEndpoint(ctx, p)
451 }
452
453 if err != nil {
khenaidoo26721882021-08-11 17:42:52 -0400454 c.stateLock.Lock()
455 c.state = stateDisconnected
456 c.stateLock.Unlock()
khenaidoo0927c722021-12-15 16:49:32 -0500457 logger.Errorw(ctx, "connection-failed", log.Fields{"api-endpoint": c.serverEndPoint, "attempt": attempt, "client": c.clientEndpoint, "error": err})
khenaidoo26721882021-08-11 17:42:52 -0400458
459 // Retry connection after a delay
460 if err = backoff.Backoff(ctx); err != nil {
461 // Context has closed or reached maximum elapsed time, if set
khenaidoo0927c722021-12-15 16:49:32 -0500462 logger.Errorw(ctx, "retry-aborted", log.Fields{"api-endpoint": c.serverEndPoint, "client": c.clientEndpoint, "error": err})
khenaidoo26721882021-08-11 17:42:52 -0400463 return
464 }
465 attempt += 1
khenaidoofe90ac32021-11-08 18:17:32 -0500466 c.connectionLock.RLock()
467 if !c.done {
468 c.events <- eventConnecting
469 }
470 c.connectionLock.RUnlock()
khenaidoo26721882021-08-11 17:42:52 -0400471 }
472 }()
473 }
474 c.stateLock.Unlock()
475
khenaidoo0927c722021-12-15 16:49:32 -0500476 case eventValidatingConnection:
477 logger.Debugw(ctx, "connection-validation", log.Fields{"api-endpoint": c.serverEndPoint, "client": c.clientEndpoint})
478 c.stateLock.Lock()
479 if c.state != stateConnected {
480 c.state = stateValidatingConnection
481 }
482 c.stateLock.Unlock()
483 monitorConnectionCtx, monitorConnectionDone = context.WithCancel(context.Background())
484 go c.monitorConnection(monitorConnectionCtx)
485
khenaidoo26721882021-08-11 17:42:52 -0400486 case eventConnected:
khenaidoo26721882021-08-11 17:42:52 -0400487 attempt = 1
khenaidoo0927c722021-12-15 16:49:32 -0500488 backoff.Reset()
khenaidoo26721882021-08-11 17:42:52 -0400489 c.stateLock.Lock()
khenaidoo0927c722021-12-15 16:49:32 -0500490 logger.Debugw(ctx, "endpoint-connected", log.Fields{"api-endpoint": c.serverEndPoint, "curr-state": c.state, "client": c.clientEndpoint})
khenaidoo26721882021-08-11 17:42:52 -0400491 if c.state != stateConnected {
khenaidoo0927c722021-12-15 16:49:32 -0500492 // Setup the service
493 c.connectionLock.RLock()
494 conn := c.connection
495 c.connectionLock.RUnlock()
496
497 subCtx, cancel := context.WithTimeout(ctx, c.backoffMaxInterval)
498 svc := handler(subCtx, conn)
499 if svc != nil {
500 c.service = svc
501 if p != nil {
502 p.UpdateStatus(ctx, c.serverEndPoint, probe.ServiceStatusRunning)
503 }
504 logger.Infow(ctx, "connected-to-endpoint", log.Fields{"api-endpoint": c.serverEndPoint, "client": c.clientEndpoint})
505 } else {
506 // Should never happen, but just in case
507 logger.Warnw(ctx, "service-is-nil", log.Fields{"api-endpoint": c.serverEndPoint, "client": c.clientEndpoint})
508 c.events <- eventDisconnected
509 }
510 cancel()
khenaidoo26721882021-08-11 17:42:52 -0400511 c.state = stateConnected
512 if initialConnection {
khenaidoo0927c722021-12-15 16:49:32 -0500513 logger.Debugw(ctx, "initial-endpoint-connection", log.Fields{"api-endpoint": c.serverEndPoint, "client": c.clientEndpoint})
khenaidoo26721882021-08-11 17:42:52 -0400514 initialConnection = false
515 } else {
khenaidoo0927c722021-12-15 16:49:32 -0500516 logger.Debugw(ctx, "endpoint-reconnection", log.Fields{"api-endpoint": c.serverEndPoint, "client": c.clientEndpoint})
khenaidoo26721882021-08-11 17:42:52 -0400517 // Trigger any callback on a restart
518 go func() {
khenaidoob9503212021-12-08 14:22:21 -0500519 err := c.onRestart(log.WithSpanFromContext(context.Background(), ctx), c.serverEndPoint)
khenaidoo26721882021-08-11 17:42:52 -0400520 if err != nil {
khenaidoo0927c722021-12-15 16:49:32 -0500521 logger.Errorw(ctx, "unable-to-restart-endpoint", log.Fields{"error": err, "api-endpoint": c.serverEndPoint, "client": c.clientEndpoint})
khenaidoo26721882021-08-11 17:42:52 -0400522 }
523 }()
524 }
525 }
526 c.stateLock.Unlock()
527
528 case eventDisconnected:
529 if p != nil {
khenaidoob9503212021-12-08 14:22:21 -0500530 p.UpdateStatus(ctx, c.serverEndPoint, probe.ServiceStatusNotReady)
khenaidoo26721882021-08-11 17:42:52 -0400531 }
khenaidoo0927c722021-12-15 16:49:32 -0500532 connectionValidationFail := false
533 c.stateLock.Lock()
534 logger.Debugw(ctx, "endpoint-disconnected", log.Fields{"api-endpoint": c.serverEndPoint, "curr-state": c.state, "client": c.clientEndpoint})
535 if c.state == stateValidatingConnection {
536 connectionValidationFail = true
537 c.state = stateDisconnected
538 }
539 c.stateLock.Unlock()
khenaidoo26721882021-08-11 17:42:52 -0400540
khenaidoo0927c722021-12-15 16:49:32 -0500541 // Stop the streaming connection
542 if monitorConnectionDone != nil {
543 monitorConnectionDone()
544 monitorConnectionDone = nil
545 }
546
547 if connectionValidationFail {
548 // Retry connection after a delay
549 if err := backoff.Backoff(ctx); err != nil {
550 // Context has closed or reached maximum elapsed time, if set
551 logger.Errorw(ctx, "retry-aborted", log.Fields{"api-endpoint": c.serverEndPoint, "client": c.clientEndpoint, "error": err})
552 return
553 }
554 }
555 c.connectionLock.RLock()
556 if !c.done {
557 c.events <- eventConnecting
558 }
559 c.connectionLock.RUnlock()
khenaidoo26721882021-08-11 17:42:52 -0400560
561 case eventStopped:
khenaidoo0927c722021-12-15 16:49:32 -0500562 logger.Debugw(ctx, "endpoint-stopped", log.Fields{"api-endpoint": c.serverEndPoint, "client": c.clientEndpoint})
563
564 if monitorConnectionDone != nil {
565 monitorConnectionDone()
566 monitorConnectionDone = nil
567 }
568 if err := c.closeConnection(ctx, p); err != nil {
569 logger.Errorw(ctx, "endpoint-closing-connection-failed", log.Fields{"api-endpoint": c.serverEndPoint, "client": c.clientEndpoint, "error": err})
570 }
khenaidoo26721882021-08-11 17:42:52 -0400571 break loop
572 case eventError:
khenaidoo0927c722021-12-15 16:49:32 -0500573 logger.Errorw(ctx, "endpoint-error-event", log.Fields{"api-endpoint": c.serverEndPoint, "client": c.clientEndpoint})
khenaidoo26721882021-08-11 17:42:52 -0400574 default:
khenaidoo0927c722021-12-15 16:49:32 -0500575 logger.Errorw(ctx, "endpoint-unknown-event", log.Fields{"api-endpoint": c.serverEndPoint, "client": c.clientEndpoint, "error": event})
khenaidoo26721882021-08-11 17:42:52 -0400576 }
577 }
578 }
khenaidoo0927c722021-12-15 16:49:32 -0500579
580 // Stop the streaming connection
581 if monitorConnectionDone != nil {
582 logger.Debugw(ctx, "closing-connection-monitoring", log.Fields{"api-endpoint": c.serverEndPoint, "client": c.clientEndpoint})
583 monitorConnectionDone()
584 }
585
586 logger.Infow(ctx, "client-stopped", log.Fields{"api-endpoint": c.serverEndPoint, "client": c.clientEndpoint})
khenaidoo26721882021-08-11 17:42:52 -0400587}
588
nikesh.krishnanb547c1a2023-03-11 03:05:16 +0530589func (c *Client) connectToEndpoint(ctx context.Context, p *probe.Probe, retry_interceptor ...grpc.UnaryClientInterceptor) error {
khenaidoo26721882021-08-11 17:42:52 -0400590 if p != nil {
khenaidoob9503212021-12-08 14:22:21 -0500591 p.UpdateStatus(ctx, c.serverEndPoint, probe.ServiceStatusPreparing)
khenaidoo26721882021-08-11 17:42:52 -0400592 }
593
594 c.connectionLock.Lock()
595 defer c.connectionLock.Unlock()
596
597 if c.connection != nil {
598 _ = c.connection.Close()
599 c.connection = nil
600 }
601
602 c.service = nil
603
604 // Use Interceptors to:
605 // 1. automatically inject
606 // 2. publish Open Tracing Spans by this GRPC Client
607 // 3. detect connection failure on client calls such that the reconnection process can begin
nikesh.krishnanb547c1a2023-03-11 03:05:16 +0530608 interceptor_opts := []grpc.UnaryClientInterceptor{grpc_opentracing.UnaryClientInterceptor(grpc_opentracing.WithTracer(log.ActiveTracerProxy{}))}
609
610 if len(retry_interceptor) > 0 {
611 interceptor_opts = append(interceptor_opts, retry_interceptor...)
612 }
khenaidoob9503212021-12-08 14:22:21 -0500613 conn, err := grpc.Dial(c.serverEndPoint,
khenaidoo26721882021-08-11 17:42:52 -0400614 grpc.WithInsecure(),
615 grpc.WithStreamInterceptor(grpc_middleware.ChainStreamClient(
616 grpc_opentracing.StreamClientInterceptor(grpc_opentracing.WithTracer(log.ActiveTracerProxy{})),
617 )),
nikesh.krishnanb547c1a2023-03-11 03:05:16 +0530618 grpc.WithUnaryInterceptor(grpc_middleware.ChainUnaryClient(interceptor_opts...)),
khenaidoo26721882021-08-11 17:42:52 -0400619 )
620
621 if err == nil {
khenaidoo0927c722021-12-15 16:49:32 -0500622 c.connection = conn
623 c.events <- eventValidatingConnection
624 return nil
625 } else {
626 logger.Warnw(ctx, "no-connection-to-endpoint", log.Fields{"api-endpoint": c.serverEndPoint, "client": c.clientEndpoint, "error": err})
khenaidoo26721882021-08-11 17:42:52 -0400627 }
khenaidoo26721882021-08-11 17:42:52 -0400628
629 if p != nil {
khenaidoob9503212021-12-08 14:22:21 -0500630 p.UpdateStatus(ctx, c.serverEndPoint, probe.ServiceStatusFailed)
khenaidoo26721882021-08-11 17:42:52 -0400631 }
khenaidoo0927c722021-12-15 16:49:32 -0500632 return fmt.Errorf("no connection to api endpoint %s", c.serverEndPoint)
khenaidoo26721882021-08-11 17:42:52 -0400633}
634
635func (c *Client) closeConnection(ctx context.Context, p *probe.Probe) error {
636 if p != nil {
khenaidoob9503212021-12-08 14:22:21 -0500637 p.UpdateStatus(ctx, c.serverEndPoint, probe.ServiceStatusStopped)
khenaidoo26721882021-08-11 17:42:52 -0400638 }
khenaidoo0927c722021-12-15 16:49:32 -0500639 logger.Infow(ctx, "client-closing-connection", log.Fields{"api-endpoint": c.serverEndPoint, "client": c.clientEndpoint})
khenaidoo26721882021-08-11 17:42:52 -0400640
641 c.connectionLock.Lock()
642 defer c.connectionLock.Unlock()
643
644 if c.connection != nil {
645 err := c.connection.Close()
khenaidoo0927c722021-12-15 16:49:32 -0500646 c.service = nil
khenaidoo26721882021-08-11 17:42:52 -0400647 c.connection = nil
648 return err
649 }
650
651 return nil
652}
653
654func (c *Client) Stop(ctx context.Context) {
khenaidoo0927c722021-12-15 16:49:32 -0500655 logger.Infow(ctx, "client-stop-request-event-received", log.Fields{"api-endpoint": c.serverEndPoint, "client": c.clientEndpoint})
khenaidoofe90ac32021-11-08 18:17:32 -0500656 c.connectionLock.Lock()
657 defer c.connectionLock.Unlock()
khenaidoo26721882021-08-11 17:42:52 -0400658 if !c.done {
khenaidoofe90ac32021-11-08 18:17:32 -0500659 c.done = true
khenaidoo26721882021-08-11 17:42:52 -0400660 c.events <- eventStopped
661 close(c.events)
khenaidoo26721882021-08-11 17:42:52 -0400662 }
khenaidoo0927c722021-12-15 16:49:32 -0500663 logger.Infow(ctx, "client-stop-request-event-sent", log.Fields{"api-endpoint": c.serverEndPoint, "client": c.clientEndpoint})
khenaidoo26721882021-08-11 17:42:52 -0400664}
665
666// SetService is used for testing only
667func (c *Client) SetService(srv interface{}) {
668 c.connectionLock.Lock()
669 defer c.connectionLock.Unlock()
670 c.service = srv
671}
672
673func (c *Client) SubscribeForLiveness(callback func(timestamp time.Time)) {
khenaidoo0927c722021-12-15 16:49:32 -0500674 c.livenessLock.Lock()
675 defer c.livenessLock.Unlock()
khenaidoo26721882021-08-11 17:42:52 -0400676 c.livenessCallback = callback
677}