Add otel collector installation to the bootstrap script

Author: emasab

tests/soak/README.md

@@ -10,10 +10,24 @@ OpenTelemetry reporting supported through OTLP.
 
 # Installation
 
-TESTID=normal \
-LK_VERSION=v2.2.0 \
-CKPY_VERSION=v2.2.0 \
-CC_BOOSTRAP_SERVERS=_ \
-CC_USERNAME=_ \
-CC_PASSWORD=_ \
-DOCKER_REPOSITORY=_ ./install.sh
+1. Edit `ccloud.config`
+
+2. Edit `otel-config.yaml`
+
+3. the first time:
+```bash
+./bootstrap.sh <python-branch/tag> <librdkafka-branch/tag>
+```
+4. next times:
+```bash
+./build.sh <python-branch/tag> <librdkafka-branch/tag>
+```
+
+5. 
+```bash
+. venv/bin/activate
+
+5. Run some tests
+```bash
+TESTID=<testid> ./run.sh ccloud.config
+```

tests/soak/bootstrap.sh

@@ -16,11 +16,20 @@ fi
 
 python_branch=$1
 librdkafka_branch=$2
+otel_collector_version=0.130.0
+otel_collector_package_url="https://github.com/open-telemetry/"\
+"opentelemetry-collector-releases/releases/download/"\
+"v${otel_collector_version}/otelcol-contrib_${otel_collector_version}_linux_amd64.deb"
 venv=$PWD/venv
 
 sudo apt update
-sudo apt install -y git curl make gcc g++ zlib1g-dev libssl-dev libzstd-dev \
-    python3-dev python3-pip python3-venv
+sudo apt install -y git curl wget make gcc g++ zlib1g-dev libssl-dev \
+    libzstd-dev python3-dev python3-pip python3-venv
+wget -O otel_collector_package.deb $otel_collector_package_url
+sudo dpkg -i otel_collector_package.deb
+rm otel_collector_package.deb
+sudo cp otel-config.yaml /etc/otelcol-contrib/config.yaml
+sudo systemctl restart otelcol-contrib
 
 if [[ ! -d confluent-kafka-python ]]; then
     git clone https://github.com/confluentinc/confluent-kafka-python

tests/soak/soakclient.py

@@ -26,12 +26,11 @@
 #
 
 from confluent_kafka import KafkaError, KafkaException, version
-from confluent_kafka import Producer
+from confluent_kafka import Producer, Consumer
 from confluent_kafka.admin import AdminClient, NewTopic
 from collections import defaultdict
 from builtins import int
 from opentelemetry import metrics
-from common import TestConsumer
 import argparse
 import threading
 import time
@@ -447,7 +446,7 @@ def filter_config(conf, filter_out, strip_prefix):
         cconf['error_cb'] = self.consumer_error_cb
         cconf['on_commit'] = self.consumer_commit_cb
         self.logger.info("consumer: using group.id {}".format(cconf['group.id']))
-        self.consumer = TestConsumer(cconf)
+        self.consumer = Consumer(cconf)
 
         # Create and start producer thread
         self.producer_thread = threading.Thread(target=self.producer_thread_main)