use chip ingress witout otel

Author: Tofel

book/src/SUMMARY.md

@@ -54,6 +54,7 @@
       - [NodeSet](framework/components/chainlink/nodeset.md)
     - [Storage](framework/components/storage.md)
       - [S3](framework/components/storage/s3.md)
+    - [Chip Ingress Set](framework/components//chipingresset/chip_ingress.md)
   - [Clients]()
     - [Chainlink]()
     - [RPC]()

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

@@ -0,0 +1,116 @@
+# Chip Ingress Set
+
+Chip Ingress Set is a composite component that collects Beholder events. It is a thin `testcontainers-go` wrapper over a Docker Compose file copied from the [Atlas](https://github.com/smartcontractkit/atlas/blob/master/chip-ingress/docker-compose.yml) repo.
+
+It consists of 3 components:
+- Chip Ingress
+- Red Panda
+- Red Panda Console
+
+## Configuration
+
+To add it to your stack use following TOML:
+```toml
+[chip_ingress]
+  compose_file='../../components/chip_ingress_set/docker-compose.yml'
+  extra_docker_networks = ["my-existing-network"]
+```
+
+Where compose file indicates the location of the `docker-compose.yml` file (remote URLs are supported) and `extra_docker_networks` an optional slice of existing Docker networks, to which whole stack should be connected to.
+
+## Exposed ports
+
+These 3 components expose a variety of ports, but the most important ones from the point of view of user interaction are:
+- schema registry port: `18081`
+- Kafka port: `19092`
+- Red Panda console port: `8080`
+
+## Useful helper methods
+
+Packge contains also a bunch of helper functions tha can:
+- create and delete Kafka topics
+- fetch `.proto` files from remote repositories and register them with Red Panda
+
+
+### Topic management
+```go
+import chipingressset "github.com/smartcontractkit/chainlink-testing-framework/framework/components/chip_ingress_set"
+
+topicsErr := chipingressset.DeleteAllTopics(cmd.Context(), redPandaKafkaURLFlag)
+if topicsErr != nil {
+    panic(topicsErr)
+}
+
+createTopicsErr := chipingressset.CreateTopics(ctx, out.RedPanda.KafkaExternalURL, []string{"cre"})
+if createTopicsErr != nil {
+    panic(createTopicsErr)
+}
+```
+
+### Protobuf schema registration
+```go
+out, outErr := chipingressset.New(in.ChipIngress)
+if outErr != nil {
+    panic(outErr)
+}
+
+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{
+    {
+        Owner:   "smartcontractkit",
+        Repo:    "chainlink-protos",
+        Ref:     "626c42d55bdcb36dffe0077fff58abba40acc3e5",
+        Folders: []string{"workflows"},
+    },
+}, out.RedPanda.SchemaRegistryExternalURL)
+if protoErr != nil {
+    panic(protoErr)
+}
+```
+
+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'
+ref = '626c42d55bdcb36dffe0077fff58abba40acc3e5'
+folders = ['workflows']
+```
+
+using this code:
+```go
+var protoSchemaSets []chipingressset.ProtoSchemaSet
+for _, schemaSet := range configFiles {
+    file, fileErr := os.ReadFile(schemaSet)
+    if fileErr != nil {
+        return errors.Wrapf(fileErr, "failed to read proto schema set config file: %s", schemaSet)
+    }
+
+    type protoSchemaSets struct {
+        Sets []chipingressset.ProtoSchemaSet `toml:"proto_schema_set"`
+    }
+
+    var sets protoSchemaSets
+    if err := toml.Unmarshal(file, &sets); err != nil {
+        return errors.Wrapf(err, "failed to unmarshal proto config file: %s", protoConfig)
+    }
+
+    protoSchemaSets = append(reposConfigs, sets.Sets...)
+}
+```
+
+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.

framework/components/chip_ingress_set/chip_ingress.go

@@ -32,13 +32,14 @@ type RedPandaOutput struct {
 	SchemaRegistryExternalURL string
 	KafkaInternalURL          string
 	KafkaExternalURL          string
+	ConsoleExternalURL        string
 }
 
 type Input struct {
-	ComposeFile string   `toml:"compose_file"`
-	Topics      []string `toml:"topics"`
-	Output      *Output  `toml:"output"`
-	UseCache    bool     `toml:"use_cache"`
+	ComposeFile         string   `toml:"compose_file"`
+	ExtraDockerNetworks []string `toml:"extra_docker_networks"`
+	Output              *Output  `toml:"output"`
+	UseCache            bool     `toml:"use_cache"`
 }
 
 func defaultChipIngress(in *Input) *Input {
@@ -49,12 +50,17 @@ func defaultChipIngress(in *Input) *Input {
 }
 
 const (
+	DEFAULT_STACK_NAME = "chip-ingress"
+
 	DEFAULT_CHIP_INGRESS_GRPC_PORT    = "50051"
 	DEFAULT_CHIP_INGRESS_SERVICE_NAME = "chip-ingress"
 
 	DEFAULT_RED_PANDA_SCHEMA_REGISTRY_PORT = "18081"
 	DEFAULT_RED_PANDA_KAFKA_PORT           = "19092"
 	DEFAULT_RED_PANDA_SERVICE_NAME         = "redpanda-0"
+
+	DEFAULT_RED_PANDA_CONSOLE_SERVICE_NAME = "redpanda-console"
+	DEFAULT_RED_PANDA_CONSOLE_PORT         = "8080"
 )
 
 func New(in *Input) (*Output, error) {
@@ -69,7 +75,8 @@ func New(in *Input) (*Output, error) {
 	}
 
 	in = defaultChipIngress(in)
-	identifier := framework.DefaultTCName(DEFAULT_CHIP_INGRESS_SERVICE_NAME)
+	identifier := framework.DefaultTCName(DEFAULT_STACK_NAME)
+	framework.L.Debug().Str("Compose file", in.ComposeFile).Msgf("Starting Chip Ingress stack with identifier %s", framework.DefaultTCName(DEFAULT_STACK_NAME))
 
 	composeFilePath, fileErr := composeFilePath(in.ComposeFile)
 	if fileErr != nil {
@@ -119,45 +126,50 @@ func New(in *Input) (*Output, error) {
 	timeout := time.After(1 * time.Minute)
 	tick := time.Tick(500 * time.Millisecond)
 
-	// so let's try to connect to the default network a couple of times, there must be a race condition in Docker
+	// so let's try to connect to a Docker network a couple of times, there must be a race condition in Docker
 	// and even when network sandbox has been created and container is running, this call can still fail
 	// retrying is simpler than trying to figure out how to correctly wait for the network sandbox to be ready
-	var connectDefaultNetwork = func() error {
+	var connectNetwork = func(networkName string) error {
 		for {
 			select {
 			case <-timeout:
 				return fmt.Errorf("timeout while trying to connect chip-ingress to default network")
 			case <-tick:
 				if networkErr := cli.NetworkConnect(
 					ctx,
-					framework.DefaultNetworkName,
+					networkName,
 					chipIngressContainer.ID,
 					&networkTypes.EndpointSettings{
 						Aliases: []string{identifier},
 					},
 				); networkErr != nil && !strings.Contains(networkErr.Error(), "already exists in network") {
-					fmt.Println("failed to connect to default network", networkErr)
+					framework.L.Trace().Msgf("failed to connect to default network: %v", networkErr)
 					continue
 				}
-				fmt.Println("connected to default network")
+				framework.L.Trace().Msgf("connected to %s network", networkName)
 				return nil
 			}
 		}
 	}
 
-	if connectErr := connectDefaultNetwork(); connectErr != nil {
-		return nil, errors.Wrap(connectErr, "failed to connect chip-ingress to default network")
-	}
+	networks := []string{framework.DefaultNetworkName}
+	networks = append(networks, in.ExtraDockerNetworks...)
 
-	// verify that the container is connected to framework's network
-	inspected, inspectErr := cli.ContainerInspect(ctx, chipIngressContainer.ID)
-	if inspectErr != nil {
-		return nil, errors.Wrapf(inspectErr, "failed to inspect container %s", chipIngressContainer.ID)
-	}
+	for _, networkName := range networks {
+		framework.L.Debug().Msgf("Connecting chip-ingress to %s network", networkName)
+		if connectErr := connectNetwork(networkName); connectErr != nil {
+			return nil, errors.Wrapf(connectErr, "failed to connect chip-ingress to %s network", networkName)
+		}
+		// verify that the container is connected to framework's network
+		inspected, inspectErr := cli.ContainerInspect(ctx, chipIngressContainer.ID)
+		if inspectErr != nil {
+			return nil, errors.Wrapf(inspectErr, "failed to inspect container %s", chipIngressContainer.ID)
+		}
 
-	_, ok := inspected.NetworkSettings.Networks[framework.DefaultNetworkName]
-	if !ok {
-		return nil, fmt.Errorf("container %s is NOT on network %s", chipIngressContainer.ID, framework.DefaultNetworkName)
+		_, ok := inspected.NetworkSettings.Networks[networkName]
+		if !ok {
+			return nil, fmt.Errorf("container %s is NOT on network %s", chipIngressContainer.ID, networkName)
+		}
 	}
 
 	// get hosts and ports for chip-ingress and redpanda
@@ -188,6 +200,20 @@ func New(in *Input) (*Output, error) {
 		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 {
+		return nil, errors.Wrap(redpandaConsoleErr, "failed to get redpanda-console container")
+	}
+
+	redpandaExternalConsoleHost, redpandaExternalConsoleHostErr := redpandaConsoleContainer.Host(ctx)
+	if redpandaExternalConsoleHostErr != nil {
+		return nil, errors.Wrap(redpandaExternalConsoleHostErr, "failed to get host for Red Panda Console")
+	}
+	redpandaExternalConsolePort, redpandaExternalConsolePortErr := redpandaConsoleContainer.MappedPort(ctx, 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),
@@ -198,9 +224,12 @@ func New(in *Input) (*Output, error) {
 			SchemaRegistryExternalURL: fmt.Sprintf("http://%s:%s", redpandaExternalHost, redpandaExternalSchemaRegistryPort.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()),
 		},
 	}
 
+	framework.L.Info().Msg("Chip Ingress stack start")
+
 	return output, nil
 }
 
@@ -216,7 +245,7 @@ func composeFilePath(rawFilePath string) (string, error) {
 	}
 	defer resp.Body.Close()
 
-	tempFile, tempErr := os.CreateTemp(".", "chip-ingress-docker-compose.yml")
+	tempFile, tempErr := os.CreateTemp("", "chip-ingress-docker-compose-*.yml")
 	if tempErr != nil {
 		return "", errors.Wrap(tempErr, "failed to create temp file")
 	}

framework/components/chip_ingress_set/docker-compose.yml

@@ -1,11 +1,11 @@
+# copied from https://github.com/smartcontractkit/atlas/blob/master/chip-ingress/docker-compose.yml
 services:
 
   chip-ingress:
-    image: chip-ingress:latest
+    image: ${CHIP_INGRESS_IMAGE:-chip-ingress:latest}
     container_name: chip-ingress
     depends_on:
       - redpanda-0
-      - otel-lgtm
     restart: on-failure
     tty: true
     command: ["write-service"]
@@ -15,7 +15,9 @@ services:
       SERVER_METRICS_PORT: "9090"
       SERVER_HEALTHCHECKS_PORT: "9091"
       SERVER_METRICS_IMPLEMENTATION: "otel"
-      SERVER_METRICS_OTEL_EXPORTER_GRPC_ENDPOINT: ${SERVER_METRICS_OTEL_EXPORTER_GRPC_ENDPOINT:-otel-lgtm:4317}
+      # this has been modified from the original docker-compose.yml to use localhost:4317 as we don't have otel-lgtm service in this stack
+      # but we have our own observability stack, which is accessible via localhost (after starting it with ctf obs u)
+      SERVER_METRICS_OTEL_EXPORTER_GRPC_ENDPOINT: ${SERVER_METRICS_OTEL_EXPORTER_GRPC_ENDPOINT:-localhost:4317}
       KAFKA_BROKERS: "redpanda-0:9092"
       SCHEMA_REGISTRY_URL: "http://redpanda-0:8081"
       # this has been modified and default values were removed, somehow we cannot override them in testcontainers-go
@@ -33,8 +35,8 @@ services:
       - "9090:9090"
       - "9092:9091"
     healthcheck:
+      # this has been modified from the original docker-compose.yml to use grpc_health_probe
       test: [ "CMD", "grpc_health_probe", "-addr=localhost:50051" ]
-      # test: [ "CMD-SHELL", "sleep 1"]
       interval: 200ms
       timeout: 10s
       retries: 10
@@ -96,11 +98,5 @@ services:
     depends_on:
       - redpanda-0
 
-  otel-lgtm:
-    image: grafana/otel-lgtm
-    ports:
-      - "3000:3000"
-      - "4317:4317"
-      - "4318:4318"
-      - "3100:3100"
-
+# removed otel-lgtm service as it's not needed for the stack to work
+# and we have our own observability stack