Refactor CLI organisation, and improve termination (googleforgames#570)

docs/src/filters/writing_custom_filters.md

@@ -73,10 +73,11 @@ impl StaticFilter for Greet {
 
 ## Running
 
-We can run the proxy in the exact manner as the default Quilkin binary using the
-[run][runner::run] function, passing in our custom [FilterFactory]. Let's add a
-main function that does that. Quilkin relies on the [Tokio] async runtime, so we
-need to import that crate and wrap our main function with it.
+We can run the proxy using [`Server::TryFrom`][Server::TryFrom] function. Let's
+add a main function that does that. Quilkin relies on the [Tokio] async
+runtime, so we need to import that crate and wrap our main function with it.
+
+We can also register custom filters in quilkin using [`FilterRegistry::register`][FilterRegistry::register]
 
 Add Tokio as a dependency in `Cargo.toml`.
 
@@ -241,7 +242,8 @@ filter. Try it out with the following configuration:
 [FilterFactory]: ../../api/quilkin/filters/trait.FilterFactory.html
 [filter-factory-name]: ../../api/quilkin/filters/trait.FilterFactory.html#tymethod.name
 [FilterRegistry]: ../../api/quilkin/filters/struct.FilterRegistry.html
-[runner::run]: ../../api/quilkin/runner/fn.run.html
+[FilterRegistry::register]: ../../api/quilkin/filters/struct.FilterRegistry.html#method.register
+[Server::try_from]: ../../api/struct.Server.html#impl-TryFrom%3CConfig%3E
 [CreateFilterArgs::config]: ../../api/quilkin/filters/prelude/struct.CreateFilterArgs.html#structfield.config
 [ConfigType::dynamic]: ../../api/quilkin/config/enum.ConfigType.html#variant.Dynamic
 [ConfigType::static]: ../../api/quilkin/config/enum.ConfigType.html#variant.Static

examples/quilkin-filter-example/Cargo.toml

@@ -21,7 +21,7 @@ name = "quilkin-filter-example"
 version = "0.1.0"
 homepage = "https://github.com/googleforgames/quilkin"
 repository = "https://github.com/googleforgames/quilkin"
-edition = "2018"
+edition = "2021"
 
 [dependencies]
 # If lifting this example, you will want to be explicit about the Quilkin version, e.g.

examples/quilkin-filter-example/src/main.rs

@@ -90,22 +90,22 @@ impl StaticFilter for Greet {
 
 // ANCHOR: run
 #[tokio::main]
-async fn main() {
-    quilkin::run(
-        quilkin::Config::builder()
-            .port(7001)
-            .filters(vec![quilkin::config::Filter {
-                name: Greet::NAME.into(),
-                config: None,
-            }])
-            .endpoints(vec![quilkin::endpoint::Endpoint::new(
-                (std::net::Ipv4Addr::LOCALHOST, 4321).into(),
-            )])
-            .build()
-            .unwrap(),
-        vec![Greet::factory()].into_iter(),
-    )
-    .await
-    .unwrap();
+async fn main() -> quilkin::Result<()> {
+    quilkin::filters::FilterRegistry::register(vec![Greet::factory()].into_iter());
+
+    let (_shutdown_tx, shutdown_rx) = tokio::sync::watch::channel(());
+    let server: quilkin::Server = quilkin::Config::builder()
+        .port(7001)
+        .filters(vec![quilkin::config::Filter {
+            name: Greet::NAME.into(),
+            config: None,
+        }])
+        .endpoints(vec![quilkin::endpoint::Endpoint::new(
+            (std::net::Ipv4Addr::LOCALHOST, 4321).into(),
+        )])
+        .build()?
+        .try_into()?;
+
+    server.run(shutdown_rx).await
 }
 // ANCHOR_END: run

src/cli.rs

@@ -0,0 +1,102 @@
+/*
+ * Copyright 2022 Google LLC
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+mod generate_config_schema;
+mod manage;
+mod run;
+
+use std::path::PathBuf;
+
+use crate::Config;
+
+const VERSION: &str = env!("CARGO_PKG_VERSION");
+
+#[derive(clap::Parser)]
+pub struct Cli {
+    #[clap(
+        short,
+        long,
+        env = "QUILKIN_CONFIG",
+        default_value = "quilkin.yaml",
+        help = "The YAML configuration file."
+    )]
+    config: PathBuf,
+    #[clap(
+        short,
+        long,
+        env,
+        help = "Whether Quilkin will report any results to stdout/stderr."
+    )]
+    quiet: bool,
+    #[clap(subcommand)]
+    command: Commands,
+}
+
+#[derive(clap::Subcommand)]
+enum Commands {
+    Run(run::Run),
+    GenerateConfigSchema(generate_config_schema::GenerateConfigSchema),
+    Manage(manage::Manage),
+}
+
+impl Cli {
+    /// Drives the main quilkin application lifecycle using the command line
+    /// arguments.
+    pub async fn drive(self) -> crate::Result<()> {
+        let version: std::borrow::Cow<'static, str> = if cfg!(debug_assertions) {
+            format!("{VERSION}+debug").into()
+        } else {
+            VERSION.into()
+        };
+
+        if !self.quiet {
+            let env_filter = tracing_subscriber::EnvFilter::builder()
+                .with_default_directive(tracing_subscriber::filter::LevelFilter::INFO.into())
+                .from_env_lossy();
+            tracing_subscriber::fmt()
+                .json()
+                .with_env_filter(env_filter)
+                .init();
+        }
+
+        tracing::info!(
+            version = &*version,
+            commit = crate::metadata::build::GIT_COMMIT_HASH,
+            "Starting Quilkin"
+        );
+
+        match &self.command {
+            Commands::Run(runner) => runner.run(&self).await,
+            Commands::Manage(manager) => manager.manage(&self).await,
+            Commands::GenerateConfigSchema(generator) => generator.generate_config_schema(),
+        }
+    }
+
+    /// Searches for the configuration file, and panics if not found.
+    fn read_config(&self) -> Config {
+        std::fs::File::open(&self.config)
+            .or_else(|error| {
+                if cfg!(unix) {
+                    std::fs::File::open("/etc/quilkin/quilkin.yaml")
+                } else {
+                    Err(error)
+                }
+            })
+            .map_err(eyre::Error::from)
+            .and_then(|file| Config::from_reader(file).map_err(From::from))
+            .unwrap()
+    }
+}

src/cli/generate_config_schema.rs

@@ -0,0 +1,70 @@
+/*
+ * Copyright 2022 Google LLC
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+#[derive(clap::Args)]
+pub struct GenerateConfigSchema {
+    #[clap(
+        short,
+        long,
+        default_value = ".",
+        help = "The directory to write configuration files."
+    )]
+    output_directory: std::path::PathBuf,
+    #[clap(
+        min_values = 1,
+        default_value = "all",
+        help = "A list of one or more filter IDs to generate or 'all' to generate all available filter schemas."
+    )]
+    filter_ids: Vec<String>,
+}
+
+impl GenerateConfigSchema {
+    pub fn generate_config_schema(&self) -> crate::Result<()> {
+        let set = crate::filters::FilterSet::default();
+        type SchemaIterator<'r> =
+            Box<dyn Iterator<Item = (&'static str, schemars::schema::RootSchema)> + 'r>;
+
+        let schemas = (self.filter_ids.len() == 1 && self.filter_ids[0].to_lowercase() == "all")
+            .then(|| {
+                Box::new(
+                    set.iter()
+                        .map(|factory| (factory.name(), factory.config_schema())),
+                ) as SchemaIterator
+            })
+            .unwrap_or_else(|| {
+                Box::new(self.filter_ids.iter().filter_map(|id| {
+                    let item = set.get(id);
+
+                    if item.is_none() {
+                        tracing::error!("{id} not found in filter set.");
+                    }
+
+                    item.map(|item| (item.name(), item.config_schema()))
+                })) as SchemaIterator
+            });
+
+        for (id, schema) in schemas {
+            let mut path = self.output_directory.join(id);
+            path.set_extension("yaml");
+
+            tracing::info!("Writing {id} schema to {}", path.display());
+
+            std::fs::write(path, serde_yaml::to_string(&schema)?)?;
+        }
+
+        Ok(())
+    }
+}

src/cli/manage.rs

@@ -0,0 +1,68 @@
+/*
+ * Copyright 2022 Google LLC
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+#[derive(clap::Args)]
+pub struct Manage {
+    #[clap(subcommand)]
+    provider: Providers,
+}
+
+#[derive(clap::Subcommand)]
+enum Providers {
+    Agones {
+        #[clap(
+            short,
+            long,
+            default_value = "default",
+            help = "Namespace under which the proxies run."
+        )]
+        config_namespace: String,
+        #[clap(
+            short,
+            long,
+            default_value = "default",
+            help = "Namespace under which the game servers run."
+        )]
+        gameservers_namespace: String,
+    },
+
+    File,
+}
+
+impl Manage {
+    pub async fn manage(&self, cli: &crate::Cli) -> crate::Result<()> {
+        let config = std::sync::Arc::new(cli.read_config());
+
+        let provider_task = match &self.provider {
+            Providers::Agones {
+                gameservers_namespace,
+                config_namespace,
+            } => tokio::spawn(crate::config::watch::agones(
+                gameservers_namespace.clone(),
+                config_namespace.clone(),
+                config.clone(),
+            )),
+            Providers::File => {
+                tokio::spawn(crate::config::watch::fs(config.clone(), cli.config.clone()))
+            }
+        };
+
+        tokio::select! {
+            result = crate::xds::server::spawn(config) => result,
+            result = provider_task => result.map_err(From::from).and_then(|result| result),
+        }
+    }
+}

src/cli/run.rs

@@ -0,0 +1,68 @@
+/*
+ * Copyright 2021 Google LLC
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+use tokio::{signal, sync::watch};
+use tracing::{debug, info, span, Level};
+
+#[cfg(doc)]
+use crate::filters::FilterFactory;
+
+#[derive(clap::Args)]
+pub struct Run {}
+
+impl Run {
+    /// Start and run a proxy. Any passed in [`FilterFactory`]s are included
+    /// alongside the default filter factories.
+    pub async fn run(&self, cli: &crate::Cli) -> crate::Result<()> {
+        let config = cli.read_config();
+        let span = span!(Level::INFO, "source::run");
+        let _enter = span.enter();
+
+        let server = crate::Server::try_from(config)?;
+
+        #[cfg(target_os = "linux")]
+        let mut sig_term_fut = signal::unix::signal(signal::unix::SignalKind::terminate())?;
+
+        let (shutdown_tx, shutdown_rx) = watch::channel::<()>(());
+        tokio::spawn(async move {
+            #[cfg(target_os = "linux")]
+            let sig_term = sig_term_fut.recv();
+            #[cfg(not(target_os = "linux"))]
+            let sig_term = std::future::pending();
+
+            tokio::select! {
+                _ = signal::ctrl_c() => {
+                    debug!("Received SIGINT")
+                }
+                _ = sig_term => {
+                    debug!("Received SIGTERM")
+                }
+            }
+
+            info!("Shutting down");
+            // Don't unwrap in order to ensure that we execute
+            // any subsequent shutdown tasks.
+            shutdown_tx.send(()).ok();
+        });
+
+        if let Err(err) = server.run(shutdown_rx).await {
+            info! (error = %err, "Shutting down with error");
+            Err(err)
+        } else {
+            Ok(())
+        }
+    }
+}