in application-load-balancer-copy-utility/copy_classic_load_balancer.py [0:0]
def get_alb_data(elb_data, region, load_balancer_name):
if debug:
print("building the Application Load Balancer data structure")
# this is used for building the load balancer spec
alb_data = {'VpcId': elb_data['LoadBalancerDescriptions'][0]['VPCId'], 'Region': region,
'Alb_name': elb_data['LoadBalancerDescriptions'][0]['LoadBalancerName'],
'Subnets': elb_data['LoadBalancerDescriptions'][0]['Subnets'],
'Security_groups': elb_data['LoadBalancerDescriptions'][0]['SecurityGroups'],
'Scheme': elb_data['LoadBalancerDescriptions'][0]['Scheme'],
'Tags': elb_data['TagDescriptions'][0]['Tags'],
'listeners': [],
'target_group_attributes': [],
'target_group_arns': []}
# this is used for building the listeners specs
for elb_listener in elb_data['LoadBalancerDescriptions'][0]['ListenerDescriptions']:
alb_listener = {'Protocol': elb_listener['Listener']['Protocol'],
'Port': elb_listener['Listener']['LoadBalancerPort'],
'TargetGroup_Port': elb_listener['Listener']['InstancePort'],
'TargetGroup_Protocol': elb_listener['Listener']['InstanceProtocol']}
TargetGroup_Attribute = {
'dereg_timeout_seconds_delay': str(
elb_data['LoadBalancerAttributes']['ConnectionDraining']['Timeout']),
'TargetGroup_Port': elb_listener['Listener']['InstancePort'],
}
if alb_listener['Protocol'] == "HTTPS":
alb_listener['Certificates'] = [
{'CertificateArn': elb_listener['Listener']['SSLCertificateId']}]
# check if Classic Load Balancer has any listener policy
if len(elb_listener['PolicyNames']) > 0:
for listener_policy in elb_listener['PolicyNames']:
# If there is a LBCookieStickinessPolicy, append TG attriubtes
if 'LBCookieStickinessPolicy' in listener_policy:
for policy in elb_data['PolicyDescriptions']:
if listener_policy == policy['PolicyName']:
TargetGroup_Attribute['stickiness.enabled'] = 'true'
TargetGroup_Attribute['stickiness.type'] = 'lb_cookie'
TargetGroup_Attribute['stickiness_policy'] = policy['PolicyName'].split('-')[3]
TargetGroup_Attribute['stickiness.lb_cookie.duration_seconds'] = \
policy['PolicyAttributeDescriptions'][0]['AttributeValue']
# If there is a SSLNegotiationPolicy, set Application Load Balancer listener policy
if 'SSLNegotiationPolicy' in listener_policy:
for policy in elb_data['PolicyDescriptions']:
if policy['PolicyName'] == listener_policy:
for policy_attribute_description in policy['PolicyAttributeDescriptions']:
if 'Reference-Security-Policy'in policy_attribute_description['AttributeName']:
alb_listener['SslPolicy']=policy_attribute_description['AttributeValue']
# TGs is not per unique backend port as two TGs might have two
# different stickiness policy
alb_data['listeners'].append(alb_listener)
alb_data['target_group_attributes'].append(TargetGroup_Attribute)
# this is used for building the target groups
'''
# We need to create more target group if ELB front port has Duration-Based sticky policy
# '''
alb_data['target_groups'] = []
hc_target = elb_data['LoadBalancerDescriptions'][
0]['HealthCheck']['Target']
# Append unique stickiness policy name to Target Group Name
alb_target_group_attributes_list = list(alb_data['target_group_attributes'])
for listener in alb_data['listeners']:
target_group = {'HealthCheckTimeoutSeconds': elb_data[
'LoadBalancerDescriptions'][0]['HealthCheck']['Timeout']}
# We only offer 15 seconds minimum health check interval
if elb_data['LoadBalancerDescriptions'][0]['HealthCheck']['Interval'] < 15:
print(
"HealthCheck Interval is less than 15 seconds! Setting it to 15 seconds")
target_group['HealthCheckIntervalSeconds'] = 15
else:
target_group['HealthCheckIntervalSeconds'] = elb_data['LoadBalancerDescriptions'][0]['HealthCheck'][
'Interval']
target_group['HealthyThresholdCount'] = elb_data['LoadBalancerDescriptions'][0]['HealthCheck'][
'HealthyThreshold']
target_group['UnhealthyThresholdCount'] = elb_data['LoadBalancerDescriptions'][0]['HealthCheck'][
'UnhealthyThreshold']
target_group['HealthCheckPath'] = '/' + hc_target.split('/', 1)[1]
target_group['HealthCheckPort'] = hc_target[
hc_target.index(':') + 1: hc_target.index('/')]
target_group['HealthCheckProtocol'] = hc_target.split(':')[0]
target_group['VpcId'] = elb_data[
'LoadBalancerDescriptions'][0]['VPCId']
for target_group_attribute in alb_target_group_attributes_list:
if listener['TargetGroup_Port'] == target_group_attribute['TargetGroup_Port']:
target_group['Port'] = listener['TargetGroup_Port']
target_group['Protocol'] = listener['TargetGroup_Protocol']
random_id = (''.join(choice(ascii_uppercase) for i in range(3)))
if 'stickiness.type' in target_group_attribute:
target_group['Name'] = "{}-tg-stickiness-{}-{}".format(load_balancer_name[: 8],
str(listener['TargetGroup_Port']),random_id)
else:
target_group['Name'] = "{}-tg-{}-{}".format(
load_balancer_name[: 8],str(listener['TargetGroup_Port']),random_id)
alb_target_group_attributes_list.remove(target_group_attribute)
alb_data['target_groups'].append(target_group)
break
# create alb attributes
alb_data['attributes'] = []
attributes = []
attribute = {'Key': 'idle_timeout.timeout_seconds',
'Value': str(elb_data['LoadBalancerAttributes']['ConnectionSettings']['IdleTimeout'])}
attributes.append(attribute)
if elb_data['LoadBalancerAttributes']['AccessLog']['Enabled']:
attribute = {'Key': 'access_logs.s3.enabled',
'Value': str(elb_data['LoadBalancerAttributes']['AccessLog']['Enabled']).lower()}
attributes.append(attribute)
attribute = {'Key': 'access_logs.s3.bucket',
'Value': elb_data['LoadBalancerAttributes']['AccessLog']['S3BucketName']}
attributes.append(attribute)
# we don't specify the prefix key if the prefix is root
if elb_data['LoadBalancerAttributes']['AccessLog']['S3BucketPrefix'] != '':
attribute = {'Key': 'access_logs.s3.prefix',
'Value': elb_data['LoadBalancerAttributes']['AccessLog']['S3BucketPrefix']}
attributes.append(attribute)
alb_data['attributes'] = attributes
alb_data['instanceIds'] = []
for instance in elb_data['LoadBalancerDescriptions'][0]['Instances']:
alb_data['instanceIds'].append(instance['InstanceId'])
if debug:
print("alb_data:")
pprint(alb_data)
return alb_data