feat: Add Configurable Nested KeyValue Support for ClickHouse JSON Export

.github/workflows/processor-release.yml

@@ -12,7 +12,6 @@ jobs:
       fail-fast: false
       matrix:
         include:
-
           - target: x86_64-unknown-linux-gnu
             runner: ubuntu-22.04
             os: ubuntu
@@ -29,6 +28,10 @@ jobs:
             runner: ubuntu-22.04
             os: ubuntu
             python-version: 3.13
+          - target: x86_64-unknown-linux-gnu
+            runner: ubuntu-22.04
+            os: ubuntu
+            python-version: 3.14
           - target: aarch64-unknown-linux-gnu
             runner: ubuntu-22.04-arm
             os: ubuntu
@@ -45,6 +48,10 @@ jobs:
             runner: ubuntu-22.04-arm
             os: ubuntu
             python-version: 3.13
+          - target: aarch64-unknown-linux-gnu
+            runner: ubuntu-22.04-arm
+            os: ubuntu
+            python-version: 3.14
           - target: aarch64-apple-darwin
             runner: macos-14
             os: macos
@@ -61,6 +68,10 @@ jobs:
             runner: macos-14
             os: macos
             python-version: 3.13
+          - target: aarch64-apple-darwin
+            runner: macos-14
+            os: macos
+            python-version: 3.14
     env:
       TARGET: ${{ matrix.target }}
     steps:
@@ -74,26 +85,10 @@ jobs:
       - name: install protoc macos
         if: matrix.os == 'macos'
         run: brew install protobuf
-      - name: update python 3.10
-        if: matrix.python-version == '3.10'
-        uses: actions/setup-python@v5
-        with:
-          python-version: "3.10"
-      - name: update python 3.11
-        if: matrix.python-version == '3.11'
-        uses: actions/setup-python@v5
-        with:
-          python-version: 3.11
-      - name: update python 3.12
-        if: matrix.python-version == '3.12'
-        uses: actions/setup-python@v5
-        with:
-          python-version: 3.12
-      - name: update python 3.13
-        if: matrix.python-version == '3.13'
+      - name: update python ${{ matrix.python-version }}
         uses: actions/setup-python@v5
         with:
-          python-version: 3.13
+          python-version: ${{ matrix.python-version }}
       - name: Set build env
         run: echo "BUILD_SHORT_SHA=$(echo -n $GITHUB_SHA | cut -c 1-7)" >> $GITHUB_ENV
       - uses: actions-rust-lang/setup-rust-toolchain@v1
@@ -112,7 +107,7 @@ jobs:
           command: build
           args: --release --features pyo3 --locked --target ${{ matrix.target }}
       - name: upload build
-        if: matrix.python-version == '3.13'
+        if: matrix.python-version == '3.14'
         uses: actions/upload-artifact@v4
         with:
           name: rotel-processor-build-${{ matrix.target }}

Cargo.toml

@@ -1,6 +1,6 @@
 [package]
 name = "rotel"
-version = "0.1.6"
+version = "0.1.7"
 edition = "2024"
 homepage = "https://github.com/streamfold/rotel"
 readme = "README.md"
@@ -64,7 +64,7 @@ libc = "0.2.170"
 pin-project = "1.1.10"
 futures-util = "0.3.31"
 rotel_python_processor_sdk = { path = "rotel_python_processor_sdk", optional = true }
-pyo3 = { version = "0.24.1", optional = true }
+pyo3 = { version = "0.27.2", optional = true }
 chrono = { version = "0.4.40", features = ["serde"] }
 serde = { version = "1.0.217", features = ["derive"] }
 thiserror = "2.0.12"

README.md

@@ -232,6 +232,7 @@ and traces.
 | --clickhouse-exporter-async-insert           | true                           | true, false |
 | --clickhouse-exporter-request-timeout        | 5s                             |             |
 | --clickhouse-exporter-enable-json            |                                |             |
+| --clickhouse-exporter-nested-kv-max-depth    |                                |             |
 | --clickhouse-exporter-json-underscore        |                                |             |
 | --clickhouse-exporter-user                   |                                |             |
 | --clickhouse-exporter-password               |                                |             |
@@ -263,6 +264,18 @@ a nested JSON object. You can replace periods in JSON keys with underscores by p
 `--clickhouse-exporter-json-underscore` which will keep the JSON keys flat. For example, the resource attribute
 `service.name` will be inserted as `service_name`.
 
+When exporting OpenTelemetry attributes that contain nested `KeyValueList` structures (such as GenAI message
+attributes like `gen_ai.input.messages`), use `--clickhouse-exporter-nested-kv-max-depth` to convert them to
+proper JSON objects. Without this option, nested structures are serialized as JSON strings (the protobuf
+representation) for backwards compatibility. Set to a value like `10` to enable nested conversion:
+
+```shell
+rotel start --exporter clickhouse \
+  --clickhouse-exporter-endpoint "http://localhost:8123" \
+  --clickhouse-exporter-enable-json \
+  --clickhouse-exporter-nested-kv-max-depth 10
+```
+
 _The ClickHouse exporter is built using code from the official Rust [clickhouse-rs](https://crates.io/crates/clickhouse)
 crate._
 

rotel_python_processor_sdk/Cargo.toml

@@ -4,7 +4,7 @@ version = "0.1.0"
 edition = "2021"
 
 [dependencies]
-pyo3 = { version = "0.24.1" }
+pyo3 = { version = "0.27.2" }
 opentelemetry-proto = { version = "0.30.0" }
 tower = "0.5.2"
 utilities = { path = "../utilities" }

rotel_python_processor_sdk/src/model/common.rs

@@ -1,7 +1,6 @@
 use crate::py::common::*;
 use opentelemetry_proto::tonic::common::v1::KeyValue;
-#[allow(deprecated)]
-use pyo3::{IntoPy, PyObject, PyResult, Python};
+use pyo3::{IntoPyObjectExt, Py, PyAny, PyResult, Python};
 use std::sync::{Arc, Mutex};
 
 #[derive(Debug, Clone)]
@@ -26,10 +25,9 @@ pub struct RArrayValue {
     pub values: Arc<Mutex<Vec<Arc<Mutex<Option<RAnyValue>>>>>>,
 }
 
-#[allow(deprecated)]
 impl RArrayValue {
-    pub(crate) fn convert_to_py(&self, py: Python) -> PyResult<PyObject> {
-        Ok(ArrayValue(self.values.clone()).into_py(py))
+    pub(crate) fn convert_to_py(&self, py: Python) -> PyResult<Py<PyAny>> {
+        Ok(ArrayValue(self.values.clone()).into_py_any(py)?)
     }
 }
 
@@ -38,10 +36,9 @@ pub struct RKeyValueList {
     pub values: Arc<Mutex<Vec<RKeyValue>>>,
 }
 
-#[allow(deprecated)]
 impl RKeyValueList {
-    pub(crate) fn convert_to_py(&self, py: Python) -> PyResult<PyObject> {
-        Ok(KeyValueList(self.values.clone()).into_py(py))
+    pub(crate) fn convert_to_py(&self, py: Python) -> PyResult<Py<PyAny>> {
+        Ok(KeyValueList(self.values.clone()).into_py_any(py)?)
     }
 }
 

rotel_python_processor_sdk/src/model/mod.rs

@@ -22,10 +22,10 @@ static PROCESSOR_INIT: Once = Once::new();
 pub fn register_processor(code: String, script: String, module: String) -> Result<(), BoxError> {
     PROCESSOR_INIT.call_once(|| {
         pyo3::append_to_inittab!(rotel_sdk);
-        pyo3::prepare_freethreaded_python();
+        Python::initialize();
     });
 
-    let res = Python::with_gil(|py| -> PyResult<()> {
+    let res = Python::attach(|py| -> PyResult<()> {
         PyModule::from_code(
             py,
             CString::new(code)?.as_c_str(),
@@ -48,7 +48,7 @@ impl PythonProcessable for opentelemetry_proto::tonic::trace::v1::ResourceSpans
     fn process(self, processor: &str, request_context: Option<RequestContext>) -> Self {
         let inner = otel_transform::transform_resource_spans(self);
         // Build the PyObject
-        let res = Python::with_gil(|py| -> PyResult<()> {
+        let res = Python::attach(|py| -> PyResult<()> {
             let spans = ResourceSpans {
                 resource: inner.resource.clone(),
                 scope_spans: inner.scope_spans.clone(),
@@ -101,7 +101,7 @@ impl PythonProcessable for opentelemetry_proto::tonic::metrics::v1::ResourceMetr
     fn process(self, processor: &str, request_context: Option<RequestContext>) -> Self {
         let inner = otel_transform::transform_resource_metrics(self);
         // Build the PyObject
-        let res = Python::with_gil(|py| -> PyResult<()> {
+        let res = Python::attach(|py| -> PyResult<()> {
             let spans = ResourceMetrics {
                 resource: inner.resource.clone(),
                 scope_metrics: inner.scope_metrics.clone(),
@@ -142,7 +142,7 @@ impl PythonProcessable for opentelemetry_proto::tonic::logs::v1::ResourceLogs {
     fn process(self, processor: &str, request_context: Option<RequestContext>) -> Self {
         let inner = otel_transform::transform_resource_logs(self);
         // Build the PyObject
-        let res = Python::with_gil(|py| -> PyResult<()> {
+        let res = Python::attach(|py| -> PyResult<()> {
             let spans = ResourceLogs {
                 resource: inner.resource.clone(),
                 scope_logs: inner.scope_logs.clone(),