Add opt-in Elasticsearch/OpenSearch data streams (#937)

* Add opt-in Elasticsearch/OpenSearch data streams (#842)

New use_data_streams option in [elasticsearch] and [opensearch] saves
each report type to a data stream named like today's index without the
date suffix, instead of daily or monthly indexes. parsedmarc puts an
exact-name index template (priority 501, @timestamp mapped as date,
<stream>@custom component template on Elasticsearch) and creates the
stream; retention is left to operator-managed ILM/ISM policies, with
documented examples. Dated indexes are untouched and still searched for
duplicates and matched by the dashboards. Every document now carries
@timestamp and is saved with op_type=create in both modes.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

* Fix data stream prose found by cross-check review (layer 1)

Aggregate @timestamp is each record's begin date, not the report's; the
cluster only rolls backing indices over under an attached policy; move
the Elasticsearch 8.7 note after the template it refers to; clarify the
name-collision paragraph, OpenSearch-side comments, the migrate_indexes
docstring, and the test-module headers.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

* Tighten names and wiring tests found by cross-check review (layer 2)

Split the use_data_streams CLI passthrough test per backend so a call
site reading the other section's flag fails; name the dated-mode tests
for both changes they assert; compute index_date only in dated mode in
every save function.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

* Align data stream docs with Elastic/OpenSearch references (layer 3)

Say the stream must exist before reindexing into it (ES reindex API:
auto-creation needs a data-stream template); an existing stream's write
index needs the ILM policy attached directly since nothing rolls it over
otherwise; target the stream name in the _settings example; note
OpenSearch's leading-dot rule for stream names and Fleet's template
priorities.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

* Cover the Loki client-initialization error wrapper

_parse_config accepts a [loki] token alongside username/password; only
the LokiClient constructor rejects the combination. Exercise that path
through _init_output_clients with the real constructor so the
RuntimeError("Loki: ...") wrapper Codecov flagged as uncovered is tested.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

---------

Co-authored-by: Claude Opus 5.5 <noreply@anthropic.com>

CHANGELOG.md

@@ -11,6 +11,10 @@
 
 - **S3 object keys and metadata are now sanitized, and each key ends in a hash of the fields that identify the report** ([GHSA-7wqr-f6g9-cm29](https://github.com/domainaware/parsedmarc/security/advisories/GHSA-7wqr-f6g9-cm29), affects 11.1.0 and earlier). `S3Client.save_report_to_s3()` put the sender-controlled report ID into the object key as-is, and sent `org_name` and the other metadata values as-is. A report ID such as `../../x` added path segments to the key, an ID could carry `\`, `%`, and the other characters AWS lists as "characters to avoid", or control characters, which AWS says "might require special handling", and a long ID could push the key past the 1,024-byte limit ([AWS, "Naming Amazon S3 objects"](https://docs.aws.amazon.com/AmazonS3/latest/userguide/object-keys.html)). botocore rejects non-ASCII metadata with `ParamValidationError`, so a report from an organization with a non-ASCII name was never saved, and its batch was held back and retried. A bare CR or LF in a metadata value made botocore raise `HTTPClientError: Invalid header value` before sending, which held the batch back the same way, while CR or LF followed by a space or tab (an obs-fold) was sent in the `x-amz-meta-*` header; the 2 KB metadata limit ([AWS, "Working with object metadata"](https://docs.aws.amazon.com/AmazonS3/latest/userguide/UsingMetadata.html)) was not enforced. The key also identified a report only by its type, begin day, and report ID, so anyone who could send reports to a monitored mailbox could replace another reporter's object for that day by reusing its report ID. The report ID is now percent-encoded into a single key segment and shortened so the whole key fits in 1,024 bytes. The encoded characters are `/`, every AWS "character to avoid" (`` \ { } ^ [ ] " < > ~ # | ` `` and `%`), control characters, and non-ASCII characters, so `%` appears in a key only as the escape character. The punctuation real report IDs use, such as `:`, `@`, `.`, and `_`, is kept. Each metadata value is percent-encoded (`%`, control characters, and non-ASCII characters) so `urllib.parse.unquote()` reverses it, and is cut to 256 characters, which keeps the total under 2 KB. Each key now ends in `_<hash>.json`, where the hash is the first 16 hex characters of a SHA-256 hash of the report type and the fields that identify the report (for aggregate reports: organization name, report ID, policy domain, begin date, and end date; for SMTP TLS reports: organization name, report ID, begin date, end date, and each policy domain). Enrichment such as reverse DNS is not hashed, so a retried report still overwrites its own object. This narrows the overwrite but does not close it: these fields are not secret, and a forged report whose reporter name, report ID, policy domain, and dates all match a real one still replaces it. Turning on S3 Versioning on the bucket keeps both. Compatibility: every report's key changes. Objects saved by earlier versions keep their old keys and are not rewritten, and a batch that is retried across the upgrade is saved again under its new key. Reported by helixnebulatech.
 
+### New features
+
+- **Elasticsearch and OpenSearch output can now write to data streams** ([#842](https://github.com/domainaware/parsedmarc/issues/842), relates to [#476](https://github.com/domainaware/parsedmarc/issues/476)). The new `use_data_streams` option in `[elasticsearch]` and `[opensearch]` (default `False`) saves each report type to one data stream named like today's indexes without the date suffix (`dmarc_aggregate`, `dmarc_failure`, `smtp_tls`, with `index_prefix`, `index_suffix`, or the `index_prefix_domain_map` prefix applied as before), instead of creating an index per day or month. That gives each report type one stable name, keeps the shard count down, and lets the cluster roll over and delete old data. parsedmarc puts an index template named after each stream, whose `index_patterns` is the exact stream name so it never applies to the dated indexes, and creates the stream; it does not create or manage retention policies. On Elasticsearch, attach an ILM policy through an optional `<stream>@custom` component template, which the template includes; on OpenSearch, an ISM policy attaches itself through its `ism_template`. `docs/source/elasticsearch.md` and `docs/source/opensearch.md` have example policies. Existing dated indexes are left alone and keep matching the dashboards' `dmarc_aggregate*`, `dmarc_f*`, and `smtp_tls*` patterns next to the stream, and the duplicate check searches both, so a report saved before the switch is not saved again after it. `monthly_indexes` is ignored when `use_data_streams` is on, and a warning says so. If a concrete index or alias already holds a stream's name, as an undated index from a release before 5.0.0 might, saving fails with an error that says so instead of writing into it. Every saved document now also carries an `@timestamp` field, in both modes: each record's begin date (`date_begin`) for aggregate reports, which is the report's begin date unless parsedmarc split a report spanning more than a day into daily records; the arrival date for failure reports; and the report's begin date for SMTP TLS reports. These are the same dates that pick the dated index. Documents are now saved with `op_type=create`, which data streams require; without a document ID this is the same as the default for dated indexes.
+
 ### Changes
 
 - **mailsuite 2.3.3 or later is now required.** mailsuite 2.3.3 ([GHSA-9jx3-p8wm-g8vw](https://github.com/seanthegeek/mailsuite/security/advisories/GHSA-9jx3-p8wm-g8vw)) makes the Maildir connection return a message's stored bytes from `fetch_message()` and rename the file in `move_message()`, instead of re-serializing it. A report email that cannot be re-serialized (the [GHSA-qxwv-2vwj-v828](https://github.com/domainaware/parsedmarc/security/advisories/GHSA-qxwv-2vwj-v828) messages) is therefore now moved to the `Invalid` folder on Maildir too, instead of staying in the inbox and being logged on every run.

docs/source/elasticsearch.md

@@ -405,13 +405,170 @@ curl -X POST "http://localhost:9200/smtp_tls*/_update_by_query?conflicts=proceed
 It works identically on OpenSearch; just adjust the URL and credentials, same
 as the aggregate command above.
 
-## Records retention
+## Data streams and records retention
 
 Starting in version 5.0.0, `parsedmarc` stores data in a separate
-index for each day to make it easy to comply with records
-retention regulations such as GDPR. For more information,
+index for each day (or each month, with `monthly_indexes`), named after
+the report's date, such as `dmarc_aggregate-2024-03-31`. That makes it
+easy to comply with records retention regulations such as GDPR: delete
+the indexes for the dates you no longer keep. For more information,
 check out the Elastic guide to [managing time-based indexes efficiently](https://www.elastic.co/blog/managing-time-based-indices-efficiently).
 
+Set `use_data_streams = True` in the `[elasticsearch]` section to write
+to [data streams] instead. Each report type then goes to one data stream
+named like the indexes without the date: `dmarc_aggregate`,
+`dmarc_failure`, and `smtp_tls`, with `index_prefix`, `index_suffix`, or
+the `index_prefix_domain_map` prefix applied as usual. The cluster
+creates the stream's backing indices (`.ds-dmarc_aggregate-…`), and an
+ILM policy you attach rolls them over and deletes old ones, so there is
+one stable name per report type and far fewer shards.
+
+Before a parsedmarc process first saves to a stream, it puts an index
+template named after the stream and creates the stream if it does not
+exist yet. The template:
+
+- matches exactly the stream's name, with no wildcard, so it never applies
+  to the dated indexes;
+- has priority 501 and sets `number_of_shards` and `number_of_replicas`
+  from the configuration, for each new backing index;
+- maps `@timestamp` as a date. Every document parsedmarc saves carries
+  `@timestamp`: each record's begin date (`date_begin`) for aggregate
+  reports, which is the report's begin date unless parsedmarc split a
+  report spanning more than a day into daily records; the arrival date
+  for failure reports; and the report's begin date for SMTP TLS reports.
+  The other fields are mapped dynamically, as in the dated indexes;
+- includes a component template named `<stream>@custom`, such as
+  `dmarc_aggregate@custom`, if one exists.
+
+Data streams need Elasticsearch 8.7 or later, because the template uses
+`ignore_missing_component_templates` to make the `@custom` component
+template optional.
+
+parsedmarc puts the template again in every process, so changes made to
+it directly do not last. Put your own settings, such as the ILM policy,
+in the `@custom` component template instead.
+
+parsedmarc does not create or change any ILM policy. Here is an example
+that rolls each stream over every 30 days, or sooner when a primary
+shard reaches 50 GB, and deletes a backing index 180 days after its
+rollover:
+
+```text
+PUT _ilm/policy/parsedmarc
+{
+  "policy": {
+    "phases": {
+      "hot": {
+        "actions": {
+          "rollover": {
+            "max_age": "30d",
+            "max_primary_shard_size": "50gb"
+          }
+        }
+      },
+      "delete": {
+        "min_age": "180d",
+        "actions": {
+          "delete": {}
+        }
+      }
+    }
+  }
+}
+```
+
+Attach it to each stream through its `@custom` component template:
+
+```text
+PUT _component_template/dmarc_aggregate@custom
+{
+  "template": {
+    "settings": {
+      "index.lifecycle.name": "parsedmarc"
+    }
+  }
+}
+```
+
+Repeat that for `dmarc_failure@custom` and `smtp_tls@custom`, and for the
+prefixed or suffixed names if you use them. The component template
+applies only to backing indices created after it. A stream that already
+exists has a write index with no policy, so nothing would ever roll it
+over; attach the policy to its existing backing indices as well:
+
+```text
+PUT dmarc_aggregate/_settings
+{
+  "index.lifecycle.name": "parsedmarc"
+}
+```
+
+The delete phase's `min_age` counts from the rollover, not from the
+report dates, so a backing index is deleted 180 days after it stopped
+receiving documents, whatever dates the reports in it cover. If you need
+retention by report date, keep using dated indexes and delete them by
+name.
+
+On Elastic Cloud Serverless, which has no ILM, use a
+[data stream lifecycle] instead:
+
+```text
+PUT _data_stream/dmarc_aggregate/_lifecycle
+{
+  "data_retention": "180d"
+}
+```
+
+### Switching an existing installation to data streams
+
+Dated indexes that already exist are left alone. They keep working next
+to the streams: the shipped dashboards use the `dmarc_aggregate*`,
+`dmarc_f*`, and `smtp_tls*` index patterns, which match both, and the
+check for duplicate reports searches both, so a report that is already
+in a dated index is not saved again into a stream. You can delete the
+dated indexes when they age out of your retention period, or move their
+documents into the streams. Documents saved by earlier releases have no
+`@timestamp`, which a data stream requires, so add it while reindexing.
+Run parsedmarc once with `use_data_streams = True` first, so the stream
+and its template exist; a reindex into a name that is not a data stream
+yet creates an ordinary index of that name instead:
+
+```text
+POST _reindex
+{
+  "source": {"index": "dmarc_aggregate-*"},
+  "dest": {"index": "dmarc_aggregate", "op_type": "create"},
+  "script": {
+    "lang": "painless",
+    "source": "ctx._source['@timestamp'] = ctx._source.date_begin"
+  }
+}
+```
+
+Use `date_begin` for `smtp_tls-*` too, and `arrival_date` for
+`dmarc_failure-*`. Delete the dated indexes once the reindex has
+finished; until then, the dashboards count every document twice.
+
+If saving fails with `cannot create the data stream dmarc_aggregate: an
+index or alias with that name already exists`, an index or alias already
+uses the stream's name. parsedmarc releases before 5.0.0 wrote to an
+undated `dmarc_aggregate` index, for example. parsedmarc refuses to
+write into such an index rather than mixing new documents into it. Move
+its documents to a dated name, which the
+dashboards still match, and delete it:
+
+```text
+POST _reindex
+{
+  "source": {"index": "dmarc_aggregate"},
+  "dest": {"index": "dmarc_aggregate-legacy"}
+}
+
+DELETE dmarc_aggregate
+```
+
+[data stream lifecycle]: https://www.elastic.co/docs/manage-data/lifecycle/data-stream
+[data streams]: https://www.elastic.co/docs/manage-data/data-store/data-streams
 [elasticsearch]: https://www.elastic.co/guide/en/elasticsearch/reference/current/rpm.html
 [opensearch_dashboards.ndjson]: https://raw.githubusercontent.com/domainaware/parsedmarc/master/dashboards/opensearch/opensearch_dashboards.ndjson
 [kibana]: https://www.elastic.co/guide/en/kibana/current/rpm.html

docs/source/opensearch.md

@@ -7,8 +7,155 @@ To set up visual dashboards of DMARC data, install OpenSearch and Grafana.
 OpenSearch: https://opensearch.org/docs/latest/install-and-configure/install-opensearch/index/
 Grafana: https://grafana.com/docs/grafana/latest/setup-grafana/installation/
 
-## Records retention
+## Data streams and records retention
 
 Starting in version 5.0.0, `parsedmarc` stores data in a separate
-index for each day to make it easy to comply with records
-retention regulations such as GDPR.
+index for each day (or each month, with `monthly_indexes`), named after
+the report's date, such as `dmarc_aggregate-2024-03-31`. That makes it
+easy to comply with records retention regulations such as GDPR: delete
+the indexes for the dates you no longer keep.
+
+Set `use_data_streams = True` in the `[opensearch]` section to write to
+[data streams] instead. Each report type then goes to one data stream
+named like the indexes without the date: `dmarc_aggregate`,
+`dmarc_failure`, and `smtp_tls`, with `index_prefix`, `index_suffix`, or
+the `index_prefix_domain_map` prefix applied as usual. The cluster
+creates the stream's backing indices (`.ds-dmarc_aggregate-000001`, …),
+and an [ISM] policy you attach rolls them over and deletes old data, so
+there is one stable name per report type and far fewer shards. Amazon
+OpenSearch Serverless (`aws_service = aoss`) has no data streams or
+ISM, so leave `use_data_streams` off there.
+
+Before a parsedmarc process first saves to a stream, it puts an index
+template named after the stream and creates the stream if it does not
+exist yet. The template:
+
+- matches exactly the stream's name, with no wildcard, so it never applies
+  to the dated indexes;
+- has priority 501 and sets `number_of_shards` and `number_of_replicas`
+  from the configuration, for each new backing index;
+- maps `@timestamp` as a date. Every document parsedmarc saves carries
+  `@timestamp`: each record's begin date (`date_begin`) for aggregate
+  reports, which is the report's begin date unless parsedmarc split a
+  report spanning more than a day into daily records; the arrival date
+  for failure reports; and the report's begin date for SMTP TLS reports.
+  The other fields are mapped dynamically, as in the dated indexes.
+
+parsedmarc puts the template again in every process, so changes made to
+it directly do not last.
+
+parsedmarc does not create or change any ISM policy. Here is an example
+that rolls each stream over every 30 days, or sooner when a primary
+shard reaches 50 GB, and deletes a backing index 180 days after it was
+created. Its `ism_template` attaches it to every backing index the
+listed streams create from now on. List the stream names themselves,
+including any prefix or suffix: a `.ds-dmarc_aggregate-*` pattern does
+not match the backing indices.
+
+```text
+PUT _plugins/_ism/policies/parsedmarc
+{
+  "policy": {
+    "description": "parsedmarc data stream retention",
+    "default_state": "hot",
+    "states": [
+      {
+        "name": "hot",
+        "actions": [
+          {
+            "rollover": {
+              "min_index_age": "30d",
+              "min_primary_shard_size": "50gb"
+            }
+          }
+        ],
+        "transitions": [
+          {
+            "state_name": "delete",
+            "conditions": {"min_index_age": "180d"}
+          }
+        ]
+      },
+      {
+        "name": "delete",
+        "actions": [{"delete": {}}],
+        "transitions": []
+      }
+    ],
+    "ism_template": [
+      {
+        "index_patterns": ["dmarc_aggregate", "dmarc_failure", "smtp_tls"],
+        "priority": 100
+      }
+    ]
+  }
+}
+```
+
+The `ism_template` only applies to backing indices created after the
+policy. To attach the policy to the backing indices of a stream that
+already exists:
+
+```text
+POST _plugins/_ism/add/dmarc_aggregate
+{
+  "policy_id": "parsedmarc"
+}
+```
+
+`min_index_age` counts from when the backing index was created, not
+from the report dates, so a backing index is deleted 180 days after it
+was created, whatever dates the reports in it cover. If you need
+retention by report date, keep using dated indexes and delete them by
+name.
+
+### Switching an existing installation to data streams
+
+Dated indexes that already exist are left alone. They keep working next
+to the streams: the shipped dashboards use the `dmarc_aggregate*`,
+`dmarc_f*`, and `smtp_tls*` index patterns, which match both, and the
+check for duplicate reports searches both, so a report that is already
+in a dated index is not saved again into a stream. You can delete the
+dated indexes when they age out of your retention period, or move their
+documents into the streams. Documents saved by earlier releases have no
+`@timestamp`, which a data stream requires, so add it while reindexing.
+Run parsedmarc once with `use_data_streams = True` first, so the stream
+and its template exist; a reindex into a name that is not a data stream
+yet creates an ordinary index of that name instead:
+
+```text
+POST _reindex
+{
+  "source": {"index": "dmarc_aggregate-*"},
+  "dest": {"index": "dmarc_aggregate", "op_type": "create"},
+  "script": {
+    "lang": "painless",
+    "source": "ctx._source['@timestamp'] = ctx._source.date_begin"
+  }
+}
+```
+
+Use `date_begin` for `smtp_tls-*` too, and `arrival_date` for
+`dmarc_failure-*`. Delete the dated indexes once the reindex has
+finished; until then, the dashboards count every document twice.
+
+If saving fails with `cannot create the data stream dmarc_aggregate: an
+index or alias with that name already exists`, an index or alias already
+uses the stream's name. parsedmarc releases before 5.0.0 wrote to an
+undated `dmarc_aggregate` index, for example. parsedmarc refuses to
+write into such an index rather than mixing new documents into it. Move
+its documents to a dated name, which the
+dashboards still match, and delete it:
+
+```text
+POST _reindex
+{
+  "source": {"index": "dmarc_aggregate"},
+  "dest": {"index": "dmarc_aggregate-legacy"}
+}
+
+DELETE dmarc_aggregate
+```
+
+[data streams]: https://docs.opensearch.org/latest/im-plugin/data-streams/
+[ISM]: https://docs.opensearch.org/latest/im-plugin/ism/index/

docs/source/usage.md

@@ -591,16 +591,27 @@ The full set of configuration options is:
   - `index_suffix` - str: A suffix to apply to the index names
   - `index_prefix` - str: A prefix to apply to the index names
   - `monthly_indexes` - bool: Use monthly indexes instead of daily indexes
+    (ignored, with a warning, when `use_data_streams` is `True`)
   - `number_of_shards` - int: The number of shards to use when
-    creating the index (Default: `1`)
+    creating the index, or each backing index of a data stream
+    (Default: `1`)
   - `number_of_replicas` - int: The number of replicas to use when
-    creating the index (Default: `0`)
+    creating the index, or each backing index of a data stream
+    (Default: `0`)
+  - `use_data_streams` - bool: Save each report type to a data stream
+    named like the index without the date suffix (e.g. `dmarc_aggregate`)
+    instead of to daily or monthly indexes. parsedmarc creates the stream
+    and its index template; retention is left to an ILM policy you attach. Requires Elasticsearch
+    8.7 or later.
+    See [Data streams and records retention](elasticsearch.md#data-streams-and-records-retention)
+    (Default: `False`)
   - `serverless` - bool: Set to `True` when targeting an Elastic Cloud
     Serverless project. Serverless manages sharding and replication itself
     and rejects the `number_of_shards` / `number_of_replicas` index settings
     with HTTP 400. With this flag set, parsedmarc strips those keys from the
-    settings sent at index creation; any other settings (e.g.
-    `refresh_interval`) are passed through unchanged (Default: `False`)
+    settings sent at index creation (or in a data stream's index template);
+    any other settings (e.g. `refresh_interval`) are passed through
+    unchanged (Default: `False`)
 - `opensearch`
 
   Requires the `opensearch` extra: `pip install "parsedmarc[opensearch]"`
@@ -629,10 +640,21 @@ The full set of configuration options is:
   - `index_suffix` - str: A suffix to apply to the index names
   - `index_prefix` - str: A prefix to apply to the index names
   - `monthly_indexes` - bool: Use monthly indexes instead of daily indexes
+    (ignored, with a warning, when `use_data_streams` is `True`)
   - `number_of_shards` - int: The number of shards to use when
-    creating the index (Default: `1`)
+    creating the index, or each backing index of a data stream
+    (Default: `1`)
   - `number_of_replicas` - int: The number of replicas to use when
-    creating the index (Default: `0`)
+    creating the index, or each backing index of a data stream
+    (Default: `0`)
+  - `use_data_streams` - bool: Save each report type to a data stream
+    named like the index without the date suffix (e.g. `dmarc_aggregate`)
+    instead of to daily or monthly indexes. parsedmarc creates the stream
+    and its index template; retention is left to an ISM policy you attach.
+    OpenSearch rejects data stream names that start with `.`, so an
+    `index_prefix` starting with `.` cannot be used with this option.
+    See [Data streams and records retention](opensearch.md#data-streams-and-records-retention)
+    (Default: `False`)
 - `splunk_hec`
   - `url` - str: The URL of the Splunk HTTP Events Collector (HEC)
   - `token` - str: The HEC token

parsedmarc/cli.py

@@ -1193,6 +1193,17 @@ def _parse_config(config: ConfigParser, opts):
             opts.elasticsearch_serverless = elasticsearch_config.getboolean(
                 "serverless"
             )
+        if "use_data_streams" in elasticsearch_config:
+            opts.elasticsearch_use_data_streams = bool(
+                elasticsearch_config.getboolean("use_data_streams")
+            )
+        if getattr(opts, "elasticsearch_use_data_streams", False) and getattr(
+            opts, "elasticsearch_monthly_indexes", False
+        ):
+            logger.warning(
+                "[elasticsearch] monthly_indexes is ignored because "
+                "use_data_streams is enabled"
+            )
 
     if "opensearch" in config:
         opensearch_config = config["opensearch"]
@@ -1246,6 +1257,17 @@ def _parse_config(config: ConfigParser, opts):
             opts.opensearch_aws_region = opensearch_config["aws_region"].strip()
         if "aws_service" in opensearch_config:
             opts.opensearch_aws_service = opensearch_config["aws_service"].strip()
+        if "use_data_streams" in opensearch_config:
+            opts.opensearch_use_data_streams = bool(
+                opensearch_config.getboolean("use_data_streams")
+            )
+        if getattr(opts, "opensearch_use_data_streams", False) and getattr(
+            opts, "opensearch_monthly_indexes", False
+        ):
+            logger.warning(
+                "[opensearch] monthly_indexes is ignored because "
+                "use_data_streams is enabled"
+            )
 
     if "splunk_hec" in config.sections():
         hec_config = config["splunk_hec"]
@@ -1813,9 +1835,12 @@ def _search_alias_snapshot() -> list[tuple[Any, Any, Any]]:
         # private is the same liberty this function already takes with
         # ``connections``; without it, a failed reload that flipped
         # ``[elasticsearch] serverless`` would pair the restored old client
-        # with the new config's flag. opensearch.py declares no module
-        # globals at all (no ``global`` statement in the file), so there is
-        # nothing to pair with it.
+        # with the new config's flag. opensearch.py has no equivalent flag,
+        # so there is nothing to pair with it. Both set_hosts() also clear
+        # their module's _READY_DATA_STREAMS memo, which is deliberately not
+        # snapshotted: an emptied memo only makes the next save re-put its
+        # data stream's index template, which is idempotent, so it is
+        # correct for the restored client too.
         serverless = elastic._SERVERLESS if module is elastic else None
         snapshot.append((module, connection, serverless))
     return snapshot
@@ -2140,8 +2165,10 @@ def _build_output_clients(opts, clients, index_prefix_domain_map=None):
     # places rather than one: connections.create_connection() registers the
     # new client under the ``default`` alias, and elastic.set_hosts() also
     # assigns elastic._SERVERLESS. _init_output_clients() rolls both back if
-    # a later step fails. They are still initialized last, so that a failure
-    # in any other output happens before either registry has been touched.
+    # a later step fails. (Both set_hosts() also clear their data stream
+    # memo, which needs no rollback; see _search_alias_snapshot().) They are
+    # still initialized last, so that a failure in any other output happens
+    # before either registry has been touched.
     if opts.save_aggregate or opts.save_failure or opts.save_smtp_tls:
         # Scoped to the same condition as the constructors below, which is
         # also the condition under which process_reports() dereferences
@@ -2562,6 +2589,7 @@ def _main():
                             index_prefix=opts.elasticsearch_index_prefix
                             or get_index_prefix(report),
                             monthly_indexes=opts.elasticsearch_monthly_indexes,
+                            use_data_streams=opts.elasticsearch_use_data_streams,
                             number_of_shards=shards,
                             number_of_replicas=replicas,
                         )
@@ -2582,6 +2610,7 @@ def _main():
                             index_prefix=opts.opensearch_index_prefix
                             or get_index_prefix(report),
                             monthly_indexes=opts.opensearch_monthly_indexes,
+                            use_data_streams=opts.opensearch_use_data_streams,
                             number_of_shards=shards,
                             number_of_replicas=replicas,
                         )
@@ -2661,6 +2690,7 @@ def _main():
                             index_prefix=opts.elasticsearch_index_prefix
                             or get_index_prefix(report),
                             monthly_indexes=opts.elasticsearch_monthly_indexes,
+                            use_data_streams=opts.elasticsearch_use_data_streams,
                             number_of_shards=shards,
                             number_of_replicas=replicas,
                         )
@@ -2681,6 +2711,7 @@ def _main():
                             index_prefix=opts.opensearch_index_prefix
                             or get_index_prefix(report),
                             monthly_indexes=opts.opensearch_monthly_indexes,
+                            use_data_streams=opts.opensearch_use_data_streams,
                             number_of_shards=shards,
                             number_of_replicas=replicas,
                         )
@@ -2760,6 +2791,7 @@ def _main():
                             index_prefix=opts.elasticsearch_index_prefix
                             or get_index_prefix(report),
                             monthly_indexes=opts.elasticsearch_monthly_indexes,
+                            use_data_streams=opts.elasticsearch_use_data_streams,
                             number_of_shards=shards,
                             number_of_replicas=replicas,
                         )
@@ -2780,6 +2812,7 @@ def _main():
                             index_prefix=opts.opensearch_index_prefix
                             or get_index_prefix(report),
                             monthly_indexes=opts.opensearch_monthly_indexes,
+                            use_data_streams=opts.opensearch_use_data_streams,
                             number_of_shards=shards,
                             number_of_replicas=replicas,
                         )
@@ -3071,6 +3104,7 @@ def _main():
         elasticsearch_password=None,
         elasticsearch_api_key=None,
         elasticsearch_serverless=False,
+        elasticsearch_use_data_streams=False,
         opensearch_hosts=None,
         opensearch_timeout=60,
         opensearch_number_of_shards=1,
@@ -3087,6 +3121,7 @@ def _main():
         opensearch_auth_type="basic",
         opensearch_aws_region=None,
         opensearch_aws_service="es",
+        opensearch_use_data_streams=False,
         kafka_hosts=None,
         kafka_username=None,
         kafka_password=None,

parsedmarc/elastic.py

@@ -4,7 +4,7 @@ from __future__ import annotations
 
 from typing import TYPE_CHECKING, Any
 
-from elasticsearch import Elasticsearch
+from elasticsearch import Elasticsearch, NotFoundError
 from elasticsearch.dsl import (
     Boolean,
     Date,
@@ -43,6 +43,22 @@ _SERVERLESS = False
 # settings (e.g. ``refresh_interval``) are accepted and pass through.
 _SERVERLESS_REJECTED_SETTINGS = frozenset({"number_of_shards", "number_of_replicas"})
 
+# Names of the data streams whose index template and stream
+# ``create_data_streams`` has already set up in this process, so each one
+# costs its API calls once per process rather than once per saved document.
+# ``set_hosts`` clears it, so a new connection (including a SIGHUP reload,
+# which may point at another cluster or change the index settings)
+# re-applies the template. Forgetting an entry is always safe: it only
+# repeats idempotent calls.
+_READY_DATA_STREAMS: set[str] = set()
+
+# Priority of the index template ``create_data_streams`` installs.
+# Elasticsearch's built-in templates have priority 100 and Fleet's
+# integration templates up to 200, and its documentation recommends a
+# priority higher than 500 for custom templates whose patterns overlap them. The template's ``index_patterns`` is the
+# exact stream name, so an overlap is unlikely in the first place.
+_DATA_STREAM_TEMPLATE_PRIORITY = 501
+
 # Guard query for the dkim_results_combined/spf_results_combined backfill
 # (see ``migrate_indexes``). Matches only documents that have at least one
 # DKIM or SPF auth result and are missing the corresponding combined field.
@@ -620,6 +636,7 @@ def set_hosts(
     # Module-global; see the _SERVERLESS comment at the top of the module.
     global _SERVERLESS
     _SERVERLESS = serverless
+    _READY_DATA_STREAMS.clear()
     if not isinstance(hosts, list):
         hosts = [hosts]
     scheme = "https://" if use_ssl else "http://"
@@ -639,6 +656,27 @@ def set_hosts(
     return connections.create_connection(**conn_params)
 
 
+def _effective_index_settings(settings: dict[str, Any] | None) -> dict[str, Any]:
+    """Return the index settings to send, adjusted for Serverless.
+
+    Args:
+        settings (dict | None): The requested index settings, or ``None``
+            for the defaults (one shard, no replicas).
+
+    Returns:
+        dict: In Serverless mode, the requested settings minus the keys in
+        ``_SERVERLESS_REJECTED_SETTINGS``, and no defaults at all;
+        otherwise a copy of the requested settings, or the defaults.
+    """
+    if settings is None:
+        return {} if _SERVERLESS else {"number_of_shards": 1, "number_of_replicas": 0}
+    if _SERVERLESS:
+        return {
+            k: v for k, v in settings.items() if k not in _SERVERLESS_REJECTED_SETTINGS
+        }
+    return dict(settings)
+
+
 def create_indexes(names: list[str], settings: dict[str, Any] | None = None):
     """
     Create Elasticsearch indexes
@@ -649,16 +687,7 @@ def create_indexes(names: list[str], settings: dict[str, Any] | None = None):
             ``_SERVERLESS_REJECTED_SETTINGS`` are filtered out and the
             remaining keys are passed through; defaults are skipped entirely.
     """
-    if settings is None:
-        effective_settings: dict[str, Any] = (
-            {} if _SERVERLESS else {"number_of_shards": 1, "number_of_replicas": 0}
-        )
-    elif _SERVERLESS:
-        effective_settings = {
-            k: v for k, v in settings.items() if k not in _SERVERLESS_REJECTED_SETTINGS
-        }
-    else:
-        effective_settings = dict(settings)
+    effective_settings = _effective_index_settings(settings)
 
     for name in names:
         index = Index(name)
@@ -684,6 +713,88 @@ def create_indexes(names: list[str], settings: dict[str, Any] | None = None):
             raise ElasticsearchError(f"Elasticsearch error: {e.__str__()}")
 
 
+def create_data_streams(names: list[str], settings: dict[str, Any] | None = None):
+    """
+    Create Elasticsearch data streams and their index templates
+
+    For each name, puts an index template named after the stream whose
+    ``index_patterns`` is exactly that name -- no wildcard, so it never
+    applies to the date-suffixed indexes parsedmarc writes without data
+    streams -- and then creates the data stream if it does not exist yet.
+    The template composes an optional ``<name>@custom`` component template,
+    which is where an operator attaches an ILM policy or other settings:
+    parsedmarc puts its own template again in every process that writes to
+    the stream, so edits made to that template directly do not last. Each
+    name is set up once per process; ``set_hosts`` resets that.
+
+    The template maps only ``@timestamp``. Like ``create_indexes``, it
+    deliberately installs no mappings for the report fields themselves (see
+    issue #169), so they keep their dynamic ``object`` mapping.
+
+    Args:
+        names (list): A list of data stream names
+        settings (dict): Index settings for the stream's backing indices,
+            adjusted for Serverless the same way ``create_indexes`` does.
+
+    Raises:
+        ElasticsearchError: A concrete index or alias already holds one of
+            the names, or the cluster returned an error. A name that fails
+            is not remembered, so the next save retries it.
+    """
+    effective_settings = _effective_index_settings(settings)
+
+    for name in names:
+        if name in _READY_DATA_STREAMS:
+            continue
+        custom_template = f"{name}@custom"
+        try:
+            client = connections.get_connection()
+            # A missing name is a 404, but a name held by a concrete index
+            # answers 200 with an empty list (verified against
+            # Elasticsearch 8.19), so look for the stream in the response.
+            try:
+                response = client.indices.get_data_stream(name=name)
+                stream_exists = any(
+                    stream.get("name") == name
+                    for stream in response.get("data_streams", [])
+                )
+            except NotFoundError:
+                stream_exists = False
+            # Checked before anything is written, so a name collision leaves
+            # no template behind.
+            collision = not stream_exists and Index(name).exists()
+            if not collision:
+                client.indices.put_index_template(
+                    name=name,
+                    index_patterns=[name],
+                    data_stream={},
+                    composed_of=[custom_template],
+                    ignore_missing_component_templates=[custom_template],
+                    priority=_DATA_STREAM_TEMPLATE_PRIORITY,
+                    template={
+                        "settings": effective_settings,
+                        "mappings": {"properties": {"@timestamp": {"type": "date"}}},
+                    },
+                    meta={
+                        "description": (
+                            "Managed by parsedmarc, which overwrites it. Put "
+                            f"customizations in {custom_template}."
+                        )
+                    },
+                )
+                if not stream_exists:
+                    logger.debug(f"Creating Elasticsearch data stream: {name}")
+                    client.indices.create_data_stream(name=name)
+        except Exception as e:
+            raise ElasticsearchError(f"Elasticsearch error: {e.__str__()}")
+        if collision:
+            raise ElasticsearchError(
+                f"Elasticsearch error: cannot create the data stream {name}: "
+                "an index or alias with that name already exists"
+            )
+        _READY_DATA_STREAMS.add(name)
+
+
 _LEGACY_FO_FIELD = "published_policy.fo"
 # The same field split into the object/leaf names a mapping body nests it
 # under. Derived from the dotted name so the mapping written below cannot
@@ -757,6 +868,12 @@ def migrate_indexes(
     next startup, and the manual ``_update_by_query`` commands documented in
     ``docs/source/elasticsearch.md`` remain available in the meantime.
 
+    Both work unchanged with ``use_data_streams``: a data stream carries the
+    undated name itself, so the backfill patterns match it as well as any
+    dated indexes, and a stream that holds that exact name is skipped by
+    the ``published_policy.fo`` migration, because the backing index it
+    inspects maps the field as ``text``, not ``long``.
+
     Args:
         aggregate_indexes (list): A list of aggregate index names
         failure_indexes (list): A list of failure index names
@@ -922,6 +1039,7 @@ def save_aggregate_report_to_elasticsearch(
     monthly_indexes: bool | None = False,
     number_of_shards: int = 1,
     number_of_replicas: int = 0,
+    use_data_streams: bool = False,
 ):
     """
     Saves a parsed DMARC aggregate report to Elasticsearch
@@ -931,8 +1049,11 @@ def save_aggregate_report_to_elasticsearch(
         index_suffix (str): The suffix of the name of the index to save to
         index_prefix (str): The prefix of the name of the index to save to
         monthly_indexes (bool): Use monthly indexes instead of daily indexes
+            (ignored when ``use_data_streams`` is True)
         number_of_shards (int): The number of shards to use in the index
         number_of_replicas (int): The number of replicas to use in the index
+        use_data_streams (bool): Save to a data stream named like the index,
+            without the date suffix, instead of to a dated index
 
     Raises:
             AlreadySaved
@@ -959,6 +1080,8 @@ def save_aggregate_report_to_elasticsearch(
         search_index = "dmarc_aggregate*"
     if index_prefix is not None:
         search_index = f"{index_prefix}{search_index}"
+    # The trailing wildcard matches the data stream as well as the dated
+    # indexes, so duplicates are caught across a switch between the two.
     search = Search(index=search_index)
     query = org_name_query & report_id_query & domain_query
     query = query & begin_date_query & end_date_query
@@ -1005,10 +1128,6 @@ def save_aggregate_report_to_elasticsearch(
         )
         normalized_timespan = record["normalized_timespan"]
 
-        if monthly_indexes:
-            index_date = begin_date.strftime("%Y-%m")
-        else:
-            index_date = begin_date.strftime("%Y-%m-%d")
         aggregate_report["begin_date"] = begin_date
         aggregate_report["end_date"] = end_date
         date_range = [aggregate_report["begin_date"], aggregate_report["end_date"]]
@@ -1049,6 +1168,7 @@ def save_aggregate_report_to_elasticsearch(
                 "discovery_method"
             ),
             generator=metadata.get("generator"),
+            **{"@timestamp": begin_date},
         )
 
         for override in record["policy_evaluated"]["policy_override_reasons"]:
@@ -1078,15 +1198,24 @@ def save_aggregate_report_to_elasticsearch(
         if index_prefix:
             index = f"{index_prefix}{index}"
 
-        index = f"{index}-{index_date}"
         index_settings = dict(
             number_of_shards=number_of_shards, number_of_replicas=number_of_replicas
         )
-        create_indexes([index], index_settings)
+        if use_data_streams:
+            create_data_streams([index], index_settings)
+        else:
+            if monthly_indexes:
+                index_date = begin_date.strftime("%Y-%m")
+            else:
+                index_date = begin_date.strftime("%Y-%m-%d")
+            index = f"{index}-{index_date}"
+            create_indexes([index], index_settings)
         agg_doc.meta.index = index  # pyright: ignore[reportOptionalMemberAccess, reportAttributeAccessIssue]
 
         try:
-            agg_doc.save()
+            # Data streams accept only op_type=create. No _id is set, so
+            # for a dated index it means the same as the default.
+            agg_doc.save(op_type="create")
         except Exception as e:
             raise ElasticsearchError(f"Elasticsearch error: {e.__str__()}")
 
@@ -1098,6 +1227,7 @@ def save_failure_report_to_elasticsearch(
     monthly_indexes: bool | None = False,
     number_of_shards: int = 1,
     number_of_replicas: int = 0,
+    use_data_streams: bool = False,
 ):
     """
     Saves a parsed DMARC failure report to Elasticsearch
@@ -1107,10 +1237,14 @@ def save_failure_report_to_elasticsearch(
         index_suffix (str): The suffix of the name of the index to save to
         index_prefix (str): The prefix of the name of the index to save to
         monthly_indexes (bool): Use monthly indexes instead of daily
-                                indexes
+                                indexes (ignored when ``use_data_streams``
+                                is True)
         number_of_shards (int): The number of shards to use in the index
         number_of_replicas (int): The number of replicas to use in the
                                   index
+        use_data_streams (bool): Save to a data stream named like the
+                                 index, without the date suffix, instead
+                                 of to a dated index
 
     Raises:
         AlreadySaved
@@ -1145,6 +1279,8 @@ def save_failure_report_to_elasticsearch(
         search_index = ",".join(
             f"{index_prefix}{part}" for part in search_index.split(",")
         )
+    # The trailing wildcards match the data stream as well as the dated
+    # indexes, so duplicates are caught across a switch between the two.
     search = Search(index=search_index)
     q = Q(dict(match=dict(arrival_date=arrival_date_epoch_milliseconds)))  # pyright: ignore[reportArgumentType]
 
@@ -1253,6 +1389,7 @@ def save_failure_report_to_elasticsearch(
             dkim_domain=failure_report["dkim_domain"],
             original_rcpt_to=failure_report["original_rcpt_to"],
             sample=sample,
+            **{"@timestamp": arrival_date},
         )
 
         index = "dmarc_failure"
@@ -1260,18 +1397,23 @@ def save_failure_report_to_elasticsearch(
             index = f"{index}_{index_suffix}"
         if index_prefix:
             index = f"{index_prefix}{index}"
-        if monthly_indexes:
-            index_date = arrival_date.strftime("%Y-%m")
-        else:
-            index_date = arrival_date.strftime("%Y-%m-%d")
-        index = f"{index}-{index_date}"
         index_settings = dict(
             number_of_shards=number_of_shards, number_of_replicas=number_of_replicas
         )
-        create_indexes([index], index_settings)
+        if use_data_streams:
+            create_data_streams([index], index_settings)
+        else:
+            if monthly_indexes:
+                index_date = arrival_date.strftime("%Y-%m")
+            else:
+                index_date = arrival_date.strftime("%Y-%m-%d")
+            index = f"{index}-{index_date}"
+            create_indexes([index], index_settings)
         failure_doc.meta.index = index  # pyright: ignore[reportAttributeAccessIssue, reportOptionalMemberAccess]
         try:
-            failure_doc.save()
+            # Data streams accept only op_type=create. No _id is set, so
+            # for a dated index it means the same as the default.
+            failure_doc.save(op_type="create")
         except Exception as e:
             raise ElasticsearchError(f"Elasticsearch error: {e.__str__()}")
     except KeyError as e:
@@ -1287,6 +1429,7 @@ def save_smtp_tls_report_to_elasticsearch(
     monthly_indexes: bool = False,
     number_of_shards: int = 1,
     number_of_replicas: int = 0,
+    use_data_streams: bool = False,
 ):
     """
     Saves a parsed SMTP TLS report to Elasticsearch
@@ -1296,8 +1439,11 @@ def save_smtp_tls_report_to_elasticsearch(
         index_suffix (str): The suffix of the name of the index to save to
         index_prefix (str): The prefix of the name of the index to save to
         monthly_indexes (bool): Use monthly indexes instead of daily indexes
+            (ignored when ``use_data_streams`` is True)
         number_of_shards (int): The number of shards to use in the index
         number_of_replicas (int): The number of replicas to use in the index
+        use_data_streams (bool): Save to a data stream named like the index,
+            without the date suffix, instead of to a dated index
 
     Raises:
             AlreadySaved
@@ -1310,10 +1456,6 @@ def save_smtp_tls_report_to_elasticsearch(
     end_date = human_timestamp_to_datetime(report["end_date"], to_utc=True)
     begin_date_human = begin_date.strftime("%Y-%m-%d %H:%M:%SZ")
     end_date_human = end_date.strftime("%Y-%m-%d %H:%M:%SZ")
-    if monthly_indexes:
-        index_date = begin_date.strftime("%Y-%m")
-    else:
-        index_date = begin_date.strftime("%Y-%m-%d")
     report = report.copy()
     report["begin_date"] = begin_date
     report["end_date"] = end_date
@@ -1329,6 +1471,8 @@ def save_smtp_tls_report_to_elasticsearch(
         search_index = "smtp_tls*"
     if index_prefix is not None:
         search_index = f"{index_prefix}{search_index}"
+    # The trailing wildcard matches the data stream as well as the dated
+    # indexes, so duplicates are caught across a switch between the two.
     search = Search(index=search_index)
     query = org_name_query & report_id_query
     query = query & begin_date_query & end_date_query
@@ -1355,7 +1499,12 @@ def save_smtp_tls_report_to_elasticsearch(
         index = f"{index}_{index_suffix}"
     if index_prefix:
         index = f"{index_prefix}{index}"
-    index = f"{index}-{index_date}"
+    if not use_data_streams:
+        if monthly_indexes:
+            index_date = begin_date.strftime("%Y-%m")
+        else:
+            index_date = begin_date.strftime("%Y-%m-%d")
+        index = f"{index}-{index_date}"
     index_settings = dict(
         number_of_shards=number_of_shards, number_of_replicas=number_of_replicas
     )
@@ -1367,6 +1516,7 @@ def save_smtp_tls_report_to_elasticsearch(
         date_end=report["end_date"],
         contact_info=report["contact_info"],
         report_id=report["report_id"],
+        **{"@timestamp": report["begin_date"]},
     )
 
     for policy in report["policies"]:
@@ -1448,11 +1598,16 @@ def save_smtp_tls_report_to_elasticsearch(
                 )
         smtp_tls_doc.policies.append(policy_doc)
 
-    create_indexes([index], index_settings)
+    if use_data_streams:
+        create_data_streams([index], index_settings)
+    else:
+        create_indexes([index], index_settings)
     smtp_tls_doc.meta.index = index  # pyright: ignore[reportOptionalMemberAccess, reportAttributeAccessIssue]
 
     try:
-        smtp_tls_doc.save()
+        # Data streams accept only op_type=create. No _id is set, so for a
+        # dated index it means the same as the default.
+        smtp_tls_doc.save(op_type="create")
     except Exception as e:
         raise ElasticsearchError(f"Elasticsearch error: {e.__str__()}")
 

parsedmarc/opensearch.py

@@ -16,6 +16,7 @@ from opensearchpy import (
     Ip,
     Keyword,
     Nested,
+    NotFoundError,
     Object,
     OpenSearch,
     Q,
@@ -35,6 +36,23 @@ class OpenSearchError(Exception):
     """Raised when an OpenSearch error occurs"""
 
 
+# Names of the data streams whose index template and stream
+# ``create_data_streams`` has already set up in this process, so each one
+# costs its API calls once per process rather than once per saved document.
+# ``set_hosts`` clears it, so a new connection (including a SIGHUP reload,
+# which may point at another cluster or change the index settings)
+# re-applies the template. Forgetting an entry is always safe: it only
+# repeats idempotent calls.
+_READY_DATA_STREAMS: set[str] = set()
+
+# Priority of the index template ``create_data_streams`` installs. OpenSearch
+# itself installs no index templates, but plugins may; 501 stays above the
+# usual priority-100 convention and matches the Elasticsearch module. The
+# template's ``index_patterns`` is the exact stream name, so an overlap with
+# another template is unlikely in the first place.
+_DATA_STREAM_TEMPLATE_PRIORITY = 501
+
+
 # Guard query for the dkim_results_combined/spf_results_combined backfill
 # (see ``migrate_indexes``). Matches only documents that have at least one
 # DKIM or SPF auth result and are missing the corresponding combined field.
@@ -531,6 +549,7 @@ def set_hosts(
         alias. Callers that need to close this exact client later (rather than
         whatever holds the alias at that point) should hold on to it.
     """
+    _READY_DATA_STREAMS.clear()
     if not isinstance(hosts, list):
         hosts = [hosts]
     logger.debug("Connecting to OpenSearch: hosts=%s, use_ssl=%s", hosts, use_ssl)
@@ -605,6 +624,98 @@ def create_indexes(names: list[str], settings: dict[str, Any] | None = None):
             raise OpenSearchError(f"OpenSearch error: {e.__str__()}")
 
 
+def create_data_streams(names: list[str], settings: dict[str, Any] | None = None):
+    """
+    Create OpenSearch data streams and their index templates
+
+    For each name, puts an index template named after the stream whose
+    ``index_patterns`` is exactly that name -- no wildcard, so it never
+    applies to the date-suffixed indexes parsedmarc writes without data
+    streams -- and then creates the data stream if it does not exist yet.
+    parsedmarc puts its template again in every process that writes to the
+    stream, so edits made to it directly do not last; an ISM policy attaches
+    itself to the stream's backing indices through its own
+    ``ism_template`` instead. Each name is set up once per process;
+    ``set_hosts`` resets that.
+
+    The template maps only ``@timestamp``. Like ``create_indexes``, it
+    deliberately installs no mappings for the report fields themselves (see
+    issue #169), so they keep their dynamic ``object`` mapping.
+
+    Args:
+        names (list): A list of data stream names
+        settings (dict): Index settings for the stream's backing indices;
+            one shard and no replicas when ``None``, as in
+            ``create_indexes``.
+
+    Raises:
+        OpenSearchError: A concrete index or alias already holds one of the
+            names, or the cluster returned an error. A name that fails is
+            not remembered, so the next save retries it.
+    """
+    if settings is None:
+        effective_settings: dict[str, Any] = {
+            "number_of_shards": 1,
+            "number_of_replicas": 0,
+        }
+    else:
+        effective_settings = dict(settings)
+
+    for name in names:
+        if name in _READY_DATA_STREAMS:
+            continue
+        try:
+            client = connections.get_connection()
+            # OpenSearch 3 answers both a missing name and a name held by a
+            # concrete index with a 404. Elasticsearch answers the latter
+            # with 200 and an empty list, so, as in the Elasticsearch module,
+            # look for the stream in the response rather than trust the
+            # status alone.
+            try:
+                response = client.indices.get_data_stream(name=name)
+                stream_exists = any(
+                    stream.get("name") == name
+                    for stream in response.get("data_streams", [])
+                )
+            except NotFoundError:
+                stream_exists = False
+            # Checked before anything is written, so a name collision leaves
+            # no template behind.
+            collision = not stream_exists and Index(name).exists()
+            if not collision:
+                client.indices.put_index_template(
+                    name=name,
+                    body={
+                        "index_patterns": [name],
+                        "data_stream": {},
+                        "priority": _DATA_STREAM_TEMPLATE_PRIORITY,
+                        "template": {
+                            "settings": effective_settings,
+                            "mappings": {
+                                "properties": {"@timestamp": {"type": "date"}}
+                            },
+                        },
+                        "_meta": {
+                            "description": (
+                                "Managed by parsedmarc, which overwrites it. "
+                                "Attach ISM policies with an ism_template."
+                            )
+                        },
+                    },
+                )
+                if not stream_exists:
+                    logger.debug(f"Creating OpenSearch data stream: {name}")
+                    client.indices.create_data_stream(name=name)
+        except Exception as e:
+            raise OpenSearchError(f"OpenSearch error: {e.__str__()}")
+        if collision:
+            raise OpenSearchError(
+                f"OpenSearch error: cannot create the data stream {name}: "
+                "an index or alias with that name already exists"
+            )
+        _READY_DATA_STREAMS.add(name)
+
+
 _LEGACY_FO_FIELD = "published_policy.fo"
 # The same field split into the object/leaf names a mapping body nests it
 # under. Derived from the dotted name so the mapping written below cannot
@@ -678,6 +789,12 @@ def migrate_indexes(
     ``failure_details_combined`` fields (same issue #169) on SMTP TLS
     report documents, for each name in ``smtp_tls_indexes``.
 
+    All three work unchanged with ``use_data_streams``: a data stream
+    carries the undated name itself, so the backfill patterns match it as
+    well as any dated indexes, and a stream that holds that exact name is
+    skipped by the ``published_policy.fo`` migration, because the backing
+    index it inspects maps the field as ``text``, not ``long``.
+
     Args:
         aggregate_indexes (list): A list of aggregate index names
         failure_indexes (list): A list of failure index names
@@ -853,6 +970,7 @@ def save_aggregate_report_to_opensearch(
     monthly_indexes: bool = False,
     number_of_shards: int = 1,
     number_of_replicas: int = 0,
+    use_data_streams: bool = False,
 ):
     """
     Saves a parsed DMARC aggregate report to OpenSearch
@@ -862,8 +980,11 @@ def save_aggregate_report_to_opensearch(
         index_suffix (str): The suffix of the name of the index to save to
         index_prefix (str): The prefix of the name of the index to save to
         monthly_indexes (bool): Use monthly indexes instead of daily indexes
+            (ignored when ``use_data_streams`` is True)
         number_of_shards (int): The number of shards to use in the index
         number_of_replicas (int): The number of replicas to use in the index
+        use_data_streams (bool): Save to a data stream named like the index,
+            without the date suffix, instead of to a dated index
 
     Raises:
             AlreadySaved
@@ -890,6 +1011,8 @@ def save_aggregate_report_to_opensearch(
         search_index = "dmarc_aggregate*"
     if index_prefix is not None:
         search_index = f"{index_prefix}{search_index}"
+    # The trailing wildcard matches the data stream as well as the dated
+    # indexes, so duplicates are caught across a switch between the two.
     search = Search(index=search_index)
     query = org_name_query & report_id_query & domain_query
     query = query & begin_date_query & end_date_query
@@ -933,10 +1056,6 @@ def save_aggregate_report_to_opensearch(
         )
         normalized_timespan = record["normalized_timespan"]
 
-        if monthly_indexes:
-            index_date = begin_date.strftime("%Y-%m")
-        else:
-            index_date = begin_date.strftime("%Y-%m-%d")
         aggregate_report["begin_date"] = begin_date
         aggregate_report["end_date"] = end_date
         date_range = [aggregate_report["begin_date"], aggregate_report["end_date"]]
@@ -977,6 +1096,7 @@ def save_aggregate_report_to_opensearch(
                 "discovery_method"
             ),
             generator=metadata.get("generator"),
+            **{"@timestamp": begin_date},
         )
 
         for override in record["policy_evaluated"]["policy_override_reasons"]:
@@ -1006,15 +1126,24 @@ def save_aggregate_report_to_opensearch(
         if index_prefix:
             index = f"{index_prefix}{index}"
 
-        index = f"{index}-{index_date}"
         index_settings = dict(
             number_of_shards=number_of_shards, number_of_replicas=number_of_replicas
         )
-        create_indexes([index], index_settings)
+        if use_data_streams:
+            create_data_streams([index], index_settings)
+        else:
+            if monthly_indexes:
+                index_date = begin_date.strftime("%Y-%m")
+            else:
+                index_date = begin_date.strftime("%Y-%m-%d")
+            index = f"{index}-{index_date}"
+            create_indexes([index], index_settings)
         agg_doc.meta.index = index
 
         try:
-            agg_doc.save()
+            # Data streams accept only op_type=create. No _id is set, so
+            # for a dated index it means the same as the default.
+            agg_doc.save(op_type="create")
         except Exception as e:
             raise OpenSearchError(f"OpenSearch error: {e.__str__()}")
 
@@ -1026,6 +1155,7 @@ def save_failure_report_to_opensearch(
     monthly_indexes: bool = False,
     number_of_shards: int = 1,
     number_of_replicas: int = 0,
+    use_data_streams: bool = False,
 ):
     """
     Saves a parsed DMARC failure report to OpenSearch
@@ -1035,10 +1165,14 @@ def save_failure_report_to_opensearch(
         index_suffix (str): The suffix of the name of the index to save to
         index_prefix (str): The prefix of the name of the index to save to
         monthly_indexes (bool): Use monthly indexes instead of daily
-                                indexes
+                                indexes (ignored when ``use_data_streams``
+                                is True)
         number_of_shards (int): The number of shards to use in the index
         number_of_replicas (int): The number of replicas to use in the
                                   index
+        use_data_streams (bool): Save to a data stream named like the
+                                 index, without the date suffix, instead
+                                 of to a dated index
 
     Raises:
         AlreadySaved
@@ -1073,6 +1207,8 @@ def save_failure_report_to_opensearch(
         search_index = ",".join(
             f"{index_prefix}{part}" for part in search_index.split(",")
         )
+    # The trailing wildcards match the data stream as well as the dated
+    # indexes, so duplicates are caught across a switch between the two.
     search = Search(index=search_index)
     q = Q(dict(match=dict(arrival_date=arrival_date_epoch_milliseconds)))
 
@@ -1179,6 +1315,7 @@ def save_failure_report_to_opensearch(
             dkim_domain=failure_report["dkim_domain"],
             original_rcpt_to=failure_report["original_rcpt_to"],
             sample=sample,
+            **{"@timestamp": arrival_date},
         )
 
         index = "dmarc_failure"
@@ -1186,18 +1323,23 @@ def save_failure_report_to_opensearch(
             index = f"{index}_{index_suffix}"
         if index_prefix:
             index = f"{index_prefix}{index}"
-        if monthly_indexes:
-            index_date = arrival_date.strftime("%Y-%m")
-        else:
-            index_date = arrival_date.strftime("%Y-%m-%d")
-        index = f"{index}-{index_date}"
         index_settings = dict(
             number_of_shards=number_of_shards, number_of_replicas=number_of_replicas
         )
-        create_indexes([index], index_settings)
+        if use_data_streams:
+            create_data_streams([index], index_settings)
+        else:
+            if monthly_indexes:
+                index_date = arrival_date.strftime("%Y-%m")
+            else:
+                index_date = arrival_date.strftime("%Y-%m-%d")
+            index = f"{index}-{index_date}"
+            create_indexes([index], index_settings)
         failure_doc.meta.index = index
         try:
-            failure_doc.save()
+            # Data streams accept only op_type=create. No _id is set, so
+            # for a dated index it means the same as the default.
+            failure_doc.save(op_type="create")
         except Exception as e:
             raise OpenSearchError(f"OpenSearch error: {e.__str__()}")
     except KeyError as e:
@@ -1213,6 +1355,7 @@ def save_smtp_tls_report_to_opensearch(
     monthly_indexes: bool = False,
     number_of_shards: int = 1,
     number_of_replicas: int = 0,
+    use_data_streams: bool = False,
 ):
     """
     Saves a parsed SMTP TLS report to OpenSearch
@@ -1222,8 +1365,11 @@ def save_smtp_tls_report_to_opensearch(
         index_suffix (str): The suffix of the name of the index to save to
         index_prefix (str): The prefix of the name of the index to save to
         monthly_indexes (bool): Use monthly indexes instead of daily indexes
+            (ignored when ``use_data_streams`` is True)
         number_of_shards (int): The number of shards to use in the index
         number_of_replicas (int): The number of replicas to use in the index
+        use_data_streams (bool): Save to a data stream named like the index,
+            without the date suffix, instead of to a dated index
 
     Raises:
             AlreadySaved
@@ -1236,10 +1382,6 @@ def save_smtp_tls_report_to_opensearch(
     end_date = human_timestamp_to_datetime(report["end_date"], to_utc=True)
     begin_date_human = begin_date.strftime("%Y-%m-%d %H:%M:%SZ")
     end_date_human = end_date.strftime("%Y-%m-%d %H:%M:%SZ")
-    if monthly_indexes:
-        index_date = begin_date.strftime("%Y-%m")
-    else:
-        index_date = begin_date.strftime("%Y-%m-%d")
     report = report.copy()
     report["begin_date"] = begin_date
     report["end_date"] = end_date
@@ -1255,6 +1397,8 @@ def save_smtp_tls_report_to_opensearch(
         search_index = "smtp_tls*"
     if index_prefix is not None:
         search_index = f"{index_prefix}{search_index}"
+    # The trailing wildcard matches the data stream as well as the dated
+    # indexes, so duplicates are caught across a switch between the two.
     search = Search(index=search_index)
     query = org_name_query & report_id_query
     query = query & begin_date_query & end_date_query
@@ -1281,7 +1425,12 @@ def save_smtp_tls_report_to_opensearch(
         index = f"{index}_{index_suffix}"
     if index_prefix:
         index = f"{index_prefix}{index}"
-    index = f"{index}-{index_date}"
+    if not use_data_streams:
+        if monthly_indexes:
+            index_date = begin_date.strftime("%Y-%m")
+        else:
+            index_date = begin_date.strftime("%Y-%m-%d")
+        index = f"{index}-{index_date}"
     index_settings = dict(
         number_of_shards=number_of_shards, number_of_replicas=number_of_replicas
     )
@@ -1293,6 +1442,7 @@ def save_smtp_tls_report_to_opensearch(
         date_end=report["end_date"],
         contact_info=report["contact_info"],
         report_id=report["report_id"],
+        **{"@timestamp": report["begin_date"]},
     )
 
     for policy in report["policies"]:
@@ -1374,11 +1524,16 @@ def save_smtp_tls_report_to_opensearch(
                 )
         smtp_tls_doc.policies.append(policy_doc)
 
-    create_indexes([index], index_settings)
+    if use_data_streams:
+        create_data_streams([index], index_settings)
+    else:
+        create_indexes([index], index_settings)
     smtp_tls_doc.meta.index = index
 
     try:
-        smtp_tls_doc.save()
+        # Data streams accept only op_type=create. No _id is set, so for a
+        # dated index it means the same as the default.
+        smtp_tls_doc.save(op_type="create")
     except Exception as e:
         raise OpenSearchError(f"OpenSearch error: {e.__str__()}")
 

tests/test_cli.py

@@ -7959,6 +7959,175 @@ class TestParseConfigOpenSearch(unittest.TestCase):
             _parse_config(cp, _opts())
 
 
+class TestParseConfigUseDataStreams(unittest.TestCase):
+    """``use_data_streams`` in [elasticsearch] and [opensearch] (issue
+    #842), and the warning that ``monthly_indexes`` is ignored with it."""
+
+    SECTIONS = ("elasticsearch", "opensearch")
+
+    def test_option_parsed_for_each_section(self):
+        from parsedmarc.cli import _parse_config
+
+        for section in self.SECTIONS:
+            for raw, expected in (("true", True), ("false", False)):
+                with self.subTest(section=section, value=raw):
+                    cp = _config_with(
+                        section, {"hosts": "h:9200", "use_data_streams": raw}
+                    )
+                    opts = _opts()
+                    _parse_config(cp, opts)
+                    self.assertIs(
+                        getattr(opts, f"{section}_use_data_streams"), expected
+                    )
+
+    def test_option_absent_leaves_attribute_unset(self):
+        """The default comes from the CLI's Namespace defaults, which the
+        real CLI builds before _parse_config runs."""
+        from parsedmarc.cli import _parse_config
+
+        for section in self.SECTIONS:
+            with self.subTest(section=section):
+                cp = _config_with(section, {"hosts": "h:9200"})
+                opts = _opts()
+                _parse_config(cp, opts)
+                self.assertFalse(hasattr(opts, f"{section}_use_data_streams"))
+
+    def test_monthly_indexes_with_data_streams_warns(self):
+        from parsedmarc.cli import _parse_config
+
+        for section in self.SECTIONS:
+            with self.subTest(section=section):
+                cp = _config_with(
+                    section,
+                    {
+                        "hosts": "h:9200",
+                        "monthly_indexes": "true",
+                        "use_data_streams": "true",
+                    },
+                )
+                opts = _opts()
+                with self.assertLogs("parsedmarc.log", level="WARNING") as cm:
+                    _parse_config(cp, opts)
+                self.assertEqual(len(cm.records), 1)
+                self.assertIn(f"[{section}] monthly_indexes", cm.output[0])
+                self.assertIn("use_data_streams", cm.output[0])
+                # Still parsed; the save functions ignore it in stream mode.
+                self.assertTrue(getattr(opts, f"{section}_monthly_indexes"))
+
+    def test_no_warning_unless_both_are_enabled(self):
+        from parsedmarc.cli import _parse_config
+
+        combinations = (
+            {"monthly_indexes": "true"},
+            {"use_data_streams": "true"},
+            {"monthly_indexes": "true", "use_data_streams": "false"},
+            {"monthly_indexes": "false", "use_data_streams": "true"},
+        )
+        for section in self.SECTIONS:
+            for settings in combinations:
+                with self.subTest(section=section, settings=settings):
+                    cp = _config_with(section, {"hosts": "h:9200", **settings})
+                    with self.assertNoLogs("parsedmarc.log", level="WARNING"):
+                        _parse_config(cp, _opts())
+
+
+class TestUseDataStreamsPassthrough(unittest.TestCase):
+    """Each backend's parsed option reaches that backend's three save
+    calls, and only those: enabling it in one section leaves the other
+    backend's saves False, and it is False when absent from both."""
+
+    def _run(self, es_extra: str = "", os_extra: str = "") -> dict[str, MagicMock]:
+        names = (
+            "elastic.save_aggregate_report_to_elasticsearch",
+            "elastic.save_failure_report_to_elasticsearch",
+            "elastic.save_smtp_tls_report_to_elasticsearch",
+            "opensearch.save_aggregate_report_to_opensearch",
+            "opensearch.save_failure_report_to_opensearch",
+            "opensearch.save_smtp_tls_report_to_opensearch",
+        )
+        config = f"""[general]
+save_aggregate = true
+save_failure = true
+save_smtp_tls = true
+silent = true
+
+[imap]
+host = imap.example.com
+user = test-user
+password = test-password
+
+[elasticsearch]
+hosts = localhost
+{es_extra}
+
+[opensearch]
+hosts = localhost
+{os_extra}
+"""
+        with tempfile.NamedTemporaryFile(
+            "w", suffix=".ini", delete=False
+        ) as config_file:
+            config_file.write(config)
+            config_path = config_file.name
+        self.addCleanup(lambda: os.path.exists(config_path) and os.remove(config_path))
+
+        saves = {name: MagicMock() for name in names}
+        patches = [patch(f"parsedmarc.cli.{name}", saves[name]) for name in names]
+        patches += [
+            patch("parsedmarc.cli.elastic.set_hosts"),
+            patch("parsedmarc.cli.elastic.migrate_indexes"),
+            patch("parsedmarc.cli.opensearch.set_hosts"),
+            patch("parsedmarc.cli.opensearch.migrate_indexes"),
+            patch("parsedmarc.cli.IMAPConnection", return_value=object()),
+            patch(
+                "parsedmarc.cli.get_dmarc_reports_from_mailbox",
+                side_effect=_fetch_invoking_save_callback(
+                    {
+                        "aggregate_reports": [
+                            {"policy_published": {"domain": "example.com"}}
+                        ],
+                        "failure_reports": [{"reported_domain": "example.com"}],
+                        "smtp_tls_reports": [
+                            {"policies": [{"policy_domain": "example.com"}]}
+                        ],
+                    }
+                ),
+            ),
+            patch.object(sys, "argv", ["parsedmarc", "-c", config_path]),
+        ]
+        for p in patches:
+            p.start()
+            self.addCleanup(p.stop)
+        parsedmarc.cli._main()
+        return saves
+
+    def _assert_flags(
+        self, saves: dict[str, MagicMock], elastic: bool, opensearch: bool
+    ) -> None:
+        for name, save in saves.items():
+            expected = elastic if name.startswith("elastic.") else opensearch
+            with self.subTest(save=name):
+                save.assert_called_once()
+                self.assertIs(save.call_args.kwargs["use_data_streams"], expected)
+
+    def test_enabled_in_both_sections_reaches_every_save_call(self):
+        enabled = "use_data_streams = true"
+        saves = self._run(es_extra=enabled, os_extra=enabled)
+        self._assert_flags(saves, elastic=True, opensearch=True)
+
+    def test_elasticsearch_section_only_reaches_elasticsearch_saves(self):
+        saves = self._run(es_extra="use_data_streams = true")
+        self._assert_flags(saves, elastic=True, opensearch=False)
+
+    def test_opensearch_section_only_reaches_opensearch_saves(self):
+        saves = self._run(os_extra="use_data_streams = true")
+        self._assert_flags(saves, elastic=False, opensearch=True)
+
+    def test_absent_defaults_to_false(self):
+        saves = self._run()
+        self._assert_flags(saves, elastic=False, opensearch=False)
+
+
 class TestParseConfigSplunkHec(unittest.TestCase):
     def test_splunk_hec_complete(self):
         from parsedmarc.cli import _parse_config
@@ -9487,8 +9656,9 @@ class TestParseConfigLoki(unittest.TestCase):
 
 class TestLokiCliWiring(unittest.TestCase):
     """End-to-end: a [loki] config reaches LokiClient, every report type
-    reaches its save method, and a failed push counts as an output error
-    (so mailbox messages are held back instead of archived)."""
+    reaches its save method, a failed push counts as an output error
+    (so mailbox messages are held back instead of archived), and a config
+    LokiClient rejects fails client initialization under the Loki name."""
 
     def _run_main(self, reports, extra_general="", save_side_effect=None):
         config = f"""[general]
@@ -9581,6 +9751,35 @@ tenant_id = fosdem
             )
         self.assertEqual(ctx.exception.code, 1)
 
+    def test_rejected_client_config_names_the_loki_section(self):
+        """A [loki] config that parses but that LokiClient rejects is
+        reported as a ``Loki:`` RuntimeError.
+
+        _parse_config accepts ``token`` alongside ``username``/``password``;
+        only the real LokiClient constructor enforces that they are
+        exclusive, so this is the path a misconfigured section takes at
+        startup and on a SIGHUP reload.
+        """
+        config = _config_with(
+            "loki",
+            {"url": "http://loki.example.com:3100", "token": "t", "username": "u"},
+        )
+        # The non-None Namespace defaults _main() gives the loki_* options.
+        opts = _output_client_opts(
+            loki_timestamp="ingest",
+            loki_skip_certificate_verification=False,
+            loki_timeout=60,
+            loki_batch_size=1000,
+        )
+        parsedmarc.cli._parse_config(config, opts)
+        with self.assertRaises(RuntimeError) as context:
+            parsedmarc.cli._init_output_clients(opts)
+        self.assertEqual(
+            str(context.exception),
+            "Loki: Loki token and username/password are exclusive",
+        )
+        self.assertIsInstance(context.exception.__cause__, ValueError)
+
 
 if __name__ == "__main__":
     unittest.main(verbosity=2)

tests/test_elastic.py

@@ -1,21 +1,29 @@
 """Tests for parsedmarc.elastic
 
 Mocks at the elasticsearch.dsl SDK boundary (connections.create_connection,
-Index, Search, Document.save) so the tests verify the parsedmarc-side
-transformation logic — document construction, index naming, deduplication
-queries, error wrapping — without needing a running Elasticsearch cluster.
+Index, Search, Document.save, or a mock client registered as the
+``default`` connection so the real Document.save runs) so the tests
+verify the parsedmarc-side transformation logic — document
+construction, index naming, deduplication queries, error wrapping —
+without needing a running Elasticsearch cluster.
 """
 
 import time
 import unittest
+from collections.abc import Iterator
+from contextlib import contextmanager
+from datetime import datetime, timezone
 from typing import cast
 from unittest.mock import MagicMock, call, patch
 
+from elasticsearch import NotFoundError
+
 import parsedmarc.elastic as elastic_module
 from parsedmarc import InvalidFailureReport, parse_report_file
 from parsedmarc.elastic import (
     AlreadySaved,
     ElasticsearchError,
+    create_data_streams,
     create_indexes,
     migrate_indexes,
     save_aggregate_report_to_elasticsearch,
@@ -1528,6 +1536,418 @@ class TestSaveSmtpTlsReport(unittest.TestCase):
         self.assertEqual(failure_detail.failure_reason_code, "no-policy-served")
 
 
+# ---------------------------------------------------------------------------
+# Data streams (issue #842)
+# ---------------------------------------------------------------------------
+
+
+def _not_found() -> NotFoundError:
+    """The exception elasticsearch-py raises for an HTTP 404."""
+    return NotFoundError(
+        "index_not_found_exception", meta=MagicMock(status=404), body={}
+    )
+
+
+def _stream_listing(*names: str) -> dict:
+    """A GET _data_stream/<name> response body listing *names*."""
+    return {"data_streams": [{"name": name} for name in names]}
+
+
+@contextmanager
+def _registered_client() -> Iterator[MagicMock]:
+    """Register a mock client as the ``default`` connection.
+
+    Both ``create_data_streams`` and ``Document.save`` resolve the client
+    through the connection registry, so registering the mock there (rather
+    than patching either function) lets the real save path run and the test
+    read what reached the transport. The previous registration is put back
+    afterwards.
+    """
+    client = MagicMock()
+    client.index.return_value = {"_index": "x", "_id": "1", "result": "created"}
+    try:
+        previous = elastic_module.connections.get_connection("default")
+    except KeyError:
+        previous = None
+    elastic_module.connections.add_connection("default", client)
+    try:
+        yield client
+    finally:
+        if previous is None:
+            elastic_module.connections.remove_connection("default")
+        else:
+            elastic_module.connections.add_connection("default", previous)
+
+
+class _DataStreamTestCase(unittest.TestCase):
+    """Isolates each test from the module's data stream memo and the
+    Serverless flag."""
+
+    def setUp(self):
+        self._serverless = elastic_module._SERVERLESS
+        elastic_module._SERVERLESS = False
+        elastic_module._READY_DATA_STREAMS.clear()
+
+    def tearDown(self):
+        elastic_module._SERVERLESS = self._serverless
+        elastic_module._READY_DATA_STREAMS.clear()
+
+
+class TestCreateDataStreams(_DataStreamTestCase):
+    def test_creates_template_and_stream_when_missing(self):
+        with (
+            _registered_client() as client,
+            patch("parsedmarc.elastic.Index") as mock_index_cls,
+        ):
+            client.indices.get_data_stream.side_effect = _not_found()
+            mock_index_cls.return_value.exists.return_value = False
+            create_data_streams(
+                ["dmarc_aggregate"],
+                {"number_of_shards": 2, "number_of_replicas": 1},
+            )
+        template = client.indices.put_index_template.call_args.kwargs
+        self.assertEqual(template["name"], "dmarc_aggregate")
+        # Exact name, no wildcard: the template must never claim the
+        # date-suffixed indexes written without data streams.
+        self.assertEqual(template["index_patterns"], ["dmarc_aggregate"])
+        self.assertEqual(template["data_stream"], {})
+        self.assertEqual(template["priority"], 501)
+        self.assertEqual(template["composed_of"], ["dmarc_aggregate@custom"])
+        self.assertEqual(
+            template["ignore_missing_component_templates"],
+            ["dmarc_aggregate@custom"],
+        )
+        self.assertEqual(
+            template["template"],
+            {
+                "settings": {"number_of_shards": 2, "number_of_replicas": 1},
+                "mappings": {"properties": {"@timestamp": {"type": "date"}}},
+            },
+        )
+        self.assertIn("dmarc_aggregate@custom", template["meta"]["description"])
+        client.indices.create_data_stream.assert_called_once_with(
+            name="dmarc_aggregate"
+        )
+        # No concrete index is created for a stream.
+        mock_index_cls.return_value.create.assert_not_called()
+        self.assertIn("dmarc_aggregate", elastic_module._READY_DATA_STREAMS)
+
+    def test_default_settings_are_one_shard_no_replicas(self):
+        with (
+            _registered_client() as client,
+            patch("parsedmarc.elastic.Index") as mock_index_cls,
+        ):
+            client.indices.get_data_stream.side_effect = _not_found()
+            mock_index_cls.return_value.exists.return_value = False
+            create_data_streams(["smtp_tls"])
+        settings = client.indices.put_index_template.call_args.kwargs["template"][
+            "settings"
+        ]
+        self.assertEqual(settings, {"number_of_shards": 1, "number_of_replicas": 0})
+
+    def test_existing_stream_gets_template_but_is_not_recreated(self):
+        """The template is put again so that settings changes reach the
+        next backing index; the stream itself already exists."""
+        with (
+            _registered_client() as client,
+            patch("parsedmarc.elastic.Index") as mock_index_cls,
+        ):
+            client.indices.get_data_stream.return_value = _stream_listing(
+                "dmarc_failure"
+            )
+            create_data_streams(["dmarc_failure"])
+        client.indices.put_index_template.assert_called_once()
+        client.indices.create_data_stream.assert_not_called()
+        mock_index_cls.return_value.exists.assert_not_called()
+        self.assertIn("dmarc_failure", elastic_module._READY_DATA_STREAMS)
+
+    def test_concrete_index_holding_the_name_raises_without_writing(self):
+        """Elasticsearch answers GET _data_stream/<name> for a name held by
+        a concrete index with 200 and an empty list, not a 404 (verified
+        against 8.19). Treating that 200 as "stream exists" would send
+        every document into the old undated index, such as one left by a
+        pre-5.0.0 release."""
+        with (
+            _registered_client() as client,
+            patch("parsedmarc.elastic.Index") as mock_index_cls,
+        ):
+            client.indices.get_data_stream.return_value = _stream_listing()
+            mock_index_cls.return_value.exists.return_value = True
+            with self.assertRaises(ElasticsearchError) as ctx:
+                create_data_streams(["dmarc_aggregate"])
+        mock_index_cls.assert_called_with("dmarc_aggregate")
+        self.assertIn("already exists", str(ctx.exception))
+        self.assertIn("dmarc_aggregate", str(ctx.exception))
+        client.indices.put_index_template.assert_not_called()
+        client.indices.create_data_stream.assert_not_called()
+        self.assertNotIn("dmarc_aggregate", elastic_module._READY_DATA_STREAMS)
+
+    def test_alias_holding_the_name_raises(self):
+        """A missing stream is a 404; an alias with the name still exists
+        as far as Index.exists() is concerned."""
+        with (
+            _registered_client() as client,
+            patch("parsedmarc.elastic.Index") as mock_index_cls,
+        ):
+            client.indices.get_data_stream.side_effect = _not_found()
+            mock_index_cls.return_value.exists.return_value = True
+            with self.assertRaises(ElasticsearchError):
+                create_data_streams(["smtp_tls"])
+        client.indices.put_index_template.assert_not_called()
+        client.indices.create_data_stream.assert_not_called()
+
+    def test_ready_stream_is_not_set_up_again(self):
+        with (
+            _registered_client() as client,
+            patch("parsedmarc.elastic.Index") as mock_index_cls,
+        ):
+            client.indices.get_data_stream.side_effect = _not_found()
+            mock_index_cls.return_value.exists.return_value = False
+            create_data_streams(["dmarc_aggregate"])
+            create_data_streams(["dmarc_aggregate"])
+        self.assertEqual(client.indices.get_data_stream.call_count, 1)
+        self.assertEqual(client.indices.put_index_template.call_count, 1)
+        self.assertEqual(client.indices.create_data_stream.call_count, 1)
+
+    def test_set_hosts_forgets_ready_streams(self):
+        """A new connection may be a different cluster or carry new index
+        settings, so the template has to be put again."""
+        elastic_module._READY_DATA_STREAMS.add("dmarc_aggregate")
+        with patch("parsedmarc.elastic.connections.create_connection"):
+            set_hosts("es:9200")
+        self.assertEqual(elastic_module._READY_DATA_STREAMS, set())
+
+    def test_sdk_error_is_wrapped_and_retried_next_time(self):
+        with (
+            _registered_client() as client,
+            patch("parsedmarc.elastic.Index") as mock_index_cls,
+        ):
+            client.indices.get_data_stream.side_effect = _not_found()
+            mock_index_cls.return_value.exists.return_value = False
+            client.indices.put_index_template.side_effect = RuntimeError("cluster down")
+            with self.assertRaises(ElasticsearchError) as ctx:
+                create_data_streams(["dmarc_aggregate"])
+            self.assertIn("cluster down", str(ctx.exception))
+            self.assertNotIn("dmarc_aggregate", elastic_module._READY_DATA_STREAMS)
+            client.indices.create_data_stream.assert_not_called()
+
+            client.indices.put_index_template.side_effect = None
+            create_data_streams(["dmarc_aggregate"])
+        client.indices.create_data_stream.assert_called_once_with(
+            name="dmarc_aggregate"
+        )
+        self.assertIn("dmarc_aggregate", elastic_module._READY_DATA_STREAMS)
+
+    def test_serverless_strips_rejected_settings_from_template(self):
+        elastic_module._SERVERLESS = True
+        with (
+            _registered_client() as client,
+            patch("parsedmarc.elastic.Index") as mock_index_cls,
+        ):
+            client.indices.get_data_stream.side_effect = _not_found()
+            mock_index_cls.return_value.exists.return_value = False
+            create_data_streams(
+                ["dmarc_aggregate"],
+                {
+                    "number_of_shards": 1,
+                    "number_of_replicas": 0,
+                    "refresh_interval": "5s",
+                },
+            )
+        settings = client.indices.put_index_template.call_args.kwargs["template"][
+            "settings"
+        ]
+        self.assertEqual(settings, {"refresh_interval": "5s"})
+
+
+class TestSaveToDataStreams(_DataStreamTestCase):
+    """Each save function, run through the real Document.save() into a
+    mock client: in stream mode the document goes to the undated name
+    with op_type=create and an @timestamp; in the default mode it goes to
+    the dated index as before, with the same op_type and @timestamp."""
+
+    def _save(self, save_function, report, **kwargs):
+        with (
+            _registered_client() as client,
+            patch("parsedmarc.elastic.Search") as mock_search_cls,
+            patch("parsedmarc.elastic.Index") as mock_index_cls,
+        ):
+            mock_search_cls.return_value.execute.return_value = []
+            client.indices.get_data_stream.side_effect = _not_found()
+            mock_index_cls.return_value.exists.return_value = False
+            save_function(report, **kwargs)
+        return client, mock_search_cls, mock_index_cls
+
+    def test_aggregate_report_goes_to_the_stream(self):
+        client, mock_search_cls, mock_index_cls = self._save(
+            save_aggregate_report_to_elasticsearch,
+            _aggregate_report(),
+            index_prefix="cust_",
+            index_suffix="tenant_a",
+            # Ignored in stream mode.
+            monthly_indexes=True,
+            use_data_streams=True,
+        )
+        index_call = client.index.call_args.kwargs
+        self.assertEqual(index_call["index"], "cust_dmarc_aggregate_tenant_a")
+        self.assertEqual(index_call["op_type"], "create")
+        self.assertEqual(
+            index_call["body"]["@timestamp"],
+            datetime(2024, 1, 15, tzinfo=timezone.utc),
+        )
+        self.assertEqual(index_call["body"]["report_id"], "agg-1")
+        self.assertEqual(
+            client.indices.put_index_template.call_args.kwargs["index_patterns"],
+            ["cust_dmarc_aggregate_tenant_a"],
+        )
+        client.indices.create_data_stream.assert_called_once_with(
+            name="cust_dmarc_aggregate_tenant_a"
+        )
+        mock_index_cls.return_value.create.assert_not_called()
+        # The dedup search still uses the wildcard, which matches the
+        # stream as well as dated indexes written before the switch.
+        self.assertEqual(
+            mock_search_cls.call_args.kwargs["index"],
+            "cust_dmarc_aggregate_tenant_a*",
+        )
+
+    def test_aggregate_report_dated_mode_unchanged_apart_from_timestamp_and_op_type(
+        self,
+    ):
+        client, _, mock_index_cls = self._save(
+            save_aggregate_report_to_elasticsearch, _aggregate_report()
+        )
+        index_call = client.index.call_args.kwargs
+        self.assertEqual(index_call["index"], "dmarc_aggregate-2024-01-15")
+        self.assertEqual(index_call["op_type"], "create")
+        self.assertEqual(
+            index_call["body"]["@timestamp"],
+            datetime(2024, 1, 15, tzinfo=timezone.utc),
+        )
+        mock_index_cls.return_value.create.assert_called_once()
+        client.indices.put_index_template.assert_not_called()
+        client.indices.create_data_stream.assert_not_called()
+
+    def test_failure_report_goes_to_the_stream(self):
+        client, mock_search_cls, mock_index_cls = self._save(
+            save_failure_report_to_elasticsearch,
+            _failure_report(),
+            monthly_indexes=True,
+            use_data_streams=True,
+        )
+        index_call = client.index.call_args.kwargs
+        self.assertEqual(index_call["index"], "dmarc_failure")
+        self.assertEqual(index_call["op_type"], "create")
+        self.assertEqual(
+            index_call["body"]["@timestamp"],
+            datetime(2024, 1, 1, tzinfo=timezone.utc),
+        )
+        client.indices.create_data_stream.assert_called_once_with(name="dmarc_failure")
+        mock_index_cls.return_value.create.assert_not_called()
+        self.assertEqual(
+            mock_search_cls.call_args.kwargs["index"],
+            "dmarc_failure*,dmarc_forensic*",
+        )
+
+    def test_failure_report_dated_mode_unchanged_apart_from_timestamp_and_op_type(self):
+        client, _, _ = self._save(
+            save_failure_report_to_elasticsearch, _failure_report()
+        )
+        index_call = client.index.call_args.kwargs
+        self.assertEqual(index_call["index"], "dmarc_failure-2024-01-01")
+        self.assertEqual(index_call["op_type"], "create")
+        self.assertEqual(
+            index_call["body"]["@timestamp"],
+            datetime(2024, 1, 1, tzinfo=timezone.utc),
+        )
+        client.indices.create_data_stream.assert_not_called()
+
+    def test_smtp_tls_report_goes_to_the_stream(self):
+        client, mock_search_cls, mock_index_cls = self._save(
+            save_smtp_tls_report_to_elasticsearch,
+            _smtp_tls_report(),
+            index_prefix="cust_",
+            use_data_streams=True,
+        )
+        index_call = client.index.call_args.kwargs
+        self.assertEqual(index_call["index"], "cust_smtp_tls")
+        self.assertEqual(index_call["op_type"], "create")
+        self.assertEqual(
+            index_call["body"]["@timestamp"],
+            datetime(2024, 2, 3, tzinfo=timezone.utc),
+        )
+        client.indices.create_data_stream.assert_called_once_with(name="cust_smtp_tls")
+        mock_index_cls.return_value.create.assert_not_called()
+        self.assertEqual(mock_search_cls.call_args.kwargs["index"], "cust_smtp_tls*")
+
+    def test_smtp_tls_report_dated_mode_unchanged_apart_from_timestamp_and_op_type(
+        self,
+    ):
+        client, _, _ = self._save(
+            save_smtp_tls_report_to_elasticsearch, _smtp_tls_report()
+        )
+        index_call = client.index.call_args.kwargs
+        self.assertEqual(index_call["index"], "smtp_tls-2024-02-03")
+        self.assertEqual(index_call["op_type"], "create")
+        self.assertEqual(
+            index_call["body"]["@timestamp"],
+            datetime(2024, 2, 3, tzinfo=timezone.utc),
+        )
+        client.indices.create_data_stream.assert_not_called()
+
+    def test_stream_set_up_once_for_a_multi_record_report(self):
+        report = _aggregate_report()
+        report["records"].append(report["records"][0].copy())
+        client, _, _ = self._save(
+            save_aggregate_report_to_elasticsearch, report, use_data_streams=True
+        )
+        self.assertEqual(client.index.call_count, 2)
+        self.assertEqual(client.indices.put_index_template.call_count, 1)
+
+    def test_name_collision_saves_nothing(self):
+        with (
+            _registered_client() as client,
+            patch("parsedmarc.elastic.Search") as mock_search_cls,
+            patch("parsedmarc.elastic.Index") as mock_index_cls,
+        ):
+            mock_search_cls.return_value.execute.return_value = []
+            client.indices.get_data_stream.side_effect = _not_found()
+            mock_index_cls.return_value.exists.return_value = True
+            with self.assertRaises(ElasticsearchError):
+                save_smtp_tls_report_to_elasticsearch(
+                    _smtp_tls_report(), use_data_streams=True
+                )
+        client.index.assert_not_called()
+
+
+class TestMigrateIndexesWithDataStream(unittest.TestCase):
+    def test_legacy_fo_migration_skips_a_stream_with_the_name(self):
+        """With use_data_streams the undated legacy name is the stream
+        itself. Index.exists() is true for it, and the field mapping comes
+        back keyed by a backing index; its fo is text, so nothing is
+        migrated or deleted."""
+        with (
+            patch("parsedmarc.elastic.Index") as mock_index_cls,
+            patch("parsedmarc.elastic.reindex") as mock_reindex,
+        ):
+            index = mock_index_cls.return_value
+            index.exists.return_value = True
+            index.get_field_mapping.return_value = {
+                ".ds-dmarc_aggregate-2026.10.31-000001": {
+                    "mappings": {
+                        "published_policy.fo": {
+                            "full_name": "published_policy.fo",
+                            "mapping": {"fo": {"type": "text"}},
+                        }
+                    }
+                }
+            }
+            migrate_indexes(legacy_fo_indexes=["dmarc_aggregate"])
+        mock_reindex.assert_not_called()
+        index.create.assert_not_called()
+        index.delete.assert_not_called()
+
+
 class TestBackwardCompatAlias(unittest.TestCase):
     def test_save_forensic_alias_points_to_save_failure(self):
         self.assertIs(

tests/test_opensearch.py

@@ -1,21 +1,29 @@
 """Tests for parsedmarc.opensearch
 
 Mocks at the opensearch-dsl SDK boundary (connections.create_connection,
-Index, Search, Document.save) so the tests verify the parsedmarc-side
-transformation logic — document construction, index naming, deduplication
-queries, error wrapping — without needing a running OpenSearch cluster.
+Index, Search, Document.save, or a mock client registered as the
+``default`` connection so the real Document.save runs) so the tests
+verify the parsedmarc-side transformation logic — document
+construction, index naming, deduplication queries, error wrapping —
+without needing a running OpenSearch cluster.
 """
 
 import time
 import unittest
+from collections.abc import Iterator
+from contextlib import contextmanager
+from datetime import datetime, timezone
 from typing import cast
 from unittest.mock import MagicMock, call, patch
 
+from opensearchpy import NotFoundError
+
 import parsedmarc.opensearch as opensearch_module
 from parsedmarc import InvalidFailureReport, parse_report_file
 from parsedmarc.opensearch import (
     AlreadySaved,
     OpenSearchError,
+    create_data_streams,
     create_indexes,
     migrate_indexes,
     save_aggregate_report_to_opensearch,
@@ -1519,6 +1527,389 @@ class TestSaveSmtpTlsReport(unittest.TestCase):
         self.assertEqual(failure_detail.failure_reason_code, "no-policy-served")
 
 
+# ---------------------------------------------------------------------------
+# Data streams (issue #842)
+# ---------------------------------------------------------------------------
+
+
+def _not_found() -> NotFoundError:
+    """The exception opensearch-py raises for an HTTP 404."""
+    return NotFoundError(404, "index_not_found_exception", {})
+
+
+def _stream_listing(*names: str) -> dict:
+    """A GET _data_stream/<name> response body listing *names*."""
+    return {"data_streams": [{"name": name} for name in names]}
+
+
+@contextmanager
+def _registered_client() -> Iterator[MagicMock]:
+    """Register a mock client as the ``default`` connection.
+
+    Both ``create_data_streams`` and ``Document.save`` resolve the client
+    through the connection registry, so registering the mock there (rather
+    than patching either function) lets the real save path run and the test
+    read what reached the transport. The previous registration is put back
+    afterwards.
+    """
+    client = MagicMock()
+    client.index.return_value = {"_index": "x", "_id": "1", "result": "created"}
+    try:
+        previous = opensearch_module.connections.get_connection("default")
+    except KeyError:
+        previous = None
+    opensearch_module.connections.add_connection("default", client)
+    try:
+        yield client
+    finally:
+        if previous is None:
+            opensearch_module.connections.remove_connection("default")
+        else:
+            opensearch_module.connections.add_connection("default", previous)
+
+
+class _DataStreamTestCase(unittest.TestCase):
+    """Isolates each test from the module's data stream memo."""
+
+    def setUp(self):
+        opensearch_module._READY_DATA_STREAMS.clear()
+
+    def tearDown(self):
+        opensearch_module._READY_DATA_STREAMS.clear()
+
+
+class TestCreateDataStreams(_DataStreamTestCase):
+    def test_creates_template_and_stream_when_missing(self):
+        with (
+            _registered_client() as client,
+            patch("parsedmarc.opensearch.Index") as mock_index_cls,
+        ):
+            client.indices.get_data_stream.side_effect = _not_found()
+            mock_index_cls.return_value.exists.return_value = False
+            create_data_streams(
+                ["dmarc_aggregate"],
+                {"number_of_shards": 2, "number_of_replicas": 1},
+            )
+        template_call = client.indices.put_index_template.call_args.kwargs
+        self.assertEqual(template_call["name"], "dmarc_aggregate")
+        body = template_call["body"]
+        # Exact name, no wildcard: the template must never claim the
+        # date-suffixed indexes written without data streams.
+        self.assertEqual(body["index_patterns"], ["dmarc_aggregate"])
+        self.assertEqual(body["data_stream"], {})
+        self.assertEqual(body["priority"], 501)
+        self.assertNotIn("composed_of", body)
+        self.assertEqual(
+            body["template"],
+            {
+                "settings": {"number_of_shards": 2, "number_of_replicas": 1},
+                "mappings": {"properties": {"@timestamp": {"type": "date"}}},
+            },
+        )
+        self.assertIn("ism_template", body["_meta"]["description"])
+        client.indices.create_data_stream.assert_called_once_with(
+            name="dmarc_aggregate"
+        )
+        # No concrete index is created for a stream.
+        mock_index_cls.return_value.create.assert_not_called()
+        self.assertIn("dmarc_aggregate", opensearch_module._READY_DATA_STREAMS)
+
+    def test_default_settings_are_one_shard_no_replicas(self):
+        with (
+            _registered_client() as client,
+            patch("parsedmarc.opensearch.Index") as mock_index_cls,
+        ):
+            client.indices.get_data_stream.side_effect = _not_found()
+            mock_index_cls.return_value.exists.return_value = False
+            create_data_streams(["smtp_tls"])
+        settings = client.indices.put_index_template.call_args.kwargs["body"][
+            "template"
+        ]["settings"]
+        self.assertEqual(settings, {"number_of_shards": 1, "number_of_replicas": 0})
+
+    def test_existing_stream_gets_template_but_is_not_recreated(self):
+        """The template is put again so that settings changes reach the
+        next backing index; the stream itself already exists."""
+        with (
+            _registered_client() as client,
+            patch("parsedmarc.opensearch.Index") as mock_index_cls,
+        ):
+            client.indices.get_data_stream.return_value = _stream_listing(
+                "dmarc_failure"
+            )
+            create_data_streams(["dmarc_failure"])
+        client.indices.put_index_template.assert_called_once()
+        client.indices.create_data_stream.assert_not_called()
+        mock_index_cls.return_value.exists.assert_not_called()
+        self.assertIn("dmarc_failure", opensearch_module._READY_DATA_STREAMS)
+
+    def test_concrete_index_holding_the_name_raises_without_writing(self):
+        """OpenSearch 3 answers GET _data_stream/<name> for a name held by
+        a concrete index with a 404, so the Index.exists() check is what
+        catches it, before anything is written. Writing to that name would
+        otherwise go into the old undated index, such as one left by a
+        pre-5.0.0 release."""
+        with (
+            _registered_client() as client,
+            patch("parsedmarc.opensearch.Index") as mock_index_cls,
+        ):
+            client.indices.get_data_stream.side_effect = _not_found()
+            mock_index_cls.return_value.exists.return_value = True
+            with self.assertRaises(OpenSearchError) as ctx:
+                create_data_streams(["dmarc_aggregate"])
+        mock_index_cls.assert_called_with("dmarc_aggregate")
+        self.assertIn("already exists", str(ctx.exception))
+        self.assertIn("dmarc_aggregate", str(ctx.exception))
+        client.indices.put_index_template.assert_not_called()
+        client.indices.create_data_stream.assert_not_called()
+        self.assertNotIn("dmarc_aggregate", opensearch_module._READY_DATA_STREAMS)
+
+    def test_empty_listing_is_not_taken_as_an_existing_stream(self):
+        """Elasticsearch answers a name held by a concrete index with 200
+        and an empty list; the shared logic must not read that as "the
+        stream exists" if an OpenSearch release ever does the same."""
+        with (
+            _registered_client() as client,
+            patch("parsedmarc.opensearch.Index") as mock_index_cls,
+        ):
+            client.indices.get_data_stream.return_value = _stream_listing()
+            mock_index_cls.return_value.exists.return_value = True
+            with self.assertRaises(OpenSearchError):
+                create_data_streams(["smtp_tls"])
+        client.indices.put_index_template.assert_not_called()
+        client.indices.create_data_stream.assert_not_called()
+
+    def test_ready_stream_is_not_set_up_again(self):
+        with (
+            _registered_client() as client,
+            patch("parsedmarc.opensearch.Index") as mock_index_cls,
+        ):
+            client.indices.get_data_stream.side_effect = _not_found()
+            mock_index_cls.return_value.exists.return_value = False
+            create_data_streams(["dmarc_aggregate"])
+            create_data_streams(["dmarc_aggregate"])
+        self.assertEqual(client.indices.get_data_stream.call_count, 1)
+        self.assertEqual(client.indices.put_index_template.call_count, 1)
+        self.assertEqual(client.indices.create_data_stream.call_count, 1)
+
+    def test_set_hosts_forgets_ready_streams(self):
+        """A new connection may be a different cluster or carry new index
+        settings, so the template has to be put again."""
+        opensearch_module._READY_DATA_STREAMS.add("dmarc_aggregate")
+        with patch("parsedmarc.opensearch.connections.create_connection"):
+            set_hosts("os:9200")
+        self.assertEqual(opensearch_module._READY_DATA_STREAMS, set())
+
+    def test_sdk_error_is_wrapped_and_retried_next_time(self):
+        with (
+            _registered_client() as client,
+            patch("parsedmarc.opensearch.Index") as mock_index_cls,
+        ):
+            client.indices.get_data_stream.side_effect = _not_found()
+            mock_index_cls.return_value.exists.return_value = False
+            client.indices.put_index_template.side_effect = RuntimeError("cluster down")
+            with self.assertRaises(OpenSearchError) as ctx:
+                create_data_streams(["dmarc_aggregate"])
+            self.assertIn("cluster down", str(ctx.exception))
+            self.assertNotIn("dmarc_aggregate", opensearch_module._READY_DATA_STREAMS)
+            client.indices.create_data_stream.assert_not_called()
+
+            client.indices.put_index_template.side_effect = None
+            create_data_streams(["dmarc_aggregate"])
+        client.indices.create_data_stream.assert_called_once_with(
+            name="dmarc_aggregate"
+        )
+        self.assertIn("dmarc_aggregate", opensearch_module._READY_DATA_STREAMS)
+
+
+class TestSaveToDataStreams(_DataStreamTestCase):
+    """Each save function, run through the real Document.save() into a
+    mock client: in stream mode the document goes to the undated name
+    with op_type=create and an @timestamp; in the default mode it goes to
+    the dated index as before, with the same op_type and @timestamp."""
+
+    def _save(self, save_function, report, **kwargs):
+        with (
+            _registered_client() as client,
+            patch("parsedmarc.opensearch.Search") as mock_search_cls,
+            patch("parsedmarc.opensearch.Index") as mock_index_cls,
+        ):
+            mock_search_cls.return_value.execute.return_value = []
+            client.indices.get_data_stream.side_effect = _not_found()
+            mock_index_cls.return_value.exists.return_value = False
+            save_function(report, **kwargs)
+        return client, mock_search_cls, mock_index_cls
+
+    def test_aggregate_report_goes_to_the_stream(self):
+        client, mock_search_cls, mock_index_cls = self._save(
+            save_aggregate_report_to_opensearch,
+            _aggregate_report(),
+            index_prefix="cust_",
+            index_suffix="tenant_a",
+            # Ignored in stream mode.
+            monthly_indexes=True,
+            use_data_streams=True,
+        )
+        index_call = client.index.call_args.kwargs
+        self.assertEqual(index_call["index"], "cust_dmarc_aggregate_tenant_a")
+        self.assertEqual(index_call["op_type"], "create")
+        self.assertEqual(
+            index_call["body"]["@timestamp"],
+            datetime(2024, 1, 15, tzinfo=timezone.utc),
+        )
+        self.assertEqual(index_call["body"]["report_id"], "agg-1")
+        self.assertEqual(
+            client.indices.put_index_template.call_args.kwargs["body"][
+                "index_patterns"
+            ],
+            ["cust_dmarc_aggregate_tenant_a"],
+        )
+        client.indices.create_data_stream.assert_called_once_with(
+            name="cust_dmarc_aggregate_tenant_a"
+        )
+        mock_index_cls.return_value.create.assert_not_called()
+        # The dedup search still uses the wildcard, which matches the
+        # stream as well as dated indexes written before the switch.
+        self.assertEqual(
+            mock_search_cls.call_args.kwargs["index"],
+            "cust_dmarc_aggregate_tenant_a*",
+        )
+
+    def test_aggregate_report_dated_mode_unchanged_apart_from_timestamp_and_op_type(
+        self,
+    ):
+        client, _, mock_index_cls = self._save(
+            save_aggregate_report_to_opensearch, _aggregate_report()
+        )
+        index_call = client.index.call_args.kwargs
+        self.assertEqual(index_call["index"], "dmarc_aggregate-2024-01-15")
+        self.assertEqual(index_call["op_type"], "create")
+        self.assertEqual(
+            index_call["body"]["@timestamp"],
+            datetime(2024, 1, 15, tzinfo=timezone.utc),
+        )
+        mock_index_cls.return_value.create.assert_called_once()
+        client.indices.put_index_template.assert_not_called()
+        client.indices.create_data_stream.assert_not_called()
+
+    def test_failure_report_goes_to_the_stream(self):
+        client, mock_search_cls, mock_index_cls = self._save(
+            save_failure_report_to_opensearch,
+            _failure_report(),
+            monthly_indexes=True,
+            use_data_streams=True,
+        )
+        index_call = client.index.call_args.kwargs
+        self.assertEqual(index_call["index"], "dmarc_failure")
+        self.assertEqual(index_call["op_type"], "create")
+        self.assertEqual(
+            index_call["body"]["@timestamp"],
+            datetime(2024, 1, 1, tzinfo=timezone.utc),
+        )
+        client.indices.create_data_stream.assert_called_once_with(name="dmarc_failure")
+        mock_index_cls.return_value.create.assert_not_called()
+        self.assertEqual(
+            mock_search_cls.call_args.kwargs["index"],
+            "dmarc_failure*,dmarc_forensic*",
+        )
+
+    def test_failure_report_dated_mode_unchanged_apart_from_timestamp_and_op_type(self):
+        client, _, _ = self._save(save_failure_report_to_opensearch, _failure_report())
+        index_call = client.index.call_args.kwargs
+        self.assertEqual(index_call["index"], "dmarc_failure-2024-01-01")
+        self.assertEqual(index_call["op_type"], "create")
+        self.assertEqual(
+            index_call["body"]["@timestamp"],
+            datetime(2024, 1, 1, tzinfo=timezone.utc),
+        )
+        client.indices.create_data_stream.assert_not_called()
+
+    def test_smtp_tls_report_goes_to_the_stream(self):
+        client, mock_search_cls, mock_index_cls = self._save(
+            save_smtp_tls_report_to_opensearch,
+            _smtp_tls_report(),
+            index_prefix="cust_",
+            use_data_streams=True,
+        )
+        index_call = client.index.call_args.kwargs
+        self.assertEqual(index_call["index"], "cust_smtp_tls")
+        self.assertEqual(index_call["op_type"], "create")
+        self.assertEqual(
+            index_call["body"]["@timestamp"],
+            datetime(2024, 2, 3, tzinfo=timezone.utc),
+        )
+        client.indices.create_data_stream.assert_called_once_with(name="cust_smtp_tls")
+        mock_index_cls.return_value.create.assert_not_called()
+        self.assertEqual(mock_search_cls.call_args.kwargs["index"], "cust_smtp_tls*")
+
+    def test_smtp_tls_report_dated_mode_unchanged_apart_from_timestamp_and_op_type(
+        self,
+    ):
+        client, _, _ = self._save(
+            save_smtp_tls_report_to_opensearch, _smtp_tls_report()
+        )
+        index_call = client.index.call_args.kwargs
+        self.assertEqual(index_call["index"], "smtp_tls-2024-02-03")
+        self.assertEqual(index_call["op_type"], "create")
+        self.assertEqual(
+            index_call["body"]["@timestamp"],
+            datetime(2024, 2, 3, tzinfo=timezone.utc),
+        )
+        client.indices.create_data_stream.assert_not_called()
+
+    def test_stream_set_up_once_for_a_multi_record_report(self):
+        report = _aggregate_report()
+        report["records"].append(report["records"][0].copy())
+        client, _, _ = self._save(
+            save_aggregate_report_to_opensearch, report, use_data_streams=True
+        )
+        self.assertEqual(client.index.call_count, 2)
+        self.assertEqual(client.indices.put_index_template.call_count, 1)
+
+    def test_name_collision_saves_nothing(self):
+        with (
+            _registered_client() as client,
+            patch("parsedmarc.opensearch.Search") as mock_search_cls,
+            patch("parsedmarc.opensearch.Index") as mock_index_cls,
+        ):
+            mock_search_cls.return_value.execute.return_value = []
+            client.indices.get_data_stream.side_effect = _not_found()
+            mock_index_cls.return_value.exists.return_value = True
+            with self.assertRaises(OpenSearchError):
+                save_smtp_tls_report_to_opensearch(
+                    _smtp_tls_report(), use_data_streams=True
+                )
+        client.index.assert_not_called()
+
+
+class TestMigrateIndexesWithDataStream(unittest.TestCase):
+    def test_legacy_fo_migration_skips_a_stream_with_the_name(self):
+        """With use_data_streams the undated legacy name is the stream
+        itself. Index.exists() is true for it, and the field mapping comes
+        back keyed by a backing index; its fo is text, so nothing is
+        migrated or deleted."""
+        with (
+            patch("parsedmarc.opensearch.Index") as mock_index_cls,
+            patch("parsedmarc.opensearch.reindex") as mock_reindex,
+        ):
+            index = mock_index_cls.return_value
+            index.exists.return_value = True
+            index.get_field_mapping.return_value = {
+                ".ds-dmarc_aggregate-000001": {
+                    "mappings": {
+                        "published_policy.fo": {
+                            "full_name": "published_policy.fo",
+                            "mapping": {"fo": {"type": "text"}},
+                        }
+                    }
+                }
+            }
+            migrate_indexes(legacy_fo_indexes=["dmarc_aggregate"])
+        mock_reindex.assert_not_called()
+        index.create.assert_not_called()
+        index.delete.assert_not_called()
+
+
 class TestBackwardCompatAlias(unittest.TestCase):
     def test_save_forensic_alias_points_to_save_failure(self):
         self.assertIs(