Validation framework testcontainer env var update (open-telemetry#2506)

# Change Summary

Added a new struct to describe a template env var to allow for
situations where env var needs the host port which the framework sets
dynamically

## What issue does this PR close?

* Closes open-telemetry#2499 

## How are these changes tested?

unit test

## Are there any user-facing changes?

no

Author: c1ly

rust/otap-dataflow/crates/validation/README.md

@@ -472,9 +472,45 @@ let localstack = ContainerConfig::new("localstack/localstack", "3.4")
   Docker image and tag
 - `.env(key, value)` - set an environment variable on the container
   - can be chained multiple times
+- `.env_host_port(key, value_template, internal_port)` - set an environment
+  variable whose value is a Jinja2 template; `{{ host_port }}` is replaced
+  with the host port mapped to `internal_port` after port allocation
+  - can be chained multiple times
+  - if no connection maps the given `internal_port`, the framework
+    auto-allocates a host port during config wiring
+  - the resolved host port is consistent with the port used for Docker
+    port mapping and any `ContainerConnection` or
+    `PipelineContainerConnection` referencing the same internal port
 - `.entrypoint(cmd)` - override the container's entrypoint
   - optional
 
+##### Templated environment variables
+
+Some containers require environment variables that reference the host port
+assigned to one of their exposed ports. For example, Kafka's advertised
+listeners must contain the host-reachable address so that clients outside
+the container can connect. Use `env_host_port` for this:
+
+```rust
+use otap_df_validation::ContainerConfig;
+
+let kafka = ContainerConfig::new("confluentinc/cp-kafka", "7.5.0")
+    .env("KAFKA_NODE_ID", "1")
+    .env_host_port(
+        "KAFKA_ADVERTISED_LISTENERS",
+        "PLAINTEXT://127.0.0.1:{{ host_port }}",
+        9092,
+    );
+```
+
+After config wiring, if host port 54321 was allocated for container port
+9092, the container starts with `KAFKA_ADVERTISED_LISTENERS` set to
+`PLAINTEXT://127.0.0.1:54321`.
+
+If a `PipelineContainerConnection` or `ContainerConnection` also
+references internal port 9092 on the same container, they all share the
+same allocated host port.
+
 Register the container on the scenario with `add_container`:
 
 ```rust
@@ -671,7 +707,12 @@ Scenario::new()
         "localstack",
         ContainerConfig::new("localstack/localstack", "3.4")
             .env("SERVICES", "s3")
-            .env("DEFAULT_REGION", "us-east-1"),
+            .env("DEFAULT_REGION", "us-east-1")
+            .env_host_port(
+                "LOCALSTACK_HOST",
+                "127.0.0.1:{{ host_port }}",
+                4566,
+            ),
     )
     .add_generator(
         "gen",

rust/otap-dataflow/crates/validation/src/container.rs

@@ -11,12 +11,24 @@
 //! interact with plain Rust types defined here.
 
 use crate::error::ValidationError;
+use crate::template::render_jinja;
+use minijinja::context;
 use std::collections::HashMap;
 use testcontainers::core::IntoContainerPort;
 use testcontainers::core::WaitFor;
 use testcontainers::runners::AsyncRunner;
 use testcontainers::{ContainerAsync, GenericImage, ImageExt};
 
+/// An environment variable whose value is a Jinja2 template resolved
+/// after host port allocation. `{{ host_port }}` in the template is
+/// replaced with the host port mapped to `internal_port`.
+#[derive(Debug, Clone)]
+pub(crate) struct TemplatedEnvVar {
+    pub(crate) key: String,
+    pub(crate) value_template: String,
+    pub(crate) internal_port: u16,
+}
+
 /// Describes a Docker container to run alongside the validation scenario.
 ///
 /// Use the builder methods to configure the image and environment variables.
@@ -43,6 +55,9 @@ pub struct ContainerConfig {
     /// with host port as the value. Populated by the framework during
     /// config wiring for container connections.
     pub(crate) mapped_ports: HashMap<u16, u16>,
+    /// Templated environment variables to be resolved after port allocation.
+    /// `None` when no templated env vars have been set.
+    pub(crate) templated_env_vars: Option<Vec<TemplatedEnvVar>>,
 }
 
 impl ContainerConfig {
@@ -55,6 +70,7 @@ impl ContainerConfig {
             env_vars: Vec::new(),
             entrypoint: None,
             mapped_ports: HashMap::new(),
+            templated_env_vars: None,
         }
     }
 
@@ -72,6 +88,61 @@ impl ContainerConfig {
         self
     }
 
+    /// Set an environment variable whose value is a Jinja2 template.
+    /// After port allocation, `{{ host_port }}` is replaced with the host
+    /// port mapped to `internal_port`. If no connection maps that port,
+    /// the framework auto-allocates one during config wiring.
+    ///
+    /// # Example
+    ///
+    /// ```ignore
+    /// ContainerConfig::new("confluentinc/cp-kafka", "7.5.0")
+    ///     .env_host_port(
+    ///         "KAFKA_ADVERTISED_LISTENERS",
+    ///         "PLAINTEXT://127.0.0.1:{{ host_port }}",
+    ///         9092,
+    ///     )
+    /// ```
+    #[must_use]
+    pub fn env_host_port(
+        mut self,
+        key: impl Into<String>,
+        value_template: impl Into<String>,
+        internal_port: u16,
+    ) -> Self {
+        self.templated_env_vars
+            .get_or_insert_with(Vec::new)
+            .push(TemplatedEnvVar {
+                key: key.into(),
+                value_template: value_template.into(),
+                internal_port,
+            });
+        self
+    }
+
+    /// Resolve all templated environment variables by rendering their
+    /// Jinja2 templates with `host_port` set to the allocated host port
+    /// from `mapped_ports`. Resolved values are appended to `env_vars`
+    /// in FIFO order. After resolution, `templated_env_vars` is set to
+    /// `None`.
+    pub(crate) fn resolve_templated_env_vars(&mut self) -> Result<(), ValidationError> {
+        let vars = match self.templated_env_vars.take() {
+            Some(v) => v,
+            None => return Ok(()),
+        };
+        for tev in vars {
+            let &host_port = self.mapped_ports.get(&tev.internal_port).ok_or_else(|| {
+                ValidationError::Config(format!(
+                    "templated env var '{}' references unmapped internal port {}",
+                    tev.key, tev.internal_port
+                ))
+            })?;
+            let rendered = render_jinja(&tev.value_template, context! { host_port => host_port })?;
+            self.env_vars.push((tev.key, rendered));
+        }
+        Ok(())
+    }
+
     /// Start the container described by this configuration.
     ///
     /// Builds the Docker image, applies port mappings, environment variables,
@@ -119,6 +190,7 @@ mod tests {
         assert!(config.env_vars.is_empty());
         assert!(config.entrypoint.is_none());
         assert!(config.mapped_ports.is_empty());
+        assert!(config.templated_env_vars.is_none());
     }
 
     #[test]
@@ -160,4 +232,98 @@ mod tests {
         // Stop the container via testcontainers stop().
         container.stop().await.expect("container should stop");
     }
+
+    #[test]
+    fn env_host_port_stores_templated_var() {
+        let config = ContainerConfig::new("kafka", "7.5.0").env_host_port(
+            "KAFKA_ADVERTISED_LISTENERS",
+            "PLAINTEXT://127.0.0.1:{{ host_port }}",
+            9092,
+        );
+        let tevs = config.templated_env_vars.as_ref().expect("should be Some");
+        assert_eq!(tevs.len(), 1);
+        assert_eq!(tevs[0].key, "KAFKA_ADVERTISED_LISTENERS");
+        assert_eq!(
+            tevs[0].value_template,
+            "PLAINTEXT://127.0.0.1:{{ host_port }}"
+        );
+        assert_eq!(tevs[0].internal_port, 9092);
+    }
+
+    #[test]
+    fn env_host_port_multiple_preserves_order() {
+        let config = ContainerConfig::new("kafka", "7.5.0")
+            .env_host_port("FIRST", "{{ host_port }}", 9092)
+            .env_host_port("SECOND", "{{ host_port }}", 8080);
+        let tevs = config.templated_env_vars.as_ref().expect("should be Some");
+        assert_eq!(tevs.len(), 2);
+        assert_eq!(tevs[0].key, "FIRST");
+        assert_eq!(tevs[0].internal_port, 9092);
+        assert_eq!(tevs[1].key, "SECOND");
+        assert_eq!(tevs[1].internal_port, 8080);
+    }
+
+    #[test]
+    fn resolve_templated_env_vars_renders_and_consumes() {
+        let mut config = ContainerConfig::new("kafka", "7.5.0").env_host_port(
+            "LISTENERS",
+            "PLAINTEXT://127.0.0.1:{{ host_port }}",
+            9092,
+        );
+        // Simulate port allocation during wiring.
+        let _ = config.mapped_ports.insert(9092, 54321);
+
+        config
+            .resolve_templated_env_vars()
+            .expect("resolve should succeed");
+
+        assert!(config.templated_env_vars.is_none());
+        assert_eq!(config.env_vars.len(), 1);
+        assert_eq!(config.env_vars[0].0, "LISTENERS");
+        assert_eq!(config.env_vars[0].1, "PLAINTEXT://127.0.0.1:54321");
+    }
+
+    #[test]
+    fn resolve_templated_env_vars_preserves_order() {
+        let mut config = ContainerConfig::new("img", "tag")
+            .env_host_port("FIRST", "a:{{ host_port }}", 9092)
+            .env_host_port("SECOND", "b:{{ host_port }}", 8080);
+        let _ = config.mapped_ports.insert(9092, 11111);
+        let _ = config.mapped_ports.insert(8080, 22222);
+
+        config
+            .resolve_templated_env_vars()
+            .expect("resolve should succeed");
+
+        assert!(config.templated_env_vars.is_none());
+        assert_eq!(config.env_vars.len(), 2);
+        assert_eq!(config.env_vars[0], ("FIRST".into(), "a:11111".into()));
+        assert_eq!(config.env_vars[1], ("SECOND".into(), "b:22222".into()));
+    }
+
+    #[test]
+    fn resolve_templated_env_vars_unmapped_port_errors() {
+        let mut config =
+            ContainerConfig::new("img", "tag").env_host_port("MY_VAR", "{{ host_port }}", 5432);
+
+        let err = config
+            .resolve_templated_env_vars()
+            .expect_err("should error on unmapped port");
+        assert!(matches!(err, ValidationError::Config(_)));
+        assert!(err.to_string().contains("unmapped internal port"));
+        assert!(err.to_string().contains("MY_VAR"));
+    }
+
+    #[test]
+    fn resolve_templated_env_vars_none_is_noop() {
+        let mut config = ContainerConfig::new("img", "tag");
+        assert!(config.templated_env_vars.is_none());
+
+        config
+            .resolve_templated_env_vars()
+            .expect("noop resolve should succeed");
+
+        assert!(config.templated_env_vars.is_none());
+        assert!(config.env_vars.is_empty());
+    }
 }