[VOL-4290] Voltha go library updates for gRPC migration
Change-Id: I1aa2774beb6b7ed7419bc45aeb53fcae8a8ecda0
diff --git a/pkg/grpc/client_test.go b/pkg/grpc/client_test.go
new file mode 100644
index 0000000..d8ea31d
--- /dev/null
+++ b/pkg/grpc/client_test.go
@@ -0,0 +1,530 @@
+/*
+ * Copyright 2021-present Open Networking Foundation
+
+ * 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"
+ "os"
+ "strconv"
+ "sync"
+ "testing"
+ "time"
+
+ "github.com/golang/protobuf/ptypes/empty"
+ "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/common"
+ "github.com/opencord/voltha-protos/v5/go/core"
+ "github.com/opencord/voltha-protos/v5/go/voltha"
+ "github.com/phayes/freeport"
+ "github.com/stretchr/testify/assert"
+ "google.golang.org/grpc"
+)
+
+const (
+ testGrpcServer = "test-grpc-server"
+ initialInterval = 100 * time.Millisecond
+ maxInterval = 5000 * time.Millisecond
+ maxElapsedTime = 0 * time.Millisecond
+ monitorInterval = 2 * time.Second
+ timeout = 10 * time.Second
+)
+
+var testForNoActivityCh = make(chan time.Time, 10)
+
+type testCoreServer struct {
+ apiEndPoint string
+ server *GrpcServer
+ probe *probe.Probe
+}
+
+func newTestCoreServer(apiEndpoint string) *testCoreServer {
+ return &testCoreServer{
+ apiEndPoint: apiEndpoint,
+ probe: &probe.Probe{},
+ }
+}
+
+func (s *testCoreServer) registerService(ctx context.Context, t *testing.T) {
+ assert.NotEqual(t, "", s.apiEndPoint)
+
+ probePort, err := freeport.GetFreePort()
+ assert.Nil(t, err)
+ probeEndpoint := "127.0.0.1:" + strconv.Itoa(probePort)
+ go s.probe.ListenAndServe(ctx, probeEndpoint)
+ s.probe.RegisterService(ctx, testGrpcServer)
+
+ s.server = NewGrpcServer(s.apiEndPoint, nil, false, s.probe)
+
+ s.server.AddService(func(server *grpc.Server) {
+ core.RegisterCoreServiceServer(server, &MockCoreServiceHandler{})
+ })
+}
+
+func (s *testCoreServer) start(ctx context.Context, t *testing.T) {
+ assert.NotNil(t, s.server)
+ assert.NotEqual(t, "", s.apiEndPoint)
+
+ s.probe.UpdateStatus(ctx, testGrpcServer, probe.ServiceStatusRunning)
+ s.server.Start(ctx)
+ s.probe.UpdateStatus(ctx, testGrpcServer, probe.ServiceStatusStopped)
+}
+
+func (s *testCoreServer) stop() {
+ if s.server != nil {
+ s.server.Stop()
+ }
+}
+
+type testClient struct {
+ apiEndPoint string
+ probe *probe.Probe
+ client *Client
+}
+
+func serverRestarted(ctx context.Context, endPoint string) error {
+ logger.Infow(ctx, "remote-restarted", log.Fields{"endpoint": endPoint})
+ return nil
+}
+
+func newTestClient(apiEndpoint string, handler RestartedHandler) *testClient {
+ tc := &testClient{
+ apiEndPoint: apiEndpoint,
+ probe: &probe.Probe{},
+ }
+ // Set the environment variables that this client will use
+ var err error
+ err = os.Setenv(grpcBackoffInitialInterval, initialInterval.String())
+ if err != nil {
+ logger.Warnw(context.Background(), "setting-env-variable-failed", log.Fields{"error": err, "variable": grpcBackoffInitialInterval})
+ return nil
+ }
+ err = os.Setenv(grpcBackoffInitialInterval, maxInterval.String())
+ if err != nil {
+ logger.Warnw(context.Background(), "setting-env-variable-failed", log.Fields{"error": err, "variable": grpcBackoffInitialInterval})
+ return nil
+ }
+ err = os.Setenv(grpcBackoffMaxElapsedTime, maxElapsedTime.String())
+ if err != nil {
+ logger.Warnw(context.Background(), "setting-env-variable-failed", log.Fields{"error": err, "variable": grpcBackoffMaxElapsedTime})
+ return nil
+ }
+
+ err = os.Setenv(grpcMonitorInterval, monitorInterval.String())
+ if err != nil {
+ logger.Warnw(context.Background(), "setting-env-variable-failed", log.Fields{"error": err, "variable": grpcMonitorInterval})
+ return nil
+ }
+
+ tc.client, err = NewClient(apiEndpoint,
+ handler,
+ ActivityCheck(true))
+ if err != nil {
+ return nil
+ }
+ return tc
+}
+
+func setAndTestCoreServiceHandler(ctx context.Context, conn *grpc.ClientConn) interface{} {
+ if conn == nil {
+ return nil
+ }
+ svc := core.NewCoreServiceClient(conn)
+ if h, err := svc.GetHealthStatus(ctx, &empty.Empty{}); err != nil || h.State != voltha.HealthStatus_HEALTHY {
+ return nil
+ }
+ return svc
+}
+
+func idleConnectionTest(ctx context.Context, conn *grpc.ClientConn) interface{} {
+ if conn == nil {
+ return nil
+ }
+ svc := core.NewCoreServiceClient(conn)
+ if h, err := svc.GetHealthStatus(ctx, &empty.Empty{}); err != nil || h.State != voltha.HealthStatus_HEALTHY {
+ return nil
+ }
+ testForNoActivityCh <- time.Now()
+ return svc
+}
+
+func (c *testClient) start(ctx context.Context, t *testing.T, handler SetAndTestServiceHandler) {
+ assert.NotNil(t, c.client)
+
+ probePort, err := freeport.GetFreePort()
+ assert.Nil(t, err)
+ probeEndpoint := "127.0.0.1:" + strconv.Itoa(probePort)
+ go c.probe.ListenAndServe(ctx, probeEndpoint)
+
+ probeCtx := context.WithValue(ctx, probe.ProbeContextKey, c.probe)
+ c.client.Start(probeCtx, handler)
+}
+
+func (c *testClient) getClient(t *testing.T) core.CoreServiceClient {
+ gc, err := c.client.GetClient()
+ assert.Nil(t, err)
+ coreClient, ok := gc.(core.CoreServiceClient)
+ assert.True(t, ok)
+ return coreClient
+}
+
+func serverStartsFirstTest(t *testing.T) {
+ // Setup
+ ctx, cancel := context.WithCancel(context.Background())
+ defer cancel()
+
+ // Create and start the test server
+ grpcPort, err := freeport.GetFreePort()
+ assert.Nil(t, err)
+ apiEndpoint := "127.0.0.1:" + strconv.Itoa(grpcPort)
+ ts := newTestCoreServer(apiEndpoint)
+ ts.registerService(ctx, t)
+ go ts.start(ctx, t)
+
+ // Create the test client and start it
+ tc := newTestClient(apiEndpoint, serverRestarted)
+ assert.NotNil(t, tc)
+ go tc.start(ctx, t, setAndTestCoreServiceHandler)
+
+ // Test 1: Verify that probe status shows ready eventually
+ var servicesReady isConditionSatisfied = func() bool {
+ return ts.probe.IsReady() && tc.probe.IsReady()
+ }
+ err = waitUntilCondition(timeout, servicesReady)
+ assert.Nil(t, err)
+
+ // Test 2: Verify we get a valid client and can make grpc requests with it
+ coreClient := tc.getClient(t)
+ assert.NotNil(t, coreClient)
+
+ device, err := coreClient.GetDevice(context.Background(), &common.ID{Id: "1234"})
+ assert.Nil(t, err)
+ assert.NotNil(t, device)
+ assert.Equal(t, "test-1234", device.Type)
+}
+
+func clientStartsFirstTest(t *testing.T) {
+ ctx, cancel := context.WithCancel(context.Background())
+ defer cancel()
+
+ // Create a grpc endpoint for the server
+ grpcPort, err := freeport.GetFreePort()
+ assert.Nil(t, err)
+ apiEndpoint := "127.0.0.1:" + strconv.Itoa(grpcPort)
+
+ // Create the test client and start it
+ tc := newTestClient(apiEndpoint, serverRestarted)
+ assert.NotNil(t, tc)
+ go tc.start(ctx, t, setAndTestCoreServiceHandler)
+
+ // Verify client is not ready
+ var clientNotReady isConditionSatisfied = func() bool {
+ serviceStatus := tc.probe.GetStatus(apiEndpoint)
+ return serviceStatus == probe.ServiceStatusNotReady ||
+ serviceStatus == probe.ServiceStatusPreparing ||
+ serviceStatus == probe.ServiceStatusFailed
+ }
+ err = waitUntilCondition(timeout, clientNotReady)
+ assert.Nil(t, err)
+
+ // Create and start the test server
+ ts := newTestCoreServer(apiEndpoint)
+ ts.registerService(ctx, t)
+ go ts.start(ctx, t)
+
+ // Test 1: Verify that probe status shows ready eventually
+ var servicesReady isConditionSatisfied = func() bool {
+ return ts.probe.IsReady() && tc.probe.IsReady()
+ }
+ err = waitUntilCondition(timeout, servicesReady)
+ assert.Nil(t, err)
+
+ // Test 2: Verify we get a valid client and can make grpc requests with it
+ coreClient := tc.getClient(t)
+ assert.NotNil(t, coreClient)
+
+ device, err := coreClient.GetDevice(context.Background(), &common.ID{Id: "1234"})
+ assert.Nil(t, err)
+ assert.NotNil(t, device)
+ assert.Equal(t, "test-1234", device.Type)
+}
+
+// Liveness function
+func livessness(timestamp time.Time) {
+ logger.Debugw(context.Background(), "received-liveness", log.Fields{"timestamp": timestamp})
+}
+
+func serverRestarts(t *testing.T, numRestartRuns int) {
+ // Setup
+ ctx, cancel := context.WithCancel(context.Background())
+ defer cancel()
+
+ // Create and start the test server
+ grpcPort, err := freeport.GetFreePort()
+ assert.Nil(t, err)
+ apiEndpoint := "127.0.0.1:" + strconv.Itoa(grpcPort)
+ ts := newTestCoreServer(apiEndpoint)
+ ts.registerService(ctx, t)
+ go ts.start(ctx, t)
+
+ // Create the test client and start it
+ tc := newTestClient(apiEndpoint, serverRestarted)
+ assert.NotNil(t, tc)
+
+ // Subscribe for liveness
+ tc.client.SubscribeForLiveness(livessness)
+ go tc.start(ctx, t, setAndTestCoreServiceHandler)
+
+ // Test 1: Verify that probe status shows ready eventually
+ var servicesReady isConditionSatisfied = func() bool {
+ return ts.probe.IsReady() && tc.probe.IsReady()
+ }
+ err = waitUntilCondition(timeout, servicesReady)
+ assert.Nil(t, err)
+
+ // Test 2: Verify we get a valid client and can make grpc requests with it
+ coreClient := tc.getClient(t)
+ assert.NotNil(t, coreClient)
+
+ device, err := coreClient.GetDevice(context.Background(), &common.ID{Id: "1234"})
+ assert.Nil(t, err)
+ assert.NotNil(t, device)
+ assert.Equal(t, "test-1234", device.Type)
+
+ for i := 1; i <= numRestartRuns; i++ {
+ //Test 3: Stop server and verify server status
+ ts.stop()
+ var serverDown isConditionSatisfied = func() bool {
+ return ts.probe.GetStatus(testGrpcServer) == probe.ServiceStatusStopped
+ }
+ err = waitUntilCondition(timeout, serverDown)
+ assert.Nil(t, err)
+
+ // Make a grpc request - this will detect the server being down and automatically trigger the grpc client
+ // to reconnect
+ _, err = coreClient.GetDevice(context.Background(), &common.ID{Id: "1234"})
+ assert.NotNil(t, err)
+
+ // Wait until the client service shows as not ready. A wait is not needed. It's just to verify that the
+ // client changes connection state.
+ var clientNotReady isConditionSatisfied = func() bool {
+ serviceStatus := tc.probe.GetStatus(apiEndpoint)
+ return serviceStatus == probe.ServiceStatusNotReady ||
+ serviceStatus == probe.ServiceStatusPreparing ||
+ serviceStatus == probe.ServiceStatusFailed
+ }
+ err = waitUntilCondition(timeout, clientNotReady)
+
+ assert.Nil(t, err)
+
+ // Keep the server down for 1/2 second
+ time.Sleep(500 * time.Millisecond)
+
+ // Test 4: Restart the server and verify the server is back online
+ go ts.start(ctx, t)
+ err = waitUntilCondition(timeout, servicesReady)
+ assert.Nil(t, err)
+
+ // Test 5: verify we can pull new device with a new client instance
+ coreClient = tc.getClient(t)
+ assert.NotNil(t, coreClient)
+ device, err := coreClient.GetDevice(context.Background(), &common.ID{Id: "1234"})
+ assert.Nil(t, err)
+ assert.Equal(t, "test-1234", device.Type)
+ }
+ // Stop the server
+ ts.stop()
+}
+
+func testNoActivity(t *testing.T) {
+ ctx, cancel := context.WithCancel(context.Background())
+ defer cancel()
+
+ // Create a grpc endpoint for the server
+ grpcPort, err := freeport.GetFreePort()
+ assert.Nil(t, err)
+ apiEndpoint := "127.0.0.1:" + strconv.Itoa(grpcPort)
+
+ // Create the test client and start it
+ tc := newTestClient(apiEndpoint, serverRestarted)
+ assert.NotNil(t, tc)
+ go tc.start(ctx, t, idleConnectionTest)
+
+ // Create and start the test server
+ ts := newTestCoreServer(apiEndpoint)
+ ts.registerService(ctx, t)
+ go ts.start(ctx, t)
+
+ // Test 1: Verify that probe status shows ready eventually
+ var servicesReady isConditionSatisfied = func() bool {
+ return ts.probe.IsReady() && tc.probe.IsReady()
+ }
+ err = waitUntilCondition(timeout, servicesReady)
+ assert.Nil(t, err)
+
+ // Test 2: Verify we get a valid client and can make grpc requests with it
+ coreClient := tc.getClient(t)
+ assert.NotNil(t, coreClient)
+
+ device, err := coreClient.GetDevice(context.Background(), &common.ID{Id: "1234"})
+ assert.Nil(t, err)
+ assert.NotNil(t, device)
+ assert.Equal(t, "test-1234", device.Type)
+
+ start := time.Now()
+ numChecks := 3 // Test for 3 checks
+ // Wait on the the idle channel - on no activity a connection probe will be attempted by the client
+ timer := time.NewTimer((monitorInterval + 1*time.Second) * time.Duration(numChecks))
+ defer timer.Stop()
+ count := 0
+loop:
+ for {
+ select {
+ case timestamp := <-testForNoActivityCh:
+ if timestamp.After(start) {
+ count += 1
+ if count > numChecks {
+ break loop
+ }
+ }
+ case <-timer.C:
+ t.Fatal("no activity on the idle channel")
+ }
+ }
+}
+
+func testServerLimit(t *testing.T) {
+ t.Skip() // Not needed for regular unit tests
+
+ ctx, cancel := context.WithCancel(context.Background())
+ defer cancel()
+
+ // Create a grpc endpoint for the server
+ grpcPort, err := freeport.GetFreePort()
+ assert.Nil(t, err)
+ apiEndpoint := "127.0.0.1:" + strconv.Itoa(grpcPort)
+
+ // Create the test client and start it
+ tc := newTestClient(apiEndpoint, serverRestarted)
+ assert.NotNil(t, tc)
+ go tc.start(ctx, t, idleConnectionTest)
+
+ // Create and start the test server
+ ts := newTestCoreServer(apiEndpoint)
+ ts.registerService(ctx, t)
+ go ts.start(ctx, t)
+
+ // Test 1: Verify that probe status shows ready eventually
+ var servicesReady isConditionSatisfied = func() bool {
+ return ts.probe.IsReady() && tc.probe.IsReady()
+ }
+ err = waitUntilCondition(timeout, servicesReady)
+ assert.Nil(t, err)
+
+ // Test 2: Verify we get a valid client and can make grpc requests with it
+ coreClient := tc.getClient(t)
+ assert.NotNil(t, coreClient)
+
+ var lock sync.RWMutex
+ bad := []time.Duration{}
+ bad_err := []string{}
+ good := []time.Duration{}
+ var wg sync.WaitGroup
+ numRPCs := 10
+ total_good := time.Duration(0)
+ max_good := time.Duration(0)
+ total_bad := time.Duration(0)
+ max_bad := time.Duration(0)
+ order_received := []uint32{}
+ for i := 1; i <= numRPCs; i++ {
+ wg.Add(1)
+ go func(seq int) {
+ local := time.Now()
+ ctx, cancel := context.WithTimeout(context.Background(), 10000*time.Millisecond)
+ defer cancel()
+ var err error
+ var d *voltha.Device
+ d, err = coreClient.GetDevice(ctx, &common.ID{Id: strconv.Itoa(seq)})
+ if err != nil {
+ lock.Lock()
+ bad = append(bad, time.Since(local))
+ bad_err = append(bad_err, err.Error())
+ total_bad += time.Since(local)
+ if time.Since(local) > max_bad {
+ max_bad = time.Since(local)
+ }
+ logger.Errorw(ctx, "error produced", log.Fields{"error": err})
+ lock.Unlock()
+ } else {
+ lock.Lock()
+ good = append(good, time.Since(local))
+ total_good += time.Since(local)
+ if time.Since(local) > max_good {
+ max_good = time.Since(local)
+ }
+ if d != nil {
+ order_received = append(order_received, d.Vlan)
+ }
+ lock.Unlock()
+ }
+ wg.Done()
+ }(i)
+ }
+ wg.Wait()
+ assert.Equal(t, 0, len(bad))
+ assert.Equal(t, numRPCs, len(good))
+ //fmt.Println("Bad:", bad[:10])
+ if len(bad_err) > 0 {
+ fmt.Println("Bad Err Last:", bad_err[len(bad_err)-1:])
+ fmt.Println("Bad Err First:", bad_err[:1])
+ }
+ fmt.Println("Good:", good[len(good)-10:])
+ fmt.Println("Good average time:", total_good.Milliseconds()/int64(numRPCs))
+ fmt.Println("Bad average time:", total_bad.Milliseconds()/int64(numRPCs))
+ fmt.Println("Bad Max:", max_bad)
+ fmt.Println("Good Max:", max_good)
+ //fmt.Println("Order received:", order_received)
+
+ prev := order_received[0]
+
+ for i := 1; i < len(order_received); i++ {
+ if order_received[i] < prev {
+ fmt.Println("Prev:", prev, " curr:", order_received[i])
+ }
+ prev = order_received[i]
+ }
+}
+
+func TestSuiteClient3(t *testing.T) {
+ // Setup
+ log.SetAllLogLevel(volthaTestLogLevel)
+
+ // Test starting server before client
+ serverStartsFirstTest(t)
+
+ // Test starting client before server
+ clientStartsFirstTest(t)
+
+ // Test server restarts
+ serverRestarts(t, 1)
+
+ //Test that the client test the grpc connection on no activity
+ testNoActivity(t)
+
+ // Test client queueing with server limit
+ testServerLimit(t)
+}