run-resourcewatch: Re-add a retry loop for git commands

Author: mdbooth

pkg/resourcewatch/operator/starter.go

@@ -19,8 +19,8 @@ import (
 	"k8s.io/klog/v2"
 )
 
-type observationSource func(ctx context.Context, log logr.Logger, resourceC chan<- *observe.ResourceObservation) chan struct{}
-type observationSink func(<-chan *observe.ResourceObservation) chan struct{}
+type observationSource func(context.Context, logr.Logger, chan<- *observe.ResourceObservation) chan struct{}
+type observationSink func(context.Context, <-chan *observe.ResourceObservation) chan struct{}
 
 // this doesn't appear to handle restarts cleanly.  To do so it would need to compare the resource version that it is applying
 // to the resource version present and it would need to handle unobserved deletions properly.  both are possible, neither is easy.
@@ -88,7 +88,7 @@ func RunResourceWatch(toJsonPath, fromJsonPath string) error {
 	log := klog.FromContext(ctx)
 
 	sourceFinished := source(ctx, log, resourceC)
-	sinkFinished := sink(resourceC)
+	sinkFinished := sink(ctx, resourceC)
 
 	// Wait for the source and sink to finish.
 	select {
@@ -263,16 +263,16 @@ func gitSink() (observationSink, error) {
 		return nil, err
 	}
 
-	return func(resourceC <-chan *observe.ResourceObservation) chan struct{} {
+	return func(ctx context.Context, resourceC <-chan *observe.ResourceObservation) chan struct{} {
 		finished := make(chan struct{})
 		go func() {
 			defer close(finished)
 			for observation := range resourceC {
-				gitWrite(gitStorage, observation)
+				gitWrite(ctx, gitStorage, observation)
 			}
 
 			// We disable GC while we're writing, so run it after we're done.
-			if err := gitStorage.GC(); err != nil {
+			if err := gitStorage.GC(ctx); err != nil {
 				klog.Errorf("Failed to run git gc with error %v", err)
 			}
 		}()
@@ -288,26 +288,26 @@ func gitInitStorage() (*storage.GitStorage, error) {
 	return storage.NewGitStorage(repositoryPath)
 }
 
-func gitWrite(gitStorage *storage.GitStorage, observation *observe.ResourceObservation) {
+func gitWrite(ctx context.Context, gitStorage *storage.GitStorage, observation *observe.ResourceObservation) {
 	gvr := schema.GroupVersionResource{
 		Group:    observation.Group,
 		Version:  observation.Version,
 		Resource: observation.Resource,
 	}
 	switch observation.ObservationType {
 	case observe.ObservationTypeAdd:
-		gitStorage.OnAdd(observation.ObservationTime, gvr, observation.Object)
+		gitStorage.OnAdd(ctx, observation.ObservationTime, gvr, observation.Object)
 	case observe.ObservationTypeUpdate:
-		gitStorage.OnUpdate(observation.ObservationTime, gvr, observation.OldObject, observation.Object)
+		gitStorage.OnUpdate(ctx, observation.ObservationTime, gvr, observation.OldObject, observation.Object)
 	case observe.ObservationTypeDelete:
-		gitStorage.OnDelete(observation.ObservationTime, gvr, observation.Object)
+		gitStorage.OnDelete(ctx, observation.ObservationTime, gvr, observation.Object)
 	}
 }
 
 func jsonSink(file io.WriteCloser) (observationSink, error) {
 	encoder := json.NewEncoder(file)
 
-	return func(resourceC <-chan *observe.ResourceObservation) chan struct{} {
+	return func(ctx context.Context, resourceC <-chan *observe.ResourceObservation) chan struct{} {
 		finished := make(chan struct{})
 		go func() {
 			defer func() {

pkg/resourcewatch/operator/starter_test.go

@@ -44,6 +44,6 @@ func BenchmarkGitSink(b *testing.B) {
 	source(b.Context(), logr.Discard(), resourceC)
 
 	for b.Loop() {
-		gitWrite(gitStorage, <-resourceC)
+		gitWrite(b.Context(), gitStorage, <-resourceC)
 	}
 }

pkg/resourcewatch/storage/git_store.go

@@ -1,6 +1,7 @@
 package storage
 
 import (
+	"context"
 	"fmt"
 	"os"
 	"os/exec"
@@ -17,6 +18,7 @@ import (
 	"k8s.io/apimachinery/pkg/runtime"
 	"k8s.io/apimachinery/pkg/runtime/schema"
 	utilruntime "k8s.io/apimachinery/pkg/util/runtime"
+	"k8s.io/apimachinery/pkg/util/wait"
 	"k8s.io/client-go/tools/cache"
 	"k8s.io/klog/v2"
 	"sigs.k8s.io/yaml"
@@ -56,12 +58,12 @@ func NewGitStorage(path string) (*GitStorage, error) {
 	return storage, nil
 }
 
-func (s *GitStorage) GC() error {
-	return s.execGit("gc")
+func (s *GitStorage) GC(ctx context.Context) error {
+	return s.execGit(ctx, "gc")
 }
 
 // handle handles different operations on git
-func (s *GitStorage) handle(timestamp time.Time, gvr schema.GroupVersionResource, oldObj, obj *unstructured.Unstructured, delete bool) {
+func (s *GitStorage) handle(ctx context.Context, timestamp time.Time, gvr schema.GroupVersionResource, oldObj, obj *unstructured.Unstructured, delete bool) {
 	filePath, content, err := decodeUnstructuredObject(gvr, obj)
 	if err != nil {
 		klog.Warningf("Decoding %q failed: %v", filePath, err)
@@ -89,14 +91,14 @@ func (s *GitStorage) handle(timestamp time.Time, gvr schema.GroupVersionResource
 			// Add it first before removing it.
 			if os.IsNotExist(err) {
 				klog.Info("Observed delete of file we haven't previously observed. Adding it first.")
-				s.handle(timestamp, gvr, nil, obj, false)
-				s.handle(timestamp, gvr, nil, obj, true)
+				s.handle(ctx, timestamp, gvr, nil, obj, false)
+				s.handle(ctx, timestamp, gvr, nil, obj, true)
 				return
 			} else {
 				klog.Errorf("Error removing %q: %v", filePath, err)
 			}
 		}
-		if err := s.commitRemove(timestamp, filePath, "unknown", ocCommand); err != nil {
+		if err := s.commitRemove(ctx, timestamp, filePath, "unknown", ocCommand); err != nil {
 			klog.Error(err)
 		}
 
@@ -119,33 +121,33 @@ func (s *GitStorage) handle(timestamp time.Time, gvr schema.GroupVersionResource
 	switch operation {
 	case gitOpAdded:
 		klog.Infof("Calling commitAdd for %s", filePath)
-		if err := s.commitAdd(timestamp, filePath, modifyingUser, ocCommand); err != nil {
+		if err := s.commitAdd(ctx, timestamp, filePath, modifyingUser, ocCommand); err != nil {
 			klog.Error(err)
 		}
 	case gitOpModified:
 		klog.Infof("Calling commitModify for %s", filePath)
-		if err := s.commitModify(timestamp, filePath, modifyingUser, ocCommand); err != nil {
+		if err := s.commitModify(ctx, timestamp, filePath, modifyingUser, ocCommand); err != nil {
 			klog.Error(err)
 		}
 	default:
 		klog.Errorf("unhandled case for %s: %d", filePath, operation)
 	}
 }
 
-func (s *GitStorage) OnAdd(timestamp time.Time, gvr schema.GroupVersionResource, obj interface{}) {
+func (s *GitStorage) OnAdd(ctx context.Context, timestamp time.Time, gvr schema.GroupVersionResource, obj interface{}) {
 	objUnstructured := obj.(*unstructured.Unstructured)
 
-	s.handle(timestamp, gvr, nil, objUnstructured, false)
+	s.handle(ctx, timestamp, gvr, nil, objUnstructured, false)
 }
 
-func (s *GitStorage) OnUpdate(timestamp time.Time, gvr schema.GroupVersionResource, oldObj, obj interface{}) {
+func (s *GitStorage) OnUpdate(ctx context.Context, timestamp time.Time, gvr schema.GroupVersionResource, oldObj, obj interface{}) {
 	objUnstructured := obj.(*unstructured.Unstructured)
 	oldObjUnstructured := oldObj.(*unstructured.Unstructured)
 
-	s.handle(timestamp, gvr, oldObjUnstructured, objUnstructured, false)
+	s.handle(ctx, timestamp, gvr, oldObjUnstructured, objUnstructured, false)
 }
 
-func (s *GitStorage) OnDelete(timestamp time.Time, gvr schema.GroupVersionResource, obj interface{}) {
+func (s *GitStorage) OnDelete(ctx context.Context, timestamp time.Time, gvr schema.GroupVersionResource, obj interface{}) {
 	objUnstructured, ok := obj.(*unstructured.Unstructured)
 	if !ok {
 		tombstone, ok := obj.(cache.DeletedFinalStateUnknown)
@@ -160,7 +162,7 @@ func (s *GitStorage) OnDelete(timestamp time.Time, gvr schema.GroupVersionResour
 		}
 	}
 
-	s.handle(timestamp, gvr, nil, objUnstructured, true)
+	s.handle(ctx, timestamp, gvr, nil, objUnstructured, true)
 }
 
 // guessAtModifyingUsers tries to figure out who modified the resource
@@ -223,62 +225,67 @@ func resourceFilename(gvr schema.GroupVersionResource, namespace, name string) s
 	return filepath.Join("namespaces", namespace, groupStr, gvr.Resource, name+".yaml")
 }
 
-func (s *GitStorage) execGit(args ...string) error {
+func (s *GitStorage) execGit(ctx context.Context, args ...string) error {
 	// Disable automatic garbage collection to avoid racing with other processes.
 	args = append([]string{"-c", "gc.auto=0"}, args...)
 
-	osCommand := exec.Command("git", args...)
-	osCommand.Dir = s.path
-	output, err := osCommand.CombinedOutput()
-	if err != nil {
-		klog.Errorf("Ran git %v\n%v\n\n", args, string(output))
-		return err
-	}
-	return nil
+	// NOTE(mdbooth): I thought I saw a phantom race since adding the gc.auto=0
+	// flag, but I haven't been able to reproduce it since. This retry loop is
+	// hopefully unnecessary.
+	return wait.PollUntilContextTimeout(ctx, 1*time.Second, 15*time.Second, true, func(ctx context.Context) (bool, error) {
+		osCommand := exec.Command("git", args...)
+		osCommand.Dir = s.path
+		output, err := osCommand.CombinedOutput()
+		if err != nil {
+			klog.Errorf("Ran git %v\n%v\n\n", args, string(output))
+			return false, err
+		}
+		return true, nil
+	})
 }
 
-func (s *GitStorage) commit(timestamp time.Time, path, author, commitMessage string) error {
+func (s *GitStorage) commit(ctx context.Context, timestamp time.Time, path, author, commitMessage string) error {
 	authorString := fmt.Sprintf("%s <[email protected]>", author)
 	dateString := timestamp.Format(time.RFC3339)
 
-	return s.execGit("commit", "--author", authorString, "--date", dateString, "-m", commitMessage)
+	return s.execGit(ctx, "commit", "--author", authorString, "--date", dateString, "-m", commitMessage)
 }
 
-func (s *GitStorage) commitAdd(timestamp time.Time, path, author, ocCommand string) error {
-	if err := s.execGit("add", path); err != nil {
+func (s *GitStorage) commitAdd(ctx context.Context, timestamp time.Time, path, author, ocCommand string) error {
+	if err := s.execGit(ctx, "add", path); err != nil {
 		return err
 	}
 
 	commitMessage := fmt.Sprintf("added %s", ocCommand)
-	if err := s.commit(timestamp, path, author, commitMessage); err != nil {
+	if err := s.commit(ctx, timestamp, path, author, commitMessage); err != nil {
 		return err
 	}
 
 	klog.Infof("Add: %v -- %v added %v", path, author, ocCommand)
 	return nil
 }
 
-func (s *GitStorage) commitModify(timestamp time.Time, path, author, ocCommand string) error {
-	if err := s.execGit("add", path); err != nil {
+func (s *GitStorage) commitModify(ctx context.Context, timestamp time.Time, path, author, ocCommand string) error {
+	if err := s.execGit(ctx, "add", path); err != nil {
 		return err
 	}
 
 	commitMessage := fmt.Sprintf("modifed %s", ocCommand)
-	if err := s.commit(timestamp, path, author, commitMessage); err != nil {
+	if err := s.commit(ctx, timestamp, path, author, commitMessage); err != nil {
 		return err
 	}
 
 	klog.Infof("Modified: %v -- %v updated %v", path, author, ocCommand)
 	return nil
 }
 
-func (s *GitStorage) commitRemove(timestamp time.Time, path, author, ocCommand string) error {
-	if err := s.execGit("rm", path); err != nil {
+func (s *GitStorage) commitRemove(ctx context.Context, timestamp time.Time, path, author, ocCommand string) error {
+	if err := s.execGit(ctx, "rm", path); err != nil {
 		return err
 	}
 
 	commitMessage := fmt.Sprintf("removed %s", ocCommand)
-	if err := s.commit(timestamp, path, author, commitMessage); err != nil {
+	if err := s.commit(ctx, timestamp, path, author, commitMessage); err != nil {
 		return err
 	}