I Tried Out "Conditional Log Routing," a New Feature in CloudWatch Pipelines

Introduction

On October 14, 2026, CloudWatch pipelines added support for conditional log routing and chained parsers.

https://aws.amazon.com/jp/about-aws/whats-new/2026/10/cloudwatch-pipelines-log-routing/

Conditional routing is a feature that distributes log events to destination log groups based on conditions on fields extracted and processed in the pipeline. By separating log groups by destination, you can configure retention periods and access control to suit each purpose. For example, from JSON logs arriving in a single log group, you can separate events you want to exclude from downstream ETL and analysis into a different log group.

In this article, in the Tokyo region, I routed dummy JSON logs to two log groups based on conditions on level and type, and verified the following three points.

• Which log group receives events for each condition
• Whether routed events remain in the original log group
• What happens to the timestamp at the destination

I also verified the behavior of a chained parser, which extracts fields with grok after parsing the JSON.

Test setup

The subject is JSON modeled on Lambda Web Functions platform logs. The first event looks like this.

{"level":"WARN","type":"platform.invokeReport","seq":1,"region":"ap-northeast-1","record":{"requestId":"req-001","durationMs":11}}

The log groups are created under /test/pipelines-routing-1015/, and below I refer to them by their trailing names: source (the origin), warn, and summary.

For the pipeline, I configured one parse_json processor and three sinks. Events whose level is WARN go to warn, and events whose type is platform.invokeSummary go to summary. Events matching neither condition are sent to the original log group (@original) as the default route. The configuration is the YAML below. Replace <ACCOUNT_ID> with your own 12-digit AWS account ID.

pipeline:
  source:
    cloudwatch_logs:
      aws:
        sts_role_arn: "arn:aws:iam::<ACCOUNT_ID>:role/pipelines-routing-1015-role"
      log_event_metadata:
        data_source_name: "routing1015"
        data_source_type: "default"
  processor:
    - parse_json: {}
  sink:
    - cloudwatch_logs:
        log_group: "/test/pipelines-routing-1015/warn"
        when: 'level == "WARN"'
    - cloudwatch_logs:
        log_group: "/test/pipelines-routing-1015/summary"
        when: 'type == "platform.invokeSummary"'
    - cloudwatch_logs:
        log_group: "@original"
        is_default: true

Conditions are evaluated in the order the sinks are written in the YAML. An event matching multiple conditions is sent to the first matching sink.

In a pipeline with multiple sinks, specify exactly one default route and write it last. A single pipeline can have up to 10 sinks, including the default route. A condition expression can be up to 4,096 characters.

https://docs.aws.amazon.com/AmazonCloudWatch/latest/monitoring/pipeline-sinks.html

The source log group and the pipeline are associated with data source tags. In this example, set the cw:datasource:name and cw:datasource:type tags on the source source.

The value of each tag must match the corresponding value in log_event_metadata in the YAML. That is, cw:datasource:name corresponds to data_source_name, and cw:datasource:type corresponds to data_source_type. Events in a log group where both values match become the input of that pipeline.

aws logs tag-resource --resource-arn arn:aws:logs:ap-northeast-1:<ACCOUNT_ID>:log-group:/test/pipelines-routing-1015/source --tags cw:datasource:name=routing1015,cw:datasource:type=default

For --configuration, pass a JSON file that contains the entire YAML as a string in Body. You can validate it with validate-telemetry-pipeline-configuration before creating the pipeline.

jq -Rs '{Body: .}' routing-pipeline.yaml > routing-config.json
aws observabilityadmin validate-telemetry-pipeline-configuration --configuration file://routing-config.json
aws observabilityadmin create-telemetry-pipeline --name routing-1015-test --configuration file://routing-config.json

{
  "Version": "2012-10-17",
  "Statement": [
    {
      "Effect": "Allow",
      "Principal": { "Service": "logs.amazonaws.com" },
      "Action": "sts:AssumeRole"
    }
  ]
}

{
  "Version": "2012-10-17",
  "Statement": [
    {
      "Sid": "LogsProcessing",
      "Effect": "Allow",
      "Action": "logs:ProcessWithPipeline",
      "Resource": [
        "arn:aws:logs:ap-northeast-1:<ACCOUNT_ID>:log-group:/test/pipelines-routing-1015/source",
        "arn:aws:logs:ap-northeast-1:<ACCOUNT_ID>:log-group:/test/pipelines-routing-1015/source-chain"
      ]
    },
    {
      "Sid": "CreateRoutedLogGroups",
      "Effect": "Allow",
      "Action": "logs:CreateLogGroup",
      "Resource": "*"
    },
    {
      "Sid": "WriteRoutedLogEvents",
      "Effect": "Allow",
      "Action": [
        "logs:CreateLogStream",
        "logs:PutLogEvents"
      ],
      "Resource": "arn:aws:logs:ap-northeast-1:<ACCOUNT_ID>:log-group:/test/pipelines-routing-1015/*:*"
    }
  ]
}

Routing results by condition

I sent 17 dummy events in a single put-log-events call and counted the destinations for each input pattern.

The totals per destination were 7 for warn, 4 for summary, and 6 for the original log group, which matches the 17 input events. Routed events do not remain in the original log group.

Events matching both conditions went to the warn route, which is written first in the YAML. The event with lowercase warn did not match the level == "WARN" condition and went to the original log group.

With this configuration, which uses only parse_json, the bodies that arrived in warn and summary were the same strings as the input.

Timestamps at the destination

I sent a total of 4 events, alternating WARN and INFO, with timestamps of 1 hour ago, 30 minutes ago, 10 minutes ago, and now, respectively. Times are in JST.

The two events that arrived in warn got the same timestamp whether their original time was 1 hour ago or 10 minutes ago. This value is about 0.2 seconds before the ingestion time (ingestionTime), and appears to be the time at which the pipeline processed the events. The 17 events sent in a batch in the previous section had send-time timestamps that differed by 1 millisecond each, but all 11 events that arrived at the destinations had the same timestamp.

If you want to aggregate or query the logs at the destination by time, the sender needs to include a time field in the log body.

Handling unneeded logs

In a pipeline whose source is CloudWatch Logs, I could not discard events with drop_events. To exclude unneeded logs from downstream ETL and analysis, route them to a separate log group with a short retention period.

When I set drop_events on a cloudwatch_logs source, validate-telemetry-pipeline-configuration returned PROCESSOR_SOURCE_INCOMPATIBLE, and create-telemetry-pipeline also failed with a ValidationException. This time, I created summary in advance with a retention period of 1 day and routed the 4 platform.invokeSummary events whose level is INFO to it.

Retention follows the settings of the destination log group. A destination log group for a conditional route is created automatically when events are first sent to it if it does not exist. This time, the automatically created warn had no retention period set, so its logs were retained indefinitely. For destinations whose retention you want to manage, I recommend creating the log groups before creating the pipeline.

On the other hand, the destination of the default route is not created automatically. It must be created in advance, before creating the pipeline.

Chained parser

For the chained parser, I tried a configuration that places grok after parse_json to parse the JSON message field. I prepared a separate log group source-chain as the source and attached the data source tags with the same procedure as before. The tag values are chain1015, corresponding to data_source_name in the YAML, and default, corresponding to data_source_type. The configuration is the YAML below.

pipeline:
  source:
    cloudwatch_logs:
      aws:
        sts_role_arn: "arn:aws:iam::<ACCOUNT_ID>:role/pipelines-routing-1015-role"
      log_event_metadata:
        data_source_name: "chain1015"
        data_source_type: "default"
  processor:
    - parse_json: {}
    - grok:
        match:
          message: ["%{WORD:method} %{URIPATH:path} %{NUMBER:status} %{NUMBER:duration_ms}ms"]
  sink:
    - cloudwatch_logs:
        log_group: "@original"

With this configuration, which has only one sink, validation passed and it worked without is_default.

These are the results of sending 3 JSON events whose message holds a string.

All values extracted by grok were strings. The body of an event that matched the pattern was rewritten as JSON including the additional fields and then stored. This differs from the behavior in the routing configuration with only parse_json, where the body did not change.

For details such as processor placement and limits on the number of processors, see the following documentation.

https://docs.aws.amazon.com/AmazonCloudWatch/latest/monitoring/parser-processors.html

Pricing

CloudWatch pipelines itself has no additional charge, but the usual CloudWatch Logs ingestion charges apply. When CloudWatch Logs is the source, the ingestion charge is incurred before processing by the pipeline. Therefore, ingestion charges also apply to logs that you route to a short-retention log group to exclude them from downstream processing.

https://docs.aws.amazon.com/AmazonCloudWatch/latest/monitoring/cloudwatch-pipelines.html

Summary

With conditional log routing in CloudWatch pipelines, you can route logs to different log groups at ingestion time based on their content. In addition to separating logs that need different retention periods or access control, narrowing the target log groups for downstream ETL or CloudWatch Logs Insights can reduce the amount of data processed and the scan volume of queries.

CloudWatch pipelines itself has no additional charge, but the ingestion charge is incurred before routing. What short retention reduces is the storage cost.

Note that the original event timestamp is not preserved at the destination, so if you need to aggregate or search by the original time, you must include a time field in the log body as well.