run-resourcewatch: Add a gitstorage benchmark

Author: mdbooth

go.mod

@@ -1,8 +1,6 @@
 module github.com/openshift/origin
 
-go 1.23.0
-
-toolchain go1.23.4
+go 1.24.4
 
 require (
 	github.com/Azure/azure-sdk-for-go/sdk/azcore v1.16.0
@@ -21,6 +19,7 @@ require (
 	github.com/ghodss/yaml v1.0.1-0.20190212211648-25d852aebe32
 	github.com/go-bindata/go-bindata v3.1.2+incompatible
 	github.com/go-ldap/ldap/v3 v3.4.3
+	github.com/go-logr/logr v1.4.2
 	github.com/golang/protobuf v1.5.4
 	github.com/google/gnostic-models v0.6.8
 	github.com/google/go-cmp v0.6.0
@@ -171,7 +170,6 @@ require (
 	github.com/go-errors/errors v1.4.2 // indirect
 	github.com/go-fonts/liberation v0.3.1 // indirect
 	github.com/go-latex/latex v0.0.0-20230307184459-12ec69307ad9 // indirect
-	github.com/go-logr/logr v1.4.2 // indirect
 	github.com/go-logr/stdr v1.2.2 // indirect
 	github.com/go-openapi/jsonpointer v0.21.0 // indirect
 	github.com/go-openapi/jsonreference v0.21.0 // indirect

pkg/resourcewatch/operator/starter.go

@@ -3,6 +3,8 @@ package operator
 import (
 	"context"
 	"encoding/json"
+	"fmt"
+	"io"
 	"os"
 	"os/signal"
 	"sync"
@@ -45,8 +47,12 @@ func RunResourceWatch(toJsonPath, fromJsonPath string) error {
 	var sink observationSink
 
 	if fromJsonPath != "" {
-		var err error
-		source, err = jsonSource(fromJsonPath)
+		file, err := os.Open(fromJsonPath)
+		if err != nil {
+			return fmt.Errorf("failed to open json file %q: %w", fromJsonPath, err)
+		}
+
+		source, err = jsonSource(file)
 		if err != nil {
 			return err
 		}
@@ -59,8 +65,12 @@ func RunResourceWatch(toJsonPath, fromJsonPath string) error {
 	}
 
 	if toJsonPath != "" {
-		var err error
-		sink, err = jsonSink(toJsonPath)
+		file, err := os.OpenFile(toJsonPath, os.O_CREATE|os.O_EXCL|os.O_WRONLY|os.O_APPEND, 0664)
+		if err != nil {
+			return fmt.Errorf("failed to create json file %q: %w", toJsonPath, err)
+		}
+
+		sink, err = jsonSink(file)
 		if err != nil {
 			return err
 		}
@@ -217,13 +227,7 @@ func clusterSource() (observationSource, error) {
 	}, nil
 }
 
-func jsonSource(fromJsonPath string) (observationSource, error) {
-	file, err := os.Open(fromJsonPath)
-	if err != nil {
-		klog.Errorf("Failed to open json file with error %v", err)
-		return nil, err
-	}
-
+func jsonSource(file io.ReadCloser) (observationSource, error) {
 	decoder := json.NewDecoder(file)
 
 	return func(ctx context.Context, log logr.Logger, resourceC chan<- *observe.ResourceObservation) chan struct{} {
@@ -253,12 +257,7 @@ func jsonSource(fromJsonPath string) (observationSource, error) {
 }
 
 func gitSink() (observationSink, error) {
-	repositoryPath := "/repository"
-	if repositoryPathEnv := os.Getenv("REPOSITORY_PATH"); len(repositoryPathEnv) > 0 {
-		repositoryPath = repositoryPathEnv
-	}
-
-	gitStorage, err := storage.NewGitStorage(repositoryPath)
+	gitStorage, err := gitInitStorage()
 	if err != nil {
 		klog.Errorf("Failed to create git storage with error %v", err)
 		return nil, err
@@ -269,32 +268,38 @@ func gitSink() (observationSink, error) {
 		go func() {
 			defer close(finished)
 			for observation := range resourceC {
-				gvr := schema.GroupVersionResource{
-					Group:    observation.Group,
-					Version:  observation.Version,
-					Resource: observation.Resource,
-				}
-				switch observation.ObservationType {
-				case observe.ObservationTypeAdd:
-					gitStorage.OnAdd(gvr, observation.Object)
-				case observe.ObservationTypeUpdate:
-					gitStorage.OnUpdate(gvr, observation.OldObject, observation.Object)
-				case observe.ObservationTypeDelete:
-					gitStorage.OnDelete(gvr, observation.Object)
-				}
+				gitWrite(gitStorage, observation)
 			}
 		}()
 		return finished
 	}, nil
 }
 
-func jsonSink(toJsonPath string) (observationSink, error) {
-	file, err := os.OpenFile(toJsonPath, os.O_CREATE|os.O_WRONLY|os.O_APPEND, 0664)
-	if err != nil {
-		klog.Errorf("Failed to create json file with error %v", err)
-		return nil, err
+func gitInitStorage() (*storage.GitStorage, error) {
+	repositoryPath := "/repository"
+	if repositoryPathEnv := os.Getenv("REPOSITORY_PATH"); len(repositoryPathEnv) > 0 {
+		repositoryPath = repositoryPathEnv
 	}
+	return storage.NewGitStorage(repositoryPath)
+}
+
+func gitWrite(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(gvr, observation.Object)
+	case observe.ObservationTypeUpdate:
+		gitStorage.OnUpdate(gvr, observation.OldObject, observation.Object)
+	case observe.ObservationTypeDelete:
+		gitStorage.OnDelete(gvr, observation.Object)
+	}
+}
 
+func jsonSink(file io.WriteCloser) (observationSink, error) {
 	encoder := json.NewEncoder(file)
 
 	return func(resourceC <-chan *observe.ResourceObservation) chan struct{} {

pkg/resourcewatch/operator/starter_test.go

@@ -0,0 +1,54 @@
+package operator
+
+import (
+	"compress/gzip"
+	"os"
+	"testing"
+	"time"
+
+	"github.com/go-logr/logr"
+	"github.com/openshift/origin/pkg/resourcewatch/observe"
+)
+
+func BenchmarkGitSink(b *testing.B) {
+	repoDir := b.TempDir()
+	os.Setenv("REPOSITORY_PATH", repoDir)
+
+	// Because we have no .gitconfig
+	os.Setenv("GIT_COMMITTER_NAME", "run-resourcewatch")
+	os.Setenv("GIT_COMMITTER_EMAIL", "[email protected]")
+
+	// Don't use git configuration from the user's home directory
+	os.Setenv("HOME", "")
+
+	gitStorage, err := gitInitStorage()
+	if err != nil {
+		b.Fatalf("Failed to initialise git storage: %v", err)
+	}
+
+	file, err := os.Open("testdata/observations.json.gz")
+	if err != nil {
+		b.Fatalf("Failed to open test data: %v", err)
+	}
+
+	uncompressed, err := gzip.NewReader(file)
+	if err != nil {
+		b.Fatalf("Failed to decompress test data: %v", err)
+	}
+
+	source, err := jsonSource(uncompressed)
+	if err != nil {
+		b.Fatalf("Failed to initialise json source: %v", err)
+	}
+
+	resourceC := make(chan *observe.ResourceObservation, 1000000)
+	source(b.Context(), logr.Discard(), resourceC)
+
+	for b.Loop() {
+		gitWrite(gitStorage, <-resourceC)
+	}
+
+	// gitStorage creates unmanaged go threads which can prevent us from cleaning up.
+	// Until we address that, just give them a chance to finish
+	time.Sleep(5 * time.Second)
+}