chore: Make pool tests pass

Fixes the failing pool tests in `plumbing` and `rlp/internal/ingress`
following the change to use `grpc.NewClient` instead of `grpc.Dial`.
This is because they were testing that connections were initiated when a
new doppler was added. However, with the change to NewClient,
connections won't be made until the resulting grpc client connections
are used for RPC. This also means that dopplers may be added to the pool
map when in the past they would not be added due to the connection on
Dial failing.

One approach to getting the existing tests to pass would have been to
call `Connect` on the grpc client connections to force them to leave
idle mode. However, `Connect` is experimental, and really gRPC is
encouraging us not to care about the connection state when creating
client connections.

To fix the tests we ended up just asserting on the pool size. One
downside of this was that we couldn't see a nice way to assert that
`Close` was called on the gRPC client connection when `Close` was called
for a doppler address. We could have replaced the connection creation
with an interface and mocked that but it didn't seem worth it.

Signed-off-by: Carson Long <[email protected]>
Signed-off-by: Rebecca Roberts <[email protected]>

Author: acrmp

src/plumbing/pool.go

@@ -79,3 +79,9 @@ func (p *Pool) connectToDoppler(addr string) {
 		return
 	}
 }
+
+func (p *Pool) Size() int {
+	p.mu.Lock()
+	defer p.mu.Unlock()
+	return len(p.dopplers)
+}

src/plumbing/pool_test.go

@@ -1,8 +1,6 @@
 package plumbing_test
 
 import (
-	"net"
-
 	"code.cloudfoundry.org/loggregator-release/src/plumbing"
 
 	"golang.org/x/net/context"
@@ -16,62 +14,33 @@ import (
 var _ = Describe("Pool", func() {
 	var (
 		pool *plumbing.Pool
-
-		listeners []net.Listener
-		servers   []*grpc.Server
 	)
 
 	BeforeEach(func() {
 		pool = plumbing.NewPool(grpc.WithTransportCredentials(insecure.NewCredentials()))
 	})
 
-	AfterEach(func() {
-		for _, lis := range listeners {
-			lis.Close()
-		}
-		listeners = nil
-
-		for _, server := range servers {
-			server.Stop()
-		}
-		servers = nil
-	})
-
-	Describe("Register() & Close()", func() {
-		var (
-			lis1, lis2           net.Listener
-			accepter1, accepter2 chan bool
-		)
-
+	Describe("Register()", func() {
 		BeforeEach(func() {
-			lis1, accepter1 = accepter(startListener("127.0.0.1:0"))
-			lis2, accepter2 = accepter(startListener("127.0.0.1:0"))
-			listeners = append(listeners, lis1, lis2)
+			pool.RegisterDoppler("192.0.2.10:8080")
+			pool.RegisterDoppler("192.0.2.11:8080")
 		})
 
-		Describe("Register()", func() {
-			It("fills pool with connections to each doppler", func() {
-				pool.RegisterDoppler(lis1.Addr().String())
-				pool.RegisterDoppler(lis2.Addr().String())
-
-				Eventually(accepter1).Should(HaveLen(1))
-				Eventually(accepter2).Should(HaveLen(1))
-			})
+		It("adds entries to the pool", func() {
+			Eventually(pool.Size).Should(Equal(2))
 		})
+	})
 
-		Describe("Close()", func() {
-			BeforeEach(func() {
-				pool.RegisterDoppler(lis1.Addr().String())
-			})
-
-			It("stops the gRPC connections", func() {
-				pool.Close(lis1.Addr().String())
-				lis1.Close()
+	Describe("Close()", func() {
+		BeforeEach(func() {
+			pool.RegisterDoppler("192.0.2.10:8080")
+			pool.RegisterDoppler("192.0.2.11:8080")
+		})
 
-				// Drain the channel
-				Eventually(accepter1, 5).ShouldNot(Receive())
-				Consistently(accepter1).Should(HaveLen(0))
-			})
+		It("removes entries from the pool", func() {
+			Eventually(pool.Size).Should(Equal(2))
+			pool.Close("192.0.2.11:8080")
+			Eventually(pool.Size).Should(Equal(1))
 		})
 	})
 
@@ -173,20 +142,3 @@ func fetchRx(
 	Eventually(f).ShouldNot(HaveOccurred())
 	return rx
 }
-
-func accepter(lis net.Listener) (net.Listener, chan bool) {
-	c := make(chan bool, 100)
-	go func() {
-		var dontGC []net.Conn
-		for {
-			conn, err := lis.Accept()
-			if err != nil {
-				return
-			}
-
-			dontGC = append(dontGC, conn) //nolint: staticcheck
-			c <- true
-		}
-	}()
-	return lis, c
-}

src/rlp/internal/ingress/pool.go

@@ -123,3 +123,9 @@ func (p *Pool) connectToDoppler(addr string, clients []unsafe.Pointer, idx int)
 		return
 	}
 }
+
+func (p *Pool) Size() int {
+	p.mu.Lock()
+	defer p.mu.Unlock()
+	return len(p.dopplers)
+}

src/rlp/internal/ingress/pool_test.go

@@ -23,49 +23,27 @@ var _ = Describe("Pool", func() {
 		pool = ingress.NewPool(2, grpc.WithTransportCredentials(insecure.NewCredentials()))
 	})
 
-	Describe("Register() & Close()", func() {
-		var (
-			listeners            []net.Listener
-			lis1, lis2           net.Listener
-			accepter1, accepter2 chan bool
-		)
-
+	Describe("Register()", func() {
 		BeforeEach(func() {
-			lis1, accepter1 = accepter(startListener("127.0.0.1:0"))
-			lis2, accepter2 = accepter(startListener("127.0.0.1:0"))
-			listeners = append(listeners, lis1, lis2)
+			pool.RegisterDoppler("192.0.2.10:8080")
+			pool.RegisterDoppler("192.0.2.11:8080")
 		})
 
-		AfterEach(func() {
-			for _, lis := range listeners {
-				lis.Close()
-			}
-			listeners = nil
+		It("adds entries to the pool", func() {
+			Eventually(pool.Size).Should(Equal(2))
 		})
+	})
 
-		Describe("Register()", func() {
-			It("fills pool with connections to each doppler", func() {
-				pool.RegisterDoppler(lis1.Addr().String())
-				pool.RegisterDoppler(lis2.Addr().String())
-
-				Eventually(accepter1).Should(HaveLen(2))
-				Eventually(accepter2).Should(HaveLen(2))
-			})
+	Describe("Close()", func() {
+		BeforeEach(func() {
+			pool.RegisterDoppler("192.0.2.10:8080")
+			pool.RegisterDoppler("192.0.2.11:8080")
 		})
 
-		Describe("Close()", func() {
-			BeforeEach(func() {
-				pool.RegisterDoppler(lis1.Addr().String())
-			})
-
-			It("stops the gRPC connections", func() {
-				pool.Close(lis1.Addr().String())
-				lis1.Close()
-
-				// Drain the channel
-				Eventually(accepter1, 5).ShouldNot(Receive())
-				Consistently(accepter1).Should(HaveLen(0))
-			})
+		It("removes entries from the pool", func() {
+			Eventually(pool.Size).Should(Equal(2))
+			pool.Close("192.0.2.11:8080")
+			Eventually(pool.Size).Should(Equal(1))
 		})
 	})