SOTW ADS client (#604)

Integration initial ADS client for SOTW protocol.

Signed-off-by: Renuka Fernando <renukapiyumal@gmail.com>

pkg/client/sotw/v3/client.go

@@ -0,0 +1,167 @@
+// Copyright 2022 Envoyproxy Authors
+//
+//   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 sotw provides an implementation of GRPC SoTW (State of The World) part of XDS client
+package sotw
+
+import (
+	"context"
+	"errors"
+	"io"
+	"sync"
+
+	"github.com/golang/protobuf/ptypes/any"
+	status "google.golang.org/genproto/googleapis/rpc/status"
+	"google.golang.org/grpc"
+	"google.golang.org/grpc/codes"
+	grpcStatus "google.golang.org/grpc/status"
+
+	core "github.com/envoyproxy/go-control-plane/envoy/config/core/v3"
+	discovery "github.com/envoyproxy/go-control-plane/envoy/service/discovery/v3"
+)
+
+var (
+	ErrInit    = errors.New("ads client: grpc connection is not initialized (use InitConnect() method to initialize connection)")
+	ErrNilResp = errors.New("ads client: nil response from xds management server")
+)
+
+// ADSClient is a SoTW and ADS based generic gRPC xDS client which can be used to
+// implement an xDS client which fetches resources from an xDS server and responds
+// the server back with ack or nack the resources.
+type ADSClient interface {
+	// Initialize the gRPC connection with management server and send the initial Discovery Request.
+	InitConnect(clientConn grpc.ClientConnInterface, opts ...grpc.CallOption) error
+	// Fetch waits for a response from management server and returns response or error.
+	Fetch() (*Response, error)
+	// Ack acknowledge the validity of the last received response to management server.
+	Ack() error
+	// Nack acknowledge the invalidity of the last received response to management server.
+	Nack(message string) error
+}
+
+// Response wraps the latest Resources from the xDS server.
+// For the time being it only contains the Resources from server. This can be extended with
+// other response details. For example some metadata from DiscoveryResponse.
+type Response struct {
+	Resources []*any.Any
+}
+
+type adsClient struct {
+	ctx     context.Context
+	mu      sync.Mutex
+	node    *core.Node
+	typeURL string
+
+	// streamClient is the ADS discovery client
+	streamClient discovery.AggregatedDiscoveryService_StreamAggregatedResourcesClient
+	// lastAckedResponse is the last response acked by the ADS client
+	lastAckedResponse *discovery.DiscoveryResponse
+	// lastReceivedResponse is the last response received from management server
+	lastReceivedResponse *discovery.DiscoveryResponse
+}
+
+// NewADSClient returns a new ADSClient
+func NewADSClient(ctx context.Context, node *core.Node, typeURL string) ADSClient {
+	return &adsClient{
+		ctx:     ctx,
+		node:    node,
+		typeURL: typeURL,
+	}
+}
+
+// Initialize the gRPC connection with management server and send the initial Discovery Request.
+func (c *adsClient) InitConnect(clientConn grpc.ClientConnInterface, opts ...grpc.CallOption) error {
+	streamClient, err := discovery.NewAggregatedDiscoveryServiceClient(clientConn).StreamAggregatedResources(c.ctx, opts...)
+	if err != nil {
+		return err
+	}
+	c.streamClient = streamClient
+	return c.Ack()
+}
+
+// Fetch waits for a response from management server and returns response or error.
+func (c *adsClient) Fetch() (*Response, error) {
+	c.mu.Lock()
+	defer c.mu.Unlock()
+
+	if c.streamClient == nil {
+		return nil, ErrInit
+	}
+	resp, err := c.streamClient.Recv()
+	if err != nil {
+		return nil, err
+	}
+
+	if resp == nil {
+		return nil, ErrNilResp
+	}
+	c.lastReceivedResponse = resp
+	return &Response{
+		Resources: resp.GetResources(),
+	}, err
+}
+
+// Ack acknowledge the validity of the last received response to management server.
+func (c *adsClient) Ack() error {
+	c.mu.Lock()
+	defer c.mu.Unlock()
+
+	c.lastAckedResponse = c.lastReceivedResponse
+	return c.send(nil)
+}
+
+// Nack acknowledge the invalidity of the last received response to management server.
+func (c *adsClient) Nack(message string) error {
+	c.mu.Lock()
+	defer c.mu.Unlock()
+
+	errorDetail := &status.Status{
+		Message: message,
+	}
+	return c.send(errorDetail)
+}
+
+// IsConnError checks the provided error is due to the gRPC connection
+// and returns true if it is due to the gRPC connection.
+//
+// In this case the gRPC connection with the server should be re initialized with the
+// ADSClient.InitConnect method.
+func IsConnError(err error) bool {
+	if err == nil {
+		return false
+	}
+	if errors.Is(err, io.EOF) {
+		return true
+	}
+	errStatus, ok := grpcStatus.FromError(err)
+	if !ok {
+		return false
+	}
+	return errStatus.Code() == codes.Unavailable || errStatus.Code() == codes.Canceled
+}
+
+func (c *adsClient) send(errorDetail *status.Status) error {
+	if c.streamClient == nil {
+		return ErrInit
+	}
+
+	req := &discovery.DiscoveryRequest{
+		Node:          c.node,
+		VersionInfo:   c.lastAckedResponse.GetVersionInfo(),
+		TypeUrl:       c.typeURL,
+		ResponseNonce: c.lastReceivedResponse.GetNonce(),
+		ErrorDetail:   errorDetail,
+	}
+	return c.streamClient.Send(req)
+}

pkg/client/sotw/v3/client_test.go

@@ -0,0 +1,141 @@
+package sotw_test
+
+import (
+	"context"
+	"net"
+	"sync"
+	"testing"
+	"time"
+
+	"google.golang.org/grpc"
+	"google.golang.org/grpc/credentials/insecure"
+	"google.golang.org/protobuf/proto"
+	"google.golang.org/protobuf/types/known/anypb"
+
+	clusterv3 "github.com/envoyproxy/go-control-plane/envoy/config/cluster/v3"
+	core "github.com/envoyproxy/go-control-plane/envoy/config/core/v3"
+	discovery "github.com/envoyproxy/go-control-plane/envoy/service/discovery/v3"
+	"github.com/envoyproxy/go-control-plane/pkg/cache/types"
+	"github.com/envoyproxy/go-control-plane/pkg/cache/v3"
+	client "github.com/envoyproxy/go-control-plane/pkg/client/sotw/v3"
+	"github.com/envoyproxy/go-control-plane/pkg/resource/v3"
+	"github.com/envoyproxy/go-control-plane/pkg/server/v3"
+
+	"github.com/stretchr/testify/assert"
+)
+
+func TestFetch(t *testing.T) {
+	ctx, cancel := context.WithTimeout(context.Background(), 1*time.Second)
+	defer cancel()
+
+	snapCache := cache.NewSnapshotCache(true, cache.IDHash{}, nil)
+	go func() {
+		err := startAdsServer(ctx, snapCache)
+		assert.NoError(t, err)
+	}()
+
+	conn, err := grpc.Dial(":18001", grpc.WithTransportCredentials(insecure.NewCredentials()), grpc.WithBlock())
+	assert.NoError(t, err)
+	defer conn.Close()
+
+	c := client.NewADSClient(ctx, &core.Node{Id: "node_1"}, resource.ClusterType)
+	err = c.InitConnect(conn)
+	assert.NoError(t, err)
+
+	t.Run("Test initial fetch", testInitialFetch(ctx, snapCache, c))
+	t.Run("Test next fetch", testNextFetch(ctx, snapCache, c))
+}
+
+func testInitialFetch(ctx context.Context, snapCache cache.SnapshotCache, c client.ADSClient) func(t *testing.T) {
+	return func(t *testing.T) {
+		wg := sync.WaitGroup{}
+		wg.Add(1)
+
+		go func() {
+			// watch for configs
+			resp, err := c.Fetch()
+			assert.NoError(t, err)
+			assert.Equal(t, 3, len(resp.Resources))
+			for _, r := range resp.Resources {
+				cluster := &clusterv3.Cluster{}
+				err := anypb.UnmarshalTo(r, cluster, proto.UnmarshalOptions{})
+				assert.NoError(t, err)
+				assert.Contains(t, []string{"cluster_1", "cluster_2", "cluster_3"}, cluster.Name)
+			}
+
+			err = c.Ack()
+			assert.NoError(t, err)
+			wg.Done()
+		}()
+
+		snapshot, err := cache.NewSnapshot("1", map[resource.Type][]types.Resource{
+			resource.ClusterType: {
+				&clusterv3.Cluster{Name: "cluster_1"},
+				&clusterv3.Cluster{Name: "cluster_2"},
+				&clusterv3.Cluster{Name: "cluster_3"},
+			},
+		})
+		assert.NoError(t, err)
+
+		err = snapshot.Consistent()
+		assert.NoError(t, err)
+		err = snapCache.SetSnapshot(ctx, "node_1", snapshot)
+		wg.Wait()
+		assert.NoError(t, err)
+	}
+}
+
+func testNextFetch(ctx context.Context, snapCache cache.SnapshotCache, c client.ADSClient) func(t *testing.T) {
+	return func(t *testing.T) {
+		wg := sync.WaitGroup{}
+		wg.Add(1)
+
+		go func() {
+			// watch for configs
+			resp, err := c.Fetch()
+			assert.NoError(t, err)
+			assert.Equal(t, 2, len(resp.Resources))
+			for _, r := range resp.Resources {
+				cluster := &clusterv3.Cluster{}
+				err = anypb.UnmarshalTo(r, cluster, proto.UnmarshalOptions{})
+				assert.NoError(t, err)
+				assert.Contains(t, []string{"cluster_2", "cluster_4"}, cluster.Name)
+			}
+
+			err = c.Ack()
+			assert.NoError(t, err)
+			wg.Done()
+		}()
+
+		snapshot, err := cache.NewSnapshot("2", map[resource.Type][]types.Resource{
+			resource.ClusterType: {
+				&clusterv3.Cluster{Name: "cluster_2"},
+				&clusterv3.Cluster{Name: "cluster_4"},
+			},
+		})
+		assert.NoError(t, err)
+
+		err = snapshot.Consistent()
+		assert.NoError(t, err)
+		err = snapCache.SetSnapshot(ctx, "node_1", snapshot)
+		assert.NoError(t, err)
+		wg.Wait()
+	}
+}
+
+func startAdsServer(ctx context.Context, snapCache cache.SnapshotCache) error {
+	lis, err := net.Listen("tcp", "127.0.0.1:18001")
+	if err != nil {
+		return err
+	}
+
+	grpcServer := grpc.NewServer()
+	s := server.NewServer(ctx, snapCache, nil)
+	discovery.RegisterAggregatedDiscoveryServiceServer(grpcServer, s)
+
+	if e := grpcServer.Serve(lis); e != nil {
+		err = e
+	}
+
+	return err
+}