move logs to native

Author: vasilev-alex

packages/cubejs-api-gateway/src/gateway.ts

@@ -391,32 +391,32 @@ class ApiGateway {
         try {
           await this.assertApiScope('data', req.context?.securityContext);
 
-          this.log({
-            type: 'Load Request',
-            query: {
-              sql: query,
-            },
-            apiType: 'sql',
-            streaming: true
-          }, req.context);
+          // this.log({
+          //   type: 'Load Request',
+          //   query: {
+          //     sql: query,
+          //   },
+          //   apiType: 'sql',
+          //   streaming: true
+          // }, req.context);
 
           await this.sqlServer.execSql(req.body.query, res, req.context?.securityContext);
 
-          this.log(
-            {
-              type: 'Load Request Success',
-              query: {
-                sql: query,
-              },
-              duration: this.duration(requestStarted),
-              apiType: 'sql',
-              isDataQuery: true,
-              isPlayground: Boolean(
-                req.context?.signedWithPlaygroundAuthSecret
-              ),
-            },
-            req.context,
-          );
+          // this.log(
+          //   {
+          //     type: 'Load Request Success',
+          //     query: {
+          //       sql: query,
+          //     },
+          //     duration: this.duration(requestStarted),
+          //     apiType: 'sql',
+          //     isDataQuery: true,
+          //     isPlayground: Boolean(
+          //       req.context?.signedWithPlaygroundAuthSecret
+          //     ),
+          //   },
+          //   req.context,
+          // );
         } catch (e: any) {
           this.handleError({
             e,

packages/cubejs-backend-native/src/node_export.rs

@@ -30,6 +30,7 @@ use std::net::SocketAddr;
 use std::rc::Rc;
 use std::str::FromStr;
 use std::sync::Arc;
+use std::time::SystemTime;
 
 use cubesql::{telemetry::ReportingLogger, CubeError};
 
@@ -185,6 +186,8 @@ async fn handle_sql_query(
     stream_methods: WritableStreamMethods,
     sql_query: &String,
 ) -> Result<(), CubeError> {
+    let start_time = SystemTime::now();
+
     let config = services
         .injector()
         .get_service_typed::<dyn ConfigObj>()
@@ -222,7 +225,26 @@ async fn handle_sql_query(
         .state
         .set_auth_context(Some(native_auth_ctx.clone()));
 
-    let connection_id = session.state.connection_id;
+    if let Some(auth_context) = session.state.auth_context() {
+        session
+            .session_manager
+            .server
+            .transport
+            .log_load_state(
+                None,
+                auth_context,
+                session.state.get_load_request_meta("sql"),
+                "Load Request".to_string(),
+                serde_json::json!({
+                    "query": {
+                        "sql": sql_query,
+                    }
+                }),
+            )
+            .await?;
+    }
+
+    let session_clone = Arc::clone(&session);
 
     let execute = || async move {
         // todo: can we use compiler_cache?
@@ -310,8 +332,54 @@ async fn handle_sql_query(
     };
 
     let result = execute().await;
+    let duration = start_time.elapsed().unwrap().as_millis() as u64;
+
+    match &result {
+        Ok(_) => {
+            session_clone
+                .session_manager
+                .server
+                .transport
+                .log_load_state(
+                    None,
+                    session_clone.state.auth_context().unwrap(),
+                    session_clone.state.get_load_request_meta("sql"),
+                    "Load Request Success".to_string(),
+                    serde_json::json!({
+                        "query": {
+                            "sql": sql_query,
+                        },
+                        "apiType": "sql",
+                        "duration": duration,
+                        "isDataQuery": true
+                    }),
+                )
+                .await?;
+        }
+        Err(err) => {
+            session_clone
+                .session_manager
+                .server
+                .transport
+                .log_load_state(
+                    None,
+                    session_clone.state.auth_context().unwrap(),
+                    session_clone.state.get_load_request_meta("sql"),
+                    "Cube SQL Error".to_string(),
+                    serde_json::json!({
+                        "query": {
+                            "sql": sql_query
+                        },
+                        "apiType": "sql",
+                        "duration": duration,
+                        "error": err.message,
+                    }),
+                )
+                .await?;
+        }
+    }
 
-    session_manager.drop_session(connection_id).await;
+    session_manager.drop_session(session_clone.state.connection_id).await;
 
     result
 }