Support flattened data_stream.* fields (elastic#3465)

An input configuration supports flattened fields, however the
'data_stream' field was not being correctly decoded when
flattened. This commit fixes this issue.

Some small additions and refactoring are also implemented in the
integration test framework as well as some more detailed
documentation.

---------

Co-authored-by: Paolo Chilà <[email protected]>

.buildkite/pipeline.yml

@@ -140,7 +140,7 @@ steps:
     key: "serverless-integration-tests"
     env:
       TEST_INTEG_AUTH_ESS_REGION: us-east-1
-    command: ".buildkite/scripts/steps/integration_tests.sh serverless integration:single TestMonitoringLogsShipped" #right now, run a single test in serverless mode as a sort of smoke test, instead of re-running the entire suite
+    command: ".buildkite/scripts/steps/integration_tests.sh serverless integration:single TestLogIngestionFleetManaged" #right now, run a single test in serverless mode as a sort of smoke test, instead of re-running the entire suite
     artifact_paths:
       - "build/TEST-**"
       - "build/diagnostics/*"

changelog/fragments/1695389490-Support-flattened-data_stream.-fields.yaml

@@ -0,0 +1,35 @@
+# Kind can be one of:
+# - breaking-change: a change to previously-documented behavior
+# - deprecation: functionality that is being removed in a later release
+# - bug-fix: fixes a problem in a previous version
+# - enhancement: extends functionality but does not break or fix existing behavior
+# - feature: new functionality
+# - known-issue: problems that we are aware of in a given version
+# - security: impacts on the security of a product or a user’s deployment.
+# - upgrade: important information for someone upgrading from a prior version
+# - other: does not fit into any of the other categories
+kind: feature
+
+# Change summary; a 80ish characters long description of the change.
+summary: Support flattened data_stream.* fields in input configuration
+
+# Long description; in case the summary is not enough to describe the change
+# this field accommodate a description without length limits.
+# NOTE: This field will be rendered only for breaking-change and known-issue kinds at the moment.
+description: >-
+  An input configuration supports flattened fields, however the
+  'data_stream' field was not being correctly decoded when
+  flattened. This commit fixes this issue.
+
+# Affected component; usually one of "elastic-agent", "fleet-server", "filebeat", "metricbeat", "auditbeat", "all", etc.
+component: elastic-agent
+
+# PR URL; optional; the PR number that added the changeset.
+# If not present is automatically filled by the tooling finding the PR where this changelog fragment has been added.
+# NOTE: the tooling supports backports, so it's able to fill the original PR number instead of the backport PR number.
+# Please provide it if you are adding a fragment for a different PR.
+pr: https://github.com/elastic/elastic-agent/pull/3465
+
+# Issue URL; optional; the GitHub issue related to this changeset (either closes or is part of).
+# If not present is automatically filled by the tooling with the issue linked to the PR number.
+issue: https://github.com/elastic/elastic-agent/issues/3191

pkg/component/component_test.go

@@ -2361,3 +2361,74 @@ func gatherDurationFieldPaths(s interface{}, pathSoFar string) []string {
 
 	return gatheredPaths
 }
+
+func TestFlattenedDataStream(t *testing.T) {
+	expectedNamespace := "test-namespace"
+	expectedType := "test-type"
+	expectedDataset := "test-dataset"
+
+	policy := map[string]any{
+		"outputs": map[string]any{
+			"default": map[string]any{
+				"type":    "elasticsearch",
+				"enabled": true,
+			},
+		},
+		"inputs": []any{
+			map[string]any{
+				"type":                "filestream",
+				"id":                  "filestream-0",
+				"enabled":             true,
+				"data_stream.type":    expectedType,
+				"data_stream.dataset": expectedDataset,
+				"data_stream": map[string]any{
+					"namespace": expectedNamespace,
+				},
+			},
+		},
+	}
+	runtime, err := LoadRuntimeSpecs(filepath.Join("..", "..", "specs"), PlatformDetail{}, SkipBinaryCheck())
+	if err != nil {
+		t.Fatalf("cannot load runtime specs: %s", err)
+	}
+
+	result, err := runtime.ToComponents(policy, nil, logp.DebugLevel, nil)
+	if err != nil {
+		t.Fatalf("cannot convert policy to component: %s", err)
+	}
+
+	if len(result) != 1 {
+		t.Fatalf("expecting result to have one element, got %d", len(result))
+	}
+
+	if len(result[0].Units) != 2 {
+		t.Fatalf("expecting result[0].Units to have two elements, got %d", len(result))
+	}
+
+	// We do not make assumptions about ordering.
+	// Get the input Unit
+	var dataStream *proto.DataStream
+	for _, unit := range result[0].Units {
+		if unit.Err != nil {
+			t.Fatalf("unit.Err: %s", unit.Err)
+		}
+		if unit.Type == client.UnitTypeInput {
+			dataStream = unit.Config.DataStream
+			break
+		}
+	}
+
+	if dataStream == nil {
+		t.Fatal("DataStream cannot be nil")
+	}
+
+	if dataStream.Dataset != expectedDataset {
+		t.Errorf("expecting DataStream.Dataset: %q, got: %q", expectedDataset, dataStream.Dataset)
+	}
+	if dataStream.Type != expectedType {
+		t.Errorf("expecting DataStream.Type: %q, got: %q", expectedType, dataStream.Type)
+	}
+	if dataStream.Namespace != expectedNamespace {
+		t.Errorf("expecting DataStream.Namespace: %q, got: %q", expectedNamespace, dataStream.Namespace)
+	}
+}

pkg/component/config.go

@@ -15,6 +15,7 @@ import (
 	"google.golang.org/protobuf/types/known/structpb"
 
 	"github.com/elastic/elastic-agent-client/v7/pkg/proto"
+	"github.com/elastic/elastic-agent-libs/config"
 	"github.com/elastic/elastic-agent/pkg/limits"
 )
 
@@ -100,9 +101,88 @@ func ExpectedConfig(cfg map[string]interface{}) (*proto.UnitExpectedConfig, erro
 		return nil, err
 	}
 
+	if err := updateDataStreamsFromSource(result); err != nil {
+		return nil, fmt.Errorf("could not dedot 'data_stream': %w", err)
+	}
+
 	return result, nil
 }
 
+func deDotDataStream(ds *proto.DataStream, source *structpb.Struct) (*proto.DataStream, error) {
+	if ds == nil {
+		ds = &proto.DataStream{}
+	}
+
+	cfg, err := config.NewConfigFrom(source.AsMap())
+	if err != nil {
+		return nil, fmt.Errorf("cannot generate config from source field: %w", err)
+	}
+
+	// Create a temporary struct to unpack the configuration.
+	// Unpack correctly handles any flattened fields like
+	// data_stream.type. So all we need to do is to call Unpack,
+	// ensure the DataStream does not have a different value,
+	// them merge them both.
+	tmp := struct {
+		DataStream struct {
+			Dataset   string `config:"dataset" yaml:"dataset"`
+			Type      string `config:"type" yaml:"type"`
+			Namespace string `config:"namespace" yaml:"namespace"`
+		} `config:"data_stream" yaml:"data_stream"`
+	}{}
+
+	if err := cfg.Unpack(&tmp); err != nil {
+		return nil, fmt.Errorf("cannot unpack source field into struct: %w", err)
+	}
+
+	if (ds.Dataset != tmp.DataStream.Dataset) && (ds.Dataset != "" && tmp.DataStream.Dataset != "") {
+		return nil, errors.New("duplicated key 'datastream.dataset'")
+	}
+
+	if (ds.Type != tmp.DataStream.Type) && (ds.Type != "" && tmp.DataStream.Type != "") {
+		return nil, errors.New("duplicated key 'datastream.type'")
+	}
+
+	if (ds.Namespace != tmp.DataStream.Namespace) && (ds.Namespace != "" && tmp.DataStream.Namespace != "") {
+		return nil, errors.New("duplicated key 'datastream.namespace'")
+	}
+
+	ret := &proto.DataStream{
+		Dataset:   valueOrDefault(tmp.DataStream.Dataset, ds.Dataset),
+		Type:      valueOrDefault(tmp.DataStream.Type, ds.Type),
+		Namespace: valueOrDefault(tmp.DataStream.Namespace, ds.Namespace),
+		Source:    ds.GetSource(),
+	}
+
+	return ret, nil
+}
+
+// valueOrDefault returns b if a is an empty string
+func valueOrDefault(a, b string) string {
+	if a == "" {
+		return b
+	}
+	return a
+}
+
+func updateDataStreamsFromSource(unitConfig *proto.UnitExpectedConfig) error {
+	var err error
+	unitConfig.DataStream, err = deDotDataStream(unitConfig.GetDataStream(), unitConfig.GetSource())
+	if err != nil {
+		return fmt.Errorf("could not parse data_stream from input: %w", err)
+	}
+
+	for i, stream := range unitConfig.Streams {
+		stream.DataStream, err = deDotDataStream(stream.GetDataStream(), stream.GetSource())
+		if err != nil {
+			return fmt.Errorf("could not parse data_stream from stream [%d]: %w",
+				i, err)
+		}
+	}
+
+	return nil
+}
+
 func setSource(val interface{}, cfg map[string]interface{}) error {
 	// find the source field on the val
 	resVal := reflect.ValueOf(val).Elem()

pkg/component/config_test.go

@@ -8,8 +8,10 @@ import (
 	"errors"
 	"testing"
 
+	"github.com/google/go-cmp/cmp"
 	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/require"
+	"google.golang.org/protobuf/testing/protocmp"
 	"google.golang.org/protobuf/types/known/structpb"
 
 	"github.com/elastic/elastic-agent-client/v7/pkg/proto"
@@ -197,7 +199,12 @@ func TestExpectedConfig(t *testing.T) {
 				assert.Equal(t, err.Error(), scenario.Err.Error())
 			} else {
 				require.NoError(t, err)
-				assert.EqualValues(t, scenario.Expected, observed)
+				// protocmp.Transform ensures we do not compare any internal
+				// protobuf fields
+				if !cmp.Equal(scenario.Expected, observed, protocmp.Transform()) {
+					t.Errorf("mismatch (-want +got) \n%s",
+						cmp.Diff(scenario.Expected, observed, protocmp.Transform()))
+				}
 			}
 		})
 	}

pkg/testing/fixture.go

@@ -357,7 +357,14 @@ func (f *Fixture) RunBeat(ctx context.Context) error {
 // Elastic Agent is stopped. If at any time the Elastic Agent logs an error log and the Fixture is not started
 // with `WithAllowErrors()` then `Run` will exit early and return the logged error.
 //
-// If no `states` are provided then the Elastic Agent runs until the context is or the timeout specified with WithRunLength is reached.
+// If no `states` are provided then the Elastic Agent runs until the context is cancelled.
+//
+// The Elastic-Agent is started agent in test mode (--testing-mode) this mode
+// expects the initial configuration (full YAML config) via gRPC.
+// This configuration should be passed in the State.Configure field.
+//
+// The `elastic-agent.yml` generated by `Fixture.Configure` is ignored
+// when `Run` is called.
 func (f *Fixture) Run(ctx context.Context, states ...State) error {
 	if f.binaryName != "elastic-agent" {
 		return errors.New("Run() can only be used with elastic-agent, use RunBeat()")

pkg/testing/tools/estools/elasticsearch.go

@@ -438,9 +438,9 @@ func CheckForErrorsInLogsWithContext(ctx context.Context, client elastictranspor
 
 }
 
-// GetLogsForDatastream returns any logs associated with the datastream
-func GetLogsForDatastream(client elastictransport.Interface, index string) (Documents, error) {
-	return GetLogsForDatastreamWithContext(context.Background(), client, index)
+// GetLogsForDataset returns any logs associated with the datastream
+func GetLogsForDataset(client elastictransport.Interface, index string) (Documents, error) {
+	return GetLogsForDatasetWithContext(context.Background(), client, index)
 }
 
 // GetLogsForAgentID returns any logs associated with the agent ID
@@ -478,8 +478,8 @@ func GetLogsForAgentID(client elastictransport.Interface, id string) (Documents,
 	return handleDocsResponse(res)
 }
 
-// GetLogsForDatastreamWithContext returns any logs associated with the datastream
-func GetLogsForDatastreamWithContext(ctx context.Context, client elastictransport.Interface, index string) (Documents, error) {
+// GetLogsForDatasetWithContext returns any logs associated with the datastream
+func GetLogsForDatasetWithContext(ctx context.Context, client elastictransport.Interface, index string) (Documents, error) {
 	indexQuery := map[string]interface{}{
 		"query": map[string]interface{}{
 			"match": map[string]interface{}{
@@ -536,6 +536,60 @@ func performQueryForRawQuery(ctx context.Context, queryRaw map[string]interface{
 	return handleDocsResponse(res)
 }
 
+// GetLogsForDatastream returns any logs associated with the datastream
+func GetLogsForDatastream(
+	ctx context.Context,
+	client elastictransport.Interface,
+	dsType, dataset, namespace string) (Documents, error) {
+
+	query := map[string]any{
+		"_source": []string{"message"},
+		"query": map[string]any{
+			"bool": map[string]any{
+				"must": []any{
+					map[string]any{
+						"match": map[string]any{
+							"data_stream.dataset": dataset,
+						},
+					},
+					map[string]any{
+						"match": map[string]any{
+							"data_stream.namespace": namespace,
+						},
+					},
+					map[string]any{
+						"match": map[string]any{
+							"data_stream.type": dsType,
+						},
+					},
+				},
+			},
+		},
+	}
+
+	var buf bytes.Buffer
+	if err := json.NewEncoder(&buf).Encode(query); err != nil {
+		return Documents{}, fmt.Errorf("error creating ES query: %w", err)
+	}
+
+	es := esapi.New(client)
+	res, err := es.Search(
+		es.Search.WithIndex(fmt.Sprintf(".ds-%s*", dsType)),
+		es.Search.WithExpandWildcards("all"),
+		es.Search.WithBody(&buf),
+		es.Search.WithTrackTotalHits(true),
+		es.Search.WithPretty(),
+		es.Search.WithContext(ctx),
+	)
+	if err != nil {
+		return Documents{}, fmt.Errorf("error performing ES search: %w", err)
+	}
+
+	return handleDocsResponse(res)
+}
+
+// handleDocsResponse converts the esapi.Response into Documents,
+// it closes the response.Body after reading
 func handleDocsResponse(res *esapi.Response) (Documents, error) {
 	resultBuf, err := handleResponseRaw(res)
 	if err != nil {