test: add PrivateLink cloudtest with Toxiproxy simulation

Adds cloudtest infrastructure for testing Kafka PrivateLink connections
using Toxiproxy as a network proxy to simulate VPC endpoint routing.

Includes two tests:

- `test_privatelink_e2e_connectivity`: validates basic connectivity
  through a simulated PrivateLink path, tests failure detection when
  the proxy is disabled, and recovery when re-enabled.

- `test_privatelink_pattern_matching`: patches Redpanda to advertise
  an AZ-specific broker address, then verifies that MATCHING rules
  route post-metadata traffic through the AZ-specific proxy. The
  default proxy is disabled after bootstrap to prove pattern matching
  works.

Also adds `doc/developer/testing-confluent-privatelink.md` with a
step-by-step guide for manual testing against Confluent Cloud
PrivateLink using a scratch VM with dnsmasq DNS overrides.

Fixes the cloudtest `reset` script to clean up configmaps and
vpcendpoints.

Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>

Author: jubrad

doc/developer/testing-confluent-privatelink.md

@@ -0,0 +1,331 @@
+# Testing Confluent Cloud PrivateLink Connectivity
+
+This guide walks through setting up a scratch VM to test Kafka connectivity
+to a Confluent Cloud cluster over AWS PrivateLink. This is useful for
+debugging transport-level issues (SNI, TLS, broker routing) without needing
+a full Materialize environment.
+
+## Prerequisites
+
+- A Confluent Cloud dedicated cluster with PrivateLink enabled
+- A Confluent Cloud API key for the cluster
+- Access to `bin/scratch` in the Materialize repo
+
+## 1. Deploy a scratch instance
+
+```bash
+bin/scratch create <name>
+```
+
+## 2. Set up a VPC endpoint to Confluent
+
+Create a VPC endpoint in the scratch instance's VPC pointing to the Confluent
+PrivateLink service. You'll need the VPC endpoint service name from the
+Confluent Cloud networking settings (e.g.,
+`com.amazonaws.vpce.us-east-1.vpce-svc-006836f18cb2d0819`).
+
+### Allow access from the scratch VM's AWS account
+
+Confluent requires you to explicitly allow each AWS account that connects via
+PrivateLink. In the Confluent Cloud console, go to your cluster's networking
+settings and add the scratch VM's AWS account ID to the PrivateLink access
+list. Without this, the VPC endpoint will appear "available" in AWS but all
+traffic will be silently dropped.
+
+### Enable the correct AZs
+
+AWS creates a separate ENI per AZ, each with its own private IP. Confluent
+brokers in a given AZ are only reachable through that AZ's ENI — traffic to
+the wrong AZ's ENI will black-hole.
+
+Confluent requires you to choose at least 3 AZs when configuring PrivateLink.
+However, **only the AZs where your Materialize compute is running will
+actually carry traffic.** When creating the VPC endpoint in AWS, you only
+need to enable subnets in the AZs that match your compute placement.
+
+For the scratch VM test, enable the VPC endpoint in the AZ where the scratch
+instance is running. For a production Materialize deployment, enable the AZs
+where your cluster replicas are scheduled.
+
+Note each AZ's ENI private IP and which AWS AZ it's in. You can find these in
+the AWS console under the VPC endpoint's "Subnets" tab, or via:
+
+```bash
+aws ec2 describe-vpc-endpoints --vpc-endpoint-ids <VPCE_ID> \
+  --query 'VpcEndpoints[0].NetworkInterfaceIds' --output text | \
+  xargs -I{} aws ec2 describe-network-interfaces --network-interface-ids {} \
+  --query 'NetworkInterfaces[*].[AvailabilityZone,PrivateIpAddress]' --output table
+```
+
+You'll need to map each Confluent AZ name (e.g., `use1-az1`) to the correct
+ENI IP. AWS AZ IDs (`use1-az1`) map to different physical zone names
+(`us-east-1a`, `us-east-1b`, etc.) per account, so verify the mapping
+carefully.
+
+## 3. SSH into the scratch VM
+
+```bash
+bin/scratch ssh <name>
+```
+
+## 4. Configure DNS with dnsmasq
+
+Confluent PrivateLink requires that the cluster's DNS hostnames resolve to the
+VPC endpoint IP. Since the scratch VM doesn't have Confluent's private hosted
+zones, we use dnsmasq to override DNS resolution.
+
+### Install dnsmasq
+
+```bash
+sudo apt-get update && sudo apt-get install -y dnsmasq
+```
+
+### Disable systemd-resolved stub listener
+
+systemd-resolved binds to port 53, which conflicts with dnsmasq.
+
+```bash
+sudo mkdir -p /etc/systemd/resolved.conf.d
+sudo tee /etc/systemd/resolved.conf.d/noresolvstub.conf <<EOF
+[Resolve]
+DNSStubListener=no
+DNS=127.0.0.1
+EOF
+```
+
+### Point resolv.conf at dnsmasq
+
+```bash
+sudo rm /etc/resolv.conf
+echo "nameserver 127.0.0.1" | sudo tee /etc/resolv.conf
+```
+
+### Configure dnsmasq upstream DNS
+
+The VPC DNS resolver must be configured as the upstream so that non-Confluent
+DNS queries still work. On AWS, this is typically `169.254.169.253`.
+
+```bash
+echo "server=169.254.169.253" | sudo tee /etc/dnsmasq.d/upstream.conf
+```
+
+### Add Confluent PrivateLink DNS overrides
+
+Confluent PrivateLink uses AZ-specific DNS subdomains for broker routing.
+Each Confluent AZ has its own VPC endpoint ENI, and brokers in that AZ are
+only reachable through the matching ENI. The DNS must route each AZ's broker
+hostnames to the correct ENI IP.
+
+Confluent broker addresses follow the pattern:
+- Bootstrap/API: `lkc-XXXXX.<ENDPOINT_DOMAIN>:9092` (no AZ prefix, can use any ENI)
+- Brokers: `b0-lkc-XXXXX.<AZ>.<ENDPOINT_DOMAIN>:9092` (AZ-specific, must use that AZ's ENI)
+
+Replace `<ENDPOINT_DOMAIN>` with the Confluent endpoint domain and each
+`<AZ*_IP>` with the ENI private IP for that specific Confluent AZ.
+
+```bash
+sudo tee /etc/dnsmasq.d/confluent-privatelink.conf <<EOF
+# AZ-specific: each AZ's brokers MUST resolve to that AZ's VPC endpoint ENI.
+# Brokers are only reachable through their own AZ's ENI.
+address=/<AZ1>.<ENDPOINT_DOMAIN>/<AZ1_IP>
+address=/<AZ2>.<ENDPOINT_DOMAIN>/<AZ2_IP>
+address=/<AZ3>.<ENDPOINT_DOMAIN>/<AZ3_IP>
+
+# Base domain: bootstrap and API addresses have no AZ prefix.
+# These can route through any AZ's ENI.
+address=/<ENDPOINT_DOMAIN>/<AZ1_IP>
+EOF
+```
+
+For example, with endpoint domain `dom8pmk29rw.us-east-1.aws.confluent.cloud`
+and three AZs:
+
+```bash
+sudo tee /etc/dnsmasq.d/confluent-privatelink.conf <<EOF
+# AZ-specific: broker traffic must go through the correct AZ's ENI
+address=/use1-az1.dom8pmk29rw.us-east-1.aws.confluent.cloud/172.31.10.100
+address=/use1-az4.dom8pmk29rw.us-east-1.aws.confluent.cloud/172.31.20.100
+address=/use1-az6.dom8pmk29rw.us-east-1.aws.confluent.cloud/172.31.30.100
+
+# Bootstrap/API: no AZ in hostname, can go through any ENI
+address=/dom8pmk29rw.us-east-1.aws.confluent.cloud/172.31.10.100
+EOF
+```
+
+Note: dnsmasq wildcard matching is suffix-based. `address=/foo.example.com/IP`
+matches `foo.example.com` and `anything.foo.example.com`. The more specific
+AZ entries take precedence over the base domain entry, so a broker address
+like `b0-lkc-825730.use1-az1.dom8pmk29rw.us-east-1.aws.confluent.cloud`
+resolves to the `use1-az1` ENI IP, while the bootstrap address
+`lkc-825730.dom8pmk29rw.us-east-1.aws.confluent.cloud` resolves to the
+base domain IP.
+
+### Restart services
+
+```bash
+sudo systemctl restart systemd-resolved
+sudo systemctl restart dnsmasq
+```
+
+### Verify
+
+```bash
+nslookup google.com                                       # upstream works
+nslookup lkc-XXXXX.<ENDPOINT_DOMAIN>                      # bootstrap -> base IP
+nslookup b0-lkc-XXXXX.<AZ1>.<ENDPOINT_DOMAIN>             # AZ1 broker -> AZ1 IP
+nslookup b0-lkc-XXXXX.<AZ2>.<ENDPOINT_DOMAIN>             # AZ2 broker -> AZ2 IP
+nslookup b0-lkc-XXXXX.<AZ3>.<ENDPOINT_DOMAIN>             # AZ3 broker -> AZ3 IP
+```
+
+Each AZ lookup should return that AZ's ENI IP, not the base domain IP.
+
+## 5. Install Kafka tools
+
+```bash
+sudo apt-get install -y kafkacat default-jre-headless
+curl -L https://archive.apache.org/dist/kafka/3.9.0/kafka_2.13-3.9.0.tgz | tar xz
+```
+
+## 6. Configure Kafka authentication
+
+```bash
+cat > /tmp/kafka.properties <<EOF
+security.protocol=SASL_SSL
+sasl.mechanism=PLAIN
+sasl.jaas.config=org.apache.kafka.common.security.plain.PlainLoginModule required username="<API_KEY>" password="<API_SECRET>";
+EOF
+```
+
+## 7. Verify connectivity
+
+```bash
+BOOTSTRAP="<CLUSTER_ID>.<ENDPOINT_DOMAIN>:9092"
+
+# List brokers and topics
+kcat -b "$BOOTSTRAP" \
+  -X security.protocol=SASL_SSL \
+  -X sasl.mechanisms=PLAIN \
+  -X sasl.username=<API_KEY> \
+  -X sasl.password='<API_SECRET>' \
+  -L
+```
+
+## 8. Create a test topic
+
+Confluent Cloud doesn't allow auto-topic creation. Since the cluster is
+private, you can't create topics from outside the VPC. Use `kafka-topics.sh`
+from the scratch VM:
+
+```bash
+kafka_2.13-3.9.0/bin/kafka-topics.sh \
+  --bootstrap-server "$BOOTSTRAP" \
+  --command-config /tmp/kafka.properties \
+  --create --topic test-privatelink --partitions 3 --replication-factor 3
+```
+
+## 9. Produce test data
+
+```bash
+for i in $(seq 1 1000); do
+  echo "{\"id\": $i, \"ts\": \"$(date -u +%Y-%m-%dT%H:%M:%SZ)\", \"value\": $((RANDOM % 10000))}"
+done | kcat -b "$BOOTSTRAP" \
+  -X security.protocol=SASL_SSL \
+  -X sasl.mechanisms=PLAIN \
+  -X sasl.username=<API_KEY> \
+  -X sasl.password='<API_SECRET>' \
+  -P -t test-privatelink
+```
+
+### Verify
+
+```bash
+kcat -b "$BOOTSTRAP" \
+  -X security.protocol=SASL_SSL \
+  -X sasl.mechanisms=PLAIN \
+  -X sasl.username=<API_KEY> \
+  -X sasl.password='<API_SECRET>' \
+  -C -t test-privatelink -e -q | wc -l
+```
+
+## 10. Test from Materialize
+
+Once data is flowing, create the source in your Materialize environment:
+
+```sql
+CREATE SECRET confluent_api_secret AS '<API_SECRET>';
+
+-- The AVAILABILITY ZONES here only need to include the AZs where your
+-- Materialize compute is running. Confluent requires 3 AZs on their side,
+-- but only the AZ(s) matching your compute placement will carry traffic.
+CREATE CONNECTION confluent_privatelink
+TO AWS PRIVATELINK (
+    SERVICE NAME '<VPCE_SERVICE_NAME>',
+    AVAILABILITY ZONES ('<AZ1>', '<AZ2>', '<AZ3>')
+);
+
+-- Wait for the VPC endpoint to become available. You must also add the
+-- Materialize AWS account to the Confluent PrivateLink access list.
+-- Then:
+
+CREATE CONNECTION confluent_kafka TO KAFKA (
+    AWS PRIVATELINKS (
+        '<CLUSTER_ID>.<ENDPOINT_DOMAIN>:9092'
+            TO confluent_privatelink (AVAILABILITY ZONE '<AZ>'),
+        '*<AZ1>*' TO confluent_privatelink (AVAILABILITY ZONE '<AZ1>'),
+        '*<AZ2>*' TO confluent_privatelink (AVAILABILITY ZONE '<AZ2>'),
+        '*<AZ3>*' TO confluent_privatelink (AVAILABILITY ZONE '<AZ3>')
+    ),
+    SASL MECHANISMS 'PLAIN',
+    SASL USERNAME '<API_KEY>',
+    SASL PASSWORD SECRET confluent_api_secret,
+    SECURITY PROTOCOL 'SASL_SSL'
+);
+
+CREATE SOURCE test_privatelink_source
+FROM KAFKA CONNECTION confluent_kafka (
+    TOPIC 'test-privatelink'
+);
+
+CREATE TABLE test_privatelink_tbl
+FROM SOURCE test_privatelink_source (
+    REFERENCE "test-privatelink"
+)
+FORMAT BYTES
+ENVELOPE NONE;
+
+SELECT COUNT(*) FROM test_privatelink_tbl;
+```
+
+## Troubleshooting
+
+### Transport failure on connection validation
+
+Check which AZ endpoints are reachable. Not all VPC endpoint ENIs may have
+active NLB targets behind them:
+
+```bash
+# From inside the environmentd pod
+for ip in <IP1> <IP2> <IP3>; do
+  timeout 3 bash -c "echo > /dev/tcp/$ip/9092" 2>&1 && echo "$ip: OPEN" || echo "$ip: CLOSED"
+done
+```
+
+If only some IPs are reachable, ensure the Confluent PrivateLink is enabled
+for all the AZs you specified, or change the bootstrap rule to use a working AZ.
+
+### Enable librdkafka debug logging
+
+From a Materialize SQL session:
+
+```sql
+ALTER SYSTEM SET log_filter = 'info,librdkafka=debug';
+```
+
+Or in launchdarkly set your org up with the kafka debug variant.
+
+Then recreate the connection to trigger validation. Check environmentd logs
+for SSL handshake details, connection state transitions, etc. Reset after:
+
+```sql
+ALTER SYSTEM RESET log_filter;
+```