update the GH workflow, make most of the structs private, wait for chip ingress readiness before registering anything

Author: Tofel

.github/workflows/framework-dockercompose-tests.yml

@@ -6,7 +6,7 @@ jobs:
   test:
     defaults:
       run:
-        working-directory: framework/examples/myproject
+        working-directory: framework/examples/chip_ingress
     runs-on: ubuntu-latest
     permissions:
       id-token: write
@@ -70,7 +70,8 @@ jobs:
         if: steps.changes.outputs.src == 'true'
         env:
           CTF_CONFIGS: ${{ matrix.test.config }}
-          CHIP_INGRESS_IMAGE: ${{ secrets.AWS_ACCOUNT_ID_PROD }}.dkr.ecr.us-west-2.amazonaws.com/atlas-chip-ingress:qa-latest
+          CHIP_INGRESS_IMAGE: ${{ secrets.AWS_ACCOUNT_ID_PROD }}.dkr.ecr.us-west-2.amazonaws.com/atlas-chip-ingress:da84cb72d3a160e02896247d46ab4b9806ebee2f
+          CHIP_CONFIG_IMAGE: ${{ secrets.AWS_ACCOUNT_ID_PROD }}.dkr.ecr.us-west-2.amazonaws.com/atlas-chip-config:7b4e9ee68fd1c737dd3480b5a3ced0188f29b969
           CTF_LOG_LEVEL: debug
         run: |
           go test -timeout ${{ matrix.test.timeout }} -v -count ${{ matrix.test.count }} -run ${{ matrix.test.name }}

framework/components/dockercompose/chip_ingress_set/chip_config.go

@@ -14,35 +14,12 @@ import (
 	"github.com/jhump/protocompile"
 	"google.golang.org/protobuf/reflect/protoreflect"
 
-	cc "github.com/smartcontractkit/atlas/chip-config/client" // TODO: can we move it to chainlink-common?
 	"github.com/smartcontractkit/chainlink-common/pkg/chipingress/pb"
 )
 
 // code copied from: https://github.com/smartcontractkit/atlas/blob/master/chip-cli/config/config.go and https://github.com/smartcontractkit/atlas/blob/master/chip-cli/config/proto_validator.go
-// reason: avoid dependency on the chip-cli module in the testing framework
-func chipConfigClient(ctx context.Context, chipConfigOutput *ChipConfigOutput) (cc.ChipConfigClient, error) {
-	fmt.Printf("🔌 Initiating connection to Chip Config at \033[1m%s\033[0m...\n\n", chipConfigOutput.GRPCExternalURL)
-
-	var clientOpts []cc.ClientOpt
-	clientOpts = append(clientOpts, cc.WithBasicAuth(chipConfigOutput.Username, chipConfigOutput.Password))
-
-	client, err := cc.NewChipConfigClient(chipConfigOutput.GRPCExternalURL, clientOpts...)
-	if err != nil {
-		return nil, fmt.Errorf("failed to create Chip Config client: %w", err)
-	}
-
-	// Check we can connect to the server
-	_, pErr := client.Ping(ctx)
-	if pErr != nil {
-		return nil, fmt.Errorf("failed to connect to Chip Config: %w", pErr)
-	}
-
-	fmt.Printf("🔗 Connected to Chip Config\n\n")
-
-	return client, nil
-}
-
-func convertToPbSchemas(schemas map[string]*Schema, domain string) []*pb.Schema {
+// reason: avoid dependency on the private atlas module
+func convertToPbSchemas(schemas map[string]*schema, domain string) []*pb.Schema {
 	pbSchemas := make([]*pb.Schema, len(schemas))
 
 	for i, schema := range slices.Collect(maps.Values(schemas)) {
@@ -84,54 +61,53 @@ func convertToPbSchemas(schemas map[string]*Schema, domain string) []*pb.Schema
 	return pbSchemas
 }
 
-type RegistrationConfig struct {
+type registrationConfig struct {
 	Domain  string   `json:"domain"`
-	Schemas []Schema `json:"schemas"`
+	Schemas []schema `json:"schemas"`
 }
 
-type Schema struct {
+type schema struct {
 	Entity        string            `json:"entity"`
 	Path          string            `json:"path"`
-	References    []SchemaReference `json:"references,omitempty"`
+	References    []schemaReference `json:"references,omitempty"`
 	SchemaContent string
-	Metadata      Metadata `json:"metadata,omitempty"`
+	Metadata      metadata `json:"metadata,omitempty"`
 }
 
-type Metadata struct {
-	Stores map[string]Store `json:"stores"`
+type metadata struct {
+	Stores map[string]store `json:"stores"`
 }
 
-type Store struct {
+type store struct {
 	Index     []string `json:"index"`
 	Partition []string `json:"partition"`
 }
 
-type SchemaReference struct {
+type schemaReference struct {
 	Name   string `json:"name"`
 	Entity string `json:"entity"`
 	Path   string `json:"path"`
 }
 
-func parseSchemaConfig(configFilePath, schemaDir string) (*RegistrationConfig, map[string]*Schema, error) {
+func parseSchemaConfig(configFilePath, schemaDir string) (*registrationConfig, map[string]*schema, error) {
 	cfg, err := readConfig(configFilePath)
 	if err != nil {
 		return nil, nil, err
 	}
 
-	if err := ValidateEntityNames(cfg, schemaDir); err != nil {
+	if err := validateEntityNames(cfg, schemaDir); err != nil {
 		return nil, nil, fmt.Errorf("entity name validation failed: %w", err)
 	}
 
 	// Our end goal is to generate a schema registration request to chip config
 	// We will use a map to store the schemas by entity and path
 	// this is because more than one schema may reference the same schema
 	// technically, since SR is idempotent, this is not strictly necessary, as duplicate registrations are noop
-	schemas := make(map[string]*Schema)
-
-	for _, schema := range cfg.Schemas {
+	schemas := make(map[string]*schema)
 
+	for _, s := range cfg.Schemas {
 		// For each of the schemas, we need to get the references schema content
-		for _, reference := range schema.References {
+		for _, reference := range s.References {
 
 			// read schema contents
 			refSchemaContent, err := os.ReadFile(path.Join(schemaDir, reference.Path))
@@ -147,47 +123,47 @@ func parseSchemaConfig(configFilePath, schemaDir string) (*RegistrationConfig, m
 				continue
 			}
 
-			schemas[key] = &Schema{
+			schemas[key] = &schema{
 				Entity:        reference.Entity,
 				Path:          reference.Path,
 				SchemaContent: string(refSchemaContent),
 			}
 		}
 
 		// add the root schema to the map
-		schemaContent, err := os.ReadFile(path.Join(schemaDir, schema.Path))
+		schemaContent, err := os.ReadFile(path.Join(schemaDir, s.Path))
 		if err != nil {
 			return nil, nil, fmt.Errorf("error reading schema: %v", err)
 		}
 
-		key := fmt.Sprintf("%s:%s", schema.Entity, schema.Path)
+		key := fmt.Sprintf("%s:%s", s.Entity, s.Path)
 		// if the schema already exists, that means it is referenced by another schema.
 		// so we just need to add the references to the existing schema in the map
 		if s, ok := schemas[key]; ok {
-			s.References = append(s.References, schema.References...)
+			s.References = append(s.References, s.References...)
 			continue
 		}
 
-		schemas[key] = &Schema{
-			Entity:        schema.Entity,
-			Path:          schema.Path,
+		schemas[key] = &schema{
+			Entity:        s.Entity,
+			Path:          s.Path,
 			SchemaContent: string(schemaContent),
-			References:    schema.References,
+			References:    s.References,
 		}
 
 	}
 
 	return cfg, schemas, nil
 }
 
-func readConfig(path string) (*RegistrationConfig, error) {
+func readConfig(path string) (*registrationConfig, error) {
 	f, err := os.Open(path)
 	if err != nil {
 		return nil, fmt.Errorf("failed to open config file '%s': %w", path, err)
 	}
 	defer f.Close()
 
-	var cfg RegistrationConfig
+	var cfg registrationConfig
 
 	dErr := json.NewDecoder(f).Decode(&cfg)
 	if dErr != nil {
@@ -197,10 +173,10 @@ func readConfig(path string) (*RegistrationConfig, error) {
 	return &cfg, nil
 }
 
-// ValidateEntityNames validates that all entity names in the config match the fully qualified
+// validateEntityNames validates that all entity names in the config match the fully qualified
 // protobuf names (package.MessageName) from their corresponding proto files.
 // It collects all validation errors and returns them together for better user experience.
-func ValidateEntityNames(cfg *RegistrationConfig, schemaDir string) error {
+func validateEntityNames(cfg *registrationConfig, schemaDir string) error {
 	var errors []string
 
 	for _, schema := range cfg.Schemas {

framework/components/dockercompose/chip_ingress_set/chip_ingress.go

@@ -17,6 +17,8 @@ import (
 	"github.com/testcontainers/testcontainers-go/modules/compose"
 	"github.com/testcontainers/testcontainers-go/wait"
 
+	"github.com/smartcontractkit/chainlink-common/pkg/chipingress"
+	"github.com/smartcontractkit/chainlink-common/pkg/chipingress/pb"
 	"github.com/smartcontractkit/chainlink-testing-framework/framework"
 )
 
@@ -236,8 +238,8 @@ func NewWithContext(ctx context.Context, in *Input) (*Output, error) {
 
 	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, DEFAULT_CHIP_INGRESS_GRPC_PORT),
+			GRPCInternalURL: fmt.Sprintf("%s:%s", DEFAULT_CHIP_INGRESS_SERVICE_NAME, DEFAULT_CHIP_INGRESS_GRPC_PORT),
+			GRPCExternalURL: fmt.Sprintf("%s:%s", chipIngressExternalHost, DEFAULT_CHIP_INGRESS_GRPC_PORT),
 		},
 		ChipConfig: &ChipConfigOutput{
 			GRPCInternalURL: fmt.Sprintf("%s:%s", DEFAULT_CHIP_CONFIG_SERVICE_NAME, DEFAULT_CHIP_CONFIG_INTERNAL_PORT),
@@ -258,7 +260,12 @@ func NewWithContext(ctx context.Context, in *Input) (*Output, error) {
 	in.UseCache = true
 	framework.L.Info().Msg("Chip Ingress stack started")
 
-	return output, checkSchemaRegistryReadiness(ctx, 2*time.Minute, 300*time.Millisecond, output.RedPanda.SchemaRegistryExternalURL, 3)
+	rpReadyErr := checkSchemaRegistryReadiness(ctx, 2*time.Minute, 300*time.Millisecond, output.RedPanda.SchemaRegistryExternalURL, 3)
+	if rpReadyErr != nil {
+		return nil, errors.Wrap(rpReadyErr, "redpanda not ready")
+	}
+
+	return output, checkChipIngressReadiness(ctx, 2*time.Minute, 300*time.Millisecond, 3, output.ChipIngress)
 }
 
 func composeFilePath(rawFilePath string) (string, error) {
@@ -323,9 +330,11 @@ func connectNetwork(connCtx context.Context, timeout time.Duration, dockerClient
 // checkSchemaRegistryReadiness verifies that the Schema Registry answers 2xx on GET /subjects
 // for minSuccessCount *consecutive* attempts, polling every `interval`, with an overall `timeout`.
 func checkSchemaRegistryReadiness(ctx context.Context, timeout, interval time.Duration, registryURL string, minSuccessCount int) error {
-	if minSuccessCount < 1 {
-		minSuccessCount = 1
-	}
+	framework.L.Info().Msg("Starting Schema Registry readiness check")
+	defer func() {
+		framework.L.Info().Msg("Schema Registry readiness check finished")
+	}()
+
 	u, uErr := url.Parse(registryURL)
 	if uErr != nil {
 		return fmt.Errorf("parse registry URL: %w", uErr)
@@ -356,10 +365,7 @@ func checkSchemaRegistryReadiness(ctx context.Context, timeout, interval time.Du
 	t := time.NewTicker(interval)
 	defer t.Stop()
 
-	consecutive := 0
-	var lastErr error
-
-	for {
+	return readinessCheck(ctx, timeout, interval, minSuccessCount, func(ctx context.Context) error {
 		req, _ := http.NewRequestWithContext(ctx, http.MethodGet, u.String(), nil)
 		// small belt-and-suspenders to ensure no reuse even if transport changes
 		req.Close = true
@@ -372,21 +378,99 @@ func checkSchemaRegistryReadiness(ctx context.Context, timeout, interval time.Du
 		}
 
 		if err == nil && resp.StatusCode/100 == 2 {
-			framework.L.Debug().Msgf("schema registry ready check succeeded with status %d (%d/%d)", resp.StatusCode, consecutive+1, minSuccessCount)
+			return nil
+		} else {
+			if err != nil {
+				return fmt.Errorf("GET /subjects failed: %w", err)
+			} else {
+				return fmt.Errorf("GET /subjects status %d", resp.StatusCode)
+			}
+		}
+	})
+}
+
+// checkChipIngressReadiness verifies that the Chip Ingress can register a schema
+// for minSuccessCount *consecutive* attempts, polling every `interval`, with an overall `timeout`.
+func checkChipIngressReadiness(ctx context.Context, timeout, interval time.Duration, minSuccessCount int, chipIngressOutput *ChipIngressOutput) error {
+	client, cErr := chipingress.NewClient(chipIngressOutput.GRPCExternalURL)
+	if cErr != nil {
+		return fmt.Errorf("failed to create Chip client: %w", cErr)
+	}
+
+	framework.L.Info().Msg("Starting Chip Ingress readiness check")
+	defer func() {
+		framework.L.Info().Msg("Chip Ingress readiness check finished")
+	}()
+
+	ctx, cancel := context.WithTimeout(ctx, timeout)
+	defer cancel()
+
+	t := time.NewTicker(interval)
+	defer t.Stop()
+
+	// raw schema from demo_client_payload.proto with one new field
+	schema := `
+		syntax = "proto3";
+
+		option go_package = "github.com/smartcontractkit/atlas/chip-ingress/cmd/demo_client/pb";
+
+		package pb;
+
+		// Used for testing
+		message DemoClientPayload {
+		  string id=1;
+		  string domain=2;
+		  string entity=3;
+		  int64 batch_num=4;
+		  int64 message_num=5;
+		  int64 batch_position=6;
+		  optional string new_field=7; // New field added
+		}
+	`
+
+	subject := "platform-demo"
+	schemaObj := &pb.Schema{
+		Subject: subject,
+		Format:  pb.SchemaType_PROTOBUF,
+		Schema:  schema,
+	}
+
+	request := &pb.RegisterSchemaRequest{
+		Schemas: []*pb.Schema{schemaObj},
+	}
+
+	return readinessCheck(ctx, timeout, interval, minSuccessCount, func(ctx context.Context) error {
+		_, err := client.RegisterSchema(ctx, request)
+		return err
+	})
+}
+
+func readinessCheck(ctx context.Context, timeout, interval time.Duration, minSuccessCount int, checkFunc func(context.Context) error) error {
+	if minSuccessCount < 1 {
+		minSuccessCount = 1
+	}
+
+	ctx, cancel := context.WithTimeout(ctx, timeout)
+	defer cancel()
+
+	t := time.NewTicker(interval)
+	defer t.Stop()
+
+	consecutive := 0
+	var lastErr error
+
+	for {
+		if err := checkFunc(ctx); err == nil {
+			framework.L.Debug().Msgf("readiness check succeeded (%d/%d)", consecutive+1, minSuccessCount)
 			consecutive++
 			if consecutive >= minSuccessCount {
-				framework.L.Debug().Msg("schema registry is ready")
+				framework.L.Debug().Msg("service is ready")
 				return nil
 			}
 		} else {
 			consecutive = 0
-			if err != nil {
-				framework.L.Debug().Msgf("schema registry ready check failed with error %v (need %d/%d consecutive successes)", err, consecutive, minSuccessCount)
-				lastErr = fmt.Errorf("GET /subjects failed: %w", err)
-			} else {
-				framework.L.Debug().Msgf("schema registry ready check failed with error %v and status code %d (need %d/%d consecutive successes)", err, resp.StatusCode, consecutive, minSuccessCount)
-				lastErr = fmt.Errorf("GET /subjects status %d", resp.StatusCode)
-			}
+			framework.L.Debug().Msgf("readiness check failed with error %v (need %d/%d consecutive successes)", err, consecutive, minSuccessCount)
+			lastErr = fmt.Errorf("readiness check failed: %w", err)
 		}
 
 		select {
@@ -397,7 +481,7 @@ func checkSchemaRegistryReadiness(ctx context.Context, timeout, interval time.Du
 			return fmt.Errorf("schema registry not ready after %s; needed %d consecutive successes (got %d): %w",
 				timeout, minSuccessCount, consecutive, lastErr)
 		case <-t.C:
-			framework.L.Debug().Msg("schema registry not ready yet, retrying...")
+			framework.L.Debug().Msg("Schema registry not ready yet, retrying...")
 			// poll again
 		}
 	}

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

@@ -31,6 +31,12 @@ services:
       CSA_AUTH_USERNAME: ${CSA_AUTH_USERNAME:-}
       CSA_AUTH_PASSWORD: ${CSA_AUTH_PASSWORD:-}
       CSA_AUTH_REQUIRE_TLS: ${CSA_AUTH_REQUIRE_TLS:-false}
+      CHIP_CONFIG_ENABLED: ${CHIP_CONFIG_ENABLED:-true}
+      CHIP_CONFIG_REQUIRE_TLS: ${CHIP_CONFIG_REQUIRE_TLS:-false}
+      CHIP_CONFIG_AUTH_USERNAME: ${CHIP_CONFIG_AUTH_USERNAME:-admin}
+      CHIP_CONFIG_AUTH_PASSWORD: ${CHIP_CONFIG_AUTH_PASSWORD:-password}
+      CHIP_CONFIG_HOST: ${CHIP_CONFIG_HOST:-chip-config}
+      CHIP_CONFIG_PORT: ${CHIP_CONFIG_PORT:-50051}
     ports:
       - "50051:50051"
       - "9090:9090"
@@ -75,7 +81,7 @@ services:
       POSTGRES_USER: postgres
       POSTGRES_PASSWORD: postgres
     healthcheck:
-      test: [ "CMD", "pg_isready", "-U", "${POSTGRES_USER}", "-d", "${POSTGRES_DB}", "-h", "localhost" ]
+      test: [ "CMD", "pg_isready", "-U", "postgres", "-d", "chip_config", "-h", "localhost" ]
       interval: 5s
       timeout: 10s
       retries: 3