feat(durable-buffer): Enhance durable_buffer observability with utilization and dropped/expired per-signal metrics

- Added a new storage_utilization Gauge (f64) metric to track persistent storage capacity.
- Added per-signal dropped/expired metrics (dropped_log_records, dropped_spans, dropped_metric_datapoints, expired_log_records, expired_spans, expired_metric_datapoints) utilizing Quiver's native per-slot atomic tracking.
- Appended new metrics to the end of DurableBufferMetrics to prevent index shifting.
- Renamed recompute_queued_counters to recompute_metrics to align with the expanded scope.
- Added changelog entry and unit tests verifying the metrics functionality.

Author: Dipanshusinghh

rust/otap-dataflow/.chloggen/arrow-encoder-durable-buffer-metrics.yaml

@@ -0,0 +1,31 @@
+# Use this template to create a new changelog entry.
+# Save this file to the `.chloggen/` directory in the repository root.
+# Example: `.chloggen/my-feature.yaml`
+
+# (Required) - Type of change. Allowed values are:
+#   breaking: A change that breaks backwards compatibility.
+#   deprecation: A change that deprecates a feature or component.
+#   new_component: A change that introduces a new component.
+#   enhancement: A change that introduces a new feature or enhancement.
+#   bug_fix: A change that fixes a bug.
+change_type: enhancement
+
+# (Required) - The component(s) affected by this change.
+# See the `config.yaml` file in this directory for a list of allowed components.
+component: pipeline
+
+# (Required) - A brief note explaining the change.
+# Example: "Add support for OTLP 1.0"
+note: Enhance durable_buffer observability with storage utilization and per-signal data loss metrics.
+
+# (Required) - Issue number(s) addressed by this change.
+# Example: [123, 456]
+issues: [3117]
+
+# (Optional) - User(s) who contributed to this change.
+# Omit if this change was authored by one of the project's core maintainers.
+# Note that this only applies to the top-level CHANGELOG, since we use
+# GitHub's auto-generated changelog for releases.
+# Example: [user1, user2]
+# subtext: "Thanks to @user1 and @user2 for this contribution!"
+subtext:

rust/otap-dataflow/crates/core-nodes/src/processors/durable_buffer_processor/bundle_adapter.rs

@@ -82,7 +82,7 @@ const fn to_slot_id(_signal_type: SignalType, payload_type: ArrowPayloadType) ->
 }
 
 /// Convert signal type to OTLP slot ID (for opaque binary storage)
-const fn to_otlp_slot_id(signal_type: SignalType) -> SlotId {
+pub(crate) const fn to_otlp_slot_id(signal_type: SignalType) -> SlotId {
     SlotId::new(match signal_type {
         SignalType::Logs => otlp_slots::OTLP_LOGS,
         SignalType::Traces => otlp_slots::OTLP_TRACES,
@@ -113,6 +113,11 @@ pub(crate) fn signal_type_from_slot_id(slot: SlotId) -> Option<SignalType> {
     from_slot_id(slot).map(|(st, _)| st)
 }
 
+pub(crate) const ARROW_LOGS_SLOT: SlotId = SlotId::new(30);
+pub(crate) const ARROW_TRACES_SLOT: SlotId = SlotId::new(40);
+pub(crate) const ARROW_METRICS_SLOT_1: SlotId = SlotId::new(10);
+pub(crate) const ARROW_METRICS_SLOT_2: SlotId = SlotId::new(11);
+
 /// Convert a slot ID back to payload type only (Arrow format only).
 ///
 /// Returns the `ArrowPayloadType` for the given slot, or `None` if the slot
@@ -1007,7 +1012,7 @@ mod tests {
         // HashMap iteration order is not guaranteed, but all slots should be from same signal
         #[rustfmt::skip]
         let records = OtapArrowRecords::Traces(traces!(
-            (Spans, 
+            (Spans,
                 ("id", UInt16, vec![0u16, 1])),
             (SpanEvents,
                 ("id", UInt32, vec![0u32, 1]),