[forge] Implement suggestions

.gitignore

@@ -80,7 +80,6 @@ target-out-docker
 docker/compose/indexer-grpc/data-service-grpc-server.crt
 docker/compose/indexer-grpc/data-service-grpc-server.key
 
-
 # Doc generation output
 *.md.old
 
@@ -135,3 +134,6 @@ test_indexer_grpc/*
 *.dot
 *.bytecode
 !third_party/move/move-prover/tests/xsources/design/*.bytecode
+
+# Allow forge envs
+!forge.env

testsuite/forge-cli/Cargo.toml

@@ -30,6 +30,7 @@ once_cell = { workspace = true }
 rand = { workspace = true }
 random_word = { workspace = true }
 reqwest = { workspace = true }
+serde_json = { workspace = true }
 serde_yaml = { workspace = true }
 tokio = { workspace = true }
 url = { workspace = true }

testsuite/forge-cli/src/main.rs

@@ -65,6 +65,7 @@ use clap::{Parser, Subcommand};
 use futures::stream::{FuturesUnordered, StreamExt};
 use once_cell::sync::Lazy;
 use rand::{rngs::ThreadRng, seq::SliceRandom, Rng};
+use serde_json::json;
 use std::{
     env,
     num::NonZeroUsize,
@@ -140,8 +141,8 @@ enum OperatorCommand {
     SetNodeImageTag(SetNodeImageTag),
     /// Clean up an existing cluster
     CleanUp(CleanUp),
-    /// Resize an existing cluster
-    Resize(Resize),
+    /// Create a new cluster for testing purposes
+    Create(Create),
 }
 
 #[derive(Parser, Debug)]
@@ -193,6 +194,11 @@ struct K8sSwarm {
         help = "Retain debug logs and above for all nodes instead of just the first 5 nodes"
     )]
     retain_debug_logs: bool,
+    #[clap(
+        long,
+        help = "If set, spins up an indexer stack alongside the testnet. Same as --enable-indexer"
+    )]
+    enable_indexer: bool,
 }
 
 #[derive(Parser, Debug)]
@@ -217,8 +223,8 @@ struct CleanUp {
 }
 
 #[derive(Parser, Debug)]
-struct Resize {
-    #[clap(long, help = "The kubernetes namespace to resize")]
+struct Create {
+    #[clap(long, help = "The kubernetes namespace to create in")]
     namespace: String,
     #[clap(long, default_value_t = 30)]
     num_validators: usize,
@@ -227,13 +233,13 @@ struct Resize {
     #[clap(
         long,
         help = "Override the image tag used for validators",
-        default_value = "devnet"
+        default_value = "main"
     )]
     validator_image_tag: String,
     #[clap(
         long,
         help = "Override the image tag used for testnet-specific components",
-        default_value = "devnet"
+        default_value = "main"
     )]
     testnet_image_tag: String,
     #[clap(
@@ -248,6 +254,8 @@ struct Resize {
     connect_directly: bool,
     #[clap(long, help = "If set, enables HAProxy for each of the validators")]
     enable_haproxy: bool,
+    #[clap(long, help = "If set, spins up an indexer stack alongside the testnet")]
+    enable_indexer: bool,
 }
 
 // common metrics thresholds:
@@ -393,6 +401,7 @@ fn main() -> Result<()> {
                             k8s.reuse,
                             k8s.keep,
                             k8s.enable_haproxy,
+                            k8s.enable_indexer,
                         )
                         .unwrap(),
                         &args.options,
@@ -421,19 +430,23 @@ fn main() -> Result<()> {
                 }
                 Ok(())
             },
-            OperatorCommand::Resize(resize) => {
-                runtime.block_on(install_testnet_resources(
-                    resize.namespace,
-                    resize.num_validators,
-                    resize.num_fullnodes,
-                    resize.validator_image_tag,
-                    resize.testnet_image_tag,
-                    resize.move_modules_dir,
-                    !resize.connect_directly,
-                    resize.enable_haproxy,
-                    None,
-                    None,
-                ))?;
+            OperatorCommand::Create(create) => {
+                let client = Arc::new(ForgeKubeClient::new(
+                    runtime.block_on(create_k8s_client())?,
+                    create.namespace,
+                ));
+                let era = generate_new_era();
+                let testnet_values = json!({
+                    "profile": DEFAULT_FORGE_DEPLOYER_PROFILE,
+                    "era": era,
+                });
+                let manager = ForgeBackendManager::new();
+                // NOTE: this is generally not going to run from within the cluster, do not perform any operations
+                // that might require internal DNS resolution to work, such as txn emission directly against the node service IPs.
+                runtime.block_on(manager.start(client.clone(), testnet_values.clone()))?;
+                if create.enable_indexer {
+                    runtime.block_on(manager.start(client.clone(), testnet_values.clone()))?;
+                }
                 Ok(())
             },
         },

testsuite/forge.env

@@ -0,0 +1,6 @@
+# This file is source-d when running Forge
+# LEAVE ALL LINES COMMENTED OUT UNLESS FOR TESTING. NEVER LAND THIS WITH VALUES UNCOMMENTED
+
+# FORGE_ENABLE_INDEXER=true
+# FORGE_ENABLE_HAPROXY=true
+# FORGE_RETAIN_DEBUG_LOGS=true

testsuite/forge.py

@@ -1147,6 +1147,7 @@ def create_forge_command(
     forge_namespace_reuse: Optional[str],
     forge_namespace_keep: Optional[str],
     forge_enable_haproxy: Optional[str],
+    forge_enable_indexer: Optional[str],
     cargo_args: Optional[Sequence[str]],
     forge_cli_args: Optional[Sequence[str]],
     test_args: Optional[Sequence[str]],
@@ -1216,6 +1217,8 @@ def create_forge_command(
         forge_args.append("--keep")
     if forge_enable_haproxy == "true":
         forge_args.append("--enable-haproxy")
+    if forge_enable_indexer == "true":
+        forge_args.append("--enable-indexer")
 
     if test_args:
         forge_args.extend(test_args)
@@ -1328,6 +1331,7 @@ def seeded_random_choice(namespace: str, cluster_names: Sequence[str]) -> str:
 @envoption("FORGE_NAMESPACE_KEEP")
 @envoption("FORGE_NAMESPACE_REUSE")
 @envoption("FORGE_ENABLE_HAPROXY")
+@envoption("FORGE_ENABLE_INDEXER")
 @envoption("FORGE_ENABLE_FAILPOINTS")
 @envoption("FORGE_ENABLE_PERFORMANCE")
 @envoption("FORGE_TEST_SUITE")
@@ -1373,6 +1377,7 @@ def test(
     forge_enable_failpoints: Optional[str],
     forge_enable_performance: Optional[str],
     forge_enable_haproxy: Optional[str],
+    forge_enable_indexer: Optional[str],
     forge_test_suite: str,
     forge_runner_duration_secs: str,
     forge_image_tag: Optional[str],
@@ -1598,6 +1603,7 @@ def test(
         forge_namespace_reuse=forge_namespace_reuse,
         forge_namespace_keep=forge_namespace_keep,
         forge_enable_haproxy=forge_enable_haproxy,
+        forge_enable_indexer=forge_enable_indexer,
         cargo_args=cargo_args,
         forge_cli_args=forge_cli_args,
         test_args=test_args,

testsuite/forge/src/backend/k8s/cluster_helper.rs

@@ -405,7 +405,7 @@ pub async fn uninstall_testnet_resources(kube_namespace: String) -> Result<()> {
     Ok(())
 }
 
-fn generate_new_era() -> String {
+pub fn generate_new_era() -> String {
     let mut rng = rand::thread_rng();
     let r: u8 = rng.gen();
     format!("forge{}", r)
@@ -826,15 +826,41 @@ fn dump_helm_values_to_file(helm_release_name: &str, tmp_dir: &TempDir) -> Resul
 
 #[derive(Error, Debug)]
 #[error("{0}")]
-enum ApiError {
+pub enum ApiError {
     RetryableError(String),
     FinalError(String),
 }
 
-async fn create_namespace(
+/// Does the same as create_namespace and handling the 409, but for any k8s resource T
+pub async fn maybe_create_k8s_resource<T>(
+    api: Arc<dyn ReadWrite<T>>,
+    resource: T,
+) -> Result<T, ApiError>
+where
+    T: kube::Resource + Clone + DeserializeOwned + Debug,
+    <T as kube::Resource>::DynamicType: Default,
+{
+    if let Err(KubeError::Api(api_err)) = api.create(&PostParams::default(), &resource).await {
+        if api_err.code == 409 {
+            info!(
+                "Resource {} already exists, continuing with it",
+                resource.name()
+            );
+        } else {
+            return Err(ApiError::RetryableError(format!(
+                "Failed to use existing resource {}: {:?}",
+                resource.name(),
+                api_err
+            )));
+        }
+    }
+    Ok(resource)
+}
+
+pub async fn create_namespace(
     namespace_api: Arc<dyn ReadWrite<Namespace>>,
     kube_namespace: String,
-) -> Result<(), ApiError> {
+) -> Result<Namespace, ApiError> {
     let kube_namespace_name = kube_namespace.clone();
     let namespace = Namespace {
         metadata: ObjectMeta {
@@ -866,7 +892,7 @@ async fn create_namespace(
             )));
         }
     }
-    Ok(())
+    Ok(namespace)
 }
 
 pub async fn create_management_configmap(
@@ -1067,11 +1093,11 @@ pub fn make_k8s_label(value: String) -> String {
 #[cfg(test)]
 mod tests {
     use super::*;
-    use crate::FailedNamespacesApi;
+    use crate::FailedK8sResourceApi;
 
     #[tokio::test]
     async fn test_create_namespace_final_error() {
-        let namespace_creator = Arc::new(FailedNamespacesApi::from_status_code(401));
+        let namespace_creator = Arc::new(FailedK8sResourceApi::from_status_code(401));
         let result = create_namespace(namespace_creator, "banana".to_string()).await;
         match result {
             Err(ApiError::FinalError(_)) => {},
@@ -1148,7 +1174,7 @@ labels:
 
     #[tokio::test]
     async fn test_create_namespace_retryable_error() {
-        let namespace_creator = Arc::new(FailedNamespacesApi::from_status_code(403));
+        let namespace_creator = Arc::new(FailedK8sResourceApi::from_status_code(403));
         let result = create_namespace(namespace_creator, "banana".to_string()).await;
         match result {
             Err(ApiError::RetryableError(_)) => {},

testsuite/forge/src/backend/k8s/fullnode.rs

@@ -513,9 +513,7 @@ pub async fn install_public_fullnode<'a>(
 #[cfg(test)]
 mod tests {
     use super::*;
-    use crate::{
-        MockConfigMapApi, MockPersistentVolumeClaimApi, MockServiceApi, MockStatefulSetApi,
-    };
+    use crate::MockK8sResourceApi;
     use aptos_config::config::Identity;
     use aptos_sdk::crypto::{x25519::PrivateKey, Uniform};
     use k8s_openapi::apimachinery::pkg::api::resource::Quantity;
@@ -728,15 +726,14 @@ mod tests {
         let version = Version::new(0, "banana".to_string());
 
         // create APIs
-        let stateful_set_api = Arc::new(MockStatefulSetApi::from_stateful_set(
-            get_dummy_validator_stateful_set(),
+        let stateful_set_api: Arc<MockK8sResourceApi<StatefulSet>> = Arc::new(
+            MockK8sResourceApi::from_resource(get_dummy_validator_stateful_set()),
+        );
+        let configmap_api = Arc::new(MockK8sResourceApi::new());
+        let persistent_volume_claim_api = Arc::new(MockK8sResourceApi::from_resource(
+            get_dummy_validator_persistent_volume_claim(),
         ));
-        let configmap_api = Arc::new(MockConfigMapApi::from_config_map(ConfigMap::default()));
-        let persistent_volume_claim_api =
-            Arc::new(MockPersistentVolumeClaimApi::from_persistent_volume_claim(
-                get_dummy_validator_persistent_volume_claim(),
-            ));
-        let service_api = Arc::new(MockServiceApi::from_service(Service::default()));
+        let service_api = Arc::new(MockK8sResourceApi::new());
 
         // get the base config and mutate it
         let mut node_config = get_default_pfn_node_config();