webnet/tailscale
1
0
Fork 0
Code Issues Pull Requests 1 Packages Projects Releases Wiki Activity

net/portmapper: fire an event when a port mapping is updated (#15371)

Browse Source 
When an event bus is configured publish an event each time a new port mapping
is updated. Publication is unconditional and occurs prior to calling any
callback that is registered. For now, the callback is still fired in a separate
goroutine as before -- later, those callbacks should become subscriptions to
the published event.

For now, the event type is defined as a new type here in the package. We will
want to move it to a more central package when there are subscribers. The event
wrapper is effectively a subset of the data exported by the internal mapping
interface, but on a concrete struct so the bus plumbing can inspect it.

Updates #15160

Change-Id: I951f212429ac791223af8d75b6eb39a0d2a0053a
Signed-off-by: M. J. Fromberger <fromberger@tailscale.com>
This commit is contained in:
M. J. Fromberger
2025-03-20 15:19:26 -07:00
committed by M. J. Fromberger
parent deb0b255ff
commit 2731171c5e
5 changed files with 115 additions and 50 deletions
Download Patch File Download Diff File Expand all files Collapse all files
net/portmapper/portmapper_test.go
+32 -7
View File
@@ -12,6 +12,7 @@ import (
"time"
"tailscale.com/control/controlknobs"
"tailscale.com/util/eventbus"
)
func TestCreateOrGetMapping(t *testing.T) {
@@ -25,7 +26,7 @@ func TestCreateOrGetMapping(t *testing.T) {
if i > 0 {
time.Sleep(100 * time.Millisecond)
}
ext, err := c.createOrGetMapping(context.Background())
_, ext, err := c.createOrGetMapping(context.Background())
t.Logf("Got: %v, %v", ext, err)
}
}
@@ -55,7 +56,7 @@ func TestClientProbeThenMap(t *testing.T) {
c.SetLocalPort(1234)
res, err := c.Probe(context.Background())
t.Logf("Probe: %+v, %v", res, err)
ext, err := c.createOrGetMapping(context.Background())
_, ext, err := c.createOrGetMapping(context.Background())
t.Logf("createOrGetMapping: %v, %v", ext, err)
}
@@ -66,9 +67,8 @@ func TestProbeIntegration(t *testing.T) {
}
defer igd.Close()
c := newTestClient(t, igd)
c := newTestClient(t, igd, nil)
t.Logf("Listening on pxp=%v, upnp=%v", c.testPxPPort, c.testUPnPPort)
defer c.Close()
res, err := c.Probe(context.Background())
if err != nil {
@@ -101,8 +101,7 @@ func TestPCPIntegration(t *testing.T) {
}
defer igd.Close()
c := newTestClient(t, igd)
defer c.Close()
c := newTestClient(t, igd, nil)
res, err := c.Probe(context.Background())
if err != nil {
t.Fatalf("probe failed: %v", err)
@@ -114,7 +113,7 @@ func TestPCPIntegration(t *testing.T) {
t.Fatalf("probe did not see pcp: %+v", res)
}
external, err := c.createOrGetMapping(context.Background())
_, external, err := c.createOrGetMapping(context.Background())
if err != nil {
t.Fatalf("failed to get mapping: %v", err)
}
@@ -136,3 +135,29 @@ func TestGetUPnPErrorsMetric(t *testing.T) {
getUPnPErrorsMetric(0)
getUPnPErrorsMetric(-100)
}
func TestUpdateEvent(t *testing.T) {
igd, err := NewTestIGD(t.Logf, TestIGDOptions{PCP: true})
if err != nil {
t.Fatalf("Create test gateway: %v", err)
}
bus := eventbus.New()
defer bus.Close()
sub := eventbus.Subscribe[Mapping](bus.Client("TestUpdateEvent"))
c := newTestClient(t, igd, bus)
if _, err := c.Probe(t.Context()); err != nil {
t.Fatalf("Probe failed: %v", err)
}
c.GetCachedMappingOrStartCreatingOne()
select {
case evt := <-sub.Events():
t.Logf("Received portmap update: %+v", evt)
case <-sub.Done():
t.Error("Subscriber closed prematurely")
case <-time.After(5 * time.Second):
t.Error("Timed out waiting for an update event")
}
}