Coletar registros do Metabase
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
- Faça login na sua instância do Metabase como administrador.
- Clique no ícone de engrenagem no canto superior direito.
- Selecione Configurações de administrador.
- Acesse a guia Configurações.
- Clique na guia Autenticação no menu à esquerda.
- Role até Chaves de API e clique em Gerenciar.
- Clique no botão Criar chave de API.
- 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.
- Nome da chave: insira um nome descritivo (por exemplo,
- Clique em Criar.
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.
- No Metabase, clique no ícone de engrenagem no canto superior direito.
- Selecione Configurações de administrador.
- Acesse a guia Bancos de dados.
- Clique em Adicionar um banco de dados.
- 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:
5432para PostgreSQL,3306para 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_logequery_execution. - Senha: digite a senha do banco de dados.
- Clique em Salvar.
- Depois de salvar, anote o ID do banco de dados no URL (por exemplo,
https://metabase.example.com/admin/databases/5significa 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:
- Faça login no Metabase como administrador.
- Clique no ícone de engrenagem > Configurações de administrador > Pessoas.
- Encontre o grupo atribuído à chave de API.
- Verifique se o grupo tem acesso à coleção Análise de uso.
- Acesse Configurações de administrador > Permissões > Dados.
- 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
- Acesse o Console do Google Cloud.
- Selecione seu projeto ou crie um novo.
- No menu de navegação, acesse Cloud Storage > Buckets.
- Clique em Criar bucket.
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 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
- No Console do GCP, acesse IAM e administrador > Contas de serviço.
- Clique em Criar conta de serviço.
- 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.
- Nome da conta de serviço: insira
- Clique em Criar e continuar.
- Na seção Conceder acesso a essa conta de serviço ao projeto, adicione os seguintes papéis:
- Clique em Selecionar papel.
- Pesquise e selecione Administrador de objetos do Storage.
- Clique em + Adicionar outro papel.
- Pesquise e selecione Invocador do Cloud Run.
- Clique em + Adicionar outro papel.
- Pesquise e selecione Invocador do Cloud Functions.
- Clique em Continuar.
- 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
- Acesse Cloud Storage > Buckets.
- Clique no nome do bucket (
metabase-audit-logs). - Acesse a guia Permissões.
- Clique em Conceder acesso.
- 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.
- Adicionar principais: insira o e-mail da conta de serviço (
- Clique em Salvar.
Criar tópico Pub/Sub
- No Console do GCP, acesse Pub/Sub > Tópicos.
- Selecione Criar tópico.
- Informe os seguintes detalhes de configuração:
- ID do tópico: insira
metabase-audit-trigger. - Não mude as outras configurações.
- ID do tópico: insira
- 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.
- No Console do GCP, acesse o Cloud Run.
- Clique em Criar serviço.
- Selecione Função (use um editor in-line para criar uma função).
Na seção Configurar, forneça os seguintes detalhes de configuração:
Configuração Valor Nome do serviço metabase-audit-collectorRegiã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. Na seção Acionador (opcional):
- Clique em + Adicionar gatilho.
- Selecione Cloud Pub/Sub.
- Em Selecionar um tópico do Cloud Pub/Sub, escolha
metabase-audit-trigger. - Clique em Salvar.
Na seção Autenticação:
- Selecione Exigir autenticação.
- Verifique o Identity and Access Management (IAM).
Role para baixo e abra Contêineres, rede, segurança.
Acesse a guia Segurança:
- Conta de serviço: selecione
metabase-audit-collector-sa.
- Conta de serviço: selecione
Acesse a guia Contêineres:
- Clique em Variáveis e secrets.
- Clique em + Adicionar variável para cada variável de ambiente:
Nome da variável Valor de exemplo Descrição GCS_BUCKETmetabase-audit-logsNome do bucket do GCS GCS_PREFIXmetabase-auditPrefixo para arquivos de registro STATE_KEYmetabase-audit/state.jsonCaminho do arquivo de estado METABASE_URLhttps://metabase.yourcompany.comURL de base da instância do Metabase METABASE_API_KEYmb_your_api_key_hereChave de API do Metabase DATABASE_ID5ID do banco de dados do aplicativo LOOKBACK_HOURS24Período de lookback inicial PAGE_SIZE2000Registros por consulta de API Na seção Variáveis e secrets, role para baixo até Solicitações:
- Tempo limite da solicitação: insira
600segundos (10 minutos)
- Tempo limite da solicitação: insira
Acesse a guia Configurações:
- Na seção Recursos:
- Memória: selecione 512 MiB ou mais.
- CPU: selecione 1.
- Na seção Recursos:
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.
- Número mínimo de instâncias: insira
Clique em Criar.
Aguarde a criação do serviço (1 a 2 minutos).
Depois que o serviço for criado, o editor de código inline será aberto automaticamente.
Adicionar código da função
- Insira main no campo Ponto de entrada.
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.0Clique em Implantar para salvar e implantar a função.
Aguarde a conclusão da implantação (2 a 3 minutos).
Criar o job do Cloud Scheduler
- No Console do GCP, acesse o Cloud Scheduler.
- Clique em Criar job.
Informe os seguintes detalhes de configuração:
Configuração Valor Nome metabase-audit-collector-hourlyRegiã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-triggerCorpo da mensagem {}(objeto JSON vazio)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
- No console do Cloud Scheduler, encontre seu job (
metabase-audit-collector-hourly). - Clique em Forçar execução para acionar o job manualmente.
- Aguarde alguns segundos.
- Acesse Cloud Run > Serviços.
- Clique em
metabase-audit-collector. - Clique na guia Registros.
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 recordsAcesse Cloud Storage > Buckets.
Clique em
metabase-audit-logs.Navegue até a pasta
metabase-audit/.Verifique se um novo arquivo
.ndjsonfoi 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_KEYestá 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_IDestá 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
- Acesse Configurações > Gerenciamento de dados > Feeds. Se você estiver em uma plataforma somente SIEM, acesse Configurações > Feeds.
- Clique em Adicionar novo feed.
- Clique em Configurar um único feed.
- No campo Nome do feed, insira um nome para o feed (por exemplo,
Metabase Audit Logs). - Selecione Google Cloud Storage V2 como o Tipo de origem.
- Selecione METABASE como o Tipo de registro.
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.comCopie esse endereço de e-mail para usar na próxima etapa.
Clique em Próxima.
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.
Clique em Próxima.
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.
- Acesse Cloud Storage > Buckets.
- Clique em
metabase-audit-logs. - Acesse a guia Permissões.
- Clique em Conceder acesso.
- 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.
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 |
| 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 |
| 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.