[COP-1575 COP-1573] cache protos and other changes (#2018)

Author: Tofel

book/src/framework/components/chipingresset/chip_ingress.md

@@ -60,17 +60,10 @@ if outErr != nil {
 ctx, cancel := context.WithTimeout(context.Background(), 2*time.Minute)
 defer cancel()
 
-// we recommend to use GITHUB_TOKEN with read access to repositories with protos to avoid heavy rate limiting
-var client *github.Client
-if token := os.Getenv("GITHUB_TOKEN"); token != "" {
-    ts := oauth2.StaticTokenSource(&oauth2.Token{AccessToken: token})
-    tc := oauth2.NewClient(ctx, ts)
-    client = github.NewClient(tc)
-} else {
-    client = github.NewClient(nil)
-}
-
-protoErr := chipingressset.DefaultRegisterAndFetchProtos(ctx, client, []chipingressset.RepoConfiguration{
+protoErr := chipingressset.DefaultRegisterAndFetchProtos(
+    ctx,
+    nil, // GH client will be created dynamically, if needed
+    []chipingressset.RepoConfiguration{
     {
         URI:   "https://github.com/smartcontractkit/chainlink-protostractkit",
         Ref:     "626c42d55bdcb36dffe0077fff58abba40acc3e5",
@@ -123,4 +116,8 @@ for _, schemaSet := range configFiles {
 
 Registration logic is very simple and should handle cases of protos that import other protos as long they are all available in the `ProtoSchemaSet`s provided to the registration function. That function uses an algorithm called "topological sorting by trail", which will try to register all protos in a loop until it cannot register any more protos or it has registered all of them. That allows us to skip dependency parsing completely.
 
-Kafka doesn't have any automatic discoverability mechanism for subject - schema relationship (it has to be provided out-of-band). Currenly, we create the subject in the following way: <subject_prefix>.<package>.<1st-message-name>. Subject prefix is optional and if it's not present, then subject is equal to: <package>.<1st-message-name>. Only the first message in the `.proto` file is ever registered.
+Kafka doesn't have any automatic discoverability mechanism for subject - schema relationship (it has to be provided out-of-band). Currenly, we create the subject in the following way: <subject_prefix>.<package>.<1st-message-name>. Subject prefix is optional and if it's not present, then subject is equal to: <package>.<1st-message-name>. Only the first message in the `.proto` file is ever registered.
+
+## Protobuf caching
+
+Once fetched from `https://github.com` protobuf files will be saved in `.local/share/beholder/protobufs/<OWNER>/<REPOSTIORY>/<SHA>` folder and subsequently used. If saving to cache or reading from it fails, we will load files from the original source.

framework/components/dockercompose/.changeset/v0.1.6.md

@@ -0,0 +1,3 @@
+- Added caching of protobuf files, to avoid constant download each time Beholder starts
+- Removed dynamic port mapping, since external ports are always static
+- Red Panda images will be pulled from Dockerhub instead of docker.redpanda.com to allow by-passing rate limiting by logging into Dockerhub

framework/components/dockercompose/chip_ingress_set/chip_ingress.go

@@ -11,10 +11,8 @@ import (
 
 	networkTypes "github.com/docker/docker/api/types/network"
 	"github.com/docker/docker/client"
-	"github.com/docker/go-connections/nat"
 	"github.com/pkg/errors"
 	"github.com/smartcontractkit/chainlink-testing-framework/framework"
-	"github.com/testcontainers/testcontainers-go"
 	"github.com/testcontainers/testcontainers-go/modules/compose"
 	"github.com/testcontainers/testcontainers-go/wait"
 )
@@ -166,11 +164,6 @@ func New(in *Input) (*Output, error) {
 	if chipIngressExternalHostErr != nil {
 		return nil, errors.Wrap(chipIngressExternalHostErr, "failed to get host for Chip Ingress")
 	}
-	// for some magical reason mapped port sometimes cannot be found, even though we wait for it to be ready, when starting the services
-	chipIngressExternalPort, chipIngressExternalPortErr := findMappdePort(ctx, 31*time.Second, chipIngressContainer, DEFAULT_CHIP_INGRESS_GRPC_PORT)
-	if chipIngressExternalPortErr != nil {
-		return nil, errors.Wrap(chipIngressExternalPortErr, "failed to get mapped port for Chip Ingress")
-	}
 
 	redpandaContainer, redpandaErr := stack.ServiceContainer(ctx, DEFAULT_RED_PANDA_SERVICE_NAME)
 	if redpandaErr != nil {
@@ -181,14 +174,6 @@ func New(in *Input) (*Output, error) {
 	if redpandaExternalHostErr != nil {
 		return nil, errors.Wrap(redpandaExternalHostErr, "failed to get host for Red Panda")
 	}
-	redpandaExternalKafkaPort, redpandaExternalKafkaPortErr := findMappdePort(ctx, 31*time.Second, redpandaContainer, DEFAULT_RED_PANDA_KAFKA_PORT)
-	if redpandaExternalKafkaPortErr != nil {
-		return nil, errors.Wrap(redpandaExternalKafkaPortErr, "failed to get mapped port for Red Panda")
-	}
-	redpandaExternalSchemaRegistryPort, redpandaExternalSchemaRegistryPortErr := findMappdePort(ctx, 31*time.Second, redpandaContainer, DEFAULT_RED_PANDA_SCHEMA_REGISTRY_PORT)
-	if redpandaExternalSchemaRegistryPortErr != nil {
-		return nil, errors.Wrap(redpandaExternalSchemaRegistryPortErr, "failed to get mapped port for Red Panda")
-	}
 
 	redpandaConsoleContainer, redpandaConsoleErr := stack.ServiceContainer(ctx, DEFAULT_RED_PANDA_CONSOLE_SERVICE_NAME)
 	if redpandaConsoleErr != nil {
@@ -198,22 +183,18 @@ func New(in *Input) (*Output, error) {
 	if redpandaExternalConsoleHostErr != nil {
 		return nil, errors.Wrap(redpandaExternalConsoleHostErr, "failed to get host for Red Panda Console")
 	}
-	redpandaExternalConsolePort, redpandaExternalConsolePortErr := findMappdePort(ctx, 31*time.Second, redpandaConsoleContainer, DEFAULT_RED_PANDA_CONSOLE_PORT)
-	if redpandaExternalConsolePortErr != nil {
-		return nil, errors.Wrap(redpandaExternalConsolePortErr, "failed to get mapped port for Red Panda Console")
-	}
 
 	output := &Output{
 		ChipIngress: &ChipIngressOutput{
 			GRPCInternalURL: fmt.Sprintf("http://%s:%s", DEFAULT_CHIP_INGRESS_SERVICE_NAME, DEFAULT_CHIP_INGRESS_GRPC_PORT),
-			GRPCExternalURL: fmt.Sprintf("http://%s:%s", chipIngressExternalHost, chipIngressExternalPort.Port()),
+			GRPCExternalURL: fmt.Sprintf("http://%s:%s", chipIngressExternalHost, DEFAULT_CHIP_INGRESS_GRPC_PORT),
 		},
 		RedPanda: &RedPandaOutput{
 			SchemaRegistryInternalURL: fmt.Sprintf("http://%s:%s", DEFAULT_RED_PANDA_SERVICE_NAME, DEFAULT_RED_PANDA_SCHEMA_REGISTRY_PORT),
-			SchemaRegistryExternalURL: fmt.Sprintf("http://%s:%s", redpandaExternalHost, redpandaExternalSchemaRegistryPort.Port()),
+			SchemaRegistryExternalURL: fmt.Sprintf("http://%s:%s", redpandaExternalHost, DEFAULT_RED_PANDA_SCHEMA_REGISTRY_PORT),
 			KafkaInternalURL:          fmt.Sprintf("%s:%s", DEFAULT_RED_PANDA_SERVICE_NAME, DEFAULT_RED_PANDA_KAFKA_PORT),
-			KafkaExternalURL:          fmt.Sprintf("%s:%s", redpandaExternalHost, redpandaExternalKafkaPort.Port()),
-			ConsoleExternalURL:        fmt.Sprintf("http://%s:%s", redpandaExternalConsoleHost, redpandaExternalConsolePort.Port()),
+			KafkaExternalURL:          fmt.Sprintf("%s:%s", redpandaExternalHost, DEFAULT_RED_PANDA_KAFKA_PORT),
+			ConsoleExternalURL:        fmt.Sprintf("http://%s:%s", redpandaExternalConsoleHost, DEFAULT_RED_PANDA_CONSOLE_PORT),
 		},
 	}
 
@@ -252,33 +233,6 @@ func composeFilePath(rawFilePath string) (string, error) {
 	return tempFile.Name(), nil
 }
 
-func findMappdePort(ctx context.Context, timeout time.Duration, container *testcontainers.DockerContainer, port nat.Port) (nat.Port, error) {
-	forCtx, cancel := context.WithTimeout(ctx, timeout)
-	defer cancel()
-
-	tickerInterval := 5 * time.Second
-	ticker := time.NewTicker(5 * time.Second)
-	defer ticker.Stop()
-
-	for {
-		select {
-		case <-forCtx.Done():
-			return "", fmt.Errorf("timeout while waiting for mapped port for %s", port)
-		case <-ticker.C:
-			portCtx, portCancel := context.WithTimeout(ctx, tickerInterval)
-			defer portCancel()
-			mappedPort, mappedPortErr := container.MappedPort(portCtx, port)
-			if mappedPortErr != nil {
-				return "", errors.Wrapf(mappedPortErr, "failed to get mapped port for %s", port)
-			}
-			if mappedPort.Port() == "" {
-				return "", fmt.Errorf("mapped port for %s is empty", port)
-			}
-			return mappedPort, nil
-		}
-	}
-}
-
 func connectNetwork(connCtx context.Context, timeout time.Duration, dockerClient *client.Client, containerID, networkName, stackIdentifier string) error {
 	ticker := time.NewTicker(500 * time.Millisecond)
 	defer ticker.Stop()

framework/components/dockercompose/chip_ingress_set/docker-compose.yml

@@ -43,8 +43,10 @@ services:
       retries: 10
 
   redpanda-0:
-    # using a specific version of the redpanda image to exclude potentially breaking changes
-    image: docker.redpanda.com/redpandadata/redpanda:v24.3.16
+    # modified from the original:
+    # - pin point a specific version of the redpanda image to exclude potentially breaking changes
+    # - pull image from dockerhub instead of docker.redpanda.com
+    image: redpandadata/redpanda:v24.3.16
     container_name: redpanda-0
     hostname: redpanda-0
     command:
@@ -74,9 +76,12 @@ services:
       start_period: 1s
 
   redpanda-console:
-    # using a specific version of the console image to ensure compatibility, because v3.x.x uses incompatible configuration format
+
     container_name: redpanda-console
-    image: docker.redpanda.com/redpandadata/console:v2.8.6
+    # modified from the original:
+    # - pin point a specific version of the console image to ensure compatibility, because v3.x.x uses incompatible configuration format
+    # - pull image from dockerhub instead of docker.redpanda.com
+    image: redpandadata/console:v2.8.6
     entrypoint: /bin/sh
     command: -c 'echo "$$CONSOLE_CONFIG_FILE" > /tmp/config.yml; /app/console'
     environment:

framework/components/dockercompose/chip_ingress_set/protos.go

@@ -15,6 +15,7 @@ import (
 	"github.com/google/go-github/v72/github"
 	"github.com/pkg/errors"
 	"github.com/smartcontractkit/chainlink-testing-framework/framework"
+	"golang.org/x/oauth2"
 )
 
 type protoFile struct {
@@ -83,8 +84,27 @@ func RegisterAndFetchProtos(ctx context.Context, client *github.Client, protoSch
 		}
 	}
 
+	ghClientFn := func() *github.Client {
+		if client != nil {
+			return client
+		}
+
+		var client *github.Client
+
+		if token := os.Getenv("GITHUB_TOKEN"); token != "" {
+			ts := oauth2.StaticTokenSource(&oauth2.Token{AccessToken: token})
+			tc := oauth2.NewClient(ctx, ts)
+			client = github.NewClient(tc)
+		} else {
+			framework.L.Warn().Msg("GITHUB_TOKEN is not set, using unauthenticated GitHub client. This may cause rate limiting issues when downloading proto files")
+			client = github.NewClient(nil)
+		}
+
+		return client
+	}
+
 	for _, protoSchemaSet := range protoSchemaSets {
-		protos, protosErr := fetchProtoFilesInFolders(ctx, client, protoSchemaSet.URI, protoSchemaSet.Ref, protoSchemaSet.Folders)
+		protos, protosErr := fetchProtoFilesInFolders(ctx, ghClientFn, protoSchemaSet.URI, protoSchemaSet.Ref, protoSchemaSet.Folders)
 		if protosErr != nil {
 			return errors.Wrapf(protosErr, "failed to fetch protos from %s", protoSchemaSet.URI)
 		}
@@ -167,7 +187,7 @@ func extractTopLevelMessageNamesWithRegex(protoSrc string) ([]string, error) {
 }
 
 // Fetches .proto files from a GitHub repo optionally scoped to specific folders. It is recommended to use `*github.Client` with auth token to avoid rate limiting.
-func fetchProtoFilesInFolders(ctx context.Context, client *github.Client, uri, ref string, folders []string) ([]protoFile, error) {
+func fetchProtoFilesInFolders(ctx context.Context, clientFn func() *github.Client, uri, ref string, folders []string) ([]protoFile, error) {
 	framework.L.Debug().Msgf("Fetching proto files from %s in folders: %s", uri, strings.Join(folders, ", "))
 
 	if strings.HasPrefix(uri, "file://") {
@@ -176,14 +196,20 @@ func fetchProtoFilesInFolders(ctx context.Context, client *github.Client, uri, r
 
 	parts := strings.Split(strings.TrimPrefix(uri, "https://"), "/")
 
-	return fetchProtosFromGithub(ctx, client, parts[1], parts[2], ref, folders)
+	return fetchProtosFromGithub(ctx, clientFn, parts[1], parts[2], ref, folders)
 }
 
-func fetchProtosFromGithub(ctx context.Context, client *github.Client, owner, repository, ref string, folders []string) ([]protoFile, error) {
-	if client == nil {
-		return nil, errors.New("github client cannot be nil")
+func fetchProtosFromGithub(ctx context.Context, clientFn func() *github.Client, owner, repository, ref string, folders []string) ([]protoFile, error) {
+	cachedFiles, found, cacheErr := loadCachedProtoFiles(owner, repository, ref, folders)
+	if cacheErr != nil {
+		framework.L.Warn().Msgf("Failed to load cached proto files for %s/%s at ref %s: %v", owner, repository, ref, cacheErr)
+	}
+	if cacheErr == nil && found {
+		framework.L.Debug().Msgf("Using cached proto files for %s/%s at ref %s", owner, repository, ref)
+		return cachedFiles, nil
 	}
 
+	client := clientFn()
 	var files []protoFile
 
 	sha, shaErr := resolveRefSHA(ctx, client, owner, repository, ref)
@@ -255,9 +281,61 @@ searchLoop:
 		return nil, fmt.Errorf("no proto files found in %s/%s in folders %s", owner, repository, strings.Join(folders, ", "))
 	}
 
+	saveErr := saveProtoFilesToCache(owner, repository, ref, files)
+	if saveErr != nil {
+		framework.L.Warn().Msgf("Failed to save proto files to cache for %s/%s at ref %s: %v", owner, repository, ref, saveErr)
+	}
+
 	return files, nil
 }
 
+func loadCachedProtoFiles(owner, repository, ref string, _ []string) ([]protoFile, bool, error) {
+	cachePath, cacheErr := cacheFilePath(owner, repository, ref)
+	if cacheErr != nil {
+		return nil, false, errors.Wrapf(cacheErr, "failed to get cache file path for %s/%s at ref %s", owner, repository, ref)
+	}
+
+	if _, err := os.Stat(cachePath); os.IsNotExist(err) {
+		return nil, false, nil // cache not found
+	}
+
+	cachedFiles, cachedErr := fetchProtosFromFilesystem("file://"+cachePath, []string{}) // ignore folders since, we already filtered them when fetching from GitHub
+	if cachedErr != nil {
+		return nil, false, errors.Wrapf(cachedErr, "failed to load cached proto files from %s", cachePath)
+	}
+
+	return cachedFiles, true, nil
+}
+
+func saveProtoFilesToCache(owner, repository, ref string, files []protoFile) error {
+	cachePath, cacheErr := cacheFilePath(owner, repository, ref)
+	if cacheErr != nil {
+		return errors.Wrapf(cacheErr, "failed to get cache file path for %s/%s at ref %s", owner, repository, ref)
+	}
+
+	for _, file := range files {
+		path := filepath.Join(cachePath, file.Path)
+		if err := os.MkdirAll(filepath.Dir(path), 0755); err != nil {
+			return errors.Wrapf(err, "failed to create directory for cache file %s", cachePath)
+		}
+		if writeErr := os.WriteFile(path, []byte(file.Content), 0755); writeErr != nil {
+			return errors.Wrapf(writeErr, "failed to write cached proto files to %s", cachePath)
+		}
+	}
+
+	framework.L.Debug().Msgf("Saved %d proto files to cache at %s", len(files), cachePath)
+
+	return nil
+}
+
+func cacheFilePath(owner, repository, ref string) (string, error) {
+	homeDir, homeErr := os.UserHomeDir()
+	if homeErr != nil {
+		return "", errors.Wrap(homeErr, "failed to get user home directory")
+	}
+	return filepath.Join(homeDir, ".local", "share", "beholder", "protobufs", owner, repository, ref), nil
+}
+
 func fetchProtosFromFilesystem(uri string, folders []string) ([]protoFile, error) {
 	var files []protoFile
 

framework/components/dockercompose/go.mod

@@ -7,12 +7,12 @@ replace github.com/smartcontractkit/chainlink-testing-framework/framework => ../
 require (
 	github.com/confluentinc/confluent-kafka-go v1.9.2
 	github.com/docker/docker v28.0.4+incompatible
-	github.com/docker/go-connections v0.5.0
 	github.com/google/go-github/v72 v72.0.0
 	github.com/pkg/errors v0.9.1
 	github.com/smartcontractkit/chainlink-testing-framework/framework v0.0.0-00010101000000-000000000000
 	github.com/testcontainers/testcontainers-go v0.37.0
 	github.com/testcontainers/testcontainers-go/modules/compose v0.37.0
+	golang.org/x/oauth2 v0.25.0
 )
 
 require (
@@ -65,6 +65,7 @@ require (
 	github.com/docker/distribution v2.8.3+incompatible // indirect
 	github.com/docker/docker-credential-helpers v0.8.2 // indirect
 	github.com/docker/go v1.5.1-1.0.20160303222718-d30aec9fd63c // indirect
+	github.com/docker/go-connections v0.5.0 // indirect
 	github.com/docker/go-metrics v0.0.1 // indirect
 	github.com/docker/go-units v0.5.0 // indirect
 	github.com/ebitengine/purego v0.8.2 // indirect
@@ -201,7 +202,6 @@ require (
 	golang.org/x/crypto v0.37.0 // indirect
 	golang.org/x/exp v0.0.0-20241108190413-2d47ceb2692f // indirect
 	golang.org/x/net v0.38.0 // indirect
-	golang.org/x/oauth2 v0.25.0 // indirect
 	golang.org/x/sync v0.13.0 // indirect
 	golang.org/x/sys v0.32.0 // indirect
 	golang.org/x/term v0.31.0 // indirect

framework/examples/myproject/go.mod

@@ -15,7 +15,6 @@ require (
 	github.com/blocto/solana-go-sdk v1.30.0
 	github.com/ethereum/go-ethereum v1.15.0
 	github.com/go-resty/resty/v2 v2.16.5
-	github.com/google/go-github/v72 v72.0.0
 	github.com/smartcontractkit/chainlink-testing-framework/framework v0.8.9
 	github.com/smartcontractkit/chainlink-testing-framework/framework/components/dockercompose v0.0.0-00010101000000-000000000000
 	github.com/smartcontractkit/chainlink-testing-framework/framework/components/fake v0.0.0-20250707095700-c7855f06ddd1
@@ -58,6 +57,7 @@ require (
 	github.com/go-ini/ini v1.67.0 // indirect
 	github.com/go-viper/mapstructure/v2 v2.2.1 // indirect
 	github.com/gofrs/flock v0.12.1 // indirect
+	github.com/google/go-github/v72 v72.0.0 // indirect
 	github.com/google/go-querystring v1.1.0 // indirect
 	github.com/google/shlex v0.0.0-20191202100458-e7afc7fbc510 // indirect
 	github.com/grpc-ecosystem/grpc-gateway/v2 v2.26.3 // indirect
@@ -382,7 +382,7 @@ require (
 	golang.org/x/exp v0.0.0-20250106191152-7588d65b2ba8 // indirect
 	golang.org/x/mod v0.22.0 // indirect
 	golang.org/x/net v0.40.0 // indirect
-	golang.org/x/oauth2 v0.27.0
+	golang.org/x/oauth2 v0.27.0 // indirect
 	golang.org/x/sync v0.14.0 // indirect
 	golang.org/x/sys v0.33.0 // indirect
 	golang.org/x/term v0.32.0 // indirect

framework/examples/myproject/smoke_chip_ingress_test.go

@@ -6,11 +6,9 @@ import (
 	"testing"
 	"time"
 
-	"github.com/google/go-github/v72/github"
 	"github.com/smartcontractkit/chainlink-testing-framework/framework"
 	chipingressset "github.com/smartcontractkit/chainlink-testing-framework/framework/components/dockercompose/chip_ingress_set"
 	"github.com/stretchr/testify/require"
-	"golang.org/x/oauth2"
 )
 
 type ChipConfig struct {
@@ -33,19 +31,10 @@ func TestChipIngressSmoke(t *testing.T) {
 		ctx, cancel := context.WithTimeout(context.Background(), 2*time.Minute)
 		defer cancel()
 
-		var client *github.Client
-		if token := os.Getenv("GITHUB_TOKEN"); token != "" {
-			ts := oauth2.StaticTokenSource(&oauth2.Token{AccessToken: token})
-			tc := oauth2.NewClient(ctx, ts)
-			client = github.NewClient(tc)
-		} else {
-			client = github.NewClient(nil)
-		}
-
 		createTopicsErr := chipingressset.CreateTopics(ctx, out.RedPanda.KafkaExternalURL, []string{"cre"})
 		require.NoError(t, createTopicsErr, "failed to create topics")
 
-		err := chipingressset.DefaultRegisterAndFetchProtos(ctx, client, []chipingressset.ProtoSchemaSet{
+		err := chipingressset.DefaultRegisterAndFetchProtos(ctx, nil, []chipingressset.ProtoSchemaSet{
 			{
 				URI:           "https://github.com/smartcontractkit/chainlink-protos",
 				Ref:           "95decc005a91a1fd2621af9d9f00cb36d8061067",