Escalonar automaticamente as cargas de trabalho de consumidores do Kafka

Neste tutorial, mostramos como configurar e implantar um escalonador automático do Kafka como um serviço do Cloud Run. Esse escalonador automático realiza a lógica de escalonamento para uma carga de trabalho de consumidor do Kafka, como uma implantação de pool de workers do Cloud Run. O escalonador automático do Kafka lê métricas do seu cluster do Kafka e usa o escalonamento manual para um pool de trabalhadores ou serviço do Cloud Run para escalonar uma carga de trabalho de consumidor do Kafka com base na métrica de atraso do consumidor do Kafka.

O diagrama a seguir mostra como um serviço de escalonador automático do Kafka lê métricas de um cluster do Kafka para escalonar automaticamente um pool de workers consumidores do Kafka.

Funções exigidas

Para receber as permissões necessárias para implantar e executar esse serviço, peça ao administrador para conceder a você os seguintes papéis do IAM:

• Desenvolvedor do Cloud Run (roles/run.developer)
• Usuário da conta de serviço (roles/iam.serviceAccountUser)
• Leitor do Artifact Registry (roles/artifactregistry.reader)
• Administrador do Cloud Scheduler (roles/cloudscheduler.admin) para criar o job do Cloud Scheduler que aciona as verificações de escalonamento automático
• Administrador de filas do Cloud Tasks (roles/cloudtasks.queueAdmin) para criar a fila do Cloud Tasks para verificações de escalonamento automático
• Administrador de segurança (roles/iam.securityAdmin) para conceder permissões a contas de serviço

Antes de começar

Para configurar e usar o escalonador automático do Kafka, você precisa dos seguintes recursos:

• Cluster Kafka
• Consumidor implantado

Cluster Kafka

• Um cluster do Kafka precisa estar em execução no Compute Engine, no Google Kubernetes Engine ou no Serviço gerenciado para Apache Kafka.
• Um tópico do Kafka configurado com eventos que estão sendo publicados nele.

Consumidor do Cloud Run implantado

• Uma carga de trabalho de consumidor do Kafka precisa ser implantada no Cloud Run como um serviço ou um pool de workers. Ele precisa ser configurado para se conectar ao cluster, tópico e grupo de consumidores do Kafka. Para um exemplo de consumidor do Kafka, consulte Exemplo de consumidor do escalonador automático do Kafka no Cloud Run.
• Sua carga de trabalho do consumidor precisa estar no mesmo projeto Google Cloud do cluster do Kafka.

Práticas recomendadas

• Conecte seus consumidores do Kafka à rede VPC usando a VPC direta. Com a VPC direta, é possível se conectar ao cluster do Kafka usando endereços IP particulares e manter o tráfego na rede VPC.
• Configure uma verificação de integridade de atividade para seus consumidores do Kafka que verifique se o consumidor está extraindo eventos. A verificação de integridade ajuda a garantir que as instâncias não íntegras sejam reiniciadas automaticamente se pararem de processar eventos, mesmo que o contêiner não falhe.

Criar o escalonador automático do Kafka

Use o Cloud Build para criar uma imagem de contêiner do escalonador automático do Kafka com base no código-fonte.

1. Clone o repositório:

   git clone https://github.com/GoogleCloudPlatform/cloud-run-kafka-scaler.git
   

2. Navegue até a pasta do repositório:

   cd cloud-run-kafka-scaler
   

Para especificar o nome da imagem de saída, atualize %ARTIFACT_REGISTRY_IMAGE% no arquivo cloudbuild.yaml incluído. Por exemplo: us-central1-docker.pkg.dev/my-project/my-repo/my_kafka_autoscaler.

gcloud builds submit --tag us-central1-docker.pkg.dev/my-project/my-repo/my_kafka_autoscaler

Esse comando cria a imagem do contêiner e a envia para o Artifact Registry. Registre o caminho completo da imagem (SCALER_IMAGE_PATH), porque você vai precisar dele mais tarde.

A imagem resultante não será executada localmente. Ele foi projetado para ser colocado sobre uma imagem de base Java. Para mais informações, incluindo como remontar a imagem do contêiner para execução local, consulte Configurar atualizações automáticas da imagem de base.

Definir a configuração do escalonador automático do Kafka

É possível configurar o escalonador automático do Kafka usando secrets. O escalonador automático atualiza a configuração periodicamente. Isso significa que você pode enviar novas versões de chaves secretas para mudar a configuração sem precisar fazer uma nova implantação do escalonador automático.

Configurar propriedades do cliente Kafka

É possível configurar a conexão com a API Admin do Kafka montando um secret como um volume ao implantar o escalonador automático do Kafka.

Crie um arquivo chamado kafka_client_config.txt e inclua as propriedades de configuração do cliente administrador do Kafka que você quer adicionar. A propriedade bootstrap.servers é obrigatória:

bootstrap.servers=BOOTSTRAP_SERVER_LIST

Substitua BOOTSTRAP_SERVER_LIST pela lista HOST:PORT do cluster do Kafka.

Configurar a autenticação do Kafka

Se o servidor Kafka exigir autenticação, inclua as propriedades de configuração necessárias no arquivo kafka_client_config.txt. Por exemplo, para se conectar a um cluster do Managed Service para Apache Kafka usando credenciais padrão do aplicativo com o Google OAuth, esse secret precisa incluir as seguintes propriedades:

bootstrap.servers=BOOTSTRAP_SERVER_LIST
security.protocol=SASL_SSL
sasl.mechanism=OAUTHBEARER
sasl.login.callback.handler.class=com.google.cloud.hosted.kafka.auth.GcpLoginCallbackHandler
sasl.jaas.config=org.apache.kafka.common.security.oauthbearer.OAuthBearerLoginModule required;

Substitua BOOTSTRAP_SERVER_LIST pela lista HOST:PORT do cluster do Kafka.

Usar credenciais padrão do aplicativo com um cluster do Serviço gerenciado para Apache Kafka também exige conceder o papel Cliente do Kafka gerenciado (roles/managedkafka.client) à conta de serviço do escalonador automático do Kafka:

gcloud projects add-iam-policy-binding PROJECT_ID \
--member="serviceAccount:SCALER_SERVICE_ACCOUNT@PROJECT_ID.iam.gserviceaccount.com" \
--role="roles/managedkafka.client"

Substitua:

• SCALER_SERVICE_ACCOUNT: o nome da conta de serviço do escalonador automático do Kafka.
• PROJECT_ID: o ID do projeto para o serviço de escalonamento automático do Kafka.

Para criar o secret, que será montado como um volume na implantação, use o arquivo kafka_client_config.txt:

gcloud secrets create ADMIN_CLIENT_SECRET_NAME --data-file=kafka_client_config.txt

Substitua ADMIN_CLIENT_SECRET_NAME pelo nome do secret de autenticação do Kafka.

Configurar o escalonamento

O escalonador automático do Kafka lê a configuração de escalonamento do volume /scaler-config/scaling. O conteúdo desse volume precisa ser formatado como YAML. Recomendamos montar um volume secreto para essa configuração.

Crie um arquivo chamado scaling_config.yaml com a seguinte configuração:

spec:
  scaleTargetRef:
    name: projects/PROJECT_ID/locations/REGION/workerpools/CONSUMER_SERVICE_NAME
 metrics:
  - type: Resource
    resource:
      name: cpu
      target:
        type: Utilization
        averageUtilization: TARGET_CPU_UTILIZATION
        activationThreshold: CPU_ACTIVATION_THRESHOLD
        tolerance: CPU_TOLERANCE
        windowSeconds: CPU_METRIC_WINDOW
  - type: External
    external:
      metric:
        name: consumer_lag
      target:
        type: AverageValue
        averageValue: LAG_THRESHOLD
        activationThreshold: LAG_ACTIVATION_THRESHOLD
        tolerance: LAG_TOLERANCE

Substitua:

• PROJECT_ID: o ID do projeto da carga de trabalho do consumidor do Kafka que será escalonada automaticamente.
• REGION: a região da carga de trabalho do consumidor do Kafka que será escalonada automaticamente.
• CONSUMER_SERVICE_NAME: o nome da carga de trabalho do consumidor do Kafka a ser escalonada automaticamente.
• TARGET_CPU_UTILIZATION: a meta de uso da CPU para cálculos de escalonamento automático. Por exemplo: 60.
• LAG_THRESHOLD: o limite da métrica consumer_lag para acionar o escalonamento automático. Por exemplo: 1000.
• (Opcional) CPU_ACTIVATION_THRESHOLD: o limite de ativação da CPU. Quando todas as métricas estão inativas, o consumidor de destino é dimensionado para zero. O padrão é 0.
• (Opcional) CPU_TOLERANCE: um limite que impede mudanças de escalonamento se estiver dentro do intervalo especificado. Expressa como uma porcentagem do uso pretendido de CPU. O padrão é 0.1.
• (Opcional) CPU_METRIC_WINDOW: um período de tempo, em segundos, em que o uso médio da CPU é calculado. O padrão é 120.
• (Opcional) LAG_ACTIVATION_THRESHOLD: o limite de ativação da métrica consumer_lag. Quando todas as métricas estão inativas, o consumidor de destino é dimensionado para zero. O padrão é 0.
• (Opcional) LAG_TOLERANCE: um limite que impede mudanças de escalonamento se estiver dentro do intervalo especificado. Expressa como uma porcentagem do atraso do consumidor desejado. O padrão é 0.1.

Se quiser, configure propriedades de escalonamento avançadas usando um bloco behavior:. Esse bloco é compatível com muitas das mesmas propriedades das políticas de escalonamento do HPA do Kubernetes.

Se você não especificar um bloco behavior, a seguinte configuração padrão será usada:

behavior:
  scaleDown:
    stabilizationWindowSeconds: 300
    policies:
    - type: Percent
      value: 50
      periodSeconds: 30
    selectPolicy: Min
  scaleUp:
    stabilizationWindowSeconds: 0
    policies:
    - type: Percent
      value: 100
      periodSeconds: 15
    - type: Instances
      value: 4
      periodSeconds: 15
    selectPolicy: Max

Para criar o volume de secret, que será montado em deployment, copie a configuração em um arquivo chamado scaling_config.yaml e execute o seguinte:

gcloud secrets create SCALING_CONFIG_SECRET_NAME --data-file=scaling_config.yaml

Substitua SCALING_CONFIG_SECRET_NAME pelo nome do secret de escalonamento automático.

Implantar o escalonador automático do Kafka

Depois de concluir os pré-requisitos, você poderá implantar o serviço de escalonamento automático do Kafka e a infraestrutura de suporte dele. Um módulo do Terraform e um script de shell são fornecidos para simplificar esse processo.

gcloud iam service-accounts create SCALER_SERVICE_ACCOUNT

gcloud iam service-accounts add-iam-policy-binding CONSUMER_SERVICE_ACCOUNT_EMAIL \
    --member="serviceAccount:SCALER_SERVICE_ACCOUNT@PROJECT_ID.iam.gserviceaccount.com" \
    --role="roles/iam.serviceAccountUser"

gcloud iam service-accounts add-iam-policy-binding CONSUMER_IMAGE_REPO \
    --member="serviceAccount:SCALER_SERVICE_ACCOUNT@PROJECT_ID.iam.gserviceaccount.com" \
    --role="roles/artifactregistry.reader" \
    --location=REPO_REGION

gcloud projects add-iam-policy-binding PROJECT_ID \
    --member="serviceAccount:SCALER_SERVICE_ACCOUNT@PROJECT_ID.iam.gserviceaccount.com" \
    --role="roles/run.admin"

gcloud secrets add-iam-policy-binding ADMIN_CLIENT_SECRET_NAME \
  --member="serviceAccount:SCALER_SERVICE_ACCOUNT@PROJECT_ID.iam.gserviceaccount.com" \
  --role="roles/secretmanager.secretAccessor"

gcloud secrets add-iam-policy-binding SCALING_CONFIG_SECRET_NAME \
  --member="serviceAccount:SCALER_SERVICE_ACCOUNT@PROJECT_ID.iam.gserviceaccount.com" \
  --role="roles/secretmanager.secretAccessor"

gcloud projects add-iam-policy-binding PROJECT_ID \
    --member="serviceAccount:SCALER_SERVICE_ACCOUNT@PROJECT_ID.iam.gserviceaccount.com" \
    --role="roles/monitoring.viewer" \
    --condition=None

gcloud projects add-iam-policy-binding PROJECT_ID \
    --member="serviceAccount:SCALER_SERVICE_ACCOUNT@PROJECT_ID.iam.gserviceaccount.com" \
    --role="roles/monitoring.metricWriter" \
    --condition=None

gcloud tasks queues create CLOUD_TASKS_QUEUE_NAME \
--location=REGION

gcloud iam service-accounts create TASKS_SERVICE_ACCOUNT

gcloud run services add-iam-policy-binding SCALER_SERVICE_NAME \
    --member="serviceAccount:TASKS_SERVICE_ACCOUNT@PROJECT_ID.iam.gserviceaccount.com" \
    --role="roles/run.invoker"

gcloud iam service-accounts create SCALER_SERVICE_ACCOUNT

gcloud iam service-accounts add-iam-policy-binding CONSUMER_SERVICE_ACCOUNT_EMAIL \
    --member="serviceAccount:SCALER_SERVICE_ACCOUNT@PROJECT_ID.iam.gserviceaccount.com" \
    --role="roles/iam.serviceAccountUser"

gcloud iam service-accounts add-iam-policy-binding CONSUMER_IMAGE_REPO \
    --member="serviceAccount:SCALER_SERVICE_ACCOUNT@PROJECT_ID.iam.gserviceaccount.com" \
    --role="roles/artifactregistry.reader" \
    --location=REPO_REGION

gcloud projects add-iam-policy-binding PROJECT_ID \
    --member="serviceAccount:SCALER_SERVICE_ACCOUNT@PROJECT_ID.iam.gserviceaccount.com" \
    --role="roles/run.admin"

gcloud secrets add-iam-policy-binding ADMIN_CLIENT_SECRET_NAME \
  --member="serviceAccount:SCALER_SERVICE_ACCOUNT@PROJECT_ID.iam.gserviceaccount.com" \
  --role="roles/secretmanager.secretAccessor"

gcloud secrets add-iam-policy-binding SCALING_CONFIG_SECRET_NAME \
  --member="serviceAccount:SCALER_SERVICE_ACCOUNT@PROJECT_ID.iam.gserviceaccount.com" \
  --role="roles/secretmanager.secretAccessor"

gcloud projects add-iam-policy-binding PROJECT_ID \
    --member="serviceAccount:SCALER_SERVICE_ACCOUNT@PROJECT_ID.iam.gserviceaccount.com" \
    --role="roles/monitoring.viewer" \
    --condition=None

gcloud projects add-iam-policy-binding PROJECT_ID \
    --member="serviceAccount:SCALER_SERVICE_ACCOUNT@PROJECT_ID.iam.gserviceaccount.com" \
    --role="roles/monitoring.metricWriter" \
    --condition=None

gcloud tasks queues add-iam-policy-binding CLOUD_TASKS_QUEUE_NAME \
    --member="serviceAccount:SCALER_SERVICE_ACCOUNT@PROJECT_ID.iam.gserviceaccount.com" \
    --role="roles/cloudtasks.enqueuer" \
    --location=REGION

KAFKA_TOPIC_ID: KAFKA_TOPIC_ID
CONSUMER_GROUP_ID: CONSUMER_GROUP_ID
CYCLE_SECONDS: CYCLE_SECONDS
OUTPUT_SCALER_METRICS: OUTPUT_SCALER_METRICS

KAFKA_TOPIC_ID: KAFKA_TOPIC_ID
CONSUMER_GROUP_ID: CONSUMER_GROUP_ID
CYCLE_SECONDS: CYCLE_SECONDS
OUTPUT_SCALER_METRICS: OUTPUT_SCALER_METRICS
FULLY_QUALIFIED_CLOUD_TASKS_QUEUE_NAME: CLOUD_TASKS_QUEUE_NAME
INVOKER_SERVICE_ACCOUNT_EMAIL: TASKS_SERVICE_ACCOUNT_EMAIL

gcloud run deploy SCALER_SERVICE_NAME \
    --image=SCALER_IMAGE_URI \
    --env-vars-file=scaler_env_vars.yaml \
    --service-account=SCALER_SERVICE_ACCOUNT@PROJECT_ID.iam.gserviceaccount.com \
    --no-allow-unauthenticated \
    --network=KAFKA_VPC_NETWORK \
    --subnet=KAFKA_VPC_SUBNET \
    --update-secrets=/kafka-config/kafka-client-properties=ADMIN_CLIENT_SECRET_NAME:latest \
    --update-secrets=/scaler-config/scaling=SCALING_CONFIG_SECRET_NAME:latest
    --labels=created-by=kafka-autoscaler

gcloud iam service-accounts create SCALER_INVOKER_SERVICE_ACCOUNT

gcloud run services add-iam-policy-binding SCALER_SERVICE_NAME \
  --member="serviceAccount:SCALER_INVOKER_SERVICE_ACCOUNT@PROJECT_ID.iam.gserviceaccount.com" \
  --role="roles/run.invoker"

gcloud scheduler jobs create http kafka-scaling-check \
    --location=REGION \
    --schedule="CRON_SCHEDULE" \
    --time-zone="TIMEZONE" \
    --uri=https://SCALER_SERVICE_NAME-PROJECT_NUMBER.REGION.run.app \
    --oidc-service-account-email=SCALER_INVOKER_SERVICE_ACCOUNT@PROJECT_ID.iam.gserviceaccount.com \
    --http-method=POST

gcloud scheduler jobs create http kafka-scaling-check \
    --location=REGION \
    --schedule="* * * * *" \
    --time-zone="TIMEZONE" \
    --uri=https://SCALER_SERVICE_NAME-PROJECT_NUMBER.REGION.run.app \
    --oidc-service-account-email=SCALER_INVOKER_SERVICE_ACCOUNT@PROJECT_ID.iam.gserviceaccount.com \
    --http-method=POST

# Details for already-deployed Kafka consumer
export PROJECT_ID=PROJECT_ID
export REGION=REGION
export CONSUMER_SERVICE_NAME=DEPLOYED_KAFKA_CONSUMER
export CONSUMER_SA_EMAIL=KAFKA_CONSUMER_ACCOUNT_EMAIL # For example, NAME@PROJECT_ID.iam.gserviceaccount.com
export TOPIC_ID=KAFKA_TOPIC_ID
export CONSUMER_GROUP_ID=KAFKA_CONSUMER_GROUP_ID
export NETWORK=VPC_NETWORK
export SUBNET=VPC_SUBNET

# Details for new items to be created during this setup
export CLOUD_TASKS_QUEUE_NAME=CLOUD_TASKS_QUEUE_FOR_SCALING_CHECKS
export TASKS_SERVICE_ACCOUNT=TASKS_SERVICE_ACCOUNT_NAME

export SCALER_SERVICE_NAME=KAFKA_AUTOSCALER_SERVICE_NAME
export SCALER_IMAGE_PATH=KAFKA_AUTOSCALER_IMAGE_URI
export SCALER_CONFIG_SECRET=KAFKA_AUTOSCALER_CONFIG_SECRET_NAME

export CYCLE_SECONDS=SCALER_CHECK_FREQUENCY # For example, 15; this value should be at least 5 seconds.

export OUTPUT_SCALER_METRICS=false # If you want scaling metrics to outputted to Cloud Monitoring set this to true and ensure your scaler service account has permission to write metrics (for example, via roles/monitoring.metricWriter).

./setup_kafka_scaler.sh

gcloud projects add-iam-policy-binding $PROJECT_ID \
  --member="serviceAccount:$SCALER_SA_NAME@$PROJECT_ID.iam.gserviceaccount.com" \
  --role="roles/artifactregistry.reader" # Or appropriate role for your registry

Verificar se o escalonamento automático do Kafka está funcionando

O escalonamento automático do serviço Kafka é acionado com uma solicitação para o URL do serviço (SCALER_SERVICE_NAME-PROJECT_NUMBER.REGION.run.app).

É possível enviar uma solicitação POST ao serviço de escalonamento automático do Kafka para acionar o cálculo do escalonamento automático:

curl -X POST -H "Authorization: Bearer $(gcloud auth print-identity-token)" https://SCALER_SERVICE_NAME-PROJECT_NUMBER.REGION.run.app

Substitua:

• SCALER_SERVICE_NAME: o nome do seu serviço de escalonador automático do Kafka.
• PROJECT_NUMBER: o número do projeto do serviço de escalonador automático do Kafka.
• REGION: o local do serviço de escalonador automático do Kafka.

As solicitações POST acionam o cálculo do escalonamento automático, a saída para o registro em registros, e mudam a contagem de instâncias com base na recomendação.

Os registros do seu serviço de escalonamento automático do Kafka precisam incluir mensagens como [SCALING] Recommended instances X.

Se a flag OUTPUT_SCALER_METRICS estiver ativada, você também poderá encontrar métricas de escalonador do Cloud Monitoring em custom.googleapis.com/cloud-run-kafkascaler.

Configuração avançada de escalonamento

spec:
  metrics:
  behavior:
    scaleDown:
      stabilizationWindowSeconds: [INT]
      policies:
      - type: [Percent, Instances]
        value: [INT]
        periodSeconds: [INT]
      selectPolicy: [Min, Max]
    scaleUp:
      stabilizationWindowSeconds: [INT]
      policies:
      - type: [Percent, Instances]
        value: [INT]
        periodSeconds: [INT]
      selectPolicy: [Min, Max]

A lista a seguir descreve alguns dos elementos anteriores:

• scaleDown: o comportamento ao reduzir a contagem de instâncias (redução de escalonamento).
• scaleUp: o comportamento ao aumentar a contagem de instâncias (escalonamento vertical).
• stabilizationWindowSeconds: o número mais alto (scaleDown) ou mais baixo (scaleUp) de instâncias calculado em um período contínuo. Definir o valor como 0 significa que o valor calculado mais recente será usado.
• selectPolicy: o resultado a ser aplicado quando várias políticas são configuradas.
• Min: a menor mudança
• Max: a maior mudança
• Percent: as mudanças por período são limitadas à porcentagem configurada do total de instâncias.
• Instances: as mudanças por período são limitadas ao número configurado de instâncias.
• periodSeconds: o período em que a política é aplicada.

Por exemplo, a especificação completa, usando a configuração padrão, é assim:

spec:
  scaleTargetRef:
    name: projects/PROJECT-ID/locations/us-central1/workerpools/kafka-consumer-worker
  metrics:
    - type: Resource
      resource:
        name: cpu
        target:
          type: Utilization
          averageUtilization: 60
          activationThreshold: 0
          tolerance: 0.1
          windowSeconds: 120
    - type: External
      external:
        metric:
          name: consumer_lag
        target:
          type: AverageValue
          averageValue: 1000
          activationThreshold: 0
          tolerance: 0.1
  behavior:
    scaleDown:
      stabilizationWindowSeconds: 300
      policies:
        - type: Percent
          value: 50
          periodSeconds: 30
      selectPolicy: Min
    scaleUp:
      stabilizationWindowSeconds: 0
      policies:
        - type: Percent
          value: 100
          periodSeconds: 15
        - type: Instances
          value: 4
          periodSeconds: 15
      selectPolicy: Max