> ## Documentation Index
> Fetch the complete documentation index at: https://docs-xcor.paloaltonetworks.com/llms.txt
> Use this file to discover all available pages before exploring further.

# Ingest Google Cloud logs

> Learn how to send Google Cloud logs to Palo Alto Networks Cortex XCOR with a Dataflow job.

The Google Cloud integration sends [Google Cloud](https://cloud.google.com/) 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](https://docs-xcor.paloaltonetworks.com/ingest/logs/gcp-logs.md),
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](#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](#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](#create-the-worker-service-account). |

The worker service account, created during setup, needs the roles listed in
[Create the worker service account](#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:

```mermaid actions={false} theme={null}
graph LR
  accTitle: Google Cloud log delivery
  accDescr: Cloud Logging routes logs through Pub/Sub to a Dataflow job. The job parses, normalizes, batches, compresses, and sends logs to Cortex XCOR. Failed events go to a dead-letter topic.
  logging[Cloud Logging] -- sink --> topic[Pub/Sub topic]
  topic --> subscriptions[Pub/Sub subscriptions]
  subscriptions --> job[Dataflow job]
  job -- accepted logs --> xcor[Cortex XCOR]
  job -- failed events --> deadLetter[Dead-letter topic]
```

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-with-terraform).

### Set up the project

Set these shell variables, which the later commands use:

```shell theme={null}
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:

```shell theme={null}
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

1. 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](https://docs-xcor.paloaltonetworks.com/administer/accounts-teams/service-accounts.md#create-a-restricted-service-account).
2. Copy the account's token, which Cortex XCOR shows only once.
3. Store the token in Secret Manager, replacing `<TOKEN>`:

   ```shell theme={null}
   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:

```shell theme={null}
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`:

```shell theme={null}
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`:

```shell theme={null}
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:

```shell theme={null}
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

```shell theme={null}
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`.

```shell theme={null}
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:

```shell theme={null}
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](#locked-down-environments) and launch again.

To choose `--max-workers`, see [Sizing](#sizing). For the other template parameters, see
[Parameters reference](#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:

```shell theme={null}
--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](#set-up-the-project) and
[Store an API token in Secret Manager](#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](#export-logs-to-pubsub). 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](#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](#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.

1. 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 theme={null}
   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>"
   }
   ```

2. 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](#locked-down-environments).

   ```terraform expandable Terraform example icon="square-terminal" theme={null}
   # 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](https://cloud.google.com/vpc/docs/configure-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

1. 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.
2. 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.
3. In the navigation menu, select
   **Explorers > Logs Explorer**, and run this query:

   ```text theme={null}
   ["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](https://docs-xcor.paloaltonetworks.com/investigate/querying/query-logs).

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

```shell theme={null}
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:

1. Average events per second = logs per day ÷ 86,400.
2. Peak events per second ≈ 2 × average. Adjust for your burst profile.
3. Virtual CPUs needed = peak events per second ÷ 2,000, rounded up. The estimate is
   conservative for logs of about 1 KB.
4. `--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](https://docs-xcor.paloaltonetworks.com/ingest/xcor/integrations/server/gcp/metrics.md)
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](#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](https://docs-xcor.paloaltonetworks.com/ingest/xcor/integrations/server/gcp/metrics.md).
* 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.


## Related topics

- [Route logs from Google Cloud Platform](https://docs-xcor.paloaltonetworks.com/ingest/logs/gcp-logs.md)
- [Ingest Google Cloud metrics](https://docs-xcor.paloaltonetworks.com/ingest/xcor/integrations/server/gcp/metrics.md)
- [Google Cloud](https://docs-xcor.paloaltonetworks.com/ingest/xcor/integrations/server/gcp.md)
- [Ingest log data](https://docs-xcor.paloaltonetworks.com/ingest/logs.md)
- [Google Cloud Operations Suite destination plugin](https://docs-xcor.paloaltonetworks.com/ingest/pipeline/plugins/destination-plugins/gcp-ops.md)


This documentation is built and hosted on [Mintlify](https://mintlify.com), a developer documentation platform.