Coverage for /pythoncovmergedfiles/medio/medio/usr/local/lib/python3.11/site-packages/c7n/resources/ecs.py: 41%

1# Copyright The Cloud Custodian Authors. 

2# SPDX-License-Identifier: Apache-2.0 

3import copy 

4from botocore.exceptions import ClientError 

5 

6from c7n.actions import AutoTagUser, AutoscalingBase, BaseAction 

7from c7n.exceptions import PolicyExecutionError, PolicyValidationError 

8from c7n.filters import MetricsFilter, ValueFilter, Filter 

9from c7n.filters.costhub import CostHubRecommendation 

10from c7n.filters.offhours import OffHour, OnHour 

11import c7n.filters.vpc as net_filters 

12from c7n.manager import resources 

13from c7n.utils import local_session, chunks, get_retry, type_schema, group_by, jmespath_compile 

14from c7n import query, utils 

15from c7n.query import DescribeSource, ConfigSource 

16from c7n.resources.aws import Arn 

17from c7n.tags import Tag, TagDelayedAction, RemoveTag, TagActionFilter 

18 

19 

20def ecs_tag_normalize(resources): 

21 """normalize tag format on ecs resources to match common aws format.""" 

22 for r in resources: 

23 if 'tags' in r: 

24 r['Tags'] = [{'Key': t['key'], 'Value': t['value']} for t in r['tags']] 

25 r.pop('tags') 

26 

27 

28NEW_ARN_STYLE = ('container-instance', 'service', 'task') 

29 

30 

31def ecs_taggable(model, r): 

32 # Tag support requires new arn format 

33 # https://docs.aws.amazon.com/AmazonECS/latest/developerguide/ecs-using-tags.html 

34 # 

35 # New arn format details 

36 # https://docs.aws.amazon.com/AmazonECS/latest/developerguide/ecs-resource-ids.html 

37 # 

38 path_parts = r[model.id].rsplit(':', 1)[-1].split('/') 

39 if path_parts[0] not in NEW_ARN_STYLE: 

40 return True 

41 return len(path_parts) > 2 

42 

43 

44class ContainerConfigSource(ConfigSource): 

45 

46 preserve_empty = () 

47 preserve_case = {'Tags'} 

48 mapped_keys = {} 

49 

50 @classmethod 

51 def remap_keys(cls, resource): 

52 for k, v in cls.mapped_keys.items(): 

53 if v in resource: 

54 continue 

55 if k not in resource: 

56 continue 

57 resource[v] = resource.pop(k) 

58 return resource 

59 

60 @classmethod 

61 def lower_keys(cls, data): 

62 if isinstance(data, dict): 

63 for k, v in list(data.items()): 

64 if k in cls.preserve_case: 

65 continue 

66 lk = k[0].lower() + k[1:] 

67 data[lk] = data.pop(k) 

68 # describe doesn't return empty list/dict by default 

69 if isinstance(v, (list, dict)) and not v and lk not in cls.preserve_empty: 

70 data.pop(lk) 

71 elif isinstance(v, (dict, list)): 

72 data[lk] = cls.lower_keys(v) 

73 elif isinstance(data, list): 

74 return list(map(cls.lower_keys, data)) 

75 return data 

76 

77 def load_resource(self, item): 

78 resource = self.lower_keys(super().load_resource(item)) 

79 if self.mapped_keys: 

80 return self.remap_keys(resource) 

81 return resource 

82 

83 

84class ClusterDescribe(query.DescribeSource): 

85 

86 def augment(self, resources): 

87 resources = super(ClusterDescribe, self).augment(resources) 

88 ecs_tag_normalize(resources) 

89 return resources 

90 

91 

92@resources.register('ecs') 

93class ECSCluster(query.QueryResourceManager): 

94 

95 class resource_type(query.TypeInfo): 

96 service = 'ecs' 

97 enum_spec = ('list_clusters', 'clusterArns', None) 

98 batch_detail_spec = ( 

99 'describe_clusters', 'clusters', None, 'clusters', { 

100 'include': ['TAGS', 'SETTINGS', 'CONFIGURATIONS'] 

101 }) 

102 name = "clusterName" 

103 arn = id = "clusterArn" 

104 arn_type = 'cluster' 

105 config_type = cfn_type = 'AWS::ECS::Cluster' 

106 

107 source_mapping = { 

108 'describe': ClusterDescribe, 

109 'config': query.ConfigSource 

110 } 

111 

112 

113@ECSCluster.filter_registry.register('metrics') 

114class ECSMetrics(MetricsFilter): 

115 

116 def get_dimensions(self, resource): 

117 return [{'Name': 'ClusterName', 'Value': resource['clusterName']}] 

118 

119 

120class ECSClusterResourceDescribeSource(query.ChildDescribeSource): 

121 

122 # We need an additional subclass of describe for ecs cluster. 

123 # 

124 # - Default child query just returns the child resources from 

125 # enumeration op, for ecs clusters, enumeration just returns 

126 # resources ids, we also need to retain the parent id for 

127 # augmentation. 

128 # 

129 # - The default augmentation detail_spec/batch_detail_spec need additional 

130 # handling for the string resources with parent id. 

131 # 

132 

133 def __init__(self, manager): 

134 self.manager = manager 

135 self.query = query.ChildResourceQuery( 

136 self.manager.session_factory, self.manager, capture_parent_id=True) 

137 

138 def get_resources(self, ids, cache=True): 

139 """Retrieve ecs resources for serverless policies or related resources 

140 

141 Requires arns in new format. 

142 https://docs.aws.amazon.com/AmazonECS/latest/userguide/ecs-resource-ids.html 

143 """ 

144 cluster_resources = {} 

145 for i in ids: 

146 _, ident = i.rsplit(':', 1) 

147 parts = ident.split('/', 2) 

148 if len(parts) != 3: 

149 raise PolicyExecutionError("New format ecs arn required") 

150 cluster_resources.setdefault(parts[1], []).append(parts[2]) 

151 

152 results = [] 

153 client = local_session(self.manager.session_factory).client('ecs') 

154 for cid, resource_ids in cluster_resources.items(): 

155 results.extend( 

156 self.process_cluster_resources(client, cid, resource_ids)) 

157 return results 

158 

159 def augment(self, resources): 

160 parent_child_map = {} 

161 for pid, r in resources: 

162 parent_child_map.setdefault(pid, []).append(r) 

163 results = [] 

164 with self.manager.executor_factory( 

165 max_workers=self.manager.max_workers) as w: 

166 client = local_session(self.manager.session_factory).client('ecs') 

167 futures = {} 

168 for pid, services in parent_child_map.items(): 

169 futures[ 

170 w.submit( 

171 self.process_cluster_resources, client, pid, services) 

172 ] = (pid, services) 

173 for f in futures: 

174 pid, services = futures[f] 

175 if f.exception(): 

176 self.manager.log.warning( 

177 'error fetching ecs resources for cluster %s: %s', 

178 pid, f.exception()) 

179 continue 

180 results.extend(f.result()) 

181 return results 

182 

183 

184@query.sources.register('describe-ecs-service') 

185class ECSServiceDescribeSource(ECSClusterResourceDescribeSource): 

186 

187 def process_cluster_resources(self, client, cluster_id, services): 

188 results = [] 

189 for service_set in chunks(services, self.manager.chunk_size): 

190 results.extend( 

191 client.describe_services( 

192 cluster=cluster_id, 

193 include=['TAGS'], 

194 services=service_set).get('services', [])) 

195 ecs_tag_normalize(results) 

196 return results 

197 

198 

199class ECSServiceConfigSource(ContainerConfigSource): 

200 perserve_empty = { 

201 'placementConstraints', 'placementStrategy', 

202 'serviceRegistries', 'Tags', 'loadBalancers'} 

203 

204 mapped_keys = { 

205 'role': 'roleArn', 'cluster': 'clusterArn'} 

206 

207 

208@resources.register('ecs-service') 

209class Service(query.ChildResourceManager): 

210 

211 chunk_size = 10 

212 

213 class resource_type(query.TypeInfo): 

214 service = 'ecs' 

215 name = 'serviceName' 

216 arn = id = 'serviceArn' 

217 enum_spec = ('list_services', 'serviceArns', None) 

218 parent_spec = ('ecs', 'cluster', None) 

219 supports_trailevents = True 

220 config_type = cfn_type = 'AWS::ECS::Service' 

221 

222 source_mapping = { 

223 'config': ECSServiceConfigSource, 

224 'describe-child': ECSServiceDescribeSource, 

225 'describe': ECSServiceDescribeSource, 

226 } 

227 

228 def get_resources(self, ids, cache=True, augment=True): 

229 return super(Service, self).get_resources(ids, cache, augment=False) 

230 

231 

232@Service.filter_registry.register('metrics') 

233class ServiceMetrics(MetricsFilter): 

234 

235 def get_dimensions(self, resource): 

236 return [ 

237 {'Name': 'ClusterName', 'Value': resource['clusterArn'].rsplit('/')[-1]}, 

238 {'Name': 'ServiceName', 'Value': resource['serviceName']}] 

239 

240 

241class RelatedTaskDefinitionFilter(ValueFilter): 

242 

243 schema = type_schema('task-definition', rinherit=ValueFilter.schema) 

244 schema_alias = False 

245 permissions = ('ecs:DescribeTaskDefinition', 

246 'ecs:ListTaskDefinitions') 

247 related_key = 'taskDefinition' 

248 

249 def process(self, resources, event=None): 

250 self.task_defs = {t['taskDefinitionArn']: t for t in self.get_task_defs(resources)} 

251 return super(RelatedTaskDefinitionFilter, self).process(resources) 

252 

253 def get_task_defs(self, resources): 

254 task_def_ids = list({s[self.related_key] for s in resources}) 

255 task_def_manager = self.manager.get_resource_manager( 

256 'ecs-task-definition') 

257 

258 # due to model difference (multi-level containment with 

259 # multi-step resource iteration) and potential volume of 

260 # resources, we break our abstractions a little in the name of 

261 # efficiency wrt api usage. 

262 

263 # check to see if task def cache is already populated 

264 key = task_def_manager.get_cache_key(None) 

265 if self.manager._cache.get(key): 

266 task_defs = task_def_manager.get_resources(task_def_ids) 

267 # else just augment the ids 

268 else: 

269 task_defs = task_def_manager.augment(task_def_ids) 

270 return task_defs 

271 

272 def __call__(self, i): 

273 task = self.task_defs[i[self.related_key]] 

274 return self.match(task) 

275 

276 

277@Service.filter_registry.register('task-definition') 

278class ServiceTaskDefinitionFilter(RelatedTaskDefinitionFilter): 

279 """Filter services by their task definitions. 

280 

281 :Example: 

282 

283 Find any fargate services that are running with a particular 

284 image in the task and stop them. 

285 

286 .. code-block:: yaml 

287 

288 policies: 

289 - name: fargate-find-stop-image 

290 resource: ecs-task 

291 filters: 

292 - launchType: FARGATE 

293 - type: task-definition 

294 key: "containerDefinitions[].image" 

295 value: "elasticsearch/elasticsearch:6.4.3" 

296 value_type: swap 

297 op: contains 

298 actions: 

299 - type: stop 

300 """ 

301 

302 

303@ECSCluster.filter_registry.register('ebs-storage') 

304class Storage(ValueFilter): 

305 """Filter clusters by configured EBS storage parameters. 

306 

307 :Example: 

308 

309 Find any ECS clusters that have instances that are using unencrypted EBS volumes. 

310 

311 .. code-block:: yaml 

312 

313 policies: 

314 - name: encrypted-ebs-volumes 

315 resource: ecs 

316 filters: 

317 - type: ebs-storage 

318 key: Encrypted 

319 value: true 

320 """ 

321 

322 schema = type_schema( 

323 'ebs-storage', rinherit=ValueFilter.schema, 

324 operator={'type': 'string', 'enum': ['or', 'and']}, 

325 ) 

326 schema_alias = False 

327 

328 def get_permissions(self): 

329 return (self.manager.get_resource_manager('ebs').get_permissions() + 

330 self.manager.get_resource_manager('ec2').get_permissions() + 

331 self.manager.get_resource_manager('ecs').get_permissions() 

332 ) 

333 

334 def process(self, resources, event=None): 

335 self.storage = self.get_storage(resources) 

336 self.skip = [] 

337 self.operator = self.data.get( 

338 'operator', 'or') == 'or' and any or all 

339 return list(filter(self, resources)) 

340 

341 def get_storage(self, resources): 

342 manager = self.manager.get_resource_manager('ecs-container-instance') 

343 

344 storage = {} 

345 for cluster_set in utils.chunks(resources, 200): 

346 for cluster in cluster_set: 

347 cluster["clusterArn"] 

348 instances = manager.resources({ 

349 "cluster": cluster['clusterArn'], 

350 }, augment=False) 

351 instances = manager.get_resources(instances, augment=False) 

352 storage[cluster["clusterArn"]] = [] 

353 

354 for instance in instances: 

355 storage[cluster["clusterArn"]].extend(self.get_ebs_volumes([instance["ec2InstanceId"]])) 

356 

357 return storage 

358 

359 def get_ebs_volumes(self, resources): 

360 volumes = [] 

361 ec2_manager = self.manager.get_resource_manager('ec2') 

362 ebs_manager = self.manager.get_resource_manager('ebs') 

363 for instance_set in utils.chunks(resources, 200): 

364 instance_set = ec2_manager.get_resources(instance_set) 

365 volume_ids = [] 

366 for i in instance_set: 

367 for bd in i.get('BlockDeviceMappings', ()): 

368 if 'Ebs' not in bd: 

369 continue 

370 volume_ids.append(bd['Ebs']['VolumeId']) 

371 for v in ebs_manager.get_resources(volume_ids): 

372 if not v['Attachments']: 

373 continue 

374 volumes.append(v) 

375 return volumes 

376 

377 def __call__(self, i): 

378 storage = self.storage.get(i["clusterArn"]) 

379 

380 if not storage: 

381 return False 

382 return self.operator(map(self.match, storage)) 

383 

384 

385@Service.filter_registry.register('subnet') 

386class SubnetFilter(net_filters.SubnetFilter): 

387 

388 RelatedIdsExpression = "" 

389 expressions = ('taskSets[].networkConfiguration.awsvpcConfiguration.subnets[]', 

390 'deployments[].networkConfiguration.awsvpcConfiguration.subnets[]', 

391 'networkConfiguration.awsvpcConfiguration.subnets[]') 

392 

393 def get_related_ids(self, resources): 

394 subnet_ids = set() 

395 for exp in self.expressions: 

396 cexp = jmespath_compile(exp) 

397 for r in resources: 

398 ids = cexp.search(r) 

399 if ids: 

400 subnet_ids.update(ids) 

401 return list(subnet_ids) 

402 

403 

404@Service.filter_registry.register('security-group') 

405class SGFilter(net_filters.SecurityGroupFilter): 

406 

407 RelatedIdsExpression = "" 

408 expressions = ('taskSets[].networkConfiguration.awsvpcConfiguration.securityGroups[]', 

409 'deployments[].networkConfiguration.awsvpcConfiguration.securityGroups[]', 

410 'networkConfiguration.awsvpcConfiguration.securityGroups[]') 

411 

412 def get_related_ids(self, resources): 

413 sg_ids = set() 

414 for exp in self.expressions: 

415 cexp = jmespath_compile(exp) 

416 for r in resources: 

417 ids = cexp.search(r) 

418 if ids: 

419 sg_ids.update(ids) 

420 return list(sg_ids) 

421 

422 

423@Service.filter_registry.register('network-location', net_filters.NetworkLocation) 

424@Service.action_registry.register('modify-definition') 

425class UpdateTemplate(BaseAction): 

426 

427 schema = type_schema( 

428 'modify-definition', 

429 properties={'type': 'object'}, 

430 ) 

431 

432 permissions = ("ecs:RegisterTaskDefinition", "ecs:UpdateService") 

433 

434 def validate(self): 

435 if self.data.get('properties'): 

436 return 

437 found = False 

438 for f in self.manager.iter_filters(): 

439 if isinstance(f, CostHubRecommendation): 

440 found = True 

441 if not found: 

442 raise PolicyValidationError( 

443 "modify-definition: either properties specified or am optimization filter used" 

444 ) 

445 

446 def process(self, resources): 

447 task_def_filter = ServiceTaskDefinitionFilter({}, self.manager) 

448 task_defs = {t['taskDefinitionArn']: t for t in task_def_filter.get_task_defs(resources)} 

449 client = local_session(self.manager.session_factory).client('ecs') 

450 

451 # we can only modify task definition when ecs is controlling the deployment. 

452 resources = self.filter_resources(resources, "deploymentController.type", ("ECS",)) 

453 

454 nack = 0 

455 for r in resources: 

456 r_task_def = task_defs[r[task_def_filter.related_key]] 

457 m_task_def = self.get_target_task_def(r, r_task_def) 

458 if m_task_def is None: 

459 nack += 1 

460 continue 

461 response = client.register_task_definition(**m_task_def) 

462 task_arn = response['taskDefinition']['taskDefinitionArn'] 

463 cluster, _ = Arn.parse(r['serviceArn']).resource.split('/', 1) 

464 client.update_service( 

465 cluster=cluster, 

466 service=r['serviceName'], 

467 taskDefinition=task_arn 

468 ) 

469 if nack: 

470 self.log.warning("modify-definition %d services not modified", nack) 

471 

472 task_def_normalized = [ 

473 "taskDefinitionArn", "revision", "status", 

474 "registeredAt", "registeredBy", "requiresAttributes", 

475 ] 

476 

477 task_def_remap = { 

478 "compatibilities": "requiresCompatibilities", 

479 } 

480 

481 def get_target_task_def(self, resource, current_task_def): 

482 target_task_def = copy.deepcopy(current_task_def) 

483 target_task_def.update(self.data.get('properties', {})) 

484 cost_optimization = resource.get(CostHubRecommendation.annotation_key) 

485 

486 if cost_optimization and cost_optimization['actionType'] == 'Rightsize': 

487 cpu, mem = cost_optimization["recommendedResourceSummary"].split("/") 

488 cpu = int(float(cpu.split(" ")[0])) 

489 mem = int(mem.split(" ")[0]) 

490 target_task_def["cpu"] = str(cpu) 

491 target_task_def["memory"] = str(mem) 

492 target_task_def = self.update_target_containers_size(current_task_def, target_task_def) 

493 

494 if target_task_def == current_task_def or target_task_def is None: 

495 return 

496 

497 # normalize from describe to register formats 

498 for k in self.task_def_normalized: 

499 target_task_def.pop(k, None) 

500 for ck, dk in self.task_def_remap.items(): 

501 if ck in target_task_def: 

502 target_task_def[dk] = target_task_def.pop(ck) 

503 tags = [] 

504 for t in target_task_def.pop('Tags', []): 

505 tags.append({'key': t['Key'], 'value': t['Value']}) 

506 if tags: 

507 target_task_def['tags'] = tags 

508 return target_task_def 

509 

510 def update_target_containers_size(self, current_task_def, target_task_def): 

511 """Update container memory/size targets. 

512 

513 We need to update memory/cpu requirements of the containers within 

514 the task, so the total of the containers in the task def 

515 matches the definition. 

516 

517 for a task w/ a single container this is simple, make the container match 

518 the definition. 

519 

520 for a multi container task, we need select a heuristic 

521 (proportional, largest) with some notion of a floor / min for 

522 proportional. for now we punt on multi-container tasks. 

523 """ 

524 if len(target_task_def['containerDefinitions']) > 1: 

525 return 

526 container_def = target_task_def['containerDefinitions'][0] 

527 container_def['memory'] = int(target_task_def['memory']) 

528 container_def['cpu'] = int(target_task_def['cpu']) 

529 return target_task_def 

530 

531 

532@Service.action_registry.register('modify') 

533class UpdateService(BaseAction): 

534 """Action to update service 

535 

536 :example: 

537 

538 .. code-block:: yaml 

539 

540 policies: 

541 - name: no-public-ips-services 

542 resource: ecs-service 

543 filters: 

544 - 'networkConfiguration.awsvpcConfiguration.assignPublicIp': 'ENABLED' 

545 actions: 

546 - type: modify 

547 update: 

548 networkConfiguration: 

549 awsvpcConfiguration: 

550 assignPublicIp: DISABLED 

551 """ 

552 

553 schema = type_schema('modify', 

554 update={ 

555 'desiredCount': {'type': 'integer'}, 

556 'taskDefinition': {'type': 'string'}, 

557 'deploymentConfiguration': { 

558 'type': 'object', 

559 'properties': { 

560 'maximumPercent': {'type': 'integer'}, 

561 'minimumHealthyPercent': {'type': 'integer'}, 

562 } 

563 }, 

564 'networkConfiguration': { 

565 'type': 'object', 

566 'properties': { 

567 'awsvpcConfiguration': { 

568 'type': 'object', 

569 'properties': { 

570 'subnets': { 

571 'type': 'array', 

572 'items': { 

573 'type': 'string', 

574 }, 

575 'minItems': 1 

576 }, 

577 'securityGroups': { 

578 'items': { 

579 'type': 'string', 

580 }, 

581 }, 

582 'assignPublicIp': { 

583 'type': 'string', 

584 'enum': ['ENABLED', 'DISABLED'], 

585 } 

586 } 

587 } 

588 } 

589 }, 

590 'platformVersion': {'type': 'string'}, 

591 'forceNewDeployment': {'type': 'boolean', 'default': False}, 

592 'healthCheckGracePeriodSeconds': {'type': 'integer'}, 

593 } 

594 ) 

595 

596 permissions = ('ecs:UpdateService',) 

597 

598 def process(self, resources): 

599 client = local_session(self.manager.session_factory).client('ecs') 

600 update = self.data.get('update') 

601 

602 for r in resources: 

603 param = {} 

604 

605 # Handle network separately as it requires atomic updating, and populating 

606 # defaults from the resource. 

607 net_update = update.get('networkConfiguration', {}).get('awsvpcConfiguration') 

608 if net_update: 

609 net_param = dict(r['networkConfiguration']['awsvpcConfiguration']) 

610 param['networkConfiguration'] = {'awsvpcConfiguration': net_param} 

611 for k, v in net_update.items(): 

612 net_param[k] = v 

613 

614 for k, v in update.items(): 

615 if k == 'networkConfiguration': 

616 continue 

617 elif r.get(k) != v: 

618 param[k] = v 

619 

620 if not param: 

621 continue 

622 

623 client.update_service( 

624 cluster=r['clusterArn'], service=r['serviceName'], **param) 

625 

626 

627@Service.action_registry.register('delete') 

628class DeleteService(BaseAction): 

629 """Delete service(s).""" 

630 

631 schema = type_schema('delete') 

632 permissions = ('ecs:DeleteService',) 

633 

634 def process(self, resources): 

635 client = local_session(self.manager.session_factory).client('ecs') 

636 retry = get_retry(('Throttling',)) 

637 for r in resources: 

638 try: 

639 desiredCount = 0 

640 

641 # Two different types of responses: 

642 # Deployments would appear for normal services 

643 # TaskSets would show for Blue/Green deployment 

644 if 'deployments' in r: 

645 primary = [d for d in r['deployments'] if d['status'] == 'PRIMARY'].pop() 

646 desiredCount = primary.get('desiredCount', 0) 

647 elif 'taskSets' in r: 

648 primary = [t for t in r['taskSets'] if t['status'] == 'PRIMARY'].pop() 

649 desiredCount = primary.get('computedDesiredCount', 0) 

650 

651 if desiredCount > 0: 

652 retry(client.update_service, 

653 cluster=r['clusterArn'], service=r['serviceName'], desiredCount=0) 

654 

655 retry(client.delete_service, 

656 cluster=r['clusterArn'], service=r['serviceName']) 

657 except ClientError as e: 

658 if e.response['Error']['Code'] != 'ServiceNotFoundException': 

659 raise 

660 

661 

662@query.sources.register('describe-ecs-task') 

663class ECSTaskDescribeSource(ECSClusterResourceDescribeSource): 

664 

665 def process_cluster_resources(self, client, cluster_id, tasks): 

666 results = [] 

667 for task_set in chunks(tasks, self.manager.chunk_size): 

668 results.extend( 

669 self.manager.retry( 

670 client.describe_tasks, 

671 cluster=cluster_id, 

672 include=['TAGS'], 

673 tasks=task_set).get('tasks', [])) 

674 ecs_tag_normalize(results) 

675 return results 

676 

677 

678@resources.register('ecs-task') 

679class Task(query.ChildResourceManager): 

680 

681 chunk_size = 100 

682 

683 class resource_type(query.TypeInfo): 

684 service = 'ecs' 

685 arn = id = name = 'taskArn' 

686 arn_type = 'task' 

687 enum_spec = ('list_tasks', 'taskArns', None) 

688 parent_spec = ('ecs', 'cluster', None) 

689 supports_trailevents = True 

690 cfn_type = 'AWS::ECS::TaskSet' 

691 

692 @property 

693 def source_type(self): 

694 source = self.data.get('source', 'describe') 

695 if source in ('describe', 'describe-child'): 

696 source = 'describe-ecs-task' 

697 return source 

698 

699 def get_resources(self, ids, cache=True, augment=True): 

700 return super(Task, self).get_resources(ids, cache, augment=False) 

701 

702 

703@Task.filter_registry.register('subnet') 

704class TaskSubnetFilter(net_filters.SubnetFilter): 

705 

706 RelatedIdsExpression = "attachments[].details[?name == 'subnetId'].value[]" 

707 

708 

709@Task.filter_registry.register('security-group') 

710class TaskSGFilter(net_filters.SecurityGroupFilter): 

711 

712 ecs_group_cache = None 

713 

714 RelatedIdsExpression = "" 

715 eni_expression = "attachments[].details[?name == 'networkInterfaceId'].value[]" 

716 sg_expression = "Groups[].GroupId[]" 

717 

718 def _get_related_ids(self, resources): 

719 groups = dict() 

720 eni_ids = set() 

721 

722 cexp = jmespath_compile(self.eni_expression) 

723 for r in resources: 

724 ids = cexp.search(r) 

725 if ids: 

726 eni_ids.update(ids) 

727 

728 if eni_ids: 

729 client = local_session(self.manager.session_factory).client('ec2') 

730 response = client.describe_network_interfaces( 

731 NetworkInterfaceIds=list(eni_ids) 

732 ) 

733 if response["NetworkInterfaces"]: 

734 cexp = jmespath_compile(self.sg_expression) 

735 for r in response["NetworkInterfaces"]: 

736 ids = cexp.search(r) 

737 if ids: 

738 groups[r["NetworkInterfaceId"]] = ids 

739 self.ecs_group_cache = groups 

740 

741 return groups 

742 

743 def get_related_ids(self, resources): 

744 if not self.ecs_group_cache: 

745 self.ecs_group_cache = self._get_related_ids(resources) 

746 

747 group_ids = set() 

748 cexp = jmespath_compile(self.eni_expression) 

749 for r in resources: 

750 ids = cexp.search(r) 

751 for group_id in ids: 

752 group_ids.update(self.ecs_group_cache.get(group_id, ())) 

753 return list(group_ids) 

754 

755 

756@Task.filter_registry.register('network-location', net_filters.NetworkLocation) 

757@Task.filter_registry.register('task-definition') 

758class TaskTaskDefinitionFilter(RelatedTaskDefinitionFilter): 

759 """Filter tasks by their task definition. 

760 

761 :Example: 

762 

763 Find any fargate tasks that are running without read only root 

764 and stop them. 

765 

766 .. code-block:: yaml 

767 

768 policies: 

769 - name: fargate-readonly-tasks 

770 resource: ecs-task 

771 filters: 

772 - launchType: FARGATE 

773 - type: task-definition 

774 key: "containerDefinitions[].readonlyRootFilesystem" 

775 value: None 

776 value_type: swap 

777 op: contains 

778 actions: 

779 - type: stop 

780 

781 """ 

782 related_key = 'taskDefinitionArn' 

783 

784 

785@Task.action_registry.register('stop') 

786class StopTask(BaseAction): 

787 """Stop/Delete a currently running task.""" 

788 

789 schema = type_schema('stop', reason={"type": "string"}) 

790 permissions = ('ecs:StopTask',) 

791 

792 def process(self, resources): 

793 client = local_session(self.manager.session_factory).client('ecs') 

794 retry = get_retry(('Throttling',)) 

795 reason = self.data.get('reason', 'custodian policy') 

796 

797 for r in resources: 

798 try: 

799 retry(client.stop_task, 

800 cluster=r['clusterArn'], 

801 task=r['taskArn'], 

802 reason=reason) 

803 except ClientError as e: 

804 # No error code for not found. 

805 if e.response['Error']['Message'] != "The referenced task was not found.": 

806 raise 

807 

808 

809class DescribeTaskDefinition(DescribeSource): 

810 

811 def get_resources(self, ids, cache=True): 

812 if cache: 

813 resources = self.manager._get_cached_resources(ids)