diff --git a/amqp/adapter/consul_global_id_generator.py b/amqp/adapter/consul_global_id_generator.py index e69de29..6f2dbc4 100644 --- a/amqp/adapter/consul_global_id_generator.py +++ b/amqp/adapter/consul_global_id_generator.py @@ -0,0 +1,103 @@ +import os +import logging +import random +import socket +import time +from typing import Optional + +import consul +from amqp.config.amq_configuration import AMQConfiguration + +logger = logging.getLogger(__name__) + +# Default configuration values for Adapter +DEFAULT_CONSUL_HOST = 'localhost' +DEFAULT_CONSUL_PORT = 8500 +DEFAULT_CONSUL_COUNTER_KEY = 'adapter/ids/counter' +DEFAULT_MAX_RETRIES = 5 +DEFAULT_RETRY_DELAY_SECONDS = 0.1 + +def get_config_values(): + """ + Get configuration values from AMQConfiguration or environment variables. + + Returns: + tuple: (consul_host, consul_port, consul_counter_key, max_retries, retry_delay_seconds) + """ + try: + config = AMQConfiguration("application.properties") + consul_host = os.environ.get('CONSUL_HOST', DEFAULT_CONSUL_HOST) + consul_port = int(os.environ.get('CONSUL_PORT', DEFAULT_CONSUL_PORT)) + consul_counter_key = os.environ.get('CONSUL_COUNTER_KEY', DEFAULT_CONSUL_COUNTER_KEY) + max_retries = int(os.environ.get('CONSUL_MAX_RETRIES', DEFAULT_MAX_RETRIES)) + retry_delay_seconds = float(os.environ.get('CONSUL_RETRY_DELAY_SECONDS', DEFAULT_RETRY_DELAY_SECONDS)) + return consul_host, consul_port, consul_counter_key, max_retries, retry_delay_seconds + except Exception as e: + logger.warning(f"Failed to load configuration: {str(e)}. Using default values.") + return ( + os.environ.get('CONSUL_HOST', DEFAULT_CONSUL_HOST), + int(os.environ.get('CONSUL_PORT', DEFAULT_CONSUL_PORT)), + os.environ.get('CONSUL_COUNTER_KEY', DEFAULT_CONSUL_COUNTER_KEY), + int(os.environ.get('CONSUL_MAX_RETRIES', DEFAULT_MAX_RETRIES)), + float(os.environ.get('CONSUL_RETRY_DELAY_SECONDS', DEFAULT_RETRY_DELAY_SECONDS)) + ) +def generate_fallback_id() -> int: + """ + Generate a fallback ID when Consul is not available. + Uses a combination of timestamp, hostname hash, and random number. + + Returns: + int: A reasonably unique integer ID + """ + timestamp = int(time.time() * 1000) + hostname = socket.gethostname() + hostname_hash = hash(hostname) % 10000 + random_part = random.randint(0, 9999) + unique_id = (timestamp * 100000000) + (hostname_hash * 10000) + random_part + logger.warning(f"Using fallback ID generation method: {unique_id}") + return unique_id + +def get_unique_instance_id() -> int: + """ + Get a globally unique ID from Consul. + Uses Consul's atomic Compare-And-Set operations to safely increment a counter. + Falls back to a local generation method if Consul is unavailable. + + Returns: + int: A globally unique integer ID + """ + consul_host, consul_port, consul_counter_key, max_retries, retry_delay_seconds = get_config_values() + try: + c = consul.Consul(host=consul_host, port=consul_port) + index, data = c.kv.get(consul_counter_key) + if data is None: + logger.info(f"Initializing Consul counter at {consul_counter_key}") + if c.kv.put(consul_counter_key, "1"): + return 1 + else: + logger.error("Failed to initialize Consul counter") + return generate_fallback_id() + current_value = int(data['Value'].decode('utf-8')) + new_value = current_value + 1 + for attempt in range(max_retries): + success = c.kv.put( + consul_counter_key, + str(new_value), + cas=data['ModifyIndex'] + ) + if success: + logger.debug(f"Successfully obtained unique ID: {new_value}") + return new_value + logger.debug(f"CAS update failed on attempt {attempt+1}, retrying...") + time.sleep(retry_delay_seconds) + index, data = c.kv.get(consul_counter_key) + if data is None: + logger.error("Counter disappeared during update") + return generate_fallback_id() + current_value = int(data['Value'].decode('utf-8')) + new_value = current_value + 1 + logger.error(f"Failed to update counter after {max_retries} attempts") + return generate_fallback_id() + except Exception as e: + logger.error(f"Error accessing Consul: {str(e)}") + return generate_fallback_id() diff --git a/amqp/service/consul_global_id_generator.py b/amqp/service/consul_global_id_generator.py index f97debe..1c7ba8b 100644 --- a/amqp/service/consul_global_id_generator.py +++ b/amqp/service/consul_global_id_generator.py @@ -153,9 +153,10 @@ def get_unique_instance_id() -> int: Returns: int: A globally unique integer ID """ + consul_host, consul_port, consul_counter_key, max_retries, retry_delay_seconds = get_config_values() try: # Connect to Consul - c = consul.Consul(host=CONSUL_HOST, port=CONSUL_PORT) + c = consul.Consul(host=consul_host, port=consul_port) # Try to get the current counter value index, data = c.kv.get(CONSUL_COUNTER_KEY) @@ -206,3 +207,27 @@ def get_unique_instance_id() -> int: except Exception as e: logger.error(f"Error accessing Consul: {str(e)}") return generate_fallback_id() +def get_config_values(): + """ + Get configuration values from AMQConfiguration or environment variables. + + Returns: + tuple: (consul_host, consul_port, consul_counter_key, max_retries, retry_delay_seconds) + """ + try: + config = AMQConfiguration("application.properties") + consul_host = os.environ.get('CONSUL_HOST', 'localhost') + consul_port = int(os.environ.get('CONSUL_PORT', 8500)) + consul_counter_key = os.environ.get('CONSUL_COUNTER_KEY', 'service/ids/counter') + max_retries = int(os.environ.get('CONSUL_MAX_RETRIES', 5)) + retry_delay_seconds = float(os.environ.get('CONSUL_RETRY_DELAY_SECONDS', 0.1)) + return consul_host, consul_port, consul_counter_key, max_retries, retry_delay_seconds + except Exception as e: + logger.warning(f"Failed to load configuration: {str(e)}. Using default values.") + return ( + os.environ.get('CONSUL_HOST', 'localhost'), + int(os.environ.get('CONSUL_PORT', 8500)), + os.environ.get('CONSUL_COUNTER_KEY', 'service/ids/counter'), + int(os.environ.get('CONSUL_MAX_RETRIES', 5)), + float(os.environ.get('CONSUL_RETRY_DELAY_SECONDS', 0.1)) + )