Stanislav Hejny
99b7f73447
Co-authored-by: aider (openrouter/openai/o3-mini-high) <aider@aider.chat>
104 lines
4.2 KiB
Python
104 lines
4.2 KiB
Python
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()
|