Coletar registros do Metabase

Compatível com:

Neste documento, explicamos como ingerir registros do Metabase no Google Security Operations usando o Google Cloud Storage V2.

O Metabase é uma plataforma de análise e Business Intelligence de código aberto que permite que as organizações consultem bancos de dados, criem visualizações e criem painéis. Nos planos Pro e Enterprise, o Metabase armazena análises de uso e dados de auditoria no banco de dados do aplicativo, incluindo registros de atividades, registros de visualização e registros de execução de consultas. Uma função do Cloud Run consulta a API Metabase para recuperar esses dados e os grava em um bucket do GCS para ingestão pelo Google SecOps.

Antes de começar

Verifique se você atende os seguintes pré-requisitos:

  • Uma instância do Google SecOps
  • Um projeto do GCP com as APIs Cloud Storage, Cloud Run, Pub/Sub e API Cloud Scheduler ativadas
  • Permissões para criar e gerenciar buckets do GCS
  • Permissões para gerenciar políticas do IAM em buckets do GCS
  • Permissões para criar serviços do Cloud Run, tópicos do Pub/Sub e jobs do Cloud Scheduler
  • Plano Metabase Pro ou Enterprise (as análises de uso não estão disponíveis nas edições Open Source ou Cloud Starter)
  • Acesso de administrador do Metabase para criar chaves de API
  • O banco de dados do aplicativo Metabase adicionado como uma fonte de dados no Metabase (necessário para consultar tabelas de auditoria via API)

Configurar o acesso à API do Metabase

Para permitir que o Google SecOps recupere dados de análise de uso, crie uma chave de API no Metabase e adicione o banco de dados do aplicativo como uma fonte de dados consultável.

Criar uma chave de API

  1. Faça login na sua instância do Metabase como administrador.
  2. Clique no ícone de engrenagem no canto superior direito.
  3. Selecione Configurações de administrador.
  4. Acesse a guia Configurações.
  5. Clique na guia Autenticação no menu à esquerda.
  6. Role até Chaves de API e clique em Gerenciar.
  7. Clique no botão Criar chave de API.
  8. Informe os seguintes detalhes de configuração:
    • Nome da chave: insira um nome descritivo (por exemplo, Google SecOps Integration).
    • Grupo: selecione um grupo com acesso à coleção do Usage Analytics e permissões de consulta para o banco de dados do aplicativo.
  9. Clique em Criar.
  10. Copie e salve a chave de API com segurança.

Adicionar o banco de dados do aplicativo como uma fonte de dados

Para consultar tabelas de auditoria usando a API, adicione o banco de dados do aplicativo Metabase como uma fonte de dados no Metabase.

  1. No Metabase, clique no ícone de engrenagem no canto superior direito.
  2. Selecione Configurações de administrador.
  3. Acesse a guia Bancos de dados.
  4. Clique em Adicionar um banco de dados.
  5. Configure a conexão do banco de dados:
    • Tipo de banco de dados: selecione PostgreSQL ou MySQL (dependendo do banco de dados do aplicativo)
    • Nome: insira Metabase Application Database.
    • Host: insira o nome do host do banco de dados do aplicativo.
    • Porta: insira a porta do banco de dados (padrão: 5432 para PostgreSQL, 3306 para MySQL).
    • Nome do banco de dados: insira o nome do banco de dados do aplicativo (por exemplo, metabase).
    • Nome de usuário: insira um usuário do banco de dados com permissões SELECT nas tabelas audit_log, view_log e query_execution.
    • Senha: digite a senha do banco de dados.
  6. Clique em Salvar.
  7. Depois de salvar, anote o ID do banco de dados no URL (por exemplo, https://metabase.example.com/admin/databases/5 significa que o ID do banco de dados é 5).

Verifique as permissões

Para verificar se a chave de API tem as permissões necessárias:

  1. Faça login no Metabase como administrador.
  2. Clique no ícone de engrenagem > Configurações de administrador > Pessoas.
  3. Encontre o grupo atribuído à chave de API.
  4. Verifique se o grupo tem acesso à coleção Análise de uso.
  5. Acesse Configurações de administrador > Permissões > Dados.
  6. Verifique se o grupo tem acesso Irrestrito ou de Consulta ao banco de dados do aplicativo.

Testar o acesso à API

  • Teste suas credenciais antes de prosseguir com a integração:

    # Replace with your actual values
    METABASE_URL="https://your-metabase-instance.com"
    API_KEY="mb_your_api_key_here"
    DATABASE_ID="5"
    
    # Test API key authentication
    curl -s -H "x-api-key: ${API_KEY}" \
      "${METABASE_URL}/api/database" \
      | python3 -m json.tool
    
    # Test querying the application database for audit_log records
    curl -s -X POST "${METABASE_URL}/api/dataset/json" \
      -H "x-api-key: ${API_KEY}" \
      -H "Content-Type: application/json" \
      -d "{\"database\": ${DATABASE_ID}, \"type\": \"native\", \"native\": {\"query\": \"SELECT * FROM audit_log ORDER BY id DESC LIMIT 5\", \"template-tags\": {}}, \"parameters\": []}" \
      | python3 -m json.tool
    

Uma resposta bem-sucedida retorna uma matriz JSON de registros de auditoria recentes.

Criar um bucket do Google Cloud Storage

  1. Acesse o Console do Google Cloud.
  2. Selecione seu projeto ou crie um novo.
  3. No menu de navegação, acesse Cloud Storage > Buckets.
  4. Clique em Criar bucket.
  5. Informe os seguintes detalhes de configuração:

    Configuração Valor
    Nomeie seu bucket Insira um nome globalmente exclusivo (por exemplo, metabase-audit-logs)
    Tipo de local Escolha de acordo com suas necessidades (região, birregional, multirregional)
    Local Selecione o local (por exemplo, us-central1).
    Classe de armazenamento Padrão (recomendado para registros acessados com frequência)
    Controle de acesso Uniforme (recomendado)
    Ferramentas de proteção Opcional: ativar o controle de versões de objetos ou a política de retenção
  6. Clique em Criar.

Criar uma conta de serviço para a função do Cloud Run

A função do Cloud Run precisa de uma conta de serviço com permissões para gravar no bucket do GCS e ser invocada pelo Pub/Sub.

Criar conta de serviço

  1. No Console do GCP, acesse IAM e administrador > Contas de serviço.
  2. Clique em Criar conta de serviço.
  3. Informe os seguintes detalhes de configuração:
    • Nome da conta de serviço: insira metabase-audit-collector-sa.
    • Descrição da conta de serviço: insira Service account for Cloud Run function to collect Metabase audit logs.
  4. Clique em Criar e continuar.
  5. Na seção Conceder acesso a essa conta de serviço ao projeto, adicione os seguintes papéis:
    1. Clique em Selecionar papel.
    2. Pesquise e selecione Administrador de objetos do Storage.
    3. Clique em + Adicionar outro papel.
    4. Pesquise e selecione Invocador do Cloud Run.
    5. Clique em + Adicionar outro papel.
    6. Pesquise e selecione Invocador do Cloud Functions.
  6. Clique em Continuar.
  7. Clique em Concluído.

Esses papéis são necessários para:

  • Administrador de objetos do Storage: gravar registros no bucket do GCS e gerenciar arquivos de estado
  • Invocador do Cloud Run: permite que o Pub/Sub invoque a função.
  • Invocador do Cloud Functions: permite a invocação de funções.

Conceder permissões do IAM no bucket do GCS

  1. Acesse Cloud Storage > Buckets.
  2. Clique no nome do bucket (metabase-audit-logs).
  3. Acesse a guia Permissões.
  4. Clique em Conceder acesso.
  5. Informe os seguintes detalhes de configuração:
    • Adicionar principais: insira o e-mail da conta de serviço (metabase-audit-collector-sa@PROJECT_ID.iam.gserviceaccount.com)
    • Atribuir papéis: selecione Administrador de objetos do Storage.
  6. Clique em Salvar.

Criar tópico Pub/Sub

  1. No Console do GCP, acesse Pub/Sub > Tópicos.
  2. Selecione Criar tópico.
  3. Informe os seguintes detalhes de configuração:
    • ID do tópico: insira metabase-audit-trigger.
    • Não mude as outras configurações.
  4. Clique em Criar.

Criar uma função do Cloud Run para coletar registros

A função do Cloud Run será acionada por mensagens do Pub/Sub do Cloud Scheduler para buscar dados de auditoria da API Metabase e gravá-los no GCS.

  1. No Console do GCP, acesse o Cloud Run.
  2. Clique em Criar serviço.
  3. Selecione Função (use um editor in-line para criar uma função).
  4. Na seção Configurar, forneça os seguintes detalhes de configuração:

    Configuração Valor
    Nome do serviço metabase-audit-collector
    Região Selecione a região correspondente ao bucket do GCS (por exemplo, us-central1).
    Ambiente de execução Selecione Python 3.12 ou mais recente.
  5. Na seção Acionador (opcional):

    1. Clique em + Adicionar gatilho.
    2. Selecione Cloud Pub/Sub.
    3. Em Selecionar um tópico do Cloud Pub/Sub, escolha metabase-audit-trigger.
    4. Clique em Salvar.
  6. Na seção Autenticação:

    1. Selecione Exigir autenticação.
    2. Verifique o Identity and Access Management (IAM).
  7. Role para baixo e abra Contêineres, rede, segurança.

  8. Acesse a guia Segurança:

    • Conta de serviço: selecione metabase-audit-collector-sa.
  9. Acesse a guia Contêineres:

    1. Clique em Variáveis e secrets.
    2. Clique em + Adicionar variável para cada variável de ambiente:
    Nome da variável Valor de exemplo Descrição
    GCS_BUCKET metabase-audit-logs Nome do bucket do GCS
    GCS_PREFIX metabase-audit Prefixo para arquivos de registro
    STATE_KEY metabase-audit/state.json Caminho do arquivo de estado
    METABASE_URL https://metabase.yourcompany.com URL de base da instância do Metabase
    METABASE_API_KEY mb_your_api_key_here Chave de API do Metabase
    DATABASE_ID 5 ID do banco de dados do aplicativo
    LOOKBACK_HOURS 24 Período de lookback inicial
    PAGE_SIZE 2000 Registros por consulta de API
  10. Na seção Variáveis e secrets, role para baixo até Solicitações:

    • Tempo limite da solicitação: insira 600 segundos (10 minutos)
  11. Acesse a guia Configurações:

    • Na seção Recursos:
      • Memória: selecione 512 MiB ou mais.
      • CPU: selecione 1.
  12. Na seção Escalonamento de revisão:

    • Número mínimo de instâncias: insira 0.
    • Número máximo de instâncias: insira 100.
  13. Clique em Criar.

  14. Aguarde a criação do serviço (1 a 2 minutos).

  15. Depois que o serviço for criado, o editor de código inline será aberto automaticamente.

Adicionar código da função

  1. Insira main no campo Ponto de entrada.
  2. No editor de código inline, crie dois arquivos:

    • main.py:

      import functions_framework
      from google.cloud import storage
      import json
      import os
      import urllib3
      from datetime import datetime, timezone, timedelta
      
      http = urllib3.PoolManager(
        timeout=urllib3.Timeout(connect=10.0, read=60.0),
        retries=False,
      )
      
      storage_client = storage.Client()
      
      GCS_BUCKET = os.environ.get('GCS_BUCKET')
      GCS_PREFIX = os.environ.get('GCS_PREFIX', 'metabase-audit')
      STATE_KEY = os.environ.get('STATE_KEY', 'metabase-audit/state.json')
      METABASE_URL = os.environ.get('METABASE_URL', '').rstrip('/')
      API_KEY = os.environ.get('METABASE_API_KEY')
      DATABASE_ID = int(os.environ.get('DATABASE_ID', '1'))
      LOOKBACK_HOURS = int(os.environ.get('LOOKBACK_HOURS', '24'))
      PAGE_SIZE = int(os.environ.get('PAGE_SIZE', '2000'))
      
      TABLES = ['audit_log', 'view_log', 'query_execution']
      
      TABLE_TIME_COLUMNS = {
        'audit_log': 'timestamp',
        'view_log': 'timestamp',
        'query_execution': 'started_at',
      }
      
      @functions_framework.cloud_event
      def main(cloud_event):
        if not all([GCS_BUCKET, METABASE_URL, API_KEY]):
          print('Error: Missing required environment variables')
          return
      
        try:
          bucket = storage_client.bucket(GCS_BUCKET)
          state = load_state(bucket)
          now = datetime.now(timezone.utc)
      
          if isinstance(state, dict) and state.get('last_event_time'):
            try:
              last_val = state['last_event_time']
              if last_val.endswith('Z'):
                last_val = last_val[:-1] + '+00:00'
              last_time = datetime.fromisoformat(last_val)
              last_time = last_time - timedelta(minutes=2)
            except Exception as e:
              print(f"Warning: Could not parse last_event_time: {e}")
              last_time = now - timedelta(hours=LOOKBACK_HOURS)
          else:
            last_time = now - timedelta(hours=LOOKBACK_HOURS)
      
          print(f"Fetching logs from {last_time.isoformat()} to {now.isoformat()}")
      
          all_records = []
          newest_time = None
      
          for table in TABLES:
            time_col = TABLE_TIME_COLUMNS[table]
            records = fetch_table(table, time_col, last_time, now)
            for r in records:
              r['_metabase_table'] = table
              t = r.get(time_col)
              if t and (newest_time is None or str(t) > str(newest_time)):
                newest_time = t
            all_records.extend(records)
            print(f"Table {table}: fetched {len(records)} records")
      
          if not all_records:
            print("No new records found.")
            save_state(bucket, now.isoformat())
            return
      
          timestamp = now.strftime('%Y%m%d_%H%M%S')
          object_key = f"{GCS_PREFIX}/metabase_audit_{timestamp}.ndjson"
          blob = bucket.blob(object_key)
      
          ndjson = '\n'.join(
            [json.dumps(r, ensure_ascii=False, default=str) for r in all_records]
          ) + '\n'
          blob.upload_from_string(ndjson, content_type='application/x-ndjson')
      
          print(f"Wrote {len(all_records)} records to gs://{GCS_BUCKET}/{object_key}")
      
          save_state(bucket, str(newest_time) if newest_time else now.isoformat())
      
          print(f"Successfully processed {len(all_records)} records")
      
        except Exception as e:
          print(f'Error processing logs: {str(e)}')
          raise
      
      def fetch_table(table, time_col, start_time, end_time):
        start_str = start_time.strftime('%Y-%m-%d %H:%M:%S')
        end_str = end_time.strftime('%Y-%m-%d %H:%M:%S')
      
        all_records = []
        offset = 0
        max_pages = 50
      
        for page in range(max_pages):
          sql = (
            f"SELECT * FROM {table} "
            f"WHERE {time_col} >= '{start_str}' "
            f"AND {time_col} < '{end_str}' "
            f"ORDER BY {time_col} ASC "
            f"LIMIT {PAGE_SIZE} OFFSET {offset}"
          )
      
          query_body = {
            "database": DATABASE_ID,
            "type": "native",
            "native": {
              "query": sql,
              "template-tags": {}
            },
            "parameters": []
          }
      
          url = f"{METABASE_URL}/api/dataset/json"
          response = http.request(
            'POST', url,
            body=json.dumps(query_body).encode('utf-8'),
            headers={
              'x-api-key': API_KEY,
              'Content-Type': 'application/json'
            }
          )
      
          if response.status == 429:
            print(f"Rate limited on {table} query. Stopping pagination.")
            break
      
          if response.status != 200:
            print(f"{table} query failed: {response.status} - "
                f"{response.data.decode('utf-8')}")
            break
      
          page_results = json.loads(response.data.decode('utf-8'))
      
          if not page_results:
            break
      
          all_records.extend(page_results)
          print(f"{table} page {page + 1}: {len(page_results)} records "
              f"(total: {len(all_records)})")
      
          if len(page_results) < PAGE_SIZE:
            break
      
          offset += PAGE_SIZE
      
        return all_records
      
      def load_state(bucket):
        try:
          blob = bucket.blob(STATE_KEY)
          if blob.exists():
            return json.loads(blob.download_as_text())
        except Exception as e:
          print(f"Warning: Could not load state: {e}")
        return {}
      
      def save_state(bucket, last_event_time_iso):
        try:
          state = {
            'last_event_time': last_event_time_iso,
            'last_run': datetime.now(timezone.utc).isoformat()
          }
          blob = bucket.blob(STATE_KEY)
          blob.upload_from_string(
            json.dumps(state, indent=2),
            content_type='application/json'
          )
          print(f"Saved state: last_event_time={last_event_time_iso}")
        except Exception as e:
          print(f"Warning: Could not save state: {e}")
      
    • requirements.txt:

    functions-framework==3.*
    google-cloud-storage==2.*
    urllib3>=2.0.0
    
  3. Clique em Implantar para salvar e implantar a função.

  4. Aguarde a conclusão da implantação (2 a 3 minutos).

Criar o job do Cloud Scheduler

  1. No Console do GCP, acesse o Cloud Scheduler.
  2. Clique em Criar job.
  3. Informe os seguintes detalhes de configuração:

    Configuração Valor
    Nome metabase-audit-collector-hourly
    Região Selecione a mesma região da função do Cloud Run
    Frequência 0 * * * * (de hora em hora)
    Fuso horário Selecione o fuso horário (UTC recomendado)
    Tipo de destino Pub/Sub
    Tópico Selecionar metabase-audit-trigger
    Corpo da mensagem {} (objeto JSON vazio)
  4. Clique em Criar.

Opções de frequência de programação

Escolha a frequência com base no volume de registros e nos requisitos de latência:

Frequência Expressão Cron Caso de uso
A cada 5 minutos */5 * * * * Alto volume e baixa latência
A cada 15 minutos */15 * * * * Volume médio
A cada hora 0 * * * * Padrão (recomendado)
A cada 6 horas 0 */6 * * * Processamento em lote de baixo volume
Diário 0 0 * * * Coleta de dados históricos

Testar a integração

  1. No console do Cloud Scheduler, encontre seu job (metabase-audit-collector-hourly).
  2. Clique em Forçar execução para acionar o job manualmente.
  3. Aguarde alguns segundos.
  4. Acesse Cloud Run > Serviços.
  5. Clique em metabase-audit-collector.
  6. Clique na guia Registros.
  7. Verifique se a função foi executada com sucesso. Procure por:

    Fetching logs from YYYY-MM-DDTHH:MM:SS+00:00 to YYYY-MM-DDTHH:MM:SS+00:00
    Table audit_log: fetched X records
    Table view_log: fetched X records
    Table query_execution: fetched X records
    Wrote X records to gs://metabase-audit-logs/metabase-audit/metabase_audit_YYYYMMDD_HHMMSS.ndjson
    Successfully processed X records
    
  8. Acesse Cloud Storage > Buckets.

  9. Clique em metabase-audit-logs.

  10. Navegue até a pasta metabase-audit/.

  11. Verifique se um novo arquivo .ndjson foi criado com o carimbo de data/hora atual.

Se você encontrar erros nos registros:

  • HTTP 401: verifique se a variável de ambiente METABASE_API_KEY está correta e se a chave não foi excluída ou gerada novamente.
  • HTTP 403: verifique se o grupo da chave de API tem permissões de consulta para o banco de dados do aplicativo.
  • HTTP 429: limitação de taxa. A função vai interromper a paginação e retomar na próxima execução programada.
  • Variáveis de ambiente ausentes: verifique se todas as variáveis necessárias estão definidas na configuração da função do Cloud Run.
  • Resultados vazios: verifique se o DATABASE_ID está correto, se o banco de dados do aplicativo foi adicionado como uma fonte de dados e se você está usando o Metabase Pro ou Enterprise.

Recuperar a conta de serviço do Google SecOps

  1. Acesse Configurações > Gerenciamento de dados > Feeds. Se você estiver em uma plataforma somente SIEM, acesse Configurações > Feeds.
  2. Clique em Adicionar novo feed.
  3. Clique em Configurar um único feed.
  4. No campo Nome do feed, insira um nome para o feed (por exemplo, Metabase Audit Logs).
  5. Selecione Google Cloud Storage V2 como o Tipo de origem.
  6. Selecione METABASE como o Tipo de registro.
  7. Clique em Get Service Account. Um e-mail exclusivo da conta de serviço será exibido, por exemplo:

    chronicle-12345678@chronicle-gcp-prod.iam.gserviceaccount.com
    
  8. Copie esse endereço de e-mail para usar na próxima etapa.

  9. Clique em Próxima.

  10. Especifique valores para os seguintes parâmetros de entrada:

    • URL do bucket de armazenamento: insira o URI do bucket do GCS com o caminho do prefixo:

      gs://metabase-audit-logs/metabase-audit/
      
    • Opção de exclusão da fonte: selecione a opção de exclusão de acordo com sua preferência:

      • Nunca: nunca exclui arquivos após as transferências (recomendado para testes).
      • Excluir arquivos transferidos: exclui arquivos após a transferência.
      • Excluir arquivos transferidos e diretórios vazios: exclui arquivos e diretórios vazios após a transferência.

    • Idade máxima do arquivo: inclua arquivos modificados nos últimos dias (o padrão é 180 dias).

    • Namespace do recurso: o namespace do recurso

    • Rótulos de ingestão: o rótulo que será aplicado aos eventos desse feed.

  11. Clique em Próxima.

  12. Revise a configuração do novo feed na tela Finalizar e clique em Enviar.

Conceder permissões do IAM à conta de serviço do Google SecOps

A conta de serviço do Google SecOps precisa do papel Leitor de objetos do Storage no bucket do GCS.

  1. Acesse Cloud Storage > Buckets.
  2. Clique em metabase-audit-logs.
  3. Acesse a guia Permissões.
  4. Clique em Conceder acesso.
  5. Informe os seguintes detalhes de configuração:
    • Adicionar participantes: cole o e-mail da conta de serviço do Google SecOps.
    • Atribuir papéis: selecione Leitor de objetos do Storage.
  6. Clique em Salvar.

Tabelas de análise de uso

A função do Cloud Run consulta as seguintes tabelas do banco de dados do aplicativo Metabase:

Nome da tabela Descrição
audit_log Registros de mudanças de configuração, ações do usuário e eventos do sistema
view_log Rastreia visualizações de cards (perguntas/modelos), painéis e tabelas
query_execution Informações sobre todas as consultas executadas em todos os painéis

Retenção de dados

Por padrão, o Metabase retém os dados de análise de uso por 720 dias (aproximadamente 2 anos). Duas vezes por dia, o Metabase exclui automaticamente as linhas mais antigas que esse limite. Para mudar o período de armazenamento, defina a variável de ambiente MB_AUDIT_MAX_RETENTION_DAYS ao executar o Metabase.

Tabela de mapeamento de UDM

Campo de registro Mapeamento de UDM Lógica
adicional adicional Renomeado de "adicional"
locale additional.fields Mesclado com os respectivos rótulos (por exemplo, locale_label para locale, is_active_label para is_active etc.)
is_active additional.fields
is_qbnewb additional.fields
group_ids additional.fields
is_superuser additional.fields
login_attributes additional.fields
ID additional.fields
sso_source additional.fields
personal_collection_id additional.fields
metadados metadados Renomeado de metadados
updated_at metadata.event_timestamp Data analisada de updated_at usando ISO8601, RFC 3339 ou yyyy-MM-ddTHH:mm:ss.SSSSSSZ
has_userid metadata.event_type Defina como "USER_UNCATEGORIZED" se has_userid == "true", caso contrário, defina como "GENERIC_EVENT"
participante participante Renomeado do principal
metabase_host principal.url Valor copiado diretamente
date_joined principal.user.attribute.creation_time Data analisada de date_joined usando ISO8601, RFC 3339 ou yyyy-MM-ddTHH:mm:ss.SSSSSSZ
e-mail principal.user.email_addresses Mesclado do e-mail
first_name principal.user.first_name Valor copiado diretamente
last_login principal.user.last_login_time Data analisada de last_login usando ISO8601, RFC 3339 ou yyyy-MM-ddTHH:mm:ss.SSSSSSZ
last_name principal.user.last_name Valor copiado diretamente
common_name principal.user.user_display_name Valor copiado diretamente
e-mail principal.user.userid Valor copiado diretamente do e-mail
security_result security_result Mesclado de security_result
destino destino Renomeado do destino
metadata.product_name metadata.product_name Defina como "METABASE".
metadata.vendor_name metadata.vendor_name Defina como "METABASE".

Registro de alterações

Confira o registro de alterações deste analisador

Precisa de mais ajuda? Receba respostas de membros da comunidade e profissionais do Google SecOps.