Stanislav Hejny
79 lines
2.7 KiB
Python
79 lines
2.7 KiB
Python
import logging
|
|
import os
|
|
import threading
|
|
|
|
from amqp.adapter.consul_global_id_generator import get_unique_instance_id
|
|
from amqp.config.amq_configuration import AMQConfiguration
|
|
|
|
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(AMQConfiguration("").amq_adapter)
|
|
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}")
|