1. Introdução
Um recurso ocioso continua gerando custo sem entregar valor, por exemplo, o disco de uma instância já excluída, o IP elástico que foi provisionado para associar a alguma instância, mas que não está em uso e ninguém desassociou, o Load Balancer de um ambiente de teste encerrado, etc. Recursos ociosos em excesso geram um alto custo na cloud.
Este artigo mostra como criar 4 funções Lambda em Python, uma para cada tipo de recurso ocioso, que enviam um relatório semanal para um canal do Slack. As funções apenas reportam, sem excluir nada, pois a decisão do que apagar fica a critério do time.
Outros recursos que também podem ser analisados:
- Load Balancers sem targets registrados;
- NAT Gateways sem tráfego;
- Log groups do CloudWatch sem política de retenção;
- Volumes gp2 que podem ser migrados para gp3, que é cerca de 20% mais barato;
- Instâncias RDS paradas ou sem conexões;
- Entre outros.
2. Arquitetura da solução
A solução utiliza o EventBridge Scheduler para disparar cada função no horário definido. Cada função consulta as APIs da AWS apenas com permissões de leitura e, se encontrar recursos ociosos, envia uma mensagem para o webhook do Slack. Manter uma função por recurso oferece algumas vantagens, por exemplo, cada uma pode ter sua própria frequência de execução, uma falha não afeta as demais e fica simples ativar somente as verificações que fazem sentido para cada conta.
Abaixo, tabela com as funções:
OBS: a variável “SLACK_WEBHOOK_URL” será inserida em todas as funções.
3. Implementação
3.1 Criação do canal e webhook no Slack
O webhook é uma URL que recebe mensagens via HTTP POST e as publica em um canal específico.
1.Criar o canal que receberá os alertas, por exemplo “finops-alerts”:
2.Acessar api.slack.com/apps, clicar em “Create New App”, “Blank app”, informar um nome (ex: finops-alerts) e selecionar o workspace:
3.No menu lateral, acessar “Incoming Webhooks” e clicar em “Activate Incoming Webhooks”:
4.Em seguida, clicar em “Add New Webhook”, selecionar o canal “finops-alerts” e autorizar:
5.Copiar a URL gerada, que começa com https://hooks.slack.com/services/xxx:
OBS: a URL completa do webhook não deve ser exposta, pois qualquer pessoa com acesso a ela consegue publicar no canal.
--
3.2 Criação da policy e role IAM das funções Lambda
As 4 funções Lambda compartilham uma única role com permissões somente de leitura. Em contas com requisitos mais rígidos de menor privilégio, é possível criar uma role por função, cada uma apenas com as ações que aquela função usa, mas neste artigo, será utilizada a role “finops-idle-resources-role” para as 4.
1.Para criar a policy, será necessário acessar o console da AWS e no serviço IAM, clicar em “Policies”, “Create policy”, abrir a aba JSON e colar a política abaixo:
{
"Version": "2012-10-17",
"Statement": [
{
"Sid": "FinOpsIdleResourcesReadOnly",
"Effect": "Allow",
"Action": [
"ec2:DescribeAddresses",
"ec2:DescribeVolumes",
"ec2:DescribeSnapshots",
"ec2:DescribeImages",
"ec2:DescribeInstances"
],
"Resource": "*"
}
]
}
Salvar a política com o nome “finops-idle-resources-policy”:
2.Para criar a role, será necessário permanecer no IAM, acessar a aba “Roles”, clicar em “Create role”, selecionar as opções “AWS service”, “Lambda” e prosseguir:
3.Em seguida, selecionar as políticas “AWSLambdaBasicExecutionRole”, “finops-idle-resources-policy” , definir o nome da role (ex: finops-idle-resources-role) e clicar em “Create role”:
--
3.3 Função Lambda 1: IPs elásticos ociosos
O processo é o mesmo para as 4 funções, muda apenas o nome, o código e as variáveis opcionais.
1.Para criar a função, será necessário acessar o serviço Lambda e clicar em “Create Event Function”
2.Em seguida, definir o nome “finops-idle-elastic-ips”, linguagem de programação Python, selecionar a arquitetura ARM64 (custa menos que a que a x86_64), selecionar a role criada anteriormente “finops-idle-resources-role” e clicar em “Create Event Function”:
3.Na aba “Code”, substituir o conteúdo de “lambda_function.py” pelo código da função abaixo e clicar em “Deploy”:
import json
import os
import urllib.request
import boto3
SLACK_WEBHOOK_URL = os.environ["SLACK_WEBHOOK_URL"]
IGNORE_TAG = os.environ.get("IGNORE_TAG", "finops-ignore")
MAX_ITEMS = 20
ec2 = boto3.client("ec2")
def is_ignored(tags):
return any(t.get("Key") == IGNORE_TAG and t.get("Value", "").lower() == "true" for t in tags or [])
def get_name(tags):
return next((t["Value"] for t in tags or [] if t.get("Key") == "Name"), "no name")
def find_idle_elastic_ips():
addresses = [a for a in ec2.describe_addresses()["Addresses"] if not is_ignored(a.get("Tags"))]
findings = [
f"`{a['PublicIp']}` ({get_name(a.get('Tags'))}): not associated"
for a in addresses
if "AssociationId" not in a
]
instance_ids = [a["InstanceId"] for a in addresses if a.get("InstanceId")]
if instance_ids:
response = ec2.describe_instances(
InstanceIds=instance_ids,
Filters=[{"Name": "instance-state-name", "Values": ["stopped"]}],
)
stopped = {i["InstanceId"] for r in response["Reservations"] for i in r["Instances"]}
findings += [
f"`{a['PublicIp']}` ({get_name(a.get('Tags'))}): attached to stopped instance `{a['InstanceId']}`"
for a in addresses
if a.get("InstanceId") in stopped
]
return findings
def notify_slack(title, findings, context):
account_id = context.invoked_function_arn.split(":")[4]
region = os.environ["AWS_REGION"]
lines = "\n".join(f"• {item}" for item in findings[:MAX_ITEMS])
if len(findings) > MAX_ITEMS:
lines += f"\n_...and {len(findings) - MAX_ITEMS} more_"
payload = {
"text": f"{title}: {len(findings)} found",
"blocks": [
{"type": "header", "text": {"type": "plain_text", "text": f"{title} ({len(findings)})"}},
{"type": "context", "elements": [{"type": "mrkdwn", "text": f"Account `{account_id}` | Region `{region}`"}]},
{"type": "section", "text": {"type": "mrkdwn", "text": lines[:2900]}},
],
}
request = urllib.request.Request(
SLACK_WEBHOOK_URL,
data=json.dumps(payload).encode("utf-8"),
headers={"Content-Type": "application/json"},
)
with urllib.request.urlopen(request, timeout=10) as response:
return response.status
def lambda_handler(event, context):
findings = find_idle_elastic_ips()
print(json.dumps({"count": len(findings), "findings": findings}))
if findings:
notify_slack("Idle Elastic IPs", findings, context)
return {"count": len(findings)}
4.Em seguida, clicar em “Configuration”, “General configuration” “Edit” e definir “Timeout” de 2 minutos e “Memory” de 256 MB. O timeout padrão de 3 segundos não é suficiente em contas com muitos recursos:
As configurações foram atualizadas com sucesso:
5.O próximo passo é configurar as variáveis de ambiente. Para isso, será necessário clicar em “Configuration”, “Environment variables”, “Edit” e adicionar “SLACK_WEBHOOK_URL” com a URL gerada no tópico 3.1 (https://hooks.slack.com/services/xxx) e, se necessário, as variáveis opcionais da função:
OBS: em ambientes corporativos, o webhook pode ser guardado em uma ferramenta de gerenciamento de segredos, como o AWS Secrets Manager ou outra solução, em vez de uma variável de ambiente, pois as variáveis ficam visíveis para qualquer pessoa com acesso de leitura à função no console. Essa mudança exige um pequeno ajuste no código e não será abordada aqui.
6.Em seguida, podemos testar. Para garantir que a função encontre algo, será necessário criar um IP elástico (ex: eip-test), em “EC2/ Elastic IPs”, “Allocate Elastic IP address” e deixá-lo sem associação com instâncias:
7.Depois, será necessário clicar na aba “Test”, definir o nome do evento (ex: test-01), substituir o conteúdo do campo “Event JSON” por “{}” e clicar em “Test”. O retorno mostra a quantidade de recursos encontrados (ex: {“count”: 2}), e a mensagem chega ao canal do Slack, se o resultado for maior que zero.
Foram localizados 2 IPs ociosos, sendo um sem associação e o outro associado a uma instância parada:
O resultado também foi disparado no canal “finops-alerts”, no Slack:
OBS: o IP público automático atribuído a uma instância é dinâmico e pode mudar quando a instância é parada e iniciada novamente. Já o IP elástico é fixo e permanece reservado à conta até ser liberado. Quando não está associado a uma instância em execução ou a outro recurso, o IP pode gerar cobranças. Por isso, a função verifica os IPs ociosos, não associados a recursos, identificando possíveis oportunidades de redução de custos.
--
3.4 Função Lambda 2: volumes EBS sem anexo
Um disco no estado “available” não está anexado a nenhuma instância, mas continua sendo cobrado por GB provisionado. Isso acontece quando as instâncias são excluídas sem a opção “Delete on termination”.
Nome da função: finops-unattached-ebs-volumes
import json
import os
import urllib.request
from datetime import datetime, timezone
import boto3
SLACK_WEBHOOK_URL = os.environ["SLACK_WEBHOOK_URL"]
IGNORE_TAG = os.environ.get("IGNORE_TAG", "finops-ignore")
MAX_ITEMS = 20
ec2 = boto3.client("ec2")
def is_ignored(tags):
return any(t.get("Key") == IGNORE_TAG and t.get("Value", "").lower() == "true" for t in tags or [])
def get_name(tags):
return next((t["Value"] for t in tags or [] if t.get("Key") == "Name"), "no name")
def find_unattached_volumes():
findings = []
now = datetime.now(timezone.utc)
pages = ec2.get_paginator("describe_volumes").paginate(
Filters=[{"Name": "status", "Values": ["available"]}]
)
for page in pages:
for volume in page["Volumes"]:
if is_ignored(volume.get("Tags")):
continue
age_days = (now - volume["CreateTime"]).days
findings.append(
f"`{volume['VolumeId']}` ({get_name(volume.get('Tags'))}): "
f"{volume['Size']} GiB {volume['VolumeType']}, created {age_days} days ago"
)
return findings
def notify_slack(title, findings, context):
account_id = context.invoked_function_arn.split(":")[4]
region = os.environ["AWS_REGION"]
lines = "\n".join(f"• {item}" for item in findings[:MAX_ITEMS])
if len(findings) > MAX_ITEMS:
lines += f"\n_...and {len(findings) - MAX_ITEMS} more_"
payload = {
"text": f"{title}: {len(findings)} found",
"blocks": [
{"type": "header", "text": {"type": "plain_text", "text": f"{title} ({len(findings)})"}},
{"type": "context", "elements": [{"type": "mrkdwn", "text": f"Account `{account_id}` | Region `{region}`"}]},
{"type": "section", "text": {"type": "mrkdwn", "text": lines[:2900]}},
],
}
request = urllib.request.Request(
SLACK_WEBHOOK_URL,
data=json.dumps(payload).encode("utf-8"),
headers={"Content-Type": "application/json"},
)
with urllib.request.urlopen(request, timeout=10) as response:
return response.status
def lambda_handler(event, context):
findings = find_unattached_volumes()
print(json.dumps({"count": len(findings), "findings": findings}))
if findings:
notify_slack("Unattached EBS volumes", findings, context)
return {"count": len(findings)}
1.Repetir o passo a passo realizado no tópico 3.3 para a criação da função, deploy do código e teste.
2.Realizar o teste “test-02”. Neste caso, também foi criado um disco sem anexo para teste, em “EC2”, “Elastic Block Store/ Volumes” e “Create volume”:
Abaixo, resultado no canal do Slack. Neste caso, a função identificou um volume EBS não anexado a nenhuma instância:
--
3.5 Função Lambda 3: identificar snapshots órfãos
Na AWS existem snapshots de vários serviços e os de volumes EBS (discos das instâncias EC2) estão entre os mais utilizados. Por isso, essa função foca apenas nos snapshots de disco.
Um snapshot é considerado órfão quando o disco que o originou já foi excluído. Neste caso, ele continua sendo cobrado. O prazo mínimo para um snapshot ser reportado é definido pela variável “SNAPSHOT_MIN_AGE_DAYS”. No código, essa variável tem o valor padrão de 90 dias, indicado pelo comentário “default: 90 days if the variable is not set”. Ou seja, se a variável não for cadastrada na função, serão considerados 90 dias, um prazo adequado para ambientes corporativos. Neste exemplo, a variável será cadastrada com o valor “0”, para que o snapshot criado para o teste seja identificado imediatamente.
Nome da função: finops-orphan-ebs-snapshots
import json
import os
import urllib.request
from datetime import datetime, timedelta, timezone
import boto3
SLACK_WEBHOOK_URL = os.environ["SLACK_WEBHOOK_URL"]
IGNORE_TAG = os.environ.get("IGNORE_TAG", "finops-ignore")
MAX_ITEMS = 20
SNAPSHOT_MIN_AGE_DAYS = int(os.environ.get("SNAPSHOT_MIN_AGE_DAYS", "90")) # default: 90 days if the variable is not set
ec2 = boto3.client("ec2")
def is_ignored(tags):
return any(t.get("Key") == IGNORE_TAG and t.get("Value", "").lower() == "true" for t in tags or [])
def get_name(tags):
return next((t["Value"] for t in tags or [] if t.get("Key") == "Name"), "no name")
def find_orphan_snapshots():
now = datetime.now(timezone.utc)
limit = now - timedelta(days=SNAPSHOT_MIN_AGE_DAYS)
existing_volumes = {
v["VolumeId"]
for page in ec2.get_paginator("describe_volumes").paginate()
for v in page["Volumes"]
}
ami_snapshots = {
mapping["Ebs"]["SnapshotId"]
for image in ec2.describe_images(Owners=["self"])["Images"]
for mapping in image.get("BlockDeviceMappings", [])
if mapping.get("Ebs", {}).get("SnapshotId")
}
findings = []
for page in ec2.get_paginator("describe_snapshots").paginate(OwnerIds=["self"]):
for snapshot in page["Snapshots"]:
tags = snapshot.get("Tags") or []
if is_ignored(tags) or any(t["Key"].startswith("aws:backup") for t in tags):
continue # ignored or managed by AWS Backup
if snapshot["StartTime"] > limit:
continue # too recent
if snapshot["SnapshotId"] in ami_snapshots:
continue # used by an AMI
if snapshot.get("VolumeId") in existing_volumes:
continue
age_days = (now - snapshot["StartTime"]).days
findings.append(
f"`{snapshot['SnapshotId']}` ({get_name(tags)}): "
f"{snapshot['VolumeSize']} GiB, created {age_days} days ago"
)
return findings
def notify_slack(title, findings, context):
account_id = context.invoked_function_arn.split(":")[4]
region = os.environ["AWS_REGION"]
lines = "\n".join(f"• {item}" for item in findings[:MAX_ITEMS])
if len(findings) > MAX_ITEMS:
lines += f"\n_...and {len(findings) - MAX_ITEMS} more_"
payload = {
"text": f"{title}: {len(findings)} found",
"blocks": [
{"type": "header", "text": {"type": "plain_text", "text": f"{title} ({len(findings)})"}},
{"type": "context", "elements": [{"type": "mrkdwn", "text": f"Account `{account_id}` | Region `{region}`"}]},
{"type": "section", "text": {"type": "mrkdwn", "text": lines[:2900]}},
],
}
request = urllib.request.Request(
SLACK_WEBHOOK_URL,
data=json.dumps(payload).encode("utf-8"),
headers={"Content-Type": "application/json"},
)
with urllib.request.urlopen(request, timeout=10) as response:
return response.status
def lambda_handler(event, context):
findings = find_orphan_snapshots()
print(json.dumps({"count": len(findings), "findings": findings}))
if findings:
notify_slack("Orphan EBS snapshots", findings, context)
return {"count": len(findings)}
1.Repetir o passo a passo realizado no tópico 3.3 para a criação da função, deploy do código e teste.
2.As variáveis de ambiente foram configuradas da seguinte forma:
3.Para realizar o teste, foi criado um snapshot do disco usado no tópico 3.4, em “EC2”, “Elastic Block Store/ Volumes”, selecionando o disco “test-02”, clicando em “Actions” e “Create snapshot” (ex: snapshot-test-02). Em seguida, o disco foi deletado, mantendo apenas o snapshot. Depois, foi realizado o teste da função “finops-orphan-ebs-snapshots” para identificar snapshots órfãos:
Abaixo, resultado no canal do Slack. Neste caso, a função identificou um snapshot órfão:
--
3.6 Função Lambda 4: instâncias EC2 paradas há muito tempo
Uma instância parada não cobra computação, mas os volumes EBS e IPs associados continuam sendo cobrados. O tempo mínimo de parada para uma instância ser reportada é definido pela variável “STOPPED_MIN_DAYS”. No código, essa variável tem o valor padrão de 30 dias, indicado pelo comentário “default: 30 days if the variable is not set”. Ou seja, se a variável não for cadastrada na função, serão considerados 30 dias. Neste exemplo, a variável será cadastrada com o valor “0”, para que as instâncias paradas para o teste sejam identificadas imediatamente.
Nome da função: finops-long-stopped-ec2
import json
import os
import re
import urllib.request
from datetime import datetime, timedelta, timezone
import boto3
SLACK_WEBHOOK_URL = os.environ["SLACK_WEBHOOK_URL"]
IGNORE_TAG = os.environ.get("IGNORE_TAG", "finops-ignore")
MAX_ITEMS = 20
STOPPED_MIN_DAYS = int(os.environ.get("STOPPED_MIN_DAYS", "30")) # default: 30 days if the variable is not set
STOP_DATE = re.compile(r"\((\d{4}-\d{2}-\d{2} \d{2}:\d{2}:\d{2}) (?:GMT|UTC)\)")
ec2 = boto3.client("ec2")
def is_ignored(tags):
return any(t.get("Key") == IGNORE_TAG and t.get("Value", "").lower() == "true" for t in tags or [])
def get_name(tags):
return next((t["Value"] for t in tags or [] if t.get("Key") == "Name"), "no name")
def find_long_stopped_instances():
now = datetime.now(timezone.utc)
limit = now - timedelta(days=STOPPED_MIN_DAYS)
findings = []
pages = ec2.get_paginator("describe_instances").paginate(
Filters=[{"Name": "instance-state-name", "Values": ["stopped"]}]
)
for page in pages:
for reservation in page["Reservations"]:
for instance in reservation["Instances"]:
if is_ignored(instance.get("Tags")):
continue
match = STOP_DATE.search(instance.get("StateTransitionReason", ""))
if not match:
continue
stopped_at = datetime.strptime(match.group(1), "%Y-%m-%d %H:%M:%S").replace(tzinfo=timezone.utc)
if stopped_at < limit:
findings.append(
f"`{instance['InstanceId']}` ({get_name(instance.get('Tags'))}): "
f"{instance['InstanceType']}, stopped {(now - stopped_at).days} days ago"
)
return findings
def notify_slack(title, findings, context):
account_id = context.invoked_function_arn.split(":")[4]
region = os.environ["AWS_REGION"]
lines = "\n".join(f"• {item}" for item in findings[:MAX_ITEMS])
if len(findings) > MAX_ITEMS:
lines += f"\n_...and {len(findings) - MAX_ITEMS} more_"
payload = {
"text": f"{title}: {len(findings)} found",
"blocks": [
{"type": "header", "text": {"type": "plain_text", "text": f"{title} ({len(findings)})"}},
{"type": "context", "elements": [{"type": "mrkdwn", "text": f"Account `{account_id}` | Region `{region}`"}]},
{"type": "section", "text": {"type": "mrkdwn", "text": lines[:2900]}},
],
}
request = urllib.request.Request(
SLACK_WEBHOOK_URL,
data=json.dumps(payload).encode("utf-8"),
headers={"Content-Type": "application/json"},
)
with urllib.request.urlopen(request, timeout=10) as response:
return response.status
def lambda_handler(event, context):
findings = find_long_stopped_instances()
print(json.dumps({"count": len(findings), "findings": findings}))
if findings:
notify_slack("Long-stopped EC2 instances", findings, context)
return {"count": len(findings)}
1.Repetir o passo a passo realizado no tópico 3.3 para a criação da função, deploy do código e teste.
2.As variáveis de ambiente foram configuradas da seguinte forma:
3.Para realizar o teste, foram utilizados dois servidores “srv-grafana” e “srv-test”, em “EC2/ Instances”, que estão com o status “Stopped”:
Abaixo, resultado no canal do Slack. Neste caso, a função identificou as duas instâncias que estão paradas na conta:
4. Automação da varredura com o EventBridge Scheduler
4.1 Configuração do EventBridge Scheduler
O EventBridge Scheduler é um serviço que pode ser utilizado para executar as funções Lambda automaticamente, nos horários definidos. Cada função recebe seu próprio agendamento, o que permite frequências diferentes por tipo de recurso.
1.No console do EventBridge, acessar a aba “Scheduler”, “Schedules”, “Create schedule” e preencher:
- Schedule name: nome da função (ex: finops-idle-elastic-ips-weekly)
- Schedule pattern: selecionar “Recurring schedule” e “Cron-based schedule”
-
Cron expression:
0 9 ? * MON *(executado toda segunda-feira, às 09h00) - Time zone: America/Sao_Paulo
- Flexible time window: Off
OBS: a frequência semanal evita alertas repetidos, mas ainda assim, ela pode ser ajustada conforme a necessidade de cada ambiente, por exemplo:
-
Diária (todos os dias às 09h00):
0 9 * * ? * -
Mensal (todo dia 1º às 09h00):
0 9 1 * ? *
2.Na etapa seguinte, selecionar “AWS Lambda”, na aba “Invoke” selecionar a função “finops-idle-elastic-ips” e prosseguir:
3.Na etapa a seguir, as configurações iniciais podem ser mantidas conforme o padrão. Em “Permissions”, selecionar a opção “Create new role for this schedule” e definir o nome da role (ex: finops-idle-eips-scheduler-role). Essa role permite que o EventBridge Scheduler invoque a função automaticamente:
4.Em seguida, clicar em “Next”, revisar e clicar em “Create schedule”.
5.O agendamento foi criado com sucesso:
6.Repetir o processo acima para as demais funções Lambda.
5. Conclusão
As 4 funções criadas neste artigo facilitam a identificação de recursos ociosos que continuam gerando custo na conta, como IPs elásticos sem uso, discos sem anexo, snapshots órfãos e instâncias paradas há muito tempo. Com o relatório sendo enviado toda semana no Slack, fica mais fácil revisar esses recursos e definir o que pode ser removido. O mesmo processo também pode ser utilizado para verificar outros recursos, como Load balancers sem targets, NAT gateways sem tráfego, entre outros.
Top comments (0)