hubble: move some types definition to avoid import cycles

As part of implementing unit tests for `pkg/hubble/relay/observer`,
`FakeClientConn` is needed. Prior to this commit, the implementation was
only present for testing in `pkg/hubble/relay/pool`. However, moving it
to `pkg/hubble/testutils` creates import cycles within unit test files.
In order to resolve them, some types definitions are moved to the
existing `pkg/hubble/peer/types` package and a new
`pkg/hubble/relay/pool/types` package.

There are no functional changes with this commit; only the necessary
changes to avoid build/test failures resulting from the aforementioned
reorganization of packages and types definitions.

Signed-off-by: Robin Hahling <[email protected]>

pkg/hubble/peer/peer.go → pkg/hubble/peer/types/peer.go

@@ -12,7 +12,7 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.
 
-package peer
+package types
 
 import (
 	"net"

pkg/hubble/peer/peer_test.go → pkg/hubble/peer/types/peer_test.go

@@ -14,7 +14,7 @@
 
 // +build !privileged_tests
 
-package peer
+package types
 
 import (
 	"net"

pkg/hubble/relay/observer/observer.go

@@ -21,7 +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"
+	poolTypes "github.com/cilium/cilium/pkg/hubble/relay/pool/types"
 	"github.com/cilium/cilium/pkg/hubble/relay/queue"
 	nodeTypes "github.com/cilium/cilium/pkg/node/types"
 
@@ -36,14 +36,14 @@ import (
 // which should convert generated code
 // `observerpb.NewObserverClient(cc *grpc.ClientConn)` to
 // `observerpb.NewObserverClient(cc grpc.ClientConnInterface)`.
-func newObserverClient(cc pool.ClientConn) observerpb.ObserverClient {
+func newObserverClient(cc poolTypes.ClientConn) observerpb.ObserverClient {
 	if conn, ok := cc.(*grpc.ClientConn); ok {
 		return observerpb.NewObserverClient(conn)
 	}
 	return nil
 }
 
-func isAvailable(conn pool.ClientConn) bool {
+func isAvailable(conn poolTypes.ClientConn) bool {
 	if conn == nil {
 		return false
 	}
@@ -56,7 +56,7 @@ func isAvailable(conn pool.ClientConn) bool {
 
 func retrieveFlowsFromPeer(
 	ctx context.Context,
-	conn pool.ClientConn,
+	conn poolTypes.ClientConn,
 	req *observerpb.GetFlowsRequest,
 	flows chan<- *observerpb.GetFlowsResponse,
 ) error {

pkg/hubble/relay/observer/server.go

@@ -20,7 +20,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"
+	poolTypes "github.com/cilium/cilium/pkg/hubble/relay/pool/types"
 
 	"github.com/golang/protobuf/ptypes/wrappers"
 	"github.com/sirupsen/logrus"
@@ -38,7 +38,7 @@ const numUnavailableNodesReportMax = 10
 type PeerLister interface {
 	// List returns a list of peers with active connections. If a peer cannot
 	// be connected to; its Conn attribute must be nil.
-	List() []pool.Peer
+	List() []poolTypes.Peer
 }
 
 // PeerReporter is the interface that wraps the ReportOffline method.

pkg/hubble/relay/pool/client.go

@@ -16,39 +16,13 @@ package pool
 
 import (
 	"context"
-	"io"
 	"time"
 
+	poolTypes "github.com/cilium/cilium/pkg/hubble/relay/pool/types"
+
 	"google.golang.org/grpc"
-	"google.golang.org/grpc/connectivity"
 )
 
-// ClientConn is an interface that defines the functions clients need to
-// perform unary and streaming RPCs. It is implemented by *grpc.ClientConn.
-type ClientConn interface {
-	// GetState returns the connectivity.State of ClientConn.
-	GetState() connectivity.State
-	io.Closer
-
-	// TODO: compose with grpc.ClientConnInterface once
-	// "google.golang.org/grpc" is bumped to v1.27+ and remove the following
-	// two methods (which are part of grpc.ClientConnInterface).
-
-	// Invoke performs a unary RPC and returns after the response is received
-	// into reply.
-	Invoke(ctx context.Context, method string, args interface{}, reply interface{}, opts ...grpc.CallOption) error
-	// NewStream begins a streaming RPC.
-	NewStream(ctx context.Context, desc *grpc.StreamDesc, method string, opts ...grpc.CallOption) (grpc.ClientStream, error)
-}
-
-var _ ClientConn = (*grpc.ClientConn)(nil)
-
-// ClientConnBuilder wraps the ClientConn method.
-type ClientConnBuilder interface {
-	// ClientConn creates a new ClientConn using target.
-	ClientConn(target string) (ClientConn, error)
-}
-
 // GRPCClientConnBuilder is a generic ClientConnBuilder implementation.
 type GRPCClientConnBuilder struct {
 	// DialTimeout specifies the timeout used when establishing a new
@@ -60,7 +34,7 @@ type GRPCClientConnBuilder struct {
 }
 
 // ClientConn implements ClientConnBuilder.ClientConn.
-func (b GRPCClientConnBuilder) ClientConn(target string) (ClientConn, error) {
+func (b GRPCClientConnBuilder) ClientConn(target string) (poolTypes.ClientConn, error) {
 	ctx, cancel := context.WithTimeout(context.Background(), b.DialTimeout)
 	defer cancel()
 	return grpc.DialContext(ctx, target, b.Options...)

pkg/hubble/relay/pool/manager.go

@@ -21,24 +21,18 @@ import (
 	"time"
 
 	peerpb "github.com/cilium/cilium/api/v1/peer"
-	hubblePeer "github.com/cilium/cilium/pkg/hubble/peer"
+	peerTypes "github.com/cilium/cilium/pkg/hubble/peer/types"
+	poolTypes "github.com/cilium/cilium/pkg/hubble/relay/pool/types"
 	"github.com/cilium/cilium/pkg/lock"
 
 	"github.com/sirupsen/logrus"
 	"google.golang.org/grpc/connectivity"
 )
 
-// Peer is like hubblePeer.Peer but includes a Conn attribute to reach the
-// peer's gRPC API endpoint.
-type Peer struct {
-	hubblePeer.Peer
-	Conn ClientConn
-}
-
 type peer struct {
 	mu lock.Mutex
-	hubblePeer.Peer
-	conn            ClientConn
+	peerTypes.Peer
+	conn            poolTypes.ClientConn
 	connAttempts    int
 	nextConnAttempt time.Time
 }
@@ -141,7 +135,7 @@ connect:
 				}
 			}
 			m.opts.log.WithField("change notification", cn).Debug("Received peer change notification")
-			p := hubblePeer.FromChangeNotification(cn)
+			p := peerTypes.FromChangeNotification(cn)
 			switch cn.GetType() {
 			case peerpb.ChangeNotificationType_PEER_ADDED:
 				m.add(p)
@@ -205,17 +199,17 @@ func (m *PeerManager) Stop() {
 }
 
 // List implements observer.PeerLister.List.
-func (m *PeerManager) List() []Peer {
+func (m *PeerManager) List() []poolTypes.Peer {
 	m.mu.RLock()
 	defer m.mu.RUnlock()
 	if len(m.peers) == 0 {
 		return nil
 	}
-	peers := make([]Peer, 0, len(m.peers))
+	peers := make([]poolTypes.Peer, 0, len(m.peers))
 	for _, v := range m.peers {
 		// note: there shouldn't be null entries in the map
-		peers = append(peers, Peer{
-			Peer: hubblePeer.Peer{
+		peers = append(peers, poolTypes.Peer{
+			Peer: peerTypes.Peer{
 				Name:    v.Name,
 				Address: v.Address,
 			},
@@ -253,7 +247,7 @@ func (m *PeerManager) ReportOffline(name string) {
 	}()
 }
 
-func (m *PeerManager) add(hp *hubblePeer.Peer) {
+func (m *PeerManager) add(hp *peerTypes.Peer) {
 	if hp == nil {
 		return
 	}
@@ -267,7 +261,7 @@ func (m *PeerManager) add(hp *hubblePeer.Peer) {
 	}
 }
 
-func (m *PeerManager) remove(hp *hubblePeer.Peer) {
+func (m *PeerManager) remove(hp *peerTypes.Peer) {
 	if hp == nil {
 		return
 	}
@@ -279,7 +273,7 @@ func (m *PeerManager) remove(hp *hubblePeer.Peer) {
 	m.mu.Unlock()
 }
 
-func (m *PeerManager) update(hp *hubblePeer.Peer) {
+func (m *PeerManager) update(hp *peerTypes.Peer) {
 	if hp == nil {
 		return
 	}