Stanislav Hejny
242 lines
8.9 KiB
Python
242 lines
8.9 KiB
Python
import configparser
|
|
import inspect
|
|
import os
|
|
|
|
from amqp.adapter.logging_utils import logging_error, logging_warning
|
|
|
|
"""
|
|
Convert configuration from properties file to an object.
|
|
"""
|
|
|
|
|
|
class AMQAdapter:
|
|
"""
|
|
configuration with prefix 'cm.amq-adapter'
|
|
"""
|
|
|
|
PREFIX: str = "cm.amq-adapter."
|
|
|
|
def __init__(self, config: configparser.ConfigParser):
|
|
"""
|
|
cm.amq-adapter.service-name=amq-adapter
|
|
cm.amq-adapter.generator-id=164
|
|
cm.amq-adapter.route-mapping=
|
|
|
|
:param config: config parser to read configuration values
|
|
"""
|
|
self.generator_id = 0
|
|
self.service_name = config.get(
|
|
"CleverMicro-AMQ",
|
|
self.PREFIX + "service-name",
|
|
fallback="amq-python-adapter-" + str(self.generator_id),
|
|
)
|
|
self.route_mapping = config.get(
|
|
"CleverMicro-AMQ", self.PREFIX + "route-mapping", fallback=None
|
|
)
|
|
self.require_authenticated_user = config.getboolean(
|
|
"CleverMicro-AMQ",
|
|
self.PREFIX + "require-authenticated-user",
|
|
fallback=False,
|
|
)
|
|
self.backpressure_enabled = config.getboolean(
|
|
"CleverMicro-AMQ",
|
|
self.PREFIX + "default-backpressure-enabled",
|
|
fallback=False,
|
|
)
|
|
# these 2 shall be provided by Docker swarm, format is a random alphanumeric string
|
|
self.swarm_service_id = os.getenv("SWARM_SERVICE_ID", default="clever-amqp-service")
|
|
self.swarm_task_id = os.getenv("SWARM_TASK_ID", default="python-adapter")
|
|
#
|
|
# Consul related configuration
|
|
self.consul_port = (
|
|
int(
|
|
os.getenv(
|
|
"CONSUL_PORT",
|
|
config.get("CleverMicro-AMQ", self.PREFIX + "consul-port", fallback="8500"),
|
|
)
|
|
)
|
|
if os.getenv("CONSUL_PORT") is not None
|
|
else config.getint("CleverMicro-AMQ", self.PREFIX + "consul-port", fallback=8500)
|
|
)
|
|
self.consul_host = os.getenv(
|
|
"CONSUL_HOST",
|
|
config.get("CleverMicro-AMQ", self.PREFIX + "consul-host", fallback="consul"),
|
|
)
|
|
self.consul_max_retries = (
|
|
int(
|
|
os.getenv(
|
|
"CONSUL_MAX_RETRIES",
|
|
config.get("CleverMicro-AMQ", self.PREFIX + "consul-max-retries", fallback="5"),
|
|
)
|
|
)
|
|
if os.getenv("CONSUL_MAX_RETRIES") is not None
|
|
else config.getint("CleverMicro-AMQ", self.PREFIX + "consul-max-retries", fallback=5)
|
|
)
|
|
self.consul_initial_retry_delay = (
|
|
float(
|
|
os.getenv(
|
|
"CONSUL_INITIAL_RETRY_DELAY",
|
|
config.get(
|
|
"CleverMicro-AMQ",
|
|
self.PREFIX + "consul-initial-retry-delay",
|
|
fallback="0.2",
|
|
),
|
|
)
|
|
)
|
|
if os.getenv("CONSUL_INITIAL_RETRY_DELAY") is not None
|
|
else config.getfloat(
|
|
"CleverMicro-AMQ", self.PREFIX + "consul-initial-retry-delay", fallback=0.2
|
|
)
|
|
)
|
|
self.consul_counter_key = (
|
|
os.getenv(
|
|
"CONSUL_COUNTER_KEY",
|
|
config.get(
|
|
"CleverMicro-AMQ",
|
|
self.PREFIX + "consul-counter-key",
|
|
fallback="service/ids/counter",
|
|
),
|
|
)
|
|
if os.getenv("CONSUL_COUNTER_KEY") is not None
|
|
else config.get(
|
|
"CleverMicro-AMQ",
|
|
self.PREFIX + "consul-counter-key",
|
|
fallback="service/ids/counter",
|
|
)
|
|
)
|
|
|
|
def adapter_prefix(self) -> str:
|
|
return f"{self.service_name}-{self.generator_id}-"
|
|
|
|
|
|
class Dispatch:
|
|
"""
|
|
configuration with prefix 'cm.dispatch'
|
|
"""
|
|
|
|
PREFIX: str = "cm.dispatch."
|
|
|
|
def __init__(self, config: configparser.ConfigParser):
|
|
"""
|
|
cm.dispatch.use-dlq=true
|
|
cm.dispatch.amq-host=rabbitmq
|
|
cm.dispatch.amq-port=5672
|
|
cm.dispatch.amq-port-tls=5671
|
|
cm.dispatch.service-host=
|
|
cm.dispatch.service-port=8080
|
|
|
|
:param config: config parser to read configuration values
|
|
"""
|
|
self.use_dlq = config.getboolean("CleverMicro-AMQ", self.PREFIX + "use-dlq", fallback=False)
|
|
self.amq_host = config.get("CleverMicro-AMQ", self.PREFIX + "amq-host", fallback="rabbitmq")
|
|
self.amq_port = config.getint("CleverMicro-AMQ", self.PREFIX + "amq-port", fallback=5672)
|
|
self.amq_port_tls = config.getint(
|
|
"CleverMicro-AMQ", self.PREFIX + "amq-port-tls", fallback=5671
|
|
)
|
|
self.service_host = config.get(
|
|
"CleverMicro-AMQ", self.PREFIX + "service-host", fallback="localhost"
|
|
)
|
|
self.service_port = config.getint(
|
|
"CleverMicro-AMQ", self.PREFIX + "service-port", fallback=8080
|
|
)
|
|
self.rabbit_mq_user = os.getenv("RABBIT_MQ_USER", "springuser")
|
|
self.rabbit_mq_password = os.getenv("RABBIT_MQ_PASSWORD", "TheCleverWho")
|
|
self.rabbit_mq_url = (
|
|
f"amqp://{self.rabbit_mq_user}:{self.rabbit_mq_password}@{self.amq_host}/"
|
|
)
|
|
self.download_dir = config.get(
|
|
"CleverMicro-AMQ", self.PREFIX + "download-dir", fallback="downloaded_files"
|
|
)
|
|
|
|
|
|
class Backpressure:
|
|
"""
|
|
configuration with prefix 'cm.backpressure'
|
|
"""
|
|
|
|
PREFIX: str = "cm.backpressure."
|
|
|
|
def __init__(self, config: configparser.ConfigParser):
|
|
"""
|
|
cm.backpressure.threshold=20
|
|
cm.backpressure.time-window=3000
|
|
|
|
:param config: config parser to read configuration values
|
|
"""
|
|
self.threshold_load = config.getint(
|
|
"CleverMicro-AMQ",
|
|
self.PREFIX + "threshold",
|
|
fallback=int(os.getenv("PARALLEL_WORKERS", 1)),
|
|
)
|
|
# time-window in seconds, duration between backpressure reports
|
|
self.time_window = config.getint(
|
|
"CleverMicro-AMQ", self.PREFIX + "time-window", fallback=10
|
|
)
|
|
|
|
# Define maximum CPU usage before triggering backpressure OVERLOAD alert
|
|
self.threshold_cpu = config.getint(
|
|
"CleverMicro-AMQ", self.PREFIX + "threshold-cpu", fallback=90
|
|
)
|
|
# Define the time window length, in seconds, during which, if CPU usage is consistently above the threshold, the system will trigger OVERLOAD alert
|
|
self.cpu_overload_duration = config.getint(
|
|
"CleverMicro-AMQ", self.PREFIX + "cpu-overload-duration", fallback=30
|
|
)
|
|
# Define the time window length, in seconds, during which, if no request is made to the Service, the system will trigger IDLE alert
|
|
self.idle_duration = config.getint(
|
|
"CleverMicro-AMQ", self.PREFIX + "idle-duration", fallback=30
|
|
)
|
|
# Flag to enable/disable CPU monitoring
|
|
self.cpu_monitoring_enabled = config.getboolean(
|
|
"CleverMicro-AMQ", self.PREFIX + "cpu-monitoring-enabled", fallback=False
|
|
)
|
|
|
|
|
|
class AMQConfiguration:
|
|
"""
|
|
Top level configuration context holder / object
|
|
"""
|
|
|
|
def __init__(self, config_file_name: str = None):
|
|
config = configparser.ConfigParser()
|
|
# Read the application.properties file
|
|
if os.path.exists(config_file_name):
|
|
config.read(config_file_name)
|
|
else:
|
|
# Get the directory of the caller (the module where this function is called)
|
|
caller_frame = inspect.stack()[1] # 0 is current frame, 1 is caller
|
|
caller_module = inspect.getmodule(caller_frame[0])
|
|
if caller_module:
|
|
caller_dir = os.path.dirname(os.path.abspath(caller_module.__file__))
|
|
config_file_path = os.path.join(caller_dir, config_file_name)
|
|
config.read(config_file_path)
|
|
else:
|
|
logging_error(f"Configuration file not found: {config_file_name}")
|
|
logging_warning(
|
|
"Defaulting to preset values, which will likely be wrong, causing errors."
|
|
)
|
|
|
|
# Override values with environment variables
|
|
for section in config.sections():
|
|
for option in config.options(section):
|
|
# env_var = f"{section.upper()}_{option.upper()}"
|
|
if option in os.environ:
|
|
config.set(section, option, os.environ[option])
|
|
|
|
#
|
|
# Add individual config areas
|
|
#
|
|
self.amq_adapter: AMQAdapter = AMQAdapter(config)
|
|
self.dispatch: Dispatch = Dispatch(config)
|
|
self.backpressure: Backpressure = Backpressure(config)
|
|
|
|
def get_unique_instance_id(self) -> int:
|
|
"""
|
|
Get a unique instance ID for this AMQ service.
|
|
|
|
Returns:
|
|
int: A unique instance ID
|
|
"""
|
|
if self.instance_id == -1:
|
|
logging_warning(f"Generated unique instance ID: {self.instance_id}")
|
|
return self.instance_id
|