in source/idea/idea-sdk/src/ideasdk/metrics/cloudwatch/cloudwatch_agent_config.py [0:0]
def __init__(self, cluster_config: ClusterConfig, options: CloudWatchAgentConfigOptions):
self.config = cluster_config
if Utils.is_empty(options.module_id):
raise exceptions.invalid_params('module_id is required')
if Utils.is_empty(options.base_os):
raise exceptions.invalid_params('base_os is required')
self.options = options
# defaults
self.DEFAULT_METRICS_COLLECTION_INTERVAL = self.config.get_int('metrics.cloudwatch.metrics_collection_interval', 60)
self.DEFAULT_METRICS_FORCE_FLUSH_INTERVAL = self.config.get_int('metrics.cloudwatch.force_flush_interval', 60)
self.DEFAULT_LOGS_FORCE_FLUSH_INTERVAL = self.config.get_int('cluster.cloudwatch_logs.force_flush_interval', 5)
self.DEFAULT_LOGS_RETENTION_IN_DAYS = self.config.get_int('cluster.cloudwatch_logs.retention_in_days', 90)
# module id
self.module_id = options.module_id
# cluster name, aws region
self.cluster_name = self.config.get_string('cluster.cluster_name', required=True)
self.aws_region = self.config.get_string('cluster.aws.region', required=True)
# platform
self.platform = Utils.get_platform(base_os=options.base_os)
# vpc endpoints
self.logs_endpoint_override: Optional[str] = None
self.metrics_endpoint_override: Optional[str] = None
use_vpc_endpoints = self.config.get_bool('cluster.network.use_vpc_endpoints', False)
if use_vpc_endpoints:
logs_vpc_endpoint_enabled = self.config.get_bool('cluster.network.vpc_interface_endpoints.logs.enabled', False)
if logs_vpc_endpoint_enabled:
logs_endpoint_url = self.config.get_string('cluster.network.vpc_interface_endpoints.logs.endpoint_url', required=True)
# "endpoint_override": "vpce-XXXXXXXXXXXXXXXXXXXXXXXXX.logs.us-east-1.vpce.amazonaws.com"
self.logs_endpoint_override = logs_endpoint_url.replace('https://', '')
metrics_vpc_endpoint_enabled = self.config.get_bool('cluster.network.vpc_interface_endpoints.monitoring.enabled', False)
if metrics_vpc_endpoint_enabled:
metrics_endpoint_url = self.config.get_string('cluster.network.vpc_interface_endpoints.monitoring.endpoint_url', required=True)
# "endpoint_override": "vpce-XXXXXXXXXXXXXXXXXXXXXXXXX.monitoring.us-east-1.vpce.amazonaws.com"
self.metrics_endpoint_override = metrics_endpoint_url.replace('https://', '')
# metrics
self.enable_metrics = Utils.get_as_bool(options.enable_metrics, False)
self.metrics_collection_interval: Optional[int] = None
self.metrics_namespace: Optional[str] = None
self.metrics_force_flush_interval: Optional[int] = None
self.aggregation_dimensions: Optional[Union[List[str], List[List[str]]]] = None
self.metrics_collected: Dict[str, Dict[str, Any]] = {}
self.append_dimensions: Optional[Dict] = None
self.include_dimensions: Optional[List[str]] = None
if self.enable_metrics:
if Utils.is_empty(options.metrics):
raise exceptions.invalid_params('metrics is required when enable_metrics = True')
if Utils.is_empty(options.metrics.metrics_collected):
raise exceptions.invalid_params('metrics.metrics_collected is required')
if Utils.is_empty(options.metrics.namespace):
self.metrics_namespace = f'{self.cluster_name}/{self.module_id}'
else:
self.metrics_namespace = options.metrics.namespace
self.metrics_collection_interval = Utils.get_as_int(options.metrics.metrics_collection_interval, self.DEFAULT_METRICS_COLLECTION_INTERVAL)
self.metrics_force_flush_interval = Utils.get_as_int(options.metrics.force_flush_interval, self.DEFAULT_METRICS_FORCE_FLUSH_INTERVAL)
self.aggregation_dimensions = options.metrics.aggregation_dimensions
self.append_dimensions = options.metrics.append_dimensions
self.include_dimensions = options.metrics.include_dimensions
for collected_metric in options.metrics.metrics_collected:
metric_config = {
'enabled': Utils.get_as_bool(collected_metric.enabled, False),
}
if collected_metric.config is not None:
metric_config['config'] = collected_metric.config
self.metrics_collected[collected_metric.name] = metric_config
# logs
self.enable_logs = Utils.get_as_bool(options.enable_logs, False)
self.collected_log_files: Optional[List[Dict]] = None
self.logs_force_flush_interval: Optional[int] = None
self.default_log_stream_name: Optional[str] = None
if self.enable_logs:
if Utils.is_empty(options.logs):
raise exceptions.invalid_params('logs is required when enable_logs = True')
if Utils.is_empty(options.logs.files):
raise exceptions.invalid_params('logs.files is required')
for index, file_config in enumerate(options.logs.files):
if Utils.is_empty(file_config.file_path):
raise exceptions.invalid_params(f'logs.files[{index}].file_path is required')
if Utils.is_empty(file_config.log_group_name):
raise exceptions.invalid_params(f'logs.files[{index}].log_group_name is required')
if Utils.is_empty(file_config.log_stream_name):
raise exceptions.invalid_params(f'logs.files[{index}].log_stream_name is required')
self.default_log_stream_name = Utils.get_as_string(options.logs.default_log_stream_name, f'{self.module_id}_default_{{ip_address}}')
self.logs_force_flush_interval = Utils.get_as_int(options.logs.force_flush_interval, self.DEFAULT_LOGS_FORCE_FLUSH_INTERVAL)
collected_log_files = []
for file_config in options.logs.files:
if file_config.retention_in_days is None:
file_config.retention_in_days = self.DEFAULT_LOGS_RETENTION_IN_DAYS
collected_log_files.append(Utils.to_dict(file_config))
self.collected_log_files = collected_log_files