"""
Connection module for Amazon ELB (Classic) using boto3.
=======================================================
Renamed from ``boto_elb`` to ``boto3_elb`` and rewritten to use the
boto3 ``elb`` client APIs directly via
:py:mod:`saltext.boto3.utils.boto3mod`. The legacy boto2 code path
(object-style access, retry loops) has been removed.
:depends:
- boto3 >= 1.28.0
- botocore >= 1.31.0
:configuration: This module accepts explicit ELB credentials but can
also utilize IAM roles assigned to the instance through Instance Profiles.
Dynamic credentials are then automatically obtained from AWS API and no
further configuration is necessary. More Information available at:
.. code-block:: text
http://docs.aws.amazon.com/AWSEC2/latest/UserGuide/iam-roles-for-amazon-ec2.html
If IAM roles are not used you need to specify them either in the minion's
config file or as a profile. For example, to specify them in the minion's
config file:
.. code-block:: yaml
elb.keyid: GKTADJGHEIQSXMKKRBJ08H
elb.key: askdjghsdfjkghWupUjasdflkdfklgjsdfjajkghs
A region may also be specified in the configuration:
.. code-block:: yaml
elb.region: us-east-1
It's also possible to specify key, keyid and region via a profile, either
as a passed in dict, or as a string to pull from pillars or minion config:
.. code-block:: yaml
myprofile:
keyid: GKTADJGHEIQSXMKKRBJ08H
key: askdjghsdfjkghWupUjasdflkdfklgjsdfjajkghs
region: us-east-1
.. versionadded:: 1.0.0
"""
import logging
import salt.utils.json
from salt.utils import odict
from saltext.boto3.utils import boto3mod
try:
from botocore.exceptions import ClientError
logging.getLogger("boto3").setLevel(logging.CRITICAL)
HAS_BOTO3 = True
except ImportError:
HAS_BOTO3 = False
log = logging.getLogger(__name__)
__virtualname__ = "boto3_elb"
def _get_conn(service, region=None, key=None, keyid=None, profile=None):
"""
Return a boto3 client for ``service`` using this module's dunders.
"""
return boto3mod.get_connection(
service,
opts=__opts__,
context=__context__,
region=region,
key=key,
keyid=keyid,
profile=profile,
)
[docs]
def __virtual__():
"""
Only load if boto3 is available.
"""
if HAS_BOTO3:
return __virtualname__
return (False, "The boto3_elb module could not be loaded: boto3 is not available.")
def _error_code(exc):
try:
return exc.response["Error"]["Code"]
except (AttributeError, KeyError, TypeError):
return ""
[docs]
def exists(name, region=None, key=None, keyid=None, profile=None):
"""
Check to see if an ELB exists.
name (str):
The name of the load balancer to check for existence.
region (str, optional):
The AWS region where the load balancer is located.
key (str, optional):
The AWS secret access key.
keyid (str, optional):
The AWS access key ID.
profile (str, optional):
The profile to use for AWS credentials.
CLI Example:
.. code-block:: bash
salt myminion boto3_elb.exists myelb region=us-east-1
"""
conn = _get_conn("elb", region=region, key=key, keyid=keyid, profile=profile)
try:
result = conn.describe_load_balancers(LoadBalancerNames=[name])
if result.get("LoadBalancerDescriptions"):
return True
log.debug("The load balancer does not exist in region %s", region)
return False
except ClientError as error:
log.warning(error)
return False
[docs]
def get_all_elbs(region=None, key=None, keyid=None, profile=None):
"""
Return all load balancers associated with an account.
region (str, optional):
The AWS region where the load balancers are located.
key (str, optional):
The AWS secret access key.
keyid (str, optional):
The AWS access key ID.
profile (str, optional):
The profile to use for AWS credentials.
CLI Example:
.. code-block:: bash
salt myminion boto3_elb.get_all_elbs region=us-east-1
"""
conn = _get_conn("elb", region=region, key=key, keyid=keyid, profile=profile)
try:
paginator = conn.get_paginator("describe_load_balancers")
result = []
for page in paginator.paginate():
result.extend(page.get("LoadBalancerDescriptions", []))
return result
except ClientError as error:
log.warning(error)
return []
[docs]
def list_elbs(region=None, key=None, keyid=None, profile=None):
"""
Return names of all load balancers associated with an account.
region (str, optional):
The AWS region where the load balancers are located.
key (str, optional):
The AWS secret access key.
keyid (str, optional):
The AWS access key ID.
profile (str, optional):
The profile to use for AWS credentials.
CLI Example:
.. code-block:: bash
salt myminion boto3_elb.list_elbs region=us-east-1
"""
return [
lb["LoadBalancerName"]
for lb in get_all_elbs(region=region, key=key, keyid=keyid, profile=profile)
]
def _tags_for_lb(conn, name):
try:
resp = conn.describe_tags(LoadBalancerNames=[name])
except ClientError as error:
log.warning(error)
return None
for desc in resp.get("TagDescriptions", []):
if desc.get("LoadBalancerName") == name:
return {t["Key"]: t.get("Value") for t in desc.get("Tags", [])}
return None
[docs]
def get_elb_config(name, region=None, key=None, keyid=None, profile=None):
"""
Get an ELB configuration.
name (str):
The name of the load balancer to get the configuration for.
region (str, optional):
The AWS region where the load balancer is located.
key (str, optional):
The AWS secret access key.
keyid (str, optional):
The AWS access key ID.
profile (str, optional):
The profile to use for AWS credentials.
CLI Example:
.. code-block:: bash
salt myminion boto3_elb.get_elb_config myelb region=us-east-1
"""
conn = _get_conn("elb", region=region, key=key, keyid=keyid, profile=profile)
try:
resp = conn.describe_load_balancers(LoadBalancerNames=[name])
except ClientError as error:
log.error("Error fetching config for ELB %s: %s", name, error)
return {}
descs = resp.get("LoadBalancerDescriptions", [])
if not descs:
return {}
lb = descs[0]
ret = {}
ret["availability_zones"] = lb.get("AvailabilityZones", [])
listeners = []
for ld in lb.get("ListenerDescriptions", []):
listener_info = ld.get("Listener", {})
listener_dict = {
"elb_port": listener_info.get("LoadBalancerPort"),
"elb_protocol": listener_info.get("Protocol"),
"instance_port": listener_info.get("InstancePort"),
"instance_protocol": listener_info.get("InstanceProtocol"),
"policies": list(ld.get("PolicyNames", [])),
}
if listener_info.get("SSLCertificateId"):
listener_dict["certificate"] = listener_info["SSLCertificateId"]
listeners.append(listener_dict)
ret["listeners"] = listeners
backends = []
for backend in lb.get("BackendServerDescriptions", []):
backends.append(
{
"instance_port": backend.get("InstancePort"),
"policies": list(backend.get("PolicyNames", [])),
}
)
ret["backends"] = backends
ret["subnets"] = lb.get("Subnets", [])
ret["security_groups"] = lb.get("SecurityGroups", [])
ret["scheme"] = lb.get("Scheme")
ret["dns_name"] = lb.get("DNSName")
ret["tags"] = _tags_for_lb(conn, name)
policies = []
pols = lb.get("Policies", {})
for p in pols.get("AppCookieStickinessPolicies", []):
policies.append(p.get("PolicyName"))
for p in pols.get("LBCookieStickinessPolicies", []):
policies.append(p.get("PolicyName"))
for p in pols.get("OtherPolicies", []):
policies.append(p)
ret["policies"] = policies
ret["canonical_hosted_zone_name"] = lb.get("CanonicalHostedZoneName")
ret["canonical_hosted_zone_name_id"] = lb.get("CanonicalHostedZoneNameID")
ret["vpc_id"] = lb.get("VPCId")
return ret
[docs]
def listener_dict_to_tuple(listener):
"""
Convert an ELB listener dict into a listener tuple used by certain parts of
the AWS ELB API.
listener (dict):
The listener dictionary to convert into a listener tuple.
Example:
.. code-block:: json
{
"elb_port": 80,
"instance_port": 80,
"elb_protocol": "HTTP"
}
CLI Example:
.. code-block:: bash
salt myminion boto3_elb.listener_dict_to_tuple '{"elb_port":80,"instance_port":80,"elb_protocol":"HTTP"}'
"""
if "instance_protocol" not in listener:
instance_protocol = listener["elb_protocol"].upper()
else:
instance_protocol = listener["instance_protocol"].upper()
listener_tuple = [
listener["elb_port"],
listener["instance_port"],
listener["elb_protocol"],
instance_protocol,
]
if "certificate" in listener:
listener_tuple.append(listener["certificate"])
return tuple(listener_tuple)
def _listener_dict_to_api(listener):
api = {
"Protocol": listener["elb_protocol"],
"LoadBalancerPort": listener["elb_port"],
"InstancePort": listener["instance_port"],
"InstanceProtocol": listener.get("instance_protocol", listener["elb_protocol"]).upper(),
}
if listener.get("certificate"):
api["SSLCertificateId"] = listener["certificate"]
return api
[docs]
def create(
name,
availability_zones,
listeners,
subnets=None,
security_groups=None,
scheme="internet-facing",
region=None,
key=None,
keyid=None,
profile=None,
):
"""
Create an ELB.
name (str):
The name of the load balancer to create.
availability_zones (list):
A list of availability zones for the load balancer.
listeners (list):
A list of listener dictionaries for the load balancer.
subnets (list, optional):
A list of subnets for the load balancer.
security_groups (list, optional):
A list of security groups for the load balancer.
scheme (str, optional):
The scheme for the load balancer. Defaults to "internet-facing".
region (str, optional):
The AWS region where the load balancer will be created.
key (str, optional):
The AWS secret access key.
keyid (str, optional):
The AWS access key ID.
profile (str, optional):
The profile to use for AWS credentials.
CLI Example:
.. code-block:: bash
salt myminion boto3_elb.create myelb '["us-east-1a"]' '[{"elb_port": 443, "elb_protocol": "HTTPS"}]'
"""
if exists(name, region, key, keyid, profile):
return True
if isinstance(availability_zones, str):
availability_zones = salt.utils.json.loads(availability_zones)
if isinstance(listeners, str):
listeners = salt.utils.json.loads(listeners)
api_listeners = [_listener_dict_to_api(listener) for listener in listeners]
conn = _get_conn("elb", region=region, key=key, keyid=keyid, profile=profile)
kwargs = {
"LoadBalancerName": name,
"Listeners": api_listeners,
"Scheme": scheme,
}
if availability_zones:
kwargs["AvailabilityZones"] = availability_zones
if subnets:
kwargs["Subnets"] = subnets
if security_groups:
kwargs["SecurityGroups"] = security_groups
try:
lb = conn.create_load_balancer(**kwargs)
except ClientError as error:
log.error(
"Failed to create ELB %s: %s: %s",
name,
_error_code(error),
error.response.get("Error", {}).get("Message", ""),
exc_info_on_loglevel=logging.DEBUG,
)
return False
if lb:
log.info("Created ELB %s", name)
return True
log.error("Failed to create ELB %s", name)
return False
[docs]
def delete(name, region=None, key=None, keyid=None, profile=None):
"""
Delete an ELB.
name (str):
The name of the load balancer to delete.
region (str, optional):
The AWS region where the load balancer is located.
key (str, optional):
The AWS secret access key.
keyid (str, optional):
The AWS access key ID.
profile (str, optional):
The profile to use for AWS credentials.
CLI Example:
.. code-block:: bash
salt myminion boto3_elb.delete myelb region=us-east-1
"""
if not exists(name, region, key, keyid, profile):
return True
conn = _get_conn("elb", region=region, key=key, keyid=keyid, profile=profile)
try:
conn.delete_load_balancer(LoadBalancerName=name)
log.info("Deleted ELB %s.", name)
return True
except ClientError:
log.error("Failed to delete ELB %s", name, exc_info_on_loglevel=logging.DEBUG)
return False
[docs]
def create_listeners(name, listeners, region=None, key=None, keyid=None, profile=None):
"""
Create listeners on an ELB.
name (str):
The name of the load balancer to create listeners on.
listeners (list):
A list of listener dictionaries to create on the load balancer.
region (str, optional):
The AWS region where the load balancer is located.
key (str, optional):
The AWS secret access key.
keyid (str, optional):
The AWS access key ID.
profile (str, optional):
The profile to use for AWS credentials.
CLI Example:
.. code-block:: bash
salt myminion boto3_elb.create_listeners myelb '[{"elb_port":443,"instance_port":80,"elb_protocol":"HTTPS","certificate":"arn:..."}]'
"""
if isinstance(listeners, str):
listeners = salt.utils.json.loads(listeners)
api_listeners = [_listener_dict_to_api(listener) for listener in listeners]
conn = _get_conn("elb", region=region, key=key, keyid=keyid, profile=profile)
try:
conn.create_load_balancer_listeners(LoadBalancerName=name, Listeners=api_listeners)
log.info("Created ELB listeners on %s", name)
return True
except ClientError as error:
log.error(
"Failed to create ELB listeners on %s: %s",
name,
error,
exc_info_on_loglevel=logging.DEBUG,
)
return False
[docs]
def delete_listeners(name, ports, region=None, key=None, keyid=None, profile=None):
"""
Delete listeners on an ELB.
name (str):
The name of the load balancer to delete listeners from.
ports (list):
A list of ports corresponding to the listeners to delete.
region (str, optional):
The AWS region where the load balancer is located.
key (str, optional):
The AWS secret access key.
keyid (str, optional):
The AWS access key ID.
profile (str, optional):
The profile to use for AWS credentials.
CLI Example:
.. code-block:: bash
salt myminion boto3_elb.delete_listeners myelb '[80,443]'
"""
if isinstance(ports, str):
ports = salt.utils.json.loads(ports)
conn = _get_conn("elb", region=region, key=key, keyid=keyid, profile=profile)
try:
conn.delete_load_balancer_listeners(LoadBalancerName=name, LoadBalancerPorts=list(ports))
log.info("Deleted ELB listeners on %s", name)
return True
except ClientError as error:
log.error(
"Failed to delete ELB listeners on %s: %s",
name,
error,
exc_info_on_loglevel=logging.DEBUG,
)
return False
[docs]
def apply_security_groups(name, security_groups, region=None, key=None, keyid=None, profile=None):
"""
Apply security groups to an ELB.
name (str):
The name of the load balancer to apply security groups to.
security_groups (list):
A list of security groups to apply to the load balancer.
region (str, optional):
The AWS region where the load balancer is located.
key (str, optional):
The AWS secret access key.
keyid (str, optional):
The AWS access key ID.
profile (str, optional):
The profile to use for AWS credentials.
CLI Example:
.. code-block:: bash
salt myminion boto3_elb.apply_security_groups myelb '["mysecgroup1"]'
"""
if isinstance(security_groups, str):
security_groups = salt.utils.json.loads(security_groups)
conn = _get_conn("elb", region=region, key=key, keyid=keyid, profile=profile)
try:
conn.apply_security_groups_to_load_balancer(
LoadBalancerName=name, SecurityGroups=list(security_groups)
)
log.info("Applied security_groups on ELB %s", name)
return True
except ClientError as error:
log.error("Failed to apply security_groups on ELB %s: %s", name, error)
return False
[docs]
def enable_availability_zones(
name, availability_zones, region=None, key=None, keyid=None, profile=None
):
"""
Enable availability zones for ELB.
name (str):
The name of the load balancer to enable availability zones for.
availability_zones (list):
A list of availability zones to enable for the load balancer.
region (str, optional):
The AWS region where the load balancer is located.
key (str, optional):
The AWS secret access key.
keyid (str, optional):
The AWS access key ID.
profile (str, optional):
The profile to use for AWS credentials.
CLI Example:
.. code-block:: bash
salt myminion boto3_elb.enable_availability_zones myelb '["us-east-1a"]'
"""
if isinstance(availability_zones, str):
availability_zones = salt.utils.json.loads(availability_zones)
conn = _get_conn("elb", region=region, key=key, keyid=keyid, profile=profile)
try:
conn.enable_availability_zones_for_load_balancer(
LoadBalancerName=name, AvailabilityZones=list(availability_zones)
)
log.info("Enabled availability_zones on ELB %s", name)
return True
except ClientError as error:
log.error("Failed to enable availability_zones on ELB %s: %s", name, error)
return False
[docs]
def disable_availability_zones(
name, availability_zones, region=None, key=None, keyid=None, profile=None
):
"""
Disable availability zones for an ELB.
name (str):
The name of the load balancer to disable availability zones for.
availability_zones (list):
A list of availability zones to disable for the load balancer.
region (str, optional):
The AWS region where the load balancer is located.
key (str, optional):
The AWS secret access key.
keyid (str, optional):
The AWS access key ID.
profile (str, optional):
The profile to use for AWS credentials.
CLI Example:
.. code-block:: bash
salt myminion boto3_elb.disable_availability_zones myelb '["us-east-1a"]'
"""
if isinstance(availability_zones, str):
availability_zones = salt.utils.json.loads(availability_zones)
conn = _get_conn("elb", region=region, key=key, keyid=keyid, profile=profile)
try:
conn.disable_availability_zones_for_load_balancer(
LoadBalancerName=name, AvailabilityZones=list(availability_zones)
)
log.info("Disabled availability_zones on ELB %s", name)
return True
except ClientError as error:
log.error(
"Failed to disable availability_zones on ELB %s: %s",
name,
error,
exc_info_on_loglevel=logging.DEBUG,
)
return False
[docs]
def attach_subnets(name, subnets, region=None, key=None, keyid=None, profile=None):
"""
Attach an ELB to subnets.
name (str):
The name of the load balancer to attach to subnets.
subnets (list):
A list of subnets to attach the load balancer to.
region (str, optional):
The AWS region where the load balancer is located.
key (str, optional):
The AWS secret access key.
keyid (str, optional):
The AWS access key ID.
profile (str, optional):
The profile to use for AWS credentials.
CLI Example:
.. code-block:: bash
salt myminion boto3_elb.attach_subnets myelb '["mysubnet"]'
"""
if isinstance(subnets, str):
subnets = salt.utils.json.loads(subnets)
conn = _get_conn("elb", region=region, key=key, keyid=keyid, profile=profile)
try:
conn.attach_load_balancer_to_subnets(LoadBalancerName=name, Subnets=list(subnets))
log.info("Attached ELB %s on subnets.", name)
return True
except ClientError as error:
log.error(
"Failed to attach ELB %s on subnets: %s",
name,
error,
exc_info_on_loglevel=logging.DEBUG,
)
return False
[docs]
def detach_subnets(name, subnets, region=None, key=None, keyid=None, profile=None):
"""
Detach an ELB from subnets.
name (str):
The name of the load balancer to detach from subnets.
subnets (list):
A list of subnets to detach the load balancer from.
region (str, optional):
The AWS region where the load balancer is located.
key (str, optional):
The AWS secret access key.
keyid (str, optional):
The AWS access key ID.
profile (str, optional):
The profile to use for AWS credentials.
CLI Example:
.. code-block:: bash
salt myminion boto3_elb.detach_subnets myelb '["mysubnet"]'
"""
if isinstance(subnets, str):
subnets = salt.utils.json.loads(subnets)
conn = _get_conn("elb", region=region, key=key, keyid=keyid, profile=profile)
try:
conn.detach_load_balancer_from_subnets(LoadBalancerName=name, Subnets=list(subnets))
log.info("Detached ELB %s from subnets.", name)
return True
except ClientError as error:
log.error(
"Failed to detach ELB %s from subnets: %s",
name,
error,
exc_info_on_loglevel=logging.DEBUG,
)
return False
[docs]
def get_attributes(name, region=None, key=None, keyid=None, profile=None):
"""
Check to see if attributes are set on an ELB.
name (str):
The name of the load balancer to check attributes for.
region (str, optional):
The AWS region where the load balancer is located.
key (str, optional):
The AWS secret access key.
keyid (str, optional):
The AWS access key ID.
profile (str, optional):
The profile to use for AWS credentials.
CLI Example:
.. code-block:: bash
salt myminion boto3_elb.get_attributes myelb
"""
conn = _get_conn("elb", region=region, key=key, keyid=keyid, profile=profile)
try:
resp = conn.describe_load_balancer_attributes(LoadBalancerName=name)
except ClientError as error:
log.error("ELB %s does not exist: %s", name, error)
return {}
attrs = resp.get("LoadBalancerAttributes", {})
al = attrs.get("AccessLog", {})
czlb = attrs.get("CrossZoneLoadBalancing", {})
cd = attrs.get("ConnectionDraining", {})
cs = attrs.get("ConnectionSettings", {})
ret = odict.OrderedDict()
ret["access_log"] = odict.OrderedDict()
ret["cross_zone_load_balancing"] = odict.OrderedDict()
ret["connection_draining"] = odict.OrderedDict()
ret["connecting_settings"] = odict.OrderedDict()
ret["access_log"]["enabled"] = al.get("Enabled", False)
ret["access_log"]["s3_bucket_name"] = al.get("S3BucketName")
ret["access_log"]["s3_bucket_prefix"] = al.get("S3BucketPrefix")
ret["access_log"]["emit_interval"] = al.get("EmitInterval")
ret["cross_zone_load_balancing"]["enabled"] = czlb.get("Enabled", False)
ret["connection_draining"]["enabled"] = cd.get("Enabled", False)
ret["connection_draining"]["timeout"] = cd.get("Timeout")
ret["connecting_settings"]["idle_timeout"] = cs.get("IdleTimeout")
return ret
[docs]
def set_attributes(name, attributes, region=None, key=None, keyid=None, profile=None):
"""
Set attributes on an ELB.
name (str):
The name of the load balancer to set attributes for.
attributes (dict):
A dictionary of attributes to set on the load balancer.
region (str, optional):
The AWS region where the load balancer is located.
key (str, optional):
The AWS secret access key.
keyid (str, optional):
The AWS access key ID.
profile (str, optional):
The profile to use for AWS credentials.
CLI Example:
.. code-block:: bash
salt myminion boto3_elb.set_attributes myelb '{"access_log": {"enabled": true, "s3_bucket_name": "mybucket"}}'
"""
al = attributes.get("access_log", {})
czlb = attributes.get("cross_zone_load_balancing", {})
cd = attributes.get("connection_draining", {})
cs = attributes.get("connecting_settings", {})
if not al and not czlb and not cd and not cs:
log.error("No supported attributes for ELB.")
return False
api_attrs = {}
if al:
if not al.get("enabled", False):
log.error("Access log attribute configured, but enabled config missing")
return False
access_log = {"Enabled": True}
if al.get("s3_bucket_name") is not None:
access_log["S3BucketName"] = al["s3_bucket_name"]
if al.get("s3_bucket_prefix") is not None:
access_log["S3BucketPrefix"] = al["s3_bucket_prefix"]
if al.get("emit_interval") is not None:
access_log["EmitInterval"] = int(al["emit_interval"])
api_attrs["AccessLog"] = access_log
if czlb:
api_attrs["CrossZoneLoadBalancing"] = {"Enabled": bool(czlb["enabled"])}
if cd:
api_attrs["ConnectionDraining"] = {
"Enabled": bool(cd["enabled"]),
"Timeout": int(cd.get("timeout", 300)),
}
if cs:
api_attrs["ConnectionSettings"] = {"IdleTimeout": int(cs.get("idle_timeout", 60))}
conn = _get_conn("elb", region=region, key=key, keyid=keyid, profile=profile)
try:
conn.modify_load_balancer_attributes(
LoadBalancerName=name, LoadBalancerAttributes=api_attrs
)
except ClientError as error:
log.error("Failed to set attributes on ELB %s: %s", name, error)
return False
log.info("Set attributes on ELB %s.", name)
return True
[docs]
def get_health_check(name, region=None, key=None, keyid=None, profile=None):
"""
Get the health check configured for this ELB.
name (str):
The name of the load balancer to get the health check for.
region (str, optional):
The AWS region where the load balancer is located.
key (str, optional):
The AWS secret access key.
keyid (str, optional):
The AWS access key ID.
profile (str, optional):
The profile to use for AWS credentials.
CLI Example:
.. code-block:: bash
salt myminion boto3_elb.get_health_check myelb
"""
conn = _get_conn("elb", region=region, key=key, keyid=keyid, profile=profile)
try:
resp = conn.describe_load_balancers(LoadBalancerNames=[name])
except ClientError:
log.error("ELB %s not found.", name, exc_info_on_loglevel=logging.DEBUG)
return {}
descs = resp.get("LoadBalancerDescriptions", [])
if not descs:
return {}
hc = descs[0].get("HealthCheck", {})
ret = odict.OrderedDict()
ret["interval"] = hc.get("Interval")
ret["target"] = hc.get("Target")
ret["healthy_threshold"] = hc.get("HealthyThreshold")
ret["timeout"] = hc.get("Timeout")
ret["unhealthy_threshold"] = hc.get("UnhealthyThreshold")
return ret
[docs]
def set_health_check(name, health_check, region=None, key=None, keyid=None, profile=None):
"""
Set a health check on an ELB.
name (str):
The name of the load balancer to set the health check for.
health_check (dict):
A dictionary representing the health check configuration.
region (str, optional):
The AWS region where the load balancer is located.
key (str, optional):
The AWS secret access key.
keyid (str, optional):
The AWS access key ID.
profile (str, optional):
The profile to use for AWS credentials.
CLI Example:
.. code-block:: bash
salt myminion boto3_elb.set_health_check myelb '{"target": "HTTP:80/"}'
"""
hc = {}
mapping = {
"interval": "Interval",
"target": "Target",
"healthy_threshold": "HealthyThreshold",
"timeout": "Timeout",
"unhealthy_threshold": "UnhealthyThreshold",
}
for src, dst in mapping.items():
if src in health_check and health_check[src] is not None:
hc[dst] = health_check[src]
conn = _get_conn("elb", region=region, key=key, keyid=keyid, profile=profile)
try:
conn.configure_health_check(LoadBalancerName=name, HealthCheck=hc)
log.info("Configured health check on ELB %s", name)
return True
except ClientError:
log.exception("Failed to configure health check on ELB %s", name)
return False
[docs]
def register_instances(name, instances, region=None, key=None, keyid=None, profile=None):
"""
Register instances with an ELB.
name (str):
The name of the load balancer to register instances with.
instances (list):
A list of instance IDs to register with the load balancer.
region (str, optional):
The AWS region where the load balancer is located.
key (str, optional):
The AWS secret access key.
keyid (str, optional):
The AWS access key ID.
profile (str, optional):
The profile to use for AWS credentials.
CLI Example:
.. code-block:: bash
salt myminion boto3_elb.register_instances myelb instance_id
"""
if isinstance(instances, str):
instances = [instances]
instances = list(instances)
conn = _get_conn("elb", region=region, key=key, keyid=keyid, profile=profile)
try:
resp = conn.register_instances_with_load_balancer(
LoadBalancerName=name,
Instances=[{"InstanceId": i} for i in instances],
)
except ClientError as error:
log.warning(error)
return False
registered_ids = {i.get("InstanceId") for i in resp.get("Instances", [])}
register_failures = set(instances).difference(registered_ids)
if register_failures:
log.warning(
"Instance(s): %s not registered with ELB %s.",
list(register_failures),
name,
)
return False
return True
[docs]
def deregister_instances(name, instances, region=None, key=None, keyid=None, profile=None):
"""
Deregister instances with an ELB.
name (str):
The name of the load balancer to deregister instances from.
instances (list):
A list of instance IDs to deregister from the load balancer.
region (str, optional):
The AWS region where the load balancer is located.
key (str, optional):
The AWS secret access key.
keyid (str, optional):
The AWS access key ID.
profile (str, optional):
The profile to use for AWS credentials.
CLI Example:
.. code-block:: bash
salt myminion boto3_elb.deregister_instances myelb instance_id
"""
if isinstance(instances, str):
instances = [instances]
instances = list(instances)
conn = _get_conn("elb", region=region, key=key, keyid=keyid, profile=profile)
try:
resp = conn.deregister_instances_from_load_balancer(
LoadBalancerName=name,
Instances=[{"InstanceId": i} for i in instances],
)
except ClientError as error:
if _error_code(error) == "InvalidInstance":
log.warning(
"One or more of instance(s) %s are not part of ELB %s. "
"deregister_instances not performed.",
instances,
name,
)
return None
log.warning(error)
return False
remaining_ids = {i.get("InstanceId") for i in resp.get("Instances", [])}
deregister_failures = set(instances).intersection(remaining_ids)
if deregister_failures:
log.warning(
"Instance(s): %s not deregistered from ELB %s.",
list(deregister_failures),
name,
)
return False
return True
[docs]
def set_instances(name, instances, test=False, region=None, key=None, keyid=None, profile=None):
"""
Set the instances assigned to an ELB to exactly the list given.
name (str):
The name of the load balancer to set instances for.
instances (list):
A list of instance IDs to assign to the load balancer.
test (bool, optional):
If True, only test what changes would be made without actually making them.
region (str, optional):
The AWS region where the load balancer is located.
key (str, optional):
The AWS secret access key.
keyid (str, optional):
The AWS access key ID.
profile (str, optional):
The profile to use for AWS credentials.
CLI Example:
.. code-block:: bash
salt myminion boto3_elb.set_instances myelb region=us-east-1 instances="[instance_id,instance_id]"
"""
ret = True
current = {i["instance_id"] for i in get_instance_health(name, region, key, keyid, profile)}
desired = set(instances)
add = desired - current
remove = current - desired
if test:
return bool(add or remove)
if remove:
if deregister_instances(name, list(remove), region, key, keyid, profile) is False:
ret = False
if add:
if register_instances(name, list(add), region, key, keyid, profile) is False:
ret = False
return ret
[docs]
def get_instance_health(name, region=None, key=None, keyid=None, profile=None, instances=None):
"""
Get a list of instances and their health state.
name (str):
The name of the load balancer to get the health state for.
region (str, optional):
The AWS region where the load balancer is located.
key (str, optional):
The AWS secret access key.
keyid (str, optional):
The AWS access key ID.
profile (str, optional):
The profile to use for AWS credentials.
instances (list, optional):
A list of instance IDs to get the health state for. If not provided, all instances are returned.
CLI Example:
.. code-block:: bash
salt myminion boto3_elb.get_instance_health myelb
"""
conn = _get_conn("elb", region=region, key=key, keyid=keyid, profile=profile)
kwargs = {"LoadBalancerName": name}
if instances:
kwargs["Instances"] = [{"InstanceId": i} for i in instances]
try:
resp = conn.describe_instance_health(**kwargs)
except ClientError as error:
log.debug(error)
return []
ret = []
for state in resp.get("InstanceStates", []):
ret.append(
{
"instance_id": state.get("InstanceId"),
"description": state.get("Description"),
"state": state.get("State"),
"reason_code": state.get("ReasonCode"),
}
)
return ret
[docs]
def create_policy(
name,
policy_name,
policy_type,
policy,
region=None,
key=None,
keyid=None,
profile=None,
):
"""
Create an ELB policy.
name (str):
The name of the load balancer to create the policy for.
policy_name (str):
The name of the policy to create.
policy_type (str):
The type of the policy to create.
policy (dict):
A dictionary representing the policy attributes.
region (str, optional):
The AWS region where the load balancer is located.
key (str, optional):
The AWS secret access key.
keyid (str, optional):
The AWS access key ID.
profile (str, optional):
The profile to use for AWS credentials.
CLI Example:
.. code-block:: bash
salt myminion boto3_elb.create_policy myelb mypolicy LBCookieStickinessPolicyType '{"CookieExpirationPeriod": 3600}'
"""
if not exists(name, region, key, keyid, profile):
return False
attributes = []
if isinstance(policy, dict):
for attr_name, attr_value in policy.items():
attributes.append({"AttributeName": attr_name, "AttributeValue": str(attr_value)})
conn = _get_conn("elb", region=region, key=key, keyid=keyid, profile=profile)
try:
conn.create_load_balancer_policy(
LoadBalancerName=name,
PolicyName=policy_name,
PolicyTypeName=policy_type,
PolicyAttributes=attributes,
)
log.info("Created policy %s on ELB %s", policy_name, name)
return True
except ClientError as error:
log.error(
"Failed to create policy %s on ELB %s: %s",
policy_name,
name,
error,
exc_info_on_loglevel=logging.DEBUG,
)
return False
[docs]
def delete_policy(name, policy_name, region=None, key=None, keyid=None, profile=None):
"""
Delete an ELB policy.
name (str):
The name of the load balancer to delete the policy from.
policy_name (str):
The name of the policy to delete.
region (str, optional):
The AWS region where the load balancer is located.
key (str, optional):
The AWS secret access key.
keyid (str, optional):
The AWS access key ID.
profile (str, optional):
The profile to use for AWS credentials.
CLI Example:
.. code-block:: bash
salt myminion boto3_elb.delete_policy myelb mypolicy
"""
if not exists(name, region, key, keyid, profile):
return True
conn = _get_conn("elb", region=region, key=key, keyid=keyid, profile=profile)
try:
conn.delete_load_balancer_policy(LoadBalancerName=name, PolicyName=policy_name)
log.info("Deleted policy %s on ELB %s", policy_name, name)
return True
except ClientError as error:
log.error(
"Failed to delete policy %s on ELB %s: %s",
policy_name,
name,
error,
exc_info_on_loglevel=logging.DEBUG,
)
return False
[docs]
def set_listener_policy(name, port, policies=None, region=None, key=None, keyid=None, profile=None):
"""
Set the policies of an ELB listener.
name (str):
The name of the load balancer to set the listener policies for.
port (int):
The port of the listener to set the policies for.
policies (list, optional):
A list of policy names to set on the listener.
region (str, optional):
The AWS region where the load balancer is located.
key (str, optional):
The AWS secret access key.
keyid (str, optional):
The AWS access key ID.
profile (str, optional):
The profile to use for AWS credentials.
CLI Example:
.. code-block:: bash
salt myminion boto3_elb.set_listener_policy myelb 443 "[policy1,policy2]"
"""
if not exists(name, region, key, keyid, profile):
return True
if policies is None:
policies = []
conn = _get_conn("elb", region=region, key=key, keyid=keyid, profile=profile)
try:
conn.set_load_balancer_policies_of_listener(
LoadBalancerName=name,
LoadBalancerPort=port,
PolicyNames=list(policies),
)
log.info("Set policies %s on ELB %s listener %s", policies, name, port)
except ClientError as error:
log.info(
"Failed to set policy %s on ELB %s listener %s: %s",
policies,
name,
port,
error,
exc_info_on_loglevel=logging.DEBUG,
)
return False
return True
[docs]
def set_backend_policy(name, port, policies=None, region=None, key=None, keyid=None, profile=None):
"""
Set the policies of an ELB backend server.
name (str):
The name of the load balancer to set the backend server policies for.
port (int):
The port of the backend server to set the policies for.
policies (list, optional):
A list of policy names to set on the backend server.
region (str, optional):
The AWS region where the load balancer is located.
key (str, optional):
The AWS secret access key.
keyid (str, optional):
The AWS access key ID.
profile (str, optional):
The profile to use for AWS credentials.
CLI Example:
.. code-block:: bash
salt myminion boto3_elb.set_backend_policy myelb 443 "[policy1,policy2]"
"""
if not exists(name, region, key, keyid, profile):
return True
if policies is None:
policies = []
conn = _get_conn("elb", region=region, key=key, keyid=keyid, profile=profile)
try:
conn.set_load_balancer_policies_for_backend_server(
LoadBalancerName=name,
InstancePort=port,
PolicyNames=list(policies),
)
log.info("Set policies %s on ELB %s backend server %s", policies, name, port)
except ClientError as error:
log.info(
"Failed to set policy %s on ELB %s backend server %s: %s",
policies,
name,
port,
error,
exc_info_on_loglevel=logging.DEBUG,
)
return False
return True