תבנית Pub/Sub ל-Splunk

תבנית Pub/Sub to Splunk היא צינור סטרימינג שקורא הודעות ממינוי Pub/Sub וכותב את המטען הייעודי (payload) של ההודעה ל-Splunk באמצעות HTTP Event Collector ‏(HEC) של Splunk. התרחיש הנפוץ ביותר לשימוש בתבנית הזו הוא ייצוא יומנים ל-Splunk. דוגמה לתהליך העבודה הבסיסי מופיעה במאמר פריסת ייצוא יומנים שמוכן לייצור ב-Splunk באמצעות Dataflow.

לפני הכתיבה ל-Splunk, אפשר גם להחיל פונקציה בהגדרת המשתמש (UDF) ב-JavaScript על מטען ייעודי (payload) של ההודעה. הודעות שעיבוד שלהן נכשל מועברות לנושא לא מעובד ב-Pub/Sub, כדי לפתור את הבעיה ולעבד אותן מחדש.

כדי להוסיף שכבת הגנה לאסימון HEC, אפשר גם להעביר מפתח Cloud KMS יחד עם פרמטר האסימון HEC שמקודד ב-Base64 ומוצפן באמצעות מפתח Cloud KMS. פרטים נוספים על הצפנה של פרמטר אסימון 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: כתובת ה-URL של Splunk HEC. צריך להיות אפשר לנתב את כתובת ה-URL מ-VPC שצינור עיבוד הנתונים פועל בו. לדוגמה, https://splunk-hec-host:8088.
• ‫outputDeadletterTopic: נושא Pub/Sub להעברת הודעות שלא ניתן למסור. לדוגמה, projects/<PROJECT_ID>/topics/<TOPIC_NAME>.

פרמטרים אופציונליים

• ‫token: אסימון האימות של Splunk HEC. חובה לספק את הפרמטר הזה אם הפרמטר tokenSource מוגדר לערך PLAINTEXT או KMS.
• ‫batchCount: גודל האצווה לשליחת מספר אירועים אל Splunk. ברירת המחדל היא 1 (ללא חלוקה לקבוצות).
• ‫disableCertificateValidation: השבתת אימות אישור SSL. ברירת מחדל false (אימות מופעל). אם הערך הוא true, האישורים לא מאומתים (כל האישורים מהימנים) והמערכת מתעלמת מהפרמטר rootCaCertificatePath.
• מקביליות: המספר המקסימלי של בקשות מקבילות. ברירת המחדל היא 1 (ללא מקביליות).
• ‫includePubsubMessage: הכללת ההודעה המלאה של Pub/Sub במטען הייעודי (payload). ברירת המחדל היא false (רק רכיב הנתונים נכלל במטען הייעודי).
• ‫tokenKMSEncryptionKey: מפתח Cloud KMS שמשמש לפענוח מחרוזת האסימון של HEC. חובה לספק את הפרמטר הזה כשמגדירים את tokenSource ל-KMS. אם מספקים את מפתח Cloud KMS, צריך להעביר את מחרוזת האסימון של HEC בצורה מוצפנת. לדוגמה, projects/your-project-id/locations/global/keyRings/your-keyring/cryptoKeys/your-key-name.
• ‫tokenSecretId: המזהה הסודי של הטוקן ב-Secret Manager. חובה לספק את הפרמטר הזה כשמגדירים את tokenSource לערך SECRET_MANAGER. לדוגמה, projects/your-project-id/secrets/your-secret/versions/your-secret-version.
• ‫tokenSource: המקור של הטוקן. הערכים המותרים: PLAINTEXT,‏ KMS ו-SECRET_MANAGER. חובה לציין את הפרמטר הזה כשמשתמשים ב-Secret Manager. אם הערך של tokenSource הוא KMS, ‏tokenKMSEncryptionKey והוא מוצפן, צריך לספק את token. אם הערך של tokenSource הוא SECRET_MANAGER, צריך לציין את tokenSecretId. אם הערך של tokenSource הוא PLAINTEXT, צריך לציין את token.
• ‫rootCaCertificatePath: כתובת ה-URL המלאה לאישור ה-CA הבסיסי ב-Cloud Storage. האישור שמועבר ב-Cloud Storage צריך להיות בקידוד DER, ואפשר לספק אותו בקידוד בינארי או בקידוד שניתן להדפסה (Base64). אם האישור מסופק בקידוד Base64, הוא צריך להתחיל ב-‎-----BEGIN CERTIFICATE-----‎ ולהסתיים ב-‎-----END CERTIFICATE-----‎. אם הפרמטר הזה מסופק, קובץ אישור ה-CA הפרטי מאוחזר ומוסף למאגר האישורים של העובד ב-Dataflow כדי לאמת את אישור ה-SSL של נקודת הקצה של Splunk HEC. אם לא מציינים את הפרמטר הזה, נעשה שימוש במאגר אישורים ברירת המחדל. לדוגמה, gs://mybucket/mycerts/privateCA.crt.
• ‫enableBatchLogs: מציין אם להפעיל יומנים עבור קבוצות של נתונים שנכתבים ב-Splunk. ברירת מחדל: true.
• ‫enableGzipHttpCompression: מציין אם בקשות HTTP שנשלחות ל-Splunk HEC צריכות להיות דחוסות (תוכן בקידוד gzip). ברירת מחדל: true.
• ‫javascriptTextTransformGcsPath: ה-URI של Cloud Storage של קובץ ה-‎ .js שמגדיר את הפונקציה בהגדרת המשתמש (UDF) ב-JavaScript שבה רוצים להשתמש. לדוגמה, gs://my-bucket/my-udfs/my_file.js.
• ‫javascriptTextTransformFunctionName: השם של פונקציית JavaScript בהגדרת המשתמש (UDF) שבה רוצים להשתמש. לדוגמה, אם קוד הפונקציה ב-JavaScript הוא myTransform(inJson) { /*...do stuff...*/ }, אז שם הפונקציה הוא myTransform. דוגמאות לפונקציות מוגדרות על ידי המשתמש (UDF) ב-JavaScript זמינות במאמר UDF Examples (https://github.com/GoogleCloudPlatform/DataflowTemplates#udf-examples).
• ‫javascriptTextTransformReloadIntervalMinutes: הגדרה של המרווח שבו עובדים יכולים לבדוק אם יש שינויים בפונקציות UDF של JavaScript כדי לטעון מחדש את הקבצים. ברירת המחדל היא 0.
• ‫unwrapHecForDeadletter: אם האפשרות הזו מופעלת, והודעה לא נכתבת ב-Splunk ונשלחת לתור של הודעות שלא ניתן להעביר, המטען הייעודי (payload) המקורי יחולץ מפורמט Splunk HEC לפני הכתיבה לנושא של הודעות שלא ניתן להעביר. כך נמנעת הוספה של אירועים בתוך אירועים אחרים כשמפעילים מחדש הודעות. ברירת מחדל: false.

פונקציה בהגדרת המשתמש

אפשר גם להרחיב את התבנית הזו על ידי כתיבת פונקציה בהגדרת המשתמש (UDF). התבנית קוראת ל-UDF עבור כל רכיב קלט. מטענים ייעודיים של רכיבים עוברים סריאליזציה כמחרוזות JSON. למידע נוסף, ראו יצירת פונקציות מוגדרות על ידי המשתמש לתבניות Dataflow.

מפרט הפונקציה

המאפיינים של פונקציית UDF:

• קלט: שדה הנתונים של הודעת Pub/Sub, שעבר סריאליזציה כמחרוזת JSON.
• פלט: נתוני האירוע שיישלחו לנקודת הקצה (endpoint) של אירועים ב-Splunk HEC. הפלט חייב להיות מחרוזת או אובייקט JSON במחרוזת.

הפעלת התבנית

המסוף

1. עוברים לדף Dataflow Create job from template (יצירת משימה מתבנית).
כניסה לדף Create job from template
2. בשדה שם המשימה, מזינים שם ייחודי למשימה.
3. אופציונלי: בשדה Regional endpoint (נקודת קצה אזורית), בוחרים ערך מהתפריט הנפתח. אזור ברירת המחדל הוא us-central1.

   רשימת האזורים שבהם אפשר להריץ משימת Dataflow מופיעה במאמר מיקומי Dataflow.

4. בתפריט הנפתח Dataflow template (תבנית של העברת נתונים), בוחרים באפשרות the Pub/Sub to Splunk template.
5. בשדות הפרמטרים שמופיעים, מזינים את ערכי הפרמטרים.
6. אופציונלי: כדי לעבור מעיבוד של כל נתון בדיוק פעם אחת אל מצב סטרימינג של כל נתון לפחות פעם אחת, בוחרים באפשרות לפחות פעם אחת.
7. לוחצים על הפעלת העבודה.

gcloud

הערה: כדי להשתמש ב-Google Cloud CLI להפעלת תבניות classic, צריך להתקין את Google Cloud CLI בגרסה 138.0.0, או בגרסה מאוחרת יותר.

במעטפת או בטרמינל, מריצים את התבנית:

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: הגרסה של התבנית שבה רוצים להשתמש

  אפשר להשתמש בערכים הבאים:

  • ‫latest כדי להשתמש בגרסה העדכנית של התבנית, שזמינה בתיקיית ההורה ללא תאריך בדלי – gs://dataflow-templates-REGION_NAME/latest/
  • שם הגרסה, כמו 2023-09-12-00_RC00, כדי להשתמש בגרסה ספציפית של התבנית, שאפשר למצוא אותה בתיקיית האב המתאימה עם התאריך בדלי – gs://dataflow-templates-REGION_NAME/
  זהירות: יכול להיות שגרסת התבניות העדכנית תתעדכן עם שינויים שעלולים לשבור את התאימות לאחור. כדי למנוע מהשינויים האלה לשבש את תהליכי העבודה בסביבת הייצור, צריך להשתמש בתבניות שנשמרות בתיקייה הראשית עם תאריך העדכנית ביותר בסביבות הייצור.
• ‫STAGING_LOCATION: המיקום של קבצים מקומיים להעברה (לדוגמה, gs://your-bucket/staging)
• ‫INPUT_SUBSCRIPTION_NAME: השם של המינוי ל-Pub/Sub
• ‫TOKEN: אסימון Splunk Http Event Collector
• ‫URL: נתיב כתובת ה-URL של Splunk's Http Event Collector (לדוגמה, https://splunk-hec-host:8088)
• ‫DEADLETTER_TOPIC_NAME: שם נושא ה-Pub/Sub
• ‫JAVASCRIPT_FUNCTION: השם של פונקציית JavaScript בהגדרת המשתמש (UDF) שרוצים להשתמש בה

  לדוגמה, אם קוד הפונקציה ב-JavaScript הוא myTransform(inJson) { /*...do stuff...*/ }, אז שם הפונקציה הוא myTransform. דוגמאות ל-UDF ב-JavaScript זמינות במאמר דוגמאות ל-UDF.

• ‫PATH_TO_JAVASCRIPT_UDF_FILE: ה-URI של Cloud Storage של קובץ .js שמגדיר את הפונקציה בהגדרת המשתמש (UDF) ב-JavaScript שרוצים להשתמש בה – לדוגמה, gs://my-bucket/my-udfs/my_file.js
• ‫BATCH_COUNT: גודל האצווה שמשמש לשליחת מספר אירועים ל-Splunk
• ‫PARALLELISM: מספר הבקשות המקבילות שמשמשות לשליחת אירועים אל Splunk
• ‫DISABLE_VALIDATION: true אם רוצים להשבית את האימות של אישור ה-SSL
• ROOT_CA_CERTIFICATE_PATH: הנתיב לאישור CA הבסיסי ב-Cloud Storage (לדוגמה, gs://your-bucket/privateCA.crt)

API

כדי להריץ את התבנית באמצעות API בארכיטקטורת REST, שולחים בקשת 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 Google Cloud
• ‫JOB_NAME: שם ייחודי של המשימה לפי בחירתכם
• ‫LOCATION: האזור שבו רוצים לפרוס את עבודת Dataflow, לדוגמה: us-central1
• ‫VERSION: הגרסה של התבנית שבה רוצים להשתמש

  אפשר להשתמש בערכים הבאים:

  • ‫latest כדי להשתמש בגרסה העדכנית של התבנית, שזמינה בתיקיית ההורה ללא תאריך בדלי – gs://dataflow-templates-REGION_NAME/latest/
  • שם הגרסה, כמו 2023-09-12-00_RC00, כדי להשתמש בגרסה ספציפית של התבנית, שאפשר למצוא אותה בתיקיית האב המתאימה עם התאריך בדלי – gs://dataflow-templates-REGION_NAME/
  זהירות: יכול להיות שגרסת התבניות העדכנית תתעדכן עם שינויים שעלולים לשבור את התאימות לאחור. כדי למנוע מהשינויים האלה לשבש את תהליכי העבודה בסביבת הייצור, צריך להשתמש בתבניות שנשמרות בתיקייה הראשית עם תאריך העדכנית ביותר בסביבות הייצור.
• ‫STAGING_LOCATION: המיקום של קבצים מקומיים להעברה (לדוגמה, gs://your-bucket/staging)
• ‫INPUT_SUBSCRIPTION_NAME: השם של המינוי ל-Pub/Sub
• ‫TOKEN: אסימון Splunk Http Event Collector
• ‫URL: נתיב כתובת ה-URL של Splunk's Http Event Collector (לדוגמה, https://splunk-hec-host:8088)
• ‫DEADLETTER_TOPIC_NAME: שם נושא ה-Pub/Sub
• ‫JAVASCRIPT_FUNCTION: השם של פונקציית JavaScript בהגדרת המשתמש (UDF) שרוצים להשתמש בה

  לדוגמה, אם קוד הפונקציה ב-JavaScript הוא myTransform(inJson) { /*...do stuff...*/ }, אז שם הפונקציה הוא myTransform. דוגמאות ל-UDF ב-JavaScript זמינות במאמר דוגמאות ל-UDF.

• ‫PATH_TO_JAVASCRIPT_UDF_FILE: ה-URI של Cloud Storage של קובץ .js שמגדיר את הפונקציה בהגדרת המשתמש (UDF) ב-JavaScript שרוצים להשתמש בה – לדוגמה, gs://my-bucket/my-udfs/my_file.js
• ‫BATCH_COUNT: גודל האצווה שמשמש לשליחת מספר אירועים ל-Splunk
• ‫PARALLELISM: מספר הבקשות המקבילות שמשמשות לשליחת אירועים אל Splunk
• ‫DISABLE_VALIDATION: true אם רוצים להשבית את האימות של אישור ה-SSL
• ROOT_CA_CERTIFICATE_PATH: הנתיב לאישור CA הבסיסי ב-Cloud Storage (לדוגמה, gs://your-bucket/privateCA.crt)

קוד המקור של התבנית

Java

/*
 * Copyright (C) 2019 Google LLC
 *
 * Licensed under the Apache License, Version 2.0 (the "License"); you may not
 * use this file except in compliance with the License. You may obtain a copy of
 * the License at
 *
 *   http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
 * WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
 * License for the specific language governing permissions and limitations under
 * the License.
 */
package com.google.cloud.teleport.templates;

import com.google.cloud.teleport.coders.FailsafeElementCoder;
import com.google.cloud.teleport.metadata.Template;
import com.google.cloud.teleport.metadata.TemplateCategory;
import com.google.cloud.teleport.metadata.TemplateParameter;
import com.google.cloud.teleport.splunk.SplunkEvent;
import com.google.cloud.teleport.splunk.SplunkEventCoder;
import com.google.cloud.teleport.splunk.SplunkIO;
import com.google.cloud.teleport.splunk.SplunkWriteError;
import com.google.cloud.teleport.templates.PubSubToSplunk.PubSubToSplunkOptions;
import com.google.cloud.teleport.templates.common.ErrorConverters;
import com.google.cloud.teleport.templates.common.JavascriptTextTransformer.FailsafeJavascriptUdf;
import com.google.cloud.teleport.templates.common.JavascriptTextTransformer.JavascriptTextTransformerOptions;
import com.google.cloud.teleport.templates.common.PubsubConverters.PubsubReadSubscriptionOptions;
import com.google.cloud.teleport.templates.common.PubsubConverters.PubsubWriteDeadletterTopicOptions;
import com.google.cloud.teleport.templates.common.SplunkConverters;
import com.google.cloud.teleport.templates.common.SplunkConverters.SplunkOptions;
import com.google.cloud.teleport.util.TokenNestedValueProvider;
import com.google.cloud.teleport.values.FailsafeElement;
import com.google.common.annotations.VisibleForTesting;
import com.google.gson.Gson;
import com.google.gson.JsonObject;
import com.google.gson.JsonSyntaxException;
import java.nio.charset.StandardCharsets;
import java.util.Map;
import org.apache.beam.sdk.Pipeline;
import org.apache.beam.sdk.PipelineResult;
import org.apache.beam.sdk.coders.CoderRegistry;
import org.apache.beam.sdk.coders.StringUtf8Coder;
import org.apache.beam.sdk.io.gcp.pubsub.PubsubIO;
import org.apache.beam.sdk.io.gcp.pubsub.PubsubMessage;
import org.apache.beam.sdk.metrics.Counter;
import org.apache.beam.sdk.metrics.Metrics;
import org.apache.beam.sdk.options.PipelineOptionsFactory;
import org.apache.beam.sdk.options.ValueProvider;
import org.apache.beam.sdk.transforms.DoFn;
import org.apache.beam.sdk.transforms.Flatten;
import org.apache.beam.sdk.transforms.MapElements;
import org.apache.beam.sdk.transforms.PTransform;
import org.apache.beam.sdk.transforms.ParDo;
import org.apache.beam.sdk.values.PBegin;
import org.apache.beam.sdk.values.PCollection;
import org.apache.beam.sdk.values.PCollectionList;
import org.apache.beam.sdk.values.PCollectionTuple;
import org.apache.beam.sdk.values.TupleTag;
import org.apache.beam.vendor.guava.v32_1_2_jre.com.google.common.base.MoreObjects;
import org.apache.beam.vendor.guava.v32_1_2_jre.com.google.common.collect.ImmutableList;
import org.json.JSONException;
import org.json.JSONObject;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

/**
 * The {@link PubSubToSplunk} pipeline is a streaming pipeline which ingests data from Cloud
 * Pub/Sub, executes a UDF, converts the output to {@link SplunkEvent}s and writes those records
 * into Splunk's HEC endpoint. Any errors which occur in the execution of the UDF, conversion to
 * {@link SplunkEvent} or writing to HEC will be streamed into a Pub/Sub topic.
 *
 * <p><b>Pipeline Requirements</b>
 *
 * <ul>
 *   <li>The source Pub/Sub subscription exists.
 *   <li>HEC end-point is routable from the VPC where the Dataflow job executes.
 *   <li>Deadletter topic exists.
 * </ul>
 *
 * <p>Check out <a
 * href="https://github.com/GoogleCloudPlatform/DataflowTemplates/blob/main/v1/README_Cloud_PubSub_to_Splunk.md">README</a>
 * for instructions on how to use or modify this template.
 */
@Template(
    name = "Cloud_PubSub_to_Splunk",
    category = TemplateCategory.STREAMING,
    displayName = "Pub/Sub to Splunk",
    description = {
      "The Pub/Sub to Splunk template is a streaming pipeline that reads messages from a Pub/Sub subscription and writes the message payload to Splunk via Splunk's HTTP Event Collector (HEC). The most common use case of this template is to export logs to Splunk. "
          + "To see an example of the underlying workflow, see <a href=\"https://cloud.google.com/architecture/deploying-production-ready-log-exports-to-splunk-using-dataflow\">Deploying production-ready log exports to Splunk using Dataflow</a>.\n",
      "Before writing to Splunk, you can also apply a JavaScript user-defined function to the message payload. "
          + "Any messages that experience processing failures are forwarded to a Pub/Sub unprocessed topic for further troubleshooting and reprocessing.\n",
      "As an extra layer of protection for your HEC token, you can also pass in a Cloud KMS key along with the base64-encoded HEC token parameter encrypted with the Cloud KMS key. "
          + "See the <a href=\"https://cloud.google.com/kms/docs/reference/rest/v1/projects.locations.keyRings.cryptoKeys/encrypt\">Cloud KMS API encryption endpoint</a> for additional details on encrypting your HEC token parameter."
    },
    optionsClass = PubSubToSplunkOptions.class,
    optionsOrder = {
      PubsubReadSubscriptionOptions.class,
      SplunkOptions.class,
      JavascriptTextTransformerOptions.class,
      PubsubWriteDeadletterTopicOptions.class
    },
    documentation =
        "https://cloud.google.com/dataflow/docs/guides/templates/provided/pubsub-to-splunk",
    contactInformation = "https://cloud.google.com/support",
    requirements = {
      "The source Pub/Sub subscription must exist prior to running the pipeline.",
      "The Pub/Sub unprocessed topic must exist prior to running the pipeline.",
      "The Splunk HEC endpoint must be accessible from the Dataflow workers' network.",
      "The Splunk HEC token must be generated and available."
    },
    streaming = true,
    supportsAtLeastOnce = true,
    supportsExactlyOnce = true)
public class PubSubToSplunk {

  /** String/String Coder for FailsafeElement. */
  public static final FailsafeElementCoder<String, String> FAILSAFE_ELEMENT_CODER =
      FailsafeElementCoder.of(StringUtf8Coder.of(), StringUtf8Coder.of());

  /** Counter to track inbound messages from source. */
  private static final Counter INPUT_MESSAGES_COUNTER =
      Metrics.counter(PubSubToSplunk.class, "inbound-pubsub-messages");

  /** The tag for successful {@link SplunkEvent} conversion. */
  private static final TupleTag<SplunkEvent> SPLUNK_EVENT_OUT = new TupleTag<SplunkEvent>() {};

  /** The tag for failed {@link SplunkEvent} conversion. */
  private static final TupleTag<FailsafeElement<String, String>> SPLUNK_EVENT_DEADLETTER_OUT =
      new TupleTag<FailsafeElement<String, String>>() {};

  /** The tag for the main output for the UDF. */
  private static final TupleTag<FailsafeElement<String, String>> UDF_OUT =
      new TupleTag<FailsafeElement<String, String>>() {};

  /** The tag for the dead-letter output of the udf. */
  private static final TupleTag<FailsafeElement<String, String>> UDF_DEADLETTER_OUT =
      new TupleTag<FailsafeElement<String, String>>() {};

  /** GSON to process a {@link PubsubMessage}. */
  private static final Gson GSON = new Gson();

  /** Logger for class. */
  private static final Logger LOG = LoggerFactory.getLogger(PubSubToSplunk.class);

  private static final Boolean DEFAULT_INCLUDE_PUBSUB_MESSAGE = false;

  @VisibleForTesting protected static final String PUBSUB_MESSAGE_ATTRIBUTE_FIELD = "attributes";
  @VisibleForTesting protected static final String PUBSUB_MESSAGE_DATA_FIELD = "data";
  private static final String PUBSUB_MESSAGE_ID_FIELD = "messageId";

  /**
   * The main entry-point for pipeline execution. This method will start the pipeline but will not
   * wait for it's execution to finish. If blocking execution is required, use the {@link
   * PubSubToSplunk#run(PubSubToSplunkOptions)} method to start the pipeline and invoke {@code
   * result.waitUntilFinish()} on the {@link PipelineResult}.
   *
   * @param args The command-line args passed by the executor.
   */
  public static void main(String[] args) {

    PubSubToSplunkOptions options =
        PipelineOptionsFactory.fromArgs(args).withValidation().as(PubSubToSplunkOptions.class);

    run(options);
  }

  /**
   * Runs the pipeline to completion with the specified options. This method does not wait until the
   * pipeline is finished before returning. Invoke {@code result.waitUntilFinish()} on the result
   * object to block until the pipeline is finished running if blocking programmatic execution is
   * required.
   *
   * @param options The execution options.
   * @return The pipeline result.
   */
  public static PipelineResult run(PubSubToSplunkOptions options) {

    Pipeline pipeline = Pipeline.create(options);

    // Register coders.
    CoderRegistry registry = pipeline.getCoderRegistry();
    registry.registerCoderForClass(SplunkEvent.class, SplunkEventCoder.of());
    registry.registerCoderForType(
        FAILSAFE_ELEMENT_CODER.getEncodedTypeDescriptor(), FAILSAFE_ELEMENT_CODER);

    /*
     * Steps:
     *  1) Read messages in from Pub/Sub
     *  2) Convert message to FailsafeElement for processing.
     *  3) Apply user provided UDF (if any) on the input strings.
     *  4) Convert successfully transformed messages into SplunkEvent objects
     *  5) Write SplunkEvents to Splunk's HEC end point.
     *  5a) Wrap write failures into a FailsafeElement.
     *  6) Collect errors from UDF transform (#3), SplunkEvent transform (#4)
     *     and writing to Splunk HEC (#5) and stream into a Pub/Sub deadletter topic.
     */

    // 1) Read messages in from Pub/Sub
    PCollection<String> stringMessages =
        pipeline.apply(
            "ReadMessages",
            new ReadMessages(options.getInputSubscription(), options.getIncludePubsubMessage()));

    // 2) Convert message to FailsafeElement for processing.
    PCollectionTuple transformedOutput =
        stringMessages
            .apply(
                "ConvertToFailsafeElement",
                MapElements.into(FAILSAFE_ELEMENT_CODER.getEncodedTypeDescriptor())
                    .via(input -> FailsafeElement.of(input, input)))

            // 3) Apply user provided UDF (if any) on the input strings.
            .apply(
                "ApplyUDFTransformation",
                FailsafeJavascriptUdf.<String>newBuilder()
                    .setFileSystemPath(options.getJavascriptTextTransformGcsPath())
                    .setFunctionName(options.getJavascriptTextTransformFunctionName())
                    .setReloadIntervalMinutes(
                        options.getJavascriptTextTransformReloadIntervalMinutes())
                    .setLoggingEnabled(ValueProvider.StaticValueProvider.of(true))
                    .setSuccessTag(UDF_OUT)
                    .setFailureTag(UDF_DEADLETTER_OUT)
                    .build());

    // 4) Convert successfully transformed messages into SplunkEvent objects
    PCollectionTuple convertToEventTuple =
        transformedOutput
            .get(UDF_OUT)
            .apply(
                "ConvertToSplunkEvent",
                SplunkConverters.failsafeStringToSplunkEvent(
                    SPLUNK_EVENT_OUT, SPLUNK_EVENT_DEADLETTER_OUT));

    // 5) Write SplunkEvents to Splunk's HEC end point.
    PCollection<SplunkWriteError> writeErrors =
        convertToEventTuple
            .get(SPLUNK_EVENT_OUT)
            .apply(
                "WriteToSplunk",
                SplunkIO.writeBuilder()
                    .withToken(
                        new TokenNestedValueProvider(
                            options.getTokenSecretId(),
                            options.getTokenKMSEncryptionKey(),
                            options.getToken(),
                            options.getTokenSource()))
                    .withUrl(options.getUrl())
                    .withBatchCount(options.getBatchCount())
                    .withParallelism(options.getParallelism())
                    .withDisableCertificateValidation(options.getDisableCertificateValidation())
                    .withRootCaCertificatePath(options.getRootCaCertificatePath())
                    .withEnableBatchLogs(options.getEnableBatchLogs())
                    .withEnableGzipHttpCompression(options.getEnableGzipHttpCompression())
                    .build());

    final ValueProvider<Boolean> unwrapHecProvider = options.getUnwrapHecForDeadletter();

    // 5a) Wrap write failures into a FailsafeElement.
    PCollection<FailsafeElement<String, String>> wrappedSplunkWriteErrors =
        writeErrors.apply(
            "WrapSplunkWriteErrors",
            ParDo.of(
                new DoFn<SplunkWriteError, FailsafeElement<String, String>>() {

                  private boolean unwrapHecForDeadletter = false;

                  @Setup
                  public void setup() {
                    if (unwrapHecProvider != null) {
                      unwrapHecForDeadletter =
                          MoreObjects.firstNonNull(unwrapHecProvider.get(), false);
                    }
                    LOG.info("unwrapHecForDeadletter set to: {}", unwrapHecForDeadletter);
                  }

                  @ProcessElement
                  public void processElement(ProcessContext context) {
                    SplunkWriteError error = context.element();

                    // Extract original payload if unwrap is enabled
                    String payload = error.payload();
                    if (unwrapHecForDeadletter) {
                      payload = extractOriginalPayloadFromHec(payload);
                    }

                    FailsafeElement<String, String> failsafeElement =
                        FailsafeElement.of(payload, payload);

                    if (error.statusMessage() != null) {
                      failsafeElement.setErrorMessage(error.statusMessage());
                    }

                    if (error.statusCode() != null) {
                      failsafeElement.setErrorMessage(
                          String.format("Splunk write status code: %d", error.statusCode()));
                    }
                    context.output(failsafeElement);
                  }
                }));

    // 6) Collect errors from UDF transform (#4), SplunkEvent transform (#5)
    //     and writing to Splunk HEC (#6) and stream into a Pub/Sub deadletter topic.
    PCollectionList.of(
            ImmutableList.of(
                convertToEventTuple.get(SPLUNK_EVENT_DEADLETTER_OUT),
                wrappedSplunkWriteErrors,
                transformedOutput.get(UDF_DEADLETTER_OUT)))
        .apply("FlattenErrors", Flatten.pCollections())
        .apply(
            "WriteFailedRecords",
            ErrorConverters.WriteStringMessageErrorsToPubSub.newBuilder()
                .setErrorRecordsTopic(options.getOutputDeadletterTopic())
                .build());

    return pipeline.run();
  }

  /**
   * The {@link PubSubToSplunkOptions} class provides the custom options passed by the executor at
   * the command line.
   */
  public interface PubSubToSplunkOptions
      extends SplunkOptions,
          PubsubReadSubscriptionOptions,
          PubsubWriteDeadletterTopicOptions,
          JavascriptTextTransformerOptions {

    @TemplateParameter.Boolean(
        optional = true,
        description = "Unwrap Splunk HEC format from deadletter messages.",
        helpText =
            "When enabled, if a message fails to write to Splunk and is sent to the deadletter "
                + "queue, the original payload will be extracted from the Splunk HEC format before "
                + "writing to the deadletter topic. This prevents event nesting when messages are "
                + "replayed. Default: `false`.")
    ValueProvider<Boolean> getUnwrapHecForDeadletter();

    void setUnwrapHecForDeadletter(ValueProvider<Boolean> unwrapHecForDeadletter);
  }

  /**
   * A {@link PTransform} that reads messages from a Pub/Sub subscription, increments a counter and
   * returns a {@link PCollection} of {@link String} messages.
   */
  private static class ReadMessages extends PTransform<PBegin, PCollection<String>> {
    private final ValueProvider<String> subscriptionName;
    private final ValueProvider<Boolean> inputIncludePubsubMessageFlag;
    private Boolean includePubsubMessage;

    ReadMessages(
        ValueProvider<String> subscriptionName,
        ValueProvider<Boolean> inputIncludePubsubMessageFlag) {
      this.subscriptionName = subscriptionName;
      this.inputIncludePubsubMessageFlag = inputIncludePubsubMessageFlag;
    }

    @Override
    public PCollection<String> expand(PBegin input) {
      return input
          .apply(
              "ReadPubsubMessage",
              PubsubIO.readMessagesWithAttributes().fromSubscription(subscriptionName))
          .apply(
              "ExtractMessageIfRequired",
              ParDo.of(
                  new DoFn<PubsubMessage, String>() {

                    @Setup
                    public void setup() {
                      if (inputIncludePubsubMessageFlag != null) {
                        includePubsubMessage = inputIncludePubsubMessageFlag.get();
                      }
                      includePubsubMessage =
                          MoreObjects.firstNonNull(
                              includePubsubMessage, DEFAULT_INCLUDE_PUBSUB_MESSAGE);
                      LOG.info("includePubsubMessage set to: {}", includePubsubMessage);
                    }

                    @ProcessElement
                    public void processElement(ProcessContext context) {
                      if (includePubsubMessage) {
                        context.output(formatPubsubMessage(context.element()));
                      } else {
                        context.output(
                            new String(context.element().getPayload(), StandardCharsets.UTF_8));
                      }
                    }
                  }))
          .apply(
              "CountMessages",
              ParDo.of(
                  new DoFn<String, String>() {
                    @ProcessElement
                    public void processElement(ProcessContext context) {
                      INPUT_MESSAGES_COUNTER.inc();
                      context.output(context.element());
                    }
                  }));
    }
  }

  /**
   * Utility method that formats {@link org.apache.beam.sdk.io.gcp.pubsub.PubsubMessage} according
   * to the model defined in {@link com.google.pubsub.v1.PubsubMessage}.
   *
   * @param pubsubMessage {@link org.apache.beam.sdk.io.gcp.pubsub.PubsubMessage}
   * @return JSON String that adheres to the model defined in {@link
   *     com.google.pubsub.v1.PubsubMessage}
   */
  @VisibleForTesting
  protected static String formatPubsubMessage(PubsubMessage pubsubMessage) {
    JsonObject messageJson = new JsonObject();

    String payload = new String(pubsubMessage.getPayload(), StandardCharsets.UTF_8);
    try {
      JsonObject data = GSON.fromJson(payload, JsonObject.class);
      messageJson.add(PUBSUB_MESSAGE_DATA_FIELD, data);
    } catch (JsonSyntaxException e) {
      messageJson.addProperty(PUBSUB_MESSAGE_DATA_FIELD, payload);
    }

    JsonObject attributes = getAttributesJson(pubsubMessage.getAttributeMap());
    messageJson.add(PUBSUB_MESSAGE_ATTRIBUTE_FIELD, attributes);

    if (pubsubMessage.getMessageId() != null) {
      messageJson.addProperty(PUBSUB_MESSAGE_ID_FIELD, pubsubMessage.getMessageId());
    }

    return messageJson.toString();
  }

  /**
   * Constructs a {@link JsonObject} from a {@link Map} of Pub/Sub attributes.
   *
   * @param attributesMap {@link Map} of Pub/Sub attributes
   * @return {@link JsonObject} of Pub/Sub attributes
   */
  private static JsonObject getAttributesJson(Map<String, String> attributesMap) {
    JsonObject attributesJson = new JsonObject();
    for (String key : attributesMap.keySet()) {
      attributesJson.addProperty(key, attributesMap.get(key));
    }

    return attributesJson;
  }

  /**
   * Extracts the original payload from a Splunk HEC-formatted JSON string.
   *
   * <p>When a SplunkEvent fails to write to Splunk, it's stored in HEC format like: {"event":
   * "original payload", "time": ..., "host": ..., etc.}
   *
   * <p>This method extracts just the "event" field to prevent nesting when replaying messages from
   * the deadletter queue.
   *
   * @param hecPayload The HEC-formatted JSON string
   * @return The original event payload, or the input if extraction fails
   */
  @VisibleForTesting
  protected static String extractOriginalPayloadFromHec(String hecPayload) {
    try {
      JSONObject json = new JSONObject(hecPayload);

      // Check if this looks like HEC format (has "event" field)
      if (json.has("event")) {
        Object eventObj = json.get("event");

        // The event field could be a JSON object/array or a string
        if (eventObj instanceof String) {
          return (String) eventObj;
        } else {
          // If it's a JSONObject or JSONArray, return its string
          // representation
          return eventObj.toString();
        }
      }
    } catch (JSONException e) {
      // If parsing fails, log and return the original
      LOG.debug("Failed to parse payload as HEC JSON, returning as-is. Error: {}", e.getMessage());
    }

    // If no "event" field found or parsing failed, return original
    return hecPayload;
  }
}

המאמרים הבאים

• מידע נוסף על תבניות Dataflow
• רשימת התבניות ש-Google סיפקה