Support multiple groups and pipelines

rust/otap-dataflow/README.md

@@ -22,6 +22,20 @@ data.
 > provided as a means to test and validate OTAP pipelines built using
 > the dataflow engine.
 
+## Architecture
+
+![OTAP Dataflow Engine architecture](docs/images/architecture-high-level.svg)
+
+The controller is the local control plane for pipeline groups. It allocates CPU
+cores, spawns one worker thread per core, and owns lifecycle, coordination, and
+runtime observability. Each pipeline runs a single-threaded engine instance per
+assigned core, hot data paths stay within that thread, while cross-thread
+coordination is handled through control messages and internal telemetry.
+
+The admin HTTP server and observed-state store are driven by the controller for
+runtime visibility and control. For details, see the controller and engine crate
+READMEs.
+
 ## Features
 
 The OTAP Dataflow engine consists of a number of major pieces. Here
@@ -71,9 +85,12 @@ the many N-to-1 relationships expressed within an OTAP request.
 
 ## Major components
 
-### Engine
+### Controller and Engine
+
+See the controller and engine crate READMEs:
 
-[See crate README.](./crates/engine/README.md)
+- [controller](./crates/controller/README.md).
+- [engine](./crates/engine/README.md),
 
 The `otap_df_engine` crate is located in `crates/engine`, here we
 find the engine's overall architecture expressed:
@@ -105,7 +122,7 @@ crates/engine/lib.rs:    Effect handler extensions, pipeline factory
 |-- runtime_pipeline.rs: Builds the graph of component channels
 ```
 
-### OTAP: OTel-Arrow Protocol pipline data
+### OTAP: OTel-Arrow Protocol pipeline data
 
 [See crate README.](./crates/otap/README.md)
 
@@ -234,10 +251,10 @@ establish the performance of the OTAP Dataflow system.
 
 [See crate README.](./crates/controller/README.md)
 
-The `otap_df_controller` crate is located in `crates/controller` is
+The `otap_df_controller` crate is located in `crates/controller` and is
 the main entry point to construct an OTAP Dataflow pipeline instance. The
-controller type, `otap_df_controller::Controller<PData>`, manages building
-and running one or more pipelines.
+controller type, `otap_df_controller::Controller<PData>`, manages building,
+running, and supervising one or more pipelines.
 
 This component is responsible for making the assignment between OTAP
 dataflow pipeline and individually-numbered CPU instances. The

rust/otap-dataflow/configs/engine-conf/continuous_benchmark.yaml

@@ -0,0 +1,168 @@
+# This configuration file reproduces the continuous benchmarking setup used
+# in our CI pipelines. However, the systems: traffic generator, system under
+# test, and backend are all included in a single configuration for easier
+# local testing and debugging.
+
+# Engine-wide settings
+settings:
+  http_admin:
+    bind_address: 127.0.0.1:8085
+
+# Pipeline groups are used to logically separate different sets of
+# pipelines. We will introduce progressively more features around
+# pipeline groups in the future such as resource quotas, scheduling
+# policies, named channels, group life-cycle management, ...
+pipeline_groups:
+  # This group contains all the pipelines required for continuous benchmarking
+  continuous_benchmark:
+    pipelines:
+      # ======================================================================
+      # Pipeline generating traffic
+      # ======================================================================
+      traffic_gen:
+        quota:
+          core_allocation:
+            type: core_set
+            set:
+              - type: CoreRange
+                start: 10
+                end: 35
+
+        nodes:
+          receiver:
+            kind: receiver
+            plugin_urn: "urn:otel:otap:fake_data_generator:receiver"
+            out_ports:
+              out_port:
+                destinations:
+                  - exporter
+                dispatch_strategy: round_robin
+            config:
+              traffic_config:
+                signals_per_second: 100000
+                max_signal_count: null
+                metric_weight: 0
+                trace_weight: 0
+                log_weight: 30
+              registry_path: https://github.com/open-telemetry/semantic-conventions.git[model]
+          exporter:
+            kind: exporter
+            plugin_urn: "urn:otel:otlp:exporter"
+            config:
+              grpc_endpoint: "http://127.0.0.1:4327"
+
+      # ======================================================================
+      # System Under Test pipeline
+      # ======================================================================
+      sut:
+        quota:
+          core_allocation:
+            type: core_set
+            set:
+              - type: CoreRange
+                start: 0
+                end: 0
+        nodes:
+          otlp_recv:
+            kind: receiver
+            plugin_urn: "urn:otel:otlp:receiver"
+            out_ports:
+              out_port:
+                destinations:
+                  - router
+                dispatch_strategy: round_robin
+            config:
+              listening_addr: "127.0.0.1:4327"
+              wait_for_result: true
+
+          otap_recv:
+            kind: receiver
+            plugin_urn: "urn:otel:otap:receiver"
+            out_ports:
+              out_port:
+                destinations:
+                  - otap_exporter
+                dispatch_strategy: round_robin
+            config:
+              listening_addr: "127.0.0.1:4329"
+              response_stream_channel_size: 256
+
+          router:
+            kind: processor
+            plugin_urn: "urn:otap:processor:signal_type_router"
+            out_ports:
+              logs:
+                destinations:
+                  - retry
+                dispatch_strategy: round_robin
+              metrics:
+                destinations:
+                  - metrics_exporter
+                dispatch_strategy: round_robin
+              traces:
+                destinations:
+                  - spans_exporter
+                dispatch_strategy: round_robin
+            config: {}
+
+          retry:
+            kind: processor
+            plugin_urn: "urn:otel:retry:processor"
+            out_ports:
+              out_port:
+                destinations:
+                  - logs_exporter
+                dispatch_strategy: round_robin
+            config:
+              multiplier: 1.5
+
+          logs_exporter:
+            kind: exporter
+            plugin_urn: "urn:otel:otlp:exporter"
+            config:
+              grpc_endpoint: "http://127.0.0.1:4328"
+              max_in_flight: 6
+
+          metrics_exporter:
+            kind: exporter
+            plugin_urn: "urn:otel:noop:exporter"
+            config:
+
+          spans_exporter:
+            kind: exporter
+            plugin_urn: "urn:otel:noop:exporter"
+            config:
+
+          otap_exporter:
+            kind: exporter
+            plugin_urn: "urn:otel:noop:exporter"
+            config:
+
+      # ======================================================================
+      # Performance No-Op pipeline to measure receiver performance
+      # ======================================================================
+      backend:
+        quota:
+          core_allocation:
+            type: core_set
+            set:
+              - type: CoreRange
+                start: 1
+                end: 1
+
+        nodes:
+          receiver:
+            kind: receiver
+            plugin_urn: urn:otel:otlp:receiver
+            out_ports:
+              out_port:
+                destinations:
+                  - perf_noop
+                dispatch_strategy: round_robin
+            config:
+              listening_addr: 127.0.0.1:4328
+
+          perf_noop:
+            kind: exporter
+            plugin_urn: urn:otel:noop:exporter
+            config: null

rust/otap-dataflow/configs/fake-otap.yaml

@@ -14,13 +14,14 @@ nodes:
         dispatch_strategy: round_robin
     config:
       traffic_config:
-        max_signal_count: 1000
-        max_batch_size: 1000
-        signals_per_second: 1000
-        log_weight: 100
+        signals_per_second: 100000
+        max_signal_count: null
+        metric_weight: 0
+        trace_weight: 0
+        log_weight: 30
       registry_path: https://github.com/open-telemetry/semantic-conventions.git[model]
   exporter:
     kind: exporter
     plugin_urn: "urn:otel:otap:exporter"
     config:
-      grpc_endpoint: "http://127.0.0.1:4318"
+      grpc_endpoint: "http://127.0.0.1:4327"

rust/otap-dataflow/configs/fake-otlp.yaml

@@ -24,4 +24,4 @@ nodes:
     kind: exporter
     plugin_urn: "urn:otel:otlp:exporter"
     config:
-      grpc_endpoint: "http://127.0.0.1:4317"
+      grpc_endpoint: "http://127.0.0.1:4327"

rust/otap-dataflow/configs/otap-noop.yaml

@@ -0,0 +1,22 @@
+settings:
+  default_pipeline_ctrl_msg_channel_size: 100
+  default_node_ctrl_msg_channel_size: 100
+  default_pdata_channel_size: 100
+
+nodes:
+  receiver:
+    kind: receiver
+    plugin_urn: "urn:otel:otap:receiver"
+    out_ports:
+      out_port:
+        destinations:
+          - perf_noop
+        dispatch_strategy: round_robin
+    config:
+      listening_addr: "127.0.0.1:4328"
+      response_stream_channel_size: 256
+
+  perf_noop:
+    kind: exporter
+    plugin_urn: "urn:otel:noop:exporter"
+    config:

rust/otap-dataflow/configs/otap-route-otap.yaml

@@ -0,0 +1,98 @@
+settings:
+  default_pipeline_ctrl_msg_channel_size: 100
+  default_node_ctrl_msg_channel_size: 100
+  default_pdata_channel_size: 100
+
+nodes:
+  otlp_recv:
+    kind: receiver
+    plugin_urn: "urn:otel:otap:receiver"
+    out_ports:
+      out_port:
+        destinations:
+          - router
+        dispatch_strategy: round_robin
+    config:
+      listening_addr: "127.0.0.1:4327"
+      wait_for_result: true
+      response_stream_channel_size: 256 # Required: channel buffer capacity (number of messages)
+
+  otap_recv:
+    kind: receiver
+    plugin_urn: "urn:otel:otap:receiver"
+    out_ports:
+      out_port:
+        destinations:
+          - otap_exporter
+        dispatch_strategy: round_robin
+    config:
+      listening_addr: "127.0.0.1:4329"
+      response_stream_channel_size: 256
+
+  router:
+    kind: processor
+    plugin_urn: "urn:otap:processor:signal_type_router"
+    out_ports:
+      logs:
+        destinations:
+          - retry
+        dispatch_strategy: round_robin
+      metrics:
+        destinations:
+          - metrics_exporter
+        dispatch_strategy: round_robin
+      traces:
+        destinations:
+          - spans_exporter
+        dispatch_strategy: round_robin
+    config: {}
+
+#  logs_filter:
+#    kind: processor
+#    plugin_urn: "urn:otel:filter:processor"
+#    out_ports:
+#      out_port:
+#        destinations:
+#          - logs_exporter
+#        dispatch_strategy: round_robin
+#    config:
+#      logs:
+#        include:
+#          match_type: strict
+#          record_attributes:
+#            - key: gen_ai.system
+#              value: openai
+#            - key: ios.app.state
+#              value: active
+
+  retry:
+    kind: processor
+    plugin_urn: "urn:otel:retry:processor"
+    out_ports:
+      out_port:
+        destinations:
+          - logs_exporter
+        dispatch_strategy: round_robin
+    config:
+      multiplier: 1.5
+
+  logs_exporter:
+    kind: exporter
+    plugin_urn: "urn:otel:otap:exporter"
+    config:
+      grpc_endpoint: "http://127.0.0.1:4328"
+
+  metrics_exporter:
+    kind: exporter
+    plugin_urn: "urn:otel:noop:exporter"
+    config:
+
+  spans_exporter:
+    kind: exporter
+    plugin_urn: "urn:otel:noop:exporter"
+    config:
+
+  otap_exporter:
+    kind: exporter
+    plugin_urn: "urn:otel:noop:exporter"
+    config: