diff --git a/x-pack/elastic-agent/CHANGELOG.asciidoc b/x-pack/elastic-agent/CHANGELOG.asciidoc
index a11165e28909..e75e4efafd75 100644
--- a/x-pack/elastic-agent/CHANGELOG.asciidoc
+++ b/x-pack/elastic-agent/CHANGELOG.asciidoc
@@ -78,4 +78,5 @@
 - Rename input.type logs to logfile {pull}19360[19360]
 - Agent now installs/uninstalls Elastic Endpoint {pull}19248[19248]
 - Agent now downloads Elastic Endpoint {pull}19503[19503]
+- Refuse invalid stream values in configuration {pull}19587[19587]
 - Agent now load balances across multiple Kibana instances {pull}19628[19628]
diff --git a/x-pack/elastic-agent/_meta/config/common.p2.yml.tmpl b/x-pack/elastic-agent/_meta/config/common.p2.yml.tmpl
index 3cb0bcdac463..5d853ca38eb6 100644
--- a/x-pack/elastic-agent/_meta/config/common.p2.yml.tmpl
+++ b/x-pack/elastic-agent/_meta/config/common.p2.yml.tmpl
@@ -10,10 +10,23 @@ outputs:
 
 inputs:
   - type: system/metrics
+
+    # The only two requirement are that it has only characters allowed in an Elasticsearch index name
+    # Index names must meet the following criteria:
+    #   Lowercase only
+    #   Cannot include \, /, *, ?, ", <, >, |, ` ` (space character), ,, #
+    #   Cannot start with -, _, +
+    #   Cannot be . or ..
     dataset.namespace: default
     use_output: default
     streams:
       - metricset: cpu
+        # The only two requirement are that it has only characters allowed in an Elasticsearch index name
+        # Index names must meet the following criteria:
+        #   Lowercase only
+        #   Cannot include \, /, *, ?, ", <, >, |, ` ` (space character), ,, #
+        #   Cannot start with -, _, +
+        #   Cannot be . or ..
         dataset.name: system.cpu
       - metricset: memory
         dataset.name: system.memory
diff --git a/x-pack/elastic-agent/_meta/config/common.reference.p2.yml.tmpl b/x-pack/elastic-agent/_meta/config/common.reference.p2.yml.tmpl
index 97290ee4daef..5756a84499bc 100644
--- a/x-pack/elastic-agent/_meta/config/common.reference.p2.yml.tmpl
+++ b/x-pack/elastic-agent/_meta/config/common.reference.p2.yml.tmpl
@@ -10,10 +10,23 @@ outputs:
 
 inputs:
   - type: system/metrics
+
+    # The only two requirement are that it has only characters allowed in an Elasticsearch index name
+    # Index names must meet the following criteria:
+    #   Lowercase only
+    #   Cannot include \, /, *, ?, ", <, >, |, ` ` (space character), ,, #
+    #   Cannot start with -, _, +
+    #   Cannot be . or ..
     dataset.namespace: default
     use_output: default
     streams:
       - metricset: cpu
+        # The only two requirement are that it has only characters allowed in an Elasticsearch index name
+        # Index names must meet the following criteria:
+        #   Lowercase only
+        #   Cannot include \, /, *, ?, ", <, >, |, ` ` (space character), ,, #
+        #   Cannot start with -, _, +
+        #   Cannot be . or ..
         dataset.name: system.cpu
       - metricset: memory
         dataset.name: system.memory
diff --git a/x-pack/elastic-agent/_meta/config/elastic-agent.docker.yml.tmpl b/x-pack/elastic-agent/_meta/config/elastic-agent.docker.yml.tmpl
index 70b4f496c3c0..915615ce4e56 100644
--- a/x-pack/elastic-agent/_meta/config/elastic-agent.docker.yml.tmpl
+++ b/x-pack/elastic-agent/_meta/config/elastic-agent.docker.yml.tmpl
@@ -10,10 +10,23 @@ outputs:
 
 inputs:
   - type: system/metrics
+
+    # The only two requirement are that it has only characters allowed in an Elasticsearch index name
+    # Index names must meet the following criteria:
+    #   Lowercase only
+    #   Cannot include \, /, *, ?, ", <, >, |, ` ` (space character), ,, #
+    #   Cannot start with -, _, +
+    #   Cannot be . or ..
     dataset.namespace: default
     use_output: default
     streams:
       - metricset: cpu
+        # The only two requirement are that it has only characters allowed in an Elasticsearch index name
+        # Index names must meet the following criteria:
+        #   Lowercase only
+        #   Cannot include \, /, *, ?, ", <, >, |, ` ` (space character), ,, #
+        #   Cannot start with -, _, +
+        #   Cannot be . or ..
         dataset.name: system.cpu
       - metricset: memory
         dataset.name: system.memory
diff --git a/x-pack/elastic-agent/_meta/elastic-agent.yml b/x-pack/elastic-agent/_meta/elastic-agent.yml
index 8c3c518194dd..77033171ea9a 100644
--- a/x-pack/elastic-agent/_meta/elastic-agent.yml
+++ b/x-pack/elastic-agent/_meta/elastic-agent.yml
@@ -10,10 +10,23 @@ outputs:
 
 inputs:
   - type: system/metrics
+
+    # The only two requirement are that it has only characters allowed in an Elasticsearch index name
+    # Index names must meet the following criteria:
+    #   Lowercase only
+    #   Cannot include \, /, *, ?, ", <, >, |, ` ` (space character), ,, #
+    #   Cannot start with -, _, +
+    #   Cannot be . or ..
     dataset.namespace: default
     use_output: default
     streams:
       - metricset: cpu
+        # The only two requirement are that it has only characters allowed in an Elasticsearch index name
+        # Index names must meet the following criteria:
+        #   Lowercase only
+        #   Cannot include \, /, *, ?, ", <, >, |, ` ` (space character), ,, #
+        #   Cannot start with -, _, +
+        #   Cannot be . or ..
         dataset.name: system.cpu
       - metricset: memory
         dataset.name: system.memory
diff --git a/x-pack/elastic-agent/elastic-agent.docker.yml b/x-pack/elastic-agent/elastic-agent.docker.yml
index 6f800c1dc082..2bf794785c4f 100644
--- a/x-pack/elastic-agent/elastic-agent.docker.yml
+++ b/x-pack/elastic-agent/elastic-agent.docker.yml
@@ -10,10 +10,23 @@ outputs:
 
 inputs:
   - type: system/metrics
+
+    # The only two requirement are that it has only characters allowed in an Elasticsearch index name
+    # Index names must meet the following criteria:
+    #   Lowercase only
+    #   Cannot include \, /, *, ?, ", <, >, |, ` ` (space character), ,, #
+    #   Cannot start with -, _, +
+    #   Cannot be . or ..
     dataset.namespace: default
     use_output: default
     streams:
       - metricset: cpu
+        # The only two requirement are that it has only characters allowed in an Elasticsearch index name
+        # Index names must meet the following criteria:
+        #   Lowercase only
+        #   Cannot include \, /, *, ?, ", <, >, |, ` ` (space character), ,, #
+        #   Cannot start with -, _, +
+        #   Cannot be . or ..
         dataset.name: system.cpu
       - metricset: memory
         dataset.name: system.memory
diff --git a/x-pack/elastic-agent/elastic-agent.reference.yml b/x-pack/elastic-agent/elastic-agent.reference.yml
index 0607ef8d2266..0c0eb5944dd4 100644
--- a/x-pack/elastic-agent/elastic-agent.reference.yml
+++ b/x-pack/elastic-agent/elastic-agent.reference.yml
@@ -16,10 +16,23 @@ outputs:
 
 inputs:
   - type: system/metrics
+
+    # The only two requirement are that it has only characters allowed in an Elasticsearch index name
+    # Index names must meet the following criteria:
+    #   Lowercase only
+    #   Cannot include \, /, *, ?, ", <, >, |, ` ` (space character), ,, #
+    #   Cannot start with -, _, +
+    #   Cannot be . or ..
     dataset.namespace: default
     use_output: default
     streams:
       - metricset: cpu
+        # The only two requirement are that it has only characters allowed in an Elasticsearch index name
+        # Index names must meet the following criteria:
+        #   Lowercase only
+        #   Cannot include \, /, *, ?, ", <, >, |, ` ` (space character), ,, #
+        #   Cannot start with -, _, +
+        #   Cannot be . or ..
         dataset.name: system.cpu
       - metricset: memory
         dataset.name: system.memory
diff --git a/x-pack/elastic-agent/elastic-agent.yml b/x-pack/elastic-agent/elastic-agent.yml
index b709e9c8ab52..9dd66993da73 100644
--- a/x-pack/elastic-agent/elastic-agent.yml
+++ b/x-pack/elastic-agent/elastic-agent.yml
@@ -16,10 +16,23 @@ outputs:
 
 inputs:
   - type: system/metrics
+
+    # The only two requirement are that it has only characters allowed in an Elasticsearch index name
+    # Index names must meet the following criteria:
+    #   Lowercase only
+    #   Cannot include \, /, *, ?, ", <, >, |, ` ` (space character), ,, #
+    #   Cannot start with -, _, +
+    #   Cannot be . or ..
     dataset.namespace: default
     use_output: default
     streams:
       - metricset: cpu
+        # The only two requirement are that it has only characters allowed in an Elasticsearch index name
+        # Index names must meet the following criteria:
+        #   Lowercase only
+        #   Cannot include \, /, *, ?, ", <, >, |, ` ` (space character), ,, #
+        #   Cannot start with -, _, +
+        #   Cannot be . or ..
         dataset.name: system.cpu
       - metricset: memory
         dataset.name: system.memory
diff --git a/x-pack/elastic-agent/pkg/agent/application/filters/constraints_filter_test.go b/x-pack/elastic-agent/pkg/agent/application/filters/constraints_filter_test.go
index d7a748d7fbdc..d7e3190dd923 100644
--- a/x-pack/elastic-agent/pkg/agent/application/filters/constraints_filter_test.go
+++ b/x-pack/elastic-agent/pkg/agent/application/filters/constraints_filter_test.go
@@ -22,14 +22,14 @@ func TestEvaluation(t *testing.T) {
 	}
 
 	testCases := []testCase{
-		testCase{"simple version", "validate_version(%{[agent.version]}, '" + release.Version() + "')", true},
-		testCase{"~ version release", "validate_version(%{[agent.version]}, '~" + release.Version() + "')", true},
-		testCase{"^ version release", "validate_version(%{[agent.version]}, '^" + release.Version() + "')", true},
-		testCase{"range to release", "validate_version(%{[agent.version]}, '1.0.0 - " + release.Version() + "')", true},
-		testCase{"range lower", "validate_version(%{[agent.version]}, '1.0.0 - 5.0.0')", false},
-		testCase{"range include", "validate_version(%{[agent.version]}, '1.0.0 - 100.0.0')", true},
-		testCase{"family should equal", "%{[os.family]} == '" + runtime.GOOS + "'", true},
-		testCase{"family should not equal", "%{[os.family]} != '" + runtime.GOOS + "'", false},
+		{"simple version", "validate_version(%{[agent.version]}, '" + release.Version() + "')", true},
+		{"~ version release", "validate_version(%{[agent.version]}, '~" + release.Version() + "')", true},
+		{"^ version release", "validate_version(%{[agent.version]}, '^" + release.Version() + "')", true},
+		{"range to release", "validate_version(%{[agent.version]}, '1.0.0 - " + release.Version() + "')", true},
+		{"range lower", "validate_version(%{[agent.version]}, '1.0.0 - 5.0.0')", false},
+		{"range include", "validate_version(%{[agent.version]}, '1.0.0 - 100.0.0')", true},
+		{"family should equal", "%{[os.family]} == '" + runtime.GOOS + "'", true},
+		{"family should not equal", "%{[os.family]} != '" + runtime.GOOS + "'", false},
 	}
 
 	for _, tc := range testCases {
diff --git a/x-pack/elastic-agent/pkg/agent/application/filters/stream_checker.go b/x-pack/elastic-agent/pkg/agent/application/filters/stream_checker.go
new file mode 100644
index 000000000000..470cb776ab13
--- /dev/null
+++ b/x-pack/elastic-agent/pkg/agent/application/filters/stream_checker.go
@@ -0,0 +1,199 @@
+// Copyright Elasticsearch B.V. and/or licensed to Elasticsearch B.V. under one
+// or more contributor license agreements. Licensed under the Elastic License;
+// you may not use this file except in compliance with the Elastic License.
+
+package filters
+
+import (
+	"fmt"
+	"strings"
+
+	"github.com/elastic/beats/v7/x-pack/elastic-agent/pkg/agent/errors"
+	"github.com/elastic/beats/v7/x-pack/elastic-agent/pkg/agent/transpiler"
+	"github.com/elastic/beats/v7/x-pack/elastic-agent/pkg/core/logger"
+)
+
+// ErrInvalidNamespace is error returned when namespace value provided is invalid.
+var ErrInvalidNamespace = errors.New("provided namespace is invalid", errors.TypeConfig)
+
+// ErrInvalidDataset is error returned when dataset name value provided is invalid.
+var ErrInvalidDataset = errors.New("provided dataset name is invalid", errors.TypeConfig)
+
+// ErrInvalidIndex occurs when concatenation of {dataset.type}-{dataset.name}-{dataset.namespace} does not meet index criteria.
+var ErrInvalidIndex = errors.New("provided combination of type, dataset name and namespace is invalid", errors.TypeConfig)
+
+// StreamChecker checks for invalid values in stream namespace and dataset.
+func StreamChecker(log *logger.Logger, ast *transpiler.AST) error {
+	inputsNode, found := transpiler.Lookup(ast, "inputs")
+	if !found {
+		return nil
+	}
+
+	inputsNodeList, ok := inputsNode.Value().(*transpiler.List)
+	if !ok {
+		return nil
+	}
+
+	inputsNodeListCollection, ok := inputsNodeList.Value().([]transpiler.Node)
+	if !ok {
+		return errors.New("inputs is not a list", errors.TypeConfig)
+	}
+
+	for _, inputNode := range inputsNodeListCollection {
+		namespace := "default"
+		datasetName := "generic"
+		// fail only if dataset.namespace or dataset[namespace] is found and invalid
+		// not provided values are ok and will be fixed by rules
+		if nsNode, found := inputNode.Find("dataset.namespace"); found {
+			nsKey, ok := nsNode.(*transpiler.Key)
+			if ok {
+				newNamespace := nsKey.Value().(transpiler.Node).String()
+				if !isValid(newNamespace) {
+					return ErrInvalidNamespace
+				}
+				namespace = newNamespace
+			}
+		} else {
+			dsNode, found := inputNode.Find("dataset")
+			if found {
+				// got a dataset
+				datasetMap, ok := dsNode.Value().(*transpiler.Dict)
+				if ok {
+					nsNode, found := datasetMap.Find("namespace")
+					if found {
+						nsKey, ok := nsNode.(*transpiler.Key)
+						if ok {
+							newNamespace := nsKey.Value().(transpiler.Node).String()
+							if !isValid(newNamespace) {
+								return ErrInvalidNamespace
+							}
+							namespace = newNamespace
+						}
+					}
+				}
+			}
+		}
+
+		// get the type, longest type for now is metrics
+		datasetType := "metrics"
+		if nsNode, found := inputNode.Find("dataset.type"); found {
+			nsKey, ok := nsNode.(*transpiler.Key)
+			if ok {
+				newDataset := nsKey.Value().(transpiler.Node).String()
+				datasetType = newDataset
+			}
+		} else {
+			dsNode, found := inputNode.Find("dataset")
+			if found {
+				// got a dataset
+				datasetMap, ok := dsNode.Value().(*transpiler.Dict)
+				if ok {
+					nsNode, found := datasetMap.Find("type")
+					if found {
+						nsKey, ok := nsNode.(*transpiler.Key)
+						if ok {
+							newDataset := nsKey.Value().(transpiler.Node).String()
+							datasetType = newDataset
+						}
+					}
+				}
+			}
+		}
+
+		streamsNode, ok := inputNode.Find("streams")
+		if ok {
+			streamsList, ok := streamsNode.Value().(*transpiler.List)
+			if ok {
+				streamNodes, ok := streamsList.Value().([]transpiler.Node)
+				if !ok {
+					return errors.New("streams is not a list", errors.TypeConfig)
+				}
+
+				for _, streamNode := range streamNodes {
+					streamMap, ok := streamNode.(*transpiler.Dict)
+					if !ok {
+						continue
+					}
+
+					// fix this only if in compact form
+					if dsNameNode, found := streamMap.Find("dataset.name"); found {
+						dsKey, ok := dsNameNode.(*transpiler.Key)
+						if ok {
+							newDataset := dsKey.Value().(transpiler.Node).String()
+							if !isValid(newDataset) {
+								return ErrInvalidDataset
+							}
+							datasetName = newDataset
+						}
+					} else {
+						datasetNode, found := streamMap.Find("dataset")
+						if found {
+							datasetMap, ok := datasetNode.Value().(*transpiler.Dict)
+							if !ok {
+								continue
+							}
+
+							dsNameNode, found := datasetMap.Find("name")
+							if found {
+								dsKey, ok := dsNameNode.(*transpiler.Key)
+								if ok {
+									newDataset := dsKey.Value().(transpiler.Node).String()
+									if !isValid(newDataset) {
+										return ErrInvalidDataset
+									}
+									datasetName = newDataset
+								}
+							}
+						}
+					}
+				}
+			}
+		}
+
+		if indexName := fmt.Sprintf("%s-%s-%s", datasetType, datasetName, namespace); !matchesIndexContraints(indexName) {
+			return ErrInvalidIndex
+		}
+	}
+
+	return nil
+}
+
+// The only two requirement are that it has only characters allowed in an Elasticsearch index name
+// and does NOT contain a `-`.
+func isValid(namespace string) bool {
+	return matchesIndexContraints(namespace) && !strings.Contains(namespace, "-")
+}
+
+// The only two requirement are that it has only characters allowed in an Elasticsearch index name
+// Index names must meet the following criteria:
+//     Lowercase only
+//     Cannot include \, /, *, ?, ", <, >, |, ` ` (space character), ,, #
+//     Cannot start with -, _, +
+//     Cannot be . or ..
+func matchesIndexContraints(namespace string) bool {
+	// Cannot be . or ..
+	if namespace == "." || namespace == ".." {
+		return false
+	}
+
+	if len(namespace) <= 0 || len(namespace) > 255 {
+		return false
+	}
+
+	// Lowercase only
+	if strings.ToLower(namespace) != namespace {
+		return false
+	}
+
+	// Cannot include \, /, *, ?, ", <, >, |, ` ` (space character), ,, #
+	if strings.ContainsAny(namespace, "\\/*?\"<>| ,#") {
+		return false
+	}
+
+	// Cannot start with -, _, +
+	if strings.HasPrefix(namespace, "-") || strings.HasPrefix(namespace, "_") || strings.HasPrefix(namespace, "+") {
+		return false
+	}
+
+	return true
+}
diff --git a/x-pack/elastic-agent/pkg/agent/application/filters/stream_checker_test.go b/x-pack/elastic-agent/pkg/agent/application/filters/stream_checker_test.go
new file mode 100644
index 000000000000..b6d9bc22e0d9
--- /dev/null
+++ b/x-pack/elastic-agent/pkg/agent/application/filters/stream_checker_test.go
@@ -0,0 +1,291 @@
+// Copyright Elasticsearch B.V. and/or licensed to Elasticsearch B.V. under one
+// or more contributor license agreements. Licensed under the Elastic License;
+// you may not use this file except in compliance with the Elastic License.
+
+package filters
+
+import (
+	"crypto/sha512"
+	"encoding/hex"
+	"fmt"
+	"testing"
+
+	"github.com/stretchr/testify/assert"
+
+	"github.com/elastic/beats/v7/x-pack/elastic-agent/pkg/agent/transpiler"
+	"github.com/elastic/beats/v7/x-pack/elastic-agent/pkg/core/logger"
+)
+
+const semiLongString = ""
+
+func TestStreamCheck(t *testing.T) {
+	type testCase struct {
+		name      string
+		configMap map[string]interface{}
+		result    error
+	}
+
+	h := hex.EncodeToString(sha512.New().Sum(nil))
+	semiLongString := h[:86]
+	longString := fmt.Sprintf("%s%s", h, h)
+
+	testCases := []testCase{
+		{
+			name:      "all missing",
+			configMap: map[string]interface{}{},
+			result:    nil,
+		},
+		{
+			name: "all ok - compact",
+			configMap: map[string]interface{}{
+				"inputs": []map[string]interface{}{
+					{
+						"dataset.namespace": "somenamespace",
+						"streams":           []map[string]interface{}{{"dataset.name": "somedatasetname"}},
+					},
+				},
+			},
+			result: nil,
+		},
+		{
+			name: "all ok - long",
+			configMap: map[string]interface{}{
+				"inputs": []map[string]interface{}{
+					{
+						"dataset": map[string]interface{}{
+							"namespace": "somenamespace",
+						},
+						"streams": []map[string]interface{}{
+							{
+								"dataset": map[string]interface{}{
+									"name": "somedatasetname",
+								},
+							},
+						},
+					},
+				},
+			},
+			result: nil,
+		},
+		{
+			name: "dataset.name invalid - compact",
+			configMap: map[string]interface{}{
+				"inputs": []map[string]interface{}{
+					{"streams": []map[string]interface{}{{"dataset.name": ""}}},
+				},
+			},
+			result: ErrInvalidDataset,
+		},
+		{
+			name: "dataset.name invalid - long",
+			configMap: map[string]interface{}{
+				"inputs": []map[string]interface{}{
+					{
+						"streams": []map[string]interface{}{
+							{
+								"dataset": map[string]interface{}{
+									"name": "",
+								},
+							},
+						},
+					},
+				},
+			},
+			result: ErrInvalidDataset,
+		},
+
+		{
+			name: "dataset.name invalid dot - compact",
+			configMap: map[string]interface{}{
+				"inputs": []map[string]interface{}{
+					{"streams": []map[string]interface{}{{"dataset.name": "."}}},
+				},
+			},
+			result: ErrInvalidDataset,
+		},
+		{
+			name: "dataset.name invalid dotdot- compact",
+			configMap: map[string]interface{}{
+				"inputs": []map[string]interface{}{
+					{"streams": []map[string]interface{}{{"dataset.name": ".."}}},
+				},
+			},
+			result: ErrInvalidDataset,
+		},
+		{
+			name: "dataset.name invalid uppercase - compact",
+			configMap: map[string]interface{}{
+				"inputs": []map[string]interface{}{
+					{"streams": []map[string]interface{}{{"dataset.name": "myNameIs"}}},
+				},
+			},
+			result: ErrInvalidDataset,
+		},
+		{
+			name: "dataset.name invalid space- compact",
+			configMap: map[string]interface{}{
+				"inputs": []map[string]interface{}{
+					{"streams": []map[string]interface{}{{"dataset.name": "outer space"}}},
+				},
+			},
+			result: ErrInvalidDataset,
+		},
+		{
+			name: "dataset.name invalid invalid char- compact",
+			configMap: map[string]interface{}{
+				"inputs": []map[string]interface{}{
+					{"streams": []map[string]interface{}{{"dataset.name": "is\\thisvalid"}}},
+				},
+			},
+			result: ErrInvalidDataset,
+		},
+		{
+			name: "dataset.name invalid invalid prefix- compact",
+			configMap: map[string]interface{}{
+				"inputs": []map[string]interface{}{
+					{"streams": []map[string]interface{}{{"dataset.name": "_isthisvalid"}}},
+				},
+			},
+			result: ErrInvalidDataset,
+		},
+
+		{
+			name: "namespace invalid - compact",
+			configMap: map[string]interface{}{
+				"inputs": []map[string]interface{}{{"dataset.namespace": ""}},
+			},
+			result: ErrInvalidNamespace,
+		},
+		{
+			name: "namespace invalid name 1 - compact",
+			configMap: map[string]interface{}{
+				"inputs": []map[string]interface{}{
+					{"dataset.namespace": "."},
+				},
+			},
+			result: ErrInvalidNamespace,
+		},
+		{
+			name: "namespace invalid name 2 - compact",
+			configMap: map[string]interface{}{
+				"inputs": []map[string]interface{}{{"dataset.namespace": ".."}},
+			},
+			result: ErrInvalidNamespace,
+		},
+		{
+			name: "namespace invalid name uppercase - compact",
+			configMap: map[string]interface{}{
+				"inputs": []map[string]interface{}{{"dataset.namespace": "someUpper"}},
+			},
+			result: ErrInvalidNamespace,
+		},
+		{
+			name: "namespace invalid name space - compact",
+			configMap: map[string]interface{}{
+				"inputs": []map[string]interface{}{{"dataset.namespace": "some space"}},
+			},
+			result: ErrInvalidNamespace,
+		},
+		{
+			name: "namespace invalid name invalid char - compact",
+			configMap: map[string]interface{}{
+				"inputs": []map[string]interface{}{{"dataset.namespace": "isitok?"}},
+			},
+			result: ErrInvalidNamespace,
+		},
+		{
+			name: "namespace invalid name invalid prefix - compact",
+			configMap: map[string]interface{}{
+				"inputs": []map[string]interface{}{{"dataset.namespace": "+isitok"}},
+			},
+			result: ErrInvalidNamespace,
+		},
+		{
+			name: "namespace invalid - long",
+			configMap: map[string]interface{}{
+				"inputs": []map[string]interface{}{
+					{
+						"dataset": map[string]interface{}{
+							"namespace": "",
+						},
+					},
+				},
+			},
+			result: ErrInvalidNamespace,
+		},
+		{
+			name: "type invalid name 1 - compact",
+			configMap: map[string]interface{}{
+				"inputs": []map[string]interface{}{
+					{"dataset.type": "-invalidstart"},
+				},
+			},
+			result: ErrInvalidIndex,
+		},
+		{
+			name: "type invalid combined length 1 - compact",
+			configMap: map[string]interface{}{
+				"inputs": []map[string]interface{}{
+					{
+						"dataset.type":      semiLongString,
+						"dataset.namespace": semiLongString,
+						"streams":           []map[string]interface{}{{"dataset.name": semiLongString}},
+					},
+				},
+			},
+			result: ErrInvalidIndex,
+		},
+		{
+			name: "type invalid type length 1 - compact",
+			configMap: map[string]interface{}{
+				"inputs": []map[string]interface{}{
+					{"dataset.type": longString},
+				},
+			},
+			result: ErrInvalidIndex,
+		},
+
+		{
+			name: "type invalid namespace length 1 - compact",
+			configMap: map[string]interface{}{
+				"inputs": []map[string]interface{}{
+					{"dataset.namespace": longString},
+				},
+			},
+			result: ErrInvalidNamespace,
+		},
+
+		{
+			name: "type invalid dataset.name length 1 - compact",
+			configMap: map[string]interface{}{
+				"inputs": []map[string]interface{}{
+					{"streams": []map[string]interface{}{{"dataset.name": longString}}},
+				},
+			},
+			result: ErrInvalidDataset,
+		},
+
+		{
+			name: "type empty streams - compact",
+			configMap: map[string]interface{}{
+				"inputs": []map[string]interface{}{
+					{"streams": []map[string]interface{}{}},
+				},
+			},
+			result: nil,
+		},
+	}
+
+	log, err := logger.New("")
+	assert.NoError(t, err)
+
+	for _, tc := range testCases {
+		t.Run(tc.name, func(t *testing.T) {
+			ast, err := transpiler.NewAST(tc.configMap)
+			assert.NoError(t, err)
+
+			result := StreamChecker(log, ast)
+			assert.Equal(t, tc.result, result)
+		})
+	}
+}
diff --git a/x-pack/elastic-agent/pkg/agent/application/local_mode.go b/x-pack/elastic-agent/pkg/agent/application/local_mode.go
index db6bcc607387..b565a7ed8a7b 100644
--- a/x-pack/elastic-agent/pkg/agent/application/local_mode.go
+++ b/x-pack/elastic-agent/pkg/agent/application/local_mode.go
@@ -103,7 +103,15 @@ func newLocal(
 	localApplication.router = router
 
 	discover := discoverer(pathConfigFile, c.Management.Path)
-	emit := emitter(log, router, &configModifiers{Decorators: []decoratorFunc{injectMonitoring}, Filters: []filterFunc{filters.ConstraintFilter}}, monitor)
+	emit := emitter(
+		log,
+		router,
+		&configModifiers{
+			Decorators: []decoratorFunc{injectMonitoring},
+			Filters:    []filterFunc{filters.StreamChecker, filters.ConstraintFilter},
+		},
+		monitor,
+	)
 
 	var cfgSource source
 	if !c.Management.Reload.Enabled {
diff --git a/x-pack/elastic-agent/pkg/agent/application/managed_mode.go b/x-pack/elastic-agent/pkg/agent/application/managed_mode.go
index fd43fb397824..1c8b71f1e5b1 100644
--- a/x-pack/elastic-agent/pkg/agent/application/managed_mode.go
+++ b/x-pack/elastic-agent/pkg/agent/application/managed_mode.go
@@ -148,7 +148,7 @@ func newManaged(
 		router,
 		&configModifiers{
 			Decorators: []decoratorFunc{injectMonitoring},
-			Filters:    []filterFunc{injectFleet(config), filters.ConstraintFilter},
+			Filters:    []filterFunc{filters.StreamChecker, injectFleet(config), filters.ConstraintFilter},
 		},
 		monitor,
 	)