Raccogli i log di Metabase
Questo documento spiega come importare i log di Metabase in Google Security Operations utilizzando Google Cloud Storage V2.
Metabase è una piattaforma di business intelligence e analisi open source che consente alle organizzazioni di eseguire query sui database, creare visualizzazioni e creare dashboard. Nei piani Pro ed Enterprise, Metabase archivia i dati di analisi dell'utilizzo e di audit nel database dell'applicazione, inclusi i log delle attività, i log delle visualizzazioni e i record di esecuzione delle query. Una funzione Cloud Run esegue una query sull'API Metabase per recuperare questi dati e li scrive in un bucket GCS per l'importazione da parte di Google SecOps.
Prima di iniziare
Assicurati di soddisfare i seguenti prerequisiti:
- Un'istanza Google SecOps
- Un progetto GCP con le API Cloud Storage, Cloud Run, Pub/Sub e Cloud Scheduler abilitate
- Autorizzazioni per creare e gestire bucket GCS
- Autorizzazioni per gestire le policy IAM nei bucket GCS
- Autorizzazioni per creare servizi Cloud Run, argomenti Pub/Sub e job Cloud Scheduler
- Piano Metabase Pro o Enterprise (le analisi dell'utilizzo non sono disponibili nelle versioni Open Source o Cloud Starter)
- Accesso amministratore Metabase per creare chiavi API
- Il database dell'applicazione Metabase aggiunto come origine dati in Metabase (necessario per eseguire query sulle tabelle di controllo tramite API)
Configurare l'accesso all'API Metabase
Per consentire a Google SecOps di recuperare i dati di analisi dell'utilizzo, devi creare una chiave API in Metabase e aggiungere il database dell'applicazione come origine dati su cui è possibile eseguire query.
crea una chiave API
- Accedi alla tua istanza Metabase come amministratore.
- Fai clic sull'icona a forma di ingranaggio nell'angolo in alto a destra.
- Seleziona Impostazioni amministratore.
- Vai alla scheda Impostazioni.
- Fai clic sulla scheda Autenticazione nel menu a sinistra.
- Scorri fino a Chiavi API e fai clic su Gestisci.
- Fai clic sul pulsante Crea chiave API.
- Fornisci i seguenti dettagli di configurazione:
- Nome chiave: inserisci un nome descrittivo (ad esempio,
Google SecOps Integration). - Gruppo: seleziona un gruppo con accesso alle autorizzazioni di raccolta e query di Usage Analytics per il database dell'applicazione
- Nome chiave: inserisci un nome descrittivo (ad esempio,
- Fai clic su Crea.
Copia e salva la chiave API in modo sicuro.
Aggiungere il database dell'applicazione come origine dati
Per eseguire query sulle tabelle di controllo tramite l'API, devi aggiungere il database dell'applicazione Metabase come origine dati in Metabase.
- In Metabase, fai clic sull'icona a forma di ingranaggio nell'angolo in alto a destra.
- Seleziona Impostazioni amministratore.
- Vai alla scheda Database.
- Fai clic su Aggiungi un database.
- Configura la connessione al database:
- Tipo di database: seleziona PostgreSQL o MySQL (a seconda del database dell'applicazione)
- Nome: inserisci
Metabase Application Database - Host: inserisci il nome host del database dell'applicazione
- Porta: inserisci la porta del database (valore predefinito:
5432per PostgreSQL,3306per MySQL) - Nome database: inserisci il nome del database dell'applicazione (ad esempio,
metabase) - Nome utente: inserisci un utente del database con autorizzazioni SELECT sulle tabelle
audit_log,view_logequery_execution - Password: inserisci la password del database
- Fai clic su Salva.
- Dopo il salvataggio, prendi nota dell'ID database dall'URL (ad esempio,
https://metabase.example.com/admin/databases/5significa che l'ID database è5).
Verifica le autorizzazioni
Per verificare che la chiave API disponga delle autorizzazioni necessarie:
- Accedi a Metabase come amministratore.
- Fai clic sull'icona a forma di ingranaggio > Impostazioni amministrazione > Persone.
- Trova il gruppo assegnato alla chiave API.
- Verifica che il gruppo abbia accesso alla raccolta Analisi dell'utilizzo.
- Vai a Impostazioni di amministrazione > Autorizzazioni > Dati.
- Verifica che il gruppo abbia accesso Senza restrizioni o Query al database dell'applicazione.
Testare l'accesso API
Testa le tue credenziali prima di procedere con l'integrazione:
# 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
Una risposta positiva restituisce un array JSON di record del log di controllo recenti.
Crea un bucket Google Cloud Storage
- Vai alla console Google Cloud.
- Seleziona il tuo progetto o creane uno nuovo.
- Nel menu di navigazione, vai a Cloud Storage > Bucket.
- Fai clic su Crea bucket.
Fornisci i seguenti dettagli di configurazione:
Impostazione Valore Assegna un nome al bucket Inserisci un nome univoco globale (ad esempio metabase-audit-logs).Tipo di località Scegli in base alle tue esigenze (regione singola, a due regioni, multiregionale) Località Seleziona la posizione (ad esempio, us-central1).Classe di archiviazione Standard (consigliato per i log a cui si accede di frequente) Controllo degli accessi Uniforme (consigliato) Strumenti di protezione (Facoltativo) Abilita il controllo delle versioni degli oggetti o la policy di conservazione Fai clic su Crea.
Crea un account di servizio per la funzione Cloud Run
La funzione Cloud Run richiede un account di servizio con autorizzazioni di scrittura nel bucket GCS e di invocazione da parte di Pub/Sub.
Crea service account
- Nella console Google Cloud, vai a IAM e amministrazione > Service account.
- Fai clic su Crea account di servizio.
- Fornisci i seguenti dettagli di configurazione:
- Nome del service account: inserisci
metabase-audit-collector-sa - Descrizione service account: inserisci
Service account for Cloud Run function to collect Metabase audit logs
- Nome del service account: inserisci
- Fai clic su Crea e continua.
- Nella sezione Concedi a questo account di servizio l'accesso al progetto, aggiungi i seguenti ruoli:
- Fai clic su Seleziona un ruolo.
- Cerca e seleziona Storage Object Admin.
- Fai clic su + Aggiungi un altro ruolo.
- Cerca e seleziona Cloud Run Invoker.
- Fai clic su + Aggiungi un altro ruolo.
- Cerca e seleziona Cloud Functions Invoker.
- Fai clic su Continua.
- Fai clic su Fine.
Questi ruoli sono necessari per:
- Storage Object Admin: scrive i log nel bucket GCS e gestisce i file di stato
- Cloud Run Invoker: consente a Pub/Sub di richiamare la funzione
- Cloud Functions Invoker: consente la chiamata di funzioni
Concedi autorizzazioni IAM sul bucket GCS
- Vai a Cloud Storage > Bucket.
- Fai clic sul nome del bucket (
metabase-audit-logs). - Vai alla scheda Autorizzazioni.
- Fai clic su Concedi l'accesso.
- Fornisci i seguenti dettagli di configurazione:
- Aggiungi entità: inserisci l'email del account di servizio (
metabase-audit-collector-sa@PROJECT_ID.iam.gserviceaccount.com). - Assegna i ruoli: seleziona Storage Object Admin.
- Aggiungi entità: inserisci l'email del account di servizio (
- Fai clic su Salva.
Crea argomento Pub/Sub
- Nella console Google Cloud, vai a Pub/Sub > Argomenti.
- Fai clic su Crea argomento.
- Fornisci i seguenti dettagli di configurazione:
- ID argomento: inserisci
metabase-audit-trigger - Lascia invariate le altre impostazioni predefinite.
- ID argomento: inserisci
- Fai clic su Crea.
Crea una funzione Cloud Run per raccogliere i log
La funzione Cloud Run verrà attivata dai messaggi Pub/Sub di Cloud Scheduler per recuperare i dati di audit dall'API Metabase e scriverli in GCS.
- Nella console Google Cloud, vai a Cloud Run.
- Fai clic su Crea servizio.
- Seleziona Funzione (utilizza un editor in linea per creare una funzione).
Nella sezione Configura, fornisci i seguenti dettagli di configurazione:
Impostazione Valore Nome servizio metabase-audit-collectorRegione Seleziona la regione corrispondente al tuo bucket GCS (ad esempio, us-central1)Tempo di esecuzione Seleziona Python 3.12 o versioni successive. Nella sezione Trigger (facoltativo):
- Fai clic su + Aggiungi trigger.
- Seleziona Cloud Pub/Sub.
- In Seleziona un argomento Cloud Pub/Sub, scegli
metabase-audit-trigger. - Fai clic su Salva.
Nella sezione Autenticazione:
- Seleziona Richiedi autenticazione.
- Controlla Identity and Access Management (IAM).
Scorri verso il basso ed espandi Container, networking, sicurezza.
Vai alla scheda Sicurezza:
- Service account: seleziona
metabase-audit-collector-sa.
- Service account: seleziona
Vai alla scheda Container:
- Fai clic su Variabili e secret.
- Fai clic su + Aggiungi variabile per ogni variabile di ambiente:
Nome variabile Valore di esempio Descrizione GCS_BUCKETmetabase-audit-logsNome bucket GCS GCS_PREFIXmetabase-auditPrefisso per i file di log STATE_KEYmetabase-audit/state.jsonPercorso file di stato METABASE_URLhttps://metabase.yourcompany.comURL di base dell'istanza Metabase METABASE_API_KEYmb_your_api_key_hereChiave API Metabase DATABASE_ID5ID database dell'applicazione LOOKBACK_HOURS24Periodo di riferimento iniziale PAGE_SIZE2000Record per query API Nella sezione Variabili e secret, scorri verso il basso fino a Richieste:
- Timeout richiesta: inserisci
600secondi (10 minuti)
- Timeout richiesta: inserisci
Vai alla scheda Impostazioni:
- Nella sezione Risorse:
- Memoria: seleziona 512 MiB o un valore superiore.
- CPU: seleziona 1
- Nella sezione Risorse:
Nella sezione Scalabilità della revisione:
- Numero minimo di istanze: inserisci
0 - Numero massimo di istanze: inserisci
100
- Numero minimo di istanze: inserisci
Fai clic su Crea.
Attendi la creazione del servizio (1-2 minuti).
Dopo aver creato il servizio, si aprirà automaticamente l'editor di codice incorporato.
Aggiungi codice per la funzione
- Inserisci main nel campo Entry point (Punto di ingresso).
Nell'editor di codice incorporato, crea due file:
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.0Fai clic su Esegui il deployment per salvare la funzione ed eseguirne il deployment.
Attendi il completamento del deployment (2-3 minuti).
Crea job Cloud Scheduler
- Nella console Google Cloud, vai a Cloud Scheduler.
- Fai clic su Crea job.
Fornisci i seguenti dettagli di configurazione:
Impostazione Valore Nome metabase-audit-collector-hourlyRegione Seleziona la stessa regione della funzione Cloud Run Frequenza 0 * * * *(ogni ora, allo scoccare dell'ora)Fuso orario Seleziona il fuso orario (UTC consigliato) Tipo di target Pub/Sub Argomento Seleziona metabase-audit-triggerCorpo del messaggio {}(oggetto JSON vuoto)Fai clic su Crea.
Opzioni di frequenza di pianificazione
Scegli la frequenza in base al volume dei log e ai requisiti di latenza:
| Frequenza | Espressione cron | Caso d'uso |
|---|---|---|
| Ogni 5 minuti | */5 * * * * |
Volume elevato, bassa latenza |
| Ogni 15 minuti | */15 * * * * |
Volume medio |
| Ogni ora | 0 * * * * |
Standard (consigliato) |
| Ogni 6 ore | 0 */6 * * * |
Volume basso, elaborazione batch |
| Ogni giorno | 0 0 * * * |
Raccolta dei dati storici |
Testare l'integrazione
- Nella console Cloud Scheduler, trova il tuo job (
metabase-audit-collector-hourly). - Fai clic su Forza esecuzione per attivare il job manualmente.
- Attendi alcuni secondi.
- Vai a Cloud Run > Servizi.
- Fai clic su
metabase-audit-collector. - Fai clic sulla scheda Log.
Verifica che la funzione sia stata eseguita correttamente. Cerca:
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 recordsVai a Cloud Storage > Bucket.
Fai clic su
metabase-audit-logs.Vai alla cartella
metabase-audit/.Verifica che sia stato creato un nuovo file
.ndjsoncon il timestamp corrente.
Se nei log vengono visualizzati errori:
- HTTP 401: verifica che la variabile di ambiente
METABASE_API_KEYsia corretta e che la chiave non sia stata eliminata o rigenerata - HTTP 403: verifica che il gruppo della chiave API disponga delle autorizzazioni di query per il database dell'applicazione
- HTTP 429: limitazione della frequenza: la funzione interromperà la paginazione e riprenderà alla successiva esecuzione pianificata
- Variabili di ambiente mancanti: verifica che tutte le variabili richieste siano impostate nella configurazione di Cloud Run Functions
- Risultati vuoti: verifica che
DATABASE_IDsia corretto, che il database dell'applicazione sia stato aggiunto come origine dati e che tu stia utilizzando Metabase Pro o Enterprise
Recuperare il account di servizio Google SecOps
- Vai a Impostazioni SIEM > Feed.
- Fai clic su Aggiungi nuovo feed.
- Fai clic su Configura un singolo feed.
- Nel campo Nome feed, inserisci un nome per il feed (ad esempio,
Metabase Audit Logs). - Seleziona Google Cloud Storage V2 come Tipo di origine.
- Seleziona METABASE come Tipo di log.
Fai clic su Ottieni service account. Verrà visualizzato un indirizzo email del account di servizio univoco, ad esempio:
chronicle-12345678@chronicle-gcp-prod.iam.gserviceaccount.comCopia questo indirizzo email per utilizzarlo nel passaggio successivo.
Fai clic su Avanti.
Specifica i valori per i seguenti parametri di input:
URL del bucket di archiviazione: inserisci l'URI del bucket GCS con il percorso del prefisso:
gs://metabase-audit-logs/metabase-audit/Opzione di eliminazione dell'origine: seleziona l'opzione di eliminazione in base alle tue preferenze:
- Mai: non elimina mai i file dopo i trasferimenti (opzione consigliata per i test).
- Elimina file trasferiti: elimina i file dopo il trasferimento riuscito.
Elimina file trasferiti e directory vuote: elimina i file e le directory vuote dopo il trasferimento riuscito.
Età massima del file: includi i file modificati nell'ultimo numero di giorni (il valore predefinito è 180 giorni)
Spazio dei nomi dell'asset: lo spazio dei nomi dell'asset
Etichette di importazione: l'etichetta da applicare agli eventi di questo feed
Fai clic su Avanti.
Controlla la nuova configurazione del feed nella schermata Finalizza e poi fai clic su Invia.
Concedi le autorizzazioni IAM al account di servizio Google SecOps
Il account di servizio Google SecOps deve avere il ruolo Visualizzatore oggetti Storage nel tuo bucket GCS.
- Vai a Cloud Storage > Bucket.
- Fai clic su
metabase-audit-logs. - Vai alla scheda Autorizzazioni.
- Fai clic su Concedi l'accesso.
- Fornisci i seguenti dettagli di configurazione:
- Aggiungi entità: incolla l'email del account di servizio Google SecOps
- Assegna i ruoli: seleziona Visualizzatore oggetti Storage.
Fai clic su Salva.
Tabelle di analisi dell'utilizzo
La funzione Cloud Run esegue query sulle seguenti tabelle del database dell'applicazione Metabase:
| Nome tabella | Descrizione |
|---|---|
| audit_log | Record delle modifiche alla configurazione, delle azioni degli utenti e degli eventi di sistema |
| view_log | Monitora le visualizzazioni di schede (domande/modelli), dashboard e tabelle |
| query_execution | Informazioni su tutte le query eseguite in tutte le dashboard |
Conservazione dei dati
Per impostazione predefinita, Metabase conserva i dati di analisi dell'utilizzo per 720 giorni (circa 2 anni). Due volte al giorno, Metabase elimina automaticamente le righe precedenti a questa soglia. Per modificare il periodo di conservazione, imposta la variabile di ambiente MB_AUDIT_MAX_RETENTION_DAYS quando esegui Metabase.
Tabella di mappatura UDM
| Campo log | Mappatura UDM | Logica |
|---|---|---|
| aggiuntivo | aggiuntivo | Rinominato da aggiuntivo |
| lingua | additional.fields | Unito alle rispettive etichette (ad es. locale_label per locale, is_active_label per is_active e così via) |
| 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 | |
| metadati | metadati | Rinominate dai metadati |
| updated_at | metadata.event_timestamp | Data analizzata da updated_at utilizzando ISO8601, RFC 3339 o AAAA-MM-GGTHH:mm:ss.SSSSSSZ |
| has_userid | metadata.event_type | Imposta su "USER_UNCATEGORIZED" se has_userid == "true", altrimenti "GENERIC_EVENT" |
| entità | entità | Rinominato dall'entità |
| metabase_host | principal.url | Valore copiato direttamente |
| date_joined | principal.user.attribute.creation_time | Data analizzata da date_joined utilizzando ISO8601, RFC 3339 o AAAA-MM-GGTHH:mm:ss.SSSSSSZ |
| principal.user.email_addresses | Unito dall'email | |
| first_name | principal.user.first_name | Valore copiato direttamente |
| last_login | principal.user.last_login_time | Data analizzata da last_login utilizzando ISO8601, RFC 3339 o AAAA-MM-GGTHH:mm:ss.SSSSSSZ |
| last_name | principal.user.last_name | Valore copiato direttamente |
| common_name | principal.user.user_display_name | Valore copiato direttamente |
| principal.user.userid | Valore copiato direttamente dall'email | |
| security_result | security_result | Unito da security_result |
| destinazione | destinazione | Rinominato dal target |
| metadata.product_name | metadata.product_name | Impostato su "METABASE" |
| metadata.vendor_name | metadata.vendor_name | Impostato su "METABASE" |
Log delle modifiche
Visualizza il log delle modifiche per questo parser
Hai bisogno di ulteriore assistenza? Ricevi risposte dai membri della community e dai professionisti di Google SecOps.