pkg/grpc/client.go

/*
 * Copyright 2021-2024 Open Networking Foundation (ONF) and the ONF Contributors
 *
 * 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 grpc

import (
	"context"
	"fmt"
	"reflect"
	"strings"
	"sync"
	"time"

	grpc_middleware "github.com/grpc-ecosystem/go-grpc-middleware"
	grpc_opentracing "github.com/grpc-ecosystem/go-grpc-middleware/tracing/opentracing"
	grpc_prometheus "github.com/grpc-ecosystem/go-grpc-prometheus"
	"github.com/jhump/protoreflect/dynamic/grpcdynamic"
	"github.com/jhump/protoreflect/grpcreflect"
	"github.com/opencord/voltha-lib-go/v7/pkg/log"
	"github.com/opencord/voltha-lib-go/v7/pkg/probe"
	"github.com/opencord/voltha-protos/v5/go/adapter_service"
	"github.com/opencord/voltha-protos/v5/go/common"
	"github.com/opencord/voltha-protos/v5/go/core_service"
	"github.com/opencord/voltha-protos/v5/go/olt_inter_adapter_service"
	"github.com/opencord/voltha-protos/v5/go/onu_inter_adapter_service"
	"google.golang.org/grpc"
	"google.golang.org/grpc/codes"
	rpb "google.golang.org/grpc/reflection/grpc_reflection_v1alpha"
	"google.golang.org/grpc/status"
)

type event byte
type state byte
type GetServiceClient func(context.Context, *grpc.ClientConn) interface{}
type RestartedHandler func(ctx context.Context, endPoint string) error

const (
	grpcBackoffInitialInterval = "GRPC_BACKOFF_INITIAL_INTERVAL"
	grpcBackoffMaxInterval     = "GRPC_BACKOFF_MAX_INTERVAL"
	grpcBackoffMaxElapsedTime  = "GRPC_BACKOFF_MAX_ELAPSED_TIME"
	grpcMonitorInterval        = "GRPC_MONITOR_INTERVAL"
)

const (
	DefaultBackoffInitialInterval = 100 * time.Millisecond
	DefaultBackoffMaxInterval     = 5 * time.Second
	DefaultBackoffMaxElapsedTime  = 0 * time.Second // No time limit
	DefaultGRPCMonitorInterval    = 5 * time.Second
)

const (
	// [VOL-5434] Setting max receive message size to 20 MB,
	// Default value of 'defaultServerMaxReceiveMessageSize' is 4 MB
	grpcRecvMsgSizeLimit = 20
)

const (
	eventConnecting = event(iota)
	eventValidatingConnection
	eventConnected
	eventDisconnected
	eventStopped
	eventError

	stateConnected = state(iota)
	stateValidatingConnection
	stateConnecting
	stateDisconnected
)

type Client struct {
	clientEndpoint         string
	clientContextData      string
	serverEndPoint         string
	remoteServiceName      string
	connection             *grpc.ClientConn
	connectionLock         sync.RWMutex
	stateLock              sync.RWMutex
	state                  state
	service                interface{}
	events                 chan event
	onRestart              RestartedHandler
	backoffInitialInterval time.Duration
	backoffMaxInterval     time.Duration
	backoffMaxElapsedTime  time.Duration
	monitorInterval        time.Duration
	done                   bool
	livenessLock           sync.RWMutex
	livenessCallback       func(timestamp time.Time)
}

type ClientOption func(*Client)

func ClientContextData(data string) ClientOption {
	return func(args *Client) {
		args.clientContextData = data
	}
}

func NewClient(clientEndpoint, serverEndpoint, remoteServiceName string, onRestart RestartedHandler,
	opts ...ClientOption) (*Client, error) {
	c := &Client{
		clientEndpoint:         clientEndpoint,
		serverEndPoint:         serverEndpoint,
		remoteServiceName:      remoteServiceName,
		onRestart:              onRestart,
		events:                 make(chan event, 5),
		state:                  stateDisconnected,
		backoffInitialInterval: DefaultBackoffInitialInterval,
		backoffMaxInterval:     DefaultBackoffMaxInterval,
		backoffMaxElapsedTime:  DefaultBackoffMaxElapsedTime,
		monitorInterval:        DefaultGRPCMonitorInterval,
	}
	for _, option := range opts {
		option(c)
	}

	// Check for environment variables
	if err := SetFromEnvVariable(grpcBackoffInitialInterval, &c.backoffInitialInterval); err != nil {
		logger.Warnw(context.Background(), "failure-reading-env-variable", log.Fields{"error": err, "variable": grpcBackoffInitialInterval})
	}

	if err := SetFromEnvVariable(grpcBackoffMaxInterval, &c.backoffMaxInterval); err != nil {
		logger.Warnw(context.Background(), "failure-reading-env-variable", log.Fields{"error": err, "variable": grpcBackoffMaxInterval})
	}

	if err := SetFromEnvVariable(grpcBackoffMaxElapsedTime, &c.backoffMaxElapsedTime); err != nil {
		logger.Warnw(context.Background(), "failure-reading-env-variable", log.Fields{"error": err, "variable": grpcBackoffMaxElapsedTime})
	}

	if err := SetFromEnvVariable(grpcMonitorInterval, &c.monitorInterval); err != nil {
		logger.Warnw(context.Background(), "failure-reading-env-variable", log.Fields{"error": err, "variable": grpcMonitorInterval})
	}

	logger.Infow(context.Background(), "initialized-client", log.Fields{"client": c})

	// Sanity check
	if c.backoffInitialInterval > c.backoffMaxInterval {
		return nil, fmt.Errorf("initial retry delay %v is greater than maximum retry delay %v", c.backoffInitialInterval, c.backoffMaxInterval)
	}

	grpc.EnableTracing = true

	return c, nil
}

func (c *Client) GetClient() (interface{}, error) {
	c.connectionLock.RLock()
	defer c.connectionLock.RUnlock()
	if c.service == nil {
		return nil, fmt.Errorf("no connection to %s", c.serverEndPoint)
	}
	return c.service, nil
}

// GetCoreServiceClient is a helper function that returns a concrete service instead of the GetClient() API
// which returns an interface
func (c *Client) GetCoreServiceClient() (core_service.CoreServiceClient, error) {
	c.connectionLock.RLock()
	defer c.connectionLock.RUnlock()
	if c.service == nil {
		return nil, fmt.Errorf("no core connection to %s", c.serverEndPoint)
	}
	client, ok := c.service.(core_service.CoreServiceClient)
	if ok {
		return client, nil
	}
	return nil, fmt.Errorf("invalid-service-%s", reflect.TypeOf(c.service))
}

// GetOnuAdapterServiceClient is a helper function that returns a concrete service instead of the GetClient() API
// which returns an interface
func (c *Client) GetOnuInterAdapterServiceClient() (onu_inter_adapter_service.OnuInterAdapterServiceClient, error) {
	c.connectionLock.RLock()
	defer c.connectionLock.RUnlock()
	if c.service == nil {
		return nil, fmt.Errorf("no child adapter connection to %s", c.serverEndPoint)
	}
	client, ok := c.service.(onu_inter_adapter_service.OnuInterAdapterServiceClient)
	if ok {
		return client, nil
	}
	return nil, fmt.Errorf("invalid-service-%s", reflect.TypeOf(c.service))
}

// GetOltAdapterServiceClient is a helper function that returns a concrete service instead of the GetClient() API
// which returns an interface
func (c *Client) GetOltInterAdapterServiceClient() (olt_inter_adapter_service.OltInterAdapterServiceClient, error) {
	c.connectionLock.RLock()
	defer c.connectionLock.RUnlock()
	if c.service == nil {
		return nil, fmt.Errorf("no parent adapter connection to %s", c.serverEndPoint)
	}
	client, ok := c.service.(olt_inter_adapter_service.OltInterAdapterServiceClient)
	if ok {
		return client, nil
	}
	return nil, fmt.Errorf("invalid-service-%s", reflect.TypeOf(c.service))
}

// GetAdapterServiceClient is a helper function that returns a concrete service instead of the GetClient() API
// which returns an interface
func (c *Client) GetAdapterServiceClient() (adapter_service.AdapterServiceClient, error) {
	c.connectionLock.RLock()
	defer c.connectionLock.RUnlock()
	if c.service == nil {
		return nil, fmt.Errorf("no adapter service connection to %s", c.serverEndPoint)
	}
	client, ok := c.service.(adapter_service.AdapterServiceClient)
	if ok {
		return client, nil
	}
	return nil, fmt.Errorf("invalid-service-%s", reflect.TypeOf(c.service))
}

func (c *Client) Reset(ctx context.Context) {
	logger.Debugw(ctx, "resetting-client-connection", log.Fields{"api-endpoint": c.serverEndPoint, "client": c.clientEndpoint})
	c.stateLock.Lock()
	defer c.stateLock.Unlock()
	if c.state == stateConnected {
		c.state = stateDisconnected
		c.events <- eventDisconnected
	}
}

// executeWithTimeout runs a sending function (sf) along with a receiving one(rf) and returns an error, if any.
// If the deadline  elapses first, it returns a grpc DeadlineExceeded error instead.
func (c *Client) executeWithTimeout(sf func(*common.Connection) error, rf func() (interface{}, error), conn *common.Connection, d time.Duration) error {
	errChan := make(chan error, 1)
	go func() {
		err := sf(conn)
		logger.Debugw(context.Background(), "message-sent", log.Fields{"error": err, "qpi-endpoint": c.serverEndPoint, "client": c.clientEndpoint})
		if err == nil {
			response, err := rf()
			logger.Debugw(context.Background(), "message-received", log.Fields{"error": err, "qpi-endpoint": c.serverEndPoint, "client": c.clientEndpoint, "health": response})
		}
		errChan <- err
		close(errChan)
	}()
	t := time.NewTimer(d)
	select {
	case <-t.C:
		return status.Errorf(codes.DeadlineExceeded, "timeout-on-sending-message")
	case err := <-errChan:
		if !t.Stop() {
			<-t.C
		}
		return err
	}
}

func (c *Client) monitorConnection(ctx context.Context) {
	logger.Debugw(ctx, "monitor-connection-started", log.Fields{"qpi-endpoint": c.serverEndPoint, "client": c.clientEndpoint})

	// If we exit, assume disconnected
	defer func() {
		c.stateLock.Lock()
		if !c.done && (c.state == stateConnected || c.state == stateValidatingConnection) {
			// Handle only connected state here.  We need the validating state to know if we need to backoff before a retry
			if c.state == stateConnected {
				c.state = stateDisconnected
			}
			logger.Warnw(ctx, "sending-disconnect-event", log.Fields{"api-endpoint": c.serverEndPoint, "client": c.clientEndpoint, "curr-state": stateConnected, "new-state": c.state})
			c.events <- eventDisconnected
		} else {
			logger.Debugw(ctx, "no-state-change-needed", log.Fields{"api-endpoint": c.serverEndPoint, "client": c.clientEndpoint, "state": c.state, "client-done": c.done})
		}
		c.stateLock.Unlock()
		logger.Debugw(ctx, "monitor-connection-ended", log.Fields{"api-endpoint": c.serverEndPoint, "client": c.clientEndpoint})
	}()

	c.connectionLock.RLock()
	conn := c.connection
	c.connectionLock.RUnlock()
	if conn == nil {
		logger.Errorw(ctx, "connection-nil", log.Fields{"api-endpoint": c.serverEndPoint, "client": c.clientEndpoint})
		return
	}

	// Get a new client using reflection. The server can implement any grpc service, but it
	// needs to also implement the "StartKeepAliveStream" API
	grpcReflectClient := grpcreflect.NewClient(ctx, rpb.NewServerReflectionClient(conn))
	if grpcReflectClient == nil {
		logger.Errorw(ctx, "grpc-reflect-client-nil", log.Fields{"api-endpoint": c.serverEndPoint, "client": c.clientEndpoint})
		return
	}

	// Get the list of services - there should be 2 services: a server reflection and the voltha service we are interested in
	services, err := grpcReflectClient.ListServices()
	if err != nil {
		logger.Errorw(ctx, "list-services-error", log.Fields{"api-endpoint": c.serverEndPoint, "client": c.clientEndpoint, "error": err})
		return
	}

	// Filter out the service
	logger.Debugw(ctx, "services", log.Fields{"services": services})
	serviceOfInterest := ""
	for _, service := range services {
		if strings.EqualFold(service, c.remoteServiceName) {
			serviceOfInterest = service
			break
		}
	}
	if serviceOfInterest == "" {
		logger.Errorw(ctx, "no-service-found", log.Fields{"api-endpoint": c.serverEndPoint, "client": c.clientEndpoint, "services": services, "expected-remote-service": c.remoteServiceName})
		return
	}

	// Resolve the service
	resolvedService, err := grpcReflectClient.ResolveService(serviceOfInterest)
	if err != nil {
		logger.Errorw(ctx, "service-error", log.Fields{"api-endpoint": c.serverEndPoint, "client": c.clientEndpoint, "service": resolvedService, "error": err})
		return
	}

	// Find the method of interest
	method := resolvedService.FindMethodByName("GetHealthStatus")
	if method == nil {
		logger.Errorw(ctx, "nil-method", log.Fields{"api-endpoint": c.serverEndPoint, "client": c.clientEndpoint, "service": resolvedService})
		return
	}
	logger.Debugw(ctx, "resolved-to-method", log.Fields{"service": resolvedService.GetName(), "method": method.GetName()})

	// Get a dynamic connection
	dynamicConn := grpcdynamic.NewStub(conn)

	// Get the stream and send this client information
	streamCtx, streamDone := context.WithCancel(log.WithSpanFromContext(context.Background(), ctx))
	defer streamDone()
	stream, err := dynamicConn.InvokeRpcBidiStream(streamCtx, method)
	if err != nil {
		logger.Errorw(ctx, "stream-error", log.Fields{"api-endpoint": c.serverEndPoint, "client": c.clientEndpoint, "service": resolvedService, "error": err})
		return
	}

	clientInfo := &common.Connection{
		Endpoint:          c.clientEndpoint,
		ContextInfo:       c.clientContextData,
		KeepAliveInterval: int64(c.monitorInterval),
	}

	initialConnection := true
loop:
	for {
		// Let's send a keep alive message with our info
		err := c.executeWithTimeout(
			func(conn *common.Connection) error { return stream.SendMsg(conn) },
			func() (interface{}, error) { return stream.RecvMsg() },
			clientInfo,
			c.monitorInterval)

		if err != nil {
			// Any error means the far end is gone
			logger.Errorw(ctx, "sending-stream-error", log.Fields{"error": err, "api-endpoint": c.serverEndPoint, "client": c.clientEndpoint, "context": stream.Context().Err()})
			break loop
		}
		// Send a connect event
		if initialConnection {
			logger.Debugw(ctx, "first-stream-data-sent", log.Fields{"api-endpoint": c.serverEndPoint, "client": c.clientEndpoint})
			c.events <- eventConnected
			initialConnection = false
		}
		logger.Debugw(ctx, "stream-data-sent", log.Fields{"api-endpoint": c.serverEndPoint, "client": c.clientEndpoint})
		// Update liveness, if configured
		c.livenessLock.RLock()
		if c.livenessCallback != nil {
			go c.livenessCallback(time.Now())
		}
		c.livenessLock.RUnlock()

		// Wait to send the next keep alive
		keepAliveTimer := time.NewTimer(time.Duration(clientInfo.KeepAliveInterval))
		select {
		case <-ctx.Done():
			logger.Warnw(ctx, "context-done", log.Fields{"api-endpont": c.serverEndPoint, "client": c.clientEndpoint})
			break loop
		case <-stream.Context().Done():
			logger.Debugw(ctx, "stream-context-done", log.Fields{"api-endpoint": c.serverEndPoint, "stream-info": stream.Context(), "client": c.clientEndpoint})
			break loop
		case <-keepAliveTimer.C:
			continue
		}
	}
	if stream != nil {
		if err := stream.CloseSend(); err != nil {
			logger.Warnw(ctx, "closing-stream-error", log.Fields{"error": err, "api-endpoint": c.serverEndPoint, "client": c.clientEndpoint})
		}
	}
}

// Start kicks off the adapter agent by trying to connect to the adapter
func (c *Client) Start(ctx context.Context, handler GetServiceClient, retry_interceptor ...grpc.UnaryClientInterceptor) {
	logger.Debugw(ctx, "Starting GRPC - Client", log.Fields{"api-endpoint": c.serverEndPoint})

	// If the context contains a k8s probe then register services
	p := probe.GetProbeFromContext(ctx)
	if p != nil {
		p.RegisterService(ctx, c.serverEndPoint)
	}

	var monitorConnectionCtx context.Context
	var monitorConnectionDone func()

	initialConnection := true
	c.events <- eventConnecting
	backoff := NewBackoff(c.backoffInitialInterval, c.backoffMaxInterval, c.backoffMaxElapsedTime)
	attempt := 1
loop:
	for {
		select {
		case <-ctx.Done():
			logger.Warnw(ctx, "context-closing", log.Fields{"api_endpoint": c.serverEndPoint, "client": c.clientEndpoint, "context": ctx})
			c.connectionLock.Lock()
			if !c.done {
				c.done = true
				c.events <- eventStopped
				close(c.events)
			}
			c.connectionLock.Unlock()
			// break loop
		case event := <-c.events:
			logger.Debugw(ctx, "received-event", log.Fields{"event": event, "api-endpoint": c.serverEndPoint, "client": c.clientEndpoint})
			c.connectionLock.RLock()
			// On a client stopped, just allow the stop event to go through
			if c.done && event != eventStopped {
				c.connectionLock.RUnlock()
				logger.Debugw(ctx, "ignoring-event-on-client-stop", log.Fields{"event": event, "api-endpoint": c.serverEndPoint, "client": c.clientEndpoint})
				continue
			}
			c.connectionLock.RUnlock()
			switch event {
			case eventConnecting:
				c.stateLock.Lock()
				logger.Debugw(ctx, "connection-start", log.Fields{"api-endpoint": c.serverEndPoint, "attempts": attempt, "curr-state": c.state, "client": c.clientEndpoint})
				if c.state == stateConnected {
					c.state = stateDisconnected
				}
				if c.state != stateConnecting {
					c.state = stateConnecting
					go func() {
						var err error
						if len(retry_interceptor) > 0 {
							err = c.connectToEndpoint(ctx, p, retry_interceptor...)
						} else {
							err = c.connectToEndpoint(ctx, p)
						}

						if err != nil {
							c.stateLock.Lock()
							c.state = stateDisconnected
							c.stateLock.Unlock()
							logger.Errorw(ctx, "connection-failed", log.Fields{"api-endpoint": c.serverEndPoint, "attempt": attempt, "client": c.clientEndpoint, "error": err})

							// Retry connection after a delay
							if err = backoff.Backoff(ctx); err != nil {
								// Context has closed or reached maximum elapsed time, if set
								logger.Errorw(ctx, "retry-aborted", log.Fields{"api-endpoint": c.serverEndPoint, "client": c.clientEndpoint, "error": err})
								return
							}
							attempt += 1
							c.connectionLock.RLock()
							if !c.done {
								c.events <- eventConnecting
							}
							c.connectionLock.RUnlock()
						}
					}()
				}
				c.stateLock.Unlock()

			case eventValidatingConnection:
				logger.Debugw(ctx, "connection-validation", log.Fields{"api-endpoint": c.serverEndPoint, "client": c.clientEndpoint})
				c.stateLock.Lock()
				if c.state != stateConnected {
					c.state = stateValidatingConnection
				}
				c.stateLock.Unlock()
				monitorConnectionCtx, monitorConnectionDone = context.WithCancel(context.Background())
				go c.monitorConnection(monitorConnectionCtx)

			case eventConnected:
				attempt = 1
				backoff.Reset()
				c.stateLock.Lock()
				logger.Debugw(ctx, "endpoint-connected", log.Fields{"api-endpoint": c.serverEndPoint, "curr-state": c.state, "client": c.clientEndpoint})
				if c.state != stateConnected {
					// Setup the service
					c.connectionLock.RLock()
					conn := c.connection
					c.connectionLock.RUnlock()

					subCtx, cancel := context.WithTimeout(ctx, c.backoffMaxInterval)
					svc := handler(subCtx, conn)
					if svc != nil {
						c.service = svc
						if p != nil {
							p.UpdateStatus(ctx, c.serverEndPoint, probe.ServiceStatusRunning)
						}
						logger.Infow(ctx, "connected-to-endpoint", log.Fields{"api-endpoint": c.serverEndPoint, "client": c.clientEndpoint})
					} else {
						// Should never happen, but just in case
						logger.Warnw(ctx, "service-is-nil", log.Fields{"api-endpoint": c.serverEndPoint, "client": c.clientEndpoint})
						c.events <- eventDisconnected
					}
					cancel()
					c.state = stateConnected
					if initialConnection {
						logger.Debugw(ctx, "initial-endpoint-connection", log.Fields{"api-endpoint": c.serverEndPoint, "client": c.clientEndpoint})
						initialConnection = false
					} else {
						logger.Debugw(ctx, "endpoint-reconnection", log.Fields{"api-endpoint": c.serverEndPoint, "client": c.clientEndpoint})
						// Trigger any callback on a restart
						go func() {
							err := c.onRestart(log.WithSpanFromContext(context.Background(), ctx), c.serverEndPoint)
							if err != nil {
								logger.Errorw(ctx, "unable-to-restart-endpoint", log.Fields{"error": err, "api-endpoint": c.serverEndPoint, "client": c.clientEndpoint})
							}
						}()
					}
				}
				c.stateLock.Unlock()

			case eventDisconnected:
				if p != nil {
					p.UpdateStatus(ctx, c.serverEndPoint, probe.ServiceStatusNotReady)
				}
				connectionValidationFail := false
				c.stateLock.Lock()
				logger.Debugw(ctx, "endpoint-disconnected", log.Fields{"api-endpoint": c.serverEndPoint, "curr-state": c.state, "client": c.clientEndpoint})
				if c.state == stateValidatingConnection {
					connectionValidationFail = true
					c.state = stateDisconnected
				}
				c.stateLock.Unlock()

				// Stop the streaming connection
				if monitorConnectionDone != nil {
					monitorConnectionDone()
					monitorConnectionDone = nil
				}

				if connectionValidationFail {
					// Retry connection after a delay
					if err := backoff.Backoff(ctx); err != nil {
						// Context has closed or reached maximum elapsed time, if set
						logger.Errorw(ctx, "retry-aborted", log.Fields{"api-endpoint": c.serverEndPoint, "client": c.clientEndpoint, "error": err})
						return
					}
				}
				c.connectionLock.RLock()
				if !c.done {
					c.events <- eventConnecting
				}
				c.connectionLock.RUnlock()

			case eventStopped:
				logger.Debugw(ctx, "endpoint-stopped", log.Fields{"api-endpoint": c.serverEndPoint, "client": c.clientEndpoint})

				if monitorConnectionDone != nil {
					monitorConnectionDone()
					monitorConnectionDone = nil
				}
				if err := c.closeConnection(ctx, p); err != nil {
					logger.Errorw(ctx, "endpoint-closing-connection-failed", log.Fields{"api-endpoint": c.serverEndPoint, "client": c.clientEndpoint, "error": err})
				}
				break loop
			case eventError:
				logger.Errorw(ctx, "endpoint-error-event", log.Fields{"api-endpoint": c.serverEndPoint, "client": c.clientEndpoint})
			default:
				logger.Errorw(ctx, "endpoint-unknown-event", log.Fields{"api-endpoint": c.serverEndPoint, "client": c.clientEndpoint, "error": event})
			}
		}
	}

	// Stop the streaming connection
	if monitorConnectionDone != nil {
		logger.Debugw(ctx, "closing-connection-monitoring", log.Fields{"api-endpoint": c.serverEndPoint, "client": c.clientEndpoint})
		monitorConnectionDone()
	}

	logger.Infow(ctx, "client-stopped", log.Fields{"api-endpoint": c.serverEndPoint, "client": c.clientEndpoint})
}

func (c *Client) connectToEndpoint(ctx context.Context, p *probe.Probe, retry_interceptor ...grpc.UnaryClientInterceptor) error {
	if p != nil {
		p.UpdateStatus(ctx, c.serverEndPoint, probe.ServiceStatusPreparing)
	}

	c.connectionLock.Lock()
	defer c.connectionLock.Unlock()

	if c.connection != nil {
		_ = c.connection.Close()
		c.connection = nil
	}

	c.service = nil

	// Use Interceptors to:
	// 1. automatically inject
	// 2. publish Open Tracing Spans by this GRPC Client
	// 3. detect connection failure on client calls such that the reconnection process can begin
	interceptor_opts := []grpc.UnaryClientInterceptor{
		grpc_opentracing.UnaryClientInterceptor(grpc_opentracing.WithTracer(log.ActiveTracerProxy{})),
		grpc_prometheus.UnaryClientInterceptor,
	}

	grpc_prometheus.EnableClientHandlingTimeHistogram()
	if len(retry_interceptor) > 0 {
		interceptor_opts = append(interceptor_opts, retry_interceptor...)
	}
	conn, err := grpc.Dial(c.serverEndPoint,
		grpc.WithInsecure(),
		grpc.WithDefaultCallOptions(grpc.MaxCallRecvMsgSize(grpcRecvMsgSizeLimit*1024*1024)),
		grpc.WithStreamInterceptor(grpc_middleware.ChainStreamClient(
			grpc_opentracing.StreamClientInterceptor(grpc_opentracing.WithTracer(log.ActiveTracerProxy{})),
			grpc_prometheus.StreamClientInterceptor,
		)),
		grpc.WithUnaryInterceptor(grpc_middleware.ChainUnaryClient(interceptor_opts...)),
	)

	if err == nil {
		c.connection = conn
		c.events <- eventValidatingConnection
		return nil
	} else {
		logger.Warnw(ctx, "no-connection-to-endpoint", log.Fields{"api-endpoint": c.serverEndPoint, "client": c.clientEndpoint, "error": err})
	}

	if p != nil {
		p.UpdateStatus(ctx, c.serverEndPoint, probe.ServiceStatusFailed)
	}
	return fmt.Errorf("no connection to api endpoint %s", c.serverEndPoint)
}

func (c *Client) closeConnection(ctx context.Context, p *probe.Probe) error {
	if p != nil {
		p.UpdateStatus(ctx, c.serverEndPoint, probe.ServiceStatusStopped)
	}
	logger.Infow(ctx, "client-closing-connection", log.Fields{"api-endpoint": c.serverEndPoint, "client": c.clientEndpoint})

	c.connectionLock.Lock()
	defer c.connectionLock.Unlock()

	if c.connection != nil {
		err := c.connection.Close()
		c.service = nil
		c.connection = nil
		return err
	}

	return nil
}

func (c *Client) Stop(ctx context.Context) {
	logger.Infow(ctx, "client-stop-request-event-received", log.Fields{"api-endpoint": c.serverEndPoint, "client": c.clientEndpoint})
	c.connectionLock.Lock()
	defer c.connectionLock.Unlock()
	if !c.done {
		c.done = true
		c.events <- eventStopped
		close(c.events)
	}
	logger.Infow(ctx, "client-stop-request-event-sent", log.Fields{"api-endpoint": c.serverEndPoint, "client": c.clientEndpoint})
}

// SetService is used for testing only
func (c *Client) SetService(srv interface{}) {
	c.connectionLock.Lock()
	defer c.connectionLock.Unlock()
	c.service = srv
}

func (c *Client) SubscribeForLiveness(callback func(timestamp time.Time)) {
	c.livenessLock.Lock()
	defer c.livenessLock.Unlock()
	c.livenessCallback = callback
}