[DX-1407] Added support for loading protobufs from local filesystem (#1984)

Author: Tofel

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

@@ -12,7 +12,10 @@ It consists of 3 components:
 To add it to your stack use following TOML:
 ```toml
 [chip_ingress]
-  compose_file='../../components/chip_ingress_set/docker-compose.yml'
+  # using a local docker-compose file
+  compose_file='file://../../components/chip_ingress_set/docker-compose.yml'
+  # using a remote file
+  # compose_file='https://my.awesome.resource.io/docker-compose.yml'
   extra_docker_networks = ["my-existing-network"]
 ```
 
@@ -69,8 +72,7 @@ if token := os.Getenv("GITHUB_TOKEN"); token != "" {
 
 protoErr := chipingressset.DefaultRegisterAndFetchProtos(ctx, client, []chipingressset.RepoConfiguration{
     {
-        Owner:   "smartcontractkit",
-        Repo:    "chainlink-protos",
+        URI:   "https://github.com/smartcontractkit/chainlink-protostractkit",
         Ref:     "626c42d55bdcb36dffe0077fff58abba40acc3e5",
         Folders: []string{"workflows"},
     },
@@ -83,13 +85,21 @@ if protoErr != nil {
 Since `ProtoSchemaSet` has TOML tags you can also read it from a TOML file with this content:
 ```toml
 [[proto_schema_set]]
-owner = 'smartcontractkit'
-repository = 'chainlink-protos'
+# reading from remote registry (only github.com supported)
+uri = 'https://github.com/smartcontractkit/chainlink-protos'
 ref = '626c42d55bdcb36dffe0077fff58abba40acc3e5'
 folders = ['workflows']
+subject_prefix = 'cre-'
+
+[[proto_schema_set]]
+# reading from local folder
+uri = 'file://../../chainlink-protos'
+# ref is not supported, when reading from local folders
+folders = ['workflows']
+subject_prefix = 'cre-'
 ```
 
-using this code:
+And then use this Go code to register them:
 ```go
 var protoSchemaSets []chipingressset.ProtoSchemaSet
 for _, schemaSet := range configFiles {
@@ -113,4 +123,4 @@ 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.
 
-Since Kafka doesn't have any automatic discoverability mechanism for subject - schema relationship (it has to be provided out-of-band) code currently only knows how to correctly register protos from [chainlink-protos](https://github.com/smartcontractkit/chainlink-protos) repository.
+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.

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

@@ -0,0 +1,2 @@
+- Added support for loading protobufs from local filesystem
+- Require 'file://' prefix when using local docker-compose file

framework/components/dockercompose/chip_ingress_set/chip_ingress.go

@@ -217,15 +217,18 @@ func New(in *Input) (*Output, error) {
 		},
 	}
 
-	framework.L.Info().Msg("Chip Ingress stack start")
+	framework.L.Info().Msg("Chip Ingress stack started")
 
 	return output, nil
 }
 
 func composeFilePath(rawFilePath string) (string, error) {
 	// if it's not a URL, return it as is and assume it's a local file
 	if !strings.HasPrefix(rawFilePath, "http") {
-		return rawFilePath, nil
+		if !strings.HasPrefix(rawFilePath, "file://") {
+			return "", fmt.Errorf("docker compose URI must start either with 'file://', 'http://' or 'https://', but '%s' was found", rawFilePath)
+		}
+		return strings.TrimPrefix(rawFilePath, "file://"), nil
 	}
 
 	resp, respErr := http.Get(rawFilePath)

framework/components/dockercompose/chip_ingress_set/protos.go

@@ -7,6 +7,7 @@ import (
 	"fmt"
 	"io"
 	"net/http"
+	"os"
 	"path/filepath"
 	"regexp"
 	"strings"
@@ -23,8 +24,7 @@ type protoFile struct {
 }
 
 type ProtoSchemaSet struct {
-	Owner         string   `toml:"owner"`
-	Repository    string   `toml:"repository"`
+	URI           string   `toml:"uri"`
 	Ref           string   `toml:"ref"`            // ref or tag or commit SHA
 	Folders       []string `toml:"folders"`        // if not provided, all protos will be fetched, otherwise only protos in these folders will be fetched
 	SubjectPrefix string   `toml:"subject_prefix"` // optional prefix for subjects
@@ -36,16 +36,31 @@ type SubjectNamingStrategyFn func(subjectPrefix string, protoFile protoFile, rep
 // RepositoryToSubjectNamingStrategyFn is a map of repository names to SubjectNamingStrategyFn functions
 type RepositoryToSubjectNamingStrategyFn map[string]SubjectNamingStrategyFn
 
-func ValidateRepoConfiguration(repoConfig ProtoSchemaSet) error {
-	if repoConfig.Owner == "" {
-		return errors.New("owner is required")
+func validateRepoConfiguration(repoConfig ProtoSchemaSet) error {
+	if repoConfig.URI == "" {
+		return errors.New("uri is required")
 	}
-	if repoConfig.Repository == "" {
-		return errors.New("repo is required")
+
+	if !strings.HasPrefix(repoConfig.URI, "https://") && !strings.HasPrefix(repoConfig.URI, "file://") {
+		return errors.New("uri has to start with either 'file://' or 'https://'")
+	}
+
+	trimmedURI := strings.TrimPrefix(repoConfig.URI, "https://")
+	if !strings.HasPrefix(trimmedURI, "github.com") {
+		return fmt.Errorf("only repositories hosted at github.com are supported, but %s was found", repoConfig.URI)
+	}
+
+	parts := strings.Split(trimmedURI, "/")
+	if len(parts) < 3 {
+		return fmt.Errorf("URI should have following format: 'https://github.com/<OWNER>/<REPOSITORY>', but %s was found", repoConfig.URI)
+	}
+
+	if repoConfig.Ref == "" && strings.HasPrefix(repoConfig.URI, "https://") {
+		return errors.New("ref is required, when fetching protos from Github")
 	}
 
-	if repoConfig.Ref == "" {
-		return errors.New("ref is required")
+	if repoConfig.Ref != "" && strings.HasPrefix(repoConfig.URI, "file://") {
+		return errors.New("ref is not supported with local protos")
 	}
 
 	return nil
@@ -59,9 +74,15 @@ func RegisterAndFetchProtos(ctx context.Context, client *github.Client, protoSch
 	framework.L.Debug().Msgf("Registering and fetching protos from %d repositories", len(protoSchemaSets))
 
 	for _, protoSchemaSet := range protoSchemaSets {
-		protos, protosErr := fetchProtoFilesInFolders(ctx, client, protoSchemaSet.Owner, protoSchemaSet.Repository, protoSchemaSet.Ref, protoSchemaSet.Folders)
+		if valErr := validateRepoConfiguration(protoSchemaSet); valErr != nil {
+			return errors.Wrapf(valErr, "invalid repo configuration for schema set: %v", protoSchemaSet)
+		}
+	}
+
+	for _, protoSchemaSet := range protoSchemaSets {
+		protos, protosErr := fetchProtoFilesInFolders(ctx, client, protoSchemaSet.URI, protoSchemaSet.Ref, protoSchemaSet.Folders)
 		if protosErr != nil {
-			return errors.Wrapf(protosErr, "failed to fetch protos from %s/%s", protoSchemaSet.Owner, protoSchemaSet.Repository)
+			return errors.Wrapf(protosErr, "failed to fetch protos from %s", protoSchemaSet.URI)
 		}
 
 		protoMap := make(map[string]string)
@@ -71,7 +92,7 @@ func RegisterAndFetchProtos(ctx context.Context, client *github.Client, protoSch
 			protoMap[proto.Path] = proto.Content
 
 			var subjectStrategy SubjectNamingStrategyFn
-			if strategy, ok := repoToSubjectNamingStrategy[protoSchemaSet.Owner+"/"+protoSchemaSet.Repository]; ok {
+			if strategy, ok := repoToSubjectNamingStrategy[protoSchemaSet.URI]; ok {
 				subjectStrategy = strategy
 			} else {
 				subjectStrategy = DefaultSubjectNamingStrategy
@@ -86,7 +107,7 @@ func RegisterAndFetchProtos(ctx context.Context, client *github.Client, protoSch
 
 		registerErr := registerAllWithTopologicalSortingByTrial(schemaRegistryURL, protoMap, subjects)
 		if registerErr != nil {
-			return errors.Wrapf(registerErr, "failed to register protos from %s/%s", protoSchemaSet.Owner, protoSchemaSet.Repository)
+			return errors.Wrapf(registerErr, "failed to register protos from %s", protoSchemaSet.URI)
 		}
 	}
 
@@ -142,8 +163,22 @@ 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, owner, repository, ref string, folders []string) ([]protoFile, error) {
-	framework.L.Debug().Msgf("Fetching proto files from %s/%s in folders: %s", owner, repository, strings.Join(folders, ", "))
+func fetchProtoFilesInFolders(ctx context.Context, client *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://") {
+		return fetchProtosFromFilesystem(uri, folders)
+	}
+
+	parts := strings.Split(strings.TrimPrefix(uri, "https://"), "/")
+
+	return fetchProtosFromGithub(ctx, client, 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")
+	}
 
 	var files []protoFile
 
@@ -210,7 +245,7 @@ searchLoop:
 		})
 	}
 
-	framework.L.Debug().Msgf("Fetched %d proto files from %s/%s", len(files), owner, repository)
+	framework.L.Debug().Msgf("Fetched %d proto files from Github's %s/%s", len(files), owner, repository)
 
 	if len(files) == 0 {
 		return nil, fmt.Errorf("no proto files found in %s/%s in folders %s", owner, repository, strings.Join(folders, ", "))
@@ -219,6 +254,73 @@ searchLoop:
 	return files, nil
 }
 
+func fetchProtosFromFilesystem(uri string, folders []string) ([]protoFile, error) {
+	var files []protoFile
+
+	protoDirPath := strings.TrimPrefix(uri, "file://")
+	walkErr := filepath.Walk(protoDirPath, func(path string, info os.FileInfo, err error) error {
+		if err != nil {
+			return err
+		}
+
+		if info.IsDir() {
+			return nil
+		}
+
+		var folderFound string
+		if len(folders) > 0 {
+			matched := false
+			for _, folder := range folders {
+				if strings.HasPrefix(strings.TrimPrefix(strings.TrimPrefix(path, protoDirPath), "/"), folder) {
+					matched = true
+					folderFound = folder
+					break
+				}
+			}
+
+			if !matched {
+				return nil
+			}
+		}
+
+		if !strings.HasSuffix(path, ".proto") {
+			return nil
+		}
+
+		content, contentErr := os.ReadFile(path)
+		if contentErr != nil {
+			return errors.Wrapf(contentErr, "failed to read file at %s", path)
+		}
+
+		// subtract the folder from the path if it was provided, because if it is imported by some other protos
+		// most probably it will be imported as a relative path, so we need to remove the folder from the path
+		protoPath := strings.TrimPrefix(strings.TrimPrefix(path, protoDirPath), "/")
+		if folderFound != "" {
+			protoPath = strings.TrimPrefix(strings.TrimPrefix(protoPath, folderFound), strings.TrimSuffix(folderFound, "/"))
+			protoPath = strings.TrimPrefix(protoPath, "/")
+		}
+
+		files = append(files, protoFile{
+			Name:    filepath.Base(path),
+			Path:    protoPath,
+			Content: string(content),
+		})
+
+		return nil
+	})
+	if walkErr != nil {
+		return nil, errors.Wrapf(walkErr, "failed to walk through directory %s", protoDirPath)
+	}
+
+	framework.L.Debug().Msgf("Fetched %d proto files from local %s", len(files), protoDirPath)
+
+	if len(files) == 0 {
+		return nil, fmt.Errorf("no proto files found in '%s' in folders %s", protoDirPath, strings.Join(folders, ", "))
+	}
+
+	return files, nil
+}
+
 func resolveRefSHA(ctx context.Context, client *github.Client, owner, repository, ref string) (string, error) {
 	if refObj, _, err := client.Git.GetRef(ctx, owner, repository, "refs/tags/"+ref); err == nil {
 		return refObj.GetObject().GetSHA(), nil

framework/examples/myproject/smoke_chip_ingress_test.go

@@ -26,11 +26,10 @@ func TestChipIngressSmoke(t *testing.T) {
 
 	out, err := chipingressset.New(in.ChipIngress)
 	require.NoError(t, err, "failed to create chip ingress set")
+	require.NotEmpty(t, out.ChipIngress.GRPCExternalURL, "GRPCExternalURL is not set")
+	require.NotEmpty(t, out.RedPanda.SchemaRegistryExternalURL, "SchemaRegistryExternalURL is not set")
 
-	t.Run("chainlink-protos can be registered", func(t *testing.T) {
-		require.NotEmpty(t, out.ChipIngress.GRPCExternalURL, "GRPCExternalURL is not set")
-		require.NotEmpty(t, out.RedPanda.SchemaRegistryExternalURL, "SchemaRegistryExternalURL is not set")
-
+	t.Run("remote chainlink-protos can be registered", func(t *testing.T) {
 		ctx, cancel := context.WithTimeout(context.Background(), 2*time.Minute)
 		defer cancel()
 
@@ -48,13 +47,29 @@ func TestChipIngressSmoke(t *testing.T) {
 
 		err := chipingressset.DefaultRegisterAndFetchProtos(ctx, client, []chipingressset.ProtoSchemaSet{
 			{
-				Owner:         "smartcontractkit",
-				Repository:    "chainlink-protos",
+				URI:           "https://github.com/smartcontractkit/chainlink-protos",
 				Ref:           "95decc005a91a1fd2621af9d9f00cb36d8061067",
 				Folders:       []string{"workflows"},
 				SubjectPrefix: "cre-",
 			},
 		}, out.RedPanda.SchemaRegistryExternalURL)
 		require.NoError(t, err, "failed to register protos")
 	})
+
+	t.Run("local protos can be registered", func(t *testing.T) {
+		ctx, cancel := context.WithTimeout(context.Background(), 2*time.Minute)
+		defer cancel()
+
+		createTopicsErr := chipingressset.CreateTopics(ctx, out.RedPanda.KafkaExternalURL, []string{"cre"})
+		require.NoError(t, createTopicsErr, "failed to create topics")
+
+		err := chipingressset.DefaultRegisterAndFetchProtos(ctx, nil, []chipingressset.ProtoSchemaSet{
+			{
+				URI:           "file://../../../../chainlink-protos", // works also with absolute path
+				Folders:       []string{"workflows"},
+				SubjectPrefix: "cre-",
+			},
+		}, out.RedPanda.SchemaRegistryExternalURL)
+		require.NoError(t, err, "failed to register protos")
+	})
 }