wrangler tail using WebSockets (#2005)

`wrangler tail` will no longer use `cloudflared`, instead it will make a WebSocket connection to an endpoint, powered by Durable Objects, to receive a stream of logs. This also adds support for filtering logs, which we will be announced soon.

Cargo.toml

@@ -29,6 +29,7 @@ env_logger = "0.8.4"
 eventual = "0.1.7"
 flate2 = "1.0.18"
 fs2 = "0.4.3"
+futures = "0.3.15"
 futures-util = "0.3"
 globset = "0.4.6"
 http = "0.2.1"
@@ -54,6 +55,7 @@ semver = "1.0.3"
 serde = { version = "1.0", features = ["derive"] }
 serde_json = "1.0.60"
 serde_with = "1.5.1"
+erased-serde = "0.3"
 structopt = "0.3.21"
 sys-info = "0.9"
 tempfile = "3.1.0"

src/cli/mod.rs

@@ -32,8 +32,10 @@ pub mod exec {
 
 use std::net::IpAddr;
 use std::path::PathBuf;
+use std::str::FromStr;
 
 use crate::commands::dev::Protocol;
+use crate::commands::tail::websocket::TailFormat;
 use crate::preview::HttpMethod;
 use crate::settings::toml::migrations::{
     DurableObjectsMigration, Migration, MigrationConfig, Migrations, RenameClass, TransferClass,
@@ -215,23 +217,59 @@ pub enum Command {
     #[structopt(name = "whoami")]
     Whoami,
 
-    /// Aggregate logs from production worker
+    /// View a stream of logs from a published worker
     #[structopt(name = "tail")]
     Tail {
-        /// Specify an output format
+        /// Name of the worker to tail
+        #[structopt(index = 1)]
+        name: Option<String>,
+
+        /// Output format for log messages
         #[structopt(long, short = "f", default_value = "json", possible_values = &["json", "pretty"])]
-        format: String,
+        format: TailFormat,
+
+        /// Stops the tail after receiving the first log (useful for testing)
+        #[structopt(long)]
+        once: bool,
+
+        /// Adds a sampling rate (0.01 for 1%)
+        #[structopt(long = "sampling-rate", default_value = "1")]
+        sampling_rate: f64,
+
+        /// Filter by invocation status
+        #[structopt(long, possible_values = &["ok", "error", "canceled"])]
+        status: Vec<String>,
+
+        /// Filter by HTTP method
+        #[structopt(long)]
+        method: Vec<String>,
 
-        /// Port to accept tail log requests
-        #[structopt(long = "port", short = "p")]
+        /// Filter by HTTP header
+        #[structopt(long)]
+        header: Vec<String>,
+
+        /// Filter by IP address ("self" to filter your own IP address)
+        #[structopt(long = "ip-address", parse(try_from_str = parse_ip_address))]
+        ip_address: Vec<String>,
+
+        /// Filter by a text match in console.log messages
+        #[structopt(long)]
+        search: Option<String>,
+
+        /// Set the URL to forward log messages
+        #[structopt(hidden = true)]
+        url: Option<Url>,
+
+        /// Deprecated, no longer used.
+        #[structopt(hidden = true, long = "port", short = "p")]
         tunnel_port: Option<u16>,
 
-        /// Provides endpoint for cloudflared metrics. Used to retrieve tunnel url
-        #[structopt(long = "metrics")]
+        /// Deprecated, no longer used.
+        #[structopt(hidden = true, long = "metrics")]
         metrics_port: Option<u16>,
     },
 
-    /// Authenticate Wrangler with your Cloudflare username and password
+    /// Authenticate wrangler with your Cloudflare username and password
     #[structopt(name = "login")]
     Login,
 
@@ -318,6 +356,16 @@ impl AdhocMigration {
     }
 }
 
+fn parse_ip_address(input: &str) -> Result<String, anyhow::Error> {
+    match input {
+        "self" => Ok(String::from("self")),
+        address => match IpAddr::from_str(address) {
+            Ok(_) => Ok(address.to_owned()),
+            Err(err) => anyhow::bail!("{}: {}", err, input),
+        },
+    }
+}
+
 #[cfg(test)]
 mod tests {
     use super::*;

src/cli/tail.rs

@@ -1,25 +1,69 @@
 use super::Cli;
 use crate::commands;
+use crate::commands::tail::filter::*;
+use crate::commands::tail::websocket::{TailFormat, TailOptions};
 use crate::settings::{global_user::GlobalUser, toml::Manifest};
 
 use anyhow::Result;
+use url::Url;
 
+#[allow(clippy::too_many_arguments)]
 pub fn tail(
-    format: String,
-    tunnel_port: Option<u16>,
-    metrics_port: Option<u16>,
+    name: Option<String>,
+    url: Option<Url>,
+    format: TailFormat,
+    once: bool,
+    sampling_rate: f64,
+    outcomes: Vec<String>,
+    methods: Vec<String>,
+    headers: Vec<String>,
+    client_ips: Vec<String>,
+    search: Option<String>,
     cli_params: &Cli,
 ) -> Result<()> {
+    let user = GlobalUser::new()?;
+
+    // FIXME: If `name` is defined, allow the command to be run outside a `wrangler.toml` directory.
     let manifest = Manifest::new(&cli_params.config)?;
     let target = manifest.get_target(cli_params.environment.as_deref(), false)?;
-    let user = GlobalUser::new()?;
+    let account_id = target.account_id.load()?.to_string();
+    let script_name = name.unwrap_or(target.name);
+
+    let mut filters: Vec<Box<dyn TraceFilter>> = vec![];
+    if !outcomes.is_empty() {
+        filters.push(Box::new(OutcomeFilter::from(outcomes)));
+    }
+    if !methods.is_empty() {
+        filters.push(Box::new(MethodFilter::from(methods)));
+    }
+    if !client_ips.is_empty() {
+        filters.push(Box::new(ClientIpFilter::from(client_ips)));
+    }
+    for header in headers.into_iter() {
+        filters.push(Box::new(HeaderFilter::from(header)))
+    }
+    if let Some(query) = search {
+        filters.push(Box::new(QueryFilter::from(query)));
+    };
+    if sampling_rate < 1.0 && sampling_rate > 0.0 {
+        filters.push(Box::new(SamplingRateFilter::from(sampling_rate))); // Should always be last
+    };
+
+    let tail = commands::tail::run(
+        user,
+        account_id,
+        script_name,
+        url,
+        TailOptions {
+            once,
+            format,
+            filters,
+        },
+    );
 
-    commands::tail::start(
-        &target,
-        &user,
-        format,
-        tunnel_port,
-        metrics_port,
-        cli_params.verbose,
-    )
+    tokio::runtime::Builder::new_current_thread()
+        .enable_all()
+        .build()
+        .unwrap()
+        .block_on(tail)
 }