Cloud Composer 3 | Cloud Composer 2 | Cloud Composer 1
Este tutorial fornece etapas para depurar um DAG do Airflow com falha no Cloud Composer e diagnosticar problemas relacionados a recursos de worker, como falta de memória ou espaço de armazenamento, com a ajuda de registros e monitoramento do ambiente.
Introdução
Este tutorial se concentra em problemas relacionados a recursos para demonstrar maneiras de depurar uma DAG.
A falta de recursos alocados do worker causa falhas no DAG. Se uma tarefa do Airflow ficar sem memória ou armazenamento, uma exceção do Airflow pode aparecer, como:
WARNING airflow.exceptions.AirflowException: Task received SIGTERM signal
INFO - Marking task as FAILED.
ou
Task exited with return code Negsignal.SIGKILL
Nesses casos, a recomendação geral é aumentar os recursos do worker do Airflow ou reduzir o número de tarefas por worker. No entanto, como as exceções do Airflow podem ser genéricas, pode ser difícil identificar o recurso específico que está causando o problema.
Este tutorial explica como diagnosticar a causa de uma falha do DAG e identificar o tipo de recurso que causa problemas, destravando dois exemplos de DAGs que falham devido à falta de memória e armazenamento do worker.
Objetivos
Execute DAGs de exemplo que falham pelos seguintes motivos:
- Falta de memória do worker
- Falta de armazenamento de worker
Diagnosticar os motivos da falha
Aumentar os recursos de worker alocados
Testar as DAGs com novos limites de recursos
Custos
Neste tutorial, usamos os seguintes componentes faturáveis do Google Cloud:
- Cloud Composer (consulte os custos adicionais).
- Cloud Monitoring
Ao concluir este tutorial, exclua os recursos criados para evitar o faturamento contínuo. Para mais detalhes, consulte Limpeza.
Antes de começar
Esta seção descreve as ações necessárias antes de iniciar o tutorial.
Criar e configurar um projeto
Para este tutorial, você precisa de um Google Cloud projeto. Configure o projeto da seguinte maneira:
No console do Google Cloud, selecione ou crie um projeto:
Verifique se o faturamento foi ativado para o projeto. Saiba como verificar se o faturamento está ativado em um projeto.
Verifique se o Google Cloud usuário do projeto tem os seguintes papéis para criar os recursos necessários:
- Administrador de objetos do armazenamento e do ambiente
(
roles/composer.environmentAndStorageObjectAdmin
) - Administrador do Compute (
roles/compute.admin
) - Editor do Monitoring (
roles/monitoring.editor
)
- Administrador de objetos do armazenamento e do ambiente
(
Ativar as APIs do projeto
Enable the Cloud Composer API.
Criar seu ambiente do Cloud Composer
Crie um ambiente do Cloud Composer 2.
Como parte da criação do ambiente,
conceda o papel Extensão do agente de serviço da API Cloud Composer v2
(roles/composer.ServiceAgentV2Ext
) à conta do agente de serviço do Composer. O Cloud Composer usa essa conta para realizar operações
no projeto Google Cloud .
Verificar os limites de recursos do worker
Verifique os limites de recursos do worker do Airflow no seu ambiente:
No console do Google Cloud, acesse a página Ambientes.
Na lista de ambientes, clique no nome do seu ambiente. A página Detalhes do ambiente é aberta.
Acesse a guia Configuração do ambiente.
Acesse Recursos > Configuração de cargas de trabalho > Worker.
Verifique se os valores são 0,5 vCPUs, 1,875 GB de memória e 1 GB de armazenamento. Estes são os limites de recursos do worker do Airflow com os quais você vai trabalhar nas próximas etapas deste tutorial.
Exemplo: diagnosticar problemas de falta de memória
Faça upload do DAG de exemplo abaixo para o ambiente
criado nas etapas anteriores. Neste tutorial, o DAG é chamado de
create_list_with_many_strings
.
Esse DAG contém uma tarefa que executa as seguintes etapas:
- Cria uma lista vazia
s
. - Executa um ciclo para anexar a string
More
à lista. - Mostra quanta memória a lista consome e aguarda um segundo em cada iteração de um minuto.
import time
import airflow
from airflow import DAG
from airflow.operators.python_operator import PythonOperator
import sys
from datetime import timedelta
default_args = {
'start_date': airflow.utils.dates.days_ago(0),
'retries': 0,
'retry_delay': timedelta(minutes=10)
}
dag = DAG(
'create_list_with_many_strings',
default_args=default_args,
schedule_interval=None)
def consume():
s = []
for i in range(120):
for j in range(1000000):
s.append("More")
print(f"i={i}; size={sys.getsizeof(s) / (1000**3)}GB")
time.sleep(1)
t1 = PythonOperator(
task_id='task0',
python_callable=consume,
dag=dag,
depends_on_past=False,
retries=0
)
Acionar o DAG de exemplo
Acione o DAG de exemplo, create_list_with_many_strings
:
No console do Google Cloud, acesse a página Ambientes.
Na coluna Servidor da Web do Airflow, siga o link Airflow do ambiente.
Na interface da Web do Airflow, na página DAGs, na coluna Links do DAG, clique no botão Acionar DAG.
Clique em Gatilho.
Na página DAGs, clique na tarefa acionada e analise os registros de saída para garantir que o DAG começou a ser executado.
Enquanto a tarefa estiver em execução, os registros de saída vão mostrar o tamanho da memória em GB que o DAG está usando.
Após vários minutos, a tarefa falha porque excede o limite de memória de 1,875 GB do worker do Airflow.
Diagnosticar o DAG com falha
Se você estava executando várias tarefas no momento da falha, considere executar apenas uma tarefa e diagnosticar a pressão de recursos durante esse período para identificar quais tarefas causam pressão de recursos e quais recursos você precisa aumentar.
Analisar os registros de tarefas do Airflow
Observe que a tarefa da DAG create_list_with_many_strings
tem um
estado Failed
.
Revise os registros da tarefa. Você vai encontrar a seguinte entrada de registro:
```none
{local_task_job.py:102} INFO - Task exited with return code
Negsignal.SIGKILL
```
`Netsignal.SIGKILL` might be an indication of your task using more memory
than the Airflow worker is allocated. The system sends
the `Negsignal.SIGKILL` signal to avoid further memory consumption.
Analisar cargas de trabalho
Analise as cargas de trabalho para verificar se a carga da tarefa não faz com que o nó em que o pod é executado exceda o limite de consumo de memória:
No console do Google Cloud, acesse a página Ambientes.
Na lista de ambientes, clique no nome do seu ambiente. A página Detalhes do ambiente é aberta.
Acesse a guia Configuração do ambiente.
Em Recursos > Cluster do GKE > Cargas de trabalho, clique em ver cargas de trabalho do cluster.
Verifique se alguns dos pods de carga de trabalho têm status semelhante ao seguinte:
Error with exit code 137 and 1 more issue. ContainerStatusUnknown with exit code 137 and 1 more issue
Exit code 137
significa que um contêiner ou pod está tentando usar mais memória do que o permitido. O processo é encerrado para evitar o uso de memória.
Analisar a integridade do ambiente e o monitoramento do consumo de recursos
Analise o monitoramento da integridade do ambiente e do consumo de recursos:
No console do Google Cloud, acesse a página Ambientes.
Na lista de ambientes, clique no nome do seu ambiente. A página Detalhes do ambiente é aberta.
Acesse a guia Monitoramento e selecione Visão geral.
No painel Visão geral do ambiente, localize o gráfico Integridade do ambiente (DAG de monitoramento do Airflow). Ele contém uma área vermelha, que corresponde ao momento em que os registros começaram a imprimir erros.
Selecione Workers e encontre o gráfico Total workers memory usage. Observe que a linha Uso de memória tem um pico no momento em que a tarefa estava em execução.
Mesmo que a linha de uso de memória no gráfico não atinja o limite, ao diagnosticar os motivos da falha, é necessário considerar apenas a memória alocável, enquanto a linha Limite de memória no gráfico representa a memória total disponível (incluindo a capacidade reservada pelo GKE).
Neste exemplo, o limite de memória do worker é definido como 1,875 GB. O GKE reserva 25% dos primeiros 4 GiB de memória. O GKE também reserva um limite de remoção (link em inglês) extra: 100 MiB de memória em cada nó para remoção de kubelet.
A memória alocável é calculada da seguinte maneira:
ALLOCATABLE = CAPACITY - RESERVED - EVICTION-THRESHOLD
Se o limite de memória for 1,875 GB, a memória alocável real será:
1.75 GiB (1.875GB) - 0.44 (25% GiB reserved) - 0.1 = 1.21 GiB (~1.3 GB).
Quando você anexar esse limite real ao gráfico de uso da memória, vai notar que o pico de uso da memória da tarefa atinge o limite real e você pode concluir que a tarefa falhou devido à memória insuficiente do worker.
Aumentar o limite de memória do worker
Aloque mais memória de worker para que o DAG de exemplo funcione:
No console do Google Cloud, acesse a página Ambientes.
Na lista de ambientes, clique no nome do seu ambiente. A página Detalhes do ambiente é aberta.
Acesse a guia Configuração do ambiente.
Encontre a configuração Recursos > Cargas de trabalho e clique em Editar.
Na seção Worker, no campo Memory, especifique o novo limite de memória para workers do Airflow. Neste tutorial, use 3 GB.
Salve as mudanças e aguarde alguns minutos para que os workers do Airflow sejam reiniciados.
Testar seu DAG com o novo limite de memória
Acione o DAG create_list_with_many_strings
novamente e aguarde até que ele
termine a execução.
Nos registros de saída da execução do DAG, você vai encontrar
Marking task as SUCCESS
, e o estado da tarefa vai indicar Success.Revise a seção Visão geral do ambiente na guia Monitoring e verifique se não há áreas vermelhas.
Clique na seção Workers e encontre o gráfico Total workers memory usage. A linha Limite de memória reflete a mudança no limite de memória, e a linha Uso de memória está muito abaixo do limite de memória alocável real.
Exemplo: diagnosticar problemas de falta de armazenamento
Nesta etapa, você faz upload de duas DAGs que criam arquivos grandes. O primeiro DAG cria um arquivo grande. O segundo DAG cria um arquivo grande e imita uma operação de longa duração.
O tamanho do arquivo em ambos os DAGs excede o limite de armazenamento de 1 GB do worker do Airflow, mas o segundo DAG tem uma tarefa de espera adicional para estender a duração artificialmente.
Você vai investigar as diferenças no comportamento de ambas as DAGs nas próximas etapas.
Fazer upload de um DAG que cria um arquivo grande
Faça upload do DAG de exemplo abaixo para o ambiente
criado nas etapas anteriores. Neste tutorial, o DAG é chamado de
create_large_txt_file_print_logs
.
Esse DAG contém uma tarefa que executa as seguintes etapas:
- Grava um arquivo
localfile.txt
de 1,5 GB no armazenamento do worker do Airflow. - Mostra o tamanho do arquivo criado usando o módulo
os
do Python. - Mostra a duração da execução do DAG a cada 1 minuto.
import airflow
from airflow import DAG
from airflow.operators.python_operator import PythonOperator
import os
from datetime import timedelta
import time
default_args = {
'start_date': airflow.utils.dates.days_ago(0),
'retries': 0,
'retry_delay': timedelta(minutes=10)
}
dag = DAG(
'create_large_txt_file_print_logs',
default_args=default_args,
schedule_interval=None)
def consume():
size = 1000**2 # bytes in 1 MB
amount = 100
def create_file():
print(f"Start creating a huge file")
with open("localfile.txt", "ab") as f:
for j in range(15):
f.write(os.urandom(amount) * size)
print("localfile.txt size:", os.stat("localfile.txt").st_size / (1000**3), "GB")
create_file()
print("Success!")
t1 = PythonOperator(
task_id='create_huge_file',
python_callable=consume,
dag=dag,
depends_on_past=False,
retries=0)
Fazer upload de um DAG que cria um arquivo grande em uma operação de longa duração
Para imitar um DAG de longa duração e investigar o impacto da duração da tarefa
no estado final, faça o upload do segundo DAG de exemplo para o
ambiente. Neste tutorial, o DAG é chamado de
long_running_create_large_txt_file_print_logs
.
Esse DAG contém uma tarefa que executa as seguintes etapas:
- Grava um arquivo
localfile.txt
de 1,5 GB no armazenamento do worker do Airflow. - Mostra o tamanho do arquivo criado usando o módulo
os
do Python. - Espera 1 hora e 15 minutos para imitar o tempo necessário para operações com o arquivo, por exemplo, a leitura dele.
- Mostra a duração da execução do DAG a cada 1 minuto.
import airflow
from airflow import DAG
from airflow.operators.python_operator import PythonOperator
import os
from datetime import timedelta
import time
default_args = {
'start_date': airflow.utils.dates.days_ago(0),
'retries': 0,
'retry_delay': timedelta(minutes=10)
}
dag = DAG(
'long_running_create_large_txt_file_print_logs',
default_args=default_args,
schedule_interval=None)
def consume():
size = 1000**2 # bytes in 1 MB
amount = 100
def create_file():
print(f"Start creating a huge file")
with open("localfile.txt", "ab") as f:
for j in range(15):
f.write(os.urandom(amount) * size)
print("localfile.txt size:", os.stat("localfile.txt").st_size / (1000**3), "GB")
create_file()
for k in range(75):
time.sleep(60)
print(f"{k+1} minute")
print("Success!")
t1 = PythonOperator(
task_id='create_huge_file',
python_callable=consume,
dag=dag,
depends_on_past=False,
retries=0)
Acionar DAGs de exemplo
Acione o primeiro DAG, create_large_txt_file_print_logs
:
No console do Google Cloud, acesse a página Ambientes.
Na coluna Servidor da Web do Airflow, siga o link Airflow do ambiente.
Na interface da Web do Airflow, na página DAGs, na coluna Links do DAG, clique no botão Acionar DAG.
Clique em Gatilho.
Na página DAGs, clique na tarefa acionada e analise os registros de saída para garantir que o DAG começou a ser executado.
Aguarde até que a tarefa criada com o DAG
create_large_txt_file_print_logs
seja concluída. Isso pode levar alguns minutos.Na página DAGs, clique na execução do DAG. A tarefa terá um estado
Success
, mesmo que o limite de armazenamento tenha sido excedido.
Analise os registros do Airflow da tarefa:
No console do Google Cloud, acesse a página Ambientes.
Na lista de ambientes, clique no nome do seu ambiente. A página Detalhes do ambiente é aberta.
Acesse a guia Registros e depois Todos os registros > Registros do Airflow > Workers > Visualizar no Logs Explorer.
Filtre os registros por tipo: mostre apenas mensagens de erro.
Nos registros, você vai encontrar mensagens semelhantes a esta:
Worker: warm shutdown (Main Process)
ou
A worker pod was evicted at 2023-12-01T12:30:05Z with message: Pod ephemeral
local storage usage exceeds the total limit of containers 1023Mi.
Esses registros indicam que o pod iniciou o processo de "desligamento aquecido" porque o armazenamento usado excedeu o limite e foi removido em 1 hora. No entanto, a execução do DAG não falhou porque foi concluída dentro do período de tolerância de encerramento do Kubernetes, explicado mais detalhadamente neste tutorial.
Para ilustrar o conceito do período de carência de rescisão, analise o resultado
do segundo DAG de exemplo, long_running_create_large_txt_file_print_logs
.
Acione o segundo DAG, long_running_create_large_txt_file_print_logs
:
No console do Google Cloud, acesse a página Ambientes.
Na coluna Servidor da Web do Airflow, siga o link Airflow do ambiente.
Na interface da Web do Airflow, na página DAGs, na coluna Links do DAG, clique no botão Acionar DAG.
Clique em Gatilho.
Na página DAGs, clique na tarefa acionada e analise os registros de saída para garantir que o DAG começou a ser executado.
Aguarde até que a execução do DAG
long_running_create_large_txt_file_print_logs
falha. Isso vai levar cerca de uma hora.
Analise os resultados da execução do DAG:
Na página DAGs, clique na execução do DAG
long_running_create_large_txt_file_print_logs
. Você vai notar que a tarefa tem um estadoFailed
e que a duração da execução foi exatamente 1 hora e 5 minutos, o que é menor do que o período de espera da tarefa de 1 hora e 15 minutos.Revise os registros da tarefa. Depois que o DAG cria o arquivo
localfile.txt
no contêiner do worker do Airflow, o registro mostra que o DAG começou a aguardar, e a duração da execução é impressa nos registros de tarefas a cada 1 minuto. Neste exemplo, o DAG imprime o registrolocalfile.txt size:
, e o tamanho do arquivolocalfile.txt
será de 1,5 GB.
Quando o arquivo gravado no contêiner do worker do Airflow excede o limite de armazenamento, a execução do DAG falha. No entanto, a tarefa não falha imediatamente e continua sendo executada até que a duração atinja 1 hora e 5 minutos. Isso acontece porque o Kubernetes não encerra a tarefa imediatamente e continua em execução para permitir uma hora de tempo de recuperação, conhecido como "período de tolerância de encerramento". Quando um nó fica sem recursos, o Kubernetes não encerra o pod imediatamente para processar a interrupção de forma adequada, para que haja um impacto mínimo no usuário final.
O período de carência de encerramento ajuda os usuários a recuperar arquivos após falhas de tarefas, mas pode causar confusão ao diagnosticar DAGs. Quando o limite de armazenamento do worker do Airflow é excedido, o estado final da tarefa depende da duração da execução da DAG:
Se a execução da DAG exceder o limite de armazenamento do worker, mas for concluída em menos de uma hora, a tarefa será concluída com um status
Success
porque foi concluída dentro do período de tolerância de encerramento. No entanto, o Kubernetes encerra o pod e o arquivo gravado é excluído do contêiner imediatamente.Se o DAG exceder o limite de armazenamento do worker e for executado por mais de uma hora, ele continuará sendo executado por uma hora e poderá exceder o limite de armazenamento em milhares de porcentagens antes que o Kubernetes elimine o pod e o Airflow marque a tarefa como
Failed
.
Diagnosticar o DAG com falha
Se você estava executando várias tarefas no momento da falha, considere executar apenas uma tarefa e diagnosticar a pressão de recursos durante esse período para identificar quais tarefas causam pressão de recursos e quais recursos você precisa aumentar.
Analise os registros de tarefas do segundo DAG,
long_running_create_large_txt_file_print_logs
:
No console do Google Cloud, acesse a página Ambientes.
Na lista de ambientes, clique no nome do seu ambiente. A página Detalhes do ambiente é aberta.
Acesse a guia Registros e depois Todos os registros > Registros do Airflow > Workers > Visualizar no Logs Explorer.
Filtre os registros por tipo: mostre apenas mensagens de erro.
Nos registros, você vai encontrar mensagens semelhantes a esta:
Container storage usage of worker reached 155.7% of the limit.
This likely means that the total size of local files generated by your DAGs is
close to the storage limit of worker.
You may need to decrease the storage usage or increase the worker storage limit
in your Cloud Composer environment configuration.
ou
Pod storage usage of worker reached 140.2% of the limit.
A worker pod was evicted at 2023-12-01T12:30:05Z with message: Pod ephemeral
local storage usage exceeds the total limit of containers 1023Mi.
This eviction likely means that the total size of dags and plugins folders plus
local files generated by your DAGs exceeds the storage limit of worker.
Please decrease the storage usage or increase the worker storage limit in your
Cloud Composer environment configuration.
Essas mensagens indicam que, à medida que a tarefa avançava, os registros do Airflow começaram a imprimir erros quando o tamanho dos arquivos gerados pelo DAG excedeu o limite de armazenamento do worker e o período de carência de encerramento começou. Durante o período de carência de encerramento, o consumo de armazenamento não retornou ao limite, o que levou à remoção do pod após o término do período de carência de encerramento.
Analise o monitoramento da integridade do ambiente e do consumo de recursos:
No console do Google Cloud, acesse a página Ambientes.
Na lista de ambientes, clique no nome do seu ambiente. A página Detalhes do ambiente é aberta.
Acesse a guia Monitoramento e selecione Visão geral.
No painel Visão geral do ambiente, localize o gráfico Integridade do ambiente (DAG de monitoramento do Airflow). Ele contém uma área vermelha, que corresponde ao momento em que os registros começaram a imprimir erros.
Selecione Workers e encontre o gráfico Total workers disk usage. Observe que a linha Uso do disco tem um pico e excede a linha Limite do disco no momento em que a tarefa estava em execução.
Aumentar o limite de armazenamento do trabalhador
Aloque mais armazenamento de worker do Airflow para que o DAG de exemplo venha a ser bem-sucedido:
No console do Google Cloud, acesse a página Ambientes.
Na lista de ambientes, clique no nome do seu ambiente. A página Detalhes do ambiente é aberta.
Acesse a guia Configuração do ambiente.
Encontre a configuração Recursos > Cargas de trabalho e clique em Editar.
Na seção Worker, no campo Armazenamento, especifique o novo limite de armazenamento para os workers do Airflow. Neste tutorial, defina como 2 GB.
Salve as mudanças e aguarde alguns minutos para que os workers do Airflow sejam reiniciados.
Testar a DAG com o novo limite de armazenamento
Acione o DAG long_running_create_large_txt_file_print_logs
novamente e
aguarde uma hora e 15 minutos até que ele seja concluído.
Nos logs de saída da execução do DAG, você vai encontrar
Marking task as SUCCESS
, e o estado da tarefa vai indicar Sucesso, com uma duração de 1 hora e 15 minutos, o que é igual ao tempo de espera definido no código do DAG.Revise a seção Visão geral do ambiente na guia Monitoring e verifique se não há áreas vermelhas.
Clique na seção Workers e encontre o gráfico Total workers disk usage. A linha Limite de disco reflete a mudança no limite de armazenamento, e a linha Uso de disco está dentro do intervalo permitido.
Resumo
Neste tutorial, você diagnosticou o motivo de uma falha no DAG e identificou o tipo de recurso que causa pressão, destravando dois exemplos de DAGs que falham devido à falta de memória e armazenamento do worker. Em seguida, você executou os DAGs com sucesso após alocar mais memória e armazenamento para os workers. No entanto, é recomendado otimizar seus DAGs (fluxos de trabalho) para reduzir o consumo de recursos de workers. Isso porque não é possível aumentar os recursos além de um determinado limite.
Limpar
Para evitar cobranças na sua conta do Google Cloud pelos recursos usados neste tutorial, exclua o projeto que contém os recursos ou mantenha o projeto e exclua os recursos individuais.
Excluir o projeto
- In the Google Cloud console, go to the Manage resources page.
- In the project list, select the project that you want to delete, and then click Delete.
- In the dialog, type the project ID, and then click Shut down to delete the project.
Excluir recursos individuais
Se você planeja ver vários tutoriais e guias de início rápido, a reutilização de projetos pode evitar que você exceda os limites da cota do projeto.
Exclua o ambiente do Cloud Composer. Você também exclui o bucket do ambiente durante esse procedimento.