hubble/relay: improve peer connections handling

pkg/hubble/peer/service_test.go

@@ -355,7 +355,7 @@ func TestService_Notify(t *testing.T) {
 		t.Run(tt.name, func(t *testing.T) {
 			var got []*peerpb.ChangeNotification
 			var wg sync.WaitGroup
-			fakeServer := &FakePeerNotifyServer{
+			fakeServer := &testutils.FakePeerNotifyServer{
 				OnSend: func(resp *peerpb.ChangeNotification) error {
 					got = append(got, resp)
 					wg.Done()
@@ -418,7 +418,7 @@ func TestService_Notify(t *testing.T) {
 }
 
 func TestService_NotifyWithBlockedSend(t *testing.T) {
-	fakeServer := &FakePeerNotifyServer{
+	fakeServer := &testutils.FakePeerNotifyServer{
 		OnSend: func(resp *peerpb.ChangeNotification) error {
 			<-time.After(100 * time.Millisecond)
 			return nil
@@ -470,18 +470,6 @@ func TestService_NotifyWithBlockedSend(t *testing.T) {
 	<-done
 }
 
-type FakePeerNotifyServer struct {
-	OnSend func(response *peerpb.ChangeNotification) error
-	*testutils.FakeGRPCServerStream
-}
-
-func (s *FakePeerNotifyServer) Send(response *peerpb.ChangeNotification) error {
-	if s.OnSend != nil {
-		return s.OnSend(response)
-	}
-	panic("OnSend not set")
-}
-
 type notifier struct {
 	nodes       []types.Node
 	subscribers map[datapath.NodeHandler]struct{}

pkg/hubble/peer/types/client.go

@@ -0,0 +1,67 @@
+// Copyright 2020 Authors of Cilium
+//
+// 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 types
+
+import (
+	"context"
+	"io"
+	"time"
+
+	peerpb "github.com/cilium/cilium/api/v1/peer"
+
+	"google.golang.org/grpc"
+)
+
+// Client defines an interface that Peer service client should implement.
+type Client interface {
+	peerpb.PeerClient
+	io.Closer
+}
+
+// ClientBuilder creates a new Client.
+type ClientBuilder interface {
+	// Client builds a new Client that connects to the given target.
+	Client(target string) (Client, error)
+}
+
+type client struct {
+	conn *grpc.ClientConn
+	peerpb.PeerClient
+}
+
+func (c *client) Close() error {
+	if c.conn == nil {
+		return nil
+	}
+	return c.conn.Close()
+}
+
+// LocalClientBuilder is a ClientBuilder that is suitable when the gRPC
+// connection to the Peer service is local (typically a Unix Domain Socket).
+type LocalClientBuilder struct {
+	DialTimeout time.Duration
+}
+
+// Client implements ClientBuilder.Client.
+func (b LocalClientBuilder) Client(target string) (Client, error) {
+	ctx, cancel := context.WithTimeout(context.Background(), b.DialTimeout)
+	defer cancel()
+	// the connection is local so we assume WithInsecure() is safe in this context
+	conn, err := grpc.DialContext(ctx, target, grpc.WithInsecure(), grpc.WithBlock())
+	if err != nil {
+		return nil, err
+	}
+	return &client{conn, peerpb.NewPeerClient(conn)}, nil
+}

pkg/hubble/relay/observer.go

@@ -21,6 +21,7 @@ import (
 
 	observerpb "github.com/cilium/cilium/api/v1/observer"
 	relaypb "github.com/cilium/cilium/api/v1/relay"
+	"github.com/cilium/cilium/pkg/hubble/relay/pool"
 	"github.com/cilium/cilium/pkg/hubble/relay/queue"
 	nodeTypes "github.com/cilium/cilium/pkg/node/types"
 
@@ -32,16 +33,27 @@ import (
 	"google.golang.org/grpc/status"
 )
 
+// TODO: remove this shim once "google.golang.org/grpc" is bumped to v1.28+
+// which should convert generated code
+// `observerpb.NewObserverClient(cc *grpc.ClientConn)` to
+// `observerpb.NewObserverClient(cc grpc.ClientConnInterface)`.
+func newObserverClient(cc pool.ClientConn) observerpb.ObserverClient {
+	if conn, ok := cc.(*grpc.ClientConn); ok {
+		return observerpb.NewObserverClient(conn)
+	}
+	return nil
+}
+
 // ensure that Server implements the observer.ObserverServer interface.
 var _ observerpb.ObserverServer = (*Server)(nil)
 
 func retrieveFlowsFromPeer(
 	ctx context.Context,
-	conn *grpc.ClientConn,
+	conn pool.ClientConn,
 	req *observerpb.GetFlowsRequest,
 	flows chan<- *observerpb.GetFlowsResponse,
 ) error {
-	client := observerpb.NewObserverClient(conn)
+	client := newObserverClient(conn)
 	c, err := client.GetFlows(ctx, req)
 	if err != nil {
 		return err
@@ -236,7 +248,7 @@ func (s *Server) GetFlows(req *observerpb.GetFlowsRequest, stream observerpb.Obs
 		}
 	}()
 
-	peers := s.peerList()
+	peers := s.pm.List()
 	qlen := s.opts.SortBufferMaxLen // we don't want to buffer too many flows
 	if nqlen := req.GetNumber() * uint64(len(peers)); nqlen > 0 && nqlen < uint64(qlen) {
 		// don't make the queue bigger than necessary as it would be a problem
@@ -250,20 +262,20 @@ func (s *Server) GetFlows(req *observerpb.GetFlowsRequest, stream observerpb.Obs
 
 	for _, p := range peers {
 		p := p
-		if p.conn == nil || p.connErr != nil {
+		if p.Conn == nil {
 			s.log.WithField("address", p.Address.String()).Infof(
 				"No connection to peer %s, skipping", p.Name,
 			)
+			s.pm.ReportOffline(p.Name)
 			unavailableNodes = append(unavailableNodes, p.Name)
-			go s.connectPeer(p.Name, p.Address.String())
 			continue
 		}
 		connectedNodes = append(connectedNodes, p.Name)
 		g.Go(func() error {
 			// retrieveFlowsFromPeer returns blocks until the peer finishes
 			// the request by closing the connection, an error occurs,
 			// or gctx expires.
-			err := retrieveFlowsFromPeer(gctx, p.conn, req, flows)
+			err := retrieveFlowsFromPeer(gctx, p.Conn, req, flows)
 			if err != nil {
 				s.log.WithFields(logrus.Fields{
 					"error": err,
@@ -330,19 +342,19 @@ func (s *Server) ServerStatus(ctx context.Context, req *observerpb.ServerStatusR
 		}
 	}()
 
-	peers := s.peerList()
+	peers := s.pm.List()
 	statuses := make(chan *observerpb.ServerStatusResponse, len(peers))
 	for _, p := range peers {
 		p := p
-		if p.conn == nil || p.connErr != nil {
+		if p.Conn == nil {
 			s.log.WithField("address", p.Address.String()).Infof(
 				"No connection to peer %s, skipping", p.Name,
 			)
-			go s.connectPeer(p.Name, p.Address.String())
+			s.pm.ReportOffline(p.Name)
 			continue
 		}
 		g.Go(func() error {
-			client := observerpb.NewObserverClient(p.conn)
+			client := newObserverClient(p.Conn)
 			status, err := client.ServerStatus(ctx, req)
 			if err != nil {
 				s.log.WithFields(logrus.Fields{