收集 Cisco 應用程式中心基礎架構 (ACI) 記錄

支援的國家/地區:

本文說明如何將 Cisco Application Centric Infrastructure (ACI) 記錄檔擷取至 Google Security Operations。剖析器會先嘗試使用 Grok 模式,將傳入的 Cisco ACI 記錄處理為系統記錄訊息。如果無法剖析系統記錄,系統會假設訊息採用 JSON 格式,並據此進行剖析。最後,它會將擷取的欄位對應至統合式資料模型 (UDM)。

這項整合支援兩種方法:

  • 選項 1:透過 Bindplane 代理程式的 Syslog 格式
  • 選項 2:透過 AWS S3 使用 APIC REST API,以 JSON 格式上傳

每個選項都是獨立的,您可以根據基礎架構需求和記錄格式偏好,獨立實作這些選項。

選項 1:透過 Bindplane 代理程式傳送 Syslog

這個選項會設定 Cisco ACI 網狀架構,將系統記錄檔訊息傳送至 Bindplane 代理程式,再由代理程式將訊息轉送至 Chronicle 進行分析。

事前準備

請確認您已完成下列事前準備事項:

  • Google SecOps 執行個體
  • 搭載 systemd 的 Windows 2016 以上版本或 Linux 主機
  • 如果透過 Proxy 執行,請確保防火牆通訊埠已根據 Bindplane 代理程式需求開啟
  • Cisco APIC 控制台的特殊權限

取得 Google SecOps 擷取驗證檔案

  1. 登入 Google SecOps 控制台。
  2. 依序前往「SIEM 設定」>「收集代理程式」
  3. 下載擷取驗證檔案。將檔案安全地儲存在要安裝 Bindplane 的系統上。

取得 Google SecOps 客戶 ID

  1. 登入 Google SecOps 控制台。
  2. 依序前往「SIEM 設定」>「設定檔」
  3. 複製並儲存「機構詳細資料」專區中的客戶 ID

安裝 Bindplane 代理程式

請按照下列操作說明,在 Windows 或 Linux 作業系統上安裝 Bindplane 代理程式。

Windows 安裝

  1. 以管理員身分開啟「命令提示字元」或「PowerShell」
  2. 執行下列指令:

    msiexec /i "https://github.com/observIQ/bindplane-agent/releases/latest/download/observiq-otel-collector.msi" /quiet
    

Linux 安裝

  1. 開啟具備根層級或 sudo 權限的終端機。
  2. 執行下列指令:

    sudo sh -c "$(curl -fsSlL https://github.com/observiq/bindplane-agent/releases/latest/download/install_unix.sh)" install_unix.sh
    

其他安裝資源

設定 Bindplane 代理程式,擷取系統記錄檔並傳送至 Google SecOps

  1. 存取設定檔:

    1. 找出 config.yaml 檔案。通常位於 Linux 的 /etc/bindplane-agent/ 目錄,或 Windows 的安裝目錄。
    2. 使用文字編輯器 (例如 nanovi 或記事本) 開啟檔案。
  2. 按照下列方式編輯 config.yaml 檔案:

    receivers:
      udplog:
        # Replace the port and IP address as required
        listen_address: "0.0.0.0:514"
    
    exporters:
      chronicle/chronicle_w_labels:
        compression: gzip
        # Adjust the path to the credentials file you downloaded in Step 1
        creds_file_path: '/path/to/ingestion-authentication-file.json'
        # Replace with your actual customer ID from Step 2
        customer_id: <CUSTOMER_ID>
        endpoint: malachiteingestion-pa.googleapis.com
        # Add optional ingestion labels for better organization
        log_type: 'CISCO_ACI'
        raw_log_field: body
        ingestion_labels:
    
    service:
      pipelines:
        logs/source0__chronicle_w_labels-0:
          receivers:
            - udplog
          exporters:
            - chronicle/chronicle_w_labels
    
    • 視基礎架構需求,替換通訊埠和 IP 位址。
    • <customer_id> 替換為實際的客戶 ID。
    • /path/to/ingestion-authentication-file.json 更新為「取得 Google SecOps 擷取驗證檔案」部分中驗證檔案的儲存路徑。

重新啟動 Bindplane 代理程式,以套用變更

  • 如要在 Linux 中重新啟動 Bindplane 代理程式,請執行下列指令:

    sudo systemctl restart bindplane-agent
    
  • 如要在 Windows 中重新啟動 Bindplane 代理程式,可以使用「服務」控制台,或輸入下列指令:

    net stop BindPlaneAgent && net start BindPlaneAgent
    

在 Cisco ACI 上設定 Syslog 轉送

設定頻外管理合約

  1. 登入 Cisco APIC 控制台。
  2. 依序前往「Tenants」>「mgmt」>「Contracts」>「Filters」
  3. 按一下「建立篩選器」
  4. 提供下列設定詳細資料:
    • 「Name」(名稱):輸入 syslog-udp-514
    • 項目名稱:輸入 syslog
    • EtherType:選取「IP」
    • IP 通訊協定:選取「UDP」
    • 目標通訊埠範圍 (從):輸入 514
    • 目標位置通訊埠範圍至:輸入 514
  5. 按一下「提交」

建立管理合約

  1. 依序前往「Tenants」>「mgmt」>「Contracts」>「Standard」
  2. 按一下「建立合約」
  3. 提供下列設定詳細資料:
    • 「Name」(名稱):輸入 mgmt-syslog-contract
    • 範圍:選取「內容」
  4. 按一下「提交」
  5. 展開合約,然後按一下「主題」
  6. 按一下「建立合約主題」
  7. 提供下列設定詳細資料:
    • 「Name」(名稱):輸入 syslog-subject
    • 同時套用兩個方向:勾選這個選項。
  8. 按一下「提交」
  9. 展開主題,然後按一下「篩選器」
  10. 按一下「建立篩選器繫結」
  11. 選取先前建立的 syslog-udp-514 篩選器。
  12. 按一下「提交」

設定系統記錄目的地群組

  1. 依序前往「管理」>「外部資料收集器」>「監控目的地」>「系統記錄」
  2. 在「Syslog」上按一下滑鼠右鍵,然後選取「Create Syslog Monitoring Destination Group」(建立 Syslog 監控目的地群組)
  3. 提供下列設定詳細資料:
    • 「Name」(名稱):輸入 Chronicle-Syslog-Group
    • 管理狀態:選取「已啟用」
    • 格式:選取 aci
  4. 點選「下一步」
  5. 在「建立系統記錄檔監控目的地」對話方塊中
    • 「Name」(名稱):輸入 Chronicle-BindPlane
    • 主機:輸入 Bindplane 代理程式伺服器的 IP 位址。
    • 「Port」(通訊埠):輸入 514
    • 管理狀態:選取「已啟用」
    • 嚴重程度:選取「資訊」 (擷取詳細記錄)。
  6. 按一下「提交」

設定監控政策

Fabric 監控政策

  1. 依序前往「Fabric」>「Fabric Policies」>「Policies」>「Monitoring」>「Common Policy」
  2. 展開「Callhome/Smart Callhome/SNMP/Syslog/TACACS」
  3. 在「Syslog」上按一下滑鼠右鍵,然後選取「Create Syslog Source」(建立 Syslog 來源)
  4. 提供下列設定詳細資料:
    • 「Name」(名稱):輸入 Chronicle-Fabric-Syslog
    • 稽核記錄:勾選這個選項,即可納入稽核事件。
    • 事件:勾選這個選項即可納入系統事件。
    • 故障:勾選這個選項可納入故障事件。
    • 工作階段記錄:勾選即可納入工作階段記錄。
    • 目的地群組:選取 Chronicle-Syslog-Group
  5. 按一下「提交」

存取監控政策

  1. 依序前往「Fabric」>「Access Policies」>「Policies」>「Monitoring」>「Default Policy」
  2. 展開「Callhome/Smart Callhome/SNMP/Syslog」
  3. 在「Syslog」上按一下滑鼠右鍵,然後選取「Create Syslog Source」(建立 Syslog 來源)
  4. 提供下列設定詳細資料:
    • 「Name」(名稱):輸入 Chronicle-Access-Syslog
    • 稽核記錄:勾選這個選項,即可納入稽核事件。
    • 事件:勾選這個選項即可納入系統事件。
    • 故障:勾選這個選項可納入故障事件。
    • 工作階段記錄:勾選即可納入工作階段記錄。
    • 目的地群組:選取 Chronicle-Syslog-Group
  5. 按一下「提交」

設定系統 Syslog 訊息政策

  1. 依序前往「Fabric」>「Fabric Policies」>「Policies」>「Monitoring」>「Common Policy」
  2. 展開「Syslog Messages Policies」(系統記錄訊息政策)
  3. 點選「default」
  4. 在「機構篩選器」部分:
    • 設施:選取「預設」
    • 最低嚴重程度:變更為「資訊」
  5. 按一下「提交」

選項 2:透過 AWS S3 傳送 JSON

這個選項會使用 APIC REST API,從 Cisco ACI 結構收集 JSON 格式的事件、錯誤和稽核記錄,並儲存在 AWS S3 中,供 SecOps 擷取。

事前準備

  • Google SecOps 執行個體。
  • Cisco APIC 控制台的特殊存取權。
  • AWS (S3、IAM、Lambda、EventBridge) 的特殊存取權。

收集 Cisco ACI APIC 先決條件 (ID、API 金鑰、機構 ID、權杖)

  1. 使用 HTTPS 登入 Cisco APIC 控制台。
  2. 依序前往「Admin」>「AAA」 (適用於 APIC 6.0 以上版本),或「Admin」>「Authentication」>「AAA」 (適用於舊版)。
    • 注意:自 APIC 6.0(1) 起,AAA 選單路徑已變更。
  3. 建立或使用具有適當權限的現有本機使用者。
  4. 複製下列詳細資料並儲存在安全位置:
    • APIC 使用者名稱:具有監控資料讀取權的本機使用者
    • APIC 密碼:使用者密碼
    • APIC 網址:APIC 的 HTTPS 網址 (例如 https://apic.example.com)

為 Google SecOps 設定 AWS S3 值區和 IAM

  1. 按照這份使用者指南建立 Amazon S3 bucket建立 bucket
  2. 儲存 bucket 的「名稱」和「地區」,以供日後參考 (例如 cisco-aci-logs)。
  3. 請按照這份使用者指南建立使用者建立 IAM 使用者
  4. 選取建立的「使用者」
  5. 選取「安全憑證」分頁標籤。
  6. 在「Access Keys」部分中,按一下「Create Access Key」
  7. 選取「第三方服務」做為「用途」
  8. 點選「下一步」
  9. 選用:新增說明標記。
  10. 按一下「建立存取金鑰」
  11. 按一下「下載 CSV 檔案」,儲存「存取金鑰」和「私密存取金鑰」以供日後參考。
  12. 按一下 [完成]
  13. 選取「權限」分頁標籤。
  14. 在「權限政策」部分中,按一下「新增權限」
  15. 選取「新增權限」
  16. 選取「直接附加政策」
  17. 搜尋 AmazonS3FullAccess 政策。
  18. 選取政策。
  19. 點選「下一步」
  20. 按一下「Add permissions」。

設定 S3 上傳的身分與存取權管理政策和角色

  1. 在 AWS 控制台中,依序前往「IAM」>「Policies」
  2. 按一下「建立政策」>「JSON」分頁
  3. 輸入下列政策:

    {
      "Version": "2012-10-17",
      "Statement": [
        {
          "Sid": "AllowPutObjects",
          "Effect": "Allow",
          "Action": "s3:PutObject",
          "Resource": "arn:aws:s3:::cisco-aci-logs/*"
        },
        {
          "Sid": "AllowGetStateObject",
          "Effect": "Allow",
          "Action": "s3:GetObject",
          "Resource": "arn:aws:s3:::cisco-aci-logs/cisco-aci-events/state.json"
        }
      ]
    }
    
    • 如果您輸入其他 bucket 名稱,請替換 cisco-aci-logs
  4. 依序點選「下一步」>「建立政策」

  5. 依序前往「IAM」>「Roles」>「Create role」>「AWS service」>「Lambda」

  6. 附加新建立的政策和 AWSLambdaBasicExecutionRole 代管政策。

  7. 為角色命名 cisco-aci-lambda-role,然後按一下「建立角色」

建立 Lambda 函式

  1. AWS 控制台中,依序前往「Lambda」>「Functions」>「Create function」
  2. 按一下「從頭開始撰寫」
  3. 請提供下列設定詳細資料:

    • Name (名稱):cisco-aci-events-collector
    • 執行階段:Python 3.13
    • 架構:x86_64
    • 執行角色cisco-aci-lambda-role
  4. 建立函式後,開啟「程式碼」分頁,刪除存根並輸入下列程式碼 (cisco-aci-events-collector.py):

    import json
    import boto3
    import urllib3
    import base64
    from datetime import datetime, timedelta
    import os
    import logging
    
    # Configure logging
    logger = logging.getLogger()
    logger.setLevel(logging.INFO)
    
    # AWS S3 client and HTTP pool manager
    s3_client = boto3.client('s3')
    http = urllib3.PoolManager()
    
    def lambda_handler(event, context):
        """
        AWS Lambda handler to fetch Cisco ACI events, faults, and audit logs and store them in S3
        """
    
        try:
            # Get environment variables
            s3_bucket = os.environ['S3_BUCKET']
            s3_prefix = os.environ['S3_PREFIX']
            state_key = os.environ['STATE_KEY']
            apic_url = os.environ['APIC_URL']
            apic_username = os.environ['APIC_USERNAME']
            apic_password = os.environ['APIC_PASSWORD']
    
            # Optional parameters
            page_size = int(os.environ.get('PAGE_SIZE', '100'))
            max_pages = int(os.environ.get('MAX_PAGES', '10'))
    
            logger.info(f"Starting Cisco ACI data collection for bucket: {s3_bucket}")
    
            # Get last run timestamp from state file
            last_timestamp = get_last_timestamp(s3_bucket, state_key)
            if not last_timestamp:
                last_timestamp = (datetime.utcnow() - timedelta(hours=1)).isoformat() + 'Z'
    
            # Authenticate to APIC
            session_token = authenticate_apic(apic_url, apic_username, apic_password)
    
            headers = {
                'Cookie': f'APIC-cookie={session_token}',
                'Accept': 'application/json',
                'Content-Type': 'application/json'
            }
    
            # Data types to collect
            data_types = ['faultInst', 'eventRecord', 'aaaModLR']
            all_collected_data = []
    
            for data_type in data_types:
                logger.info(f"Collecting {data_type} data")
                collected_data = collect_aci_data(apic_url, headers, data_type, last_timestamp, page_size, max_pages)
    
                # Tag each record with its type
                for record in collected_data:
                    record['_data_type'] = data_type
    
                all_collected_data.extend(collected_data)
                logger.info(f"Collected {len(collected_data)} {data_type} records")
    
            logger.info(f"Total records collected: {len(all_collected_data)}")
    
            # Store data in S3 if any were collected
            if all_collected_data:
                timestamp_str = datetime.utcnow().strftime('%Y%m%d_%H%M%S')
                s3_key = f"{s3_prefix}cisco_aci_events_{timestamp_str}.ndjson"
    
                # Convert to NDJSON format (one JSON object per line)
                ndjson_content = '\n'.join(json.dumps(record) for record in all_collected_data)
    
                # Upload to S3
                s3_client.put_object(
                    Bucket=s3_bucket,
                    Key=s3_key,
                    Body=ndjson_content.encode('utf-8'),
                    ContentType='application/x-ndjson'
                )
    
                logger.info(f"Uploaded {len(all_collected_data)} records to s3://{s3_bucket}/{s3_key}")
    
            # Update state file with latest timestamp from collected data
            latest_timestamp = get_latest_timestamp_from_records(all_collected_data)
            if not latest_timestamp:
                latest_timestamp = datetime.utcnow().isoformat() + 'Z'
    
            update_state(s3_bucket, state_key, latest_timestamp)
    
            return {
                'statusCode': 200,
                'body': json.dumps({
                    'message': 'Success',
                    'total_records_collected': len(all_collected_data),
                    'data_types_collected': data_types
                })
            }
    
        except Exception as e:
            logger.error(f"Error in lambda_handler: {str(e)}")
            return {
                'statusCode': 500,
                'body': json.dumps({
                    'error': str(e)
                })
            }
    
    def authenticate_apic(apic_url, username, password):
        """
        Authenticate to APIC and return session token
        """
        login_url = f"{apic_url}/api/aaaLogin.json"
        login_data = {
            "aaaUser": {
                "attributes": {
                    "name": username,
                    "pwd": password
                }
            }
        }
    
        response = http.request(
            'POST',
            login_url,
            body=json.dumps(login_data).encode('utf-8'),
            headers={'Content-Type': 'application/json'},
            timeout=30
        )
    
        if response.status != 200:
            raise RuntimeError(f"APIC authentication failed: {response.status} {response.data[:256]!r}")
    
        response_data = json.loads(response.data.decode('utf-8'))
        token = response_data['imdata'][0]['aaaLogin']['attributes']['token']
        logger.info("Successfully authenticated to APIC")
        return token
    
    def collect_aci_data(apic_url, headers, data_type, last_timestamp, page_size, max_pages):
        """
        Collect data from APIC REST API with pagination
        """
        all_data = []
        page = 0
    
        while page < max_pages:
            # Build API URL with pagination and time filters
            api_url = f"{apic_url}/api/class/{data_type}.json"
            params = [
                f'page-size={page_size}',
                f'page={page}',
                f'order-by={data_type}.created|asc'
            ]
    
            # Add time filter for all data types to prevent duplicates
            time_attr = 'created'
            if last_timestamp:
                params.append(f'query-target-filter=gt({data_type}.{time_attr},"{last_timestamp}")')
    
            full_url = f"{api_url}?{'&'.join(params)}"
    
            logger.info(f"Fetching {data_type} page {page} from APIC")
    
            # Make API request
            response = http.request('GET', full_url, headers=headers, timeout=60)
    
            if response.status != 200:
                logger.error(f"API request failed: {response.status} {response.data[:256]!r}")
                break
    
            data = json.loads(response.data.decode('utf-8'))
            records = data.get('imdata', [])
    
            if not records:
                logger.info(f"No more {data_type} records found")
                break
    
            # Extract the actual data from APIC format
            extracted_records = []
            for record in records:
                if data_type in record:
                    extracted_records.append(record[data_type])
    
            all_data.extend(extracted_records)
            page += 1
    
            # If we got less than page_size records, we've reached the end
            if len(records) < page_size:
                break
    
        return all_data
    
    def get_last_timestamp(bucket, state_key):
        """
        Get the last run timestamp from S3 state file
        """
        try:
            response = s3_client.get_object(Bucket=bucket, Key=state_key)
            state_data = json.loads(response['Body'].read().decode('utf-8'))
            return state_data.get('last_timestamp')
        except s3_client.exceptions.NoSuchKey:
            logger.info("No state file found, starting from 1 hour ago")
            return None
        except Exception as e:
            logger.warning(f"Error reading state file: {str(e)}")
            return None
    
    def get_latest_timestamp_from_records(records):
        """
        Get the latest timestamp from collected records to prevent missing events
        """
        if not records:
            return None
    
        latest = None
        latest_time = None
    
        for record in records:
            try:
                # Handle both direct attributes and nested structure
                attrs = record.get('attributes', record)
                created = attrs.get('created')
                modTs = attrs.get('modTs')  # Fallback for some object types
    
                timestamp = created or modTs
                if timestamp:
                    if latest_time is None or timestamp > latest_time:
                        latest_time = timestamp
                        latest = record
            except Exception as e:
                logger.debug(f"Error parsing timestamp from record: {e}")
                continue
    
        return latest_time
    
    def update_state(bucket, state_key, timestamp):
        """
        Update the state file with the current timestamp
        """
        try:
            state_data = {
                'last_timestamp': timestamp,
                'updated_at': datetime.utcnow().isoformat() + 'Z'
            }
    
            s3_client.put_object(
                Bucket=bucket,
                Key=state_key,
                Body=json.dumps(state_data).encode('utf-8'),
                ContentType='application/json'
            )
    
            logger.info(f"Updated state file with timestamp: {timestamp}")
    
        except Exception as e:
            logger.error(f"Error updating state file: {str(e)}")
    
  5. 依序前往「設定」>「環境變數」

  6. 依序點選「編輯」> 新增環境變數

  7. 輸入下列環境變數,並將 換成您的值。

    • S3_BUCKETcisco-aci-logs
    • S3_PREFIXcisco-aci-events/
    • STATE_KEYcisco-aci-events/state.json
    • APIC_URLhttps://apic.example.com
    • APIC_USERNAME<your-apic-username>
    • APIC_PASSWORD<your-apic-password>
    • PAGE_SIZE100 (選用,控制分頁大小)
    • MAX_PAGES10 (選用,限制每次執行擷取的總頁數)
  8. 建立函式後,請留在該函式的頁面 (或依序開啟「Lambda」>「Functions」>「cisco-aci-events-collector」)。

  9. 選取「設定」分頁標籤。

  10. 在「一般設定」面板中,按一下「編輯」

  11. 將「Timeout」(逾時間隔) 變更為「5 minutes (300 seconds)」(5 分鐘 (300 秒)),然後按一下「Save」(儲存)

建立 EventBridge 排程

  1. 依序前往「Amazon EventBridge」>「Scheduler」>「Create schedule」
  2. 提供下列設定詳細資料:
    • 週期性時間表費率 (15 minutes)。
    • 目標:您的 Lambda 函式 cisco-aci-events-collector
    • 名稱cisco-aci-events-collector-15m
  3. 按一下「建立時間表」

選用:為 Google SecOps 建立唯讀 IAM 使用者和金鑰

  1. 依序前往 AWS 控制台 > IAM > 使用者 > 新增使用者
  2. 點選 [Add users] (新增使用者)。
  3. 提供下列設定詳細資料:
    • 使用者:輸入 secops-reader
    • 存取類型:選取「存取金鑰 - 程式輔助存取」
  4. 按一下「建立使用者」
  5. 附加最低讀取權限政策 (自訂):依序點選「Users」(使用者) >「secops-reader」>「Permissions」(權限) >「Add permissions」(新增權限) >「Attach policies directly」(直接附加政策) >「Create policy」(建立政策)
  6. 在 JSON 編輯器中輸入下列政策:

    {
      "Version": "2012-10-17",
      "Statement": [
        {
          "Effect": "Allow",
          "Action": ["s3:GetObject"],
          "Resource": "arn:aws:s3:::cisco-aci-logs/*"
        },
        {
          "Effect": "Allow",
          "Action": ["s3:ListBucket"],
          "Resource": "arn:aws:s3:::cisco-aci-logs"
        }
      ]
    }
    
  7. 將名稱設為 secops-reader-policy

  8. 依序前往「建立政策」> 搜尋/選取 >「下一步」>「新增權限」

  9. 依序前往「安全憑證」>「存取金鑰」>「建立存取金鑰」

  10. 下載 CSV (這些值會輸入至動態饋給)。

在 Google SecOps 中設定動態饋給,擷取 Cisco ACI 記錄

  1. 依序前往「SIEM 設定」>「動態饋給」
  2. 按一下「+ 新增動態消息」
  3. 在「動態饋給名稱」欄位中輸入動態饋給名稱 (例如 Cisco ACI JSON logs)。
  4. 選取「Amazon S3 V2」做為「來源類型」
  5. 選取「Cisco Application Centric Infrastructure」做為「記錄類型」
  6. 點選「下一步」
  7. 指定下列輸入參數的值:
    • S3 URIs3://cisco-aci-logs/cisco-aci-events/
    • 來源刪除選項:根據偏好設定選取刪除選項。
    • 檔案存在時間上限:包含在過去天數內修改的檔案。預設值為 180 天。
    • 存取金鑰 ID:具有 S3 儲存空間存取權的使用者存取金鑰。
    • 存取密鑰:具有 S3 bucket 存取權的使用者私密金鑰。
    • 資產命名空間資產命名空間
    • 擷取標籤:套用至這個動態饋給事件的標籤。
  8. 點選「下一步」
  9. 在「完成」畫面中檢查新的動態饋給設定,然後按一下「提交」

UDM 對應表

記錄欄位 UDM 對應 邏輯
@timestamp read_only_udm.metadata.event_timestamp 值取自原始記錄欄位「@timestamp」,並剖析為時間戳記。
aci_tag read_only_udm.metadata.product_log_id 值取自原始記錄欄位「aci_tag」。
cisco_timestamp - 未對應。
DIP read_only_udm.target.ip 值取自原始記錄欄位「DIP」。
DPort read_only_udm.target.port 值取自原始記錄欄位「DPort」,並轉換為整數。
說明 read_only_udm.security_result.description 值取自原始記錄檔欄位「description」。
fault_cause read_only_udm.additional.fields.value.string_value 值取自原始記錄欄位「fault_cause」。金鑰設為「Fault Cause」。
主機名稱 read_only_udm.principal.hostname 值取自原始記錄檔欄位「hostname」。
lifecycle_state read_only_udm.metadata.product_event_type 值取自原始記錄欄位「lifecycle_state」。
log.source.address - 未對應。
logstash.collect.host - 未對應。
logstash.collect.timestamp read_only_udm.metadata.collected_timestamp 值取自原始記錄欄位「logstash.collect.timestamp」,並剖析為時間戳記。
logstash.ingest.host read_only_udm.intermediary.hostname 值取自原始記錄檔欄位「logstash.ingest.host」。
logstash.irm_environment read_only_udm.additional.fields.value.string_value 值取自原始記錄檔欄位「logstash.irm_environment」。金鑰設為「IRM_Environment」。
logstash.irm_region read_only_udm.additional.fields.value.string_value 值取自原始記錄欄位「logstash.irm_region」。金鑰設為「IRM_Region」。
logstash.irm_site read_only_udm.additional.fields.value.string_value 值取自原始記錄檔欄位「logstash.irm_site」。金鑰設為「IRM_Site」。
logstash.process.host read_only_udm.intermediary.hostname 值取自原始記錄欄位「logstash.process.host」。
訊息 - 未對應。
message_class - 未對應。
message_code - 未對應。
message_content - 未對應。
message_dn - 未對應。
message_type read_only_udm.metadata.product_event_type 從原始記錄檔欄位「message_type」中移除方括號後,即可取得值。
node_link read_only_udm.principal.process.file.full_path 值取自原始記錄欄位「node_link」。
PktLen read_only_udm.network.received_bytes 值取自原始記錄欄位「PktLen」,並轉換為無正負號整數。
計畫 - 未對應。
Proto read_only_udm.network.ip_protocol 值取自原始記錄欄位「Proto」,轉換為整數,並對應至相應的 IP 通訊協定名稱 (例如 6 -> TCP)。
SIP read_only_udm.principal.ip 值取自原始記錄檔欄位「SIP」。
SPort read_only_udm.principal.port 值取自原始記錄欄位「SPort」,並轉換為整數。
syslog_facility - 未對應。
syslog_facility_code - 未對應。
syslog_host read_only_udm.principal.ip、read_only_udm.observer.ip 值取自原始記錄欄位「syslog_host」。
syslog_prog - 未對應。
syslog_severity read_only_udm.security_result.severity_details 值取自原始記錄檔欄位「syslog_severity」。
syslog_severity_code read_only_udm.security_result.severity 值取自原始記錄欄位「syslog_severity_code」,並對應至相應的嚴重程度:5、6、7 -> INFORMATIONAL;3、4 -> MEDIUM;0、1、2 -> HIGH。
syslog5424_pri - 未對應。
Vlan-Id read_only_udm.principal.resource.id 值取自原始記錄檔欄位「Vlan-Id」。
- read_only_udm.metadata.event_type 邏輯:如果存在「SIP」或「主機名稱」,且存在「Proto」,則設為「NETWORK_CONNECTION」。否則,如果存在「SIP」、「hostname」或「syslog_host」,請設為「STATUS_UPDATE」。否則請設為「GENERIC_EVENT」。
- read_only_udm.metadata.log_type 邏輯:設為「CISCO_ACI」。
- read_only_udm.metadata.vendor_name 邏輯:設為「Cisco」。
- read_only_udm.metadata.product_name 邏輯:設為「ACI」。

還有其他問題嗎?向社群成員和 Google SecOps 專業人員尋求答案。