Pub/Sub 到 Splunk 範本

「將 Pub/Sub 的內容傳輸到 Splunk 範本」是一種串流管道,可讀取 Pub/Sub 訂閱的訊息,並透過 Splunk 的 HTTP Event Collector (HEC) 將訊息酬載寫入 Splunk。這個範本最常見的用途是將記錄匯出至 Splunk。如要查看基礎工作流程的範例,請參閱「使用 Dataflow 將可用於實際工作的記錄檔匯出作業部署至 Splunk」。

您也可以在寫入 Splunk 前,將 JavaScript 使用者定義函式套用至訊息酬載。系統會將發生處理問題的所有訊息轉送至未處理的 Pub/Sub 主題,以便進一步執行疑難排解和重新處理。

為了增加多一層安全防護,您也可以傳遞 Cloud KMS 金鑰,以及使用 Cloud KMS 金鑰加密的 Base64 編碼 HEC 權杖參數。如要進一步瞭解如何加密 HEC 權杖參數,請參閱 Cloud KMS API 加密端點

管道相關規定

  • 執行管道之前,來源 Pub/Sub 訂閱項目必須已經存在。
  • 執行管道之前,Pub/Sub 未處理主題必須已經存在。
  • Splunk HEC 端點必須可透過 Dataflow 工作站的網路存取。
  • 必須產生 Splunk HEC 權杖且該權杖須可供使用。

範本參數

必要參數

  • inputSubscription:要讀取輸入內容的 Pub/Sub 訂閱項目。例如:projects/your-project-id/subscriptions/your-subscription-name
  • url:Splunk HEC 網址。網址必須可從管道執行的虛擬私有雲 (VPC) 路由。例如:https://splunk-hec-host:8088
  • outputDeadletterTopic:要將無法傳送的訊息轉送至的 Pub/Sub 主題。例如:projects/<PROJECT_ID>/topics/<TOPIC_NAME>

選用參數

  • token:Splunk HEC 驗證權杖。如果 tokenSource 參數設為 PLAINTEXTKMS,則必須提供這項屬性。
  • batchCount:將多個事件傳送至 Splunk 的批次大小。預設值為 1 (不分批)。
  • disableCertificateValidation:停用 SSL 憑證驗證。預設 false (已啟用驗證)。如果 true,系統不會驗證憑證 (所有憑證都受信任),並會忽略 rootCaCertificatePath 參數。
  • 平行處理量:平行要求數量上限。預設值為 1 (無平行處理)。
  • includePubsubMessage:在酬載中加入完整的 Pub/Sub 訊息。預設 false (酬載中只會包含資料元素)。
  • tokenKMSEncryptionKey:用來解密 HEC 權杖字串的 Cloud KMS 金鑰。如果 tokenSource 設為 KMS,就必須提供這項參數。如果提供 Cloud KMS 金鑰,則必須以加密方式傳送 HEC 權杖字串。例如:projects/your-project-id/locations/global/keyRings/your-keyring/cryptoKeys/your-key-name
  • tokenSecretId:權杖的 Secret Manager 密鑰 ID。如果 tokenSource 設為 SECRET_MANAGER,就必須提供這項參數。例如:projects/your-project-id/secrets/your-secret/versions/your-secret-version
  • tokenSource:權杖來源。允許的值包括:PLAINTEXTKMSSECRET_MANAGER。使用 Secret Manager 時,您必須提供這個參數。如果 tokenSource 設為 KMStokenKMSEncryptionKey 並經過加密,則必須提供 token。如果 tokenSource 設為 SECRET_MANAGER,則必須提供 tokenSecretId。如果 tokenSource 設為 PLAINTEXT,則必須提供 token
  • rootCaCertificatePath:Cloud Storage 中根 CA 憑證的完整網址。Cloud Storage 中提供的憑證必須採用 DER 編碼,且可採用二進位或可列印 (Base64) 編碼。如果憑證是以 Base64 編碼提供,開頭必須為「-----BEGIN CERTIFICATE-----」,結尾必須為「-----END CERTIFICATE-----」。如果提供這個參數,系統會擷取這個私有 CA 憑證檔案,並新增至 Dataflow 工作人員的信任存放區,以便驗證 Splunk HEC 端點的 SSL 憑證。如未提供這個參數,系統會使用預設的信任儲存區。例如:gs://mybucket/mycerts/privateCA.crt
  • enableBatchLogs:指定是否應為寫入 Splunk 的批次啟用記錄。預設值:true
  • enableGzipHttpCompression:指定傳送至 Splunk HEC 的 HTTP 要求是否應壓縮 (gzip 內容編碼)。預設值:true
  • javascriptTextTransformGcsPath:定義要使用的 JavaScript 使用者定義函式 (UDF) 的 .js 檔案 Cloud Storage URI。例如:gs://my-bucket/my-udfs/my_file.js
  • javascriptTextTransformFunctionName:要使用的 JavaScript 使用者定義函式 (UDF) 名稱。舉例來說,如果您的 JavaScript 函式程式碼是 myTransform(inJson) { /*...do stuff...*/ },則函式名稱就是 myTransform。如需 JavaScript UDF 範例,請參閱 UDF 範例 (https://github.com/GoogleCloudPlatform/DataflowTemplates#udf-examples)。
  • javascriptTextTransformReloadIntervalMinutes:定義工作人員檢查 JavaScript UDF 變更以重新載入檔案的間隔。預設值為 0。

使用者定義函式

您可以視需要撰寫使用者定義函式 (UDF) 來擴充這個範本。範本會針對每個輸入元素呼叫 UDF。元素酬載會序列化為 JSON 字串。詳情請參閱「為 Dataflow 範本建立使用者定義函式」。

函式規格

UDF 的規格如下:

  • 輸入:Pub/Sub 訊息資料欄位, 序列化為 JSON 字串。
  • 輸出:要傳送至 Splunk HEC 事件端點的事件資料。輸出內容必須是字串或字串化的 JSON 物件。

執行範本

控制台

  1. 前往 Dataflow 的「Create job from template」(透過範本建立工作) 頁面。
  2. 前往「依據範本建立工作」
  3. 在「工作名稱」欄位中,輸入專屬工作名稱。
  4. 選用:如要使用區域端點,請從下拉式選單中選取值。預設區域為 us-central1

    如需可執行 Dataflow 工作的地區清單,請參閱「Dataflow 位置」。

  5. 從「Dataflow template」(Dataflow 範本) 下拉式選單中選取 the Pub/Sub to Splunk template。
  6. 在提供的參數欄位中輸入參數值。
  7. 選用:如要從「僅需處理一次」切換至「至少一次」串流模式,請選取「至少一次」
  8. 按一下「Run Job」(執行工作)

gcloud

在殼層或終端機中執行範本:

gcloud dataflow jobs run JOB_NAME \
    --gcs-location gs://dataflow-templates-REGION_NAME/VERSION/Cloud_PubSub_to_Splunk \
    --region REGION_NAME \
    --staging-location STAGING_LOCATION \
    --parameters \
inputSubscription=projects/PROJECT_ID/subscriptions/INPUT_SUBSCRIPTION_NAME,\
token=TOKEN,\
url=URL,\
outputDeadletterTopic=projects/PROJECT_ID/topics/DEADLETTER_TOPIC_NAME,\
javascriptTextTransformGcsPath=PATH_TO_JAVASCRIPT_UDF_FILE,\
javascriptTextTransformFunctionName=JAVASCRIPT_FUNCTION,\
batchCount=BATCH_COUNT,\
parallelism=PARALLELISM,\
disableCertificateValidation=DISABLE_VALIDATION,\
rootCaCertificatePath=ROOT_CA_CERTIFICATE_PATH

更改下列內容:

  • JOB_NAME: 您選擇的不重複工作名稱
  • REGION_NAME: 您要部署 Dataflow 工作的地區,例如 us-central1
  • VERSION: 您要使用的範本版本

    您可以使用下列值:

  • STAGING_LOCATION:用於暫存本機檔案的位置 (例如 gs://your-bucket/staging)
  • INPUT_SUBSCRIPTION_NAME:Pub/Sub 訂閱項目名稱
  • TOKEN:Splunk 的 HTTP 事件收集器權杖
  • URL:Splunk HTTP 事件收集器的網址路徑 (例如 https://splunk-hec-host:8088)
  • DEADLETTER_TOPIC_NAME:Pub/Sub 主題名稱
  • JAVASCRIPT_FUNCTION: 您要使用的 JavaScript 使用者定義函式 (UDF) 名稱

    舉例來說,如果您的 JavaScript 函式程式碼是 myTransform(inJson) { /*...do stuff...*/ },則函式名稱就是 myTransform。如需 JavaScript UDF 範例,請參閱「UDF 範例」。

  • PATH_TO_JAVASCRIPT_UDF_FILE: 定義要使用的 JavaScript 使用者定義函式 (UDF) 的 .js 檔案 Cloud Storage URI,例如 gs://my-bucket/my-udfs/my_file.js
  • BATCH_COUNT:將多個事件傳送至 Splunk 時使用的批次大小
  • PARALLELISM:用於將事件傳送至 Splunk 的平行要求數量
  • DISABLE_VALIDATIONtrue 如果要停用 SSL 憑證驗證
  • ROOT_CA_CERTIFICATE_PATH:Cloud Storage 中根 CA 憑證的路徑 (例如 gs://your-bucket/privateCA.crt)

API

如要使用 REST API 執行範本,請傳送 HTTP POST 要求。如要進一步瞭解 API 和授權範圍,請參閱 projects.templates.launch

POST https://dataflow.googleapis.com/v1b3/projects/PROJECT_ID/locations/LOCATION/templates:launch?gcsPath=gs://dataflow-templates-LOCATION/VERSION/Cloud_PubSub_to_Splunk
{
   "jobName": "JOB_NAME",
   "environment": {
       "ipConfiguration": "WORKER_IP_UNSPECIFIED",
       "additionalExperiments": []
   },
   "parameters": {
       "inputSubscription": "projects/PROJECT_ID/subscriptions/INPUT_SUBSCRIPTION_NAME",
       "token": "TOKEN",
       "url": "URL",
       "outputDeadletterTopic": "projects/PROJECT_ID/topics/DEADLETTER_TOPIC_NAME",
       "javascriptTextTransformGcsPath": "PATH_TO_JAVASCRIPT_UDF_FILE",
       "javascriptTextTransformFunctionName": "JAVASCRIPT_FUNCTION",
       "batchCount": "BATCH_COUNT",
       "parallelism": "PARALLELISM",
       "disableCertificateValidation": "DISABLE_VALIDATION",
       "rootCaCertificatePath": "ROOT_CA_CERTIFICATE_PATH"
   }
}

更改下列內容:

  • PROJECT_ID: 您要執行 Dataflow 工作的專案 ID Google Cloud
  • JOB_NAME: 您選擇的不重複工作名稱
  • LOCATION: 您要部署 Dataflow 工作的地區,例如 us-central1
  • VERSION: 您要使用的範本版本

    您可以使用下列值:

  • STAGING_LOCATION:用於暫存本機檔案的位置 (例如 gs://your-bucket/staging)
  • INPUT_SUBSCRIPTION_NAME:Pub/Sub 訂閱項目名稱
  • TOKEN:Splunk 的 HTTP 事件收集器權杖
  • URL:Splunk HTTP 事件收集器的網址路徑 (例如 https://splunk-hec-host:8088)
  • DEADLETTER_TOPIC_NAME:Pub/Sub 主題名稱
  • JAVASCRIPT_FUNCTION: 您要使用的 JavaScript 使用者定義函式 (UDF) 名稱

    舉例來說,如果您的 JavaScript 函式程式碼是 myTransform(inJson) { /*...do stuff...*/ },則函式名稱就是 myTransform。如需 JavaScript UDF 範例,請參閱「UDF 範例」。

  • PATH_TO_JAVASCRIPT_UDF_FILE: 定義要使用的 JavaScript 使用者定義函式 (UDF) 的 .js 檔案 Cloud Storage URI,例如 gs://my-bucket/my-udfs/my_file.js
  • BATCH_COUNT:將多個事件傳送至 Splunk 時使用的批次大小
  • PARALLELISM:用於將事件傳送至 Splunk 的平行要求數量
  • DISABLE_VALIDATIONtrue 如果要停用 SSL 憑證驗證
  • ROOT_CA_CERTIFICATE_PATH:Cloud Storage 中根 CA 憑證的路徑 (例如 gs://your-bucket/privateCA.crt)

後續步驟