Enhance runtime service to support optional OTLP/gRPC spans alongside OTLP/HTTP. Update configuration to include gRPC listener address and modify documentation to reflect new capabilities. Improve test coverage for gRPC functionality and ensure source summary handling in the ingestion process.

a-a-k · a-a-k · commit 958faad41b33 · 2026-03-11T17:01:36.000+03:00
diff --git a/README.md b/README.md
@@ -5,7 +5,7 @@ Bering is a discovery and publishing layer for service topology and endpoint con
 It supports two operating modes:
 
 - deterministic batch discovery from trace files and directories
-- long-running runtime discovery that accepts OTLP/HTTP spans and publishes rolling snapshot envelopes for observability consumers
+- long-running runtime discovery that accepts OTLP/HTTP and optional OTLP/gRPC spans and publishes rolling snapshot envelopes for observability consumers
 
 Bering owns discovery and discovery-side public contracts. It does not own simulation, gating, chaos execution, or policy decisions.
 
@@ -40,7 +40,7 @@ cmd/bering                    CLI entrypoint
 internal/app                  command wiring
 internal/config               serve-mode config parsing and validation
 internal/connectors/traces    file/dir trace loading and normalization
-internal/connectors/otlp      OTLP/HTTP request decoding into normalized spans
+internal/connectors/otlp      OTLP request decoding into normalized spans
 internal/discovery            source-agnostic discovery engine and overlay application
 internal/model                stable core model structs, semantic checks, canonical IO
 internal/overlay              generic discovery overlay loader
@@ -59,7 +59,7 @@ scripts/ci                    CI helper scripts
 ```bash
 bering discover --input <trace-file|dir> [--out bering-model.json] [--snapshot-out bering-snapshot.json] [--replicas replicas.yaml|json] [--overlay overlay.yaml] [--discovered-at RFC3339]
 bering validate --input <bering-model.json|bering-snapshot.json>
-bering serve --config configs/serve.sample.yaml [--listen :4318] [--window-size 30s] [--flush-interval 5s]
+bering serve --config configs/serve.sample.yaml [--listen :4318] [--grpc-listen :4317] [--window-size 30s] [--flush-interval 5s]
 ```
 
 ## Quickstart
@@ -100,11 +100,12 @@ go run ./cmd/bering serve --config configs/serve.sample.yaml
 The runtime service exposes:
 
 - `POST /v1/traces` for OTLP/HTTP trace ingest
+- OTLP/gRPC trace export service on the configured gRPC address
 - `GET /healthz`
 - `GET /readyz`
 - `GET /metrics`
 
-The primary integration path is standard OpenTelemetry Collector or SDK exporters sending spans to Bering over OTLP/HTTP. No custom Collector build is required.
+The primary integration path remains standard OpenTelemetry Collector or SDK exporters sending spans to Bering over OTLP/HTTP. OTLP/gRPC is also supported for grpc-first collector topologies. No custom Collector build is required.
 
 ### 5) Use the stable model with Sheaft
 
diff --git a/configs/serve.sample.yaml b/configs/serve.sample.yaml
@@ -1,5 +1,6 @@
 server:
   listen_address: ":4318"
+  grpc_listen_address: ":4317"
   max_request_bytes: 5242880
 runtime:
   flush_interval: 5s
diff --git a/docs/architecture.md b/docs/architecture.md
@@ -14,8 +14,8 @@ Bering now has two user-facing flows built on one normalized discovery core.
 
 ### Runtime flow
 
-1. `bering serve` accepts OTLP/HTTP `POST /v1/traces` requests.
-2. OTLP requests are normalized into the same internal `traces.Span` shape used by batch mode.
+1. `bering serve` accepts OTLP/HTTP `POST /v1/traces` requests and optional OTLP/gRPC trace export requests.
+2. Both OTLP transports are normalized into the same internal `traces.Span` shape used by batch mode.
 3. Spans are accumulated in a bounded active tumbling window.
 4. On schedule, the window closes and Bering runs discovery.
 5. Bering computes a stable topology digest, diffs against the previous snapshot, and writes the snapshot to sinks.
@@ -43,7 +43,7 @@ Fields used today:
 - timestamps: `start_time`, `end_time`
 - discovery-relevant attributes: HTTP and messaging attributes
 
-This keeps discovery logic source-agnostic. File JSON, OTLP/HTTP, and future adapters all converge on the same shape.
+This keeps discovery logic source-agnostic. File JSON, OTLP/HTTP, OTLP/gRPC, and future adapters all converge on the same shape.
 
 ## Contracts
 
diff --git a/docs/migration-notes.md b/docs/migration-notes.md
@@ -17,6 +17,7 @@ You can now opt into additional discovery-side features.
 - `--overlay` for generic discovery metadata overlays
 - `--snapshot-out` for a snapshot envelope in batch mode
 - `bering serve` for OTLP/HTTP runtime discovery
+- optional OTLP/gRPC ingest on a separate runtime listener
 
 ## Sheaft and other downstream model consumers
 
diff --git a/docs/mvp-scope-and-limits.md b/docs/mvp-scope-and-limits.md
@@ -3,7 +3,7 @@
 ## In scope
 
 - deterministic batch discovery from trace files/directories
-- long-running runtime service that accepts OTLP/HTTP spans
+- long-running runtime service that accepts OTLP/HTTP spans, with optional OTLP/gRPC ingest
 - stable core model artifacts (`io.mb3r.bering.model`)
 - snapshot envelopes for observability/runtime consumers (`io.mb3r.bering.snapshot`)
 - generic discovery overlays for metadata and predicate references
diff --git a/docs/runtime-config.md b/docs/runtime-config.md
@@ -12,6 +12,14 @@ TCP listen address for the HTTP server.
 
 Example: `":4318"`
 
+### `server.grpc_listen_address`
+
+Optional TCP listen address for the OTLP/gRPC server.
+
+Example: `":4317"`
+
+Leave empty to disable OTLP/gRPC.
+
 ### `server.max_request_bytes`
 
 Maximum OTLP request body size in bytes after decompression.
@@ -88,6 +96,7 @@ overlays:
 
 - `--config`
 - `--listen`
+- `--grpc-listen`
 - `--flush-interval`
 - `--window-size`
 - `--max-in-memory-spans`
diff --git a/docs/trace-input-format.md b/docs/trace-input-format.md
@@ -1,6 +1,6 @@
 # Trace Input Format
 
-Bering supports two batch JSON formats and one runtime network ingest path.
+Bering supports two batch JSON formats and two runtime network ingest paths.
 
 ## 1) Normalized spans JSON
 
@@ -67,6 +67,15 @@ Supported request encodings:
 
 This is the primary integration path for any standard OpenTelemetry Collector or SDK exporter.
 
+## 4) Runtime OTLP/gRPC ingest
+
+`bering serve` optionally accepts OTLP/gRPC on `server.grpc_listen_address`.
+
+Supported request shape:
+
+- standard `opentelemetry.proto.collector.trace.v1.TraceService/Export`
+- protobuf OTLP payloads normalized through the same span conversion path as OTLP/HTTP
+
 ## Discovery-relevant attributes
 
 ### HTTP endpoint inference
diff --git a/examples/collector/otelcol.sidecar.yaml b/examples/collector/otelcol.sidecar.yaml
@@ -12,6 +12,11 @@ exporters:
     endpoint: http://bering:4318
     tls:
       insecure: true
+  # Alternative OTLP/gRPC exporter when Bering is configured with grpc_listen_address.
+  otlp/bering:
+    endpoint: bering:4317
+    tls:
+      insecure: true
 
 service:
   pipelines:
diff --git a/internal/app/app.go b/internal/app/app.go
@@ -233,6 +233,7 @@ func (r Runner) runServe(args []string) int {
 
 	configPath := fs.String("config", "", "Path to serve config YAML/JSON")
 	listen := fs.String("listen", "", "Override config server listen address")
+	grpcListen := fs.String("grpc-listen", "", "Override config server OTLP/gRPC listen address")
 	flushInterval := fs.String("flush-interval", "", "Override config runtime flush interval (e.g. 5s)")
 	windowSize := fs.String("window-size", "", "Override config runtime window size (e.g. 30s)")
 	maxInMemory := fs.Int("max-in-memory-spans", -1, "Override config runtime max in-memory spans")
@@ -256,6 +257,9 @@ func (r Runner) runServe(args []string) int {
 	if strings.TrimSpace(*listen) != "" {
 		cfg.Server.ListenAddress = strings.TrimSpace(*listen)
 	}
+	if strings.TrimSpace(*grpcListen) != "" {
+		cfg.Server.GRPCListenAddress = strings.TrimSpace(*grpcListen)
+	}
 	if strings.TrimSpace(*flushInterval) != "" {
 		parsed, err := time.ParseDuration(strings.TrimSpace(*flushInterval))
 		if err != nil {
@@ -320,7 +324,7 @@ func (r Runner) printUsage() {
 	fmt.Fprintln(r.stdout, "Usage:")
 	fmt.Fprintln(r.stdout, "  bering discover --input <trace-file|dir> [--out bering-model.json] [--snapshot-out bering-snapshot.json] [--replicas replicas.yaml|json] [--overlay overlay.yaml] [--discovered-at RFC3339]")
 	fmt.Fprintln(r.stdout, "  bering validate --input <bering-model.json|bering-snapshot.json>")
-	fmt.Fprintln(r.stdout, "  bering serve --config configs/serve.sample.yaml [--listen :8080] [--window-size 30s] [--flush-interval 5s]")
+	fmt.Fprintln(r.stdout, "  bering serve --config configs/serve.sample.yaml [--listen :8080] [--grpc-listen :4317] [--window-size 30s] [--flush-interval 5s]")
 }
 
 func (r Runner) printf(format string, args ...any) {
diff --git a/internal/app/app_test.go b/internal/app/app_test.go
@@ -46,7 +46,7 @@ func TestDiscoverAndValidate_NormalizedFixture(t *testing.T) {
 	if err != nil {
 		t.Fatalf("read expected output: %v", err)
 	}
-	if !bytes.Equal(actualRaw, expectedRaw) {
+	if !bytes.Equal(normalizeNewlines(actualRaw), normalizeNewlines(expectedRaw)) {
 		t.Fatalf("discover output mismatch\nactual:\n%s\nexpected:\n%s", actualRaw, expectedRaw)
 	}
 
@@ -96,7 +96,7 @@ func TestDiscoverAndValidate_OTLPFixture(t *testing.T) {
 	if err != nil {
 		t.Fatalf("read expected output: %v", err)
 	}
-	if !bytes.Equal(actualRaw, expectedRaw) {
+	if !bytes.Equal(normalizeNewlines(actualRaw), normalizeNewlines(expectedRaw)) {
 		t.Fatalf("discover output mismatch\nactual:\n%s\nexpected:\n%s", actualRaw, expectedRaw)
 	}
 
@@ -169,3 +169,7 @@ func repoRoot(t *testing.T) string {
 	}
 	return filepath.Clean(filepath.Join(filepath.Dir(thisFile), "..", ".."))
 }
+
+func normalizeNewlines(raw []byte) []byte {
+	return bytes.ReplaceAll(raw, []byte("\r\n"), []byte("\n"))
+}
diff --git a/internal/config/serve.go b/internal/config/serve.go
@@ -67,8 +67,9 @@ type ServeConfig struct {
 }
 
 type ServerConfig struct {
-	ListenAddress   string `json:"listen_address" yaml:"listen_address"`
-	MaxRequestBytes int64  `json:"max_request_bytes" yaml:"max_request_bytes"`
+	ListenAddress     string `json:"listen_address" yaml:"listen_address"`
+	GRPCListenAddress string `json:"grpc_listen_address" yaml:"grpc_listen_address"`
+	MaxRequestBytes   int64  `json:"max_request_bytes" yaml:"max_request_bytes"`
 }
 
 type RuntimeConfig struct {
@@ -90,8 +91,9 @@ type LoggingConfig struct {
 func DefaultServeConfig() ServeConfig {
 	return ServeConfig{
 		Server: ServerConfig{
-			ListenAddress:   ":8080",
-			MaxRequestBytes: 5 << 20,
+			ListenAddress:     ":8080",
+			GRPCListenAddress: "",
+			MaxRequestBytes:   5 << 20,
 		},
 		Runtime: RuntimeConfig{
 			FlushInterval:    Duration(5 * time.Second),
@@ -132,6 +134,9 @@ func (c ServeConfig) Validate() error {
 	if strings.TrimSpace(c.Server.ListenAddress) == "" {
 		return fmt.Errorf("server.listen_address cannot be empty")
 	}
+	if grpcAddress := strings.TrimSpace(c.Server.GRPCListenAddress); grpcAddress != "" && grpcAddress == strings.TrimSpace(c.Server.ListenAddress) {
+		return fmt.Errorf("server.grpc_listen_address must differ from server.listen_address")
+	}
 	if c.Server.MaxRequestBytes <= 0 {
 		return fmt.Errorf("server.max_request_bytes must be > 0")
 	}
diff --git a/internal/config/serve_test.go b/internal/config/serve_test.go
@@ -14,6 +14,7 @@ func TestLoadServeConfigYAML(t *testing.T) {
 	path := filepath.Join(dir, "serve.yaml")
 	raw := []byte(`server:
   listen_address: "127.0.0.1:4318"
+  grpc_listen_address: "127.0.0.1:4317"
   max_request_bytes: 1048576
 runtime:
   flush_interval: 2s
@@ -39,6 +40,9 @@ overlays:
 	if cfg.Server.ListenAddress != "127.0.0.1:4318" {
 		t.Fatalf("listen address mismatch: %s", cfg.Server.ListenAddress)
 	}
+	if cfg.Server.GRPCListenAddress != "127.0.0.1:4317" {
+		t.Fatalf("gRPC listen address mismatch: %s", cfg.Server.GRPCListenAddress)
+	}
 	if got, want := cfg.Runtime.FlushInterval.Duration(), 2*time.Second; got != want {
 		t.Fatalf("flush interval mismatch: got=%s want=%s", got, want)
 	}
@@ -65,3 +69,14 @@ func TestServeConfigValidateRejectsInvalidLatePolicy(t *testing.T) {
 		t.Fatal("expected validation error for invalid late policy")
 	}
 }
+
+func TestServeConfigValidateRejectsMatchingHTTPAndGRPCAddresses(t *testing.T) {
+	t.Parallel()
+
+	cfg := DefaultServeConfig()
+	cfg.Server.ListenAddress = "127.0.0.1:4318"
+	cfg.Server.GRPCListenAddress = "127.0.0.1:4318"
+	if err := cfg.Validate(); err == nil {
+		t.Fatal("expected validation error for matching HTTP and gRPC addresses")
+	}
+}
diff --git a/internal/connectors/otlp/http.go b/internal/connectors/otlp/http.go
@@ -18,7 +18,10 @@ import (
 	"github.com/MB3R-Lab/Bering/internal/connectors/traces"
 )
 
-const ConnectorName = "otlp_http"
+const (
+	HTTPConnectorName = "otlp_http"
+	GRPCConnectorName = "otlp_grpc"
+)
 
 func DecodeHTTPRequest(r *http.Request, maxBytes int64) ([]traces.Span, error) {
 	body, err := readBody(r, maxBytes)
@@ -39,10 +42,10 @@ func DecodePayload(contentType string, body []byte) ([]traces.Span, error) {
 			return nil, fmt.Errorf("decode otlp protobuf: %w", err)
 		}
 	}
-	return normalizeRequest(&req), nil
+	return NormalizeRequest(&req), nil
 }
 
-func normalizeRequest(req *collecttracev1.ExportTraceServiceRequest) []traces.Span {
+func NormalizeRequest(req *collecttracev1.ExportTraceServiceRequest) []traces.Span {
 	out := []traces.Span{}
 	for _, resourceSpans := range req.GetResourceSpans() {
 		resourceAttrs := resourceAttributes(resourceSpans.GetResource())
diff --git a/internal/runtime/engine.go b/internal/runtime/engine.go
diff --git a/internal/runtime/service.go b/internal/runtime/service.go
diff --git a/internal/runtime/service_test.go b/internal/runtime/service_test.go

Original file line number	Diff line number	Diff line change
`@@ -46,7 +46,7 @@ func TestDiscoverAndValidate_NormalizedFixture(t *testing.T) {`
`46`	`46`	`if err != nil {`
`47`	`47`	`t.Fatalf("read expected output: %v", err)`
`48`	`48`	`}`
`49`		`- if !bytes.Equal(actualRaw, expectedRaw) {`
	`49`	`+ if !bytes.Equal(normalizeNewlines(actualRaw), normalizeNewlines(expectedRaw)) {`
`50`	`50`	`t.Fatalf("discover output mismatch\nactual:\n%s\nexpected:\n%s", actualRaw, expectedRaw)`
`51`	`51`	`}`
`52`	`52`
`@@ -96,7 +96,7 @@ func TestDiscoverAndValidate_OTLPFixture(t *testing.T) {`
`96`	`96`	`if err != nil {`
`97`	`97`	`t.Fatalf("read expected output: %v", err)`
`98`	`98`	`}`
`99`		`- if !bytes.Equal(actualRaw, expectedRaw) {`
	`99`	`+ if !bytes.Equal(normalizeNewlines(actualRaw), normalizeNewlines(expectedRaw)) {`
`100`	`100`	`t.Fatalf("discover output mismatch\nactual:\n%s\nexpected:\n%s", actualRaw, expectedRaw)`
`101`	`101`	`}`
`102`	`102`
`@@ -169,3 +169,7 @@ func repoRoot(t *testing.T) string {`
`169`	`169`	`}`
`170`	`170`	`return filepath.Clean(filepath.Join(filepath.Dir(thisFile), "..", ".."))`
`171`	`171`	`}`
	`172`	`+`
	`173`	`+func normalizeNewlines(raw []byte) []byte {`
	`174`	`+ return bytes.ReplaceAll(raw, []byte("\r\n"), []byte("\n"))`
	`175`	`+}`