diff --git a/changelogs/fragments/beats_filestream_take_over.yml b/changelogs/fragments/beats_filestream_take_over.yml
new file mode 100644
index 00000000..c539f04c
--- /dev/null
+++ b/changelogs/fragments/beats_filestream_take_over.yml
@@ -0,0 +1,6 @@
+minor_changes:
+  - The beats role takes ``beats_filebeat_take_over``. When it is true, the filestream inputs
+    from ``beats_filebeat_log_inputs`` and the MySQL slow log input get ``take_over`` set to true,
+    so hosts that still have the ``log`` inputs of older versions of this role continue at the
+    previous read positions instead of reading all files again
+    (https://www.elastic.co/docs/reference/beats/filebeat/migrate-to-filestream).
diff --git a/roles/beats/README.md b/roles/beats/README.md
index 7287c99f..06f945a2 100644
--- a/roles/beats/README.md
+++ b/roles/beats/README.md
@@ -51,6 +51,7 @@ Run only parts of the role with `--tags`:
 | `beats_filebeat_loadbalance`<br>**Type**: `bool`<br>**Default**: `true` | Enable load balancing for the Filebeat Logstash output. |
 | `beats_filebeat_log_input`<br>**Type**: `bool`<br>**Default**: `true` | Read log files with the inputs from beats_filebeat_log_inputs. |
 | `beats_filebeat_log_inputs`<br>**Type**: `dict`<br>**Default**: `{'messages': {'paths': ['/var/log/messages', '/var/log/syslog']}}` | Log files to read, keyed by a free name that becomes the filestream id. Each entry has paths and optional fields and multiline settings. See the Filebeat inputs documentation. |
+| `beats_filebeat_take_over`<br>**Type**: `bool`<br>**Default**: `false` | Set take_over on the filestream inputs from beats_filebeat_log_inputs and on the MySQL slow log input, so they continue at the read positions of the log inputs that older versions of this role generated instead of reading the files again. Only needed once when upgrading hosts that still have log inputs. See the Filebeat inputs documentation. |
 | `beats_filebeat_syslog_tcp`<br>**Type**: `bool`<br>**Default**: `false` | Enable a TCP syslog input. |
 | `beats_filebeat_syslog_tcp_port`<br>**Type**: `int`<br>**Default**: `514` | Port of the TCP syslog input. |
 | `beats_filebeat_syslog_udp`<br>**Type**: `bool`<br>**Default**: `false` | Enable a UDP syslog input. |
diff --git a/roles/beats/defaults/main.yml b/roles/beats/defaults/main.yml
index f80ed9ea..2f13b0cb 100644
--- a/roles/beats/defaults/main.yml
+++ b/roles/beats/defaults/main.yml
@@ -29,6 +29,7 @@ beats_filebeat_log_inputs:
     paths:
       - /var/log/messages
       - /var/log/syslog
+beats_filebeat_take_over: false
 beats_filebeat_enable: true
 beats_filebeat_journald: false
 beats_filebeat_journald_inputs:
diff --git a/roles/beats/docs/filebeat-inputs.md b/roles/beats/docs/filebeat-inputs.md
index a7d9e442..b52229f9 100644
--- a/roles/beats/docs/filebeat-inputs.md
+++ b/roles/beats/docs/filebeat-inputs.md
@@ -36,6 +36,16 @@ beats_filebeat_log_inputs:
 
 `multiline` maps directly to the Filebeat multiline parser (`type`, `pattern`, `negate`, `match`). See [Manage multiline messages](https://www.elastic.co/docs/reference/beats/filebeat/multiline-examples).
 
+### Upgrading from `log` inputs
+
+Older versions of this role generated `log` inputs for `beats_filebeat_log_inputs` and for the MySQL slow log. The `filestream` inputs that replace them do not use the read positions of the `log` inputs, so without further steps Filebeat reads every configured file again from the beginning and ships its content a second time.
+
+Set `beats_filebeat_take_over: true` for the run that upgrades such hosts. The role then adds `take_over: true` to these inputs, and Filebeat continues each file where the old `log` input stopped. Take over mode requires that the new inputs have the same `paths` as the old ones and that the `log` inputs are gone. The role generates both kinds of input from the same variables and no longer generates `log` inputs, so this holds as long as `paths` is not changed in the same run.
+
+In Filebeat 8.19 take over mode is beta, and Filebeat keeps a backup of the registry in `registry.path/filebeat`, which makes it reversible. Filebeat 9.1 added `take_over.enabled` as the new syntax and still accepts `take_over: true` for migrating from `log` inputs. The migration guide describes a `take_over` tag on the events read this way, which can be used to check the result. Once all hosts are migrated, `beats_filebeat_take_over` can be set back to `false`.
+
+See [Migrate `log` input configurations to `filestream`](https://www.elastic.co/docs/reference/beats/filebeat/migrate-to-filestream) and the `take_over` option in the [filestream input reference](https://www.elastic.co/docs/reference/beats/filebeat/filebeat-input-filestream).
+
 ## Fields
 
 There are two ways to add fields, and they use **different shapes**:
diff --git a/roles/beats/meta/argument_specs.yml b/roles/beats/meta/argument_specs.yml
index dc2b960e..411f5d18 100644
--- a/roles/beats/meta/argument_specs.yml
+++ b/roles/beats/meta/argument_specs.yml
@@ -131,6 +131,16 @@ argument_specs:
           Each entry has paths and optional fields and multiline settings. See the
           Filebeat inputs documentation.
 
+      beats_filebeat_take_over:
+        type: bool
+        default: false
+        description: >-
+          Set take_over on the filestream inputs from beats_filebeat_log_inputs and on
+          the MySQL slow log input, so they continue at the read positions of the log
+          inputs that older versions of this role generated instead of reading the
+          files again. Only needed once when upgrading hosts that still have log inputs.
+          See the Filebeat inputs documentation.
+
       beats_filebeat_syslog_tcp:
         type: bool
         default: false
diff --git a/roles/beats/templates/filebeat.yml.j2 b/roles/beats/templates/filebeat.yml.j2
index 453d8def..3d7e00bb 100644
--- a/roles/beats/templates/filebeat.yml.j2
+++ b/roles/beats/templates/filebeat.yml.j2
@@ -4,6 +4,9 @@ filebeat.inputs:
 {% for input in beats_filebeat_log_inputs %}
 - type: filestream
   id: {{ input }}-filestream
+{% if beats_filebeat_take_over | bool %}
+  take_over: true
+{% endif %}
   enabled: true
   paths:
 {% for path in beats_filebeat_log_inputs[input].paths %}
@@ -44,6 +47,9 @@ filebeat.inputs:
 {% if beats_filebeat_mysql_slowlog_input | bool %}
 - type: filestream
   id: mysql-slowlog-filestream
+{% if beats_filebeat_take_over | bool %}
+  take_over: true
+{% endif %}
   enabled: true
   paths:
     - /var/log/mysql/*-slow.log