Ingest Google Cloud logs

The Google Cloud integration sends Google Cloud logs to Palo Alto Networks Cortex XCOR. A streaming Dataflow job that runs in your Google Cloud project reads log messages from a Pub/Sub subscription and sends them to the Cortex XCOR logs ingest API. A typical input is logs routed from Cloud Logging to Pub/Sub by a log sink, but the job accepts any Pub/Sub topic that carries JSON messages. The job runs entirely in your project, under identities you create. No step grants Palo Alto Networks access to your project. This job replaces Route logs from Google Cloud Platform, where Cortex XCOR Support pulls logs from a subscription you create. That setup is supported only for tenants already running it. Set up new log collection with the Dataflow job on this page.

Compatibility

Log collection works with:

The pubsub-to-chronosphere Dataflow Flex Template, version 1.0.1.
Pub/Sub subscriptions whose messages are JSON objects. Cloud Logging LogEntry messages get built-in normalization. Other JSON objects pass through as received. The template adds _ingest_source, project_id, and subscription_id to every log, replacing any fields with the same names. When the template can look up the subscription’s topic, it adds topic in the same way.
One or more subscriptions for each job. A subscription can attach to a topic in another Google Cloud project.

How logs are labeled

Every log gets two labels:

xcor.integration.name: always gcp, the integration’s catalog ID.
xcor_integration: the Google Cloud service that wrote the log, such as gcp_gke, gcp_cloud_run, gcp_cloud_sql, or gcp_compute_engine. Cortex XCOR derives this label from the log’s monitored resource type and log name. Audit logs that aren’t associated with a single service get gcp_audit_logs. Logs that Cortex XCOR can’t attribute get gcp_other and are stored like any other log.

For the following fields, values from the log take precedence over values that Cortex XCOR sets, and your organization’s log normalization settings take precedence over both:

Service: The service field is the xcor_integration value without gcp_, such as gke, cloud_run, or cloud_sql. Logs labeled gcp_other get their monitored resource type, such as global. Workload names, such as a container or Cloud Run service name, stay in the resource.labels fields.
Severity: Cloud Logging severities map to DEBUG, INFO (INFO and NOTICE), WARN, ERROR, and FATAL (CRITICAL, ALERT, and EMERGENCY). DEFAULT means the log has no severity. An audit log without a severity gets ERROR when protoPayload.status.code is non-zero, and INFO otherwise. Other logs without a severity get your organization’s default.

Prerequisites

Log collection requires:

A Cortex XCOR tenant with logging enabled. Contact Cortex XCOR Support if you’re unsure whether logging is enabled.
A Google Cloud project to run the job in, where you can enable APIs and create the resources in the setup steps.
The gcloud command-line tool, authenticated against that project.
A project Owner to grant roles to the job’s service account, or the other identities that Create the worker service account lists. The Editor role can’t grant roles. If the deploying identity isn’t an Editor, only the Owner can let it launch jobs as that service account.

Required permissions

The identity that deploys the job can be a person or an automation account. It needs either the roles/editor role on the project plus the following two roles, or the full list of roles after them. The Editor role can’t create log sinks or change a topic’s Identity and Access Management (IAM) policy.

Role	Scope	Purpose
`roles/logging.configWriter`	Project	Create the log sink.
`roles/pubsub.admin`	Project	Let the sink’s writer identity publish to the export topic. `roles/pubsub.editor` can’t grant this.

If the deploying identity isn’t an Editor, it needs the following roles instead:

Role	Scope	Purpose
`roles/serviceusage.serviceUsageAdmin`	Project	Enable the required APIs.
`roles/iam.serviceAccountCreator`	Project	Create the worker service account.
`roles/logging.configWriter`	Project	Create the log sink.
`roles/pubsub.admin`	Project	Create topics and subscriptions, and let the sink’s writer identity publish to the export topic. `roles/pubsub.editor` can’t grant this.
`roles/storage.admin`	Project	Create the staging bucket and upload staging files. If the bucket already exists, `roles/storage.objectAdmin` on the bucket is enough.
`roles/secretmanager.admin`	Project	Create the secret and add the token.
`roles/dataflow.developer`	Project	Create, drain, and cancel jobs.
`roles/compute.viewer`	Project	View machine types and quotas when launching jobs.
`roles/logging.viewer`	Project	Read the job’s worker logs, which the Troubleshooting rows quote.
`roles/iam.serviceAccountUser`	The worker service account	Launch jobs that run as the worker service account. `roles/iam.serviceAccountCreator` can’t grant this, so the project Owner grants it in Create the worker service account.

The worker service account, created during setup, needs the roles listed in Create the worker service account. No step requires organization-level roles.

Configure the integration

The Integrations Hub has no configuration form for logs. You deploy the job in Google Cloud, and it sends logs to your tenant. The setup creates the following pipeline in your project: Complete the following steps in order. Skip any step whose resource already exists. To create the same resources with Terraform instead, see Set up with Terraform.

Set up the project

Set these shell variables, which the later commands use:

PROJECT=<PROJECT_ID>   # Google Cloud project the job runs in
REGION=<REGION>        # for example, us-central1
TENANT=<TENANT>        # your tenant is https://<TENANT>.chronosphere.io

Enable the required APIs:

gcloud services enable dataflow.googleapis.com pubsub.googleapis.com \
  secretmanager.googleapis.com storage.googleapis.com compute.googleapis.com \
  --project=$PROJECT

Store an API token in Secret Manager

In Cortex XCOR, create a restricted service account with Write-only access and no label restrictions. The write-only restriction covers logs ingest too, though the option’s description in Cortex XCOR mentions only metric data. For the procedure, see Create a restricted service account.
Copy the account’s token, which Cortex XCOR shows only once.
Store the token in Secret Manager, replacing <TOKEN>:
printf '%s' "<TOKEN>" | gcloud secrets create cortex-xcor-api-token \ --data-file=- --replication-policy=automatic --project=$PROJECT
If your organization policy rejects automatic replication, use --replication-policy=user-managed --locations=$REGION instead.

Each worker reads the token from Secret Manager at startup, so the token never appears in the job’s parameters or pipeline graph.

Create the worker service account

The job’s virtual machines run as a dedicated service account. Create it:

SA=cortex-xcor-dataflow@$PROJECT.iam.gserviceaccount.com
gcloud iam service-accounts create cortex-xcor-dataflow --project=$PROJECT

A project Owner must grant the service account its roles. A project IAM administrator (roles/resourcemanager.projectIamAdmin) can run the loop instead, but not the secret binding, which needs the Owner or an identity with roles/secretmanager.admin:

for ROLE in roles/dataflow.worker roles/pubsub.subscriber roles/pubsub.viewer \
            roles/pubsub.publisher roles/storage.objectAdmin roles/artifactregistry.reader \
            roles/monitoring.metricWriter; do
  gcloud projects add-iam-policy-binding $PROJECT \
    --member="serviceAccount:$SA" --role="$ROLE" --condition=None
done
gcloud secrets add-iam-policy-binding cortex-xcor-api-token --project=$PROJECT \
  --member="serviceAccount:$SA" --role=roles/secretmanager.secretAccessor --condition=None

If the deploying identity isn’t an Editor, the Owner must also let it launch jobs as the service account. Replace <DEPLOYER> with the identity, such as user:alice@example.com:

gcloud iam service-accounts add-iam-policy-binding $SA --project=$PROJECT \
  --member="<DEPLOYER>" --role=roles/iam.serviceAccountUser --condition=None

The following table lists why the job needs each role:

Role	Purpose
`roles/dataflow.worker`	Run the launcher and worker virtual machines.
`roles/pubsub.subscriber`	Pull and acknowledge messages from the input subscriptions.
`roles/pubsub.viewer`	Look up each subscription’s topic at launch, which the job adds to every log as the `topic` field.
`roles/pubsub.publisher`	Publish failed messages to the dead-letter topic. Omit this role if you don’t use a dead-letter topic.
`roles/storage.objectAdmin`	Read and write staging and temporary files.
`roles/artifactregistry.reader`	Pull the template’s container image.
`roles/monitoring.metricWriter`	Write the job’s worker metrics to Cloud Monitoring. Without it, each worker start logs a denied `monitoring.metricDescriptors.create` call.
`roles/secretmanager.secretAccessor` on the token secret only	Read the API token at worker startup.

The --condition=None flag keeps the commands non-interactive in projects that have conditional IAM bindings. If you created the service account moments earlier, the first binding can fail with service account does not exist for up to a minute while IAM propagates the account. Run the command again.

Export logs to Pub/Sub

Create a topic, a log sink that routes Cloud Logging logs to it, and a subscription for the job:

gcloud pubsub topics create logs-export --project=$PROJECT

# The sink filter controls which logs leave your project and get ingested.
gcloud logging sinks create cortex-xcor-logs \
  pubsub.googleapis.com/projects/$PROJECT/topics/logs-export \
  --log-filter='(severity>=INFO OR severity=DEFAULT) AND NOT (resource.type="dataflow_step" AND resource.labels.job_name="cortex-xcor-logs")' \
  --project=$PROJECT

# Let the sink's writer identity publish to the topic.
WRITER=$(gcloud logging sinks describe cortex-xcor-logs --project=$PROJECT --format='value(writerIdentity)')
gcloud pubsub topics add-iam-policy-binding logs-export --project=$PROJECT \
  --member="$WRITER" --role=roles/pubsub.publisher

gcloud pubsub subscriptions create logs-export-cortex-xcor --topic=logs-export \
  --project=$PROJECT --ack-deadline=120 --message-retention-duration=23h

Everything the sink exports is sent and ingested, so filter with the sink’s --log-filter:

The filter keeps logs with no severity, which Cloud Logging stores as DEFAULT and severity>=INFO alone would drop.
It leaves out the job’s own worker logs, which would otherwise flow back into the job’s input and be ingested. The job_name in the filter must match the name you launch the job with.
Because it keeps DEFAULT, the filter also exports the Kubernetes API audit logs (k8s.io) from Google Kubernetes Engine (GKE) clusters, which carry no severity. These are Admin Activity audit logs, and lease updates can make up most of their volume. To leave out the lease updates, add AND NOT protoPayload.methodName="io.k8s.coordination.v1.leases.update" to the filter. To leave out all of them, add AND NOT protoPayload.serviceName="k8s.io", which also drops the record of changes such as creates, deletes, and permission changes.
If you enable Data Access audit logs for Pub/Sub, update the sink filter to exclude audit logs for the export topic.

Use one sink per project for this integration. Each sink exports its own copy of every log it matches, so a log that two sinks match is ingested twice. This includes folder-level and organization-level sinks that route the project’s logs. If another sink already exports these logs to a job like this one, don’t add a second. If the project runs more than one such job, give each job a different source: every subscription on a topic receives every message, so two jobs reading the same topic ingest it twice. Exclude each job’s worker logs in the one sink’s filter with a clause of its own, such as AND NOT (resource.type="dataflow_step" AND resource.labels.job_name="<OTHER_JOB>"). A new sink can take a few minutes to start exporting. Logs written before then aren’t exported. Keep subscription retention under 24 hours. Cortex XCOR discards logs with timestamps more than 24 hours in the past, so an older backlog can’t be delivered. To ship other JSON messages, such as app events published directly to Pub/Sub, point the job at a subscription on that topic instead. If a source topic lives in a different project, create the subscription in the job’s project and reference the topic as projects/<TOPIC_PROJECT>/topics/<TOPIC>. The identity that creates the subscription needs roles/pubsub.subscriber on the source topic. The worker service account needs no grants in the source project.

Create the staging bucket and dead-letter topic

gcloud storage buckets create gs://$PROJECT-cortex-xcor-dataflow \
  --project=$PROJECT --location=$REGION --uniform-bucket-level-access

gcloud pubsub topics create cortex-xcor-dead-letter --project=$PROJECT
gcloud pubsub subscriptions create cortex-xcor-dead-letter-monitor \
  --topic=cortex-xcor-dead-letter --project=$PROJECT

Create the dead-letter topic in every production deployment. Without it, the job drops messages that aren’t JSON objects and events that Cortex XCOR permanently rejects. The job counts dropped events, but can’t recover them.

Launch the job

The template spec for version 1.0.1 is gs://chronosphere-release/dataflow-templates/pubsub-to-chronosphere/1.0.1/spec.json. Its container image is us-docker.pkg.dev/chronosphereio/release/pubsub-to-chronosphere:1.0.1.

SPEC=gs://chronosphere-release/dataflow-templates/pubsub-to-chronosphere/1.0.1/spec.json

gcloud dataflow flex-template run cortex-xcor-logs \
  --project=$PROJECT --region=$REGION \
  --template-file-gcs-location=$SPEC \
  --service-account-email=$SA \
  --staging-location=gs://$PROJECT-cortex-xcor-dataflow/staging \
  --temp-location=gs://$PROJECT-cortex-xcor-dataflow/temp \
  --enable-streaming-engine --max-workers=2 \
  --parameters=inputSubscription=projects/$PROJECT/subscriptions/logs-export-cortex-xcor \
  --parameters=chronosphereUrl=https://$TENANT.chronosphere.io/api/v1/data/logs/ingest/integrations/gcp \
  --parameters=chronosphereApiTokenSecretId=projects/$PROJECT/secrets/cortex-xcor-api-token/versions/latest \
  --parameters=deadLetterTopic=projects/$PROJECT/topics/cortex-xcor-dead-letter

The command returns as soon as Dataflow accepts the job, even if the job then fails to start. Confirm that it started:

gcloud dataflow jobs list --project=$PROJECT --region=$REGION \
  --filter="name=cortex-xcor-logs" --limit=1 \
  --format="value(id,state)"

The command usually shows the newest job with that name. Its state should reach Running within a few minutes. If it shows Failed, open the job’s logs on the Dataflow Jobs page. A message such as Failed to start the launcher VM followed by Operation denied by org policy means an organization policy blocked the virtual machine. Add the settings in Locked-down environments and launch again. To choose --max-workers, see Sizing. For the other template parameters, see Parameters reference. To read several subscriptions with one job, separate them with commas. Because the comma collides with the gcloud parameter syntax, use the alternate-delimiter form and quote the whole flag:

--parameters="^;^inputSubscription=projects/$PROJECT/subscriptions/sub-a,projects/$PROJECT/subscriptions/sub-b"

Set up with Terraform

Terraform can create everything after the API token. Complete Set up the project and Store an API token in Secret Manager first, so the token stays in Secret Manager and out of Terraform state. The configuration creates its own log sink, so first read the one-sink-per-project guidance in Export logs to Pub/Sub. It also grants the worker service account its roles, so apply it as a project Owner, or as an identity that holds roles/resourcemanager.projectIamAdmin and roles/secretmanager.admin along with the roles in Required permissions. An identity that has neither the Owner nor the Editor role also needs roles/iam.serviceAccountUser on the worker service account before Terraform creates the job. The account doesn’t exist until Terraform creates it, and only the Owner can grant that role on it, so apply in two passes. Run terraform apply -target=google_service_account.worker to create only the account. Have the Owner run the roles/iam.serviceAccountUser command in Create the worker service account, with SA set to the account’s email, cortex-xcor-dataflow@<PROJECT_ID>.iam.gserviceaccount.com for the default name. Then run terraform apply again. Alternatively, the Owner can grant roles/iam.serviceAccountUser on the project before the apply, which lets the identity act as every service account in the project.

The configuration needs the google and google-beta providers because the Dataflow flex template job resource is available only in google-beta. The configuration was tested with version 8 of both providers. Set the project and region on each:

terraform {
  required_providers {
    google      = { source = "hashicorp/google", version = ">= 6.0" }
    google-beta = { source = "hashicorp/google-beta", version = ">= 6.0" }
  }
}

provider "google" {
  project = "<PROJECT_ID>"
  region  = "<REGION>"
}

provider "google-beta" {
  project = "<PROJECT_ID>"
  region  = "<REGION>"
}

Set values for these variables, and apply the configuration. In a project that blocks external IP addresses, also set subnetwork and private_ips, as in Locked-down environments.

Terraform example

# Variables

variable "project_id" {
  type        = string
  description = "The Google Cloud project the job runs in."
}

variable "region" {
  type        = string
  description = "The region the job runs in, for example us-central1."
}

variable "xcor_org_name" {
  type        = string
  description = "The name of your Cortex XCOR organization, which is the subdomain name before .chronosphere.io."
}

variable "api_token_secret_id" {
  type        = string
  description = "The ID of the Secret Manager secret in this project that holds the API token."
}

variable "name" {
  type        = string
  default     = "cortex-xcor"
  description = "Prefix for the resources this configuration creates. The job is named <name>-logs."
}

variable "max_workers" {
  type    = number
  default = 2
}

variable "subnetwork" {
  type        = string
  default     = null
  description = "Subnetwork for the job's virtual machines, as regions/<REGION>/subnetworks/<SUBNET> or a full URL for Shared VPC. Empty uses the default network."
}

variable "private_ips" {
  type        = bool
  default     = false
  description = "Run the virtual machines without external IP addresses. The subnetwork then needs Private Google Access and egress to <TENANT>.chronosphere.io, such as Cloud NAT."
}

locals {
  job_name = "${var.name}-logs"
  bucket   = "${var.project_id}-${var.name}-dataflow"
  spec     = "gs://chronosphere-release/dataflow-templates/pubsub-to-chronosphere/1.0.1/spec.json"
}

# Worker service account

resource "google_service_account" "worker" {
  account_id   = "${var.name}-dataflow"
  display_name = "Cortex XCOR logs Dataflow worker"
}

resource "google_project_iam_member" "worker" {
  for_each = toset([
    "roles/dataflow.worker",
    "roles/pubsub.subscriber",
    "roles/pubsub.viewer",
    "roles/pubsub.publisher",
    "roles/storage.objectAdmin",
    "roles/artifactregistry.reader",
    "roles/monitoring.metricWriter",
  ])
  project = var.project_id
  role    = each.value
  member  = "serviceAccount:${google_service_account.worker.email}"
}

resource "google_secret_manager_secret_iam_member" "worker" {
  secret_id = var.api_token_secret_id
  role      = "roles/secretmanager.secretAccessor"
  member    = "serviceAccount:${google_service_account.worker.email}"
}

# Logs export

resource "google_pubsub_topic" "export" {
  name = "${var.name}-logs-export"
}

resource "google_logging_project_sink" "export" {
  name        = local.job_name
  destination = "pubsub.googleapis.com/${google_pubsub_topic.export.id}"
  # Keeps logs with no severity, and leaves out the job's own worker logs. To leave out GKE's
  # lease updates, append AND NOT protoPayload.methodName=\"io.k8s.coordination.v1.leases.update\"
  # to the string; to leave out all of GKE's Kubernetes API audit logs, append
  # AND NOT protoPayload.serviceName=\"k8s.io\".
  filter                 = "(severity>=INFO OR severity=DEFAULT) AND NOT (resource.type=\"dataflow_step\" AND resource.labels.job_name=\"${local.job_name}\")"
  unique_writer_identity = true
}

resource "google_pubsub_topic_iam_member" "sink_writer" {
  topic  = google_pubsub_topic.export.id
  role   = "roles/pubsub.publisher"
  member = google_logging_project_sink.export.writer_identity
}

resource "google_pubsub_subscription" "export" {
  name                       = "${var.name}-logs-export"
  topic                      = google_pubsub_topic.export.id
  ack_deadline_seconds       = 120
  message_retention_duration = "82800s"
}

# Staging bucket and dead-letter topic

resource "google_storage_bucket" "staging" {
  name                        = local.bucket
  location                    = var.region
  uniform_bucket_level_access = true
  force_destroy               = true
}

resource "google_pubsub_topic" "dead_letter" {
  name = "${var.name}-dead-letter"
}

resource "google_pubsub_subscription" "dead_letter" {
  name  = "${var.name}-dead-letter-monitor"
  topic = google_pubsub_topic.dead_letter.id
}

# Dataflow job

# The flex template job resource is only in the google-beta provider.
resource "google_dataflow_flex_template_job" "logs" {
  provider                = google-beta
  name                    = local.job_name
  region                  = var.region
  container_spec_gcs_path = local.spec
  service_account_email   = google_service_account.worker.email
  staging_location        = "gs://${local.bucket}/staging"
  temp_location           = "gs://${local.bucket}/temp"
  enable_streaming_engine = true
  max_workers             = var.max_workers
  subnetwork              = var.subnetwork
  ip_configuration        = var.private_ips ? "WORKER_IP_PRIVATE" : null
  on_delete               = "drain"
  parameters = {
    inputSubscription            = google_pubsub_subscription.export.id
    chronosphereUrl              = "https://${var.xcor_org_name}.chronosphere.io/api/v1/data/logs/ingest/integrations/gcp"
    chronosphereApiTokenSecretId = "projects/${var.project_id}/secrets/${var.api_token_secret_id}/versions/latest"
    deadLetterTopic              = google_pubsub_topic.dead_letter.id
  }
  depends_on = [google_project_iam_member.worker, google_secret_manager_secret_iam_member.worker, google_storage_bucket.staging]
}

The sink filter names the job, so changing name updates both. Destroying the configuration drains the job before deleting it.

Locked-down environments

Common organization policies need these additional launch settings:

No external IP addresses, enforced by constraints/compute.vmExternalIpAccess or constraints/compute.managed.vmExternalIpAccess: Add --disable-public-ips and run on a subnet with Private Google Access enabled. The subnet also needs the following:
- If your Virtual Private Cloud (VPC) network uses custom private DNS zones for Google APIs, such as private.googleapis.com or restricted.googleapis.com setups, the zones must cover pkg.dev for the template image and gcr.io for a logging agent that the Dataflow launcher pulls. For the options, see Private Google Access domain options. Older setups that predate Artifact Registry often cover gcr.io but miss pkg.dev.
- The job sends data to <TENANT>.chronosphere.io on port 443. Private Google Access doesn’t cover this external traffic. Provide Cloud Network Address Translation (Cloud NAT) or equivalent egress to that host. The job runs continuously, so the egress path must stay up.
No default VPC network: Add --network, --subnetwork, or both. For Shared VPC, use the full subnetwork URL, https://www.googleapis.com/compute/v1/projects/<HOST_PROJECT>/regions/<REGION>/subnetworks/<SUBNET>, and grant the deploying identity roles/compute.networkUser on the subnet.
Firewall rules keyed on network tags: Dataflow virtual machines carry only the default dataflow network tag. Add tags at launch with --additional-experiments=use_network_tags=<TAG1>;<TAG2> for worker virtual machines and --additional-experiments=use_network_tags_for_flex_templates=<TAG> for the launcher virtual machine. You can’t add tags after the job starts.
Mandatory virtual machine labels: Add --additional-user-labels=<KEY1>=<VALUE1>,<KEY2>=<VALUE2>.
VPC Service Controls: Confirm that the perimeter’s egress policy allows the job to reach the ingest endpoint, and that cross-project subscriptions stay inside the perimeter.

Validate log ingestion

Allow several minutes for Dataflow to start the job’s virtual machines. After the job is running, a message usually reaches Cortex XCOR within seconds of arriving in Pub/Sub. The job sends a partial batch after the batch timeout, 5 seconds by default.
In the Google Cloud console, open the job on the Dataflow Jobs page and review the custom counters for its steps. The valid_messages, batches_sent, and events_sent counters should increase, and invalid_messages and batches_failed should stay at 0. The events_sent counter includes logs that Cortex XCOR accepted and then discarded, so confirm arrival in the next step.
In the navigation menu, select Explorers > Logs Explorer, and run this query:
["xcor.integration.name"] = "gcp" | summarize count() by xcor_integration
Each Google Cloud service sending logs appears as a row. Each log also carries project_id and subscription_id fields that identify the subscription it came through, and a topic field when the job could look up the subscription’s topic. project_id is the subscription’s project, not the project that wrote the log. For more about queries, see Search and filter log data.

To send a test message that exercises normalization, publish a LogEntry-shaped message to the input topic:

gcloud pubsub topics publish logs-export --project=$PROJECT \
  --message='{"severity":"ERROR","textPayload":"Cortex XCOR test message","timestamp":"'"$(date -u +%Y-%m-%dT%H:%M:%SZ)"'"}'

The log appears with message set to the textPayload value.

Sizing

With Streaming Engine enabled, as in the launch command, horizontal autoscaling is on by default, and you can change the minimum and maximum workers with an in-flight update. Always set --max-workers, because it caps the job’s cost. To estimate it:

Average events per second = logs per day ÷ 86,400.
Peak events per second ≈ 2 × average. Adjust for your burst profile.
Virtual CPUs needed = peak events per second ÷ 2,000, rounded up. The estimate is conservative for logs of about 1 KB.
--max-workers = virtual CPUs needed ÷ 4, rounded up, for the default n1-standard-4 workers.

For reference, one n1-standard-4 worker sustained about 150 million logs per day at under half its CPU. The job batches by event count, not size. A batch holds up to batchSize events, with a default of 500. A Cloud Logging entry can be up to 256 KiB, so a batch of large entries can pass the 16 MiB limit on one ingest request. The job then splits the batch in half and retries, so nothing is lost. Each oversized batch costs extra requests. If your entries are large, lower batchSize, for example to 100.

Monitoring

You run the job, so you own detecting its failures. If you collect Google Cloud metrics too, add the pubsub.googleapis.com/subscription and dataflow.googleapis.com/job metric type prefixes so Cortex XCOR ingests pipeline health metrics into the same tenant as the logs. Alert on these conditions:

The subscription’s oldest_unacked_message_age approaches the retention window. Messages older than the retention window are lost.
The subscription’s num_undelivered_messages grows steadily.
The Dataflow job’s system_lag rises, or the job leaves the running state.
The dead-letter subscription holds any messages.

The job also reports these custom counters for each step in the Dataflow console: valid_messages, invalid_messages, empty_messages, batches_sent, batches_failed, batches_bisected, events_sent, events_dead_lettered, retries_performed, udf_transformed, udf_filtered, udf_errors, and udf_invalid_output. It also reports the batch_size and request_duration_ms distributions.

Parameters reference

The following table lists the template parameters:

Parameter	Required	Default	Description
`inputSubscription`	Yes	None	Subscriptions to read, in the form `projects/<PROJECT_ID>/subscriptions/<SUBSCRIPTION>`. Separate several with commas.
`chronosphereUrl`	Yes	None	The logs ingest endpoint, `https://<TENANT>.chronosphere.io/api/v1/data/logs/ingest/integrations/gcp`.
`chronosphereApiTokenSecretId`	One of the two token parameters	None	Secret Manager resource ID of the API token, in the form `projects/<PROJECT_ID>/secrets/<SECRET>/versions/<VERSION>`. Takes precedence over `chronosphereApiToken`.
`chronosphereApiToken`	One of the two token parameters	None	Plaintext API token. Don’t use it in production, because the token is visible in the job parameters in the Dataflow console.
`enableDefaultNormalization`	No	`true`	Built-in `LogEntry` normalization, described after this table.
`defaultLabels`	No	None	Comma-separated `key=value` fields added to every log, such as `env=prod,team=payments`. A label overwrites a same-named field in the log, so avoid keys your logs already use.
`batchSize`	No	`500`	Events per HTTP request.
`batchTimeoutSeconds`	No	`5`	Maximum seconds to buffer a partial batch before sending it.
`deadLetterTopic`	No	None	Topic for messages that fail parsing, invalid function output, and permanently rejected events, in the form `projects/<PROJECT_ID>/topics/<TOPIC>`. If unset, the job drops them.
`javascriptTextTransformGcsPath`	No	None	Cloud Storage path of a JavaScript user-defined function file, such as `gs://<BUCKET>/transform.js`.
`javascriptTextTransformFunctionName`	No	`transform`	Function to call in the user-defined function file.
`enableGzipCompression`	No	`true`	Compress request bodies with gzip.
`maxRetries`	No	`5`	Retry attempts for each request after a transient HTTP or network error.

Normalization

Normalization applies only to messages that contain a textPayload, jsonPayload, or protoPayload field. It only adds fields and never overwrites them:

message: Set from textPayload, from the message, msg, or log field of jsonPayload, or from the protoPayload method name and resource name for audit logs. When none of those fields exist, the payload itself becomes the message as a JSON string.
severity: Set to DEFAULT when missing.
timestamp: Set from receiveTimestamp when missing.
kubernetes: For k8s_container resources, set to an object with the cluster_name, namespace_name, and pod_name resource labels.

The job keeps the full nested payload. Every log also gets _ingest_source, project_id, subscription_id, and, when the job could look it up, topic. The job sets these fields after defaultLabels, so a default label with one of these keys has no effect, except topic when the lookup failed.

User-defined functions

A JavaScript user-defined function runs after normalization. It receives each log as a JSON string and returns either a string containing a JSON object or null to drop the log. Write it in ES5 syntax. The job loads the file at worker startup, so update the file and then update or relaunch the job to apply changes. Don’t use a user-defined function to redact data. If the function throws an error, the job sends the original log unchanged. Output that isn’t a JSON object goes to the dead-letter topic.

Delivery semantics and the dead-letter topic

Delivery is at least once. After an outage or a restart, Pub/Sub can redeliver some messages that were already sent. The job handles each response from the logs ingest API as follows:

200: Delivered.
429, any 5xx status, or a network error: The job retries with exponential backoff up to maxRetries times and honors a Retry-After header whose value is specified in seconds. When retries run out, Dataflow retries the work while Pub/Sub holds the backlog, so an endpoint outage doesn’t lose data or fill the dead-letter topic.
401 or 403: The job stops delivering and Dataflow retries the work, so an authentication problem stalls the pipeline instead of dead-lettering all traffic.
Any other status, such as 400 or 413: The API rejects a request as a whole, so the job splits the batch in half and sends each half again until it isolates the rejected events. Only those events go to the dead-letter topic.

The dead-letter topic also receives messages that aren’t JSON objects and user-defined function output that isn’t a JSON object. Empty messages are counted in empty_messages and dropped. Cortex XCOR can accept a request with a 200 status and still discard individual logs in it. It discards logs with timestamps more than 24 hours in the past or future, logs larger than 1 MB, and logs over your tenant’s ingest limits. The job counts these logs in events_sent, and they never reach the dead-letter topic. To replay dead-lettered events after you fix the cause, pull them from the monitor subscription and publish them to the input topic within 24 hours of their original timestamps. Older logs are discarded.

Upgrading

A running job doesn’t update itself. To move to a new template version, run gcloud dataflow flex-template run with the --update flag and the new spec. This updates the streaming job in place when the versions are compatible. Otherwise, drain the old job and launch a new one. If the new job has a different name, change the job_name in the sink’s filter to match. Otherwise, the new job’s worker logs are ingested. Each request carries a User-Agent: pubsub-to-chronosphere/<VERSION> header that identifies the deployed version. To upgrade a running job, run your original launch command again with --update, SPEC set to the new version, and every parameter and option you launched with. If the new version runs a later Beam SDK than your job, also pass --additional-pipeline-options=updateCompatibilityVersion=<current_job_beam_version>, using the Beam SDK version from the following table for the template version your job runs. With Terraform, set additional_pipeline_options = ["updateCompatibilityVersion=<current_job_beam_version>"] on the google_dataflow_flex_template_job resource, not in parameters. Keep the same job name. Dataflow checks compatibility before it replaces the job. If the check fails, the running job keeps running, and you can drain it and launch the new version.

Template version	Beam SDK
`0.0.1`	`2.74.0`
`1.0.0`	`2.74.0`
`1.0.1`	`2.77.0`

Troubleshooting

The following table lists common problems and how to resolve them:

Symptom	Resolution
Worker logs show `Failed to read API token from Secret Manager`	The worker service account lacks `roles/secretmanager.secretAccessor` on the secret, or the secret ID is wrong. Grant the role or fix `chronosphereApiTokenSecretId`.
Worker logs show `Chronosphere returned 401` or `Chronosphere returned 403`, and the backlog grows	The token is wrong or revoked, or its service account is read-only or has label restrictions. Use a token from a write-only service account without label restrictions: add it as a new secret version, then update or relaunch the job.
Every event goes to the dead-letter topic, and worker logs show `non-retryable status` with a status such as `404` or `405`, or a redirect	`chronosphereUrl` has the wrong scheme, host, or path. Use `https://<TENANT>.chronosphere.io/api/v1/data/logs/ingest/integrations/gcp`.
The launch command succeeds, but the job fails within seconds with `Failed to start the launcher VM`, or an organization policy otherwise blocks virtual machine creation	An organization policy blocks the virtual machine, such as `constraints/compute.managed.vmExternalIpAccess` for external IP addresses, or a policy that requires labels or network tags. See Locked-down environments.
Launch fails on `iam.serviceAccounts.actAs`	The deploying identity lacks `roles/iam.serviceAccountUser` on the worker service account.
Launch fails on `compute.subnetworks.use`	The deploying identity or the Dataflow service agent lacks `roles/compute.networkUser` on the Shared VPC subnet.
Launch fails with `Timeout in polling result file`	The launcher virtual machine can’t reach an image registry. On a subnet without external IP addresses, verify Private Google Access and that private DNS zones cover both `pkg.dev` and `gcr.io`.
Workers run but no requests leave the network	No route to the ingest endpoint. Provide Cloud NAT or another egress path to `<TENANT>.chronosphere.io` on port 443.
`PERMISSION_DENIED` when creating a cross-project subscription	The identity creating the subscription lacks `roles/pubsub.subscriber` on the source topic.
The job runs but reads no messages	The worker service account lacks `roles/pubsub.subscriber`, or the sink filter matches no logs.
`invalid_messages` increases and the dead-letter topic fills	Input messages aren’t JSON objects. Inspect the dead-letter subscription.
`events_sent` increases but some logs never appear, and the dead-letter topic is empty	Cortex XCOR discarded the logs without an error, because their timestamps are more than 24 hours from the current time or they’re larger than 1 MB. Keep subscription retention under 24 hours, clear backlogs promptly, and verify the timestamps your sources send.
Logs lack the `topic` field	The worker service account lacks `roles/pubsub.viewer`, so the job couldn’t look up the subscription’s topic at launch. Grant the role and relaunch.

Limitations

This page covers logs only. For metrics, see Google Cloud metrics.
You deploy, run, scale, and upgrade the job in your own Google Cloud project, and Google Cloud bills you for it.
Cortex XCOR discards logs with timestamps more than 24 hours in the past or future, and logs larger than 1 MB, without returning an error. A backlog older than 24 hours can’t be delivered.
Delivery is at least once, so some logs can arrive twice after an outage or restart.
A user-defined function that throws an error passes the original log through, so it can’t enforce redaction.

Cortex XCOR

Palo Alto Networks Telemetry Pipeline

More information

Compatibility

How logs are labeled

Prerequisites

Required permissions

Configure the integration

Set up the project

Store an API token in Secret Manager

Create the worker service account

Export logs to Pub/Sub

Create the staging bucket and dead-letter topic

Launch the job

Set up with Terraform

Locked-down environments

Validate log ingestion

Sizing

Monitoring

Parameters reference

Normalization

User-defined functions

Delivery semantics and the dead-letter topic

Upgrading

Troubleshooting

Limitations

​Compatibility

​How logs are labeled

​Prerequisites

​Required permissions

​Configure the integration

​Set up the project

​Store an API token in Secret Manager

​Create the worker service account

​Export logs to Pub/Sub

​Create the staging bucket and dead-letter topic

​Launch the job

​Set up with Terraform

​Locked-down environments

​Validate log ingestion

​Sizing

​Monitoring

​Parameters reference

​Normalization

​User-defined functions

​Delivery semantics and the dead-letter topic

​Upgrading

​Troubleshooting

​Limitations

Compatibility

How logs are labeled

Prerequisites

Required permissions

Configure the integration

Set up the project

Store an API token in Secret Manager

Create the worker service account

Export logs to Pub/Sub

Create the staging bucket and dead-letter topic

Launch the job

Set up with Terraform

Locked-down environments

Validate log ingestion

Sizing

Monitoring

Parameters reference

Normalization

User-defined functions

Delivery semantics and the dead-letter topic

Upgrading

Troubleshooting

Limitations