Add environment manager for kafka

scripts/integration/kafka/Cargo.toml

@@ -0,0 +1,16 @@
+[package]
+name = "vector-kafka-env-manager"
+version = "0.1.0"
+edition = "2021"
+authors = ["Vector Contributors <vector@datadoghq.com>"]
+license = "MPL-2.0"
+readme = "README.md"
+publish = false
+
+[dependencies]
+anyhow = "1.0.66"
+clap = { version = "4.0.18", features = ["derive"] }
+dunce = "1.0.3"
+serde_json = "1.0.87"
+
+[workspace]

scripts/integration/kafka/README.md

@@ -0,0 +1,11 @@
+# vector-kafka-env-manager
+
+-----
+
+## Usage
+
+```text
+vdev int show kafka
+vdev int start kafka <ENV>
+vdev int stop kafka <ENV>
+```

scripts/integration/kafka/data/docker-compose.yml

@@ -0,0 +1,40 @@
+version: '3'
+
+services:
+  zookeeper:
+    image: docker.io/wurstmeister/zookeeper:${KAFKA_VERSION}
+    ports:
+    - 2181:2181
+  kafka:
+    image: docker.io/wurstmeister/kafka:2.13-2.6.0
+    depends_on:
+    - zookeeper
+    environment:
+    - KAFKA_BROKER_ID=1
+    - KAFKA_ZOOKEEPER_CONNECT=zookeeper:2181
+    - KAFKA_LISTENERS=PLAINTEXT://:9091,SSL://:9092,SASL_PLAINTEXT://:9093
+    - KAFKA_ADVERTISED_LISTENERS=PLAINTEXT://kafka:9091,SSL://kafka:9092,SASL_PLAINTEXT://kafka:9093
+    - KAFKA_SSL_KEYSTORE_TYPE=PKCS12
+    - KAFKA_SSL_KEYSTORE_LOCATION=/certs/kafka.p12
+    - KAFKA_SSL_KEYSTORE_PASSWORD=NOPASS
+    - KAFKA_SSL_TRUSTSTORE_TYPE=PKCS12
+    - KAFKA_SSL_TRUSTSTORE_LOCATION=/certs/kafka.p12
+    - KAFKA_SSL_TRUSTSTORE_PASSWORD=NOPASS
+    - KAFKA_SSL_KEY_PASSWORD=NOPASS
+    - KAFKA_SSL_ENDPOINT_IDENTIFICATION_ALGORITHM=none
+    - KAFKA_OPTS=-Djava.security.auth.login.config=/etc/kafka/kafka_server_jaas.conf
+    - KAFKA_INTER_BROKER_LISTENER_NAME=SASL_PLAINTEXT
+    - KAFKA_SASL_ENABLED_MECHANISMS=PLAIN
+    - KAFKA_SASL_MECHANISM_INTER_BROKER_PROTOCOL=PLAIN
+    ports:
+    - 9091:9091
+    - 9092:9092
+    - 9093:9093
+    volumes:
+    - ../../../../tests/data/ca/intermediate_server/private/kafka.p12:/certs/kafka.p12:ro
+    - ../../../../tests/data/kafka_server_jaas.conf:/etc/kafka/kafka_server_jaas.conf
+
+networks:
+  default:
+    name: ${VECTOR_NETWORK}
+    external: true

scripts/integration/kafka/src/core.rs

@@ -0,0 +1,61 @@
+use anyhow::{bail, Result};
+use serde_json::Value;
+use std::path::PathBuf;
+use std::process::Command;
+use std::thread;
+use std::time::Duration;
+
+pub fn start(config: Value) -> Result<()> {
+    let mut command = compose_command();
+    command.args(["up", "-d"]);
+
+    apply_env_vars(&mut command, &config);
+
+    let status = command.status()?;
+    if status.success() {
+        thread::sleep(Duration::from_secs(20));
+        return Ok(());
+    } else {
+        bail!("failed to execute: {}", render_command(&mut command));
+    }
+}
+
+pub fn stop(config: Value) -> Result<()> {
+    let mut command = compose_command();
+    command.args(["down", "-t", "0"]);
+
+    apply_env_vars(&mut command, &config);
+
+    let status = command.status()?;
+    if status.success() {
+        return Ok(());
+    } else {
+        bail!("failed to execute: {}", render_command(&mut command));
+    }
+}
+
+fn compose_command() -> Command {
+    let path = PathBuf::from_iter(["data", "docker-compose.yml"].iter());
+    let compose_file = match dunce::canonicalize(&path) {
+        Ok(p) => p.display().to_string(),
+        Err(_) => path.display().to_string(),
+    };
+
+    let mut command = Command::new("docker");
+    command.args(["compose", "-f", &compose_file]);
+    command
+}
+
+fn apply_env_vars(command: &mut Command, config: &Value) {
+    if let Some(version) = config.get("version") {
+        command.env("KAFKA_VERSION", version.as_str().unwrap());
+    }
+}
+
+fn render_command(command: &mut Command) -> String {
+    format!(
+        "{} {}",
+        command.get_program().to_str().unwrap(),
+        Vec::from_iter(command.get_args().map(|arg| arg.to_str().unwrap())).join(" ")
+    )
+}