implement actual grpc trigger (#903)

.github/workflows/pull-request.yaml

@@ -157,7 +157,7 @@ jobs:
           --set telemetry.enabled="true" \
           --set telemetry.otelCollectorEndpoint="otel-collector.tracetest.svc.cluster.local:4317" \
           --set poolingConfig.maxWaitTimeForTrace="60s" \
-          --set poolingConfig.retryDelay="1s" \
+          --set poolingConfig.retryDelay="6s" \
           --set ingress.enabled=false \
           --wait --timeout 3m
 

cli/conversion/openapi_definition_conversion.go

@@ -61,7 +61,6 @@ func convertGRPCOpenAPIIntoDefinition(request *openapi.GRPCRequest) definition.G
 	return definition.GrpcRequest{
 		ProtobufFile: ConvertOpenapiStringIntoString(request.ProtobufFile),
 		Address:      ConvertOpenapiStringIntoString(request.Address),
-		Service:      ConvertOpenapiStringIntoString(request.Service),
 		Method:       ConvertOpenapiStringIntoString(request.Method),
 		Metadata:     metadata,
 		Auth:         getAuthDefinition(request.Auth),

cli/definition/grcp.go

@@ -5,9 +5,8 @@ import (
 )
 
 type GrpcRequest struct {
-	ProtobufFile string             `yaml:"protobuf_file"`
+	ProtobufFile string             `yaml:"protobufFile"`
 	Address      string             `yaml:"address"`
-	Service      string             `yaml:"service"`
 	Method       string             `yaml:"method"`
 	Metadata     []GRPCHeader       `yaml:"metadata,omitempty"`
 	Auth         HTTPAuthentication `yaml:"authentication,omitempty"`
@@ -16,17 +15,13 @@ type GrpcRequest struct {
 
 func (r GrpcRequest) Validate() error {
 	if r.ProtobufFile == "" {
-		return fmt.Errorf("protobuf_file cannot be empty")
+		return fmt.Errorf("protobufFile cannot be empty")
 	}
 
 	if r.Address == "" {
 		return fmt.Errorf("address cannot be empty")
 	}
 
-	if r.Service == "" {
-		return fmt.Errorf("service cannot be empty")
-	}
-
 	if r.Method == "" {
 		return fmt.Errorf("method cannot be empty")
 	}

cli/definition/test.go

@@ -29,22 +29,21 @@ type TestTrigger struct {
 }
 
 func (t TestTrigger) Validate() error {
-	validTypes := map[string]bool{
-		"http": true,
-	}
-
-	if t.Type == "" {
+	switch t.Type {
+	case "http":
+		if err := t.HTTPRequest.Validate(); err != nil {
+			return fmt.Errorf("http request must be valid: %w", err)
+		}
+	case "grpc":
+		if err := t.GRPC.Validate(); err != nil {
+			return fmt.Errorf("grpc request must be valid: %w", err)
+		}
+	case "":
 		return fmt.Errorf("type cannot be empty")
-	}
-
-	if _, ok := validTypes[t.Type]; !ok {
+	default:
 		return fmt.Errorf("type \"%s\" is not supported", t.Type)
 	}
 
-	if err := t.HTTPRequest.Validate(); err != nil {
-		return fmt.Errorf("http request must be valid: %w", err)
-	}
-
 	return nil
 }
 

cli/examples/grpc.yaml

@@ -0,0 +1,52 @@
+id: f89be07c-3efb-4389-b2ea-06ebcc5ab144
+name: Test Create with gRCP trigger
+description: ""
+trigger:
+  type: grpc
+  grpc:
+    protobufFile: |
+      syntax = "proto3";
+
+      option java_multiple_files = true;
+      option java_outer_classname = "PokeshopProto";
+      option objc_class_prefix = "PKS";
+
+      package pokeshop;
+
+      service Pokeshop {
+        rpc getPokemonList (GetPokemonRequest) returns (GetPokemonListResponse) {}
+        rpc createPokemon (Pokemon) returns (Pokemon) {}
+        rpc importPokemon (ImportPokemonRequest) returns (ImportPokemonRequest) {}
+      }
+
+      message ImportPokemonRequest {
+        int32 id = 1;
+      }
+
+      message GetPokemonRequest {
+        optional int32 skip = 1;
+        optional int32 take = 2;
+      }
+
+      message GetPokemonListResponse {
+        repeated Pokemon items = 1;
+        int32 totalCount = 2;
+      }
+
+      message Pokemon {
+        optional int32 id = 1;
+        string name = 2;
+        string type = 3;
+        bool isFeatured = 4;
+        optional string imageUrl = 5;
+      }
+    address: localhost:8002
+    method: pokeshop.Pokeshop.importPokemon
+    request: |
+      {
+        "id": 52
+      }
+testDefinition:
+- selector: span[tracetest.span.type="rpc"]
+  assertions:
+  - tracetest.selected_spans.count = 1

docker-compose.yaml

@@ -8,7 +8,7 @@ services:
       - VERSION=dev
     volumes:
       - type: bind
-        source: ./config.tests.yaml
+        source: ./local-config/config.tests.yaml
         target: /app/config.yaml
     ports:
       - 8080:8080
@@ -82,17 +82,38 @@ services:
       timeout: 5s
       retries: 60
 
-  demo:
+  demo-api:
     image: kubeshop/demo-pokemon-api:latest
+    pull_policy: always
     environment:
       REDIS_URL: cache
       DATABASE_URL: postgresql://postgres:postgres@postgres:5432/postgres?schema=public
       RABBITMQ_HOST: queue
       POKE_API_BASE_URL: https://pokeapi.co/api/v2
       JAEGER_HOST: jaeger
       JAEGER_PORT: 6832
-    ports:
-      - 8001:80
+      NPM_RUN_COMMAND: api
+    depends_on:
+      postgres:
+        condition: service_healthy
+      cache:
+        condition: service_healthy
+      queue:
+        condition: service_healthy
+      jaeger:
+        condition: service_healthy
+
+  demo-rpc:
+    image: kubeshop/demo-pokemon-api:latest
+    pull_policy: always
+    environment:
+      REDIS_URL: cache
+      DATABASE_URL: postgresql://postgres:postgres@postgres:5432/postgres?schema=public
+      RABBITMQ_HOST: queue
+      POKE_API_BASE_URL: https://pokeapi.co/api/v2
+      JAEGER_HOST: jaeger
+      JAEGER_PORT: 6832
+      NPM_RUN_COMMAND: rpc
     depends_on:
       postgres:
         condition: service_healthy
@@ -102,3 +123,4 @@ services:
         condition: service_healthy
       jaeger:
         condition: service_healthy
+

config.tests.yaml → local-config/config.tests.yaml

@@ -8,18 +8,14 @@ tracingBackend:
         insecure: true
 
 poolingConfig:
-  maxWaitTimeForTrace: 30s
-  retryDelay: 500ms
+  maxWaitTimeForTrace: 60s
+  retryDelay: 6s
 
 googleAnalytics:
   enabled: false
 
 telemetry:
+  enabled: true
   serviceName: tracetest
   sampling: 100 # 100%
-  jaeger:
-    host: jaeger
-    port: 6831
-  exporters:
-    - console
-    - jaeger
+  otelCollectorEndpoint: otel-collector:4317

docker-compose.testrunner.yaml → local-config/docker-compose.testrunner.yaml

@@ -9,6 +9,7 @@ services:
       - TARGET_URL=http://tracetest:8080
       - TRACETEST_MAIN_ENDPOINT=tracetest:8080
       - TRACETEST_TARGET_ENDPOINT=tracetest:8080
-      - DEMO_APP_URL=http://demo
+      - DEMO_APP_URL=http://demo-api:8081
+      - DEMO_APP_GRPC_URL=demo-rpc:8082
     depends_on:
       - tracetest

run-tracetesting-tests.sh

@@ -19,8 +19,8 @@ done
 
 
 docker compose -f docker-compose.yaml up -d --build --remove-orphans
-docker compose -f docker-compose.yaml -f docker-compose.testrunner.yaml build
-docker compose -f docker-compose.yaml -f docker-compose.testrunner.yaml run testrunner
+docker compose -f docker-compose.yaml -f local-config/docker-compose.testrunner.yaml build
+docker compose -f docker-compose.yaml -f local-config/docker-compose.testrunner.yaml run testrunner
 
 if [ "$STOP" == "yes" ]; then
   docker compose -f docker-compose.yaml stop

server/app/app.go

@@ -91,6 +91,7 @@ func (a *App) Start() error {
 
 	triggerReg := trigger.NewRegsitry(a.tracer)
 	triggerReg.Add(trigger.HTTP())
+	triggerReg.Add(trigger.GRPC())
 
 	subscriptionManager := subscription.NewManager()
 

server/encoding/yaml/definition/test.go

@@ -35,6 +35,7 @@ type TestTrigger struct {
 func (t TestTrigger) Validate() error {
 	validTypes := map[string]bool{
 		"http": true,
+		"grpc": true,
 	}
 
 	if t.Type == "" {

server/executor/poller_executor.go

@@ -1,7 +1,6 @@
 package executor
 
 import (
-	"encoding/hex"
 	"fmt"
 	"math"
 	"time"
@@ -29,21 +28,26 @@ func (pe InstrumentedPollerExecutor) ExecuteRequest(request *PollingRequest) (bo
 	defer span.End()
 
 	finished, run, err := pe.pollerExecutor.ExecuteRequest(request)
-	if err != nil {
-		span.RecordError(err)
-		return finished, run, err
-	}
 
-	spanCount := len(run.Trace.Flat)
+	spanCount := 0
+	if run.Trace != nil {
+		spanCount = len(run.Trace.Flat)
+	}
 
-	span.SetAttributes(
-		attribute.String("tracetest.run.trace_poller.trace_id", hex.EncodeToString(request.run.TraceID[:])),
-		attribute.String("tracetest.run.trace_poller.span_id", hex.EncodeToString(request.run.SpanID[:])),
+	attrs := []attribute.KeyValue{
+		attribute.String("tracetest.run.trace_poller.trace_id", request.run.TraceID.String()),
+		attribute.String("tracetest.run.trace_poller.span_id", request.run.SpanID.String()),
 		attribute.Bool("tracetest.run.trace_poller.succesful", finished),
 		attribute.String("tracetest.run.trace_poller.test_id", request.test.ID.String()),
 		attribute.Int("tracetest.run.trace_poller.amount_retrieved_spans", spanCount),
-	)
+	}
+
+	if err != nil {
+		attrs = append(attrs, attribute.String("tracetest.run.trace_poller.error", err.Error()))
+		span.RecordError(err)
+	}
 
+	span.SetAttributes(attrs...)
 	return finished, run, err
 }