> ## 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.

# Build custom pipelines for the CXDOT Collector

> Build custom pipelines to process and route telemetry data from the CXDOT Collector.

> Requires CXDOT Collector version 1.5.0 or later.

You can build custom pipelines to modify how the CXDOT Collector processes and routes
telemetry data. Custom pipelines affect telemetry data after an
[integration](/ingest/xcor/collector/configure/integrations) collects that data,
but before the CXDOT Collector sends it to Palo Alto Networks Cortex XCOR. For
example, you can create a custom pipeline that removes sensitive information from
logs before the Collector sends those logs to Cortex XCOR, or a custom pipeline that
sends one copy of metrics to Cortex XCOR and another copy to a different destination.

Each custom pipeline applies to either metrics, logs, or traces, and by default
affects data of that telemetry type from all enabled integrations. For example, a
pipeline for traces affects trace data from every integration that collects traces,
unless the pipeline is configured to
[only affect certain integrations](#configure-per-integration-behavior).

```mermaid actions={false} theme={null}
flowchart TD
  accTitle: Custom pipeline telemetry flow
  accDescr: The CXDOT Collector collects telemetry data from multiple integrations. That data passes through a custom pipeline, then gets sent to Cortex XCOR and other destinations.

  source(["Telemetry sources"])

  subgraph collector["CXDOT Collector"]
    direction LR

    subgraph integrations[" "]
      direction TB
      integration1["Integration"]
      integration2["Integration"]
      integration3["Integration"]
    end

    pipeline{"Custom pipeline"}

    integration1 -.->|"Telemetry data"| pipeline
    integration2 -.->|"Telemetry data"| pipeline
    integration3 -.->|"Telemetry data"| pipeline
  end

  xcor(["Palo Alto Networks Cortex XCOR"])
  destinations(["Other destinations"])

  source -.-> integration1
  source -.-> integration2
  source -.-> integration3
  pipeline -.->|"Processed telemetry data"| xcor
  pipeline -.->|"Processed telemetry data"| destinations

  style integrations fill:transparent,stroke:transparent

  classDef green fill:#60E684,stroke:#000,stroke-width:2px,color:#000;
  class pipeline green
```

## Components

Custom pipelines use a subset of OpenTelemetry components. Each custom pipeline
consists of the following components:

* At least one [receiver](https://opentelemetry.io/docs/collector/configuration/#receivers)
* At least one [exporter](https://opentelemetry.io/docs/collector/configuration/#exporters)
* (Optional) One or more [processors](https://opentelemetry.io/docs/collector/configuration/#processors).

The CXDOT Collector provides two components that can be used as receivers
and exporters. Use the `forward/cxdot/from-integrations` component as a receiver to
receive telemetry data from CXDOT Collector integrations. Use the
`forward/cxdot/to-egress` component as an exporter to send telemetry data to Cortex
XCOR.

> [!WARNING]
>
> When a custom pipeline receives telemetry data from
> `forward/cxdot/from-integrations`, that custom pipeline replaces the default path for
> that telemetry type. To continue sending all telemetry data of that type to Cortex
> XCOR, every processing or routing path must eventually export to
> `forward/cxdot/to-egress`.

### Connect multiple pipelines

If you're using multiple custom pipelines for the same telemetry type, use
`forward/cxdot/from-integrations` as a receiver in the first pipeline of your
sequence, then use
[OpenTelemetry connectors](https://opentelemetry.io/docs/collector/configuration/#connectors)
to allow telemetry data to flow from one pipeline to the next. Use
`forward/cxdot/to-egress` as an exporter in any pipelines that should send telemetry
data to Cortex XCOR.

For example, the following snippet uses the `forward/next` connector to send metrics
from one custom pipeline to another. The `forward/cxdot/from-integrations` receiver
in the first pipeline receives telemetry data, and the `forward/cxdot/to-egress`
exporter in the second pipeline sends that data to Cortex XCOR.

```yaml theme={null}
connectors:
  forward/next: {}

service:
  pipelines:
    metrics/first:
      receivers: [forward/cxdot/from-integrations]
      exporters: [forward/next]
    metrics/second:
      receivers: [forward/next]
      exporters: [forward/cxdot/to-egress]
```

If multiple pipelines are joined by a routing connector, those pipelines must all
handle the same telemetry type as that routing connector. For example, a routing
connector that handles metrics must join only pipelines for metrics.

### List compatible OpenTelemetry components

To list the OpenTelemetry components that are compatible with your version of the
CXDOT Collector, use the following command:

**Kubernetes**

```shell theme={null}
kubectl exec --namespace cxdot POD -- \
  /bin/cxdot-collector components
```

Replace *`POD`* with the name of any running CXDOT Collector Pod.

**Linux**

```shell theme={null}
cxdot-collector components
```

**Windows**

```powershell theme={null}
.\cxdot-collector.exe components
```

> [!NOTE]
>
> The `forward` component used by `forward/cxdot/from-integrations` and
> `forward/cxdot/to-egress` appears in the command's `connectors` list.

## Create a custom pipeline

To create a custom pipeline, add a `service.pipelines` block to your configuration
file that defines receivers and exporters. Optionally, add a `processors` block that
defines processors.

> [!NOTE]
>
> Before you create a custom pipeline for logs or traces, make sure to
> [enable telemetry egress](/ingest/xcor/collector/configure/telemetry-egress) for
> that telemetry type.

### Process telemetry data

The following example creates a custom pipeline named `metrics/add-env` that adds the
`env = prod` resource attribute to metrics. Logs and traces aren't affected.

**Helm values file**

Add `processors` and `service.pipelines` blocks to both the
`nodeCollector.configOverlay` and `clusterCollector.configOverlay` sections
of your Helm values file.

```yaml expandable title="values.yaml" theme={null}
nodeCollector:
  configOverlay:
    processors:
      resource/add-env:
        attributes:
          - key: env
            value: prod
            action: insert
    service:
      pipelines:
        metrics/add-env:
          receivers: [forward/cxdot/from-integrations]
          processors: [resource/add-env]
          exporters: [forward/cxdot/to-egress]

clusterCollector:
  configOverlay:
    processors:
      resource/add-env:
        attributes:
          - key: env
            value: prod
            action: insert
    service:
      pipelines:
        metrics/add-env:
          receivers: [forward/cxdot/from-integrations]
          processors: [resource/add-env]
          exporters: [forward/cxdot/to-egress]
```

**Configuration file**

Add `processors` and `service.pipelines` as top-level blocks in your configuration
file.

```yaml expandable title="config.yaml" theme={null}
cxdot:
  # ...

processors:
  resource/add-env:
    attributes:
      - key: env
        value: prod
        action: insert

service:
  pipelines:
    metrics/add-env:
      receivers: [forward/cxdot/from-integrations]
      processors: [resource/add-env]
      exporters: [forward/cxdot/to-egress]
```

### Send telemetry data to an additional destination

The following example creates a custom pipeline named `logs/copy` that sends a copy
of log data to an OTLP HTTP endpoint while continuing to send logs to Cortex XCOR.
Metrics and traces aren't affected.

**Helm values file**

Add `exporters` and `service.pipelines` blocks to both the
`nodeCollector.configOverlay` and `clusterCollector.configOverlay` sections
of your Helm values file.

```yaml expandable title="values.yaml" theme={null}
nodeCollector:
  configOverlay:
    exporters:
      otlphttp/additional:
        endpoint: https://destination.example.com:4318
    service:
      pipelines:
        logs/copy:
          receivers: [forward/cxdot/from-integrations]
          exporters: [forward/cxdot/to-egress, otlphttp/additional]

clusterCollector:
  configOverlay:
    exporters:
      otlphttp/additional:
        endpoint: https://destination.example.com:4318
    service:
      pipelines:
        logs/copy:
          receivers: [forward/cxdot/from-integrations]
          exporters: [forward/cxdot/to-egress, otlphttp/additional]
```

**Configuration file**

Add `exporters` and `service.pipelines` as top-level blocks in your configuration
file.

```yaml expandable title="config.yaml" theme={null}
cxdot:
  # ...

exporters:
  otlphttp/additional:
    endpoint: https://destination.example.com:4318

service:
  pipelines:
    logs/copy:
      receivers: [forward/cxdot/from-integrations]
      exporters: [forward/cxdot/to-egress, otlphttp/additional]
```

#### Send batched telemetry data to an additional destination

Telemetry data exported through `forward/cxdot/to-egress` is batched before being
sent to Cortex XCOR, but other exporters don't batch data by default. To batch
telemetry data sent through another exporter, add a `batch` processor to the pipeline
for that destination.

The following example creates two custom pipelines. The first custom pipeline, named
`logs/copy`, sends logs to Cortex XCOR using the `forward/cxdot/to-egress` exporter,
and also routes a copy of that log data to the second pipeline. The second custom
pipeline, named `logs/additional`, uses a batch processor to batch these logs before
sending them to an OTLP HTTP endpoint. Metrics and traces aren't affected.

**Helm values file**

Add `connectors`, `exporters`, `processors`, and `service.pipelines` blocks to
both the `nodeCollector.configOverlay` and `clusterCollector.configOverlay`
sections of your Helm values file.

```yaml expandable title="values.yaml" theme={null}
nodeCollector:
  configOverlay:
    connectors:
      forward/additional: {}
    exporters:
      otlphttp/additional:
        endpoint: https://destination.example.com:4318
    processors:
      batch/additional: {}
    service:
      pipelines:
        logs/copy:
          receivers: [forward/cxdot/from-integrations]
          exporters: [forward/cxdot/to-egress, forward/additional]
        logs/additional:
          receivers: [forward/additional]
          processors: [batch/additional]
          exporters: [otlphttp/additional]

clusterCollector:
  configOverlay:
    connectors:
      forward/additional: {}
    exporters:
      otlphttp/additional:
        endpoint: https://destination.example.com:4318
    processors:
      batch/additional: {}
    service:
      pipelines:
        logs/copy:
          receivers: [forward/cxdot/from-integrations]
          exporters: [forward/cxdot/to-egress, forward/additional]
        logs/additional:
          receivers: [forward/additional]
          processors: [batch/additional]
          exporters: [otlphttp/additional]
```

**Configuration file**

Add `connectors`, `exporters`, `processors`, and `service.pipelines` as top-level
blocks in your configuration file.

```yaml expandable title="config.yaml" theme={null}
cxdot:
  # ...

connectors:
  forward/additional: {}

exporters:
  otlphttp/additional:
    endpoint: https://destination.example.com:4318

processors:
  batch/additional: {}

service:
  pipelines:
    logs/copy:
      receivers: [forward/cxdot/from-integrations]
      exporters: [forward/cxdot/to-egress, forward/additional]
    logs/additional:
      receivers: [forward/additional]
      processors: [batch/additional]
      exporters: [otlphttp/additional]
```

## Configure per-integration behavior

Each custom pipeline applies to either metrics, logs, or traces, and by default
affects data of that telemetry type from all enabled integrations. For example, a
pipeline for traces affects trace data from every integration that collects traces.

To change this behavior, use the `xcor.integration.name` attribute and
`cxdot.integration.instance.name` attribute to configure per-integration behavior for
your custom pipelines. This lets custom pipelines selectively route and process
telemetry data from specified integrations without affecting other integrations.

The `xcor.integration.name` attribute identifies an integration type. For example,
this `condition` block matches every configured MySQL integration instance. If
multiple MySQL integration instances are configured, it will match them all.

```yaml theme={null}
condition: resource.attributes["xcor.integration.name"] == "mysql"
```

The `cxdot.integration.instance.name` attribute identifies a
[named integration instance](/ingest/xcor/collector/configure/integrations#use-multiple-instances-of-the-same-integration).
Use this attribute in conjunction with the `xcor.integration.name` attribute. For
example, this `condition` block matches a configured NGINX integration with the
instance name `bravo`:

```yaml theme={null}
condition: resource.attributes["xcor.integration.name"] == "nginx" and resource.attributes["cxdot.integration.instance.name"] == "bravo"
```

### Process telemetry data from one integration type

The following example creates a custom pipeline named `metrics/filter-mysql` that
drops the `mysql.locks` metric from all instances of the
[MySQL integration](/ingest/xcor/integrations/collector/mysql). Metrics from other
integration types aren't affected, and logs and traces aren't affected.

**Helm values file**

Add `processors` and `service.pipelines` blocks to both the
`nodeCollector.configOverlay` and `clusterCollector.configOverlay` sections of
your Helm values file.

```yaml expandable title="values.yaml" theme={null}
nodeCollector:
  configOverlay:
    processors:
      filter/mysql-noise:
        metrics:
          metric:
            - 'resource.attributes["xcor.integration.name"] == "mysql" and name == "mysql.locks"'
    service:
      pipelines:
        metrics/filter-mysql:
          receivers: [forward/cxdot/from-integrations]
          processors: [filter/mysql-noise]
          exporters: [forward/cxdot/to-egress]

clusterCollector:
  configOverlay:
    processors:
      filter/mysql-noise:
        metrics:
          metric:
            - 'resource.attributes["xcor.integration.name"] == "mysql" and name == "mysql.locks"'
    service:
      pipelines:
        metrics/filter-mysql:
          receivers: [forward/cxdot/from-integrations]
          processors: [filter/mysql-noise]
          exporters: [forward/cxdot/to-egress]
```

**Configuration file**

Add `processors` and `service.pipelines` as top-level blocks in your configuration
file.

```yaml expandable title="config.yaml" theme={null}
cxdot:
  # ...

processors:
  filter/mysql-noise:
    metrics:
      metric:
        - 'resource.attributes["xcor.integration.name"] == "mysql" and name == "mysql.locks"'

service:
  pipelines:
    metrics/filter-mysql:
      receivers: [forward/cxdot/from-integrations]
      processors: [filter/mysql-noise]
      exporters: [forward/cxdot/to-egress]
```

### Route telemetry data from one integration type

The following example creates three custom pipelines. The first custom pipeline,
named `metrics/route-integrations`, receives metrics from all integrations and sends
them to a routing connector, which routes metrics to one of two branching pipelines.
The `metrics/mysql` pipeline sends metrics from all instances of the
[MySQL integration](/ingest/xcor/integrations/collector/mysql) to Cortex XCOR and
an additional destination. The `metrics/other` pipeline sends metrics from all other
integration types to Cortex XCOR.

Metrics from all integrations are sent to Cortex XCOR, but only metrics from
instances of the MySQL integration are sent to the additional destination. Logs and
traces aren't affected.

**Helm values file**

Add `connectors`, `exporters`, and `service.pipelines` blocks to both the
`nodeCollector.configOverlay` and `clusterCollector.configOverlay` sections of
your Helm values file.

```yaml expandable title="values.yaml" theme={null}
nodeCollector:
  configOverlay:
    connectors:
      routing/metrics:
        default_pipelines: [metrics/other]
        table:
          - context: resource
            condition: resource.attributes["xcor.integration.name"] == "mysql"
            pipelines: [metrics/mysql]
    exporters:
      otlphttp/additional:
        endpoint: https://destination.example.com:4318
    service:
      pipelines:
        metrics/route-integrations:
          receivers: [forward/cxdot/from-integrations]
          exporters: [routing/metrics]
        metrics/mysql:
          receivers: [routing/metrics]
          exporters: [forward/cxdot/to-egress, otlphttp/additional]
        metrics/other:
          receivers: [routing/metrics]
          exporters: [forward/cxdot/to-egress]

clusterCollector:
  configOverlay:
    connectors:
      routing/metrics:
        default_pipelines: [metrics/other]
        table:
          - context: resource
            condition: resource.attributes["xcor.integration.name"] == "mysql"
            pipelines: [metrics/mysql]
    exporters:
      otlphttp/additional:
        endpoint: https://destination.example.com:4318
    service:
      pipelines:
        metrics/route-integrations:
          receivers: [forward/cxdot/from-integrations]
          exporters: [routing/metrics]
        metrics/mysql:
          receivers: [routing/metrics]
          exporters: [forward/cxdot/to-egress, otlphttp/additional]
        metrics/other:
          receivers: [routing/metrics]
          exporters: [forward/cxdot/to-egress]
```

**Configuration file**

Add `connectors`, `exporters`, and `service.pipelines` as top-level blocks in your
configuration file.

```yaml expandable title="config.yaml" theme={null}
cxdot:
  # ...

connectors:
  routing/metrics:
    default_pipelines: [metrics/other]
    table:
      - context: resource
        condition: resource.attributes["xcor.integration.name"] == "mysql"
        pipelines: [metrics/mysql]

exporters:
  otlphttp/additional:
    endpoint: https://destination.example.com:4318

service:
  pipelines:
    metrics/route-integrations:
      receivers: [forward/cxdot/from-integrations]
      exporters: [routing/metrics]
    metrics/mysql:
      receivers: [routing/metrics]
      exporters: [forward/cxdot/to-egress, otlphttp/additional]
    metrics/other:
      receivers: [routing/metrics]
      exporters: [forward/cxdot/to-egress]
```


## Related topics

- [CXDOT Collector](/ingest/xcor/collector.md)
- [SNMP](/ingest/xcor/integrations/collector/snmp.md)
- [NGINX](/ingest/xcor/integrations/collector/nginx.md)
- [Build telemetry pipelines](/ingest/pipeline/v2/build.md)
- [Create and modify pipelines](/ingest/pipeline/v2/build/create-modify.md)


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