איסוף יומנים של Metabase

נתמך ב:

במאמר הזה מוסבר איך להטמיע יומנים של Metabase ב-Google Security Operations באמצעות Google Cloud Storage V2.

‫Metabase היא פלטפורמה בקוד פתוח לבינה עסקית ולניתוח נתונים, שמאפשרת לארגונים לשלוח שאילתות למסדי נתונים, ליצור תצוגות חזותיות ולבנות לוחות בקרה. בתוכניות Pro ו-Enterprise,‏ Metabase מאחסנת ניתוח נתוני שימוש ונתוני ביקורת במסד הנתונים של האפליקציה, כולל יומני פעילות, יומני צפייה ורשומות של ביצוע שאילתות. פונקציית Cloud Run שולחת שאילתה ל-Metabase API כדי לאחזר את הנתונים האלה, וכותבת אותם לקטגוריית GCS לצורך הטמעה ב-Google SecOps.

לפני שמתחילים

חשוב לוודא שמתקיימות דרישות הסף הבאות:

  • מופע של Google SecOps
  • פרויקט GCP עם ממשקי Cloud Storage,‏ Cloud Run,‏ Pub/Sub ו-Cloud Scheduler API מופעלים
  • הרשאות ליצירה ולניהול של מאגרי GCS
  • הרשאות לניהול מדיניות IAM בקטגוריות GCS
  • הרשאות ליצירת שירותי Cloud Run, נושאי Pub/Sub ומשימות Cloud Scheduler
  • מינוי Metabase Pro או Enterprise (ניתוח השימוש לא זמין במהדורות Open Source או Cloud Starter)
  • גישת אדמין ל-Metabase כדי ליצור מפתחות API
  • מסד הנתונים של אפליקציית Metabase נוסף כמקור נתונים ב-Metabase (נדרש לשליחת שאילתות לטבלאות ביקורת באמצעות API)

הגדרת גישה ל-API של Metabase

כדי לאפשר ל-Google SecOps לאחזר נתוחי שימוש, צריך ליצור מפתח API ב-Metabase ולהוסיף את מסד הנתונים של האפליקציה כמקור נתונים שאפשר להריץ עליו שאילתות.

יצירה של מפתח API

  1. נכנסים לחשבון Metabase כאדמינים.
  2. לוחצים על סמל גלגל השיניים בפינה השמאלית העליונה.
  3. בוחרים באפשרות הגדרות אדמין.
  4. עוברים לכרטיסייה הגדרות.
  5. בתפריט הימני, לוחצים על הכרטיסייה אימות.
  6. גוללים אל API Keys (מפתחות API) ולוחצים על Manage (ניהול).
  7. לוחצים על הלחצן Create API Key (יצירת מפתח API).
  8. מספקים את פרטי ההגדרה הבאים:
    • שם המפתח: מזינים שם תיאורי (לדוגמה, Google SecOps Integration).
    • קבוצה: בוחרים קבוצה עם גישה לאיסוף נתוני השימוש ב-Analytics והרשאות לשאילתות במסד הנתונים של האפליקציה
  9. לוחצים על יצירה.
  10. מעתיקים את מפתח ה-API ושומרים אותו במקום מאובטח.

הוספת מסד הנתונים של האפליקציה כמקור נתונים

כדי להריץ שאילתות בטבלאות ביקורת באמצעות ה-API, צריך להוסיף את מסד הנתונים של אפליקציית Metabase כמקור נתונים ב-Metabase.

  1. ב-Metabase, לוחצים על סמל גלגל השיניים בפינה השמאלית העליונה.
  2. בוחרים באפשרות הגדרות אדמין.
  3. עוברים לכרטיסייה מסדי נתונים.
  4. לוחצים על הוספת מסד נתונים.
  5. מגדירים את החיבור למסד הנתונים:
    • סוג מסד הנתונים: בוחרים באפשרות PostgreSQL או MySQL (בהתאם למסד הנתונים של האפליקציה)
    • שם: מזינים Metabase Application Database
    • מארח: מזינים את שם המארח של מסד הנתונים של האפליקציה
    • יציאה: מזינים את היציאה של מסד הנתונים (ברירת מחדל: 5432 ל-PostgreSQL, ‏ 3306 ל-MySQL)
    • שם מסד הנתונים: מזינים את שם מסד הנתונים של האפליקציה (לדוגמה, metabase).
    • שם משתמש: מזינים משתמש במסד הנתונים עם הרשאות SELECT בטבלאות audit_log,‏ view_log ו-query_execution
    • ‫Password (סיסמה): מזינים את סיסמת מסד הנתונים
  6. לוחצים על Save.
  7. אחרי השמירה, מעתיקים את מזהה מסד הנתונים מכתובת ה-URL (לדוגמה, אם כתובת ה-URL היא https://metabase.example.com/admin/databases/5, מזהה מסד הנתונים הוא 5).

אימות ההרשאות

כדי לוודא שלמפתח ה-API יש את ההרשאות הנדרשות:

  1. נכנסים ל-Metabase כאדמינים.
  2. לוחצים על סמל גלגל השיניים > הגדרות אדמין > אנשים.
  3. מאתרים את הקבוצה שהוקצתה למפתח ה-API.
  4. מוודאים שלקבוצה יש גישה לאיסוף של ניתוח נתוני השימוש.
  5. עוברים אל הגדרות ניהול > הרשאות > נתונים.
  6. מוודאים שלקבוצה יש גישה לא מוגבלת או גישת שאילתה למסד הנתונים של האפליקציה.

בדיקת הגישה ל-API

  • לפני שממשיכים בשילוב, חשוב לבדוק את פרטי הכניסה:

    # 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
    

תגובה מוצלחת מחזירה מערך JSON של רשומות עדכניות ביומן הביקורת.

יצירת קטגוריה של Google Cloud Storage

  1. נכנסים אל מסוף Google Cloud.
  2. בוחרים פרויקט או יוצרים פרויקט חדש.
  3. בתפריט הניווט, עוברים אל Cloud Storage > Buckets.
  4. לוחצים על Create bucket.
  5. מספקים את פרטי ההגדרה הבאים:

    הגדרה ערך
    Name your bucket (שם הקטגוריה) מזינים שם ייחודי גלובלית (לדוגמה, metabase-audit-logs)
    סוג המיקום בוחרים לפי הצרכים (אזור, שני אזורים, מספר אזורים)
    Location בוחרים את המיקום (לדוגמה, us-central1).
    סוג אחסון (storage class) רגיל (מומלץ ליומנים שמתבצעת אליהם גישה לעיתים קרובות)
    בקרת גישה אחיד (מומלץ)
    כלים להגנה אופציונלי: הפעלת ניהול גרסאות של אובייקטים או מדיניות שמירת נתונים
  6. לוחצים על יצירה.

יצירת חשבון שירות לפונקציית Cloud Run

פונקציית Cloud Run צריכה חשבון שירות עם הרשאות כתיבה לקטגוריה של GCS והרשאה להפעלה על ידי Pub/Sub.

יצירת חשבון שירות

  1. במסוף GCP, עוברים אל IAM & Admin > Service Accounts.
  2. לוחצים על יצירת חשבון שירות.
  3. מספקים את פרטי ההגדרה הבאים:
    • שם חשבון השירות: מזינים metabase-audit-collector-sa
    • תיאור חשבון השירות: מזינים Service account for Cloud Run function to collect Metabase audit logs
  4. לוחצים על יצירה והמשך.
  5. בקטע Grant this service account access to project (מתן גישה של חשבון השירות הזה לפרויקט), מוסיפים את התפקידים הבאים:
    1. לוחצים על בחירת תפקיד.
    2. מחפשים את Storage Object Admin ולוחצים עליו.
    3. לוחצים על + הוספת תפקיד נוסף.
    4. מחפשים את האפשרות Cloud Run Invoker (הפעלת שירותים ב-Cloud Run) ולוחצים עליה.
    5. לוחצים על + הוספת תפקיד נוסף.
    6. מחפשים את האפשרות Cloud Functions Invoker (הפעלת פונקציות של Cloud Functions) ולוחצים עליה.
  6. לוחצים על Continue.
  7. לוחצים על סיום.

התפקידים האלה נדרשים כדי:

  • אדמין של אובייקט אחסון: כתיבת יומנים לקטגוריית GCS וניהול קבצי מצב
  • ‫Cloud Run Invoker: מתן הרשאה ל-Pub/Sub להפעיל את הפונקציה
  • Cloud Functions Invoker: מאפשר הפעלה של פונקציות

מתן הרשאות IAM בקטגוריית GCS

  1. עוברים אל Cloud Storage > Buckets (קטגוריות).
  2. לוחצים על שם הקטגוריה (metabase-audit-logs).
  3. עוברים לכרטיסייה Permissions.
  4. לוחצים על הענקת גישה.
  5. מספקים את פרטי ההגדרה הבאים:
    • הוספת ישויות מורשות: מזינים את כתובת האימייל בחשבון של חשבון השירות (metabase-audit-collector-sa@PROJECT_ID.iam.gserviceaccount.com)
    • הקצאת תפקידים: בוחרים באפשרות אדמין של אובייקטים באחסון.
  6. לוחצים על Save.

יצירת נושא Pub/Sub

  1. במסוף GCP, עוברים אל Pub/Sub > Topics.
  2. לוחצים על יצירת נושא.
  3. מספקים את פרטי ההגדרה הבאים:
    • מזהה הנושא: מזינים metabase-audit-trigger
    • השארת ההגדרות האחרות כברירת מחדל
  4. לוחצים על יצירה.

יצירת פונקציית Cloud Run לאיסוף יומנים

הפונקציה של Cloud Run תופעל על ידי הודעות Pub/Sub מ-Cloud Scheduler כדי לאחזר נתוני ביקורת מ-Metabase API ולכתוב אותם ב-GCS.

  1. במסוף GCP, עוברים אל Cloud Run.
  2. לוחצים על יצירת שירות.
  3. בוחרים באפשרות Function (שימוש בעורך ישיר ליצירת פונקציה).
  4. בקטע Configure (הגדרה), מזינים את פרטי ההגדרה הבאים:

    הגדרה ערך
    שם השירות metabase-audit-collector
    אזור בוחרים אזור שתואם ל-GCS Bucket (לדוגמה, us-central1)
    זמן ריצה בוחרים באפשרות Python 3.12 ואילך.
  5. בקטע טריגר (אופציונלי):

    1. לוחצים על + הוספת טריגר.
    2. בוחרים באפשרות Cloud Pub/Sub.
    3. בקטע Select a Cloud Pub/Sub topic, בוחרים באפשרות metabase-audit-trigger.
    4. לוחצים על Save.
  6. בקטע Authentication:

    1. בוחרים באפשרות דרישת אימות.
    2. בודקים את ניהול זהויות והרשאות גישה (IAM).
  7. גוללים למטה ומרחיבים את Containers, Networking, Security (מאגרי תגים, רשתות, אבטחה).

  8. עוברים לכרטיסייה אבטחה:

    • חשבון שירות: בוחרים באפשרות metabase-audit-collector-sa
  9. עוברים לכרטיסייה מאגרי תגים:

    1. לוחצים על Variables & Secrets (משתנים וסודות).
    2. לוחצים על + הוספת משתנה לכל משתנה סביבתי:
    שם המשתנה ערך לדוגמה תיאור
    GCS_BUCKET metabase-audit-logs שם קטגוריית GCS
    GCS_PREFIX metabase-audit קידומת לקובצי יומן
    STATE_KEY metabase-audit/state.json נתיב קובץ תמונת מצב
    METABASE_URL https://metabase.yourcompany.com כתובת URL בסיסית של מכונת Metabase
    METABASE_API_KEY mb_your_api_key_here מפתח Metabase API
    DATABASE_ID 5 מזהה מסד הנתונים של האפליקציה
    LOOKBACK_HOURS 24 תקופת המבט לאחור הראשונית
    PAGE_SIZE 2000 רשומות לכל שאילתת API
  10. בקטע Variables & Secrets, גוללים למטה אל Requests:

    • זמן קצוב לתפוגת בקשה: מזינים 600 שניות (10 דקות)
  11. עוברים לכרטיסייה הגדרות:

    • בקטע Resources (מקורות מידע):
      • זיכרון: בוחרים באפשרות 512MiB או יותר
      • ‫CPU: בוחרים באפשרות 1
  12. בקטע שינוי קנה מידה של העריכה:

    • מספר מינימלי של מופעים: מזינים 0
    • מספר מקרים מקסימלי: מזינים 100
  13. לוחצים על יצירה.

  14. ממתינים ליצירת השירות (1-2 דקות).

  15. אחרי שיוצרים את השירות, עורך הקוד המוטבע נפתח באופן אוטומטי.

הוספת קוד פונקציה

  1. מזינים main בשדה Entry point.
  2. בעורך הקוד המוטבע, יוצרים שני קבצים:

    • 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. לוחצים על Deploy כדי לשמור ולפרוס את הפונקציה.

  4. ממתינים לסיום הפריסה (2-3 דקות).

יצירת משימה ב-Cloud Scheduler

  1. במסוף GCP, עוברים אל Cloud Scheduler.
  2. לוחצים על Create Job (יצירת משימה).
  3. מספקים את פרטי ההגדרה הבאים:

    הגדרה ערך
    שם metabase-audit-collector-hourly
    אזור בחירה באותו אזור כמו פונקציית Cloud Run
    תדירות ‫0 * * * * (כל שעה, בדיוק בשעה)
    אזור זמן בחירת אזור זמן (מומלץ לבחור UTC)
    סוג הטירגוט Pub/Sub
    נושא בחירת metabase-audit-trigger
    גוף ההודעה ‫{} (אובייקט JSON ריק)
  4. לוחצים על יצירה.

אפשרויות לתדירות התזמון

בחירת התדירות בהתאם לנפח היומן ולדרישות זמן האחזור:

תדירות ביטוי Cron תרחיש לדוגמה
כל 5 דקות */5 * * * * נפח גבוה, זמן אחזור נמוך
כל 15 דקות */15 * * * * נפח חיפושים בינוני
כל שעה 0 * * * * רגיל (מומלץ)
כל 6 שעות 0 */6 * * * נפח נמוך, עיבוד באצווה
יומי 0 0 * * * איסוף נתונים היסטוריים

בדיקת השילוב

  1. במסוף Cloud Scheduler, מאתרים את המשימה (metabase-audit-collector-hourly).
  2. לוחצים על הפעלת המשימה בכוח כדי להפעיל את המשימה באופן ידני.
  3. מחכים כמה שניות.
  4. עוברים אל Cloud Run > Services.
  5. לוחצים על metabase-audit-collector.
  6. לוחצים על הכרטיסייה Logs (יומנים).
  7. מוודאים שהפונקציה בוצעה בהצלחה. מחפשים:

    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. עוברים אל Cloud Storage > Buckets (קטגוריות).

  9. לוחצים על metabase-audit-logs.

  10. עוברים לתיקייה metabase-audit/.

  11. מוודאים שנוצר קובץ .ndjson חדש עם חותמת הזמן הנוכחית.

אם מופיעות שגיאות ביומנים:

  • ‫HTTP 401: מוודאים שמשתנה הסביבה METABASE_API_KEY נכון ושלא נמחק או נוצר מחדש מפתח
  • ‫HTTP 403: מוודאים שלקבוצה של מפתח ה-API יש הרשאות לשאילתות במסד הנתונים של האפליקציה
  • ‫HTTP 429: הגבלת קצב – הפונקציה תפסיק את המספור העמודים ותמשיך בפעולה בהפעלה המתוזמנת הבאה
  • משתני סביבה חסרים: מוודאים שכל המשתנים הנדרשים מוגדרים בהגדרות של פונקציית Cloud Run
  • תוצאות ריקות: צריך לוודא ש-DATABASE_ID נכון, שמאגר הנתונים של האפליקציה נוסף כמקור נתונים ושאתם משתמשים ב-Metabase Pro או ב-Enterprise

אחזור חשבון השירות של Google SecOps

  1. עוברים אל SIEM Settings > Feeds (הגדרות SIEM > פידים).
  2. לוחצים על הוספת פיד חדש.
  3. לוחצים על Configure a single feed (הגדרת פיד יחיד).
  4. בשדה שם הפיד, מזינים שם לפיד (לדוגמה, Metabase Audit Logs).
  5. בוחרים באפשרות Google Cloud Storage V2 בתור סוג המקור.
  6. בוחרים באפשרות METABASE בתור סוג היומן.
  7. לוחצים על קבלת חשבון שירות. תוצג כתובת האימייל בחשבון שירות ייחודי, לדוגמה:

    chronicle-12345678@chronicle-gcp-prod.iam.gserviceaccount.com
    
  8. מעתיקים את כתובת האימייל הזו לשימוש בשלב הבא.

  9. לוחצים על הבא.

  10. מציינים ערכים לפרמטרים הבאים של הקלט:

    • כתובת ה-URL של קטגוריית האחסון: מזינים את ה-URI של קטגוריית GCS עם נתיב הקידומת:

      gs://metabase-audit-logs/metabase-audit/
      
    • אפשרות למחיקת המקור: בוחרים את אפשרות המחיקה לפי ההעדפה שלכם:

      • אף פעם: לא נמחקים קבצים אחרי ההעברה (מומלץ לבדיקות).
      • מחיקת קבצים שהועברו: הקבצים נמחקים אחרי שההעברה מסתיימת.
      • מחיקת קבצים שהועברו וספריות ריקות: מחיקת קבצים וספריות ריקות אחרי שההעברה מסתיימת בהצלחה.

    • הגיל המקסימלי של הקובץ: כולל קבצים ששונו במספר הימים האחרון (ברירת המחדל היא 180 ימים)

    • מרחב השמות של הנכס: מרחב השמות של הנכס

    • תוויות להוספה: התווית שתתווסף לאירועים מהפיד הזה

  11. לוחצים על הבא.

  12. בודקים את ההגדרות של הפיד החדש במסך סיום ולוחצים על שליחה.

הענקת הרשאות IAM לחשבון השירות של Google SecOps

לחשבון השירות של Google SecOps צריך להיות התפקיד צפייה באובייקט אחסון בקטגוריה של GCS.

  1. עוברים אל Cloud Storage > Buckets (קטגוריות).
  2. לוחצים על metabase-audit-logs.
  3. עוברים לכרטיסייה Permissions.
  4. לוחצים על הענקת גישה.
  5. מספקים את פרטי ההגדרה הבאים:
    • Add principals (הוספת ישויות מורשות): מדביקים את כתובת האימייל בחשבון של חשבון השירות של Google SecOps
    • הקצאת תפקידים: בוחרים באפשרות צפייה באובייקטים באחסון.
  6. לוחצים על Save.

טבלאות של ניתוח השימוש

פונקציית Cloud Run שולחת שאילתות לטבלאות הבאות ממסד הנתונים של אפליקציית Metabase:

שם טבלה תיאור
audit_log רשומות של שינויים בהגדרות, פעולות משתמשים ואירועים במערכת
view_log מעקב אחרי צפיות בכרטיסים (שאלות/מודלים), בלוחות בקרה ובטבלאות
query_execution מידע על כל השאילתות שמופעלות בכל לוחות הבקרה

שמירת נתונים

כברירת מחדל, נתוני ניתוח השימוש נשמרים ב-Metabase למשך 720 ימים (כשנתיים). פעמיים ביום, Metabase מוחקת באופן אוטומטי שורות ישנות יותר מערך הסף הזה. כדי לשנות את תקופת השמירה, מגדירים את משתנה הסביבה MB_AUDIT_MAX_RETENTION_DAYS כשמריצים את Metabase.

טבלת מיפוי UDM

שדה ביומן מיפוי UDM היגיון
נוסף נוסף השם שונה מ-additional
לוקאל additional.fields המאפיינים אוחדו עם התוויות המתאימות (לדוגמה, locale_label עבור locale,‏ is_active_label עבור is_active וכו')
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
מטא-נתונים מטא-נתונים שם חדש במקום מטא-נתונים
updated_at metadata.event_timestamp התאריך שנותח מהשדה updated_at באמצעות ISO8601,‏ RFC 3339 או yyyy-MM-ddTHH:mm:ss.SSSSSSZ
has_userid metadata.event_type הערך הוא USER_UNCATEGORIZED אם has_userid == true, אחרת GENERIC_EVENT
משתמש (principal) משתמש (principal) השם שונה מהשם של הישות המורשית
metabase_host principal.url הערך הועתק ישירות
date_joined principal.user.attribute.creation_time התאריך שנותח מ-date_joined באמצעות ISO8601,‏ RFC 3339 או yyyy-MM-ddTHH:mm:ss.SSSSSSZ
אימייל principal.user.email_addresses מוזג מאימייל
first_name principal.user.first_name הערך הועתק ישירות
last_login principal.user.last_login_time התאריך שנותח מ-last_login באמצעות ISO8601,‏ RFC 3339 או yyyy-MM-ddTHH:mm:ss.SSSSSSZ
last_name principal.user.last_name הערך הועתק ישירות
common_name principal.user.user_display_name הערך הועתק ישירות
אימייל principal.user.userid הערך הועתק ישירות מאימייל
security_result security_result התוצאות ממוזגות מ-security_result
target target השם שונה מהיעד
metadata.product_name metadata.product_name ההגדרה היא 'METABASE'
metadata.vendor_name metadata.vendor_name ההגדרה היא 'METABASE'

יומן שינויים

צפייה ביומן השינויים של מנתח התוכן הזה

הבעיה עדיין לא נפתרה? קבלת תשובות מחברי הקהילה וממומחי Google SecOps.