blob: 9b66d85851007ca6d87725dff17b615ec469dc8a [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"
28 "github.com/opencord/voltha-lib-go/v7/pkg/log"
29 "github.com/opencord/voltha-lib-go/v7/pkg/probe"
khenaidoob9503212021-12-08 14:22:21 -050030 "github.com/opencord/voltha-protos/v5/go/common"
khenaidooa5feb8e2021-10-19 17:29:22 -040031 "github.com/opencord/voltha-protos/v5/go/core_service"
32 "github.com/opencord/voltha-protos/v5/go/olt_inter_adapter_service"
33 "github.com/opencord/voltha-protos/v5/go/onu_inter_adapter_service"
khenaidoo26721882021-08-11 17:42:52 -040034 "google.golang.org/grpc"
35 "google.golang.org/grpc/keepalive"
36)
37
38type event byte
39type state byte
khenaidoob9503212021-12-08 14:22:21 -050040type SetAndTestServiceHandler func(context.Context, *grpc.ClientConn, *common.Connection) interface{}
khenaidoo26721882021-08-11 17:42:52 -040041type RestartedHandler func(ctx context.Context, endPoint string) error
42
43type contextKey string
44
45func (c contextKey) String() string {
46 return string(c)
47}
48
49var (
50 grpcMonitorContextKey = contextKey("grpc-monitor")
51)
52
53const (
54 grpcBackoffInitialInterval = "GRPC_BACKOFF_INITIAL_INTERVAL"
55 grpcBackoffMaxInterval = "GRPC_BACKOFF_MAX_INTERVAL"
56 grpcBackoffMaxElapsedTime = "GRPC_BACKOFF_MAX_ELAPSED_TIME"
57 grpcMonitorInterval = "GRPC_MONITOR_INTERVAL"
58)
59
60const (
61 DefaultBackoffInitialInterval = 100 * time.Millisecond
62 DefaultBackoffMaxInterval = 5 * time.Second
63 DefaultBackoffMaxElapsedTime = 0 * time.Second // No time limit
64 DefaultGRPCMonitorInterval = 5 * time.Second
65)
66
67const (
68 connectionErrorSubString = "SubConns are in TransientFailure"
69 connectionClosedSubstring = "client connection is closing"
70 connectionError = "connection error"
71 connectionSystemNotReady = "system is not ready"
72)
73
74const (
75 eventConnecting = event(iota)
76 eventConnected
77 eventDisconnected
78 eventStopped
79 eventError
80
81 stateConnected = state(iota)
82 stateConnecting
83 stateDisconnected
84)
85
86type Client struct {
khenaidoob9503212021-12-08 14:22:21 -050087 clientEndpoint string
88 serverEndPoint string
khenaidoo26721882021-08-11 17:42:52 -040089 connection *grpc.ClientConn
90 connectionLock sync.RWMutex
91 stateLock sync.RWMutex
92 state state
93 service interface{}
94 events chan event
95 onRestart RestartedHandler
96 backoffInitialInterval time.Duration
97 backoffMaxInterval time.Duration
98 backoffMaxElapsedTime time.Duration
khenaidoo26721882021-08-11 17:42:52 -040099 monitorInterval time.Duration
khenaidoo26721882021-08-11 17:42:52 -0400100 done bool
101 livenessCallback func(timestamp time.Time)
102}
103
104type ClientOption func(*Client)
105
khenaidoob9503212021-12-08 14:22:21 -0500106func NewClient(clientEndpoint, serverEndpoint string, onRestart RestartedHandler, 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,
khenaidoo26721882021-08-11 17:42:52 -0400110 onRestart: onRestart,
111 events: make(chan event, 1),
112 state: stateDisconnected,
113 backoffInitialInterval: DefaultBackoffInitialInterval,
114 backoffMaxInterval: DefaultBackoffMaxInterval,
115 backoffMaxElapsedTime: DefaultBackoffMaxElapsedTime,
116 monitorInterval: DefaultGRPCMonitorInterval,
117 }
118 for _, option := range opts {
119 option(c)
120 }
121
122 // Check for environment variables
123 if err := SetFromEnvVariable(grpcBackoffInitialInterval, &c.backoffInitialInterval); err != nil {
124 logger.Warnw(context.Background(), "failure-reading-env-variable", log.Fields{"error": err, "variable": grpcBackoffInitialInterval})
125 }
126
127 if err := SetFromEnvVariable(grpcBackoffMaxInterval, &c.backoffMaxInterval); err != nil {
128 logger.Warnw(context.Background(), "failure-reading-env-variable", log.Fields{"error": err, "variable": grpcBackoffMaxInterval})
129 }
130
131 if err := SetFromEnvVariable(grpcBackoffMaxElapsedTime, &c.backoffMaxElapsedTime); err != nil {
132 logger.Warnw(context.Background(), "failure-reading-env-variable", log.Fields{"error": err, "variable": grpcBackoffMaxElapsedTime})
133 }
134
135 if err := SetFromEnvVariable(grpcMonitorInterval, &c.monitorInterval); err != nil {
136 logger.Warnw(context.Background(), "failure-reading-env-variable", log.Fields{"error": err, "variable": grpcMonitorInterval})
137 }
138
139 logger.Infow(context.Background(), "initialized-client", log.Fields{"client": c})
140
141 // Sanity check
142 if c.backoffInitialInterval > c.backoffMaxInterval {
143 return nil, fmt.Errorf("initial retry delay %v is greater than maximum retry delay %v", c.backoffInitialInterval, c.backoffMaxInterval)
144 }
145
146 return c, nil
147}
148
149func (c *Client) GetClient() (interface{}, error) {
150 c.connectionLock.RLock()
151 defer c.connectionLock.RUnlock()
152 if c.service == nil {
khenaidoob9503212021-12-08 14:22:21 -0500153 return nil, fmt.Errorf("no connection to %s", c.serverEndPoint)
khenaidoo26721882021-08-11 17:42:52 -0400154 }
155 return c.service, nil
156}
157
158// GetCoreServiceClient is a helper function that returns a concrete service instead of the GetClient() API
159// which returns an interface
khenaidooa5feb8e2021-10-19 17:29:22 -0400160func (c *Client) GetCoreServiceClient() (core_service.CoreServiceClient, error) {
khenaidoo26721882021-08-11 17:42:52 -0400161 c.connectionLock.RLock()
162 defer c.connectionLock.RUnlock()
163 if c.service == nil {
khenaidoob9503212021-12-08 14:22:21 -0500164 return nil, fmt.Errorf("no core connection to %s", c.serverEndPoint)
khenaidoo26721882021-08-11 17:42:52 -0400165 }
khenaidooa5feb8e2021-10-19 17:29:22 -0400166 client, ok := c.service.(core_service.CoreServiceClient)
khenaidoo26721882021-08-11 17:42:52 -0400167 if ok {
168 return client, nil
169 }
170 return nil, fmt.Errorf("invalid-service-%s", reflect.TypeOf(c.service))
171}
172
173// GetOnuAdapterServiceClient is a helper function that returns a concrete service instead of the GetClient() API
174// which returns an interface
khenaidooa5feb8e2021-10-19 17:29:22 -0400175func (c *Client) GetOnuInterAdapterServiceClient() (onu_inter_adapter_service.OnuInterAdapterServiceClient, error) {
khenaidoo26721882021-08-11 17:42:52 -0400176 c.connectionLock.RLock()
177 defer c.connectionLock.RUnlock()
178 if c.service == nil {
khenaidoob9503212021-12-08 14:22:21 -0500179 return nil, fmt.Errorf("no child adapter connection to %s", c.serverEndPoint)
khenaidoo26721882021-08-11 17:42:52 -0400180 }
khenaidooa5feb8e2021-10-19 17:29:22 -0400181 client, ok := c.service.(onu_inter_adapter_service.OnuInterAdapterServiceClient)
khenaidoo26721882021-08-11 17:42:52 -0400182 if ok {
183 return client, nil
184 }
185 return nil, fmt.Errorf("invalid-service-%s", reflect.TypeOf(c.service))
186}
187
188// GetOltAdapterServiceClient is a helper function that returns a concrete service instead of the GetClient() API
189// which returns an interface
khenaidooa5feb8e2021-10-19 17:29:22 -0400190func (c *Client) GetOltInterAdapterServiceClient() (olt_inter_adapter_service.OltInterAdapterServiceClient, error) {
khenaidoo26721882021-08-11 17:42:52 -0400191 c.connectionLock.RLock()
192 defer c.connectionLock.RUnlock()
193 if c.service == nil {
khenaidoob9503212021-12-08 14:22:21 -0500194 return nil, fmt.Errorf("no parent adapter connection to %s", c.serverEndPoint)
khenaidoo26721882021-08-11 17:42:52 -0400195 }
khenaidooa5feb8e2021-10-19 17:29:22 -0400196 client, ok := c.service.(olt_inter_adapter_service.OltInterAdapterServiceClient)
khenaidoo26721882021-08-11 17:42:52 -0400197 if ok {
198 return client, nil
199 }
200 return nil, fmt.Errorf("invalid-service-%s", reflect.TypeOf(c.service))
201}
202
203func (c *Client) Reset(ctx context.Context) {
khenaidoob9503212021-12-08 14:22:21 -0500204 logger.Debugw(ctx, "resetting-client-connection", log.Fields{"endpoint": c.serverEndPoint})
khenaidoo26721882021-08-11 17:42:52 -0400205 c.stateLock.Lock()
206 defer c.stateLock.Unlock()
207 if c.state == stateConnected {
208 c.state = stateDisconnected
209 c.events <- eventDisconnected
210 }
211}
212
213func (c *Client) clientInterceptor(ctx context.Context, method string, req interface{}, reply interface{},
214 cc *grpc.ClientConn, invoker grpc.UnaryInvoker, opts ...grpc.CallOption) error {
215 // Nothing to do before intercepting the call
216 err := invoker(ctx, method, req, reply, cc, opts...)
217 // On connection failure, start the reconnect process depending on the error response
218 if err != nil {
khenaidoob9503212021-12-08 14:22:21 -0500219 logger.Errorw(ctx, "received-error", log.Fields{"error": err, "context": ctx, "endpoint": c.serverEndPoint})
khenaidoo26721882021-08-11 17:42:52 -0400220 if strings.Contains(err.Error(), connectionErrorSubString) ||
221 strings.Contains(err.Error(), connectionError) ||
222 strings.Contains(err.Error(), connectionSystemNotReady) ||
223 isGrpcMonitorKeyPresentInContext(ctx) {
224 c.stateLock.Lock()
225 if c.state == stateConnected {
khenaidoo26721882021-08-11 17:42:52 -0400226 c.state = stateDisconnected
khenaidoob9503212021-12-08 14:22:21 -0500227 logger.Warnw(context.Background(), "sending-disconnect-event", log.Fields{"endpoint": c.serverEndPoint, "error": err, "curr-state": stateConnected, "new-state": c.state})
khenaidoo26721882021-08-11 17:42:52 -0400228 c.events <- eventDisconnected
229 }
230 c.stateLock.Unlock()
231 } else if strings.Contains(err.Error(), connectionClosedSubstring) {
khenaidoob9503212021-12-08 14:22:21 -0500232 logger.Errorw(context.Background(), "invalid-client-connection-closed", log.Fields{"endpoint": c.serverEndPoint, "error": err})
khenaidoo26721882021-08-11 17:42:52 -0400233 }
234 return err
235 }
236 // Update activity on success only
237 c.updateActivity(ctx)
238 return nil
239}
240
khenaidoob9503212021-12-08 14:22:21 -0500241// updateActivity updates the liveness channel
khenaidoo26721882021-08-11 17:42:52 -0400242func (c *Client) updateActivity(ctx context.Context) {
khenaidoob9503212021-12-08 14:22:21 -0500243 logger.Debugw(ctx, "update-activity", log.Fields{"api-endpoint": c.serverEndPoint})
khenaidoo26721882021-08-11 17:42:52 -0400244
khenaidoob9503212021-12-08 14:22:21 -0500245 // Update liveness only in connected state
246 if c.livenessCallback != nil {
247 c.stateLock.RLock()
248 if c.state == stateConnected {
249 c.livenessCallback(time.Now())
khenaidoo26721882021-08-11 17:42:52 -0400250 }
khenaidoob9503212021-12-08 14:22:21 -0500251 c.stateLock.RUnlock()
khenaidoo26721882021-08-11 17:42:52 -0400252 }
253}
254
255func WithGrpcMonitorContext(ctx context.Context, name string) context.Context {
256 ctx = context.WithValue(ctx, grpcMonitorContextKey, name)
257 return ctx
258}
259
260func isGrpcMonitorKeyPresentInContext(ctx context.Context) bool {
261 if ctx != nil {
262 _, present := ctx.Value(grpcMonitorContextKey).(string)
263 return present
264 }
265 return false
266}
267
268// monitorActivity monitors the activity on the gRPC connection. If there are no activity after a specified
269// timeout, it will send a default API request on that connection. If the connection is good then nothing
270// happens. If it's bad this will trigger reconnection attempts.
271func (c *Client) monitorActivity(ctx context.Context, handler SetAndTestServiceHandler) {
khenaidoob9503212021-12-08 14:22:21 -0500272 logger.Infow(ctx, "start-activity-monitor", log.Fields{"endpoint": c.serverEndPoint})
khenaidoo26721882021-08-11 17:42:52 -0400273
khenaidooaa290962021-10-22 18:14:33 -0400274 grpcMonitorCheckRunning := false
275 var grpcMonitorCheckRunningLock sync.RWMutex
276
khenaidoo26721882021-08-11 17:42:52 -0400277 // Interval to wait for no activity before probing the connection
278 timeout := c.monitorInterval
279loop:
280 for {
281 timeoutTimer := time.NewTimer(timeout)
282 select {
283
khenaidoob9503212021-12-08 14:22:21 -0500284 case <-ctx.Done():
285 // Stop and drain timer
khenaidoo26721882021-08-11 17:42:52 -0400286 if !timeoutTimer.Stop() {
khenaidooaa290962021-10-22 18:14:33 -0400287 select {
288 case <-timeoutTimer.C:
289 default:
290 }
khenaidoo26721882021-08-11 17:42:52 -0400291 }
khenaidoo26721882021-08-11 17:42:52 -0400292 break loop
293
294 case <-timeoutTimer.C:
295 // Trigger an activity check if the state is connected. If the state is not connected then there is already
296 // a backoff retry mechanism in place to retry establishing connection.
297 c.stateLock.RLock()
khenaidooaa290962021-10-22 18:14:33 -0400298 grpcMonitorCheckRunningLock.RLock()
299 runCheck := (c.state == stateConnected) && !grpcMonitorCheckRunning
300 grpcMonitorCheckRunningLock.RUnlock()
khenaidoo26721882021-08-11 17:42:52 -0400301 c.stateLock.RUnlock()
302 if runCheck {
303 go func() {
khenaidooaa290962021-10-22 18:14:33 -0400304 grpcMonitorCheckRunningLock.Lock()
305 if grpcMonitorCheckRunning {
306 grpcMonitorCheckRunningLock.Unlock()
khenaidoob9503212021-12-08 14:22:21 -0500307 logger.Debugw(ctx, "connection-check-already-in-progress", log.Fields{"api-endpoint": c.serverEndPoint})
khenaidooaa290962021-10-22 18:14:33 -0400308 return
309 }
310 grpcMonitorCheckRunning = true
311 grpcMonitorCheckRunningLock.Unlock()
312
khenaidoob9503212021-12-08 14:22:21 -0500313 logger.Debugw(ctx, "connection-check-start", log.Fields{"api-endpoint": c.serverEndPoint})
khenaidoo26721882021-08-11 17:42:52 -0400314 subCtx, cancel := context.WithTimeout(ctx, c.backoffMaxInterval)
315 defer cancel()
316 subCtx = WithGrpcMonitorContext(subCtx, "grpc-monitor")
317 c.connectionLock.RLock()
318 defer c.connectionLock.RUnlock()
319 if c.connection != nil {
khenaidoob9503212021-12-08 14:22:21 -0500320 response := handler(subCtx, c.connection, &common.Connection{Endpoint: c.clientEndpoint, KeepAliveInterval: int64(c.monitorInterval)})
321 logger.Debugw(ctx, "connection-check-response", log.Fields{"api-endpoint": c.serverEndPoint, "up": response != nil})
khenaidoo26721882021-08-11 17:42:52 -0400322 }
khenaidooaa290962021-10-22 18:14:33 -0400323 grpcMonitorCheckRunningLock.Lock()
324 grpcMonitorCheckRunning = false
325 grpcMonitorCheckRunningLock.Unlock()
khenaidoo26721882021-08-11 17:42:52 -0400326 }()
327 }
328 }
329 }
khenaidoob9503212021-12-08 14:22:21 -0500330 logger.Infow(ctx, "activity-monitor-stopping", log.Fields{"endpoint": c.serverEndPoint})
khenaidoo26721882021-08-11 17:42:52 -0400331}
332
333// Start kicks off the adapter agent by trying to connect to the adapter
334func (c *Client) Start(ctx context.Context, handler SetAndTestServiceHandler) {
khenaidoob9503212021-12-08 14:22:21 -0500335 logger.Debugw(ctx, "Starting GRPC - Client", log.Fields{"api-endpoint": c.serverEndPoint})
khenaidoo26721882021-08-11 17:42:52 -0400336
337 // If the context contains a k8s probe then register services
338 p := probe.GetProbeFromContext(ctx)
339 if p != nil {
khenaidoob9503212021-12-08 14:22:21 -0500340 p.RegisterService(ctx, c.serverEndPoint)
khenaidoo26721882021-08-11 17:42:52 -0400341 }
342
khenaidoob9503212021-12-08 14:22:21 -0500343 // Enable activity check
344 go c.monitorActivity(ctx, handler)
khenaidoo26721882021-08-11 17:42:52 -0400345
346 initialConnection := true
347 c.events <- eventConnecting
348 backoff := NewBackoff(c.backoffInitialInterval, c.backoffMaxInterval, c.backoffMaxElapsedTime)
349 attempt := 1
350loop:
351 for {
352 select {
353 case <-ctx.Done():
khenaidoob9503212021-12-08 14:22:21 -0500354 logger.Debugw(ctx, "context-closing", log.Fields{"endpoint": c.serverEndPoint})
khenaidoofe90ac32021-11-08 18:17:32 -0500355 break loop
khenaidoo26721882021-08-11 17:42:52 -0400356 case event := <-c.events:
khenaidoob9503212021-12-08 14:22:21 -0500357 logger.Debugw(ctx, "received-event", log.Fields{"event": event, "endpoint": c.serverEndPoint})
khenaidoofe90ac32021-11-08 18:17:32 -0500358 c.connectionLock.RLock()
359 // On a client stopped, just allow the stop event to go through
360 if c.done && event != eventStopped {
361 c.connectionLock.RUnlock()
khenaidoob9503212021-12-08 14:22:21 -0500362 logger.Debugw(ctx, "ignoring-event-on-client-stop", log.Fields{"event": event, "endpoint": c.serverEndPoint})
khenaidoofe90ac32021-11-08 18:17:32 -0500363 continue
364 }
365 c.connectionLock.RUnlock()
khenaidoo26721882021-08-11 17:42:52 -0400366 switch event {
367 case eventConnecting:
khenaidoo26721882021-08-11 17:42:52 -0400368 c.stateLock.Lock()
khenaidoob9503212021-12-08 14:22:21 -0500369 logger.Debugw(ctx, "connection-start", log.Fields{"endpoint": c.serverEndPoint, "attempts": attempt, "curr-state": c.state})
khenaidoo26721882021-08-11 17:42:52 -0400370 if c.state == stateConnected {
371 c.state = stateDisconnected
372 }
373 if c.state != stateConnecting {
374 c.state = stateConnecting
375 go func() {
376 if err := c.connectToEndpoint(ctx, handler, p); err != nil {
377 c.stateLock.Lock()
378 c.state = stateDisconnected
379 c.stateLock.Unlock()
khenaidoob9503212021-12-08 14:22:21 -0500380 logger.Errorw(ctx, "connection-failed", log.Fields{"endpoint": c.serverEndPoint, "attempt": attempt, "error": err})
khenaidoo26721882021-08-11 17:42:52 -0400381
382 // Retry connection after a delay
383 if err = backoff.Backoff(ctx); err != nil {
384 // Context has closed or reached maximum elapsed time, if set
khenaidoob9503212021-12-08 14:22:21 -0500385 logger.Errorw(ctx, "retry-aborted", log.Fields{"endpoint": c.serverEndPoint, "error": err})
khenaidoo26721882021-08-11 17:42:52 -0400386 return
387 }
388 attempt += 1
khenaidoofe90ac32021-11-08 18:17:32 -0500389 c.connectionLock.RLock()
390 if !c.done {
391 c.events <- eventConnecting
392 }
393 c.connectionLock.RUnlock()
khenaidoo26721882021-08-11 17:42:52 -0400394 } else {
395 backoff.Reset()
396 }
397 }()
398 }
399 c.stateLock.Unlock()
400
401 case eventConnected:
khenaidoo26721882021-08-11 17:42:52 -0400402 attempt = 1
403 c.stateLock.Lock()
khenaidoob9503212021-12-08 14:22:21 -0500404 logger.Debugw(ctx, "endpoint-connected", log.Fields{"endpoint": c.serverEndPoint, "curr-state": c.state})
khenaidoo26721882021-08-11 17:42:52 -0400405 if c.state != stateConnected {
406 c.state = stateConnected
407 if initialConnection {
khenaidoob9503212021-12-08 14:22:21 -0500408 logger.Debugw(ctx, "initial-endpoint-connection", log.Fields{"endpoint": c.serverEndPoint})
khenaidoo26721882021-08-11 17:42:52 -0400409 initialConnection = false
410 } else {
khenaidoob9503212021-12-08 14:22:21 -0500411 logger.Debugw(ctx, "endpoint-reconnection", log.Fields{"endpoint": c.serverEndPoint})
khenaidoo26721882021-08-11 17:42:52 -0400412 // Trigger any callback on a restart
413 go func() {
khenaidoob9503212021-12-08 14:22:21 -0500414 err := c.onRestart(log.WithSpanFromContext(context.Background(), ctx), c.serverEndPoint)
khenaidoo26721882021-08-11 17:42:52 -0400415 if err != nil {
khenaidoob9503212021-12-08 14:22:21 -0500416 logger.Errorw(ctx, "unable-to-restart-endpoint", log.Fields{"error": err, "endpoint": c.serverEndPoint})
khenaidoo26721882021-08-11 17:42:52 -0400417 }
418 }()
419 }
420 }
421 c.stateLock.Unlock()
422
423 case eventDisconnected:
424 if p != nil {
khenaidoob9503212021-12-08 14:22:21 -0500425 p.UpdateStatus(ctx, c.serverEndPoint, probe.ServiceStatusNotReady)
khenaidoo26721882021-08-11 17:42:52 -0400426 }
khenaidooaa290962021-10-22 18:14:33 -0400427 c.stateLock.RLock()
khenaidoob9503212021-12-08 14:22:21 -0500428 logger.Debugw(ctx, "endpoint-disconnected", log.Fields{"endpoint": c.serverEndPoint, "curr-state": c.state})
khenaidooaa290962021-10-22 18:14:33 -0400429 c.stateLock.RUnlock()
khenaidoo26721882021-08-11 17:42:52 -0400430
431 // Try to connect again
432 c.events <- eventConnecting
433
434 case eventStopped:
khenaidoob9503212021-12-08 14:22:21 -0500435 logger.Debugw(ctx, "endPoint-stopped", log.Fields{"adapter": c.serverEndPoint})
khenaidoo26721882021-08-11 17:42:52 -0400436 go func() {
437 if err := c.closeConnection(ctx, p); err != nil {
khenaidoob9503212021-12-08 14:22:21 -0500438 logger.Errorw(ctx, "endpoint-closing-connection-failed", log.Fields{"endpoint": c.serverEndPoint, "error": err})
khenaidoo26721882021-08-11 17:42:52 -0400439 }
440 }()
441 break loop
442 case eventError:
khenaidoob9503212021-12-08 14:22:21 -0500443 logger.Errorw(ctx, "endpoint-error-event", log.Fields{"endpoint": c.serverEndPoint})
khenaidoo26721882021-08-11 17:42:52 -0400444 default:
khenaidoob9503212021-12-08 14:22:21 -0500445 logger.Errorw(ctx, "endpoint-unknown-event", log.Fields{"endpoint": c.serverEndPoint, "error": event})
khenaidoo26721882021-08-11 17:42:52 -0400446 }
447 }
448 }
khenaidoob9503212021-12-08 14:22:21 -0500449 logger.Infow(ctx, "endpoint-stopped", log.Fields{"endpoint": c.serverEndPoint})
khenaidoo26721882021-08-11 17:42:52 -0400450}
451
452func (c *Client) connectToEndpoint(ctx context.Context, handler SetAndTestServiceHandler, p *probe.Probe) error {
453 if p != nil {
khenaidoob9503212021-12-08 14:22:21 -0500454 p.UpdateStatus(ctx, c.serverEndPoint, probe.ServiceStatusPreparing)
khenaidoo26721882021-08-11 17:42:52 -0400455 }
456
457 c.connectionLock.Lock()
458 defer c.connectionLock.Unlock()
459
460 if c.connection != nil {
461 _ = c.connection.Close()
462 c.connection = nil
463 }
464
465 c.service = nil
466
467 // Use Interceptors to:
468 // 1. automatically inject
469 // 2. publish Open Tracing Spans by this GRPC Client
470 // 3. detect connection failure on client calls such that the reconnection process can begin
khenaidoob9503212021-12-08 14:22:21 -0500471 conn, err := grpc.Dial(c.serverEndPoint,
khenaidoo26721882021-08-11 17:42:52 -0400472 grpc.WithInsecure(),
473 grpc.WithStreamInterceptor(grpc_middleware.ChainStreamClient(
474 grpc_opentracing.StreamClientInterceptor(grpc_opentracing.WithTracer(log.ActiveTracerProxy{})),
475 )),
476 grpc.WithUnaryInterceptor(grpc_middleware.ChainUnaryClient(
477 grpc_opentracing.UnaryClientInterceptor(grpc_opentracing.WithTracer(log.ActiveTracerProxy{})),
478 )),
479 grpc.WithUnaryInterceptor(c.clientInterceptor),
480 // Set keealive parameter - use default grpc values
481 grpc.WithKeepaliveParams(keepalive.ClientParameters{
482 Time: c.monitorInterval,
483 Timeout: c.backoffMaxInterval,
484 PermitWithoutStream: true,
485 }),
486 )
487
488 if err == nil {
489 subCtx, cancel := context.WithTimeout(ctx, c.backoffMaxInterval)
490 defer cancel()
khenaidoob9503212021-12-08 14:22:21 -0500491 svc := handler(subCtx, conn, &common.Connection{Endpoint: c.clientEndpoint, KeepAliveInterval: int64(c.monitorInterval)})
khenaidoo26721882021-08-11 17:42:52 -0400492 if svc != nil {
493 c.connection = conn
494 c.service = svc
495 if p != nil {
khenaidoob9503212021-12-08 14:22:21 -0500496 p.UpdateStatus(ctx, c.serverEndPoint, probe.ServiceStatusRunning)
khenaidoo26721882021-08-11 17:42:52 -0400497 }
khenaidoob9503212021-12-08 14:22:21 -0500498 logger.Infow(ctx, "connected-to-endpoint", log.Fields{"endpoint": c.serverEndPoint})
khenaidoo26721882021-08-11 17:42:52 -0400499 c.events <- eventConnected
500 return nil
501 }
502 }
503 logger.Warnw(ctx, "Failed to connect to endpoint",
504 log.Fields{
khenaidoob9503212021-12-08 14:22:21 -0500505 "endpoint": c.serverEndPoint,
khenaidoo26721882021-08-11 17:42:52 -0400506 "error": err,
507 })
508
509 if p != nil {
khenaidoob9503212021-12-08 14:22:21 -0500510 p.UpdateStatus(ctx, c.serverEndPoint, probe.ServiceStatusFailed)
khenaidoo26721882021-08-11 17:42:52 -0400511 }
khenaidoob9503212021-12-08 14:22:21 -0500512 return fmt.Errorf("no connection to endpoint %s", c.serverEndPoint)
khenaidoo26721882021-08-11 17:42:52 -0400513}
514
515func (c *Client) closeConnection(ctx context.Context, p *probe.Probe) error {
516 if p != nil {
khenaidoob9503212021-12-08 14:22:21 -0500517 p.UpdateStatus(ctx, c.serverEndPoint, probe.ServiceStatusStopped)
khenaidoo26721882021-08-11 17:42:52 -0400518 }
519
520 c.connectionLock.Lock()
521 defer c.connectionLock.Unlock()
522
523 if c.connection != nil {
524 err := c.connection.Close()
525 c.connection = nil
526 return err
527 }
528
529 return nil
530}
531
532func (c *Client) Stop(ctx context.Context) {
khenaidoofe90ac32021-11-08 18:17:32 -0500533 c.connectionLock.Lock()
534 defer c.connectionLock.Unlock()
khenaidoo26721882021-08-11 17:42:52 -0400535 if !c.done {
khenaidoofe90ac32021-11-08 18:17:32 -0500536 c.done = true
khenaidoo26721882021-08-11 17:42:52 -0400537 c.events <- eventStopped
538 close(c.events)
khenaidoo26721882021-08-11 17:42:52 -0400539 }
khenaidoob9503212021-12-08 14:22:21 -0500540 logger.Infow(ctx, "client-stopped", log.Fields{"endpoint": c.serverEndPoint})
khenaidoo26721882021-08-11 17:42:52 -0400541}
542
543// SetService is used for testing only
544func (c *Client) SetService(srv interface{}) {
545 c.connectionLock.Lock()
546 defer c.connectionLock.Unlock()
547 c.service = srv
548}
549
550func (c *Client) SubscribeForLiveness(callback func(timestamp time.Time)) {
551 c.livenessCallback = callback
552}