Make throttling nmagent fetches for nodesubnet more dynamic

cns/nodesubnet/helper_for_ip_fetcher_test.go

@@ -2,8 +2,13 @@ package nodesubnet
 
 import "time"
 
-// This method is in this file (_test.go) because it is a test helper method.
-// The following method is built during tests, and is not part of the main code.
-func (c *IPFetcher) SetSecondaryIPQueryInterval(interval time.Duration) {
-	c.secondaryIPQueryInterval = interval
+// These methods is in this file (_test.go) because they are helpers. They are
+// built during tests, and are not part of the main code.
+
+func (c *IPFetcher) GetCurrentQueryInterval() time.Duration {
+	return c.tickerInterval
+}
+
+func (c *IPFetcher) SetTicker(tickProvider TickProvider) {
+	c.ticker = tickProvider
 }

cns/nodesubnet/ip_fetcher.go

@@ -10,46 +10,130 @@ import (
 	"github.com/pkg/errors"
 )
 
+const (
+	// Default minimum time between secondary IP fetches
+	DefaultMinRefreshInterval = 4 * time.Second
+	// Default maximum time between secondary IP fetches
+	DefaultMaxRefreshInterval = 1024 * time.Second
+)
+
 var ErrRefreshSkipped = errors.New("refresh skipped due to throttling")
 
 // InterfaceRetriever is an interface is implemented by the NMAgent Client, and also a mock client for testing.
 type InterfaceRetriever interface {
 	GetInterfaceIPInfo(ctx context.Context) (nmagent.Interfaces, error)
 }
 
-type IPFetcher struct {
-	// Node subnet state
-	secondaryIPQueryInterval   time.Duration // Minimum time between secondary IP fetches
-	secondaryIPLastRefreshTime time.Time     // Time of last secondary IP fetch
+// IPConsumer is an interface implemented by whoever consumes the secondary IPs fetched in nodesubnet
+type IPConsumer interface {
+	UpdateIPsForNodeSubnet(netip.Addr, []netip.Addr) error
+}
 
-	ipFectcherClient InterfaceRetriever
+type IPFetcher struct {
+	// Node subnet config
+	intfFetcherClient  InterfaceRetriever
+	ticker             TickProvider
+	tickerInterval     time.Duration
+	consumer           IPConsumer
+	minRefreshInterval time.Duration
+	maxRefreshInterval time.Duration
 }
 
-func NewIPFetcher(nmaClient InterfaceRetriever, queryInterval time.Duration) *IPFetcher {
+// NewIPFetcher creates a new IPFetcher. If minInterval is 0, it will default to 4 seconds.
+// If maxInterval is 0, it will default to 1024 seconds (or minInterval, if it is higher).
+func NewIPFetcher(
+	client InterfaceRetriever,
+	consumer IPConsumer,
+	minInterval time.Duration,
+	maxInterval time.Duration,
+) *IPFetcher {
+	if minInterval == 0 {
+		minInterval = DefaultMinRefreshInterval
+	}
+
+	if maxInterval == 0 {
+		maxInterval = DefaultMaxRefreshInterval
+	}
+
+	maxInterval = max(maxInterval, minInterval)
+
 	return &IPFetcher{
-		ipFectcherClient:         nmaClient,
-		secondaryIPQueryInterval: queryInterval,
+		intfFetcherClient:  client,
+		consumer:           consumer,
+		minRefreshInterval: minInterval,
+		maxRefreshInterval: maxInterval,
+		tickerInterval:     minInterval,
+	}
+}
+
+func (c *IPFetcher) UpdateFetchIntervalForNoObservedDiff() {
+	c.tickerInterval = min(c.tickerInterval*2, c.maxRefreshInterval) //nolint:gomnd // doubling interval
+
+	if c.ticker != nil {
+		c.ticker.Reset(c.tickerInterval)
 	}
 }
 
-func (c *IPFetcher) RefreshSecondaryIPsIfNeeded(ctx context.Context) (ips []netip.Addr, err error) {
-	// If secondaryIPQueryInterval has elapsed since the last fetch, fetch secondary IPs
-	if time.Since(c.secondaryIPLastRefreshTime) < c.secondaryIPQueryInterval {
-		return nil, ErrRefreshSkipped
+func (c *IPFetcher) UpdateFetchIntervalForObservedDiff() {
+	c.tickerInterval = c.minRefreshInterval
+
+	if c.ticker != nil {
+		c.ticker.Reset(c.tickerInterval)
+	}
+}
+
+func (c *IPFetcher) Start(ctx context.Context) {
+	go func() {
+		// Do an initial fetch
+		err := c.RefreshSecondaryIPs(ctx)
+		if err != nil {
+			log.Printf("Error refreshing secondary IPs: %v", err)
+		}
+
+		if c.ticker == nil {
+			c.ticker = NewTimedTickProvider(c.tickerInterval)
+		}
+
+		defer c.ticker.Stop()
+
+		for {
+			select {
+			case <-c.ticker.C():
+				err := c.RefreshSecondaryIPs(ctx)
+				if err != nil {
+					log.Printf("Error refreshing secondary IPs: %v", err)
+				}
+			case <-ctx.Done():
+				log.Println("IPFetcher stopped")
+				return
+			}
+		}
+	}()
+}
+
+// Fetch IPs from NMAgent and pass to the consumer
+func (c *IPFetcher) RefreshSecondaryIPs(ctx context.Context) error {
+	response, err := c.intfFetcherClient.GetInterfaceIPInfo(ctx)
+	if err != nil {
+		return errors.Wrap(err, "getting interface IPs")
+	}
+
+	if len(response.Entries) == 0 {
+		return errors.New("no interfaces found in response from NMAgent")
 	}
 
-	c.secondaryIPLastRefreshTime = time.Now()
-	response, err := c.ipFectcherClient.GetInterfaceIPInfo(ctx)
+	primaryIP, secondaryIPs := flattenIPListFromResponse(&response)
+	err = c.consumer.UpdateIPsForNodeSubnet(primaryIP, secondaryIPs)
 	if err != nil {
-		return nil, errors.Wrap(err, "getting interface IPs")
+		return errors.Wrap(err, "updating secondary IPs")
 	}
 
-	res := flattenIPListFromResponse(&response)
-	return res, nil
+	return nil
 }
 
 // Get the list of secondary IPs from fetched Interfaces
-func flattenIPListFromResponse(resp *nmagent.Interfaces) (res []netip.Addr) {
+func flattenIPListFromResponse(resp *nmagent.Interfaces) (primary netip.Addr, secondaryIPs []netip.Addr) {
+	var primaryIP netip.Addr
 	// For each interface...
 	for _, intf := range resp.Entries {
 		if !intf.IsPrimary {
@@ -63,15 +147,16 @@ func flattenIPListFromResponse(resp *nmagent.Interfaces) (res []netip.Addr) {
 			for _, a := range s.IPAddress {
 				// Primary addresses are reserved for the host.
 				if a.IsPrimary {
+					primaryIP = netip.Addr(a.Address)
 					continue
 				}
 
-				res = append(res, netip.Addr(a.Address))
+				secondaryIPs = append(secondaryIPs, netip.Addr(a.Address))
 				addressCount++
 			}
 			log.Printf("Got %d addresses from subnet %s", addressCount, s.Prefix)
 		}
 	}
 
-	return res
+	return primaryIP, secondaryIPs
 }

cns/nodesubnet/ip_fetcher_test.go

@@ -2,7 +2,8 @@ package nodesubnet_test
 
 import (
 	"context"
-	"errors"
+	"net/netip"
+	"sync/atomic"
 	"testing"
 	"time"
 
@@ -12,55 +13,117 @@ import (
 
 // Mock client that simply tracks if refresh has been called
 type TestClient struct {
-	fetchCalled bool
+	refreshCount int32
 }
 
 // Mock refresh
 func (c *TestClient) GetInterfaceIPInfo(_ context.Context) (nmagent.Interfaces, error) {
-	c.fetchCalled = true
+	atomic.AddInt32(&c.refreshCount, 1)
 	return nmagent.Interfaces{}, nil
 }
 
-func TestRefreshSecondaryIPsIfNeeded(t *testing.T) {
-	getTests := []struct {
-		name       string
-		shouldCall bool
-		interval   time.Duration
-	}{
-		{
-			"fetch called",
-			true,
-			-1 * time.Second, // Negative timeout to force refresh
-		},
-		{
-			"no refresh needed",
-			false,
-			10 * time.Hour, // High timeout to avoid refresh
-		},
+var _ nodesubnet.InterfaceRetriever = &TestClient{}
+
+// Mock client that simply consumes fetched IPs
+type TestConsumer struct {
+	consumeCount int32
+}
+
+// Mock IP update
+func (c *TestConsumer) UpdateIPsForNodeSubnet(_ netip.Addr, _ []netip.Addr) error {
+	atomic.AddInt32(&c.consumeCount, 1)
+	return nil
+}
+
+var _ nodesubnet.IPConsumer = &TestConsumer{}
+
+// MockTickProvider is a mock implementation of the TickProvider interface
+type MockTickProvider struct {
+	tickChan        chan time.Time
+	currentDuration time.Duration
+}
+
+// NewMockTickProvider creates a new MockTickProvider
+func NewMockTickProvider() *MockTickProvider {
+	return &MockTickProvider{
+		tickChan: make(chan time.Time, 1),
+	}
+}
+
+// C returns the channel on which ticks are delivered
+func (m *MockTickProvider) C() <-chan time.Time {
+	return m.tickChan
+}
+
+// Stop stops the ticker
+func (m *MockTickProvider) Stop() {
+	close(m.tickChan)
+}
+
+// Tick manually sends a tick to the channel
+func (m *MockTickProvider) Tick() {
+	m.tickChan <- time.Now()
+}
+
+func (m *MockTickProvider) Reset(d time.Duration) {
+	m.currentDuration = d
+}
+
+var _ nodesubnet.TickProvider = &MockTickProvider{}
+
+func TestRefresh(t *testing.T) {
+	clientPtr := &TestClient{}
+	consumerPtr := &TestConsumer{}
+	fetcher := nodesubnet.NewIPFetcher(clientPtr, consumerPtr, 0, 0)
+	ticker := NewMockTickProvider()
+	fetcher.SetTicker(ticker)
+	ctx, cancel := testContext(t)
+	defer cancel()
+	fetcher.Start(ctx)
+	ticker.Tick() // Trigger a refresh
+	ticker.Tick() // This tick will be read only after previous refresh is done
+	ticker.Tick() // This call will block until the prevous tick is read
+
+	// At least 2 refreshes - one initial and one after the first tick should be done
+	if atomic.LoadInt32(&clientPtr.refreshCount) < 2 {
+		t.Error("Not enough refreshes")
 	}
 
+	// No consumes, since the responses are empty
+	if atomic.LoadInt32(&consumerPtr.consumeCount) > 0 {
+		t.Error("Consume called unexpectedly, shouldn't be called since responses are empty")
+	}
+}
+
+func TestIntervalUpdate(t *testing.T) {
 	clientPtr := &TestClient{}
-	fetcher := nodesubnet.NewIPFetcher(clientPtr, 0)
-
-	for _, test := range getTests {
-		test := test
-		t.Run(test.name, func(t *testing.T) { // Do not parallelize, as we are using a shared client
-			fetcher.SetSecondaryIPQueryInterval(test.interval)
-			ctx, cancel := testContext(t)
-			defer cancel()
-			clientPtr.fetchCalled = false
-			_, err := fetcher.RefreshSecondaryIPsIfNeeded(ctx)
-
-			if test.shouldCall {
-				if err != nil && errors.Is(err, nodesubnet.ErrRefreshSkipped) {
-					t.Error("refresh expected, but didn't happen")
-				}
-
-				checkErr(t, err, false)
-			} else if err == nil || !errors.Is(err, nodesubnet.ErrRefreshSkipped) {
-				t.Error("refresh not expected, but happened")
-			}
-		})
+	consumerPtr := &TestConsumer{}
+	fetcher := nodesubnet.NewIPFetcher(clientPtr, consumerPtr, 0, 0)
+	interval := fetcher.GetCurrentQueryInterval()
+	ticker := NewMockTickProvider()
+	fetcher.SetTicker(ticker)
+
+	if interval != nodesubnet.DefaultMinRefreshInterval {
+		t.Error("Default min interval not used")
+	}
+
+	for i := 1; i <= 10; i++ {
+		fetcher.UpdateFetchIntervalForNoObservedDiff()
+		exp := interval * 2
+		if interval == nodesubnet.DefaultMaxRefreshInterval {
+			exp = nodesubnet.DefaultMaxRefreshInterval
+		}
+		if fetcher.GetCurrentQueryInterval() != exp || ticker.currentDuration != exp {
+			t.Error("Interval not updated correctly")
+		} else {
+			interval = exp
+		}
+	}
+
+	fetcher.UpdateFetchIntervalForObservedDiff()
+
+	if fetcher.GetCurrentQueryInterval() != nodesubnet.DefaultMinRefreshInterval || ticker.currentDuration != nodesubnet.DefaultMinRefreshInterval {
+		t.Error("Observed diff update incorrect")
 	}
 }
 
@@ -72,15 +135,3 @@ func testContext(t *testing.T) (context.Context, context.CancelFunc) {
 	}
 	return context.WithCancel(context.Background())
 }
-
-// checkErr is an assertion of the presence or absence of an error
-func checkErr(t *testing.T, err error, shouldErr bool) {
-	t.Helper()
-	if err != nil && !shouldErr {
-		t.Fatal("unexpected error: err:", err)
-	}
-
-	if err == nil && shouldErr {
-		t.Fatal("expected error but received none")
-	}
-}