diff --git a/otdemo/otdemo.properties b/otdemo/otdemo.properties index d81a7c6..e330510 100644 --- a/otdemo/otdemo.properties +++ b/otdemo/otdemo.properties @@ -5,7 +5,7 @@ # CleverMicro AMQ Adapter settings that identify this AMQ Adapter instance to other adapters # A name to identify this service for the AMQ Adapter -cm.amq-adapter.service-name=otdemo +cm.amq-adapter.service-name=cleverswarm # The CleverMicro AMQ Adapter instance ID, used to identify this instance in the CleverMicro AMQ Adapter cluster cm.amq-adapter.generator-id=1 # Route mapping. For route detail description see section 3.1 in Onboarding Guide for Python: @@ -31,7 +31,7 @@ cm.dispatch.download-dir=/tmp/downloaded_files # Backpressure monitor settings # Define maximum number of messages of being processed in parallel before triggering backpressure OVERLOAD alert -cm.backpressure.threshold=5 +cm.backpressure.threshold=1 # Define maximum CPU usage (%) before triggering backpressure OVERLOAD alert cm.backpressure.threshold-cpu-overload=90 # Define maximum CPU usage (%) before triggering backpressure IDLE alert diff --git a/pyproject.toml b/pyproject.toml index 68bd9e4..87be1c7 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -29,7 +29,8 @@ dependencies = [ "opentelemetry-instrumentation-pika", "pika", "psutil", - "python-multipart" + "python-multipart", + "consul-kv" ] [tool.setuptools.packages.find] diff --git a/requirements.txt b/requirements.txt index d4c2447..798a2a6 100644 --- a/requirements.txt +++ b/requirements.txt @@ -9,3 +9,4 @@ opentelemetry-exporter-prometheus==0.55b1 opentelemetry-instrumentation-fastapi==0.55b1 opentelemetry-instrumentation-httpx==0.55b1 protobuf==5.28.3 +consul-kv==0.7.4 diff --git a/tests/adapter/perf_test_consul_global_id_generator.py b/tests/adapter/perf_test_consul_global_id_generator.py new file mode 100644 index 0000000..9a5b36e --- /dev/null +++ b/tests/adapter/perf_test_consul_global_id_generator.py @@ -0,0 +1,77 @@ +import logging +import os +import threading + +from amqp.service.consul_global_id_generator import get_unique_instance_id + +CONSUL_HOST = os.environ.get("CONSUL_HOST", "localhost") +CONSUL_PORT = int(os.environ.get("CONSUL_PORT", 8500)) +CONSUL_COUNTER_KEY = "service/ids/counter" +MAX_RETRIES = 5 # Max attempts for atomic update +RETRY_DELAY_SECONDS = 0.1 # Delay between retries for atomic update + +# --- Logging Setup --- +logging.basicConfig( + level=logging.INFO, format="%(asctime)s - %(threadName)s - %(levelname)s - %(message)s" +) +logger = logging.getLogger(__name__) + + +# --- Demo: Simulate Concurrent Access --- +def worker_thread(results_list): + """A function to be run by multiple threads to get unique IDs.""" + instance_id = get_unique_instance_id() + results_list.append(instance_id) + logger.info(f"Thread {threading.current_thread().name} got ID: {instance_id}") + + +if __name__ == "__main__": + num_threads = 10 # Number of concurrent service instances to simulate + all_generated_ids = [] + + print(f"\n--- Simulating {num_threads} concurrent service instances requesting unique IDs ---") + + threads = [] + for i in range(num_threads): + thread = threading.Thread( + target=worker_thread, args=(all_generated_ids,), name=f"ServiceInstance-{i + 1}" + ) + threads.append(thread) + thread.start() + + # Wait for all threads to complete + for thread in threads: + thread.join() + + print("\n--- All unique IDs generated ---") + print(f"Generated IDs: {sorted(all_generated_ids)}") + + # Verify uniqueness + if len(all_generated_ids) == len(set(all_generated_ids)): + print("All generated IDs are unique (as expected).") + else: + print( + "WARNING: Duplicate IDs found! This might indicate a problem with atomicity or fallback logic." + ) + # Identify duplicates + seen = set() + duplicates = set() + for x in all_generated_ids: + if x in seen: + duplicates.add(x) + seen.add(x) + print(f"Duplicate IDs: {duplicates}") + + print("\n--- Testing Consul Unavailability Fallback ---") + # Temporarily change Consul port to simulate unavailability + original_consul_port = CONSUL_PORT + CONSUL_PORT = 1 # Use an unlikely port + + # Clear any existing ConsulKV client cache if applicable (consul_kv typically creates new client per call) + # If using a persistent client, you'd need to invalidate/reinitialize it here. + + fallback_id_1 = get_unique_instance_id() + fallback_id_2 = get_unique_instance_id() + + print(f"Fallback ID 1: {fallback_id_1}") + print(f"Fallback ID 2: {fallback_id_2}")