Google Cloud, tarefas de longo prazo e Compute Engine

Recentemente, trabalhei em um projeto que exigia a implementação de um serviço ETL que ultrapassava os limites de alguns serviços serverless do Google Cloud, como Cloud Run e Cloud Functions. Embora o Google Cloud ofereça ferramentas desenhadas para essas tarefas, outro ponto foi buscado para manter os custos baixos. Portanto, Dataflow, Dataproc ou qualquer um desses serviços orientados a pipeline de dados aumentaram o faturamento a um ponto em que não era viável.

Isso levou-me a iterar diferentes arquiteturas que cumpriam todos os requisitos e sem as limitações que outros serviços impõem, como timeouts de 30 minutos a 1 hora ou a persistência de dados conseguida integrando mais componentes à arquitetura (e, portanto, um maior grau de manutenção) de certos serviços stateless.

Depois de analisá-lo, decidi implementar uma arquitetura através do Compute Engine, mas com a variante que rodaria o container do serviço ao invés de um servidor tradicional, juntamente com outros componentes que, ao trabalharem juntos, permitiriam que o serviço rodasse sem nenhuma limitação, bem como descartar todos os recursos quando terminar.

Neste artigo, mostrarei como implementar esta arquitetura para executar um simples programa scraper, observe que este é um exemplo, se você realmente considera esta arquitetura para o seu projeto, deve fazê-lo pelos seguintes motivos:

- O tempo de execução excede os limites de outros serviços.
- Mantenha os custos baixos.
- Reduza o número de dependências em sua infraestrutura.

Requisitos para realizar este projeto

• Conhecimento intermediário em Python e Virtualenv.
• Conhecimento em Git e GitHub.
• Conhecimento em Google Cloud.
• Um projeto no Google Cloud com faturamento ativado ou com créditos atuais.
• Uma conta de serviço com as seguintes funções:

 -Administrador de computação.
 -Gravador de log.
 -Invocador de fluxo de trabalho.
 -Usuário de conta de serviço.
 -Administrador de armazenamento.

• Gerar uma chave para a conta de serviço (se for executá-la localmente).
• Instalar o Google Cloud SDK e configure o projeto localmente usando o comando gcloud init.

Este seria o diagrama de arquitetura considerado:

Antes de falar sobre infraestrutura, primeiro precisamos definir e executar o serviço responsável pela(s) tarefa(s) de longa duração. Para este artigo, escrevi um scraper em Python, mas como vamos usar o Docker, você pode usar qualquer linguagem de sua escolha. Como a arquitetura não depende de SO ou linguagem, você só precisa de um ambiente onde possa montar a imagem e rodar o container.

Agora vamos rever a estrutura do projeto a ser implementado, não vou perder muito tempo explicando o código já que este artigo é mais focado na parte de infraestrutura. No entanto, no futuro, posso fazer outro onde examinamos o código com mais detalhes.

Aqui está o link para o repositório onde explico os passos para configurar este serviço.

Hora de começar

Esta será a estrutura do projeto:

scraper/

scrape.py

Dockerfile

requirements.txt

start_vm.yml

config.json

1) Clone o repositório e crie seu virtualenv

git clone https://github.com/luigicfh/scraper.git

2) Crie o virtualenv

virtualenv venv

3) Ative o virtualenv

source venv/bin/activate

4) Instale as dependências

pip install -r requirements.txt

5) Examine o arquivo scrape.py

from bs4 import BeautifulSoup, element

import requests

import json

from google.cloud import compute_v1, storage

import google.cloud.logging

from google.cloud.exceptions import NotFound

import logging

import traceback

import os

import datetime

BASE_URL = 'https://books.toscrape.com'

PAGE_URL = BASE_URL + '/catalogue/page-{}.html'

LOG_MSG = "Scraper logs: "

def setup_env_variables() -> None:

with open('config.json') as file:

to_dict = json.loads(file.read())

os.environ['PROJECT_ID'] = to_dict['project_id']

os.environ['ZONE'] = to_dict['zone']

os.environ['INSTANCE_NAME'] = to_dict['instance_name']

os.environ['BUCKET'] = to_dict['bucket']

os.environ['FOLDER'] = to_dict['folder']

def get_products() -> list[element.ResultSet]:

product_list = []

page = 1

while True:

response = requests.get(url=PAGE_URL.format(page))

if response.status_code != 200:

break

soup = BeautifulSoup(response.content, 'html.parser')

products = soup.find_all('article', {'class': 'product_pod'})

product_list.append(products)

page += 1

return product_list

def setup_logging() -> None:

logging_client = google.cloud.logging.Client()

logging_client.setup_logging()

def delete_instance() -> None:

gce_client = compute_v1.InstancesClient()

try:

gce_client.delete(

project=os.environ.get('PROJECT_ID'),

zone=os.environ.get('ZONE'),

instance=os.environ.get('INSTANCE_NAME')

)

except NotFound:

pass

def upload_to_gcs(data: str) -> None:

gcs_client = storage.Client()

bucket = gcs_client.get_bucket(os.environ.get("BUCKET"))

blob = bucket.blob(f"{os.environ.get('FOLDER')}/results.json")

blob.upload_from_string(data=data, content_type="application/json")

def handle_exception() -> None:

logging.error(LOG_MSG + traceback.format_exc())

delete_instance()

def to_list(products: list[element.ResultSet]) -> list[dict]:

d = {}

data = []

for product in products:

d['book_title'] = product[0].find('h3').find('a')['title']

d['image'] = BASE_URL + product[0].find(attrs={'class': 'image_container'}).find(

'a').find('img')['src'].replace("..", '')

d['price'] = product[0].find(attrs={'class': 'product_price'}).find(

attrs={'class': 'price_color'}).get_text()

d['in_stock'] = product[0].find(

attrs={'class': 'instock availability'}).get_text().replace('\n', '').strip()

d['rating'] = product[0].find(attrs={'class': 'star-rating'})[

'class'][-1]

data.append(d)

d = {}

return data

def scrape() -> None:

logging.info(

LOG_MSG + f"scraping started {datetime.datetime.now().isoformat()}")

book_products = get_products()

logging.info(

LOG_MSG + f"transforming data {datetime.datetime.now().isoformat()}")

data_list = to_list(book_products)

to_json = json.dumps(data_list, indent=4)

logging.info(

LOG_MSG + f"uploading file {datetime.datetime.now().isoformat()}")

upload_to_gcs(to_json)

logging.info(

LOG_MSG + f"scraping process finished, deleting instance {datetime.datetime.now().isoformat()}")

delete_instance()

if __name__ == '__main__':

setup_env_variables()

setup_logging()

try:

scrape()

except Exception:

handle_exception()

Embora eu já tenha mencionado que não examinaremos o código em detalhes, quero observar um ponto importante: se voltarmos ao diagrama de arquitetura, você notará que existe um workflow cujo trabalho é crie a instância do Compute Engine e configure-a para iniciar a execução da tarefa. Mas quem destrói a instância quando o serviço termina sua execução? A resposta é: o mesmo serviço, pois só ele sabe quando terminou a execução, sem depender de outras técnicas como o polling para isso.

O código abaixo mostra como esse serviço destrói a instância em duas situações diferentes: se a execução foi bem-sucedida e se falhou (já que não queremos uma instância ativa do Compute Engine se algo der errado).

A tentativa de exclusão da instância está dentro de um bloco try except, porque se você executar testes locais, a instância do Compute Engine provavelmente não existe.

def delete_instance():

gce_client = compute_v1.InstancesClient()

try:

gce_client.delete(

project=os.environ.get('PROJECT_ID'),

zone=os.environ.get('ZONE'),

instance=os.environ.get('INSTANCE_NAME')

)

except NotFound:

pass

O último item a examinar, mas não menos importante, é o nosso Dockerfile, onde o ponto principal é a última linha. Como você pode ver, a execução do serviço é iniciada dentro deste mesmo arquivo.

FROM python:3.10-slim

ENV PYTHONUNBUFFERED True

ENV APP_HOME /app

WORKDIR $APP_HOME

COPY . ./

RUN apt-get update -y && apt-get update

RUN pip install --upgrade pip

RUN pip install -r requirements.txt

CMD python scrape.py

O próprio scraper já integra dois serviços do Google Cloud. A primeira delas é o Cloud Logging, utilizado para registrar o processo enquanto ele está rodando, bem como eventuais erros que possam surgir durante sua execução. As vantagens de usar o Cloud Logging é que você pode criar visualizações personalizadas que mostram apenas os logs do serviço em questão, bem como criar alertas para algum tipo de log que notifiquem os responsáveis ​​pelo monitoramento do serviço se algo estiver errado.

Para integrar o Cloud Logging ao seu projeto, você deve primeiro inicializar o cliente e, em seguida, usar a biblioteca de log Python para gerar seus logs no Google Cloud.

def setup_logging():

logging_client = google.cloud.logging.Client()

logging_client.setup_logging()

O segundo serviço usado é o Cloud Storage, que usamos para armazenar os dados convertidos em JSON que extraímos do site. Abaixo você pode examinar a função upload_to_gcs().

def upload_to_gcs(data: str) -> None:

gcs_client = storage.Client()

bucket = gcs_client.get_bucket(os.environ.get("BUCKET"))

blob = bucket.blob(f"{os.environ.get('FOLDER')}/results.json")

blob._chunk_size = 8388608  # 1024 * 1024 B * 16 = 8 MB

blob.upload_from_string(data=data, content_type="application/json")

Para gerar a imagem do nosso serviço e levá-la ao Google Cloud iremos utilizar mais dois serviços: Cloud Build para construir nossa imagem junto com todas as suas dependências, e Container Registry, onde nossa imagem estará disponível para ser baixada e implantada em qualquer serviço que execute contêineres.

Para construir nossa imagem e salvá-la no Container Registry, usaremos o seguinte comando:

Obs: Deve ser executado dentro do arquivo onde está localizado o Dockerfile.

gcloud builds submit --tag=gcr.io/<project_id>/<service_name>

Ok, agora temos um serviço que executa nossa tarefa de scraping e integra diferentes serviços do Google Cloud, agora como automatizamos esse processo de acordo com um cronograma definido? Agora falaremos sobre fluxos de trabalho e como eles podem nos ajudar a executar tarefas sequenciais ou até paralelas no Google Cloud.

Vamos examinar o arquivo start_vm.yml.

main:

steps:

- init:

assign:

- projectId: ${sys.get_env("GOOGLE_CLOUD_PROJECT_ID")}

- projectNumber: ${sys.get_env("GOOGLE_CLOUD_PROJECT_NUMBER")}

- zone: "us-east1-b"

- machineType: "e2-micro"

- instanceName: "scraper-vm"

- log_vm_creation_start:

call: sys.log

args:

data: ${"Scraper logs:initializing VM"}

- create_and_start_vm:

call: googleapis.compute.v1.instances.insert

args:

project: ${projectId}

zone: ${zone}

body:

name: ${instanceName}

machineType: ${"zones/" + zone + "/machineTypes/" + machineType}

disks:

- initializeParams:

sourceImage: "projects/cos-cloud/global/images/cos-stable-93-16623-102-1"

diskSizeGb: "10"

boot: true

autoDelete: true

networkInterfaces:

- accessConfigs:

- kind: compute#accessConfig

name: "external-nat"

networkTier: "PREMIUM"

metadata:

items:

- key: "gce-container-declaration"

value: '${"spec:\n  containers:\n  - name: scraper\n    image: gcr.io/" + projectId + "/scraper\n    stdin: false\n    tty: false\n  restartPolicy: Always\n"}'

- key: "google-logging-enabled"

value: "true"

- key: "google-monitoring-enabled"

value: "true"

serviceAccounts:

- email: "service-account-name@project-id.iam.gserviceaccount.com"

scopes:

- https://www.googleapis.com/auth/devstorage.read_only

- https://www.googleapis.com/auth/logging.write

- https://www.googleapis.com/auth/monitoring.write

- https://www.googleapis.com/auth/servicecontrol

- https://www.googleapis.com/auth/service.management.readonly

- https://www.googleapis.com/auth/trace.append

- https://www.googleapis.com/auth/cloud-platform

- log_wait_for_vm_network:

call: sys.log

args:

data: ${"Scraper logs:Waiting for VM network to initialize"}

- wait_for_vm_network:

call: sys.sleep

args:

seconds: 10

- log_vm_creation_end:

call: sys.log

args:

data: ${"Scraper logs:VM initialized successfully"}

A primeira etapa do nosso workflow é o init, onde simplesmente atribuímos algumas variáveis ​​que serão utilizadas posteriormente.

Além disso, aqui você pode configurar o tamanho da máquina que deseja criar, desde máquinas predefinidas até máquinas com especificações personalizadas.

- init:

assign:

- projectId: ${sys.get_env("GOOGLE_CLOUD_PROJECT_ID")}

- projectNumber: ${sys.get_env("GOOGLE_CLOUD_PROJECT_NUMBER")}

- zone: "us-east1-b"

- machineType: "e2-micro"

- instanceName: "scraper-vm"

A próxima etapa é simplesmente um utilitário para registrar no Cloud Logging.

- log_vm_creation_start:

call: sys.log

args:

data: ${"Scraper logs:initializing VM"}

Agora vamos revisar a etapa mais complexa, que executa a criação da máquina, instala o Container OS e nossa imagem de serviço.

• call: determina a API do Google Cloud e o método que chamamos para a criação da máquina.
• machineType: é onde configuramos o tipo de máquina que será criada.
  sourceImage é onde definimos a versão do Container OS.
• natIP: é onde atribuímos o IP estático público que definimos anteriormente.
• metadata: a estrutura da chave é usada lá. Podemos adicionar parâmetros de configuração adicionais, como scripts de inicialização, mas neste caso específico, é aqui que configuramos a imagem do nosso serviço que será instalado e executado na máquina.
• serviceAccounts: permite configurar a conta de serviço associada à máquina e os escopos que ela terá para chamar outros serviços do Google Cloud.

- create_and_start_vm:

call: googleapis.compute.v1.instances.insert

args:

project: ${projectId}

zone: ${zone}

body:

name: ${instanceName}

machineType: ${"zones/" + zone + "/machineTypes/" + machineType}

disks:

- initializeParams:

sourceImage: "projects/cos-cloud/global/images/cos-stable-93-16623-102-1"

diskSizeGb: "10"

boot: true

autoDelete: true

networkInterfaces:

- accessConfigs:

- kind: compute#accessConfig

name: "external-nat"

networkTier: "PREMIUM"

metadata:

items:

- key: "gce-container-declaration"

value: '${"spec:\n  containers:\n  - name: scraper\n    image: gcr.io/" + projectId + "/scraper\n    stdin: false\n    tty: false\n  restartPolicy: Always\n"}'

- key: "google-logging-enabled"

value: "true"

- key: "google-monitoring-enabled"

value: "true"

serviceAccounts:

- email: "service-account-name@project-id.iam.gserviceaccount.com"

scopes:

- https://www.googleapis.com/auth/devstorage.read_only

- https://www.googleapis.com/auth/logging.write

- https://www.googleapis.com/auth/monitoring.write

- https://www.googleapis.com/auth/servicecontrol

- https://www.googleapis.com/auth/service.management.readonly

- https://www.googleapis.com/auth/trace.append

- https://www.googleapis.com/auth/cloud-platform

El resto de los pasos generan registros para monitorear el estado del workflow.

- log_wait_for_vm_network:

call: sys.log

args:

data: ${"Scraper logs:Waiting for VM network to initialize"}

- wait_for_vm_network:

call: sys.sleep

args:

seconds: 10

- log_vm_creation_end:

call: sys.log

args:

data: ${"Scraper logs:VM initialized successfully"}

Para implantar nosso fluxo de trabalho e realizar testes, usamos o seguinte comando:

gcloud workflows deploy <workflow_name> --source=<file-name>.yml --service-account=<service-account-email> --location=<region>

Por fim, podemos associar nosso workflow a um Cloud Scheduler que, basicamente, é uma tarefa cron configurada no Google Cloud para definir o cronograma de execução. Para isso, seguimos os seguintes passos:

• Abrimos nosso console do GCP.
• Procuramos workflows ou fluxos de trabalho.
• Selecionamos nosso workflow na lista.
• Vamos para a guia Ativadores e clique em Editar.
• Na página de edição, procuramos a seção Ativadores.
• Clicamos em Adicionar novo ativador.
• Selecionamos Cloud Scheduler.
• Damos um nome e definimos a hora usando a sintaxe para tarefas cron.
• Definimos o fuso horário.
• Vamos para a seção Configurar execução e selecionamos a conta de serviço.
• Clique em Criar.

Aqui está uma página onde você pode gerar sua frase cron.

E isso e tudo! Se você chegou até aqui, implementou uma arquitetura que permitirá executar tarefas de longa duração sem limites. Também integramos vários serviços do Google Cloud usando as bibliotecas de cliente para Python.

Saudações.