feat: (qe) Telemetry capturing for PDP/Data proxy

Cargo.toml

@@ -39,7 +39,8 @@ user-facing-errors = { path = "./libs/user-facing-errors" }
 uuid = { version = "1", features = ["serde"] }
 
 [workspace.dependencies.quaint]
-git = "https://github.com/prisma/quaint"
+git = "https://github.com/miguelff/quaint"
+branch = "fix-traces"
 features = [
   "bigdecimal",
   "chrono",

Makefile

@@ -214,7 +214,7 @@ validate:
 	cargo run --bin test-cli -- validate-datamodel dev_datamodel.prisma
 
 qe:
-	cargo run --bin query-engine -- --enable-playground --enable-raw-queries --enable-metrics --enable-open-telemetry
+	cargo run --bin query-engine -- --enable-playground --enable-raw-queries --enable-metrics --enable-open-telemetry --enable-telemetry-in-response
 
 qe-dmmf:
 	cargo run --bin query-engine -- cli dmmf > dmmf.json

query-engine/connector-test-kit-rs/query-tests-setup/src/runner/binary.rs

@@ -20,7 +20,13 @@ pub struct BinaryRunner {
 #[async_trait::async_trait]
 impl RunnerInterface for BinaryRunner {
     async fn load(datamodel: String, connector_tag: ConnectorTag, metrics: MetricRegistry) -> TestResult<Self> {
-        let opts = PrismaOpt::from_list(&["binary", "--enable-raw-queries", "--datamodel", &datamodel]);
+        let opts = PrismaOpt::from_list(&[
+            "binary",
+            "--enable-raw-queries",
+            "--enable-telemetry-in-response",
+            "--datamodel",
+            &datamodel,
+        ]);
         let state = setup(&opts, false, Some(metrics)).await.unwrap();
 
         let configuration = opts.configuration(true).unwrap();

query-engine/core/Cargo.toml

@@ -22,11 +22,11 @@ futures = "0.3"
 indexmap = { version = "1.7", features = ["serde-1"] }
 itertools = "0.10"
 mongodb-connector = { path = "../connectors/mongodb-query-connector", package = "mongodb-query-connector", optional = true }
-once_cell = "1.3"
+once_cell = "1"
 petgraph = "0.4"
 prisma-models = { path = "../prisma-models", features = ["default_generators"] }
 prisma-value = { path = "../../libs/prisma-value" }
-opentelemetry = { version = "0.17"}
+opentelemetry = { version = "0.17.0", features = ["rt-tokio", "serialize"] }
 query-engine-metrics = {path = "../metrics"}
 serde.workspace = true
 serde_json.workspace = true
@@ -35,7 +35,7 @@ thiserror = "1.0"
 tokio.workspace = true
 tracing = { version = "0.1", features = ["attributes"] }
 tracing-futures = "0.2"
-tracing-subscriber = {version = "0.3", features = ["env-filter"]}
+tracing-subscriber = { version = "0.3", features = ["env-filter"] }
 tracing-opentelemetry = "0.17.4"
 url = "2"
 user-facing-errors = { path = "../../libs/user-facing-errors" }

query-engine/core/src/interactive_transactions/error.rs

@@ -1,5 +1,10 @@
 use thiserror::Error;
 
+use crate::{
+    response_ir::{Item, Map},
+    CoreError,
+};
+
 #[derive(Debug, Error, PartialEq)]
 pub enum TransactionError {
     #[error("Unable to start a transaction in the given time.")]
@@ -17,3 +22,27 @@ pub enum TransactionError {
     #[error("Unexpected response: {reason}.")]
     Unknown { reason: String },
 }
+
+#[derive(Debug, serde::Serialize, PartialEq)]
+pub struct ExtendedTransactionUserFacingError {
+    #[serde(flatten)]
+    user_facing_error: user_facing_errors::Error,
+
+    #[serde(skip_serializing_if = "indexmap::IndexMap::is_empty")]
+    extensions: Map,
+}
+
+impl ExtendedTransactionUserFacingError {
+    pub fn set_extension(&mut self, key: String, val: serde_json::Value) {
+        self.extensions.entry(key).or_insert(Item::Json(val));
+    }
+}
+
+impl From<CoreError> for ExtendedTransactionUserFacingError {
+    fn from(error: CoreError) -> Self {
+        ExtendedTransactionUserFacingError {
+            user_facing_error: error.into(),
+            extensions: Default::default(),
+        }
+    }
+}

query-engine/core/src/lib.rs

@@ -16,7 +16,7 @@ pub mod telemetry;
 pub use self::{
     error::{CoreError, FieldConversionError},
     executor::{QueryExecutor, TransactionOptions},
-    interactive_transactions::{TransactionError, TxId},
+    interactive_transactions::{ExtendedTransactionUserFacingError, TransactionError, TxId},
     query_document::*,
     telemetry::*,
 };