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

run-resourcewatch should no longer race with itself, but this adds
robustness in case another process (like an impatient manual tester) is
running git commands in the same repo.

Author: mdbooth

pkg/resourcewatch/git/git_store.go

@@ -1,6 +1,7 @@
 package git
 
 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,69 @@ 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
+	// The git store should no longer race with itself since we
+	// disabled auto gc so this should never happen in a CI run. However,
+	// manually executed git commands may still cause errors, so we have a retry
+	// loop for robustness.
+	return wait.PollUntilContextCancel(ctx, 1*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))
+			// Don't return the error or we'll stop polling.
+			return false, nil
+		}
+		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
 	}
 

pkg/resourcewatch/git/git_store_test.go

@@ -33,7 +33,7 @@ func BenchmarkGitSink(b *testing.B) {
 
 	b.ResetTimer()
 	for i := range b.N {
-		gitWrite(gitStorage, resources[i])
+		gitWrite(context.TODO(), gitStorage, resources[i])
 	}
 }
 

pkg/resourcewatch/git/sink.go

@@ -1,6 +1,7 @@
 package git
 
 import (
+	"context"
 	"os"
 
 	"github.com/go-logr/logr"
@@ -16,16 +17,16 @@ func Sink(log logr.Logger) (observe.ObservationSink, error) {
 		return nil, err
 	}
 
-	return func(log logr.Logger, resourceC <-chan *observe.ResourceObservation) chan struct{} {
+	return func(ctx context.Context, log logr.Logger, 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 {
 				log.Error(err, "Failed to run git gc")
 			}
 		}()
@@ -41,18 +42,18 @@ func gitInitStorage() (*GitStorage, error) {
 	return NewGitStorage(repositoryPath)
 }
 
-func gitWrite(gitStorage *GitStorage, observation *observe.ResourceObservation) {
+func gitWrite(ctx context.Context, gitStorage *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)
 	}
 }

pkg/resourcewatch/json/json.go

@@ -42,7 +42,7 @@ func Source(file io.ReadCloser) (observe.ObservationSource, error) {
 func Sink(file io.WriteCloser) (observe.ObservationSink, error) {
 	encoder := json.NewEncoder(file)
 
-	return func(log logr.Logger, resourceC <-chan *observe.ResourceObservation) chan struct{} {
+	return func(_ context.Context, log logr.Logger, resourceC <-chan *observe.ResourceObservation) chan struct{} {
 		finished := make(chan struct{})
 		go func() {
 			defer func() {

pkg/resourcewatch/observe/types.go

@@ -32,4 +32,4 @@ type ResourceObservation struct {
 }
 
 type ObservationSource func(ctx context.Context, log logr.Logger, resourceC chan<- *ResourceObservation) chan struct{}
-type ObservationSink func(log logr.Logger, resourceC <-chan *ResourceObservation) chan struct{}
+type ObservationSink func(ctx context.Context, log logr.Logger, resourceC <-chan *ResourceObservation) chan struct{}

pkg/resourcewatch/operator/starter.go

@@ -77,16 +77,18 @@ func RunResourceWatch(toJsonPath, fromJsonPath string) error {
 
 	// Observers emit observations to this channel. We use this channel as a buffer between the observers and the git writer.
 	// Memory consumption will grow if we can't write quickly enough.
-	resourceC := make(chan *observe.ResourceObservation, 1000000)
+	// For reference, the captured test data from a 4.20 installation has ~17000 observations.
+	resourceC := make(chan *observe.ResourceObservation, 10000)
 
 	sourceFinished := source(ctx, log, resourceC)
-	sinkFinished := sink(log, resourceC)
+	sinkFinished := sink(ctx, log, resourceC)
 
 	// Wait for the source and sink to finish.
 	select {
 	case <-sourceFinished:
 		// The source finished. This will also happen if the context is cancelled.
 		close(resourceC)
+		log.Info("Source finished")
 
 		// Wait for the sink to finish writing queued observations.
 		<-sinkFinished