Replace grpcio with tonic (#2112)

* Begin replacing grpio with tonic

* "Finish" porting to tonic

* Add metadata fields to manifest

* Allow optional port assignment

* Fix field name

* Remove unused generic

* Print out error when unable to get game server

* Pull out health check into its own separate task

* Update rust-simple example to work with updated SDK

* Remove unused generic parameter

* Fix rust tests

* Remove unneeded cruft from rust build image

* Remove lines causing excessive rebuilds

* Fix rust CI test, add prefix to log messages

The logs from CD runs are incredibly hard to read since they are just
intermixed with all tests, so I prefixed all of the print messages from
the rust example so I can actually read the output more easily.

* Address PR feedback

* Update rust SDK
documentation

* spawn_health_task => health_check

Removed the auto health check task since that actually made health
checking meaningless since it's _supposed_ to be sent by the application
itself when it working properly, not automatically

* Use internal timeout for SDK connection

* Do continuous health checks

This method should be better than the previous test/examples as the
health check will actually be canceled if the task/thread the health
check is owned by is canceled for any reason

* Copy protobuffers before build

* Bump example version

* Add trailing newline

* Ensure protos are copied for the test

* Vendor protos again

build/build-sdk-images/rust/Dockerfile

@@ -31,25 +31,6 @@ RUN wget -q https://static.rust-lang.org/rustup/dist/x86_64-unknown-linux-gnu/ru
     cargo --version; \
     rustc --version;
 
-# install rust tooling for SDK generation
-RUN cargo install protobuf-codegen --vers 2.16.2
-RUN cargo install grpcio-compiler --vers 0.6.0
-
-RUN wget -q https://cmake.org/files/v3.14/cmake-3.14.1-Linux-x86_64.sh
-RUN mkdir /opt/cmake
-RUN sh cmake-3.14.1-Linux-x86_64.sh --prefix=/opt/cmake --skip-license
-RUN ln -s /opt/cmake/bin/cmake /usr/local/bin/cmake
-RUN cmake --version
-
-ENV GO_VERSION=1.10.2 \
-    GO_CHECKSUM=4b677d698c65370afa33757b6954ade60347aaca310ea92a63ed717d7cb0c2ff
-RUN mkdir -p /usr/local/go \
-  && curl -fSO https://dl.google.com/go/go${GO_VERSION}.linux-amd64.tar.gz \
-  && shasum -a 256 go${GO_VERSION}.linux-amd64.tar.gz | grep ${GO_CHECKSUM} \
-  && tar xf go${GO_VERSION}.linux-amd64.tar.gz -C /usr/local/go --strip-components=1 \
-  && rm -f go${GO_VERSION}.linux-amd64.tar.gz
-ENV PATH $PATH:/usr/local/go/bin
-
 # code generation scripts
 COPY *.sh /root/
 RUN chmod +x /root/*.sh

build/build-sdk-images/rust/build-sdk-test.sh

@@ -15,6 +15,7 @@
 # limitations under the License.
 
 set -ex
+
 mkdir -p /go/src/agones.dev/agones/test/sdk/rust/.cargo
 mkdir -p /go/src/agones.dev/agones/test/sdk/rust/.cargo-targets
 cd /go/src/agones.dev/agones/test/sdk/rust

build/build-sdk-images/rust/gen.sh

@@ -16,25 +16,12 @@
 
 set -ex
 
-googleapis=/go/src/agones.dev/agones/proto/googleapis
+protos=/go/src/agones.dev/agones/proto
+dest=/go/src/agones.dev/agones/sdks/rust
 
-sdk=/go/src/agones.dev/agones/proto/sdk
-cd /go/src/agones.dev/agones
+rm -rf ${dest}/proto
 
-protoc \
-    -I ${googleapis} -I ${sdk} sdk.proto \
-    --rust_out=sdks/rust/src/grpc --grpc_out=sdks/rust/src/grpc \
-    --plugin=protoc-gen-grpc=`which grpc_rust_plugin`
-protoc \
-    -I ${googleapis} -I ${sdk}/alpha alpha.proto \
-    --rust_out=sdks/rust/src/grpc --grpc_out=sdks/rust/src/grpc \
-    --plugin=protoc-gen-grpc=`which grpc_rust_plugin`
+echo "Copying protobuffers to rust sdk"
+cp -r ${protos} ${dest}
 
-cat ./build/boilerplate.go.txt ./sdks/rust/src/grpc/sdk.rs >> ./sdk.rs
-cat ./build/boilerplate.go.txt ./sdks/rust/src/grpc/sdk_grpc.rs >> ./sdk_grpc.rs
-cat ./build/boilerplate.go.txt ./sdks/rust/src/grpc/alpha.rs >> ./alpha.rs
-cat ./build/boilerplate.go.txt ./sdks/rust/src/grpc/alpha_grpc.rs >> ./alpha_grpc.rs
-mv ./sdk.rs ./sdks/rust/src/grpc/
-mv ./sdk_grpc.rs ./sdks/rust/src/grpc/
-mv ./alpha.rs ./sdks/rust/src/grpc/
-mv ./alpha_grpc.rs ./sdks/rust/src/grpc/
+echo "Rust code is generated at build time"

examples/rust-simple/Cargo.toml

@@ -15,6 +15,17 @@
 [package]
 name = "rust-simple"
 version = "0.1.0"
+edition = "2018"
 
 [dependencies]
 agones = { path = "../../sdks/rust" }
+
+[dependencies.tokio]
+version = "1.6"
+default-features = false
+features = [
+    "macros",
+    "rt-multi-thread",
+    "sync",
+    "time",
+]

examples/rust-simple/Makefile

@@ -27,7 +27,7 @@ REPOSITORY ?= gcr.io/agones-images
 
 mkfile_path := $(abspath $(lastword $(MAKEFILE_LIST)))
 project_path := $(dir $(mkfile_path))
-server_tag = $(REPOSITORY)/rust-simple-server:0.8
+server_tag = $(REPOSITORY)/rust-simple-server:0.9
 
 #   _____                    _
 #  |_   _|_ _ _ __ __ _  ___| |_ ___

examples/rust-simple/src/main.rs

@@ -12,25 +12,13 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.
 
-extern crate agones;
-
-use std::result::Result;
-use std::thread;
 use std::time::Duration;
 
-macro_rules! enclose {
-    ( ($( $x:ident ),*) $y:expr ) => {
-        {
-            $(let mut $x = $x.clone();)*
-            $y
-        }
-    };
-}
-
-fn main() {
+#[tokio::main(flavor = "multi_thread", worker_threads = 4)]
+async fn main() {
     println!("Rust Game Server has started!");
 
-    ::std::process::exit(match run() {
+    ::std::process::exit(match run().await {
         Ok(_) => {
             println!("Rust Game Server finished.");
             0
@@ -42,57 +30,109 @@ fn main() {
     });
 }
 
-fn run() -> Result<(), String> {
+async fn run() -> Result<(), String> {
     println!("Creating SDK instance");
-    let sdk = agones::Sdk::new().map_err(|_| "Could not connect to the sidecar. Exiting!")?;
-
-    let _health = thread::spawn(enclose! {(sdk) move || {
-        loop {
-            match sdk.health() {
-                (s, Ok(_)) => {
-                    println!("Health ping sent");
-                    sdk = s;
-                },
-                (s, Err(e)) => {
-                    println!("Health ping failed : {:?}", e);
-                    sdk = s;
+    let mut sdk = agones::Sdk::new(None /* default port */, None /* keep_alive */)
+        .await
+        .map_err(|e| format!("unable to create sdk client: {}", e))?;
+
+    // Spawn a task that will send health checks every 2 seconds. If this current
+    // thread/task panics or dropped, the health check will also be stopped
+    let _health = {
+        let health_tx = sdk.health_check();
+        let (tx, mut rx) = tokio::sync::oneshot::channel::<()>();
+
+        tokio::task::spawn(async move {
+            let mut interval = tokio::time::interval(Duration::from_secs(2));
+
+            loop {
+                tokio::select! {
+                    _ = interval.tick() => {
+                        if health_tx
+                            .send(())
+                            .await.is_err() {
+                            eprintln!("Health check receiver was dropped");
+                            break;
+                        }
+                    }
+                    _ = &mut rx => {
+                        println!("Health check task canceled");
+                        break;
+                    }
                 }
             }
-            thread::sleep(Duration::from_secs(2));
-        }
-    }});
+        });
+
+        tx
+    };
 
-    let _watch = thread::spawn(enclose! {(sdk) move || {
-        println!("Starting to watch GameServer updates...");
-        let _ = sdk.watch_gameserver(|gameserver| {
-            println!("GameServer Update, name: {}", gameserver.object_meta.unwrap().name);
-            println!("GameServer Update, state: {}", gameserver.status.unwrap().state);
+    let _watch = {
+        let mut watch_client = sdk.clone();
+        let (tx, mut rx) = tokio::sync::oneshot::channel::<()>();
+
+        tokio::task::spawn(async move {
+            println!("Starting to watch GameServer updates...");
+            match watch_client.watch_gameserver().await {
+                Err(e) => println!("Failed to watch for GameServer updates: {}", e),
+                Ok(mut stream) => loop {
+                    tokio::select! {
+                        gs = stream.message() => {
+                            match gs {
+                                Ok(Some(gs)) => {
+                                    println!("GameServer Update, name: {}", gs.object_meta.unwrap().name);
+                                    println!("GameServer Update, state: {}", gs.status.unwrap().state);
+                                }
+                                Ok(None) => {
+                                    println!("Server closed the GameServer watch stream");
+                                    break;
+                                }
+                                Err(e) => {
+                                    eprintln!("GameServer Update stream encountered an error: {}", e);
+                                }
+                            }
+
+                        }
+                        _ = &mut rx => {
+                            println!("Shutting down GameServer watch loop");
+                            break;
+                        }
+                    }
+                },
+            }
         });
-    }});
+
+        tx
+    };
 
     println!("Setting a label");
     sdk.set_label("test-label", "test-value")
+        .await
         .map_err(|e| format!("Could not run SetLabel(): {}. Exiting!", e))?;
 
     println!("Setting an annotation");
     sdk.set_annotation("test-annotation", "test value")
+        .await
         .map_err(|e| format!("Could not run SetAnnotation(): {}. Exiting!", e))?;
 
     println!("Marking server as ready...");
     sdk.ready()
+        .await
         .map_err(|e| format!("Could not run Ready(): {}. Exiting!", e))?;
 
     println!("...marked Ready");
 
     println!("Setting as Reserved for 5 seconds");
-    sdk.reserve(Duration::new(5, 0)).map_err(|e| format!("Could not run Reserve(): {}. Exiting!", e))?;
+    sdk.reserve(Duration::from_secs(5))
+        .await
+        .map_err(|e| format!("Could not run Reserve(): {}. Exiting!", e))?;
     println!("...Reserved");
 
-    thread::sleep(Duration::new(6, 0));
+    tokio::time::sleep(Duration::from_secs(6)).await;
 
     println!("Getting GameServer details...");
     let gameserver = sdk
         .get_gameserver()
+        .await
         .map_err(|e| format!("Could not run GameServer(): {}. Exiting!", e))?;
 
     println!("GameServer name: {}", gameserver.object_meta.unwrap().name);
@@ -101,11 +141,12 @@ fn run() -> Result<(), String> {
         let time = i * 10;
         println!("Running for {} seconds", time);
 
-        thread::sleep(Duration::from_secs(10));
+        tokio::time::sleep(Duration::from_secs(10)).await;
 
         if i == 5 {
             println!("Shutting down after 60 seconds...");
             sdk.shutdown()
+                .await
                 .map_err(|e| format!("Could not run Shutdown: {}. Exiting!", e))?;
             println!("...marked for Shutdown");
         }

sdks/rust/.gitignore

@@ -13,5 +13,4 @@ Cargo.lock
 # These are backup files generated by rustfmt
 **/*.rs.bk
 
-
 # End of https://www.gitignore.io/api/rust

sdks/rust/Cargo.toml

@@ -16,14 +16,35 @@
 name = "agones"
 version = "0.1.0"
 edition = "2018"
-
-[features]
-openssl = ["grpcio/openssl"]
-openssl-vendored = ["grpcio/openssl-vendored"]
+license = "Apache-2.0"
+repository = "https://github.com/googleforgames/agones"
+documentation = "https://docs.rs/agones"
+homepage = "https://agones.dev/site/"
 
 [dependencies]
-grpcio = "0.6.0"
-protobuf = "2.16.2"
-futures = { version = "0.3", features = ["compat"] }
-futures01 = "0.1"
-error-chain = "0.12.3"
+async-stream = "0.3"
+http = "0.2"
+prost = "0.7"
+thiserror = "1.0"
+
+[dependencies.tokio]
+version = "1.0"
+default-features = false
+features = ["sync", "time"]
+
+[dependencies.tonic]
+version = "0.4"
+default-features = false
+features = [
+    "codegen",
+    "transport",
+    "prost",
+]
+
+[build-dependencies.tonic-build]
+version = "0.4"
+default-features = false
+features = [
+    "prost",
+    "transport",
+]

sdks/rust/build.rs

@@ -0,0 +1,10 @@
+fn main() {
+    tonic_build::configure()
+        // The SDK is just a client, no need to build the server types
+        .build_server(false)
+        .compile(
+            &["proto/sdk/alpha/alpha.proto", "proto/sdk/sdk.proto"],
+            &["proto/googleapis", "proto/sdk/alpha", "proto/sdk"],
+        )
+        .expect("failed to compile protobuffers");
+}