add --apm-server-enable-data-streams option (#1229)

scripts/modules/elastic_stack.py

@@ -247,6 +247,9 @@ def add_es_config(args, prefix="output", tls=self.es_tls):
                     ("output.file.path", self.options.get("apm_server_output_file", os.devnull)),
                 ])
 
+        if self.options.get("apm_server_enable_data_streams"):
+            self.apm_server_command_args.append(("apm-server.data_streams.enabled", "true"))
+
         for opt in options.get("apm_server_opt", []):
             self.apm_server_command_args.append(opt.split("=", 1))
 
@@ -387,6 +390,11 @@ def add_arguments(cls, parser):
             action="store_false",
             help="make apm-server act as a Jaeger collector (HTTP and gRPC).",
         )
+        parser.add_argument(
+            "--apm-server-enable-data-streams",
+            action="store_true",
+            help='enable writing to data streams'
+        )
         parser.add_argument(
             '--apm-server-enable-tls',
             action="store_true",

scripts/tests/service_tests.py

@@ -917,6 +917,10 @@ def test_apm_server_custom_pipeline(self):
         self.assertIn("foo:/usr/share/apm-server/ingest/pipeline/definition.json", apm_server["volumes"])
         self.assertIn("apm-server.register.ingest.pipeline.overwrite=true", apm_server["command"])
 
+    def test_data_streams(self):
+        apm_server = ApmServer(version="7.16.0", apm_server_enable_data_streams=True).render()["apm-server"]
+        self.assertIn("apm-server.data_streams.enabled=true", apm_server["command"])
+
     def test_debug(self):
         apm_server = ApmServer(version="6.8.0", apm_server_enable_debug=True).render()["apm-server"]
         self.assertTrue("-d" in apm_server["command"])