chore: Migrate topics, internal_telemetry_receiver, and perf_exporter… (open-telemetry#2323)

… to core-nodes crate

# Change Summary

Next part of open-telemetry#1847 and open-telemetry#2086

Moves:

* perf_exporter
* topic_exporter
* internal_telemetry_receiver
* topic_receiver

## How are these changes tested?

* Unit tests / CI
* Compiled and ran `df_engine` and confirmed all nodes are still
available

## Are there any user-facing changes?

No

Author: drewrelmas

rust/otap-dataflow/README.md

@@ -256,12 +256,14 @@ ongoing refactor.
 
 Current status:
 
-- Exporters: `console_exporter`, `error_exporter`, `noop_exporter`
+- Exporters: `console_exporter`, `error_exporter`, `noop_exporter`,
+  `perf_exporter`, `topic_exporter`
 - Processors: `attributes_processor`, `batch_processor`,
   `content_router`, `debug_processor`, `delay_processor`,
   `durable_buffer_processor`, `fanout_processor`, `filter_processor`,
   `retry_processor`, `signal_type_router`, `transform_processor`
-- Receivers: `fake_data_generator`
+- Receivers: `fake_data_generator`, `internal_telemetry_receiver`,
+  `topic_receiver`
 
 ### Contrib Nodes
 

rust/otap-dataflow/benchmarks/benches/exporter/main.rs

@@ -8,6 +8,9 @@
 use criterion::{BenchmarkId, Criterion, criterion_group, criterion_main};
 use fluke_hpack::Encoder;
 use otap_df_channel::mpsc;
+use otap_df_core_nodes::exporters::perf_exporter::{
+    OTAP_PERF_EXPORTER_URN, PerfExporter, config::Config,
+};
 use otap_df_engine::{
     Interests,
     config::ExporterConfig,
@@ -19,7 +22,6 @@ use otap_df_engine::{
 use otap_df_otap::{
     otap_exporter::OTAPExporter,
     pdata::{Context, OtapPdata},
-    perf_exporter::{config::Config, exporter::PerfExporter},
 };
 use otap_df_pdata::{
     Consumer,
@@ -63,7 +65,6 @@ use otap_df_engine::context::ControllerContext;
 use otap_df_engine::control::{Controllable, NodeControlMsg, pipeline_ctrl_msg_channel};
 use otap_df_otap::otap_exporter::OTAP_EXPORTER_URN;
 use otap_df_otap::otlp_grpc::OTLPData;
-use otap_df_otap::perf_exporter::exporter::OTAP_PERF_EXPORTER_URN;
 use otap_df_telemetry::InternalTelemetrySystem;
 use serde_json::json;
 use std::pin::Pin;

rust/otap-dataflow/crates/core-nodes/Cargo.toml

@@ -33,6 +33,7 @@ bytes.workspace = true
 data_engine_expressions.workspace = true
 data_engine_kql_parser.workspace = true
 humantime-serde.workspace = true
+futures.workspace = true
 linkme.workspace = true
 prost.workspace = true
 rand.workspace = true

rust/otap-dataflow/crates/core-nodes/README.md

@@ -14,6 +14,8 @@ Each component lives in its own subfolder within a category:
         console_exporter/
         error_exporter/
         noop_exporter/
+        perf_exporter/
+        topic_exporter/
       processors/
         mod.rs (category exports)
         attributes_processor/
@@ -30,6 +32,8 @@ Each component lives in its own subfolder within a category:
       receivers/
         mod.rs (category exports)
         fake_data_generator/
+        internal_telemetry_receiver/
+        topic_receiver/
       lib.rs
 
 ## Components
@@ -41,6 +45,8 @@ Each component lives in its own subfolder within a category:
 | console_exporter | `urn:otel:exporter:console` | `src/exporters/console_exporter/` |
 | error_exporter | `urn:otel:exporter:error` | `src/exporters/error_exporter/` |
 | noop_exporter | `urn:otel:exporter:noop` | `src/exporters/noop_exporter/` |
+| perf_exporter | `urn:otel:exporter:perf` | `src/exporters/perf_exporter/` |
+| topic_exporter | `urn:otel:exporter:topic` | `src/exporters/topic_exporter/` |
 
 #### console_exporter
 
@@ -57,6 +63,16 @@ Each component lives in its own subfolder within a category:
 - Placeholder exporter that ACKs all messages without processing
 - Lightweight for performance testing and pipeline validation
 
+#### perf_exporter
+
+- Measures item throughput by signal type for benchmarking scenarios
+- Emits pdata-oriented telemetry metrics during pipeline execution
+
+#### topic_exporter
+
+- Publishes pdata into configured runtime topics
+- Supports tracked end-to-end ack/nack propagation through topic boundaries
+
 ### Processors
 
 | Node | URN | Module |
@@ -140,9 +156,21 @@ Each component lives in its own subfolder within a category:
 | Node | URN | Module |
 | ---- | --- | ------ |
 | fake_data_generator | `urn:otel:receiver:traffic_generator` | `src/receivers/fake_data_generator/` |
+| internal_telemetry_receiver | `urn:otel:receiver:internal_telemetry` | `src/receivers/internal_telemetry_receiver/` |
+| topic_receiver | `urn:otel:receiver:topic` | `src/receivers/topic_receiver/` |
 
 #### fake_data_generator
 
 - Generates synthetic OTAP/OTLP signals for testing and benchmarking
 - Configurable signal generation strategies and volume constraints
 - Includes support for pregenerated, dynamic, and rate-based signal generation
+
+#### internal_telemetry_receiver
+
+- Receives internal engine telemetry events from the internal log channel
+- Emits them as OTLP log pdata into the configured pipeline
+
+#### topic_receiver
+
+- Subscribes to runtime topics and forwards messages into the pipeline
+- Supports broadcast/balanced subscription modes and topic ack/nack bridging

rust/otap-dataflow/crates/core-nodes/src/exporters/mod.rs

@@ -9,3 +9,9 @@ pub mod error_exporter;
 
 /// Console exporter.
 pub mod console_exporter;
+
+/// Topic exporter.
+pub mod topic_exporter;
+
+/// Perf exporter.
+pub mod perf_exporter;

…/crates/otap/src/perf_exporter/config.rs …es/src/exporters/perf_exporter/config.rsrust/otap-dataflow/crates/otap/src/perf_exporter/config.rs renamed to rust/otap-dataflow/crates/core-nodes/src/exporters/perf_exporter/config.rs rust/otap-dataflow/crates/otap/src/perf_exporter/config.rs renamed to rust/otap-dataflow/crates/core-nodes/src/exporters/perf_exporter/config.rs

@@ -3,30 +3,32 @@
 
 //! Exporter used to measure the performance of the OTAP data pipeline.
 //!
-//! ToDo – Future developments / improvements:
+//! ToDo - Future developments / improvements:
 //! - Replace this exporter with a processor that could be combined with a Noop exporter to achieve
 //!   the same functionality. The advantage would be to allow performance measurements anywhere in
 //!   the pipeline.
 //! - Measure the number of memory allocations for the current thread. This would allow measuring
 //!   the memory used by the pipeline. This is possible using `mimalloc-sys`.
-//! - Measure per-thread CPU usage. This would allow measuring the pipeline’s CPU load. This is
-//!   possible using the "libc" crate function `getrusage(RUSAGE\_THREAD)`.
+//! - Measure per-thread CPU usage. This would allow measuring the pipeline's CPU load. This is
+//!   possible using the "libc" crate function `getrusage(RUSAGE_THREAD)`.
 //! - Measure network usage either via a cgroup or via eBPF.
-//! - Measure per-thread perf counters (see crates perfcnt, perfcnt2, or direct perf\_event\_open
+//! - Measure per-thread perf counters (see crates perfcnt, perfcnt2, or direct perf_event_open
 //!   via nix/libc). We could measure task-clock, context switches, page faults, ...
 //! - Measure the latency of signals traversing the pipeline. This would require adding a timestamp
 //!   in the headers of pdata messages.
 //! - Support live reconfiguration via control message.
 
-use crate::OTAP_EXPORTER_FACTORIES;
-use crate::metrics::ExporterPDataMetrics;
-use crate::pdata::OtapPdata;
-use crate::perf_exporter::config::Config;
-use crate::perf_exporter::metrics::PerfExporterPdataMetrics;
+pub mod config;
+pub mod metrics;
+
+use crate::exporters::perf_exporter::config::Config;
+use crate::exporters::perf_exporter::metrics::PerfExporterPdataMetrics;
 use async_trait::async_trait;
+use linkme::distributed_slice;
 use otap_df_config::SignalType;
 use otap_df_config::node::NodeUserConfig;
 use otap_df_engine::ConsumerEffectHandlerExtension;
+use otap_df_engine::ExporterFactory;
 use otap_df_engine::config::ExporterConfig;
 use otap_df_engine::context::PipelineContext;
 use otap_df_engine::control::{AckMsg, NodeControlMsg};
@@ -36,7 +38,9 @@ use otap_df_engine::local::exporter as local;
 use otap_df_engine::message::{Message, MessageChannel};
 use otap_df_engine::node::NodeId;
 use otap_df_engine::terminal_state::TerminalState;
-use otap_df_engine::{ExporterFactory, distributed_slice};
+use otap_df_otap::OTAP_EXPORTER_FACTORIES;
+use otap_df_otap::metrics::ExporterPDataMetrics;
+use otap_df_otap::pdata::OtapPdata;
 use otap_df_pdata::otap::OtapArrowRecords;
 use otap_df_telemetry::metrics::{MetricSet, MetricSetHandler};
 use otap_df_telemetry::otel_info;
@@ -267,22 +271,17 @@ impl local::Exporter<OtapPdata> for PerfExporter {
 
 #[cfg(test)]
 mod tests {
-    use crate::fixtures::{
-        SimpleDataGenOptions, create_simple_logs_arrow_record_batches,
-        create_simple_metrics_arrow_record_batches, create_simple_trace_arrow_record_batches,
-    };
-    use crate::pdata::OtapPdata;
-    use crate::perf_exporter::config::Config;
-    use crate::perf_exporter::exporter::{OTAP_PERF_EXPORTER_URN, PerfExporter};
+    use super::{OTAP_PERF_EXPORTER_URN, PerfExporter};
+    use crate::exporters::perf_exporter::config::Config;
     use otap_df_config::node::NodeUserConfig;
     use otap_df_engine::context::ControllerContext;
     use otap_df_engine::error::Error;
     use otap_df_engine::exporter::ExporterWrapper;
     use otap_df_engine::testing::exporter::TestContext;
     use otap_df_engine::testing::exporter::TestRuntime;
     use otap_df_engine::testing::test_node;
-    use otap_df_pdata::Consumer;
-    use otap_df_pdata::otap::{OtapArrowRecords, from_record_messages};
+    use otap_df_otap::pdata::OtapPdata;
+    use otap_df_otap::testing::create_test_pdata;
     use otap_df_telemetry::registry::TelemetryRegistryHandle;
     use std::future::Future;
     use std::ops::Add;
@@ -297,59 +296,10 @@ mod tests {
     -> impl FnOnce(TestContext<OtapPdata>) -> std::pin::Pin<Box<dyn Future<Output = ()>>> {
         |ctx| {
             Box::pin(async move {
-                // send some messages to the exporter to calculate pipeline statistics
-                for i in 0..3 {
-                    let mut traces_batch_data =
-                        create_simple_trace_arrow_record_batches(SimpleDataGenOptions {
-                            id_offset: 3 * i,
-                            num_rows: 5,
-                            ..Default::default()
-                        });
-                    let mut logs_batch_data =
-                        create_simple_logs_arrow_record_batches(SimpleDataGenOptions {
-                            id_offset: 3 * i + 1,
-                            num_rows: 5,
-                            ..Default::default()
-                        });
-                    let mut metrics_batch_data =
-                        create_simple_metrics_arrow_record_batches(SimpleDataGenOptions {
-                            id_offset: 3 * i + 2,
-                            num_rows: 5,
-                            ..Default::default()
-                        });
-
-                    let trace_batch_data = from_record_messages(
-                        Consumer::default()
-                            .consume_bar(&mut traces_batch_data)
-                            .unwrap(),
-                    );
-                    let logs_batch_data = from_record_messages(
-                        Consumer::default()
-                            .consume_bar(&mut logs_batch_data)
-                            .unwrap(),
-                    );
-                    let metrics_batch_data = from_record_messages(
-                        Consumer::default()
-                            .consume_bar(&mut metrics_batch_data)
-                            .unwrap(),
-                    );
-
-                    // Send a data message
-                    ctx.send_pdata(OtapPdata::new_default(
-                        OtapArrowRecords::Traces(trace_batch_data).into(),
-                    ))
-                    .await
-                    .expect("Failed to send data message");
-                    ctx.send_pdata(OtapPdata::new_default(
-                        OtapArrowRecords::Logs(logs_batch_data).into(),
-                    ))
-                    .await
-                    .expect("Failed to send data message");
-                    ctx.send_pdata(OtapPdata::new_default(
-                        OtapArrowRecords::Metrics(metrics_batch_data).into(),
-                    ))
-                    .await
-                    .expect("Failed to send data message");
+                for _ in 0..3 {
+                    ctx.send_pdata(create_test_pdata())
+                        .await
+                        .expect("Failed to send data message");
                 }
 
                 // TODO ADD DELAY BETWEEN HERE

…crates/otap/src/perf_exporter/metrics.rs …s/src/exporters/perf_exporter/metrics.rsrust/otap-dataflow/crates/otap/src/perf_exporter/metrics.rs renamed to rust/otap-dataflow/crates/core-nodes/src/exporters/perf_exporter/metrics.rs rust/otap-dataflow/crates/otap/src/perf_exporter/metrics.rs renamed to rust/otap-dataflow/crates/core-nodes/src/exporters/perf_exporter/metrics.rs

@@ -3,8 +3,6 @@
 
 //! Topic exporter.
 
-use crate::OTAP_EXPORTER_FACTORIES;
-use crate::pdata::OtapPdata;
 use async_trait::async_trait;
 use futures::stream::{FuturesUnordered, StreamExt};
 use linkme::distributed_slice;
@@ -25,6 +23,8 @@ use otap_df_engine::topic::{
     PublishOutcome, TopicHandle, TrackedPublishOutcome, TrackedTryPublishOutcome,
 };
 use otap_df_engine::{ConsumerEffectHandlerExtension, ExporterFactory};
+use otap_df_otap::OTAP_EXPORTER_FACTORIES;
+use otap_df_otap::pdata::OtapPdata;
 use otap_df_telemetry::instrument::{Counter, Gauge};
 use otap_df_telemetry::metrics::MetricSet;
 use otap_df_telemetry::{otel_info, otel_warn};
@@ -363,8 +363,6 @@ impl Exporter<OtapPdata> for TopicExporter {
 #[cfg(test)]
 mod tests {
     use super::{TOPIC_EXPORTER, TOPIC_EXPORTER_URN, TopicExporter};
-    use crate::pdata::OtapPdata;
-    use crate::testing::{TestCallData, create_test_pdata, next_ack};
     use otap_df_config::node::NodeUserConfig;
     use otap_df_config::topic::{TopicAckPropagationMode, TopicQueueOnFullPolicy};
     use otap_df_engine::Interests;
@@ -381,6 +379,8 @@ mod tests {
         PipelineTopicBinding, SubscriberOptions, SubscriptionMode, TopicBroadcastOnLagPolicy,
         TopicBroker, TopicOptions, TopicSet,
     };
+    use otap_df_otap::pdata::OtapPdata;
+    use otap_df_otap::testing::{TestCallData, create_test_pdata, next_ack};
     use otap_df_telemetry::reporter::MetricsReporter;
     use serde_json::json;
     use std::sync::Arc;

…rates/otap/src/perf_exporter/exporter.rs …nodes/src/exporters/perf_exporter/mod.rsrust/otap-dataflow/crates/otap/src/perf_exporter/exporter.rs renamed to rust/otap-dataflow/crates/core-nodes/src/exporters/perf_exporter/mod.rs rust/otap-dataflow/crates/otap/src/perf_exporter/exporter.rs renamed to rust/otap-dataflow/crates/core-nodes/src/exporters/perf_exporter/mod.rs

@@ -6,8 +6,6 @@
 //! This receiver consumes internal logs from the logging channel and emits
 //! the logs as OTLP ExportLogsRequest messages into the pipeline.
 
-use crate::OTAP_RECEIVER_FACTORIES;
-use crate::pdata::{Context, OtapPdata};
 use async_trait::async_trait;
 use bytes::Bytes;
 use linkme::distributed_slice;
@@ -21,6 +19,8 @@ use otap_df_engine::local::receiver as local;
 use otap_df_engine::node::NodeId;
 use otap_df_engine::receiver::ReceiverWrapper;
 use otap_df_engine::terminal_state::TerminalState;
+use otap_df_otap::OTAP_RECEIVER_FACTORIES;
+use otap_df_otap::pdata::{Context, OtapPdata};
 use otap_df_pdata::OtlpProtoBytes;
 use otap_df_pdata::otlp::ProtoBuffer;
 use otap_df_telemetry::event::{LogEvent, ObservedEvent};