Internal logging configuration model, background thread(s)

rust/otap-dataflow/Cargo.toml

@@ -33,7 +33,10 @@ path = "src/main.rs"
 otap-df-config.workspace = true
 otap-df-controller.workspace = true
 otap-df-otap.workspace = true
+otap-df-telemetry.workspace = true
 thiserror.workspace = true
+tracing.workspace = true
+tracing-subscriber.workspace = true
 quiver = { workspace = true, optional = true }
 serde_json.workspace = true
 clap.workspace = true
@@ -147,6 +150,7 @@ sysinfo = "0.37"
 tempfile = "3"
 thiserror = "2.0.17"
 tracing = { version = ">=0.1.40", default-features = false }
+tracing-core = { version = ">=0.1.32", default-features = false }
 tracing-subscriber = { version = "0.3", default-features = false }
 tokio = { version = "1.48.0", features = ["rt", "time", "net", "io-util", "sync", "macros", "rt-multi-thread", "fs", "io-std", "process"] }
 tokio-stream = "0.1.17"

rust/otap-dataflow/configs/fake-batch-debug-noop.yaml

@@ -51,3 +51,12 @@ nodes:
     kind: exporter
     plugin_urn: "urn:otel:noop:exporter"
     config:
+
+service:
+  telemetry:
+    logs:
+      level: "debug"
+      providers:
+        global: immediate
+        engine: immediate
+      output: direct

rust/otap-dataflow/configs/internal-telemetry.yaml

@@ -0,0 +1,79 @@
+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:fake_data_generator:receiver"
+    out_ports:
+      out_port:
+        destinations:
+          - debug
+        dispatch_strategy: round_robin
+    config:
+      traffic_config:
+        max_signal_count: 100000
+        max_batch_size: 1000
+        signals_per_second: 1000
+        log_weight: 100
+      registry_path: https://github.com/open-telemetry/semantic-conventions.git[model]
+  debug:
+    kind: processor
+    plugin_urn: "urn:otel:debug:processor"
+    out_ports:
+      out_port:
+        destinations:
+          - noop
+        dispatch_strategy: round_robin
+    config:
+      verbosity: basic
+  noop:
+    kind: exporter
+    plugin_urn: "urn:otel:noop:exporter"
+    config: {}
+
+# Internal telemetry pipeline - separate from main pipeline
+# Uses hardcoded settings: single thread, no admin server
+internal:
+  telemetry:
+    kind: receiver
+    plugin_urn: "urn:otel:internal:otlp:receiver"
+    out_ports:
+      out_port:
+        destinations:
+          - batch
+        dispatch_strategy: round_robin
+    config: {}
+  batch:
+    kind: processor
+    plugin_urn: "urn:otel:batch:processor"
+    out_ports:
+      out_port:
+        destinations:
+          - console
+        dispatch_strategy: round_robin
+    config:
+      otap:
+        min_size: 1000
+        sizer: items
+      flush_timeout: 3s
+      format: preserve
+  console:
+    kind: exporter
+    plugin_urn: "urn:otel:console:exporter"
+    config: {}
+
+service:
+  telemetry:
+    logs:
+      level: "debug"
+      providers:
+        global: immediate
+        engine: immediate
+        internal: raw  # Avoid feedback in internal pipeline
+      output: internal
+    resource:
+      service.id: 1234
+      service.name: test

rust/otap-dataflow/crates/config/src/error.rs

@@ -118,6 +118,34 @@ pub enum Error {
         /// The id of the pipeline that was duplicated.
         pipeline_id: PipelineId,
     },
+
+    /// A receiver in the internal telemetry pipeline has an invalid plugin URN.
+    /// Only Internal Telemetry Receivers (ITR) are allowed in the internal pipeline.
+    #[error(
+        "Invalid receiver in internal pipeline: node `{node_id}` has plugin_urn `{plugin_urn}`, \
+         but only Internal Telemetry Receivers are allowed\nContext: {context}"
+    )]
+    #[diagnostic(code(data_plane::invalid_internal_receiver), url(docsrs))]
+    InvalidInternalReceiver {
+        /// The context in which the error occurred.
+        context: Context,
+        /// The node id of the invalid receiver.
+        node_id: NodeId,
+        /// The plugin URN of the invalid receiver.
+        plugin_urn: String,
+    },
+
+    /// The internal telemetry pipeline is required but not configured.
+    #[error(
+        "Internal telemetry pipeline required but not configured. \
+         When output mode is 'internal', the `internal` section must be present \
+         with at least one Internal Telemetry Receiver.\nContext: {context}"
+    )]
+    #[diagnostic(code(data_plane::missing_internal_pipeline), url(docsrs))]
+    MissingInternalPipeline {
+        /// The context in which the error occurred.
+        context: Context,
+    },
 }
 
 /// Information that all errors provide to help identify