Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
:type yml: dict
"""
self.serviceName = yml['name']
self.clusterName = yml['cluster']
if 'launch_type' in yml:
self.launchType = yml['launch_type']
self.environment = yml.get('environment', 'undefined')
self.family = yml['family']
# backwards compatibility for deployfish.yml < 0.16.0
if 'maximum_percent' in yml:
self.maximumPercent = yml['maximum_percent']
self.minimumHealthyPercent = yml['minimum_healthy_percent']
self.asg = ASG(yml=yml)
if 'application_scaling' in yml:
# Application Autoscaling
self.scaling = ApplicationAutoscaling(yml['name'], yml['cluster'], yml=yml['application_scaling'])
if 'load_balancer' in yml:
if 'service_role_arn' in yml:
# backwards compatibility for deployfish.yml < 0.3.6
self.roleArn = yml['service_role_arn']
else:
self.roleArn = yml['load_balancer']['service_role_arn']
if 'target_groups' in yml['load_balancer']:
# If we want the service to register itself with multiple target groups,
# the "load_balancer" section will have a list entry named "target_groups".
# Each item in the target_group_list will be a dict with keys "target_group_arn",
# "container_name" and "container_port"
self.set_alb(yml['load_balancer']['target_groups'])
else:
# We either have just one target group, or we're using an ELB
if 'load_balancer_name' in yml['load_balancer']:
# ELB
def from_aws(self):
"""
Update our service definition, task definition and tasks from the live
versions in AWS.
"""
self.__aws_service = self.__get_service()
if not self.scaling:
# This only gets executed if we don't have an "application_scaling"
# section in our service YAML definition.
#
# But we're looking here for an autoscaling setup that we previously
# had created but which we no longer want
self.scaling = ApplicationAutoscaling(self.serviceName, self.clusterName)
if not self.scaling.exists():
self.scaling = None
if self.__aws_service:
self.active_task_definition = TaskDefinition(self.taskDefinition)
# If we have helper tasks, update them from AWS now
helpers = self.active_task_definition.get_helper_tasks()
if helpers:
for t in self.tasks.values():
t.from_aws(helpers[t.family])
if self.__aws_service['serviceRegistries']:
self.serviceDiscovery = ServiceDiscovery(self.service_discovery[0]['registryArn'])
else:
self.serviceDiscovery = None
else:
self.active_task_definition = None
def needs_update(self):
if self == ApplicationAutoscaling(self.serviceName, self.clusterName, aws=self.__aws_scalable_target):
for policy in self.policies.keys():
if self.policies[policy].needs_update():
return True
return False
return True
def __setattr__(self, attr, value):
if attr in ['MinCapacity', 'MaxCapacity', 'RoleARN']:
setattr(self, "_" + attr, value)
else:
super(ApplicationAutoscaling, self).__setattr__(attr, value)
def update(self):
if self != ApplicationAutoscaling(self.serviceName, self.clusterName, aws=self.__aws_scalable_target):
# the scalable target itself needs updating
self.delete()
self.create()
else:
# the scalable target itself doesn't need updating but maybe
# the scaling policies do
for policy in self.policies.keys():
self.policies[policy].update()