feat: add creating actor by build tags, upgrading actors (#1388)

<!-- Please make sure there is an issue that this PR is correlated to. -->
Fixes RVT-4130
## Changes

<!-- If there are frontend changes, please include screenshots. -->

docker/dev-full/README.md

@@ -20,7 +20,8 @@ Start the cluster with:
 docker-compose -f docker/dev-full/docker-compose.yml up -d
 ```
 
-This will start the cluster in detached mode. Once complete, visit the dashboard at [http://localhost:8080](http://localhost:8080).
+This will start the cluster in detached mode. Once complete, visit the dashboard at
+[http://localhost:8080](http://localhost:8080).
 
 To test creating an actor end-to-end, run:
 
@@ -83,16 +84,16 @@ docker-compose -f docker/dev-full/docker-compose.yml logs -f {name}
 #### Grep
 
 It's common to use grep (or the more modern
-[ripgrep](https://www.google.com/search?q=ripgrep&oq=ripgrep&sourceid=chrome&ie=UTF-8))
-to filter logs.
+[ripgrep](https://www.google.com/search?q=ripgrep&oq=ripgrep&sourceid=chrome&ie=UTF-8)) to filter logs.
 
 For example, to find all errors in `rivet-server` with the 10 preceding lines, run:
 
 ```bash
 docker-compose -f docker/dev-full/docker-compose.yml logs rivet-server | grep -B 10 level=error
 ```
 
-Logs for `rivet-server` and `rivet-client` can also be configured via the environment. See [here](TODO) for more information.
+Logs for `rivet-server` and `rivet-client` can also be configured via the environment. See [here](TODO) for
+more information.
 
 ## Troubleshooting
 
@@ -105,4 +106,3 @@ docker-compose -f docker/dev-full/docker-compose.yml down -v -t 0
 ```
 
 This will destroy all containers & volumes immediately.
-

docker/dev-full/clickhouse/config.xml

@@ -48,4 +48,3 @@
         <shard>01</shard>
     </macros>
 </clickhouse>
-

docker/dev-full/client.Dockerfile

@@ -23,4 +23,3 @@ RUN DEBIAN_FRONTEND=noninteractive apt-get update -y && apt-get install -y --no-
 COPY --from=builder /app/dist/rivet-client /app/dist/rivet-isolate-v8-runner /app/dist/rivet-container-runner /usr/local/bin/
 ENTRYPOINT ["rivet-client"]
 CMD ["-c", "/etc/rivet-client/config.json"]
-

docker/dev-full/docker-compose.yml

@@ -74,7 +74,7 @@ services:
       # individual port converts to an iptable rule when using Docker, which
       # can cause unexpected side effects. This limits the number of actors
       # using host networking to 100.
-     - "20000-20100:20000-20100"
+      - "20000-20100:20000-20100"
     networks:
       - rivet-network
 

docker/dev-full/rivet-client/config.yaml

@@ -11,7 +11,7 @@ client:
   network:
     bind_ip: 127.0.0.1
     lan_ip: 127.0.0.1
-    # Point to lcoalhost since this is a dev instance
+    # Point to localhost since this is a dev instance
     wan_ip: 127.0.0.1
     # Corresponds to the port range configured in the `docker-compose.yml`
     wan_port_range_min: 20000
@@ -23,6 +23,6 @@ client:
     cpu: 0
     memory: 0
   logs:
-    redirect_logs: false
+    redirect_logs: true
   vector:
     address: vector-server:6100

docker/dev-full/rivet-server/config.yaml

@@ -50,4 +50,3 @@ server:
       -----BEGIN PRIVATE KEY-----
       MC4CAQAwBQYDK2VwBCIEIDI+WHFytxvdtfGot36NMCI26s2Yp0+E5u9OiPf3NQX3
       -----END PRIVATE KEY-----
-

docker/dev-full/server.Dockerfile

@@ -40,4 +40,3 @@ RUN \
 	RUSTFLAGS="--cfg tokio_unstable" cargo build --bin rivet-server && \
 	mv target/debug/rivet-server /usr/bin/rivet-server && \
 	mkdir /etc/rivet-server
-

packages/api/actor/src/route/actors.rs

@@ -101,13 +101,14 @@ pub async fn create(
 ) -> GlobalResult<models::ActorCreateActorResponse> {
 	let CheckOutput { game_id, env_id } = ctx.auth().check(ctx.op_ctx(), &query, false).await?;
 
-	let (clusters_res, game_configs_res) = tokio::try_join!(
+	let (clusters_res, game_configs_res, build_id) = tokio::try_join!(
 		ctx.op(cluster::ops::get_for_game::Input {
 			game_ids: vec![game_id],
 		}),
 		ctx.op(ds::ops::game_config::get::Input {
 			game_ids: vec![game_id],
 		}),
+		resolve_build_id(&ctx, game_id, body.build, body.build_tags.flatten()),
 	)?;
 	let cluster_id = unwrap!(clusters_res.games.first()).cluster_id;
 	let game_config = unwrap!(game_configs_res.game_configs.first());
@@ -153,7 +154,7 @@ pub async fn create(
 				durable: false,
 			}
 		}),
-		image_id: body.runtime.build,
+		image_id: build_id,
 		root_user_enabled: game_config.root_user_enabled,
 		args: body.runtime.arguments.unwrap_or_default(),
 		network_mode: network.mode.unwrap_or_default().api_into(),
@@ -317,8 +318,9 @@ pub async fn create_deprecated(
 			runtime: Box::new(models::ActorCreateActorRuntimeRequest {
 				arguments: body.runtime.arguments,
 				environment: body.runtime.environment,
-				build: body.runtime.build,
 			}),
+			build: Some(body.runtime.build),
+			build_tags: None,
 			tags: body.tags,
 		},
 		global,
@@ -396,6 +398,33 @@ pub async fn destroy_deprecated(
 	.await
 }
 
+// MARK: POST /actors/{}/upgrade
+pub async fn upgrade(
+	ctx: Ctx<Auth>,
+	actor_id: Uuid,
+	body: models::ActorUpgradeActorRequest,
+	query: GlobalQuery,
+) -> GlobalResult<serde_json::Value> {
+	let CheckOutput { game_id, env_id } = ctx.auth().check(ctx.op_ctx(), &query, false).await?;
+
+	assert::server_for_env(&ctx, actor_id, game_id, env_id).await?;
+
+	let build_id = resolve_build_id(&ctx, game_id, body.build, body.build_tags.flatten()).await?;
+
+	let mut sub = ctx
+		.subscribe::<ds::workflows::server::UpgradeStarted>(("server_id", actor_id))
+		.await?;
+
+	ctx.signal(ds::workflows::server::Upgrade { image_id: build_id })
+		.tag("server_id", actor_id)
+		.send()
+		.await?;
+
+	sub.next().await?;
+
+	Ok(json!({}))
+}
+
 // MARK: GET /actors
 #[derive(Debug, Clone, Deserialize)]
 pub struct ListQuery {
@@ -582,3 +611,33 @@ fn legacy_convert_actor_to_server(
 		tags: a.tags,
 	}
 }
+
+async fn resolve_build_id(
+	ctx: &Ctx<Auth>,
+	game_id: Uuid,
+	build_id: Option<Uuid>,
+	build_tags: Option<serde_json::Value>,
+) -> GlobalResult<Uuid> {
+	match (build_id, build_tags) {
+		(Some(build_id), None) => Ok(build_id),
+		// Resolve build from tags
+		(None, Some(build_tags)) => {
+			let builds_res = ctx
+				.op(build::ops::resolve_for_tags::Input {
+					game_id: Some(game_id),
+					tags: serde_json::from_value(build_tags)?,
+				})
+				.await?;
+
+			let build = unwrap_with!(builds_res.builds.first(), BUILDS_BUILD_NOT_FOUND_WITH_TAGS);
+
+			Ok(build.build_id)
+		}
+		_ => {
+			bail_with!(
+				API_BAD_BODY,
+				error = "must have either `build` or `buildTags`"
+			);
+		}
+	}
+}

packages/common/formatted-error/errors/builds/build-not-found-with-tags.md

@@ -0,0 +1,9 @@
+---
+name = "BUILDS_BUILD_NOT_FOUND_WITH_TAGS"
+description = "Build not found with the given tags."
+http_status = 400
+---
+
+# Build Not Found With Tags
+
+Build not found for given tags.

packages/infra/client/Cargo.toml

@@ -20,6 +20,7 @@ echo = { path = "echo" }
 isolate-v8-runner = { path = "isolate-v8-runner" }
 pegboard-config = { path = "config" }
 actor-kv = { path = "actor-kv", package = "pegboard-actor-kv" }
+manager = { path = "manager", package = "pegboard-manager" }
 
 [workspace.dependencies.sqlx]
 git = "https://github.com/rivet-gg/sqlx"

packages/services/build/src/ops/get.rs

@@ -14,7 +14,7 @@ pub struct Output {
 }
 
 #[derive(sqlx::FromRow)]
-struct BuildRow {
+pub(crate) struct BuildRow {
 	build_id: Uuid,
 	game_id: Option<Uuid>,
 	env_id: Option<Uuid>,
@@ -64,10 +64,8 @@ pub async fn get(ctx: &OperationCtx, input: &Input) -> GlobalResult<Output> {
 			kind,
 			compression,
 			tags
-		FROM
-			db_build.builds
-		WHERE
-			build_id = ANY($1)
+		FROM db_build.builds
+		WHERE build_id = ANY($1)
 		",
 		&input.build_ids,
 	)

packages/services/build/src/ops/mod.rs

@@ -1,3 +1,4 @@
 pub mod get;
 pub mod patch_tags;
 pub mod prewarm_ats;
+pub mod resolve_for_tags;

packages/services/build/src/ops/resolve_for_tags.rs

@@ -0,0 +1,49 @@
+use std::{collections::HashMap, convert::TryInto};
+
+use chirp_workflow::prelude::*;
+
+use super::get::BuildRow;
+use crate::types;
+
+#[derive(Debug)]
+pub struct Input {
+	pub game_id: Option<Uuid>,
+	pub tags: HashMap<String, String>,
+}
+
+#[derive(Debug)]
+pub struct Output {
+	pub builds: Vec<types::Build>,
+}
+
+#[operation]
+pub async fn get(ctx: &OperationCtx, input: &Input) -> GlobalResult<Output> {
+	let builds = sql_fetch_all!(
+		[ctx, BuildRow]
+		"
+		SELECT
+			build_id,
+			game_id,
+			env_id,
+			upload_id,
+			display_name,
+			image_tag,
+			create_ts,
+			kind,
+			compression,
+			tags
+		FROM db_build.builds
+		WHERE
+			($2 IS NULL OR game_id = $2) AND
+			tags @> $1
+		",
+		input.game_id,
+		serde_json::to_string(&input.tags)?,
+	)
+	.await?
+	.into_iter()
+	.map(|build| build.try_into())
+	.collect::<GlobalResult<Vec<_>>>()?;
+
+	Ok(Output { builds })
+}

packages/services/ds/src/workflows/server/mod.rs

@@ -329,7 +329,7 @@ async fn get_build_and_dc(
 			datacenter_ids: vec![input.datacenter_id],
 		})
 	)?;
-	let build = unwrap!(build_res.builds.first());
+	let build = unwrap_with!(build_res.builds.first(), BUILDS_BUILD_NOT_FOUND);
 	let upload_id = unwrap!(build.upload_id).as_uuid();
 	let build_kind = unwrap!(backend::build::BuildKind::from_i32(build.kind));
 	let build_compression = unwrap!(backend::build::BuildCompression::from_i32(
@@ -372,6 +372,29 @@ async fn set_connectable(ctx: &ActivityCtx, input: &SetConnectableInput) -> Glob
 	Ok(res.rows_affected() > 0)
 }
 
+#[derive(Debug, Serialize, Deserialize, Hash)]
+struct UpdateImageInput {
+	server_id: Uuid,
+	image_id: Uuid,
+}
+
+#[activity(UpdateImage)]
+async fn update_image(ctx: &ActivityCtx, input: &UpdateImageInput) -> GlobalResult<()> {
+	sql_execute!(
+		[ctx]
+		"
+		UPDATE db_ds.servers
+		SET image_id = $2
+		WHERE server_id = $1
+		",
+		input.server_id,
+		input.image_id,
+	)
+	.await?;
+
+	Ok(())
+}
+
 #[message("ds_server_create_complete")]
 pub struct CreateComplete {}
 
@@ -392,6 +415,17 @@ pub struct DestroyStarted {}
 #[message("ds_server_destroy_complete")]
 pub struct DestroyComplete {}
 
+#[signal("ds_server_upgrade")]
+pub struct Upgrade {
+	pub image_id: Uuid,
+}
+
+#[message("ds_server_upgrade_started")]
+pub struct UpgradeStarted {}
+
+#[message("ds_server_upgrade_complete")]
+pub struct UpgradeComplete {}
+
 #[signal("ds_server_drain")]
 pub struct Drain {
 	pub drain_timeout: i64,

packages/services/ds/src/workflows/server/nomad/mod.rs

@@ -12,7 +12,7 @@ use util::serde::AsHashableExt;
 
 use super::{
 	resolve_image_artifact_url, CreateComplete, CreateFailed, Destroy, Drain, DrainState,
-	GetBuildAndDcInput, InsertDbInput, Port, DRAIN_PADDING_MS,
+	GetBuildAndDcInput, InsertDbInput, Port, Upgrade, DRAIN_PADDING_MS,
 };
 use crate::{
 	types::{NetworkMode, Routing, ServerLifecycle, ServerResources},
@@ -176,6 +176,7 @@ pub(crate) async fn ds_server_nomad(ctx: &mut WorkflowCtx, input: &Input) -> Glo
 						}
 					}
 					Main::Destroy(sig) => return Ok(Loop::Break(sig.override_kill_timeout_ms)),
+					Main::Upgrade(_) => bail!("cannot upgrade nomad dynamic server"),
 				}
 
 				Ok(Loop::Continue)
@@ -905,6 +906,7 @@ join_signal!(Init {
 join_signal!(Main {
 	NomadAllocPlan,
 	NomadAllocUpdate,
-	Destroy,
 	Drain,
+	Upgrade,
+	Destroy,
 });