internal/agent/exposeproxy/exposeproxy_test.go
Ref: Size: 22.5 KiB History
package exposeproxy
import (
"io"
"net"
"strconv"
"strings"
"syscall"
"testing"
"time"
"github.com/a73x/eitri/internal/pb"
"github.com/stretchr/testify/assert"
"github.com/stretchr/testify/require"
)
// fakeGuest is a loopback listener standing in for a service inside a guest.
// It echoes everything it reads back, prefixed, so a test can prove bytes went
// both ways through the proxy.
type fakeGuest struct {
ln net.Listener
addr string
port uint32
}
func newFakeGuest(t *testing.T) *fakeGuest {
t.Helper()
ln, err := net.Listen("tcp", "127.0.0.1:0")
require.NoError(t, err)
t.Cleanup(func() { ln.Close() })
g := &fakeGuest{ln: ln, addr: "127.0.0.1", port: portOf(t, ln.Addr())}
go func() {
for {
c, err := ln.Accept()
if err != nil {
return
}
go func() {
defer c.Close()
buf := make([]byte, 64)
n, err := c.Read(buf)
if n > 0 {
c.Write(append([]byte("echo:"), buf[:n]...)) //nolint:errcheck
}
if err != nil {
return
}
}()
}
}()
return g
}
func portOf(t *testing.T, a net.Addr) uint32 {
t.Helper()
ta, ok := a.(*net.TCPAddr)
require.True(t, ok)
return uint32(ta.Port)
}
// newTestManager builds a Manager whose address answer is a fixed map.
func newTestManager(t *testing.T, addrs map[string]string) *Manager {
t.Helper()
m := NewManager(func(vmID string) string { return addrs[vmID] })
t.Cleanup(m.StopAll)
return m
}
// boundPort reads the port the manager's listener for id actually bound. Host
// port 0 in a desired spec lets the OS pick, which is how these tests avoid
// racing for a fixed number.
func boundPort(t *testing.T, m *Manager, id string) string {
t.Helper()
m.mu.Lock()
defer m.mu.Unlock()
ex, ok := m.live[id]
require.True(t, ok, "no listener for %q", id)
require.NotNil(t, ex.ln, "listener for %q is not bound", id)
return ex.ln.Addr().String()
}
func desired(id, vmID string, guestPort, hostPort uint32) *pb.ExposureSpec {
return &pb.ExposureSpec{Id: id, VmId: vmID, GuestPort: guestPort, HostPort: hostPort, Protocol: "tcp"}
}
func desiredUDP(id, vmID string, guestPort, hostPort uint32) *pb.ExposureSpec {
return &pb.ExposureSpec{Id: id, VmId: vmID, GuestPort: guestPort, HostPort: hostPort, Protocol: "udp"}
}
// speak dials addr, sends msg, and returns what came back.
func speak(t *testing.T, addr, msg string) string {
t.Helper()
c, err := net.Dial("tcp", addr)
require.NoError(t, err)
defer c.Close()
require.NoError(t, c.SetDeadline(time.Now().Add(5*time.Second)))
_, err = c.Write([]byte(msg))
require.NoError(t, err)
out, err := io.ReadAll(c)
require.NoError(t, err)
return string(out)
}
func TestConvergeBindsAndPipesToTheGuest(t *testing.T) {
g := newFakeGuest(t)
m := newTestManager(t, map[string]string{"vm1": g.addr})
got := m.Converge([]*pb.ExposureSpec{desired("e1", "vm1", g.port, 0)})
require.Len(t, got, 1)
assert.Equal(t, "e1", got[0].GetId())
assert.Equal(t, "active", got[0].GetState())
assert.Equal(t, "echo:hello", speak(t, boundPort(t, m, "e1"), "hello"))
}
func TestConvergeIsIdempotent(t *testing.T) {
g := newFakeGuest(t)
m := newTestManager(t, map[string]string{"vm1": g.addr})
m.Converge([]*pb.ExposureSpec{desired("e1", "vm1", g.port, 0)})
addr := boundPort(t, m, "e1")
m.Converge([]*pb.ExposureSpec{desired("e1", "vm1", g.port, 0)})
assert.Equal(t, addr, boundPort(t, m, "e1"), "an unchanged spec keeps its listener")
assert.Equal(t, "echo:hi", speak(t, addr, "hi"))
}
func TestConvergeClosesAVanishedExposure(t *testing.T) {
g := newFakeGuest(t)
m := newTestManager(t, map[string]string{"vm1": g.addr})
m.Converge([]*pb.ExposureSpec{desired("e1", "vm1", g.port, 0)})
addr := boundPort(t, m, "e1")
got := m.Converge(nil)
assert.Empty(t, got)
_, err := net.Dial("tcp", addr)
assert.Error(t, err, "a revoked exposure stops answering")
}
func TestConvergeRebindsAChangedSpec(t *testing.T) {
first := newFakeGuest(t)
second := newFakeGuest(t)
m := newTestManager(t, map[string]string{"vm1": first.addr})
m.Converge([]*pb.ExposureSpec{desired("e1", "vm1", first.port, 0)})
oldAddr := boundPort(t, m, "e1")
m.Converge([]*pb.ExposureSpec{desired("e1", "vm1", second.port, 0)})
newAddr := boundPort(t, m, "e1")
assert.NotEqual(t, oldAddr, newAddr, "a changed spec is a close and a rebind")
// The new listener reaches the new guest port.
assert.Equal(t, "echo:x", speak(t, newAddr, "x"))
}
func TestConvergeReportsABindFailureAndHealsWhenThePortFrees(t *testing.T) {
g := newFakeGuest(t)
m := newTestManager(t, map[string]string{"vm1": g.addr})
// Squat on a host port, then ask for exactly it.
squatter, err := net.Listen("tcp", "127.0.0.1:0")
require.NoError(t, err)
held := portOf(t, squatter.Addr())
got := m.Converge([]*pb.ExposureSpec{desired("e1", "vm1", g.port, held)})
require.Len(t, got, 1)
assert.Equal(t, "failed", got[0].GetState())
assert.NotEmpty(t, got[0].GetReason(), "the report carries what the OS said")
// The squatter exits; the very next converge binds. A bind-time refusal is
// not an error path, it is a level to converge on.
require.NoError(t, squatter.Close())
var state string
for i := 0; i < 20; i++ {
got = m.Converge([]*pb.ExposureSpec{desired("e1", "vm1", g.port, held)})
state = got[0].GetState()
if state == "active" {
break
}
time.Sleep(50 * time.Millisecond)
}
assert.Equal(t, "active", state)
assert.Equal(t, "echo:back", speak(t, net.JoinHostPort("127.0.0.1", strconv.Itoa(int(held))), "back"))
}
func TestConvergeRebindsAListenerWhoseAcceptLoopDied(t *testing.T) {
g := newFakeGuest(t)
m := newTestManager(t, map[string]string{"vm1": g.addr})
m.Converge([]*pb.ExposureSpec{desired("e1", "vm1", g.port, 0)})
// Break the listener behind the manager's back, the way fd exhaustion would
// end an accept loop: the manager never asked for this and still believes
// the exposure is served.
m.mu.Lock()
old := m.live["e1"].ln
require.NotNil(t, old)
require.NoError(t, old.Close())
m.mu.Unlock()
// The loop hands the listener back and records why it left. Both are read
// under the lock the loop writes them under, which is what makes the reason
// deterministic rather than a sampled guess.
var reason string
for i := 0; i < 200; i++ {
m.mu.Lock()
ex := m.live["e1"]
gone, r := ex.ln == nil, ex.reason
m.mu.Unlock()
if gone {
reason = r
break
}
time.Sleep(10 * time.Millisecond)
}
assert.Contains(t, reason, "accept:", "a dead loop must not leave a listener nothing is serving")
// The bind-failure level-trigger is the whole healing mechanism: the next
// converge sees no listener and binds one. This exposure asked for host port
// 0, so that rebind cannot fail and no converge ever reports "failed" — the
// interim report is only observable when the fresh bind is also refused,
// which is the case TestConvergeReportsABindFailure... already covers.
got := m.Converge([]*pb.ExposureSpec{desired("e1", "vm1", g.port, 0)})
require.Len(t, got, 1)
assert.Equal(t, "active", got[0].GetState())
m.mu.Lock()
rebound := m.live["e1"].ln
m.mu.Unlock()
// Compared as pointers, not with NotEqual: a deep-equality walk would read
// the live listener's internals while its accept loop is using them.
assert.True(t, rebound != old, "a dead listener is rebound, not believed")
assert.Equal(t, "echo:again", speak(t, boundPort(t, m, "e1"), "again"))
}
func TestConnectionClosesWhenTheGuestHasNoAddress(t *testing.T) {
m := newTestManager(t, map[string]string{}) // the guest is still leasing
m.Converge([]*pb.ExposureSpec{desired("e1", "vm1", 8080, 0)})
c, err := net.Dial("tcp", boundPort(t, m, "e1"))
require.NoError(t, err)
defer c.Close()
require.NoError(t, c.SetDeadline(time.Now().Add(5*time.Second)))
out, err := io.ReadAll(c)
require.NoError(t, err)
assert.Empty(t, out, "no address yet means close, not wait")
}
func TestSpliceIsHalfCloseAware(t *testing.T) {
// A guest that answers only after the client has finished speaking — the
// shape every request/response protocol over a half-closed write takes.
ln, err := net.Listen("tcp", "127.0.0.1:0")
require.NoError(t, err)
t.Cleanup(func() { ln.Close() })
go func() {
c, err := ln.Accept()
if err != nil {
return
}
defer c.Close()
body, _ := io.ReadAll(c) // blocks until the client's write half closes
c.Write(append([]byte("saw:"), body...))
}()
m := newTestManager(t, map[string]string{"vm1": "127.0.0.1"})
m.Converge([]*pb.ExposureSpec{desired("e1", "vm1", portOf(t, ln.Addr()), 0)})
c, err := net.Dial("tcp", boundPort(t, m, "e1"))
require.NoError(t, err)
defer c.Close()
require.NoError(t, c.SetDeadline(time.Now().Add(5*time.Second)))
_, err = c.Write([]byte("request"))
require.NoError(t, err)
require.NoError(t, c.(*net.TCPConn).CloseWrite())
out, err := io.ReadAll(c)
require.NoError(t, err)
assert.Equal(t, "saw:request", string(out))
}
func TestRevokeDrainsRatherThanCuts(t *testing.T) {
// A guest that echoes for as long as the client keeps speaking, so one
// connection can carry two exchanges either side of the revocation.
ln, err := net.Listen("tcp", "127.0.0.1:0")
require.NoError(t, err)
t.Cleanup(func() { ln.Close() })
go func() {
for {
c, err := ln.Accept()
if err != nil {
return
}
go func() {
defer c.Close()
_, _ = io.Copy(c, c)
}()
}
}()
m := newTestManager(t, map[string]string{"vm1": "127.0.0.1"})
m.Converge([]*pb.ExposureSpec{desired("e1", "vm1", portOf(t, ln.Addr()), 0)})
addr := boundPort(t, m, "e1")
c, err := net.Dial("tcp", addr)
require.NoError(t, err)
defer c.Close()
require.NoError(t, c.SetDeadline(time.Now().Add(5*time.Second)))
exchange := func(msg string) string {
t.Helper()
_, err := c.Write([]byte(msg))
require.NoError(t, err)
buf := make([]byte, len(msg))
_, err = io.ReadFull(c, buf)
require.NoError(t, err)
return string(buf)
}
require.Equal(t, "one", exchange("one"), "the connection is spliced through before anything is revoked")
m.Converge(nil)
_, err = net.Dial("tcp", addr)
require.Error(t, err, "a revoked exposure stops answering new dials")
assert.Equal(t, "two", exchange("two"), "closing a listener drains: only new dials die")
}
// newHoldingGuest is a guest that echoes for as long as a caller keeps
// speaking, so a test can hold connections open across the proxy.
func newHoldingGuest(t *testing.T) *fakeGuest {
t.Helper()
ln, err := net.Listen("tcp", "127.0.0.1:0")
require.NoError(t, err)
t.Cleanup(func() { ln.Close() })
go func() {
for {
c, err := ln.Accept()
if err != nil {
return
}
go func() {
defer c.Close()
_, _ = io.Copy(c, c)
}()
}
}()
return &fakeGuest{ln: ln, addr: "127.0.0.1", port: portOf(t, ln.Addr())}
}
// hold opens a connection through the proxy and proves it is spliced through to
// the guest, which is also what makes the exposure's count of it deterministic:
// a byte came back, so the connection was counted before its pipe started.
func hold(t *testing.T, addr string) net.Conn {
t.Helper()
c, err := net.Dial("tcp", addr)
require.NoError(t, err)
t.Cleanup(func() { c.Close() })
require.NoError(t, c.SetDeadline(time.Now().Add(5*time.Second)))
require.Equal(t, "ping", exchange(t, c, "ping"))
return c
}
// exchange writes msg on an open connection and reads the echo back.
func exchange(t *testing.T, c net.Conn, msg string) string {
t.Helper()
_, err := c.Write([]byte(msg))
require.NoError(t, err)
buf := make([]byte, len(msg))
_, err = io.ReadFull(c, buf)
require.NoError(t, err)
return string(buf)
}
// waitConns waits for an exposure's held-connection count to reach want. The
// increment is synchronous with the accept, but the decrement happens when the
// pipe ends, which is a goroutine finishing on its own time.
func waitConns(t *testing.T, m *Manager, id string, want int64) {
t.Helper()
var got int64
for i := 0; i < 200; i++ {
m.mu.Lock()
ex, ok := m.live[id]
m.mu.Unlock()
require.True(t, ok, "no exposure %q", id)
if got = ex.conns.Load(); got == want {
return
}
time.Sleep(10 * time.Millisecond)
}
t.Fatalf("exposure %q holds %d connections, want %d", id, got, want)
}
func TestExposureRefusesConnectionsPastItsCap(t *testing.T) {
g := newHoldingGuest(t)
m := newTestManager(t, map[string]string{"vm1": g.addr})
m.maxConns = 2
m.Converge([]*pb.ExposureSpec{desired("e1", "vm1", g.port, 0)})
addr := boundPort(t, m, "e1")
first, second := hold(t, addr), hold(t, addr)
// The listener still accepts — the refusal is a close, not a bind that went
// away — and the caller learns immediately rather than waiting on a promise.
over, err := net.Dial("tcp", addr)
require.NoError(t, err)
defer over.Close()
require.NoError(t, over.SetDeadline(time.Now().Add(5*time.Second)))
out, err := io.ReadAll(over)
require.NoError(t, err)
assert.Empty(t, out, "a connection past the cap is closed at once")
assert.Equal(t, "still", exchange(t, first, "still"), "a refusal must not disturb the connections under the cap")
assert.Equal(t, "serving", exchange(t, second, "serving"))
}
func TestTheCapReleasesWhenAConnectionEnds(t *testing.T) {
g := newHoldingGuest(t)
m := newTestManager(t, map[string]string{"vm1": g.addr})
m.maxConns = 1
m.Converge([]*pb.ExposureSpec{desired("e1", "vm1", g.port, 0)})
addr := boundPort(t, m, "e1")
first := hold(t, addr)
require.NoError(t, first.Close())
waitConns(t, m, "e1", 0)
// The cap is a level, not a ratchet: the slot the closed connection held is
// the slot this one takes.
assert.Equal(t, "after", exchange(t, hold(t, addr), "after"))
}
func TestTheCapCountsConnectionsDrainingThroughARebind(t *testing.T) {
first, second := newHoldingGuest(t), newHoldingGuest(t)
m := newTestManager(t, map[string]string{"vm1": first.addr})
m.maxConns = 1
m.Converge([]*pb.ExposureSpec{desired("e1", "vm1", first.port, 0)})
held := hold(t, boundPort(t, m, "e1"))
// The spec changes and the exposure rebinds. The connection under the old
// listener is still open and still spending its two descriptors, so it is
// still what the cap is counting — an exposure that forgot it on the rebind
// would let this port hold twice what it is allowed.
m.Converge([]*pb.ExposureSpec{desired("e1", "vm1", second.port, 0)})
waitConns(t, m, "e1", 1)
over, err := net.Dial("tcp", boundPort(t, m, "e1"))
require.NoError(t, err)
defer over.Close()
require.NoError(t, over.SetDeadline(time.Now().Add(5*time.Second)))
out, err := io.ReadAll(over)
require.NoError(t, err)
assert.Empty(t, out, "the rebound listener is at its cap, because the draining connection still counts")
// The old connection drains rather than being cut, and the slot it holds is
// the one the next caller gets when it ends.
assert.Equal(t, "drain", exchange(t, held, "drain"))
require.NoError(t, held.Close())
waitConns(t, m, "e1", 0)
assert.Equal(t, "after", exchange(t, hold(t, boundPort(t, m, "e1")), "after"))
}
// waitReason waits for an exposure's reported reason to satisfy ok, reading it
// under the lock the accept loop writes it under.
func waitReason(t *testing.T, m *Manager, id string, ok func(string) bool, what string) string {
t.Helper()
var got string
for i := 0; i < 200; i++ {
m.mu.Lock()
ex, live := m.live[id]
if live {
got = ex.reason
}
m.mu.Unlock()
if live && ok(got) {
return got
}
time.Sleep(10 * time.Millisecond)
}
t.Fatalf("exposure %q reason %q: %s", id, got, what)
return ""
}
func TestADescriptorShortageIsReportedWhileThePortStaysBound(t *testing.T) {
g := newFakeGuest(t)
m := newTestManager(t, map[string]string{"vm1": g.addr})
d := desired("e1", "vm1", g.port, 0)
ln := &scriptedListener{next: make(chan acceptResult), addr: g.ln.Addr()}
ex := &exposure{key: specKey(d), ln: ln}
m.mu.Lock()
m.live["e1"] = ex
m.mu.Unlock()
go m.accept("e1", ex, ln, "vm1", g.port)
ln.next <- acceptResult{err: syscall.EMFILE}
waitReason(t, m, "e1", func(r string) bool { return strings.Contains(r, "out of descriptors") },
"a port nothing can get through must say why")
// The port is still bound, so it is still active — with the streak beside
// it, which is the whole point: "active" alone would read as healthy.
got := m.Converge([]*pb.ExposureSpec{d})
require.Len(t, got, 1)
assert.Equal(t, "active", got[0].GetState())
assert.Contains(t, got[0].GetReason(), "out of descriptors")
// Descriptors come back. The next accept clears the streak, and the report
// stops saying something is wrong.
client, accepted := net.Pipe()
defer client.Close()
ln.next <- acceptResult{conn: accepted}
waitReason(t, m, "e1", func(r string) bool { return r == "" }, "a recovered port must stop reporting a shortage")
got = m.Converge([]*pb.ExposureSpec{d})
assert.Equal(t, "active", got[0].GetState())
assert.Empty(t, got[0].GetReason())
}
// scriptedListener hands an accept loop exactly the sequence a test writes to
// it — the errors a real listener only produces under a load a test cannot
// stage.
type scriptedListener struct {
next chan acceptResult
addr net.Addr
}
type acceptResult struct {
conn net.Conn
err error
}
func (l *scriptedListener) Accept() (net.Conn, error) {
r := <-l.next
return r.conn, r.err
}
func (l *scriptedListener) Close() error { return nil }
func (l *scriptedListener) Addr() net.Addr { return l.addr }
func TestAcceptKeepsThePortThroughADescriptorShortage(t *testing.T) {
g := newFakeGuest(t)
m := newTestManager(t, map[string]string{"vm1": g.addr})
ln := &scriptedListener{next: make(chan acceptResult), addr: g.ln.Addr()}
ex := &exposure{key: "k", ln: ln}
m.mu.Lock()
m.live["e1"] = ex
m.mu.Unlock()
go m.accept("e1", ex, ln, "vm1", g.port)
// Out of descriptors is not a broken listener. Handing the port back would
// only ask the next converge to find a descriptor the process does not have.
ln.next <- acceptResult{err: syscall.EMFILE}
ln.next <- acceptResult{err: syscall.EMFILE}
client, accepted := net.Pipe()
defer client.Close()
ln.next <- acceptResult{conn: accepted}
require.NoError(t, client.SetDeadline(time.Now().Add(5*time.Second)))
_, err := client.Write([]byte("hello"))
require.NoError(t, err)
buf := make([]byte, len("echo:hello"))
_, err = io.ReadFull(client, buf)
require.NoError(t, err)
assert.Equal(t, "echo:hello", string(buf), "the loop kept its listener and went on serving")
// A listener that is genuinely broken still ends the loop and gives the
// exposure back its listener-less state for the next converge to heal.
ln.next <- acceptResult{err: net.ErrClosed}
var reason string
for i := 0; i < 200; i++ {
m.mu.Lock()
gone, r := ex.ln == nil, ex.reason
m.mu.Unlock()
if gone {
reason = r
break
}
time.Sleep(10 * time.Millisecond)
}
assert.Contains(t, reason, "accept:")
}
func TestStopAllClosesEveryListener(t *testing.T) {
g := newFakeGuest(t)
m := NewManager(func(string) string { return g.addr })
m.Converge([]*pb.ExposureSpec{
desired("e1", "vm1", g.port, 0),
desired("e2", "vm1", g.port, 0),
})
a1, a2 := boundPort(t, m, "e1"), boundPort(t, m, "e2")
m.StopAll()
for _, addr := range []string{a1, a2} {
_, err := net.Dial("tcp", addr)
assert.Error(t, err, "listener %s survived StopAll", addr)
}
}
// counters converges the same desired set again and returns what the named
// exposure reports about what it has carried. Converge is level-triggered, so
// asking twice is how the report is taken in production too.
func counters(t *testing.T, m *Manager, d []*pb.ExposureSpec, id string) *pb.ExposureSessions {
t.Helper()
for _, a := range m.Converge(d) {
if a.GetId() == id {
require.NotNil(t, a.GetSessions(), "exposure %q reported no counters", id)
return a.GetSessions()
}
}
t.Fatalf("exposure %q not in the report", id)
return nil
}
// TestExposureCountsWhatItHoldsAndWhatItTurnsAway is the observability leg: a
// published port that is quietly refusing callers looks exactly like a healthy
// one from outside, and these three numbers are the difference.
func TestExposureCountsWhatItHoldsAndWhatItTurnsAway(t *testing.T) {
g := newHoldingGuest(t)
m := newTestManager(t, map[string]string{"vm1": g.addr})
m.maxConns = 1
spec := []*pb.ExposureSpec{desired("e1", "vm1", g.port, 0)}
m.Converge(spec)
addr := boundPort(t, m, "e1")
held := hold(t, addr)
waitConns(t, m, "e1", 1)
got := counters(t, m, spec, "e1")
assert.Equal(t, int64(1), got.GetActive())
assert.Zero(t, got.GetRefused(), "nobody has been turned away yet")
// One caller past the cap, refused and closed.
over, err := net.Dial("tcp", addr)
require.NoError(t, err)
defer over.Close()
require.NoError(t, over.SetDeadline(time.Now().Add(5*time.Second)))
_, err = io.ReadAll(over)
require.NoError(t, err)
got = counters(t, m, spec, "e1")
assert.Equal(t, int64(1), got.GetActive(), "the refusal did not disturb the connection under the cap")
assert.Equal(t, int64(1), got.GetRefused())
// The refusal is the whole point of a TOTAL: once the port frees up, a gauge
// would read zero of everything and the burst would be gone.
held.Close()
waitConns(t, m, "e1", 0)
got = counters(t, m, spec, "e1")
assert.Zero(t, got.GetActive(), "active is a gauge and comes back down")
assert.Equal(t, int64(1), got.GetRefused(), "a refusal already counted is not un-counted")
}
// TestExposureCountsACallerItCannotCarry separates the port's own limit from the
// guest not being there: an operator reading "refused" should raise the cap, and
// one reading "dropped" should look at the guest.
func TestExposureCountsACallerItCannotCarry(t *testing.T) {
m := newTestManager(t, map[string]string{}) // no address for vm1: still leasing
spec := []*pb.ExposureSpec{desired("e1", "vm1", 8080, 0)}
m.Converge(spec)
addr := boundPort(t, m, "e1")
c, err := net.Dial("tcp", addr)
require.NoError(t, err)
defer c.Close()
require.NoError(t, c.SetDeadline(time.Now().Add(5*time.Second)))
_, err = io.ReadAll(c)
require.NoError(t, err)
waitConns(t, m, "e1", 0)
got := counters(t, m, spec, "e1")
assert.Equal(t, int64(1), got.GetDropped(), "a caller the host could not pipe to the guest is a drop")
assert.Zero(t, got.GetRefused(), "the cap refused nobody — the guest was not there")
}
// TestAFailedExposureStillReportsItsCounters: a port whose bind failed carries
// the message anyway, so "this agent counts" and "this port has counted
// nothing" stay one answer apart from "nobody said" for every exposure alike.
func TestAFailedExposureStillReportsItsCounters(t *testing.T) {
held, err := net.Listen("tcp", "127.0.0.1:0")
require.NoError(t, err)
defer held.Close()
m := newTestManager(t, map[string]string{"vm1": "127.0.0.1"})
port := portOf(t, held.Addr())
out := m.Converge([]*pb.ExposureSpec{desired("e1", "vm1", 8080, port)})
require.Len(t, out, 1)
require.Equal(t, "failed", out[0].GetState())
require.NotNil(t, out[0].GetSessions())
assert.Zero(t, out[0].GetSessions().GetActive())
}