Add fsnotify to watch files in the gke-pod-network directory

By doing this, we will not parse the entire gke-pod-network directory
every 30 seconds. Instead, we will only parse the directory every time
there is a change in the pod IP allocation. For most cases, this will
be more efficient.

go.mod

@@ -6,6 +6,7 @@ require (
 	github.com/beorn7/perks v0.0.0-20180321164747-3a771d992973 // indirect
 	github.com/containernetworking/plugins v0.7.3
 	github.com/coreos/go-iptables v0.4.0
+	github.com/fsnotify/fsnotify v1.5.1
 	github.com/golang/glog v0.0.0-20160126235308-23def4e6c14b
 	github.com/matttproud/golang_protobuf_extensions v1.0.1 // indirect
 	github.com/prometheus/client_golang v0.9.0
@@ -14,7 +15,8 @@ require (
 	github.com/spf13/pflag v1.0.5
 	github.com/vishvananda/netlink v1.1.1-0.20200210145206-a2e8781202af
 	github.com/vishvananda/netns v0.0.0-20191106174202-0a2b9b5464df
-	golang.org/x/sys v0.0.0-20210616094352-59db8d763f22
+	golang.org/x/crypto v0.0.0-20201002170205-7f63de1d35b0 // indirect
+	golang.org/x/sys v0.0.0-20210630005230-0f9fa26af87c
 	k8s.io/api v0.20.0
 	k8s.io/apimachinery v0.20.0
 	k8s.io/client-go v0.20.0

go.sum

@@ -61,6 +61,8 @@ github.com/evanphx/json-patch v4.9.0+incompatible/go.mod h1:50XU6AFN0ol/bzJsmQLi
 github.com/form3tech-oss/jwt-go v3.2.2+incompatible/go.mod h1:pbq4aXjuKjdthFRnoDwaVPLA+WlJuPGy+QneDUgJi2k=
 github.com/fsnotify/fsnotify v1.4.7/go.mod h1:jwhsz4b93w/PPRr/qN1Yymfu8t87LnFCMoQvtojpjFo=
 github.com/fsnotify/fsnotify v1.4.9/go.mod h1:znqG4EE+3YCdAaPaxE2ZRY/06pZUdp0tY4IgpuI1SZQ=
+github.com/fsnotify/fsnotify v1.5.1 h1:mZcQUHVQUQWoPXXtuf9yuEXKudkV2sx1E06UadKWpgI=
+github.com/fsnotify/fsnotify v1.5.1/go.mod h1:T3375wBYaZdLLcVNkcVbzGHY7f1l/uK5T5Ai1i3InKU=
 github.com/ghodss/yaml v0.0.0-20150909031657-73d445a93680/go.mod h1:4dBDuWmgqj2HViK6kFavaiC9ZROes6MMH2rRYeMEF04=
 github.com/go-gl/glfw v0.0.0-20190409004039-e6da0acd62b1/go.mod h1:vR7hzQXu2zJy9AVAgeJqvqgH9Q5CA+iKCZ2gyEVpxRU=
 github.com/go-gl/glfw/v3.3/glfw v0.0.0-20191125211704-12ad95a8df72/go.mod h1:tQ2UAYgL5IevRw8kRxooKSPJfGvJ9fJQFa0TUsXzTg8=
@@ -297,6 +299,8 @@ golang.org/x/sys v0.0.0-20200930185726-fdedc70b468f/go.mod h1:h1NjWce9XRLGQEsW7w
 golang.org/x/sys v0.0.0-20201112073958-5cba982894dd/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs=
 golang.org/x/sys v0.0.0-20210616094352-59db8d763f22 h1:RqytpXGR1iVNX7psjB3ff8y7sNFinVFvkx1c8SjBkio=
 golang.org/x/sys v0.0.0-20210616094352-59db8d763f22/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=
+golang.org/x/sys v0.0.0-20210630005230-0f9fa26af87c h1:F1jZWGFhYfh0Ci55sIpILtKKK8p3i2/krTr0H1rg74I=
+golang.org/x/sys v0.0.0-20210630005230-0f9fa26af87c/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=
 golang.org/x/text v0.0.0-20170915032832-14c0d48ead0c/go.mod h1:NqM8EUOU14njkJ3fqMW+pc6Ldnwhi/IjpwHt7yyuwOQ=
 golang.org/x/text v0.3.0/go.mod h1:NqM8EUOU14njkJ3fqMW+pc6Ldnwhi/IjpwHt7yyuwOQ=
 golang.org/x/text v0.3.1-0.20180807135948-17ff2d5776d2/go.mod h1:NqM8EUOU14njkJ3fqMW+pc6Ldnwhi/IjpwHt7yyuwOQ=

pkg/metrics/collector/pod_ip_metrics.go

@@ -23,18 +23,19 @@ import (
 	"os"
 	"strings"
 
+	"github.com/fsnotify/fsnotify"
 	"github.com/golang/glog"
 	"github.com/prometheus/client_golang/prometheus"
 )
 
 type ipFamily int
 
 const (
-	gkePodNetworkDir = "/host/var/lib/cni/networks/gke-pod-network"
-	ipv4 ipFamily = iota
-	ipv6 = 1
-	dual = 2
-	dualStack = "IPV4_IPV6"
+	gkePodNetworkDir          = "/host/var/lib/cni/networks/gke-pod-network"
+	ipv4             ipFamily = iota
+	ipv6                      = 1
+	dual                      = 2
+	dualStack                 = "IPV4_IPV6"
 )
 
 var (
@@ -63,14 +64,15 @@ var (
 		"Indicates how many pods had more than one IP per family.",
 		nil, nil,
 	)
+	podIpMetricsWatcherSetup = false
 )
 
 type podIpMetricsCollector struct {
-	usedIpv4AddrCount uint64
-	usedIpv6AddrCount uint64
-	dualStackCount uint64
+	usedIpv4AddrCount   uint64
+	usedIpv6AddrCount   uint64
+	dualStackCount      uint64
 	dualStackErrorCount uint64
-	duplicateIpCount uint64
+	duplicateIpCount    uint64
 }
 
 func (ip ipFamily) String() string {
@@ -164,10 +166,57 @@ func (c *podIpMetricsCollector) listIpAddresses(dir string) error {
 	return nil
 }
 
+func (c *podIpMetricsCollector) setupDirectoryWatcher(dir string) error {
+	if err := c.listIpAddresses(dir); err != nil {
+		return err
+	}
+	watcher, err := fsnotify.NewWatcher()
+	if err != nil {
+		glog.Errorf("NewWatcher failed: %v", err)
+		return err
+	}
+
+	go func() {
+		defer func() {
+			watcher.Close()
+			podIpMetricsWatcherSetup = false
+		}()
+
+		for {
+			select {
+			case event, ok := <-watcher.Events:
+				if !ok {
+					glog.Errorf("%s %s is not ok\n", event.Name, event.Op)
+					return
+				}
+				if err := c.listIpAddresses(dir); err != nil {
+					return
+				}
+
+			case err, ok := <-watcher.Errors:
+				glog.Errorf("Received error from watcher %v, ok: %t", err, ok)
+				if !ok {
+					return
+				}
+			}
+		}
+	}()
+
+	err = watcher.Add(dir)
+	if err != nil {
+		glog.Errorf("Failed to add watcher for directory %s: %v", dir, err)
+	}
+	podIpMetricsWatcherSetup = true
+	return nil
+}
+
 func (c *podIpMetricsCollector) Update(ch chan<- prometheus.Metric) error {
-	if err := c.listIpAddresses(gkePodNetworkDir); err != nil {
-		glog.Errorf("ListIpAddresses returned error: %v", err)
-		return nil
+	if !podIpMetricsWatcherSetup {
+		err := c.setupDirectoryWatcher(gkePodNetworkDir)
+		if err != nil {
+			glog.Errorf("setupDirectoryWatcher returned error: %v", err)
+			return nil
+		}
 	}
 	ch <- prometheus.MustNewConstMetric(usedIpv4AddrCountDesc, prometheus.GaugeValue, float64(c.usedIpv4AddrCount))
 	ch <- prometheus.MustNewConstMetric(usedIpv6AddrCountDesc, prometheus.GaugeValue, float64(c.usedIpv6AddrCount))

pkg/metrics/collector/pod_ip_metrics_test.go

@@ -4,6 +4,7 @@ import (
 	"io/ioutil"
 	"os"
 	"testing"
+	"time"
 )
 
 func mustCreateFile(t *testing.T, dir, name, content string) {
@@ -18,6 +19,14 @@ func mustCreateFile(t *testing.T, dir, name, content string) {
 	}
 }
 
+func mustDeleteFile(t *testing.T, dir, name string) {
+	path := dir + "/" + name
+	err := os.Remove(path)
+	if err != nil {
+		t.Fatalf("Error removing file path %s: %v", path, err)
+	}
+}
+
 func mustCreateIpAddrDir(t *testing.T, dir string) string {
 	tempDir, err := ioutil.TempDir("", dir)
 	if err != nil {
@@ -54,53 +63,53 @@ func TestListIpAddresses(t *testing.T) {
 	mustCreateFile(t, dir3, "2600:1900::2", "hash1")
 
 	for _, tc := range []struct {
-		desc string
-		dir string
-		stackType string
-		wantUsedIpv4Count uint64
-		wantUsedIpv6Count uint64
-		wantDualStackCount uint64
+		desc                  string
+		dir                   string
+		stackType             string
+		wantUsedIpv4Count     uint64
+		wantUsedIpv6Count     uint64
+		wantDualStackCount    uint64
 		wantDualStackErrCount uint64
-		wantDuplicateIpCount uint64
-		wantErr bool
-	} {
+		wantDuplicateIpCount  uint64
+		wantErr               bool
+	}{
 		{
-			desc: "bad directory",
-			dir: "bogus",
+			desc:      "bad directory",
+			dir:       "bogus",
 			stackType: "IPV4",
-			wantErr: true,
+			wantErr:   true,
 		},
 		{
-			desc: "dir1 with 2 IPv4 and 2 IPv6 addresses. Single stack",
-			dir: dir1,
-			stackType: "IPV4",
+			desc:              "dir1 with 2 IPv4 and 2 IPv6 addresses. Single stack",
+			dir:               dir1,
+			stackType:         "IPV4",
 			wantUsedIpv4Count: 2,
 			wantUsedIpv6Count: 2,
 		},
 		{
-			desc: "dir1 with 2 IPv4 and 2 IPv6 addresses. dual stack",
-			dir: dir1,
-			stackType: "IPV4_IPV6",
-			wantUsedIpv4Count: 2,
-			wantUsedIpv6Count: 2,
+			desc:               "dir1 with 2 IPv4 and 2 IPv6 addresses. dual stack",
+			dir:                dir1,
+			stackType:          "IPV4_IPV6",
+			wantUsedIpv4Count:  2,
+			wantUsedIpv6Count:  2,
 			wantDualStackCount: 2,
 		},
 		{
-			desc: "dir2 with dual stack errors",
-			dir: dir2,
-			stackType: "IPV4_IPV6",
-			wantUsedIpv4Count: 2,
-			wantUsedIpv6Count: 2,
-			wantDualStackCount: 1,
+			desc:                  "dir2 with dual stack errors",
+			dir:                   dir2,
+			stackType:             "IPV4_IPV6",
+			wantUsedIpv4Count:     2,
+			wantUsedIpv6Count:     2,
+			wantDualStackCount:    1,
 			wantDualStackErrCount: 2,
 		},
 		{
-			desc: "dir2 with dual stack and dup IPs",
-			dir: dir3,
-			stackType: "IPV4_IPV6",
-			wantUsedIpv4Count: 2,
-			wantUsedIpv6Count: 2,
-			wantDualStackCount: 1,
+			desc:                 "dir2 with dual stack and dup IPs",
+			dir:                  dir3,
+			stackType:            "IPV4_IPV6",
+			wantUsedIpv4Count:    2,
+			wantUsedIpv6Count:    2,
+			wantDualStackCount:   1,
 			wantDuplicateIpCount: 2,
 		},
 	} {
@@ -130,3 +139,75 @@ func TestListIpAddresses(t *testing.T) {
 		})
 	}
 }
+
+func TestSetupDirectoryWatcher(t *testing.T) {
+	// dir1 has 2 IPv4 addresses and 2 IPv6 addresses for dual stack pods
+	dir1 := mustCreateIpAddrDir(t, "dir1")
+	mustCreateFile(t, dir1, "10.0.0.1", "hash1")
+	mustCreateFile(t, dir1, "10.0.0.2", "hash2")
+	mustCreateFile(t, dir1, "2600:1900::1", "hash1")
+	mustCreateFile(t, dir1, "2600:1900::2", "hash2")
+	stackType = "IPV4_IPV6"
+
+	// Setup directory watcher and verify metrics
+	mc := podIpMetricsCollector{}
+	if err := mc.setupDirectoryWatcher(dir1); err != nil {
+		t.Fatalf("Got error %v while setting up directory watcher", err)
+	}
+	if mc.usedIpv4AddrCount != 2 {
+		t.Errorf("usedIpv4AddrCount. want: 2, got %d", mc.usedIpv4AddrCount)
+	}
+	if mc.usedIpv6AddrCount != 2 {
+		t.Errorf("usedIpv6AddrCount. want: 2, got %d", mc.usedIpv6AddrCount)
+	}
+	if mc.dualStackCount != 2 {
+		t.Errorf("dualStackCount. want: 2, got %d", mc.dualStackCount)
+	}
+	if mc.dualStackErrorCount != 0 {
+		t.Errorf("dualStackErrorCount. want: 0, got %d", mc.dualStackErrorCount)
+	}
+	if mc.duplicateIpCount != 0 {
+		t.Errorf("duplicateIpCount. want: 0, got %d", mc.duplicateIpCount)
+	}
+	if !podIpMetricsWatcherSetup {
+		t.Errorf("podIpMetricsWatcherSetup is not set to true")
+	}
+
+	//Add a new file to the directory. Verify metrics
+	mustCreateFile(t, dir1, "10.0.0.3", "hash3")
+	time.Sleep(1 * time.Second)
+	if mc.usedIpv4AddrCount != 3 {
+		t.Errorf("usedIpv4AddrCount. want: 3, got %d", mc.usedIpv4AddrCount)
+	}
+	if mc.usedIpv6AddrCount != 2 {
+		t.Errorf("usedIpv6AddrCount. want: 2, got %d", mc.usedIpv6AddrCount)
+	}
+	if mc.dualStackCount != 2 {
+		t.Errorf("dualStackCount. want: 2, got %d", mc.dualStackCount)
+	}
+	if mc.dualStackErrorCount != 1 {
+		t.Errorf("dualStackErrorCount. want: 1, got %d", mc.dualStackErrorCount)
+	}
+	if mc.duplicateIpCount != 0 {
+		t.Errorf("duplicateIpCount. want: 0, got %d", mc.duplicateIpCount)
+	}
+
+	//Remove a file from the directory. Verify metrics
+	mustDeleteFile(t, dir1, "10.0.0.3")
+	time.Sleep(1 * time.Second)
+	if mc.usedIpv4AddrCount != 2 {
+		t.Errorf("usedIpv4AddrCount. want: 2, got %d", mc.usedIpv4AddrCount)
+	}
+	if mc.usedIpv6AddrCount != 2 {
+		t.Errorf("usedIpv6AddrCount. want: 2, got %d", mc.usedIpv6AddrCount)
+	}
+	if mc.dualStackCount != 2 {
+		t.Errorf("dualStackCount. want: 2, got %d", mc.dualStackCount)
+	}
+	if mc.dualStackErrorCount != 0 {
+		t.Errorf("dualStackErrorCount. want: 0, got %d", mc.dualStackErrorCount)
+	}
+	if mc.duplicateIpCount != 0 {
+		t.Errorf("duplicateIpCount. want: 0, got %d", mc.duplicateIpCount)
+	}
+}