chore: test

Author: ovr

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

@@ -1,5 +1,6 @@
 import {
   setupLogger,
+  resetLogger,
   registerInterface,
   shutdownInterface,
   execSql,
@@ -382,5 +383,9 @@ export class SQLServer {
 
   public async shutdown(mode: ShutdownMode): Promise<void> {
     await shutdownInterface(this.sqlInterfaceInstance!, mode);
+
+    resetLogger(
+      process.env.CUBEJS_LOG_LEVEL === 'trace' ? 'trace' : 'warn'
+    );
   }
 }

packages/cubejs-api-gateway/test/auth.test.ts

@@ -3,7 +3,7 @@ import express, { Application as ExpressApplication, RequestHandler } from 'expr
 // eslint-disable-next-line import/no-extraneous-dependencies
 import request from 'supertest';
 import jwt from 'jsonwebtoken';
-import { pausePromise } from '@cubejs-backend/shared';
+import {getEnv, pausePromise} from '@cubejs-backend/shared';
 
 import { ApiGateway, ApiGatewayOptions, CubejsHandlerError, Request } from '../src';
 import { AdapterApiMock, DataSourceStorageMock } from './mocks';
@@ -113,7 +113,7 @@ describe('test authorization with native gateway', () => {
     await apiGateway.shutdownSQLServer();
   });
 
-  fit('default authorization', async () => {
+  it('default authorization', async () => {
     const token = generateAuthToken({ uid: 5, });
 
     await request(app)

packages/cubejs-backend-native/js/index.ts

@@ -352,6 +352,12 @@ export const setupLogger = (logger: (extra: any) => unknown, logLevel: LogLevel)
   native.setupLogger({ logger: wrapNativeFunctionWithChannelCallback(logger), logLevel });
 };
 
+/// Reset local to default implementation, which uses STDOUT
+export const resetLogger = (logLevel: LogLevel): void => {
+  const native = loadNative();
+  native.resetLogger({ logLevel });
+};
+
 export const isFallbackBuild = (): boolean => {
   const native = loadNative();
   return native.isFallbackBuild();

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

@@ -398,7 +398,9 @@ impl SqlGenerator for NodeSqlGenerator {
 impl Drop for NodeSqlGenerator {
     fn drop(&mut self) {
         let channel = self.channel.clone();
+        /// Safety: Safe, because on_track take is used only for dropping
         let sql_generator_obj = self.sql_generator_obj.take().unwrap();
+
         channel.send(move |mut cx| {
             let _ = match Arc::try_unwrap(sql_generator_obj) {
                 Ok(v) => v.into_inner(&mut cx),

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

@@ -48,29 +48,13 @@ pub fn tokio_runtime() -> Result<&'static Runtime, CubeError> {
     })
 }
 
-pub fn create_logger(log_level: log::Level) -> SimpleLogger {
-    SimpleLogger::new()
-        .with_level(log::Level::Error.to_level_filter())
-        .with_module_level("cubesql", log_level.to_level_filter())
-        .with_module_level("cubejs_native", log_level.to_level_filter())
-        .with_module_level("datafusion", log::Level::Warn.to_level_filter())
-        .with_module_level("pg_srv", log::Level::Warn.to_level_filter())
-}
-
 #[cfg(feature = "neon-entrypoint")]
 #[neon::main]
 fn main(cx: ModuleContext) -> NeonResult<()> {
     // We use log_rerouter to swap logger, because we init logger from js side in api-gateway
     log_reroute::init().unwrap();
 
-    let logger = Box::new(create_logger(log::Level::Error));
-    log_reroute::reroute_boxed(logger);
-
-    ReportingLogger::init(
-        Box::new(LocalReporter::new()),
-        log::Level::Error.to_level_filter(),
-    )
-    .unwrap();
+    node_export::setup_local_logger(log::Level::Error);
 
     node_export::register_module_exports::<NodeConfigurationImpl>(cx)?;
 

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

@@ -11,14 +11,14 @@ use crate::channel::call_js_with_channel_as_callback;
 #[derive(Debug)]
 pub struct NodeBridgeLogger {
     channel: Arc<Channel>,
-    on_track: Arc<Root<JsFunction>>,
+    on_track: Option<Arc<Root<JsFunction>>>,
 }
 
 impl NodeBridgeLogger {
     pub fn new(channel: Channel, on_track: Root<JsFunction>) -> Self {
         Self {
             channel: Arc::new(channel),
-            on_track: Arc::new(on_track),
+            on_track: Some(Arc::new(on_track)),
         }
     }
 }
@@ -35,12 +35,34 @@ impl LogReporter for NodeBridgeLogger {
         let extra = serde_json::to_string(&EventBox { event: props }).unwrap();
 
         let channel = self.channel.clone();
-        let on_track = self.on_track.clone();
+        /// Safety: Safe, because on_track take is used only for dropping
+        let on_track = self.on_track.as_ref().expect("Unable to unwrap on_track to log event for NodeBridgeLogger. Logger was dropped?").clone();
+
         // TODO: Move to spawning loops
         spawn(async move { log(channel, on_track, Some(extra)).await });
     }
 }
 
+impl Drop for NodeBridgeLogger {
+    fn drop(&mut self) {
+        let channel = self.channel.clone();
+        let on_track = self.on_track.take().expect("NodeBridgeLogger.on_track was t already dropped");
+
+        channel.send(move |mut cx| {
+            let _ = match Arc::try_unwrap(on_track) {
+                Ok(v) => {
+                    v.into_inner(&mut cx)
+                },
+                Err(_) => {
+                    log::error!("Unable to drop sql generator: reference is copied somewhere else. Potential memory leak");
+                    return Ok(());
+                },
+            };
+            Ok(())
+        });
+    }
+}
+
 async fn log(channel: Arc<Channel>, on_track: Arc<Root<JsFunction>>, extra: Option<String>) {
     let _ = call_js_with_channel_as_callback::<String>(channel, on_track, extra).await;
 }

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

@@ -16,7 +16,7 @@ use crate::cubesql_utils::with_session;
 use crate::logger::NodeBridgeLogger;
 use crate::sql4sql::sql4sql;
 use crate::stream::OnDrainHandler;
-use crate::tokio_runtime_node;
+use crate::{tokio_runtime_node};
 use crate::transport::NodeBridgeTransport;
 use crate::utils::batch_to_rows;
 use cubenativeutils::wrappers::neon::context::neon_run_with_guarded_lifetime;
@@ -31,9 +31,10 @@ use std::rc::Rc;
 use std::sync::Arc;
 
 use cubesql::{telemetry::ReportingLogger, CubeError};
-
+use cubesql::telemetry::LocalReporter;
 use neon::prelude::*;
 use neon::result::Throw;
+use simple_logger::SimpleLogger;
 
 pub(crate) struct SQLInterface {
     pub(crate) services: Arc<NodeCubeServices>,
@@ -497,29 +498,34 @@ fn is_fallback_build(mut cx: FunctionContext) -> JsResult<JsBoolean> {
     Ok(JsBoolean::new(&mut cx, true))
 }
 
-pub fn setup_logger(mut cx: FunctionContext) -> JsResult<JsUndefined> {
-    let options = cx.argument::<JsObject>(0)?;
-    let cube_logger = options
-        .get::<JsFunction, _, _>(&mut cx, "logger")?
-        .root(&mut cx);
-
-    let log_level_handle = options.get_value(&mut cx, "logLevel")?;
-    let log_level = if log_level_handle.is_a::<JsString, _>(&mut cx) {
-        let value = log_level_handle.downcast_or_throw::<JsString, _>(&mut cx)?;
-        let log_level = match value.value(&mut cx).as_str() {
+fn get_log_level_from_variable(log_level_handle: Handle<JsValue>, cx: &mut FunctionContext) -> NeonResult<log::Level> {
+    if log_level_handle.is_a::<JsString, _>(cx) {
+        let value = log_level_handle.downcast_or_throw::<JsString, _>(cx)?;
+        let log_level = match value.value(cx).as_str() {
             "error" => log::Level::Error,
             "warn" => log::Level::Warn,
             "info" => log::Level::Info,
             "debug" => log::Level::Debug,
             "trace" => log::Level::Trace,
             x => cx.throw_error(format!("Unrecognized log level: {}", x))?,
         };
-        log_level
+
+        Ok(log_level)
     } else {
-        log::Level::Trace
-    };
+        Ok(log::Level::Trace)
+    }
+}
+
+pub fn setup_logger(mut cx: FunctionContext) -> JsResult<JsUndefined> {
+    let options = cx.argument::<JsObject>(0)?;
+    let cube_logger = options
+        .get::<JsFunction, _, _>(&mut cx, "logger")?
+        .root(&mut cx);
 
-    let logger = crate::create_logger(log_level);
+    let log_level_handle = options.get_value(&mut cx, "logLevel")?;
+    let log_level = get_log_level_from_variable(log_level_handle, &mut cx)?;
+
+    let logger = create_logger(log_level);
     log_reroute::reroute_boxed(Box::new(logger));
 
     ReportingLogger::init(
@@ -531,6 +537,37 @@ pub fn setup_logger(mut cx: FunctionContext) -> JsResult<JsUndefined> {
     Ok(cx.undefined())
 }
 
+pub fn create_logger(log_level: log::Level) -> SimpleLogger {
+    SimpleLogger::new()
+        .with_level(log::Level::Error.to_level_filter())
+        .with_module_level("cubesql", log_level.to_level_filter())
+        .with_module_level("cubejs_native", log_level.to_level_filter())
+        .with_module_level("datafusion", log::Level::Warn.to_level_filter())
+        .with_module_level("pg_srv", log::Level::Warn.to_level_filter())
+}
+
+pub fn setup_local_logger(log_level: log::Level) {
+    let logger = create_logger(log_level);
+    log_reroute::reroute_boxed(Box::new(logger));
+
+    ReportingLogger::init(
+        Box::new(LocalReporter::new()),
+        log::Level::Error.to_level_filter(),
+    )
+        .unwrap();
+}
+
+pub fn reset_logger(mut cx: FunctionContext) -> JsResult<JsUndefined> {
+    let options = cx.argument::<JsObject>(0)?;
+
+    let log_level_handle = options.get_value(&mut cx, "logLevel")?;
+    let log_level = get_log_level_from_variable(log_level_handle, &mut cx)?;
+
+    setup_local_logger(log_level);
+
+    Ok(cx.undefined())
+}
+
 //============ sql planner ===================
 
 fn build_sql_and_params(cx: FunctionContext) -> JsResult<JsValue> {
@@ -570,6 +607,7 @@ pub fn register_module_exports<C: NodeConfiguration + 'static>(
     mut cx: ModuleContext,
 ) -> NeonResult<()> {
     cx.export_function("setupLogger", setup_logger)?;
+    cx.export_function("resetLogger", reset_logger)?;
     cx.export_function("registerInterface", register_interface::<C>)?;
     cx.export_function("shutdownInterface", shutdown_interface)?;
     cx.export_function("execSql", exec_sql)?;