Merge branch 'master' into release/1.70

Cargo.toml

@@ -98,7 +98,7 @@ slog-scope = "4.4"
 slog-stdlog = "4.1"
 slog-term = "2.6"
 thiserror = "1.0"
-tokio = "1.17"
+tokio = "1.36"
 tokio-compat-02 = "0.2"
 tokio-core = "0.1"
 tokio-io = "0.1"

autoconnect/Cargo.toml

@@ -53,7 +53,7 @@ actix-service = "2.0"
 docopt = "1.1"
 
 [features]
-default = ["dual"]
+default = ["dual", "emulator"]
 bigtable = ["autopush_common/bigtable", "autoconnect_settings/bigtable"]
 dynamodb = ["autopush_common/dynamodb", "autoconnect_settings/dynamodb"]
 dual = ["bigtable", "dynamodb"]

autoconnect/src/main.rs

@@ -4,7 +4,7 @@
 #[macro_use]
 extern crate slog_scope;
 
-use std::{env, vec::Vec};
+use std::{env, time::Duration, vec::Vec};
 
 use actix_http::HttpService;
 use actix_server::Server;
@@ -16,6 +16,7 @@ use serde::Deserialize;
 use autoconnect_settings::{AppState, Settings};
 use autoconnect_web::{build_app, config, config_router};
 use autopush_common::{
+    db::spawn_pool_periodic_reporter,
     errors::{ApcErrorKind, Result},
     logging,
 };
@@ -79,12 +80,17 @@ async fn main() -> Result<()> {
     let actix_workers = settings.actix_workers;
     let app_state = AppState::from_settings(settings)?;
     app_state.init_and_spawn_megaphone_updater().await?;
+    spawn_pool_periodic_reporter(
+        Duration::from_secs(10),
+        app_state.db.clone(),
+        app_state.metrics.clone(),
+    );
 
     info!(
-        "Starting autoconnect on port: {} router_port: {} (available_parallelism: {:?})",
+        "Starting autoconnect on port: {} router_port: {} ({})",
         port,
         router_port,
-        std::thread::available_parallelism()
+        logging::parallelism_banner()
     );
 
     let router_app_state = app_state.clone();

autoendpoint/src/main.rs

@@ -59,9 +59,9 @@ async fn main() -> Result<(), Box<dyn Error>> {
         .await
         .expect("Could not start server");
     info!(
-        "Starting autoendpoint on port: {} (available_parallelism: {:?})",
+        "Starting autoendpoint on port: {} ({})",
         host_port,
-        std::thread::available_parallelism()
+        logging::parallelism_banner()
     );
     server.await?;
 

autoendpoint/src/server.rs

@@ -7,19 +7,18 @@ use actix_cors::Cors;
 use actix_web::{
     dev, http::StatusCode, middleware::ErrorHandlers, web, web::Data, App, HttpServer,
 };
-#[cfg(feature = "bigtable")]
-use autopush_common::db::bigtable::BigTableClientImpl;
-#[cfg(feature = "dual")]
-use autopush_common::db::dual::DualClientImpl;
 use cadence::StatsdClient;
 use fernet::MultiFernet;
 use serde_json::json;
 
+#[cfg(feature = "bigtable")]
+use autopush_common::db::bigtable::BigTableClientImpl;
+#[cfg(feature = "dual")]
+use autopush_common::db::dual::DualClientImpl;
 #[cfg(feature = "dynamodb")]
 use autopush_common::db::dynamodb::DdbClientImpl;
-
 use autopush_common::{
-    db::{client::DbClient, DbSettings, StorageType},
+    db::{client::DbClient, spawn_pool_periodic_reporter, DbSettings, StorageType},
     middleware::sentry::SentryWrapper,
 };
 
@@ -130,6 +129,12 @@ impl Server {
             adm_router,
         };
 
+        spawn_pool_periodic_reporter(
+            Duration::from_secs(10),
+            app_state.db.clone(),
+            app_state.metrics.clone(),
+        );
+
         let server = HttpServer::new(move || {
             // These have a bad habit of being reset. Specify them explicitly.
             let cors = Cors::default()

autopush-common/Cargo.toml

@@ -56,9 +56,10 @@ url.workspace = true
 
 again = "0.1"
 async-trait = "0.1"
-deadpool = "0.10"
+deadpool = { version = "0.10", features = ["managed", "rt_tokio_1"] }
 gethostname = "0.4"
 futures-backoff = "0.1.0"
+num_cpus = "1.16"
 woothee = "0.13"
 
 # #[cfg(bigtable)] for this section.
@@ -67,6 +68,7 @@ google-cloud-rust-raw = { version = "0.16", default-features = false, features =
     "bigtable",
 ], optional = true }
 grpcio = { version = "=0.13.0", features = ["openssl"], optional = true }
+grpcio-sys = { version = "=0.13.0", optional = true }
 protobuf = { version = "=2.28.0", optional = true } # grpcio does not support protobuf 3+
 form_urlencoded = { version = "1.2", optional = true }
 
@@ -82,6 +84,7 @@ actix-rt = "2.8"
 bigtable = [
     "dep:google-cloud-rust-raw",
     "dep:grpcio",
+    "dep:grpcio-sys",
     "dep:protobuf",
     "dep:form_urlencoded",
 ]

autopush-common/src/db/bigtable/bigtable_client/Bigtable_Learnings.md

@@ -46,3 +46,7 @@ fn expiry_filter() -> Result<data::RowFilter, error::BigTableError> {
 ```
 
 Adding this filter to a query will return a result that only includes the "expiry" cells which have a value that is greater than the current time. No other cells or values will be included in the return set, however each "row" will include the row meta information, including the key.
+
+## Connections
+
+GRPC is not very smart. It's often best to include the port number when including the DSN (e.g. `grpc://bigtable.googleapis.com:443`)