Source code for ecs_composex.rds.rds_features
# SPDX-License-Identifier: MPL-2.0
# Copyright 2020-2022 John Mille <john@compose-x.io>
"""
Module to handle RDS Features definition
"""
from compose_x_common.compose_x_common import keyisset
from troposphere import AWS_NO_VALUE, AWS_STACK_NAME, GetAtt, Ref, Sub
from troposphere.iam import Role as IamRole
from troposphere.rds import DBClusterRole
import ecs_composex.common.troposphere_tools
from ecs_composex.common.logging import LOG
from ecs_composex.iam import define_iam_policy, service_role_trust_policy
from ecs_composex.rds.rds_features_define import (
define_s3_export_feature_policy,
define_s3_import_feature_policy,
)
S3_KEY = "x-s3"
[docs]def validate_rds_features(features_list, db_family):
"""
Function to validate the features listed are supported by the given family.
:param features_list:
:param db_family:
:return:
"""
[docs]def validate_features_content(data):
"""
Function to ensure the data given in compose file is valid for S3/MacroParameters/IamAccess
:param dict allowed_keys:
:param dict data:
:return:
"""
feature_structure = {"Name": (str, None), "Resources": (list, None)}
for feature in data:
for name, rtype in feature_structure.items():
if not keyisset(name, feature):
raise KeyError(f"Features requires {name}. Got", feature.keys())
elif not isinstance(feature[name], rtype[0]):
raise TypeError(
f"Feature property {name} is of type",
type(feature),
"Expected",
rtype[0],
)
[docs]def define_associated_roles(db):
"""
Function to define the AssociatedRoles, either present or empty
:param ecs_composex.rds.rds_stack.RdsDb db:
:return: the list of Associated Roles
:rtype: list
"""
if db.cfn_resource and hasattr(db.cfn_resource, "AssociatedRoles"):
LOG.warning(
"The db properties already had AssociatedRoles defined."
" Only will add ones without the feature already defined"
)
roles = getattr(db.cfn_resource, "AssociatedRoles")
else:
roles = []
return roles
[docs]def add_rds_features(settings, db, db_stack, features, boundary):
"""
Function to add AssociatedRoles and Features if not already defined in the DB properties for that feature.
"""
features_settings = {
"s3Import": define_s3_import_feature_policy,
"s3Export": define_s3_export_feature_policy,
"Lambda": None,
"SageMaker": None,
"Comprehend": None,
}
validate_features_content(features)
roles = define_associated_roles(db)
to_add = [
feature
for feature in features
if feature["Name"] not in [role.FeatureName for role in roles]
]
excluded = [
feature
for feature in features
if feature["Name"] in [role.FeatureName for role in roles]
]
if excluded:
LOG.warning(
f"Features {excluded} are not being processed as already defined in AssociatedRoles of the DB properties"
)
if not to_add:
LOG.warning("No features were found to be added at all!!")
return
policies = []
iam_role = IamRole(
f"{db.logical_name}FeaturesIamRole",
AssumeRolePolicyDocument=service_role_trust_policy("rds"),
Description=Sub(
f"{db.logical_name} RDS Features IAM Role in ${{{AWS_STACK_NAME}}}"
),
Policies=policies,
PermissionsBoundary=boundary,
MaxSessionDuration=3600,
)
features_definition = []
for feature in to_add:
if feature["Name"] not in features_settings.keys() or not (
feature["Name"] in features_settings.keys()
and features_settings[feature["Name"]]
):
LOG.warning(
f"The feature {feature['Name']} is not currently supported. Sorry."
)
policies.append(
features_settings[feature["Name"]](
settings, db, db_stack, feature["Resources"]
)
)
features_definition.append(
DBClusterRole(FeatureName=feature["Name"], RoleArn=GetAtt(iam_role, "Arn"))
)
db_stack.stack_template.add_resource(iam_role)
if policies and not hasattr(db.cfn_resource, "AssociatedRoles"):
setattr(db.cfn_resource, "AssociatedRoles", features_definition)