Skip to content
Merged
Show file tree
Hide file tree
Changes from 6 commits
Commits
File filter

Filter by extension

Filter by extension


Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
213 changes: 197 additions & 16 deletions Cargo.lock

Large diffs are not rendered by default.

16 changes: 10 additions & 6 deletions codegen/src/shuttle_main.rs
Original file line number Diff line number Diff line change
Expand Up @@ -12,23 +12,27 @@ pub(crate) fn tokens(_attr: TokenStream, item: TokenStream) -> TokenStream {
let mut user_main_fn = parse_macro_input!(item as ItemFn);
let loader_runner = LoaderAndRunner::from_item_fn(&mut user_main_fn);

quote! {
Into::into(quote! {
fn main() {
// manual expansion of #[tokio::main]
::shuttle_runtime::tokio::runtime::Builder::new_multi_thread()
.enable_all()
.build()
.unwrap()
.block_on(async {
::shuttle_runtime::__internals::start(__loader, __runner).await;
})
::shuttle_runtime::__internals::start(
__loader,
__runner,
env!("CARGO_CRATE_NAME"),
env!("CARGO_PKG_VERSION"),
).await;
})
}

#loader_runner

#user_main_fn
}
.into()
})
}

struct LoaderAndRunner {
Expand Down Expand Up @@ -358,7 +362,7 @@ mod tests {
assert_eq!(actual.fn_inputs, expected_inputs);

// Make sure attributes was removed from input
if let syn::FnArg::Typed(param) = input.sig.inputs.first().unwrap() {
if let FnArg::Typed(param) = input.sig.inputs.first().unwrap() {
assert!(
param.attrs.is_empty(),
"some attributes were not removed: {:?}",
Expand Down
11 changes: 10 additions & 1 deletion common/src/resource.rs
Original file line number Diff line number Diff line change
Expand Up @@ -165,7 +165,16 @@ pub enum Type {
}

#[derive(
Clone, Copy, Debug, strum::EnumString, strum::Display, Deserialize, Serialize, Eq, PartialEq,
Clone,
Copy,
Debug,
Deserialize,
Eq,
PartialEq,
Serialize,
strum::AsRefStr,
strum::EnumString,
strum::Display,
)]
#[typeshare::typeshare]
// is a flat enum instead of nested enum to allow typeshare
Expand Down
48 changes: 45 additions & 3 deletions runtime/Cargo.toml
Original file line number Diff line number Diff line change
Expand Up @@ -19,15 +19,37 @@ shuttle-service = { workspace = true }
anyhow = { workspace = true }
async-trait = { workspace = true }
hyper = { workspace = true, features = ["http1", "server", "tcp"] }
log = { version = "0.4", optional = true, default-features = false }
opentelemetry = { version = "0.27", optional = true, default-features = false, features = ["logs", "metrics", "trace", "tracing"] }
opentelemetry-otlp = { version = "0.27", optional = true, default-features = false, features = [
"http-proto",
"logs",
"metrics",
"reqwest-client",
"reqwest-rustls",
"trace",
] }
opentelemetry_sdk = { version = "0.27", optional = true, default-features = false, features = [
"http",
"logs",
"metrics",
"rt-tokio",
"trace",
"spec_unstable_logs_enabled",
] }
opentelemetry-semantic-conventions = { version = "0.27", optional = true, default-features = false, features = ["semconv_experimental"] }
serde = { workspace = true }
serde_json = { workspace = true }
strfmt = { workspace = true }
tokio = { workspace = true, features = ["full"] }
tokio-util = { workspace = true }
tokio-stream = { workspace = true }
tonic = { workspace = true }
tracing = { workspace = true }
tracing-subscriber = { workspace = true, optional = true }
tracing = { workspace = true, features = ["attributes", "std"] }
tracing-core = { version = "0.1", optional = true, default-features = false, features = ["std"] }
tracing-log = { version = "0.2", optional = true, default-features = false, features = ["log-tracer", "std"] }
tracing-opentelemetry = { version = "0.28.0", optional = true, default-features = false, features = ["metrics"] }
tracing-subscriber = { workspace = true, optional = true, default-features = false }

[dev-dependencies]
portpicker = { workspace = true }
Expand All @@ -40,6 +62,26 @@ default = ["setup-tracing"]
api-client-tracing = ["shuttle-api-client/tracing"]

setup-tracing = [
"tracing-subscriber/default",
"tracing-subscriber/ansi",
"tracing-subscriber/env-filter",
"tracing-subscriber/fmt",
"tracing-subscriber/smallvec",
"tracing-subscriber/std",
"tracing-subscriber/tracing-log",
]
setup-telemetry = [
"setup-tracing",
"dep:log",
"dep:opentelemetry",
"dep:opentelemetry-otlp",
"dep:opentelemetry_sdk",
"dep:opentelemetry-semantic-conventions",
"dep:tracing-core",
"dep:tracing-log",
"dep:tracing-opentelemetry",
"tracing-subscriber/alloc",
"tracing-subscriber/parking_lot",
"tracing-subscriber/registry",
"tracing-subscriber/tracing",
"tracing-subscriber/tracing-serde",
]
16 changes: 8 additions & 8 deletions runtime/src/alpha.rs
Original file line number Diff line number Diff line change
Expand Up @@ -118,7 +118,7 @@ where
}
},
Err(error) => {
if error.is_panic() {
return if error.is_panic() {
let panic = error.into_panic();
let msg = match panic.downcast_ref::<String>() {
Some(msg) => msg.to_string(),
Expand All @@ -128,18 +128,18 @@ where
},
};
println!("loading service panicked: {msg}");
return Ok(Response::new(LoadResponse {
Ok(Response::new(LoadResponse {
success: false,
message: msg,
resources: vec![],
}));
}))
} else {
println!("loading service crashed: {error:#}");
return Ok(Response::new(LoadResponse {
Ok(Response::new(LoadResponse {
success: false,
message: error.to_string(),
resources: vec![],
}));
}))
}
}
};
Expand Down Expand Up @@ -214,7 +214,7 @@ where

println!("Starting on {service_address}");

let (kill_tx, kill_rx) = tokio::sync::oneshot::channel();
let (kill_tx, kill_rx) = oneshot::channel();
*self.kill_tx.lock().unwrap() = Some(kill_tx);

let handle = tokio::runtime::Handle::current();
Expand Down Expand Up @@ -296,7 +296,7 @@ where
} else {
println!("failed to stop deployment");

Ok(tonic::Response::new(StopResponse { success: false }))
Ok(Response::new(StopResponse { success: false }))
}
}

Expand Down Expand Up @@ -324,7 +324,7 @@ where
Ok(Response::new(ReceiverStream::new(rx)))
}

async fn version(&self, _requset: Request<Ping>) -> Result<Response<VersionInfo>, Status> {
async fn version(&self, _request: Request<Ping>) -> Result<Response<VersionInfo>, Status> {
Ok(Response::new(VersionInfo {
version: crate::VERSION_STRING.to_owned(),
}))
Expand Down
7 changes: 5 additions & 2 deletions runtime/src/lib.rs
Original file line number Diff line number Diff line change
Expand Up @@ -12,6 +12,9 @@ mod plugins;
mod rt;
mod start;

#[cfg(feature = "setup-telemetry")]
mod trace;

// Public API
pub use plugins::{Metadata, Secrets};
pub use shuttle_codegen::main;
Expand Down Expand Up @@ -51,7 +54,7 @@ pub mod __internals {
O: Future<Output = Result<Vec<Vec<u8>>, Error>> + Send,
{
async fn load(self, factory: ResourceFactory) -> Result<Vec<Vec<u8>>, Error> {
(self)(factory).await
self(factory).await
}
}

Expand All @@ -72,7 +75,7 @@ pub mod __internals {
type Service = S;

async fn run(self, resources: Vec<Vec<u8>>) -> Result<Self::Service, Error> {
(self)(resources).await
self(resources).await
}
}
}
4 changes: 2 additions & 2 deletions runtime/src/rt.rs
Original file line number Diff line number Diff line change
Expand Up @@ -93,7 +93,7 @@ pub async fn start(loader: impl Loader + Send + 'static, runner: impl Runner + S
// light hyper server
let make_service = make_service_fn(|_conn| async {
Ok::<_, Infallible>(service_fn(|_req| async move {
trace!("Receivied health check");
trace!("Received health check");
// TODO: A hook into the `Service` trait can be added here
trace!("Responding to health check");
Result::<Response<Body>, hyper::Error>::Ok(Response::new(Body::empty()))
Expand Down Expand Up @@ -130,7 +130,7 @@ pub async fn start(loader: impl Loader + Send + 'static, runner: impl Runner + S
// Sort secrets by key
let secrets = BTreeMap::from_iter(secrets.into_iter().map(|(k, v)| (k, Secret::new(v))));

// TODO: rework resourcefactory
// TODO: rework `ResourceFactory`
let factory = ResourceFactory::new(project_name, secrets.clone(), env);
let mut resources = match loader.load(factory).await {
Ok(r) => r,
Expand Down
45 changes: 30 additions & 15 deletions runtime/src/start.rs
Original file line number Diff line number Diff line change
Expand Up @@ -48,7 +48,14 @@ impl Args {
}
}

pub async fn start(loader: impl Loader + Send + 'static, runner: impl Runner + Send + 'static) {
pub async fn start(
loader: impl Loader + Send + 'static,
runner: impl Runner + Send + 'static,
#[cfg_attr(not(feature = "setup-telemetry"), allow(unused_variables))]
project_name: &'static str,
#[cfg_attr(not(feature = "setup-telemetry"), allow(unused_variables))]
project_version: &'static str,
) {
// `--version` overrides any other arguments. Used by cargo-shuttle to check compatibility on local runs.
if std::env::args().any(|arg| arg == "--version") {
println!("{}", crate::VERSION_STRING);
Expand All @@ -69,33 +76,41 @@ pub async fn start(loader: impl Loader + Send + 'static, runner: impl Runner + S
};

// this is handled after arg parsing to not interfere with --version above
#[cfg(feature = "setup-tracing")]
#[cfg(all(feature = "setup-tracing", not(feature = "setup-telemetry")))]
{
use tracing_subscriber::{fmt, prelude::*, registry, EnvFilter};
registry()
.with(fmt::layer().without_time())
.with(
// let user override RUST_LOG in local run if they want to
EnvFilter::try_from_default_env()
EnvFilter::try_from_default_env().unwrap_or_else(|_| {
// otherwise use our default
.or_else(|_| {
EnvFilter::try_new(if args.beta {
Into::into(format!(
"{},{}=debug",
if args.beta {
"info"
} else {
"info,shuttle=trace"
})
})
.unwrap(),
},
project_name
))
}),
)
.init();
}

if args.beta {
tracing::warn!(
"Default tracing subscriber initialized (https://docs.shuttle.dev/docs/logs)"
);
} else {
tracing::warn!("Default tracing subscriber initialized (https://docs.shuttle.rs/configuration/logs)");
}
#[cfg(feature = "setup-telemetry")]
let _guard = crate::trace::init_tracing_subscriber(project_name, project_version);

#[cfg(any(feature = "setup-tracing", feature = "setup-telemetry"))]
if args.beta {
tracing::warn!(
"Default tracing subscriber initialized (https://docs.shuttle.dev/docs/logs)"
);
} else {
tracing::warn!(
"Default tracing subscriber initialized (https://docs.shuttle.rs/configuration/logs)"
);
}

if args.beta {
Expand Down
Loading