Add a way to select the stream under test in system test runner (#187)

This provides a means to select a stream to test by input type.

Closes #186

docs/howto/system_testing.md

@@ -75,10 +75,11 @@ Next, we must define configuration for each data stream that we want to system t
             config.yml
 ```
 
-The `config.yml` file allows you define values for package and data stream-level variables. For example, the `apache/access` data stream's `config.yml` is shown below.
+The `config.yml` file allows you to define values for package and data stream-level variables. For example, the `apache/access` data stream's `config.yml` is shown below.
 
 ```
 vars: ~
+input: logfile
 data_stream:
   vars:
     paths:
@@ -91,6 +92,11 @@ The `data_stream.vars` field corresponds to data stream-level variables for the
 
 Notice the use of the `{{SERVICE_LOGS_DIR}}` placeholder. This corresponds to the `${SERVICE_LOGS_DIR}` variable we saw in the `docker-compose.yml` file earlier. In the above example, the net effect is as if the `/usr/local/apache2/logs/access.log*` files located inside the Apache integration service container become available at the same path from Elastic Agent's perspective.
 
+When a data stream's manifest declares multiple streams with different inputs
+you can use the `input` option to select the stream to test. The first stream
+whose input type matches the `input` value will be tested. By default, the first
+stream declared in the manifest will be tested. 
+
 #### Placeholders
 
 The `SERVICE_LOGS_DIR` placeholder is not the only one available for use in a data stream's `config.yml` file. The complete list of available placeholders is shown below.

internal/testrunner/runners/system/runner.go

@@ -354,7 +354,8 @@ func createPackageDatastream(
 	ds packages.DataStreamManifest,
 	c testConfig,
 ) ingestmanager.PackageDataStream {
-	streamInput := ds.Streams[0].Input
+	stream := ds.Streams[getDataStreamIndex(c.Input, ds)]
+	streamInput := stream.Input
 	r := ingestmanager.PackageDataStream{
 		Name:      fmt.Sprintf("%s-%s", pkg.Name, ds.Name),
 		Namespace: "ep",
@@ -386,7 +387,7 @@ func createPackageDatastream(
 
 	// Add dataStream-level vars
 	dsVars := ingestmanager.Vars{}
-	for _, dsVar := range ds.Streams[0].Vars {
+	for _, dsVar := range stream.Vars {
 		val := dsVar.Default
 
 		cfgVar, exists := c.DataStream.Vars[dsVar.Name]
@@ -427,6 +428,17 @@ func createPackageDatastream(
 	return r
 }
 
+// getDataStreamIndex returns the index of the data stream whose input name
+// matches. Otherwise it returns the 0.
+func getDataStreamIndex(inputName string, ds packages.DataStreamManifest) int {
+	for i, s := range ds.Streams {
+		if s.Input == inputName {
+			return i
+		}
+	}
+	return 0
+}
+
 func deleteDataStreamDocs(esClient *es.Client, dataStream string) error {
 	body := strings.NewReader(`{ "query": { "match_all": {} } }`)
 	_, err := esClient.DeleteByQuery([]string{dataStream}, body)

internal/testrunner/runners/system/test_config.go

@@ -22,6 +22,7 @@ import (
 const configFileName = "config.yml"
 
 type testConfig struct {
+	Input      string                       `config:"input"`
 	Vars       map[string]packages.VarValue `config:"vars"`
 	DataStream struct {
 		Vars map[string]packages.VarValue `config:"vars"`