do not program status file until endpoint has status "up" (#8692)

Author: aaaaaaaalex

felix/fv/pod_setup_wait_test.go

@@ -131,6 +131,8 @@ var _ = infrastructure.DatastoreDescribe("_BPF-SAFE_ Pod setup status wait", []a
 			By("creating a workload before Felix starts")
 			wl := workload.New(tc.Felixes[0], "workload-endpoint-status-tests-0", "default", "10.65.0.10", "8080", "tcp")
 			wl.ConfigureInInfra(infra)
+			err := wl.Start()
+			Expect(err).NotTo(HaveOccurred(), "Couldn't start a test workload")
 
 			By("determining the filename Felix will look for")
 			wKey, err := names.V3WorkloadEndpointToWorkloadEndpointKey(wl.WorkloadEndpoint)

felix/statusrep/status_file_reporter.go

@@ -32,7 +32,9 @@ import (
 )
 
 const (
-	dirStatus = "endpoint-status"
+	dirStatus  = "endpoint-status"
+	statusUp   = "up"
+	statusDown = "down"
 )
 
 // EndpointStatusFileReporter writes a file to the FS
@@ -55,6 +57,8 @@ type EndpointStatusFileReporter struct {
 	reapplyInterval time.Duration
 
 	hostname string
+
+	filesys
 }
 
 // Backoff wraps a timer-based-retry type which can be stepped.
@@ -68,6 +72,35 @@ type backoffManager struct {
 	newBackoffFunc func() Backoff
 }
 
+type filesys interface {
+	Create(name string) (*os.File, error)
+	Remove(name string) error
+	Mkdir(name string, perm os.FileMode) error
+	ReadDir(name string) ([]os.DirEntry, error)
+}
+
+type defaultFilesys struct{}
+
+// Create wraps os.Create.
+func (fs *defaultFilesys) Create(name string) (*os.File, error) {
+	return os.Create(name)
+}
+
+// Remove wraps os.Remove.
+func (fs *defaultFilesys) Remove(name string) error {
+	return os.Remove(name)
+}
+
+// Mkdir wraps os.Mkdir.
+func (fs *defaultFilesys) Mkdir(name string, perm os.FileMode) error {
+	return os.Mkdir(name, perm)
+}
+
+// ReadDir wraps os.ReadDir.
+func (fs *defaultFilesys) ReadDir(name string) ([]os.DirEntry, error) {
+	return os.ReadDir(name)
+}
+
 // newBackoffManager creates a backoffManager which uses the
 // passed func to create backoffs.
 func newBackoffManager(newBackoffFunc func() Backoff) backoffManager {
@@ -99,6 +132,7 @@ func NewEndpointStatusFileReporter(
 		bom:                     newBackoffManager(newDefaultBackoff),
 		hostname:                "",
 		reapplyInterval:         10 * time.Second,
+		filesys:                 &defaultFilesys{},
 	}
 
 	for _, o := range opts {
@@ -124,6 +158,13 @@ func WithHostname(hostname string) FileReporterOption {
 	}
 }
 
+// WithFilesys allows shimming into filesystem calls.
+func WithFilesys(f filesys) FileReporterOption {
+	return func(fr *EndpointStatusFileReporter) {
+		fr.filesys = f
+	}
+}
+
 // SyncForever blocks until ctx is cancelled.
 // Continuously pulls status-updates from updates C,
 // and reconciles the filesystem with internal state.
@@ -245,7 +286,20 @@ func (fr *EndpointStatusFileReporter) handleEndpointUpdate(e interface{}) {
 		}
 		key := names.WorkloadEndpointIDToWorkloadEndpointKey(m.Id, fr.hostname)
 		fn := names.WorkloadEndpointKeyToStatusFilename(key)
-		fr.statusDirDeltaTracker.Desired().Add(fn)
+
+		if m.Status.Status == statusDown {
+			logrus.WithField("update", e).Debug("Skipping WorkloadEndpointStatusUpdate with down status")
+			fr.statusDirDeltaTracker.Desired().Delete(fn)
+			return
+		} else if m.Status.Status == statusUp {
+			// Explicitly checking the opposite case here (rather than fallthrough)
+			// in-case of a terrible failure where status is neither "up" nor "down".
+			logrus.WithField("update", e).Debug("Handling WorkloadEndpointUpdate with up status")
+			fr.statusDirDeltaTracker.Desired().Add(fn)
+		} else {
+			logrus.WithField("update", e).Warn("Skipping update with unrecognized status")
+		}
+
 	case *proto.WorkloadEndpointStatusRemove:
 		if m.Id == nil {
 			logrus.WithField("update", m).Warn("Couldn't handle nil WorkloadEndpointStatusRemove")
@@ -259,10 +313,11 @@ func (fr *EndpointStatusFileReporter) handleEndpointUpdate(e interface{}) {
 	}
 }
 
+// A sub-call of SyncForever.
 // Overwrites our user-space representation of the kernel with a fresh snapshot.
 func (fr *EndpointStatusFileReporter) resyncDataplaneWithKernel() error {
 	// Load any pre-existing committed dataplane entries.
-	entries, err := ensureStatusDir(fr.endpointStatusDirPrefix)
+	entries, err := fr.ensureStatusDir(fr.endpointStatusDirPrefix)
 	if err != nil {
 		return err
 	}
@@ -310,7 +365,7 @@ func (fr *EndpointStatusFileReporter) writeStatusFile(name string) error {
 	// Write file to dir.
 	logrus.WithField("filename", name).Debug("Writing endpoint-status file to status-dir")
 	filename := filepath.Join(fr.endpointStatusDirPrefix, dirStatus, name)
-	f, err := os.Create(filename)
+	f, err := fr.filesys.Create(filename)
 	if err != nil {
 		return err
 	}
@@ -319,19 +374,19 @@ func (fr *EndpointStatusFileReporter) writeStatusFile(name string) error {
 
 func (fr *EndpointStatusFileReporter) deleteStatusFile(name string) error {
 	filename := filepath.Join(fr.endpointStatusDirPrefix, dirStatus, name)
-	return os.Remove(filename)
+	return fr.filesys.Remove(filename)
 }
 
 // ensureStatusDir ensures there is a directory named "endpoint-status", within
 // the parent dir specified by prefix. Attempts to create the dir if it doesn't exist.
 // Returns all entries within the dir if any exist.
-func ensureStatusDir(prefix string) (entries []fs.DirEntry, err error) {
+func (fr *EndpointStatusFileReporter) ensureStatusDir(prefix string) (entries []fs.DirEntry, err error) {
 	filename := filepath.Join(prefix, dirStatus)
 
-	entries, err = os.ReadDir(filename)
+	entries, err = fr.filesys.ReadDir(filename)
 	if err != nil && errors.Is(err, fs.ErrNotExist) {
 		// Discard ErrNotExist and return the result of attempting to create it.
-		return entries, os.Mkdir(filename, fs.FileMode(0655))
+		return entries, fr.filesys.Mkdir(filename, fs.FileMode(0655))
 	}
 
 	return entries, err

felix/statusrep/status_file_reporter_test.go

@@ -12,17 +12,20 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.
 
-package statusrep_test
+package statusrep
 
 import (
 	"context"
+	"os"
+	"path/filepath"
 	"time"
 
 	. "github.com/onsi/ginkgo"
 	. "github.com/onsi/gomega"
+	"github.com/sirupsen/logrus"
 
 	"github.com/projectcalico/calico/felix/proto"
-	"github.com/projectcalico/calico/felix/statusrep"
+	"github.com/projectcalico/calico/libcalico-go/lib/names"
 )
 
 type mockBackoff struct {
@@ -39,7 +42,43 @@ func (b *mockBackoff) Step() time.Duration {
 	return b.StepDuration
 }
 
+type mockFilesys struct {
+	createCB  func(string)
+	removeCB  func(string)
+	mkdirCB   func(name string)
+	readdirCB func(name string)
+}
+
+func (f *mockFilesys) Create(name string) (*os.File, error) {
+	if f.createCB != nil {
+		f.createCB(name)
+	}
+	return os.Create(name)
+}
+
+func (f *mockFilesys) Remove(name string) error {
+	if f.removeCB != nil {
+		f.removeCB(name)
+	}
+	return os.Remove(name)
+}
+
+func (f *mockFilesys) Mkdir(name string, perm os.FileMode) error {
+	if f.mkdirCB != nil {
+		f.mkdirCB(name)
+	}
+	return os.Mkdir(name, perm)
+}
+
+func (f *mockFilesys) ReadDir(name string) ([]os.DirEntry, error) {
+	if f.readdirCB != nil {
+		f.readdirCB(name)
+	}
+	return os.ReadDir(name)
+}
+
 var _ = Describe("Endpoint Policy Status Reports [file-reporting]", func() {
+	logrus.SetLevel(logrus.DebugLevel)
 	var endpointUpdatesC chan interface{}
 	var ctx context.Context
 	var cancel context.CancelFunc
@@ -56,18 +95,24 @@ var _ = Describe("Endpoint Policy Status Reports [file-reporting]", func() {
 	})
 
 	It("should create a new directory and retry if it fails", func() {
-		var fileReporter *statusrep.EndpointStatusFileReporter
+		var fileReporter *EndpointStatusFileReporter
 
 		backoffCalledC := make(chan struct{})
-		newMockBackoff := func() statusrep.Backoff {
+		newMockBackoff := func() Backoff {
 			return &mockBackoff{
 				C:            backoffCalledC,
-				StepDuration: 1 * time.Nanosecond,
+				StepDuration: 1 * time.Second,
 			}
 		}
+		readdirC := make(chan string, 100)
+		mockfs := mockFilesys{
+			readdirCB: func(name string) {
+				readdirC <- name
+			},
+		}
 
 		// Use a path we think the reporter cannot write to.
-		fileReporter = statusrep.NewEndpointStatusFileReporter(endpointUpdatesC, "/root/", statusrep.WithNewBackoffFunc(newMockBackoff))
+		fileReporter = NewEndpointStatusFileReporter(endpointUpdatesC, "/root/", WithNewBackoffFunc(newMockBackoff), WithFilesys(&mockfs))
 
 		By("Starting a fileReporter which cannot create the necessary directory")
 
@@ -79,6 +124,54 @@ var _ = Describe("Endpoint Policy Status Reports [file-reporting]", func() {
 		}()
 
 		Eventually(endpointUpdatesC, "10s").Should(BeSent(&proto.DataplaneInSync{}))
+
+		Eventually(readdirC, "10s").Should(Receive(Equal("/root/endpoint-status")), "Expected reporter to try reading a forbidden directory")
 		Eventually(backoffCalledC, "10s").Should(BeClosed(), "Backoff wasn't called by the reporter (is the file-reporting unexpectedly succeeding?).")
 	})
+
+	It("should only add a desired file to the delta-tracker when update has status up", func() {
+		fileCreatedC := make(chan string, 100)
+		mockfs := mockFilesys{
+			createCB: func(name string) {
+				fileCreatedC <- name
+			},
+		}
+		reporter := NewEndpointStatusFileReporter(endpointUpdatesC, "/tmp", WithHostname("host"), WithFilesys(&mockfs))
+
+		go func() {
+			defer close(doneC)
+			reporter.SyncForever(ctx)
+		}()
+
+		By("Sending a status-down update to the reporter")
+
+		wepID := &proto.WorkloadEndpointID{
+			OrchestratorId: "abc",
+			WorkloadId:     "default/pod1",
+			EndpointId:     "eth0",
+		}
+		key := names.WorkloadEndpointIDToWorkloadEndpointKey(wepID, "host")
+		mapKey := names.WorkloadEndpointKeyToStatusFilename(key)
+		filename := filepath.Join("/tmp/endpoint-status", mapKey)
+
+		Eventually(endpointUpdatesC, "10s").Should(BeSent(&proto.WorkloadEndpointStatusUpdate{
+			Id: wepID,
+			Status: &proto.EndpointStatus{
+				Status: "down",
+			},
+		}))
+		Eventually(endpointUpdatesC, "10s").Should(BeSent(&proto.DataplaneInSync{}))
+
+		Consistently(fileCreatedC, "3s").ShouldNot(Receive(), "Tracker wrote a file for an endpoint before status was up")
+
+		By("Sending a status-up update to the reporter")
+		Eventually(endpointUpdatesC, "10s").Should(BeSent(&proto.WorkloadEndpointStatusUpdate{
+			Id: wepID,
+			Status: &proto.EndpointStatus{
+				Status: "up",
+			},
+		}))
+
+		Eventually(fileCreatedC, "10s").Should(Receive(Equal(filename)), "Tracker did not add desired file for endpoint with status up")
+	})
 })