Balaji/etna forge indexer

.github/workflows/forge-stable.yaml

@@ -168,7 +168,7 @@ jobs:
               { TEST_NAME: 'changing-working-quorum-test-high-load', FORGE_RUNNER_DURATION_SECS: 900, FORGE_TEST_SUITE: 'changing_working_quorum_test_high_load', FORGE_ENABLE_FAILPOINTS: true },
               { TEST_NAME: 'pfn-const-tps-realistic-env', FORGE_RUNNER_DURATION_SECS: 900, FORGE_TEST_SUITE: 'pfn_const_tps_with_realistic_env' },
               { TEST_NAME: 'realistic-env-max-load-long', FORGE_RUNNER_DURATION_SECS: 7200, FORGE_TEST_SUITE: 'realistic_env_max_load_large' },
-              { TEST_NAME: 'etna', FORGE_RUNNER_DURATION_SECS: 3600, FORGE_TEST_SUITE: 'etna', FORGE_IMAGE_NAME: 'etna-rust' }
+              { TEST_NAME: 'etna', FORGE_RUNNER_DURATION_SECS: 5400, FORGE_TEST_SUITE: 'etna', FORGE_IMAGE_NAME: 'etna-rust', FORGE_ENABLE_INDEXER: true }
             ];
 
             const matrix = testName != "all" ? tests.filter(test => test.TEST_NAME === testName) : tests;
@@ -284,6 +284,7 @@ jobs:
       FORGE_ENABLE_PERFORMANCE: ${{ matrix.FORGE_ENABLE_PERFORMANCE || false }}
       FORGE_ENABLE_FAILPOINTS: ${{ matrix.FORGE_ENABLE_FAILPOINTS || false }}
       FORGE_IMAGE_NAME: ${{ matrix.FORGE_IMAGE_NAME }}
+      FORGE_ENABLE_INDEXER: ${{ matrix.FORGE_ENABLE_INDEXER || false }}
       POST_TO_SLACK: true
       SEND_RESULTS_TO_TRUNK: true
       FORGE_CONTINUOUS_TEST_MODE: true

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

@@ -3,8 +3,9 @@
 // SPDX-License-Identifier: Apache-2.0
 
 use crate::{
-    Factory, GenesisConfig, GenesisConfigFn, NodeConfigFn, Result, Swarm, Version,
-    INDEXER_GRPC_DOCKER_IMAGE_REPO, VALIDATOR_DOCKER_IMAGE_REPO,
+    Factory, GenesisConfig, GenesisConfigFn, IndexerDeployConfig, NodeConfigFn, Result, Swarm,
+    Version, INDEXER_GRPC_CONFIG_KEY, INDEXER_GRPC_DOCKER_IMAGE_REPO, INDEXER_GRPC_V2_CONFIG_KEY,
+    VALIDATOR_DOCKER_IMAGE_REPO,
 };
 use anyhow::bail;
 use futures::{future, FutureExt};
@@ -116,6 +117,7 @@ impl Factory for K8sFactory {
         genesis_config_fn: Option<GenesisConfigFn>,
         node_config_fn: Option<NodeConfigFn>,
         existing_db_tag: Option<String>,
+        indexer_deploy_config: Option<IndexerDeployConfig>,
     ) -> Result<Box<dyn Swarm>> {
         let genesis_modules_path = match genesis_config {
             Some(config) => match config {
@@ -196,28 +198,55 @@ impl Factory for K8sFactory {
             .boxed();
 
             let deploy_indexer_fut = async {
-            if self.enable_indexer {
-                // NOTE: by default, use a deploy profile and no additional configuration values
-                let config = serde_json::from_value(json!({
-                    "profile": self.deployer_profile.clone(),
-                    "era": new_era.clone(),
-                    "namespace": self.kube_namespace.clone(),
-                    "indexer-grpc-values": {
+                if let Some(deploy_config) = indexer_deploy_config {
+                    let mut config_json = json!({
+                        "profile":  deploy_config.deployer_profile.clone(),
+                        "era": new_era.clone(),
+                        "namespace": self.kube_namespace.clone(), 
+                    });
+                    let indexer_grpc_key = if deploy_config.use_indexer_v2 {
+                        INDEXER_GRPC_V2_CONFIG_KEY
+                    } else {
+                        INDEXER_GRPC_CONFIG_KEY
+                    };
+                    config_json[indexer_grpc_key] = json!({
                         "indexerGrpcImage": format!("{}:{}", INDEXER_GRPC_DOCKER_IMAGE_REPO, init_version),
                         "fullnodeConfig": {
                             "image": format!("{}:{}", VALIDATOR_DOCKER_IMAGE_REPO, init_version),
                         }
-                    },
-                }))?;
+                    });
+                    json_patch::merge(&mut config_json, &deploy_config.values);
+                    let indexer_deployer = ForgeDeployerManager::new(
+                        kube_client.clone(),
+                        self.kube_namespace.clone(),
+                        FORGE_INDEXER_DEPLOYER_DOCKER_IMAGE_REPO.to_string(),
+                        None,
+                    );
+                    indexer_deployer.start(config_json).await?;
+                    indexer_deployer.wait_completed().await?;
+                    Ok(())
+                } else if self.enable_indexer {
+                    // NOTE: by default, use a deploy profile and no additional configuration values
+                    let config = serde_json::from_value(json!({
+                        "profile": self.deployer_profile.clone(),
+                        "era": new_era.clone(),
+                        "namespace": self.kube_namespace.clone(),
+                        "indexer-grpc-values": {
+                            "indexerGrpcImage": format!("{}:{}", INDEXER_GRPC_DOCKER_IMAGE_REPO, init_version),
+                            "fullnodeConfig": {
+                                "image": format!("{}:{}", VALIDATOR_DOCKER_IMAGE_REPO, init_version),
+                            }
+                        },
+                    }))?;
 
-                let indexer_deployer = ForgeDeployerManager::new(
-                    kube_client.clone(),
-                    self.kube_namespace.clone(),
-                    FORGE_INDEXER_DEPLOYER_DOCKER_IMAGE_REPO.to_string(),
-                    None,
-                );
-                indexer_deployer.start(config).await?;
-                indexer_deployer.wait_completed().await
+                    let indexer_deployer = ForgeDeployerManager::new(
+                        kube_client.clone(),
+                        self.kube_namespace.clone(),
+                        FORGE_INDEXER_DEPLOYER_DOCKER_IMAGE_REPO.to_string(),
+                        None,
+                    );
+                    indexer_deployer.start(config).await?;
+                    indexer_deployer.wait_completed().await
                 } else {
                     Ok(())
                 }

testsuite/forge/src/backend/k8s_deployer/constants.rs

@@ -10,7 +10,7 @@ pub const INDEXER_GRPC_DOCKER_IMAGE_REPO: &str =
     "us-docker.pkg.dev/aptos-registry/docker/indexer-grpc";
 
 /// The version of the forge deployer image to use.
-pub const DEFAULT_FORGE_DEPLOYER_IMAGE_TAG: &str = "0769f1eb0c17e101e6e9d852cfe1c22e2ca82190"; // default to the latest stable build from the main branch (2025-08-02)
+pub const DEFAULT_FORGE_DEPLOYER_IMAGE_TAG: &str = "84562c2931bac69ff12ad1154d7b3686efee4c60"; // default to the latest stable build from the main branch (2025-08-02)
 
 /// This is the service account name that the deployer will use to deploy the forge components. It may require extra permissions and additonal setup
 pub const FORGE_DEPLOYER_SERVICE_ACCOUNT_NAME: &str = "forge";
@@ -19,3 +19,7 @@ pub const FORGE_DEPLOYER_SERVICE_ACCOUNT_NAME: &str = "forge";
 pub const FORGE_DEPLOYER_VALUES_ENV_VAR_NAME: &str = "FORGE_DEPLOY_VALUES_JSON";
 
 pub const DEFAULT_FORGE_DEPLOYER_PROFILE: &str = "forge";
+
+pub const INDEXER_GRPC_CONFIG_KEY: &str = "indexer-grpc-values";
+
+pub const INDEXER_GRPC_V2_CONFIG_KEY: &str = "indexer-grpc-v2-values";

testsuite/forge/src/backend/k8s_deployer/deployer.rs

@@ -19,6 +19,22 @@ use kube::{
 use log::info;
 use std::{collections::BTreeMap, sync::Arc, time::Duration};
 
+#[derive(Clone)]
+pub struct IndexerDeployConfig {
+    pub deployer_profile: String,
+    pub use_indexer_v2: bool,
+    pub values: serde_json::Value,
+}
+
+impl IndexerDeployConfig {
+    pub fn new(profile: String, use_indexer_v2: bool, values: serde_json::Value) -> Self {
+        Self {
+            deployer_profile: profile,
+            use_indexer_v2,
+            values,
+        }
+    }
+}
 /// The ForgeDeployerManager is responsible for managing the lifecycle of forge deployers, which deploy the
 /// forge components to the k8s cluster.
 pub struct ForgeDeployerManager {

testsuite/forge/src/backend/local/mod.rs

@@ -2,7 +2,10 @@
 // Parts of the project are originally copyright © Meta Platforms, Inc.
 // SPDX-License-Identifier: Apache-2.0
 
-use crate::{Factory, GenesisConfig, GenesisConfigFn, NodeConfigFn, Result, Swarm, Version};
+use crate::{
+    Factory, GenesisConfig, GenesisConfigFn, IndexerDeployConfig, NodeConfigFn, Result, Swarm,
+    Version,
+};
 use anyhow::{bail, Context};
 use aptos_config::config::{NodeConfig, OverrideNodeConfig};
 use aptos_framework::ReleaseBundle;
@@ -184,6 +187,7 @@ impl Factory for LocalFactory {
         _genesis_config_fn: Option<GenesisConfigFn>,
         _node_config_fn: Option<NodeConfigFn>,
         _existing_db_tag: Option<String>,
+        _indexer_config: Option<IndexerDeployConfig>,
     ) -> Result<Box<dyn Swarm>> {
         let framework = match genesis_config {
             Some(config) => match config {

testsuite/forge/src/config.rs

@@ -58,6 +58,8 @@ pub struct ForgeConfig {
 
     /// Retain debug logs and above for all nodes instead of just the first 5 nodes
     pub retain_debug_logs: bool,
+
+    pub indexer_config: Option<IndexerDeployConfig>,
 }
 
 impl ForgeConfig {
@@ -281,6 +283,11 @@ impl ForgeConfig {
         self
     }
 
+    pub fn with_indexer_config(mut self, indexer_config: IndexerDeployConfig) -> Self {
+        self.indexer_config = Some(indexer_config);
+        self
+    }
+
     pub fn get_emit_job(&self) -> &EmitJobRequest {
         &self.emit_job_request
     }
@@ -357,6 +364,7 @@ impl Default for ForgeConfig {
             validator_resource_override: NodeResourceOverride::default(),
             fullnode_resource_override: NodeResourceOverride::default(),
             retain_debug_logs: false,
+            indexer_config: None,
         }
     }
 }

testsuite/forge/src/interface/factory.rs

@@ -3,7 +3,7 @@
 // SPDX-License-Identifier: Apache-2.0
 
 use super::{GenesisConfig, Swarm, Version};
-use crate::{GenesisConfigFn, NodeConfigFn, Result};
+use crate::{GenesisConfigFn, IndexerDeployConfig, NodeConfigFn, Result};
 use rand::rngs::StdRng;
 use std::{num::NonZeroUsize, time::Duration};
 
@@ -24,5 +24,6 @@ pub trait Factory {
         genesis_config_fn: Option<GenesisConfigFn>,
         node_config_fn: Option<NodeConfigFn>,
         existing_db_tag: Option<String>,
+        indexer_config: Option<IndexerDeployConfig>,
     ) -> Result<Box<dyn Swarm>>;
 }

testsuite/forge/src/runner.rs

@@ -290,6 +290,7 @@ impl<'cfg, F: Factory> Forge<'cfg, F> {
                 self.tests.genesis_helm_config_fn.clone(),
                 self.tests.build_node_helm_config_fn(retain_debug_logs),
                 self.tests.existing_db_tag.clone(),
+                self.tests.indexer_config.clone(),
             ))?;
 
             // Run AptosTests

testsuite/testcases/src/load_vs_perf_benchmark.rs

@@ -33,7 +33,7 @@ pub struct SingleRunStats {
 pub enum Workloads {
     TPS(Vec<usize>),
     TRANSACTIONS(Vec<TransactionWorkload>),
-    RawTransactions(Vec<(String, TransactionType)>),
+    RawTransactions(Vec<RawTransactionWorkload>),
 }
 
 impl Workloads {
@@ -74,7 +74,7 @@ impl Workloads {
             },
             Self::TRANSACTIONS(workloads) => format!("TRANSACTIONS({:?})", workloads[index]),
             Self::RawTransactions(workloads) => {
-                format!("RAW TRANSACTIONS({:?})", workloads[index].0)
+                format!("RAW TRANSACTIONS({:?})", workloads[index].name)
             },
         }
     }
@@ -95,7 +95,7 @@ impl Workloads {
                 },
                 workloads[index].phase_name(phase)
             ),
-            Self::RawTransactions(workloads) => format!("{}: {}", index, workloads[index].0),
+            Self::RawTransactions(workloads) => format!("{}: {}", index, workloads[index].name),
         }
     }
 
@@ -104,7 +104,10 @@ impl Workloads {
             Self::TPS(tpss) => request.mode(EmitJobMode::ConstTps { tps: tpss[index] }),
             Self::TRANSACTIONS(workloads) => workloads[index].configure(request),
             Self::RawTransactions(workloads) => {
-                request.transaction_type(workloads[index].1.clone())
+                let workload = &workloads[index];
+                request
+                    .mode(workload.emit_job_mode.clone())
+                    .transaction_type(workload.workload.clone())
             },
         }
     }
@@ -248,6 +251,23 @@ impl TransactionWorkload {
     }
 }
 
+#[derive(Debug, Clone)]
+pub struct RawTransactionWorkload {
+    name: String,
+    workload: TransactionType,
+    emit_job_mode: EmitJobMode,
+}
+
+impl RawTransactionWorkload {
+    pub fn new(name: String, workload: TransactionType, emit_job_mode: EmitJobMode) -> Self {
+        Self {
+            name,
+            workload,
+            emit_job_mode,
+        }
+    }
+}
+
 pub struct BackgroundTraffic {
     pub traffic: EmitJobRequest,
     pub criteria: Vec<SuccessCriteria>,