enhancement(ci): combine build steps for integration test workflows (#…

…17724)

- vdev integration test logic has ability to build with all integration
features flag
- CI workflows use the new vdev flag, and are run within the same job,
thus each step leverages the cached runner image
- Adds retries for integration tests both at nextest and between
bringup/teardown of the container services
- Reduces billable time for Integration Test Suite workflow by 90% 🚀

.github/workflows/integration-test.yml

@@ -1,22 +1,12 @@
-# This workflow is used to run an integration test.
-# The most common use case is that it is triggered by another workflow,
-# such as the Master Merge Queue Suite, or the Integration Comment.
+# Integration Test
 #
-# It can also be triggered on manual dispatch in CI however.
-# In that use case, an input for the test name needs to be provided.
+# This workflow is used to run an integration test on demand.
+# An input for the test name needs to be provided.
 # TODO: check if the input is "all" , and run all, without a timeout?
 
 name: Integration Test
 
 on:
-  workflow_call:
-    inputs:
-      if:
-         required: false
-         type: boolean
-      test_name:
-        required: true
-        type: string
   workflow_dispatch:
     inputs:
       test_name:

scripts/ci-integration-test.sh

@@ -0,0 +1,30 @@
+#!/bin/bash
+
+# Used in CI to run and stop an integration test and upload the results of it.
+# This is useful to allow retrying the integration test at a higher level than
+# the nextest and reduce code duplication in the workflow file.
+
+set -u
+
+if [[ -z "${CI:-}" ]]; then
+  echo "Aborted: this script is for use in CI." >&2
+  exit 1
+fi
+
+if [ $# -ne 1 ]
+then
+  echo "usage: $0 INTEGRATION"
+  exit 1
+fi
+
+set -x
+
+INTEGRATION=$1
+
+cargo vdev -v int start "${INTEGRATION}"
+sleep 15
+cargo vdev -v int test --retries 2 -a "${INTEGRATION}"
+RET=$?
+cargo vdev -v int stop "${INTEGRATION}"
+./scripts/upload-test-results.sh
+exit $RET

scripts/integration/gcp/compose.yaml

@@ -6,3 +6,12 @@ services:
     environment:
     - PUBSUB_PROJECT1=testproject,topic1:subscription1
     - PUBSUB_PROJECT2=sourceproject,topic2:subscription2
+  chronicle-emulator:
+    image: docker.io/plork/chronicle-emulator:${CONFIG_VERSION}
+    ports:
+    - 3000:3000
+    volumes:
+    - ./public.pem:/public.pem:ro
+    command:
+    - -p
+    - /public.pem

scripts/integration/gcp/test.yaml

@@ -1,10 +1,12 @@
 features:
 - gcp-integration-tests
+- chronicle-integration-tests
 
 test_filter: '::gcp::'
 
 env:
   EMULATOR_ADDRESS: http://gcloud-pubsub:8681
+  CHRONICLE_ADDRESS: http://chronicle-emulator:3000
 
 matrix:
   version: [latest]

src/sinks/gcp/chronicle_unstructured.rs

@@ -550,12 +550,10 @@ mod integration_tests {
         trace_init();
 
         let log_type = random_string(10);
-        let (sink, healthcheck) = config_build(
-            &log_type,
-            "/home/vector/scripts/integration/chronicle/auth.json",
-        )
-        .await
-        .expect("Building sink failed");
+        let (sink, healthcheck) =
+            config_build(&log_type, "/home/vector/scripts/integration/gcp/auth.json")
+                .await
+                .expect("Building sink failed");
 
         healthcheck.await.expect("Health check failed");
 
@@ -585,7 +583,7 @@ mod integration_tests {
         // Test with an auth file that doesnt match the public key sent to the dummy chronicle server.
         let sink = config_build(
             &log_type,
-            "/home/vector/scripts/integration/chronicle/invalidauth.json",
+            "/home/vector/scripts/integration/gcp/invalidauth.json",
         )
         .await;
 
@@ -599,12 +597,10 @@ mod integration_tests {
         // The chronicle-emulator we are testing against is setup so a `log_type` of "INVALID"
         // will return a `400 BAD_REQUEST`.
         let log_type = "INVALID";
-        let (sink, healthcheck) = config_build(
-            log_type,
-            "/home/vector/scripts/integration/chronicle/auth.json",
-        )
-        .await
-        .expect("Building sink failed");
+        let (sink, healthcheck) =
+            config_build(log_type, "/home/vector/scripts/integration/gcp/auth.json")
+                .await
+                .expect("Building sink failed");
 
         healthcheck.await.expect("Health check failed");
 

vdev/src/commands/integration/start.rs

@@ -24,6 +24,6 @@ impl Cli {
             let env = envs.keys().next().expect("Integration has no environments");
             env.clone()
         };
-        IntegrationTest::new(self.integration, environment)?.start()
+        IntegrationTest::new(self.integration, environment, false, 0)?.start()
     }
 }

vdev/src/commands/integration/stop.rs

@@ -14,7 +14,7 @@ pub struct Cli {
 impl Cli {
     pub fn exec(self) -> Result<()> {
         if let Some(active) = EnvsDir::new(&self.integration).active()? {
-            IntegrationTest::new(self.integration, active)?.stop()
+            IntegrationTest::new(self.integration, active, false, 0)?.stop()
         } else {
             println!("No environment for {:?} is active.", self.integration);
             Ok(())

vdev/src/commands/integration/test.rs

@@ -20,6 +20,14 @@ pub struct Cli {
     /// The desired environment (optional)
     environment: Option<String>,
 
+    /// Whether to compile the test runner with all integration test features
+    #[arg(short = 'a', long)]
+    build_all: bool,
+
+    /// Number of retries to allow on each integration test case.
+    #[arg(short = 'r', long)]
+    retries: u8,
+
     /// Extra test command arguments
     args: Vec<String>,
 }
@@ -30,17 +38,28 @@ impl Cli {
         let envs = config.environments();
 
         let active = EnvsDir::new(&self.integration).active()?;
+
         match (self.environment, active) {
             (Some(environment), Some(active)) if environment != active => {
                 bail!("Requested environment {environment:?} does not match active one {active:?}")
             }
             (Some(environment), _) => {
-                IntegrationTest::new(self.integration, environment)?.test(self.args)
+                IntegrationTest::new(self.integration, environment, self.build_all, self.retries)?
+                    .test(self.args)
+            }
+            (None, Some(active)) => {
+                IntegrationTest::new(self.integration, active, self.build_all, self.retries)?
+                    .test(self.args)
             }
-            (None, Some(active)) => IntegrationTest::new(self.integration, active)?.test(self.args),
             (None, None) => {
                 for env_name in envs.keys() {
-                    IntegrationTest::new(&self.integration, env_name)?.test(self.args.clone())?;
+                    IntegrationTest::new(
+                        &self.integration,
+                        env_name,
+                        self.build_all,
+                        self.retries,
+                    )?
+                    .test(self.args.clone())?;
                 }
                 Ok(())
             }

vdev/src/testing/integration.rs

@@ -21,10 +21,17 @@ pub struct IntegrationTest {
     runner: IntegrationTestRunner,
     compose: Option<Compose>,
     env_config: Environment,
+    build_all: bool,
+    retries: u8,
 }
 
 impl IntegrationTest {
-    pub fn new(integration: impl Into<String>, environment: impl Into<String>) -> Result<Self> {
+    pub fn new(
+        integration: impl Into<String>,
+        environment: impl Into<String>,
+        build_all: bool,
+        retries: u8,
+    ) -> Result<Self> {
         let integration = integration.into();
         let environment = environment.into();
         let (test_dir, config) = IntegrationTestConfig::load(&integration)?;
@@ -34,8 +41,12 @@ impl IntegrationTest {
         };
         let network_name = format!("vector-integration-tests-{integration}");
         let compose = Compose::new(test_dir, env_config.clone(), network_name.clone())?;
+
+        // None if compiling with all integration test feature flag.
+        let runner_name = (!build_all).then(|| integration.clone());
+
         let runner = IntegrationTestRunner::new(
-            integration.clone(),
+            runner_name,
             &config.runner,
             compose.is_some().then_some(network_name),
         )?;
@@ -48,6 +59,8 @@ impl IntegrationTest {
             runner,
             compose,
             env_config,
+            build_all,
+            retries,
         })
     }
 
@@ -69,7 +82,12 @@ impl IntegrationTest {
         let mut args = self.config.args.clone().unwrap_or_default();
 
         args.push("--features".to_string());
-        args.push(self.config.features.join(","));
+
+        args.push(if self.build_all {
+            "all-integration-tests".to_string()
+        } else {
+            self.config.features.join(",")
+        });
 
         // If the test field is not present then use the --lib flag
         match self.config.test {
@@ -91,6 +109,11 @@ impl IntegrationTest {
             args.push("--no-capture".to_string());
         }
 
+        if self.retries > 0 {
+            args.push("--retries".to_string());
+            args.push(self.retries.to_string());
+        }
+
         self.runner
             .test(&env_vars, &self.config.runner.env, &args)?;
 

vdev/src/testing/runner.rs

@@ -47,6 +47,13 @@ fn detect_container_tool() -> OsString {
     fatal!("No container tool could be detected.");
 }
 
+fn get_rust_version() -> String {
+    match RustToolchainConfig::parse() {
+        Ok(config) => config.channel,
+        Err(error) => fatal!("Could not read `rust-toolchain.toml` file: {error}"),
+    }
+}
+
 fn dockercmd<I: AsRef<OsStr>>(args: impl IntoIterator<Item = I>) -> Command {
     let mut command = Command::new(&*CONTAINER_TOOL);
     command.args(args);
@@ -93,13 +100,6 @@ pub trait ContainerTestRunner: TestRunner {
             .wait(format!("Stopping container {}", self.container_name()))
     }
 
-    fn get_rust_version(&self) -> String {
-        match RustToolchainConfig::parse() {
-            Ok(config) => config.channel,
-            Err(error) => fatal!("Could not read `rust-toolchain.toml` file: {error}"),
-        }
-    }
-
     fn state(&self) -> Result<RunnerState> {
         let mut command = dockercmd(["ps", "-a", "--format", "{{.Names}} {{.State}}"]);
         let container_name = self.container_name();
@@ -183,8 +183,10 @@ pub trait ContainerTestRunner: TestRunner {
             &self.image_name(),
             "--file",
             dockerfile.to_str().unwrap(),
+            "--label",
+            "vector-test-runner=true",
             "--build-arg",
-            &format!("RUST_VERSION={}", self.get_rust_version()),
+            &format!("RUST_VERSION={}", get_rust_version()),
             ".",
         ]);
 
@@ -295,15 +297,16 @@ where
 }
 
 pub(super) struct IntegrationTestRunner {
-    integration: String,
+    // The integration is None when compiling the runner image with the `all-integration-tests` feature.
+    integration: Option<String>,
     needs_docker_socket: bool,
     network: Option<String>,
     volumes: Vec<String>,
 }
 
 impl IntegrationTestRunner {
     pub(super) fn new(
-        integration: String,
+        integration: Option<String>,
         config: &IntegrationRunnerConfig,
         network: Option<String>,
     ) -> Result<Self> {
@@ -344,11 +347,11 @@ impl ContainerTestRunner for IntegrationTestRunner {
     }
 
     fn container_name(&self) -> String {
-        format!(
-            "vector-test-runner-{}-{}",
-            self.integration,
-            self.get_rust_version()
-        )
+        if let Some(integration) = self.integration.as_ref() {
+            format!("vector-test-runner-{}-{}", integration, get_rust_version())
+        } else {
+            format!("vector-test-runner-{}", get_rust_version())
+        }
     }
 
     fn image_name(&self) -> String {
@@ -372,7 +375,7 @@ impl ContainerTestRunner for DockerTestRunner {
     }
 
     fn container_name(&self) -> String {
-        format!("vector-test-runner-{}", self.get_rust_version())
+        format!("vector-test-runner-{}", get_rust_version())
     }
 
     fn image_name(&self) -> String {