add few simple metrics to sqld

- libsql_server_total_response_size_before_lock - histogram of
  TOTAL_RESPOSE_SIZE values before attempt to acquire connection
  semaphore
- libsql_server_stream_handles - guage with current in-memory stream
  handles
- libsql_server_hrana_step_errors - counter with Hrana error codes

Author: sivukhin

libsql-server/src/connection/mod.rs

@@ -13,12 +13,13 @@ use crate::error::Error;
 use crate::http::user::timing::sample_time;
 use crate::metrics::{
     CONCURRENT_CONNECTIONS_COUNT, CONNECTION_ALIVE_DURATION, CONNECTION_CREATE_TIME,
+    TOTAL_RESPONSE_SIZE_HIST,
 };
 use crate::namespace::meta_store::MetaStore;
 use crate::namespace::NamespaceName;
 use crate::query::{Params, Query};
 use crate::query_analysis::Statement;
-use crate::query_result_builder::{IgnoreResult, QueryResultBuilder};
+use crate::query_result_builder::{IgnoreResult, QueryResultBuilder, TOTAL_RESPONSE_SIZE};
 use crate::replication::FrameNo;
 use crate::Result;
 
@@ -352,6 +353,8 @@ impl<F: MakeConnection> MakeConnection for MakeThrottledConnection<F> {
             "Available semaphore units: {}",
             self.semaphore.available_permits()
         );
+        TOTAL_RESPONSE_SIZE_HIST
+            .record(TOTAL_RESPONSE_SIZE.load(std::sync::atomic::Ordering::Relaxed) as f64);
         let units = self.units_to_take();
         let waiters_guard = WaitersGuard::new(&self.waiters);
         if (waiters_guard.waiters.load(Ordering::Relaxed) as u64) >= self.max_concurrent_requests {

libsql-server/src/hrana/http/stream.rs

@@ -12,6 +12,7 @@ use tokio::time::{Duration, Instant};
 
 use crate::connection::MakeConnection;
 use crate::database::Connection;
+use crate::metrics::STREAM_HANDLES_COUNT;
 
 use super::super::ProtocolError;
 use super::Server;
@@ -169,6 +170,8 @@ pub async fn acquire<'srv>(
                 baton_seq: rand::random(),
             });
             state.handles.insert(stream.stream_id, Handle::Acquired);
+            STREAM_HANDLES_COUNT.increment(1.0);
+
             tracing::debug!(
                 "Stream {} was created with baton seq {}",
                 stream.stream_id,
@@ -253,6 +256,7 @@ impl<'srv> Drop for Guard<'srv> {
             tracing::debug!("Stream {stream_id} was released for further use");
         } else {
             tracing::debug!("Stream {stream_id} was closed");
+            STREAM_HANDLES_COUNT.decrement(1.0);
         }
     }
 }
@@ -374,6 +378,7 @@ fn pump_expire(state: &mut ServerStreamState, cx: &mut task::Context) {
 
         match state.handles.get_mut(&stream_id) {
             Some(handle @ Handle::Available(_)) => {
+                STREAM_HANDLES_COUNT.decrement(1.0);
                 *handle = Handle::Expired;
             }
             _ => continue,

libsql-server/src/hrana/stmt.rs

@@ -1,4 +1,5 @@
 use anyhow::{anyhow, bail, Result};
+use metrics::counter;
 use std::collections::HashMap;
 
 use super::result_builder::SingleStatementBuilder;
@@ -206,39 +207,46 @@ fn catch_stmt_error(sqld_error: SqldError) -> anyhow::Error {
 }
 
 pub fn stmt_error_from_sqld_error(sqld_error: SqldError) -> Result<StmtError, SqldError> {
-    Ok(match sqld_error {
-        SqldError::LibSqlInvalidQueryParams(source) => StmtError::ArgsInvalid { source },
-        SqldError::LibSqlTxTimeout => StmtError::TransactionTimeout,
-        SqldError::LibSqlTxBusy => StmtError::TransactionBusy,
+    let result = match sqld_error {
+        SqldError::LibSqlInvalidQueryParams(source) => Ok(StmtError::ArgsInvalid { source }),
+        SqldError::LibSqlTxTimeout => Ok(StmtError::TransactionTimeout),
+        SqldError::LibSqlTxBusy => Ok(StmtError::TransactionBusy),
         SqldError::BuilderError(QueryResultBuilderError::ResponseTooLarge(_)) => {
-            StmtError::ResponseTooLarge
+            Ok(StmtError::ResponseTooLarge)
         }
-        SqldError::Blocked(reason) => StmtError::Blocked { reason },
-        SqldError::RpcQueryError(e) => StmtError::Proxy(e.message),
+        SqldError::Blocked(reason) => Ok(StmtError::Blocked { reason }),
+        SqldError::RpcQueryError(e) => Ok(StmtError::Proxy(e.message)),
         SqldError::RusqliteError(rusqlite_error)
         | SqldError::RusqliteErrorExtended(rusqlite_error, _) => match rusqlite_error {
-            rusqlite::Error::SqliteFailure(sqlite_error, Some(message)) => StmtError::SqliteError {
-                source: sqlite_error,
-                message,
-            },
-            rusqlite::Error::SqliteFailure(sqlite_error, None) => StmtError::SqliteError {
+            rusqlite::Error::SqliteFailure(sqlite_error, Some(message)) => {
+                Ok(StmtError::SqliteError {
+                    source: sqlite_error,
+                    message,
+                })
+            }
+            rusqlite::Error::SqliteFailure(sqlite_error, None) => Ok(StmtError::SqliteError {
                 message: sqlite_error.to_string(),
                 source: sqlite_error,
-            },
+            }),
             rusqlite::Error::SqlInputError {
                 error: sqlite_error,
                 msg: message,
                 offset,
                 ..
-            } => StmtError::SqlInputError {
+            } => Ok(StmtError::SqlInputError {
                 source: sqlite_error,
                 message,
                 offset,
-            },
-            rusqlite_error => return Err(SqldError::RusqliteError(rusqlite_error)),
+            }),
+            rusqlite_error => Err(SqldError::RusqliteError(rusqlite_error)),
         },
-        sqld_error => return Err(sqld_error),
-    })
+        sqld_error => Err(sqld_error),
+    };
+
+    let code = result.as_ref().map(|x| x.code()).unwrap_or("UKNOWN");
+    counter!("libsql_server_hrana_step_errors", 1, "code" => code);
+
+    result
 }
 
 pub fn proto_error_from_stmt_error(error: &StmtError) -> hrana::proto::Error {

libsql-server/src/metrics.rs

@@ -27,6 +27,16 @@ pub static CONCURRENT_CONNECTIONS_COUNT: Lazy<Gauge> = Lazy::new(|| {
     describe_gauge!(NAME, "number of concurrent connections");
     register_gauge!(NAME)
 });
+pub static TOTAL_RESPONSE_SIZE_HIST: Lazy<Histogram> = Lazy::new(|| {
+    const NAME: &str = "libsql_server_total_response_size_before_lock";
+    describe_histogram!(NAME, "total response size value before connection lock");
+    register_histogram!(NAME)
+});
+pub static STREAM_HANDLES_COUNT: Lazy<Gauge> = Lazy::new(|| {
+    const NAME: &str = "libsql_server_stream_handles";
+    describe_gauge!(NAME, "amount of in-memory stream handles");
+    register_gauge!(NAME)
+});
 pub static NAMESPACE_LOAD_LATENCY: Lazy<Histogram> = Lazy::new(|| {
     const NAME: &str = "libsql_server_namespace_load_latency";
     describe_histogram!(NAME, "latency is us when loading a namespace");