diff --git a/feature/conn25/addrAssignments.go b/feature/conn25/addrAssignments.go index 7c700adb4..3c05b8cc7 100644 --- a/feature/conn25/addrAssignments.go +++ b/feature/conn25/addrAssignments.go @@ -101,16 +101,50 @@ func (a *addrAssignments) lookupByTransitIP(tip netip.Addr) (*addrs, bool) { return v, true } -// popExpired returns the member of addrAssignments that expired earliest, -// or an invalid addrs if there are no expired members of addrAssignments. +const ( + // deadFlowWaitTimeout is the minimum time after the active flow count + // drops to zero that we keep an address mapping in our table of address + // mappings. + deadFlowWaitTimeout = 2 * time.Minute + + // extendForActiveFlowDuration is the minimum time we will wait to recheck + // an address mapping with a positive active flow count for removal from + // the table of address mappings. + extendForActiveFlowDuration = 24 * time.Hour +) + +// popExpired attempts to remove from all the indexes one address +// mapping, and return that mapping, or nil if there were no eligible mappings. +// An address mapping is eligible for removal if: +// - the current time is past the expiresAt time on the mapping +// - and, the active flow count is 0 +// - and, it's been long enough since the active flow count dropped to 0 +// We're using a heap on expiresAt to efficiently find addresses that +// are eligible for removal. expiresAt is initially set according to the +// TTL on the DNS response. If the current time is past the expiresAt, but +// there are active flows, we extend the expiresAt time into the future. func (a *addrAssignments) popExpired(now time.Time) *addrs { if a.byExpiresAt.Len() == 0 { return nil } - if !a.byExpiresAt.peek().expiresAt.Before(now) { - return nil + var v *addrs + // Look for an address we can remove. + for { + if !a.byExpiresAt.peek().expiresAt.Before(now) { + // There's no longer anything outside the expiry window. + return nil + } + candidate := heap.Pop(&a.byExpiresAt).(*addrs) + if candidate.activeFlowCount == 0 && candidate.zeroFlowTime.Add(deadFlowWaitTimeout).Before(now) { + // Found one. + v = candidate + break + } + // Candidate can't be removed due to active flows. Extend expiresAt, and put it back in the heap. + candidate.expiresAt = now.Add(extendForActiveFlowDuration) + // TODO(mzb/fran): This is an expensive operation we could consider optimizing. + heap.Push(&a.byExpiresAt, candidate) } - v := heap.Pop(&a.byExpiresAt).(*addrs) delete(a.byMagicIP, v.magic) delete(a.byTransitIP, v.transit) dd := domainDst{domain: v.domain, dst: v.dst} diff --git a/feature/conn25/conn25.go b/feature/conn25/conn25.go index 504698871..e5433cd54 100644 --- a/feature/conn25/conn25.go +++ b/feature/conn25/conn25.go @@ -292,13 +292,17 @@ func (c *Conn25) ClientTransitIPForMagicIP(m netip.Addr) (netip.Addr, error) { } // ClientFlowCreated implements [Conn25Datapath]. +// The datapath notifies Conn25 that a flow with transitIP has been created so +// that Conn25 can prevent that transit IP and associated addresses from being +// removed from its state and returned to their pools. func (c *Conn25) ClientFlowCreated(transitIP netip.Addr) { - // TODO(tailscale/corp#43180): manage state for address assignment expiry + c.client.flowCreated(transitIP) } // ClientFlowRemoved implements [Conn25Datapath]. +// See [Conn25.ClientFlowCreated]. func (c *Conn25) ClientFlowRemoved(transitIP netip.Addr) { - // TODO(tailscale/corp#43180): manage state for address assignment expiry + c.client.flowRemoved(transitIP) } // ConnectorRealIPForTransitIPConnection implements [Conn25Datapath]. @@ -947,6 +951,29 @@ func (c *client) enqueueAddressAssignment(addrs *addrs) error { } } +func (c *client) flowCreated(transit netip.Addr) { + c.mu.Lock() + defer c.mu.Unlock() + entry, ok := c.assignments.byTransitIP[transit] + if !ok { + return + } + entry.activeFlowCount++ +} + +func (c *client) flowRemoved(transit netip.Addr) { + c.mu.Lock() + defer c.mu.Unlock() + entry, ok := c.assignments.byTransitIP[transit] + if !ok { + return + } + entry.activeFlowCount-- + if entry.activeFlowCount == 0 { + entry.zeroFlowTime = c.assignments.clock.Now() + } +} + func (c *client) extraWireGuardAllowedIPs(k key.NodePublic) views.Slice[netip.Prefix] { c.mu.Lock() defer c.mu.Unlock() @@ -1344,12 +1371,14 @@ func (c *connector) lookupBySrcIPAndTransitIP(srcIP, transitIP netip.Addr) (appA } type addrs struct { - dst netip.Addr - magic netip.Addr - transit netip.Addr - domain dnsname.FQDN - app string - expiresAt time.Time + dst netip.Addr + magic netip.Addr + transit netip.Addr + domain dnsname.FQDN + app string + expiresAt time.Time + activeFlowCount int + zeroFlowTime time.Time } func (as addrs) isValid() bool { diff --git a/feature/conn25/conn25_test.go b/feature/conn25/conn25_test.go index 082dcbc78..0c404c76b 100644 --- a/feature/conn25/conn25_test.go +++ b/feature/conn25/conn25_test.go @@ -1531,6 +1531,29 @@ func parseResponse(t *testing.T, buf []byte) ([]dnsmessage.Resource, []dnsmessag return answers, additionals } +func compareToRecords(t *testing.T, resources []dnsmessage.Resource, want []netip.Addr) { + t.Helper() + var got []netip.Addr + for _, r := range resources { + if b, ok := r.Body.(*dnsmessage.AResource); ok { + got = append(got, netip.AddrFrom4(b.A)) + } else if b, ok := r.Body.(*dnsmessage.AAAAResource); ok { + got = append(got, netip.AddrFrom16(b.AAAA)) + } + } + if diff := cmp.Diff(want, got, cmpopts.EquateComparable(netip.Addr{})); diff != "" { + t.Fatalf("A/AAAA records mismatch (-want +got):\n%s", diff) + } +} + +func assertParsesToAnswers(want []netip.Addr) func(t *testing.T, bs []byte) { + return func(t *testing.T, bs []byte) { + t.Helper() + answers, _ := parseResponse(t, bs) + compareToRecords(t, answers, want) + } +} + func TestMapDNSResponseRewritesResponses(t *testing.T) { configuredDomain := "example.com" domainName := configuredDomain + "." @@ -1548,29 +1571,6 @@ func TestMapDNSResponseRewritesResponses(t *testing.T) { cfg := mustConfig(t, sn) - compareToRecords := func(t *testing.T, resources []dnsmessage.Resource, want []netip.Addr) { - t.Helper() - var got []netip.Addr - for _, r := range resources { - if b, ok := r.Body.(*dnsmessage.AResource); ok { - got = append(got, netip.AddrFrom4(b.A)) - } else if b, ok := r.Body.(*dnsmessage.AAAAResource); ok { - got = append(got, netip.AddrFrom16(b.AAAA)) - } - } - if diff := cmp.Diff(want, got, cmpopts.EquateComparable(netip.Addr{})); diff != "" { - t.Fatalf("A/AAAA records mismatch (-want +got):\n%s", diff) - } - } - - assertParsesToAnswers := func(want []netip.Addr) func(t *testing.T, bs []byte) { - return func(t *testing.T, bs []byte) { - t.Helper() - answers, _ := parseResponse(t, bs) - compareToRecords(t, answers, want) - } - } - assertParsesToAdditionals := func(want []netip.Addr) func(t *testing.T, bs []byte) { return func(t *testing.T, bs []byte) { t.Helper() @@ -2596,3 +2596,212 @@ func TestReconfigDoesNotReissueInUseAddresses(t *testing.T) { }) } } + +// TestAddressExpiryDependsOnActiveFlows creates a Conn25 and +// +// 1. runs a DNS response through it +// 2. uses the ClientFlowCreated/Removed API and advances the clock +// 3. runs a second DNS response through the Conn25 +// 4. asserts things about the expected state of the clients assignments +// table based on 1-3 +// +// to try and verify how the assignments table entries expiration is affected +// by the presence of active flows for the addresses in the entry +func TestAddressExpiryDependsOnActiveFlows(t *testing.T) { + configuredDomain := "example.com" + domainName := configuredDomain + "." + dnsMessageName := dnsmessage.MustNewName(domainName) + sn := makeSelfNode(t, []appctype.Conn25Attr{{ + Name: "app1", + Connectors: []string{"tag:woo"}, + Domains: []string{configuredDomain}, + }}, appctype.Conn25PoolsAttr{ + V4MagicIPPool: []netipx.IPRange{v4RangeFrom("0", "10")}, + V4TransitIPPool: []netipx.IPRange{v4RangeFrom("40", "50")}, + V6MagicIPPool: []netipx.IPRange{netipx.IPRangeFrom(netip.MustParseAddr("2606:4700::6812:100"), netip.MustParseAddr("2606:4700::6812:1ff"))}, + V6TransitIPPool: []netipx.IPRange{netipx.IPRangeFrom(netip.MustParseAddr("2606:4700::6813:100"), netip.MustParseAddr("2606:4700::6813:1ff"))}, + }, nil) + + var ttlSecs uint32 = 300 + ttlDur := time.Duration(ttlSecs) * time.Second + + ipOne := netip.MustParseAddr("1.0.0.1") + dnsRespIPOne := makeDNSResponseForSections(t, + []dnsmessage.Question{{Name: dnsMessageName, Type: dnsmessage.TypeA, Class: dnsmessage.ClassINET}}, + []dnsmessage.Resource{ + { + Header: dnsmessage.ResourceHeader{Name: dnsMessageName, Type: dnsmessage.TypeA, Class: dnsmessage.ClassINET, TTL: ttlSecs}, + Body: &dnsmessage.AResource{A: ipOne.As4()}, + }, + }, + nil, + ) + + ipTwo := netip.MustParseAddr("1.0.0.2") + dnsRespIPTwo := makeDNSResponseForSections(t, + []dnsmessage.Question{{Name: dnsMessageName, Type: dnsmessage.TypeA, Class: dnsmessage.ClassINET}}, + []dnsmessage.Resource{ + { + Header: dnsmessage.ResourceHeader{Name: dnsMessageName, Type: dnsmessage.TypeA, Class: dnsmessage.ClassINET, TTL: ttlSecs}, + Body: &dnsmessage.AResource{A: ipTwo.As4()}, + }, + }, + nil, + ) + + tests := []struct { + name string + flowsAndTimeFx func(*Conn25, *tstest.Clock, netip.Addr) + secondDNSResponse []byte + assertSecondDNSResponse func(*testing.T, []byte) + wantUnexpiredDstIPs set.Set[netip.Addr] + wantExpiredAtTime map[netip.Addr]time.Duration // since the startTime + }{ + { + // The first dns response should create an assignments entry for ipOne + // (tested elsewhere). + // Then time advances past that entry's expiresAt. + // Then a second dns response creates an assignments entry for ipTwo. + // We clean up some expired assignments entries when we create a new one + // and so we expect the entry for ipOne to be removed, and the entry for + // ipTwo to be present. + name: "flows-zero", + flowsAndTimeFx: func(c *Conn25, clock *tstest.Clock, transit netip.Addr) { + clock.Advance(30 * time.Hour) + }, + wantUnexpiredDstIPs: set.SetOf([]netip.Addr{ipTwo}), + wantExpiredAtTime: map[netip.Addr]time.Duration{ + ipTwo: (30 * time.Hour) + ttlDur, + }, + }, + { + // Same as flows-zero except this time the datapath has let us know that + // there is a flow for the transit address that was assigned to the entry for + // ipOne. + // And so that entry does not get expired. + name: "flows-not-zero", + flowsAndTimeFx: func(c *Conn25, clock *tstest.Clock, transit netip.Addr) { + c.ClientFlowCreated(transit) + clock.Advance(30 * time.Hour) + }, + wantUnexpiredDstIPs: set.SetOf([]netip.Addr{ipOne, ipTwo}), + wantExpiredAtTime: map[netip.Addr]time.Duration{ + ipOne: (30 * time.Hour) + extendForActiveFlowDuration, + ipTwo: (30 * time.Hour) + ttlDur, + }, + }, + { + // Like flows-not-zero except that this time the datapath removed the + // client flow after creating it. + // So the expired entry is removed. + name: "last-flow-removed-a-while-ago", + flowsAndTimeFx: func(c *Conn25, clock *tstest.Clock, transit netip.Addr) { + c.ClientFlowCreated(transit) + clock.Advance(30 * time.Hour) + c.ClientFlowRemoved(transit) + clock.Advance(3 * time.Minute) + }, + wantUnexpiredDstIPs: set.SetOf([]netip.Addr{ipTwo}), + wantExpiredAtTime: map[netip.Addr]time.Duration{ + ipTwo: (30 * time.Hour) + (3 * time.Minute) + ttlDur, + }, + }, + { + // Like last-flow-removed-a-while-ago except the flow was removed recently, + // within the cooldown period. + // And so the expired entry is not removed. + name: "last-flow-recently-removed", + flowsAndTimeFx: func(c *Conn25, clock *tstest.Clock, transit netip.Addr) { + c.ClientFlowCreated(transit) + clock.Advance(30 * time.Hour) + c.ClientFlowRemoved(transit) + clock.Advance(1 * time.Second) + }, + wantUnexpiredDstIPs: set.SetOf([]netip.Addr{ipOne, ipTwo}), + wantExpiredAtTime: map[netip.Addr]time.Duration{ + ipOne: (30 * time.Hour) + extendForActiveFlowDuration + (1 * time.Second), + ipTwo: (30 * time.Hour) + (1 * time.Second) + ttlDur, + }, + }, + { + // Like flows-not-zero except that the second dns response is for the same address as the first. + // So the entry is not removed. + name: "repeated-response-with-expired-and-active-flow", + secondDNSResponse: dnsRespIPOne, + flowsAndTimeFx: func(c *Conn25, clock *tstest.Clock, transit netip.Addr) { + c.ClientFlowCreated(transit) + clock.Advance(30 * time.Hour) + }, + wantUnexpiredDstIPs: set.SetOf([]netip.Addr{ipOne}), + wantExpiredAtTime: map[netip.Addr]time.Duration{ + ipOne: (30 * time.Hour) + ttlDur, + }, + assertSecondDNSResponse: assertParsesToAnswers( + []netip.Addr{ + netip.MustParseAddr("100.64.0.0"), + }, + ), + }, + } + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + c := newConn25(logger.Discard) + startTime := time.Now() + clock := tstest.NewClock(tstest.ClockOpts{Start: startTime}) + c.client.assignments.clock = clock + cfg := mustConfig(t, sn) + c.reconfig(cfg) + + // we get a dns response for ipone + bs1 := c.mapDNSResponse(dnsRespIPOne) + assertParsesToAnswers( + []netip.Addr{ + netip.MustParseAddr("100.64.0.0"), + }, + )(t, bs1) + + ipOneDD := domainDst{ + domain: dnsname.FQDN(domainName), + dst: ipOne, + } + + // there are client flows and time passes + tt.flowsAndTimeFx(c, clock, c.client.assignments.byDomainDst[ipOneDD].transit) + + // then a second dns response + dnsR2 := tt.secondDNSResponse + assertSecondResponseFx := tt.assertSecondDNSResponse + if dnsR2 == nil { + dnsR2 = dnsRespIPTwo + assertSecondResponseFx = assertParsesToAnswers( + []netip.Addr{ + netip.MustParseAddr("100.64.0.1"), + }, + ) + } + bs2 := c.mapDNSResponse(dnsR2) + assertSecondResponseFx(t, bs2) + + // assert which addresses have expired / remain unexpired + assignmentsDsts := set.Set[netip.Addr]{} + for _, a := range c.client.assignments.byMagicIP { + assignmentsDsts.Add(a.dst) + } + if !assignmentsDsts.Equal(tt.wantUnexpiredDstIPs) { + t.Fatalf("unexpired dst IPs: want: %v, got %v", tt.wantUnexpiredDstIPs, assignmentsDsts) + } + + for a, dur := range tt.wantExpiredAtTime { + dd := domainDst{ + domain: dnsname.FQDN(domainName), + dst: a, + } + as := c.client.assignments.byDomainDst[dd] + expected := startTime.Add(dur) + if !as.expiresAt.Equal(expected) { + t.Fatalf("a: %v, as.ExpiredAt: %v, expected: %v, dur: %v", a, as.expiresAt, expected, dur) + } + } + }) + } +}