feat: 🌈 are prometheus remote reads actually *working* ?

Author: fungiboletus

.gitignore

@@ -26,3 +26,5 @@ test.db*
 
 # Claude
 .claude/settings.local.json
+
+reference/

src/ingestors/http/prometheus_read.rs

@@ -124,6 +124,13 @@ pub async fn prometheus_remote_read(
 
     // Log detailed information about each query for debugging
     for (i, query) in read_request.queries.iter().enumerate() {
+        println!(
+            "[DEBUG PROM READ] Query {}: time range {}ms - {}ms ({} matchers)",
+            i,
+            query.start_timestamp_ms,
+            query.end_timestamp_ms,
+            query.matchers.len()
+        );
         info!(
             "Query {}: time range {}ms - {}ms ({} matchers)",
             i,
@@ -133,6 +140,10 @@ pub async fn prometheus_remote_read(
         );
 
         for matcher in &query.matchers {
+            println!(
+                "[DEBUG PROM READ]   Matcher: {}={} (type={})",
+                matcher.name, matcher.value, matcher.r#type
+            );
             debug!(
                 "  Matcher: {}={} (type={})",
                 matcher.name, matcher.value, matcher.r#type
@@ -226,7 +237,9 @@ async fn handle_streamed_response(
     state: &HttpServerState,
     read_request: &crate::parsing::prometheus::remote_read_models::ReadRequest,
 ) -> Result<Response<axum::body::Body>, AppError> {
-    let mut chunked_responses = Vec::with_capacity(read_request.queries.len());
+    // Prometheus expects ONE ChunkedReadResponse per series, not one per query
+    // Each message contains exactly one series
+    let mut chunked_responses = Vec::new();
 
     for (query_index, query) in read_request.queries.iter().enumerate() {
         // Convert Prometheus matchers to SensApp matchers
@@ -245,36 +258,88 @@ async fn handle_streamed_response(
                 AppError::internal_server_error(anyhow::anyhow!("Storage query failed: {}", e))
             })?;
 
+        println!(
+            "[DEBUG PROM READ] Query {} returned {} sensors from storage",
+            query_index,
+            sensor_data.len()
+        );
         debug!(
             "Query {} returned {} sensors",
             query_index,
             sensor_data.len()
         );
 
-        // Convert to ChunkedSeries
-        let chunked_series: Vec<_> = sensor_data
-            .iter()
-            .filter_map(|sd| {
-                // Extract labels
-                let labels = build_prometheus_labels(&sd.sensor);
+        // Log each sensor found
+        for sd in &sensor_data {
+            println!(
+                "[DEBUG PROM READ]   Sensor: {} (type={:?}, {} samples)",
+                sd.sensor.name,
+                sd.sensor.sensor_type,
+                sd.samples.len()
+            );
+        }
 
-                // Extract samples as Prometheus format
-                let samples = extract_prom_samples_for_chunks(sd)?;
+        // Convert each sensor to a separate ChunkedReadResponse
+        // Prometheus expects ONE series per response message!
+        for sd in &sensor_data {
+            // Extract labels
+            let labels = build_prometheus_labels(&sd.sensor);
+            println!(
+                "[DEBUG PROM READ]   Labels for sensor {}: {:?}",
+                sd.sensor.name,
+                labels
+                    .iter()
+                    .map(|l| format!("{}={}", l.name, l.value))
+                    .collect::<Vec<_>>()
+            );
 
-                // Encode as XOR chunks
-                ChunkEncoder::encode_series(labels, samples).ok()
-            })
-            .collect();
+            // Extract samples as Prometheus format
+            let samples = match extract_prom_samples_for_chunks(sd) {
+                Some(s) => s,
+                None => continue, // Skip non-numeric types
+            };
+
+            println!(
+                "[DEBUG PROM READ]   Encoding {} samples for sensor {} (first ts: {}, last ts: {})",
+                samples.len(),
+                sd.sensor.name,
+                samples.first().map(|s| s.timestamp).unwrap_or(0),
+                samples.last().map(|s| s.timestamp).unwrap_or(0)
+            );
+
+            // Encode as XOR chunks
+            let chunked_series = match ChunkEncoder::encode_series(labels, samples) {
+                Ok(cs) => cs,
+                Err(_) => continue, // Skip on encoding error
+            };
+
+            // Create ONE response per series (this is what Prometheus expects!)
+            let chunked_response =
+                ChunkEncoder::create_response(query_index as i64, vec![chunked_series]);
+            chunked_responses.push(chunked_response);
+        }
 
-        let chunked_response = ChunkEncoder::create_response(query_index as i64, chunked_series);
-        chunked_responses.push(chunked_response);
+        println!(
+            "[DEBUG PROM READ] Query {} produced {} chunked_responses (one per series)",
+            query_index,
+            chunked_responses.len()
+        );
     }
 
+    println!(
+        "[DEBUG PROM READ] Total {} chunked_responses to write",
+        chunked_responses.len()
+    );
+
     // Create the streaming response body
     let body = StreamWriter::create_stream_body(&chunked_responses).map_err(|e| {
         AppError::internal_server_error(anyhow::anyhow!("Failed to create stream body: {}", e))
     })?;
 
+    println!(
+        "[DEBUG PROM READ] Returning STREAMED_XOR_CHUNKS response with {} bytes",
+        body.len()
+    );
     info!(
         "Prometheus remote read: Returning STREAMED_XOR_CHUNKS response with {} bytes",
         body.len()

src/parsing/prometheus/chunk_encoder.rs

@@ -1,8 +1,7 @@
 use super::remote_read_models::{Chunk as ProtoChunk, ChunkedReadResponse, ChunkedSeries, chunk};
 use super::remote_write_models::{Label, Sample};
 use anyhow::Result;
-use rusty_chunkenc::chunk::Chunk;
-use rusty_chunkenc::xor::XORSample;
+use rusty_chunkenc::xor::{XORChunk, XORSample};
 use tracing::debug;
 
 /// Encodes time series samples into XOR-compressed chunks for Prometheus remote read.
@@ -45,11 +44,12 @@ impl ChunkEncoder {
         let max_time_ms = samples.last().map(|s| s.timestamp).unwrap_or(0);
 
         // Create the XOR chunk
-        let chunk = Chunk::new_xor(xor_samples);
+        let xor_chunk = XORChunk::new(xor_samples);
 
-        // Encode the chunk to bytes
+        // Encode just the raw XOR chunk data (NOT the full chunk format with length/type/crc)
+        // Prometheus remote read expects raw XOR data starting with 2-byte BE sample count
         let mut encoded_data = Vec::new();
-        chunk.write(&mut encoded_data)?;
+        xor_chunk.write(&mut encoded_data)?;
 
         debug!(
             "Encoded chunk: {} samples into {} bytes (time range: {}ms - {}ms)",

src/parsing/prometheus/remote_read_parser.rs

@@ -1,5 +1,5 @@
 use super::common::decompress_snappy;
-use super::remote_read_models::{QueryResult, ReadRequest, ReadResponse};
+use super::remote_read_models::{ReadRequest, ReadResponse};
 use anyhow::Result;
 use prost::Message;
 use std::io::Cursor;
@@ -47,19 +47,6 @@ pub fn parse_remote_read_request(input: &[u8]) -> Result<ReadRequest> {
     Ok(request)
 }
 
-pub fn create_empty_read_response(request: &ReadRequest) -> Result<ReadResponse> {
-    // Create an empty response with the same number of query results as queries
-    let results = request
-        .queries
-        .iter()
-        .map(|_| QueryResult {
-            timeseries: vec![], // Empty timeseries for now
-        })
-        .collect();
-
-    Ok(ReadResponse { results })
-}
-
 pub fn serialize_read_response(response: &ReadResponse) -> Result<Vec<u8>> {
     let encoded = response.encode_to_vec();
     debug!("Encoded ReadResponse to {} bytes", encoded.len());
@@ -136,32 +123,6 @@ mod tests {
         assert_eq!(output.accepted_response_types.len(), 1);
     }
 
-    #[test]
-    fn test_create_empty_read_response() {
-        let request = ReadRequest {
-            queries: vec![
-                Query {
-                    start_timestamp_ms: 1000,
-                    end_timestamp_ms: 2000,
-                    matchers: vec![],
-                    hints: None,
-                },
-                Query {
-                    start_timestamp_ms: 3000,
-                    end_timestamp_ms: 4000,
-                    matchers: vec![],
-                    hints: None,
-                },
-            ],
-            accepted_response_types: vec![],
-        };
-
-        let response = create_empty_read_response(&request).unwrap();
-        assert_eq!(response.results.len(), 2);
-        assert_eq!(response.results[0].timeseries.len(), 0);
-        assert_eq!(response.results[1].timeseries.len(), 0);
-    }
-
     #[test]
     fn test_serialize_read_response() {
         let response = ReadResponse {