add consul configuration

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

pyproject.toml

@@ -29,7 +29,8 @@ dependencies = [
     "opentelemetry-instrumentation-pika",
     "pika",
     "psutil",
-    "python-multipart"
+    "python-multipart",
+    "consul-kv"
 ]
 
 [tool.setuptools.packages.find]

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

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}")