diff --git a/.gitignore b/.gitignore
index c73997313..224784804 100644
--- a/.gitignore
+++ b/.gitignore
@@ -1,2 +1,3 @@
 .idea/
-temp/
\ No newline at end of file
+temp/
+fuzz
diff --git a/code/go/internal/spectypes/contenttype.go b/code/go/internal/spectypes/contenttype.go
new file mode 100644
index 000000000..f64955c36
--- /dev/null
+++ b/code/go/internal/spectypes/contenttype.go
@@ -0,0 +1,76 @@
+// 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 spectypes
+
+import (
+	"encoding/json"
+	"fmt"
+	"mime"
+
+	"gopkg.in/yaml.v3"
+)
+
+// ContentType contains a content media type with its parameters.
+type ContentType struct {
+	MediaType string
+	Params    map[string]string
+}
+
+// Ensure ContentType implements these interfaces.
+var (
+	_ json.Marshaler   = new(ContentType)
+	_ json.Unmarshaler = new(ContentType)
+	_ yaml.Marshaler   = new(ContentType)
+	_ yaml.Unmarshaler = new(ContentType)
+)
+
+// MarshalJSON implements the json.Marshaler interface for ContentType. Returned
+// value is a string representation of the content media type and its parameters.
+func (t ContentType) MarshalJSON() ([]byte, error) {
+	return []byte(`"` + t.String() + `"`), nil
+}
+
+// MarshalYAML implements the json.Marshaler interface for ContentType. Returned
+// value is a string representation of the content media type and its parameters.
+func (t ContentType) MarshalYAML() (interface{}, error) {
+	return t.String(), nil
+}
+
+// UnmarshalJSON implements the json.Marshaler interface for ContentType.
+func (t *ContentType) UnmarshalJSON(d []byte) error {
+	var raw string
+	err := json.Unmarshal(d, &raw)
+	if err != nil {
+		return err
+	}
+
+	return t.unmarshalString(raw)
+}
+
+// UnmarshalYAML implements the yaml.Marshaler interface for ContentType.
+func (t *ContentType) UnmarshalYAML(value *yaml.Node) error {
+	// For some reason go-yaml doesn't like the UnmarshalJSON function above.
+	return t.unmarshalString(value.Value)
+}
+
+func (t *ContentType) unmarshalString(text string) error {
+	mediatype, params, err := mime.ParseMediaType(text)
+	if err != nil {
+		return err
+	}
+	if mime.FormatMediaType(mediatype, params) == "" {
+		// Bug in mime library? Happens when parsing something like "0;*0=0"
+		return fmt.Errorf("invalid token in media type")
+	}
+
+	t.MediaType = mediatype
+	t.Params = params
+	return nil
+}
+
+// String returns a string representation of the content type.
+func (t ContentType) String() string {
+	return mime.FormatMediaType(t.MediaType, t.Params)
+}
diff --git a/code/go/internal/spectypes/contenttype_test.go b/code/go/internal/spectypes/contenttype_test.go
new file mode 100644
index 000000000..fbcfa734e
--- /dev/null
+++ b/code/go/internal/spectypes/contenttype_test.go
@@ -0,0 +1,120 @@
+// 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 spectypes
+
+import (
+	"encoding/json"
+	"testing"
+
+	"gopkg.in/yaml.v3"
+
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+)
+
+func TestContentTypeMarshalJSON(t *testing.T) {
+	jsonContentType := ContentType{"application/json", nil}
+	yamlContentType := ContentType{"application/x-yaml", map[string]string{
+		"require-document-dashes": "true",
+	}}
+
+	cases := []struct {
+		contentType ContentType
+		expected    string
+	}{
+		{ContentType{}, `""`},
+		{jsonContentType, `"application/json"`},
+		{yamlContentType, `"application/x-yaml; require-document-dashes=true"`},
+	}
+
+	for _, c := range cases {
+		t.Run(c.expected, func(t *testing.T) {
+			d, err := json.Marshal(c.contentType)
+			require.NoError(t, err)
+			assert.Equal(t, c.expected, string(d))
+		})
+	}
+}
+
+func TestContentTypeMarshalYAML(t *testing.T) {
+	jsonContentType := ContentType{"application/json", nil}
+	yamlContentType := ContentType{"application/x-yaml", map[string]string{
+		"require-document-dashes": "true",
+	}}
+
+	cases := []struct {
+		contentType ContentType
+		expected    string
+	}{
+		{ContentType{}, "\"\"\n"},
+		{jsonContentType, "application/json\n"},
+		{yamlContentType, "application/x-yaml; require-document-dashes=true\n"},
+	}
+
+	for _, c := range cases {
+		t.Run(c.expected, func(t *testing.T) {
+			d, err := yaml.Marshal(c.contentType)
+			require.NoError(t, err)
+			assert.Equal(t, c.expected, string(d))
+		})
+	}
+}
+
+func TestContentTypeUnmarshal(t *testing.T) {
+	t.Run("json", func(t *testing.T) {
+		testContentTypeUnmarshalFormat(t, json.Unmarshal)
+	})
+	t.Run("yaml", func(t *testing.T) {
+		testContentTypeUnmarshalFormat(t, yaml.Unmarshal)
+	})
+}
+
+func testContentTypeUnmarshalFormat(t *testing.T, unmarshaler func([]byte, interface{}) error) {
+	cases := []struct {
+		json           string
+		expectedType   string
+		expectedParams map[string]string
+		valid          bool
+	}{
+		{`"application/json"`, "application/json", nil, true},
+		{
+			`"application/x-yaml; require-document-dashes=true"`,
+			"application/x-yaml",
+			map[string]string{"require-document-dashes": "true"},
+			true,
+		},
+		{
+			`"application/x-yaml; require-document-dashes=true; charset=utf-8"`,
+			"application/x-yaml",
+			map[string]string{
+				"require-document-dashes": "true",
+				"charset":                 "utf-8",
+			},
+			true,
+		},
+		{`"application`, "", nil, false},
+		{`""`, "", nil, false},
+		{`"application/json; charset"`, "", nil, false},
+	}
+
+	for _, c := range cases {
+		t.Run(c.json, func(t *testing.T) {
+			var found ContentType
+			err := unmarshaler([]byte(c.json), &found)
+			if c.valid {
+				require.NoError(t, err)
+				assert.Equal(t, c.expectedType, found.MediaType)
+				if len(c.expectedParams) == 0 {
+					assert.Empty(t, found.Params)
+				} else {
+					assert.EqualValues(t, c.expectedParams, found.Params)
+				}
+			} else {
+				t.Log(found)
+				require.Error(t, err)
+			}
+		})
+	}
+}
diff --git a/code/go/internal/spectypes/filesize.go b/code/go/internal/spectypes/filesize.go
new file mode 100644
index 000000000..3fbd03b2d
--- /dev/null
+++ b/code/go/internal/spectypes/filesize.go
@@ -0,0 +1,133 @@
+// 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 spectypes
+
+import (
+	"encoding/json"
+	"fmt"
+	"regexp"
+	"strconv"
+
+	"gopkg.in/yaml.v3"
+)
+
+// Common units for file sizes.
+const (
+	Byte     = FileSize(1)
+	KiloByte = 1024 * Byte
+	MegaByte = 1024 * KiloByte
+)
+
+const (
+	byteString     = "B"
+	kiloByteString = "KB"
+	megaByteString = "MB"
+)
+
+// FileSize represents the size of a file.
+type FileSize uint64
+
+// Ensure FileSize implements these interfaces.
+var (
+	_ json.Marshaler   = new(FileSize)
+	_ json.Unmarshaler = new(FileSize)
+	_ yaml.Marshaler   = new(FileSize)
+	_ yaml.Unmarshaler = new(FileSize)
+)
+
+func parseFileSizeInt(s string) (uint64, error) {
+	// os.FileInfo reports size as int64, don't support bigger numbers.
+	maxBitSize := 63
+	return strconv.ParseUint(s, 10, maxBitSize)
+}
+
+// MarshalJSON implements the json.Marshaler interface for FileSize, it returns
+// the string representation in a format that can be unmarshaled back to an
+// equivalent value.
+func (s FileSize) MarshalJSON() ([]byte, error) {
+	return []byte(`"` + s.String() + `"`), nil
+}
+
+// MarshalYAML implements the yaml.Marshaler interface for FileSize, it returns
+// the string representation in a format that can be unmarshaled back to an
+// equivalent value.
+func (s FileSize) MarshalYAML() (interface{}, error) {
+	return s.String(), nil
+}
+
+// UnmarshalJSON implements the json.Unmarshaler interface for FileSize.
+func (s *FileSize) UnmarshalJSON(d []byte) error {
+	// Support unquoted plain numbers.
+	bytes, err := parseFileSizeInt(string(d))
+	if err == nil {
+		*s = FileSize(bytes)
+		return nil
+	}
+
+	var text string
+	err = json.Unmarshal(d, &text)
+	if err != nil {
+		return err
+	}
+
+	return s.unmarshalString(text)
+}
+
+// UnmarshalYAML implements the yaml.Unmarshaler interface for FileSize.
+func (s *FileSize) UnmarshalYAML(value *yaml.Node) error {
+	// Support unquoted plain numbers.
+	bytes, err := parseFileSizeInt(value.Value)
+	if err == nil {
+		*s = FileSize(bytes)
+		return nil
+	}
+
+	return s.unmarshalString(value.Value)
+}
+
+var bytesPattern = regexp.MustCompile(fmt.Sprintf(`^(\d+)(%s|%s|%s|)$`, byteString, kiloByteString, megaByteString))
+
+func (s *FileSize) unmarshalString(text string) error {
+	match := bytesPattern.FindStringSubmatch(text)
+	if len(match) < 3 {
+		return fmt.Errorf("invalid format for file size (%s)", text)
+	}
+
+	q, err := parseFileSizeInt(match[1])
+	if err != nil {
+		return fmt.Errorf("invalid format for file size (%s): %w", text, err)
+	}
+
+	unit := match[2]
+	switch unit {
+	case megaByteString:
+		*s = FileSize(q) * MegaByte
+	case kiloByteString:
+		*s = FileSize(q) * KiloByte
+	case byteString, "":
+		*s = FileSize(q) * Byte
+	default:
+		return fmt.Errorf("invalid unit for filesize (%s): %s", text, unit)
+	}
+
+	return nil
+}
+
+// String returns the string representation of the FileSize.
+func (s FileSize) String() string {
+	format := func(q FileSize, unit string) string {
+		return fmt.Sprintf("%d%s", q, unit)
+	}
+
+	if s >= MegaByte && (s%MegaByte == 0) {
+		return format(s/MegaByte, megaByteString)
+	}
+
+	if s >= KiloByte && (s%KiloByte == 0) {
+		return format(s/KiloByte, kiloByteString)
+	}
+
+	return format(s, byteString)
+}
diff --git a/code/go/internal/spectypes/filesize_test.go b/code/go/internal/spectypes/filesize_test.go
new file mode 100644
index 000000000..84ae95fe7
--- /dev/null
+++ b/code/go/internal/spectypes/filesize_test.go
@@ -0,0 +1,99 @@
+// 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 spectypes
+
+import (
+	"encoding/json"
+	"testing"
+
+	"gopkg.in/yaml.v3"
+
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+)
+
+func TestFileSizeMarshallJSON(t *testing.T) {
+	cases := []struct {
+		fileSize FileSize
+		expected string
+	}{
+		{FileSize(0), `"0B"`},
+		{FileSize(1024), `"1KB"`},
+		{FileSize(1025), `"1025B"`},
+		{5 * MegaByte, `"5MB"`},
+	}
+
+	for _, c := range cases {
+		t.Run(c.expected, func(t *testing.T) {
+			d, err := json.Marshal(c.fileSize)
+			require.NoError(t, err)
+			assert.Equal(t, c.expected, string(d))
+		})
+	}
+}
+
+func TestFileSizeMarshallYAML(t *testing.T) {
+	cases := []struct {
+		fileSize FileSize
+		expected string
+	}{
+		{FileSize(0), "0B\n"},
+		{FileSize(1024), "1KB\n"},
+		{FileSize(1025), "1025B\n"},
+		{5 * MegaByte, "5MB\n"},
+	}
+
+	for _, c := range cases {
+		t.Run(c.expected, func(t *testing.T) {
+			d, err := yaml.Marshal(c.fileSize)
+			require.NoError(t, err)
+			assert.Equal(t, c.expected, string(d))
+		})
+	}
+}
+
+func TestFileSizeUnmarshal(t *testing.T) {
+	t.Run("json", func(t *testing.T) {
+		testFileSizeUnmarshalFormat(t, json.Unmarshal)
+	})
+	t.Run("yaml", func(t *testing.T) {
+		testFileSizeUnmarshalFormat(t, yaml.Unmarshal)
+	})
+}
+
+func testFileSizeUnmarshalFormat(t *testing.T, unmarshaler func([]byte, interface{}) error) {
+	cases := []struct {
+		json     string
+		expected FileSize
+		valid    bool
+	}{
+		{"0", 0, true},
+		{"1024", 1024 * Byte, true},
+		{`"1024"`, 1024 * Byte, true},
+		{`"1024B"`, 1024 * Byte, true},
+		{`"10MB"`, 10 * MegaByte, true},
+		{`"2KB"`, 2 * KiloByte, true},
+		{`"KB"`, 0, false},
+		{`"1s"`, 0, false},
+		{`""`, 0, false},
+		{`"B"`, 0, false},
+		{`"-200MB"`, 0, false},
+		{`"-1"`, 0, false},
+		{`"10000000000000000000MB"`, 0, false},
+	}
+
+	for _, c := range cases {
+		t.Run(c.json, func(t *testing.T) {
+			var found FileSize
+			err := unmarshaler([]byte(c.json), &found)
+			if c.valid {
+				require.NoError(t, err)
+				assert.Equal(t, c.expected, found)
+			} else {
+				require.Error(t, err)
+			}
+		})
+	}
+}
diff --git a/code/go/internal/spectypes/spectypes_test.go b/code/go/internal/spectypes/spectypes_test.go
new file mode 100644
index 000000000..1e59426e7
--- /dev/null
+++ b/code/go/internal/spectypes/spectypes_test.go
@@ -0,0 +1,70 @@
+// 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.
+
+//go:build go1.18
+
+package spectypes
+
+import (
+	"encoding/json"
+	"testing"
+
+	"github.com/stretchr/testify/require"
+)
+
+func FuzzContentTypeMarshalling(f *testing.F) {
+	f.Add(`"application/json"`)
+	f.Add(`"application/x-yaml; require-document-dashes=true"`)
+	f.Add(`"application/x-yaml; require-document-dashes=true; charset=utf-8"`)
+
+	f.Fuzz(func(t *testing.T, contentType string) {
+		t.Log("original: " + contentType)
+
+		var first, second ContentType
+		err := json.Unmarshal([]byte(contentType), &first)
+		if err != nil {
+			return
+		}
+
+		t.Logf("first: (%s)", first)
+		d, err := json.Marshal(first)
+		require.NoError(t, err)
+
+		err = json.Unmarshal(d, &second)
+		require.NoError(t, err)
+
+		t.Logf("second: (%s)", second)
+		require.Equal(t, first.MediaType, second.MediaType)
+		require.EqualValues(t, first.Params, second.Params)
+	})
+}
+
+func FuzzFileSizeMarshalling(f *testing.F) {
+	f.Add(`0`)
+	f.Add(`1024`)
+	f.Add(`"0B"`)
+	f.Add(`"1KB"`)
+	f.Add(`"1025B"`)
+	f.Add(`"5MB"`)
+
+	f.Fuzz(func(t *testing.T, fileSize string) {
+		t.Log("original: " + fileSize)
+
+		var first, second FileSize
+		err := json.Unmarshal([]byte(fileSize), &first)
+		if err != nil {
+			return
+		}
+
+		t.Logf("first: (%s)", first)
+		d, err := json.Marshal(first)
+		require.NoError(t, err)
+
+		err = json.Unmarshal(d, &second)
+		require.NoError(t, err)
+
+		t.Logf("second: (%s)", second)
+		require.Equal(t, first, second)
+	})
+}
diff --git a/code/go/internal/validator/common_spec.go b/code/go/internal/validator/common_spec.go
index 2348349b5..b049cf080 100644
--- a/code/go/internal/validator/common_spec.go
+++ b/code/go/internal/validator/common_spec.go
@@ -5,14 +5,51 @@
 package validator
 
 import (
+	"reflect"
+
 	"github.com/creasty/defaults"
 	"github.com/pkg/errors"
+
+	"github.com/elastic/package-spec/code/go/internal/spectypes"
 )
 
 type commonSpec struct {
 	AdditionalContents bool             `yaml:"additionalContents"`
 	Contents           []folderItemSpec `yaml:"contents"`
 	DevelopmentFolder  bool             `yaml:"developmentFolder"`
+
+	Limits commonSpecLimits `yaml:",inline"`
+}
+
+type commonSpecLimits struct {
+	// Limit to the total number of elements in a directory.
+	TotalContentsLimit int `yaml:"totalContentsLimit"`
+
+	// Limit to the total size of files in a directory.
+	TotalSizeLimit spectypes.FileSize `yaml:"totalSizeLimit"`
+
+	// Limit to individual files.
+	SizeLimit spectypes.FileSize `yaml:"sizeLimit"`
+
+	// Limit to individual configuration files (yaml files).
+	ConfigurationSizeLimit spectypes.FileSize `yaml:"configurationSizeLimit"`
+
+	// Limit to files referenced as relative paths (images).
+	RelativePathSizeLimit spectypes.FileSize `yaml:"relativePathSizeLimit"`
+
+	// Maximum number of fields per data stream, can only be set at the root level spec.
+	FieldsPerDataStreamLimit int `yaml:"fieldsPerDataStreamLimit"`
+}
+
+func (l *commonSpecLimits) update(o commonSpecLimits) {
+	target := reflect.ValueOf(l).Elem()
+	source := reflect.ValueOf(&o).Elem()
+	for i := 0; i < target.NumField(); i++ {
+		field := target.Field(i)
+		if field.IsZero() {
+			field.Set(source.Field(i))
+		}
+	}
 }
 
 func setDefaultValues(spec *commonSpec) error {
@@ -31,5 +68,14 @@ func setDefaultValues(spec *commonSpec) error {
 			return err
 		}
 	}
+
 	return nil
 }
+
+func propagateContentLimits(spec *commonSpec) {
+	for i := range spec.Contents {
+		content := &spec.Contents[i].commonSpec
+		content.Limits.update(spec.Limits)
+		propagateContentLimits(content)
+	}
+}
diff --git a/code/go/internal/validator/folder_item_content.go b/code/go/internal/validator/folder_item_content.go
index c059337b3..0cc09d222 100644
--- a/code/go/internal/validator/folder_item_content.go
+++ b/code/go/internal/validator/folder_item_content.go
@@ -9,59 +9,115 @@ import (
 	"encoding/json"
 	"fmt"
 	"io/fs"
-	"mime"
 
 	"github.com/pkg/errors"
 	"gopkg.in/yaml.v3"
+
+	ve "github.com/elastic/package-spec/code/go/internal/errors"
+	"github.com/elastic/package-spec/code/go/internal/spectypes"
 )
 
-func loadItemContent(fsys fs.FS, itemPath, mediaType string) ([]byte, error) {
-	itemData, err := fs.ReadFile(fsys, itemPath)
+func loadItemSchema(fsys fs.FS, path string, contentType *spectypes.ContentType) ([]byte, error) {
+	data, err := fs.ReadFile(fsys, path)
 	if err != nil {
-		return nil, errors.Wrap(err, "reading item file failed")
+		return nil, ve.ValidationErrors{errors.Wrap(err, "reading item file failed")}
+	}
+	if contentType != nil && contentType.MediaType == "application/x-yaml" {
+		return convertYAMLToJSON(data)
 	}
+	return data, nil
+}
 
-	// There might be a situation in which we'd like to keep a directory without any file content under version control.
-	// Usually it can be solved by adding the .empty file. There is no media type defined for this file.
-	// It's expected of the file with media type defined not to be empty - the file can be marked as non-required.
-	if len(itemData) == 0 && mediaType != "" {
-		return nil, errors.New("file is empty, but media type is defined")
+func validateContentType(fsys fs.FS, path string, contentType spectypes.ContentType) error {
+	switch contentType.MediaType {
+	case "application/x-yaml":
+		v, _ := contentType.Params["require-document-dashes"]
+		requireDashes := (v == "true")
+		if requireDashes {
+			err := validateYAMLDashes(fsys, path)
+			if err != nil {
+				return err
+			}
+		}
+	case "application/json":
+	case "text/markdown":
+	case "text/plain":
+	default:
+		return fmt.Errorf("unsupported media type (%s)", contentType)
 	}
+	return nil
+}
 
-	if mediaType == "" {
-		return itemData, nil // no item's schema defined
+func validateYAMLDashes(fsys fs.FS, path string) error {
+	f, err := fsys.Open(path)
+	if err != nil {
+		return err
+	}
+	defer f.Close()
+
+	dashes := []byte("---\n")
+	b := make([]byte, len(dashes))
+	_, err = f.Read(b)
+	if err != nil {
+		return err
 	}
 
-	basicMediaType, params, err := mime.ParseMediaType(mediaType)
+	if !bytes.Equal(dashes, b) {
+		return errors.New("document dashes are required (start the document with '---')")
+	}
+	return nil
+}
+
+func validateContentTypeSize(fsys fs.FS, path string, contentType spectypes.ContentType, limits commonSpecLimits) error {
+	info, err := fs.Stat(fsys, path)
 	if err != nil {
-		return nil, errors.Wrapf(err, "invalid media type (%s)", mediaType)
+		return err
+	}
+	size := spectypes.FileSize(info.Size())
+	if size <= 0 {
+		return errors.New("file is empty, but media type is defined")
 	}
 
-	switch basicMediaType {
+	var sizeLimit spectypes.FileSize
+	switch contentType.MediaType {
 	case "application/x-yaml":
-		// TODO Determine if special handling of `---` is required (issue: https://github.com/elastic/package-spec/pull/54)
-		if v, _ := params["require-document-dashes"]; v == "true" && !bytes.HasPrefix(itemData, []byte("---\n")) {
-			return nil, errors.New("document dashes are required (start the document with '---')")
-		}
+		sizeLimit = limits.ConfigurationSizeLimit
+	}
+	if sizeLimit > 0 && size > sizeLimit {
+		return errors.Errorf("file size (%s) is bigger than expected (%s)", size, sizeLimit)
+	}
+	return nil
+}
 
-		var c interface{}
-		err = yaml.Unmarshal(itemData, &c)
-		if err != nil {
-			return nil, errors.Wrapf(err, "unmarshalling YAML file failed (path: %s)", itemPath)
-		}
-		c = expandItemKey(c)
+func validateMaxSize(fsys fs.FS, path string, limits commonSpecLimits) error {
+	if limits.SizeLimit == 0 {
+		return nil
+	}
 
-		itemData, err = json.Marshal(&c)
-		if err != nil {
-			return nil, errors.Wrapf(err, "converting YAML file to JSON failed (path: %s)", itemPath)
-		}
-	case "application/json": // no need to convert the item content
-	case "text/markdown": // text/markdown can't be transformed into JSON format
-	case "text/plain": // text/plain should be left as-is
-	default:
-		return nil, fmt.Errorf("unsupported media type (%s)", mediaType)
+	info, err := fs.Stat(fsys, path)
+	if err != nil {
+		return err
+	}
+	size := spectypes.FileSize(info.Size())
+	if size > limits.SizeLimit {
+		return errors.Errorf("file size (%s) is bigger than expected (%s)", size, limits.SizeLimit)
+	}
+	return nil
+}
+
+func convertYAMLToJSON(data []byte) ([]byte, error) {
+	var c interface{}
+	err := yaml.Unmarshal(data, &c)
+	if err != nil {
+		return nil, errors.Wrapf(err, "unmarshalling YAML file failed")
+	}
+	c = expandItemKey(c)
+
+	data, err = json.Marshal(&c)
+	if err != nil {
+		return nil, errors.Wrapf(err, "converting YAML to JSON failed")
 	}
-	return itemData, nil
+	return data, nil
 }
 
 func expandItemKey(c interface{}) interface{} {
diff --git a/code/go/internal/validator/folder_item_spec.go b/code/go/internal/validator/folder_item_spec.go
index 2455f0489..fa1e279ae 100644
--- a/code/go/internal/validator/folder_item_spec.go
+++ b/code/go/internal/validator/folder_item_spec.go
@@ -15,20 +15,21 @@ import (
 	"github.com/xeipuuv/gojsonschema"
 
 	ve "github.com/elastic/package-spec/code/go/internal/errors"
+	"github.com/elastic/package-spec/code/go/internal/spectypes"
 	"github.com/elastic/package-spec/code/go/internal/validator/semantic"
 	"github.com/elastic/package-spec/code/go/internal/yamlschema"
 )
 
 type folderItemSpec struct {
-	Description       string   `yaml:"description"`
-	ItemType          string   `yaml:"type"`
-	ContentMediaType  string   `yaml:"contentMediaType"`
-	ForbiddenPatterns []string `yaml:"forbiddenPatterns"`
-	Name              string   `yaml:"name"`
-	Pattern           string   `yaml:"pattern"`
-	Required          bool     `yaml:"required"`
-	Ref               string   `yaml:"$ref"`
-	Visibility        string   `yaml:"visibility" default:"public"`
+	Description       string                 `yaml:"description"`
+	ItemType          string                 `yaml:"type"`
+	ContentMediaType  *spectypes.ContentType `yaml:"contentMediaType"`
+	ForbiddenPatterns []string               `yaml:"forbiddenPatterns"`
+	Name              string                 `yaml:"name"`
+	Pattern           string                 `yaml:"pattern"`
+	Required          bool                   `yaml:"required"`
+	Ref               string                 `yaml:"$ref"`
+	Visibility        string                 `yaml:"visibility" default:"public"`
 	commonSpec        `yaml:",inline"`
 }
 
@@ -56,13 +57,13 @@ func (s *folderItemSpec) matchingFileExists(files []fs.DirEntry) (bool, error) {
 	return false, nil
 }
 
-// sameFileChecker is the interface that parameters of isSameType should implement,
+// sameTypeChecker is the interface that parameters of isSameType should implement,
 // this is intended to accept both fs.DirEntry and fs.FileInfo.
-type sameFileChecker interface {
+type sameTypeChecker interface {
 	IsDir() bool
 }
 
-func (s *folderItemSpec) isSameType(file sameFileChecker) bool {
+func (s *folderItemSpec) isSameType(file sameTypeChecker) bool {
 	switch s.ItemType {
 	case itemTypeFile:
 		return !file.IsDir()
@@ -73,22 +74,43 @@ func (s *folderItemSpec) isSameType(file sameFileChecker) bool {
 	return false
 }
 
-func (s *folderItemSpec) validate(schemaFS fs.FS, fsys fs.FS, folderSpecPath string, itemPath string) ve.ValidationErrors {
-	// loading item content
-	itemData, err := loadItemContent(fsys, itemPath, s.ContentMediaType)
+func (s *folderItemSpec) validate(schemaFsys fs.FS, fsys fs.FS, folderSpecPath string, itemPath string) ve.ValidationErrors {
+	err := validateMaxSize(fsys, itemPath, s.Limits)
 	if err != nil {
 		return ve.ValidationErrors{err}
 	}
+	if s.ContentMediaType != nil {
+		err := validateContentType(fsys, itemPath, *s.ContentMediaType)
+		if err != nil {
+			return ve.ValidationErrors{err}
+		}
+		err = validateContentTypeSize(fsys, itemPath, *s.ContentMediaType, s.Limits)
+		if err != nil {
+			return ve.ValidationErrors{err}
+		}
+	}
+
+	errs := s.validateSchema(schemaFsys, fsys, folderSpecPath, itemPath)
+	if len(errs) > 0 {
+		return errs
+	}
+
+	return nil
+}
 
-	var schemaLoader gojsonschema.JSONLoader
-	if s.Ref != "" {
-		schemaPath := filepath.Join(filepath.Dir(folderSpecPath), s.Ref)
-		schemaLoader = yamlschema.NewReferenceLoaderFileSystem("file:///"+schemaPath, schemaFS)
-	} else {
+func (s *folderItemSpec) validateSchema(schemaFsys fs.FS, fsys fs.FS, folderSpecPath, itemPath string) ve.ValidationErrors {
+	if s.Ref == "" {
 		return nil // item's schema is not defined
 	}
 
+	schemaPath := filepath.Join(filepath.Dir(folderSpecPath), s.Ref)
+	schemaLoader := yamlschema.NewReferenceLoaderFileSystem("file:///"+schemaPath, schemaFsys)
+
 	// validation with schema
+	itemData, err := loadItemSchema(fsys, itemPath, s.ContentMediaType)
+	if err != nil {
+		return ve.ValidationErrors{err}
+	}
 	documentLoader := gojsonschema.NewBytesLoader(itemData)
 
 	formatCheckersMutex.Lock()
@@ -98,20 +120,20 @@ func (s *folderItemSpec) validate(schemaFS fs.FS, fsys fs.FS, folderSpecPath str
 		formatCheckersMutex.Unlock()
 	}()
 
-	semantic.LoadRelativePathFormatChecker(fsys, filepath.Dir(itemPath))
+	semantic.LoadRelativePathFormatChecker(fsys, filepath.Dir(itemPath), s.Limits.RelativePathSizeLimit)
 	semantic.LoadDataStreamNameFormatChecker(fsys, filepath.Dir(itemPath))
 	result, err := gojsonschema.Validate(schemaLoader, documentLoader)
 	if err != nil {
 		return ve.ValidationErrors{err}
 	}
 
-	if result.Valid() {
-		return nil // item content is valid according to the loaded schema
+	if !result.Valid() {
+		var errs ve.ValidationErrors
+		for _, re := range result.Errors() {
+			errs = append(errs, fmt.Errorf("field %s: %s", re.Field(), adjustErrorDescription(re.Description())))
+		}
+		return errs
 	}
 
-	var errs ve.ValidationErrors
-	for _, re := range result.Errors() {
-		errs = append(errs, fmt.Errorf("field %s: %s", re.Field(), adjustErrorDescription(re.Description())))
-	}
-	return errs
+	return nil // item content is valid according to the loaded schema
 }
diff --git a/code/go/internal/validator/folder_item_spec_errors.go b/code/go/internal/validator/folder_item_spec_errors.go
index a97e177ce..90384f38c 100644
--- a/code/go/internal/validator/folder_item_spec_errors.go
+++ b/code/go/internal/validator/folder_item_spec_errors.go
@@ -4,11 +4,15 @@
 
 package validator
 
-import "github.com/elastic/package-spec/code/go/internal/validator/semantic"
+import (
+	"fmt"
+
+	"github.com/elastic/package-spec/code/go/internal/validator/semantic"
+)
 
 func adjustErrorDescription(description string) string {
 	if description == "Does not match format '"+semantic.RelativePathFormat+"'" {
-		return "relative path is invalid or target doesn't exist"
+		return fmt.Sprintf("relative path is invalid, target doesn't exist or it exceeds the file size limit")
 	} else if description == "Does not match format '"+semantic.DataStreamNameFormat+"'" {
 		return "data stream doesn't exist"
 	}
diff --git a/code/go/internal/validator/folder_spec.go b/code/go/internal/validator/folder_spec.go
index 5bb98f31c..53e24e41a 100644
--- a/code/go/internal/validator/folder_spec.go
+++ b/code/go/internal/validator/folder_spec.go
@@ -17,6 +17,7 @@ import (
 
 	ve "github.com/elastic/package-spec/code/go/internal/errors"
 	"github.com/elastic/package-spec/code/go/internal/fspath"
+	"github.com/elastic/package-spec/code/go/internal/spectypes"
 )
 
 const (
@@ -31,39 +32,43 @@ type folderSpec struct {
 	fs       fs.FS
 	specPath string
 	commonSpec
+
+	// These "validation-time" fields don't actually belong to the spec, storing
+	// them here for convenience by now.
+	totalSize     spectypes.FileSize
+	totalContents int
 }
 
-func newFolderSpec(fs fs.FS, specPath string) (*folderSpec, error) {
+func (s *folderSpec) load(fs fs.FS, specPath string) error {
 	specFile, err := fs.Open(specPath)
 	if err != nil {
-		return nil, errors.Wrap(err, "could not open folder specification file")
+		return errors.Wrap(err, "could not open folder specification file")
 	}
 	defer specFile.Close()
 
 	data, err := ioutil.ReadAll(specFile)
 	if err != nil {
-		return nil, errors.Wrap(err, "could not read folder specification file")
+		return errors.Wrap(err, "could not read folder specification file")
 	}
 
 	var wrapper struct {
-		Spec commonSpec `yaml:"spec"`
+		Spec *commonSpec `yaml:"spec"`
 	}
-
+	wrapper.Spec = &s.commonSpec
 	if err := yaml.Unmarshal(data, &wrapper); err != nil {
-		return nil, errors.Wrap(err, "could not parse folder specification file")
-	}
-
-	spec := folderSpec{
-		fs:         fs,
-		specPath:   specPath,
-		commonSpec: wrapper.Spec,
+		return errors.Wrap(err, "could not parse folder specification file")
 	}
 
-	err = setDefaultValues(&spec.commonSpec)
+	err = setDefaultValues(&s.commonSpec)
 	if err != nil {
-		return nil, errors.Wrap(err, "could not set default values")
+		return errors.Wrap(err, "could not set default values")
 	}
-	return &spec, nil
+
+	propagateContentLimits(&s.commonSpec)
+
+	s.fs = fs
+	s.specPath = specPath
+	return nil
 }
 
 func (s *folderSpec) validate(packageName string, fsys fspath.FS, path string) ve.ValidationErrors {
@@ -74,6 +79,13 @@ func (s *folderSpec) validate(packageName string, fsys fspath.FS, path string) v
 		return errs
 	}
 
+	// This is not taking into account if the folder is for development. Enforce
+	// this limit in all cases to avoid having to read too many files.
+	if contentsLimit := s.Limits.TotalContentsLimit; contentsLimit > 0 && len(files) > contentsLimit {
+		errs = append(errs, errors.Errorf("folder [%s] exceeds the limit of %d files", fsys.Path(path), contentsLimit))
+		return errs
+	}
+
 	for _, file := range files {
 		fileName := file.Name()
 		itemSpec, err := s.findItemSpec(packageName, fileName)
@@ -116,23 +128,21 @@ func (s *folderSpec) validate(packageName string, fsys fspath.FS, path string) v
 				continue
 			}
 
-			var subFolderSpec *folderSpec
+			var subFolderSpec folderSpec
+			// Inherit limits from parent directory.
+			subFolderSpec.Limits = s.Limits
 			if itemSpec.Ref != "" {
 				subFolderSpecPath := filepath.Join(filepath.Dir(s.specPath), itemSpec.Ref)
-				subFolderSpec, err = newFolderSpec(s.fs, subFolderSpecPath)
+				err := subFolderSpec.load(s.fs, subFolderSpecPath)
 				if err != nil {
 					errs = append(errs, err)
 					continue
 				}
 			} else if itemSpec.Contents != nil {
-				subFolderSpec = &folderSpec{
-					fs:       s.fs,
-					specPath: s.specPath,
-					commonSpec: commonSpec{
-						AdditionalContents: itemSpec.AdditionalContents,
-						Contents:           itemSpec.Contents,
-					},
-				}
+				subFolderSpec.fs = s.fs
+				subFolderSpec.specPath = s.specPath
+				subFolderSpec.commonSpec.AdditionalContents = itemSpec.AdditionalContents
+				subFolderSpec.commonSpec.Contents = itemSpec.Contents
 			}
 
 			// Subfolders of development folders are also considered development folders.
@@ -146,6 +156,11 @@ func (s *folderSpec) validate(packageName string, fsys fspath.FS, path string) v
 				errs = append(errs, subErrs...)
 			}
 
+			// Don't count files in development folders.
+			if !subFolderSpec.DevelopmentFolder {
+				s.totalContents += subFolderSpec.totalContents
+				s.totalSize += subFolderSpec.totalSize
+			}
 		} else {
 			if !itemSpec.isSameType(file) {
 				errs = append(errs, fmt.Errorf("[%s] is a file but is expected to be a folder", fsys.Path(fileName)))
@@ -159,9 +174,21 @@ func (s *folderSpec) validate(packageName string, fsys fspath.FS, path string) v
 					errs = append(errs, errors.Wrapf(ive, "file \"%s\" is invalid", fsys.Path(itemPath)))
 				}
 			}
+
+			info, err := fs.Stat(fsys, itemPath)
+			if err != nil {
+				errs = append(errs, errors.Wrapf(err, "failed to obtain file size for \"%s\"", fsys.Path(itemPath)))
+			} else {
+				s.totalContents++
+				s.totalSize += spectypes.FileSize(info.Size())
+			}
 		}
 	}
 
+	if sizeLimit := s.Limits.TotalSizeLimit; sizeLimit > 0 && s.totalSize > sizeLimit {
+		errs = append(errs, errors.Errorf("folder [%s] exceeds the total size limit of %s", fsys.Path(path), sizeLimit))
+	}
+
 	// validate that required items in spec are all accounted for
 	for _, itemSpec := range s.Contents {
 		if !itemSpec.Required {
diff --git a/code/go/internal/validator/semantic/format_checkers.go b/code/go/internal/validator/semantic/format_checkers.go
index 4280fc823..af99aca9c 100644
--- a/code/go/internal/validator/semantic/format_checkers.go
+++ b/code/go/internal/validator/semantic/format_checkers.go
@@ -9,6 +9,8 @@ import (
 	"path/filepath"
 
 	"github.com/xeipuuv/gojsonschema"
+
+	"github.com/elastic/package-spec/code/go/internal/spectypes"
 )
 
 const (
@@ -28,6 +30,7 @@ const (
 type relativePathChecker struct {
 	fsys        fs.FS
 	currentPath string
+	sizeLimit   spectypes.FileSize
 }
 
 // IsFormat method checks if the path exists.
@@ -38,19 +41,28 @@ func (r relativePathChecker) IsFormat(input interface{}) bool {
 	}
 
 	path := filepath.Join(r.currentPath, asString)
-	_, err := fs.Stat(r.fsys, path)
+	info, err := fs.Stat(r.fsys, path)
 	if err != nil {
 		return false
 	}
+
+	// TODO: It happens that we want the same max size for all the files we reference with
+	// relative paths, but it'd be better if we could find a way to parameterize format
+	// checkers so we can configure specific max sizes, and we can provide better feedback.
+	if r.sizeLimit > 0 && spectypes.FileSize(info.Size()) > r.sizeLimit {
+		return false
+	}
+
 	return true
 }
 
 // LoadRelativePathFormatChecker loads the relative-path format checker into the
 // json-schema validation library.
-func LoadRelativePathFormatChecker(fsys fs.FS, currentPath string) {
+func LoadRelativePathFormatChecker(fsys fs.FS, currentPath string, sizeLimit spectypes.FileSize) {
 	gojsonschema.FormatCheckers.Add(RelativePathFormat, relativePathChecker{
 		fsys:        fsys,
 		currentPath: currentPath,
+		sizeLimit:   sizeLimit,
 	})
 }
 
diff --git a/code/go/internal/validator/semantic/validate_fields_limits.go b/code/go/internal/validator/semantic/validate_fields_limits.go
new file mode 100644
index 000000000..429f00f97
--- /dev/null
+++ b/code/go/internal/validator/semantic/validate_fields_limits.go
@@ -0,0 +1,69 @@
+// 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 semantic
+
+import (
+	"fmt"
+	"path/filepath"
+	"strings"
+
+	"github.com/pkg/errors"
+
+	ve "github.com/elastic/package-spec/code/go/internal/errors"
+	"github.com/elastic/package-spec/code/go/internal/fspath"
+)
+
+// ValidateFieldsLimits verifies limits on fields.
+func ValidateFieldsLimits(limit int) func(fspath.FS) ve.ValidationErrors {
+	return func(fsys fspath.FS) ve.ValidationErrors {
+		return validateFieldsLimits(fsys, limit)
+	}
+}
+
+func validateFieldsLimits(fsys fspath.FS, limit int) ve.ValidationErrors {
+	counts := make(map[string]int)
+	countField := func(fieldsFile string, f field) ve.ValidationErrors {
+		if len(f.Fields) > 0 {
+			// Don't count groups
+			return nil
+		}
+
+		dataStream, err := dataStreamFromFieldsPath(fsys.Path(), fieldsFile)
+		if err != nil {
+			return ve.ValidationErrors{err}
+		}
+		count, _ := counts[dataStream]
+		counts[dataStream] = count + 1
+		return nil
+	}
+
+	err := validateFields(fsys, countField)
+	if err != nil {
+		return err
+	}
+
+	var errs ve.ValidationErrors
+	for dataStream, count := range counts {
+		if count > limit {
+			errs = append(errs, errors.Errorf("data stream %s has more than %d fields (%d)", dataStream, limit, count))
+		}
+	}
+	return errs
+}
+
+func dataStreamFromFieldsPath(pkgRoot, fieldsFile string) (string, error) {
+	dataStreamPath := filepath.Clean(filepath.Join(pkgRoot, "data_stream"))
+	relPath, err := filepath.Rel(dataStreamPath, filepath.Clean(fieldsFile))
+	if err != nil {
+		return "", fmt.Errorf("looking for fields file (%s) in data streams path (%s): %w", fieldsFile, dataStreamPath, err)
+	}
+
+	parts := strings.SplitN(relPath, string(filepath.Separator), 2)
+	if len(parts) != 2 {
+		return "", errors.Errorf("could not find data stream for fields file %s", fieldsFile)
+	}
+	dataStream := parts[0]
+	return dataStream, nil
+}
diff --git a/code/go/internal/validator/spec.go b/code/go/internal/validator/spec.go
index c8b57c434..fb74f05c6 100644
--- a/code/go/internal/validator/spec.go
+++ b/code/go/internal/validator/spec.go
@@ -50,8 +50,9 @@ func NewSpec(version semver.Version) (*Spec, error) {
 func (s Spec) ValidatePackage(pkg Package) ve.ValidationErrors {
 	var errs ve.ValidationErrors
 
+	var rootSpec folderSpec
 	rootSpecPath := path.Join(s.specPath, "spec.yml")
-	rootSpec, err := newFolderSpec(s.fs, rootSpecPath)
+	err := rootSpec.load(s.fs, rootSpecPath)
 	if err != nil {
 		errs = append(errs, errors.Wrap(err, "could not read root folder spec file"))
 		return errs
@@ -69,6 +70,7 @@ func (s Spec) ValidatePackage(pkg Package) ve.ValidationErrors {
 		semantic.ValidateVersionIntegrity,
 		semantic.ValidatePrerelease,
 		semantic.ValidateFieldGroups,
+		semantic.ValidateFieldsLimits(rootSpec.Limits.FieldsPerDataStreamLimit),
 		semantic.ValidateDimensionFields,
 	}
 
diff --git a/code/go/pkg/validator/limits_test.go b/code/go/pkg/validator/limits_test.go
new file mode 100644
index 000000000..3775ee526
--- /dev/null
+++ b/code/go/pkg/validator/limits_test.go
@@ -0,0 +1,382 @@
+// 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 validator
+
+import (
+	_ "embed"
+	"fmt"
+	"io"
+	"io/fs"
+	"os"
+	"path/filepath"
+	"strings"
+	"testing"
+	"time"
+
+	"github.com/stretchr/testify/assert"
+
+	"github.com/elastic/package-spec/code/go/internal/spectypes"
+)
+
+func TestLimitsValidation(t *testing.T) {
+	cases := []struct {
+		title string
+		fsys  fs.FS
+		valid bool
+	}{
+		{
+			title: "all good",
+			fsys:  newMockFS().Good(),
+			valid: true,
+		},
+		{
+			title: "configurationSizeLimit exceeded",
+			fsys: newMockFS().Good().Override(func(o *overrideFS) {
+				o.File("manifest.yml").WithSize(10 * spectypes.MegaByte)
+			}),
+			valid: false,
+		},
+		{
+			title: "sizeLimit exceeded",
+			fsys: newMockFS().Good().Override(func(o *overrideFS) {
+				o.File("docs/README.md").WithSize(200 * spectypes.MegaByte)
+			}),
+			valid: false,
+		},
+		{
+			title: "totalSizeLimit exceeded",
+			fsys: newMockFS().Good().WithFiles(
+				newMockFile("docs/other.md").WithSize(140*spectypes.MegaByte),
+				newMockFile("docs/someother.md").WithSize(140*spectypes.MegaByte),
+			),
+			valid: false,
+		},
+		{
+			title: "totalContentsLimit exceeded",
+			fsys: newMockFS().Good().Override(func(o *overrideFS) {
+				o.File("docs").WithGeneratedFiles(70000, ".md", 512*spectypes.Byte)
+			}),
+			valid: false,
+		},
+		{
+			title: "relativePathSizeLimit exceeded",
+			fsys: newMockFS().Good().Override(func(o *overrideFS) {
+				o.File("img/kibana-system.png").WithSize(10 * spectypes.MegaByte)
+			}),
+			valid: false,
+		},
+		{
+			title: "data streams limit exceeded",
+			fsys: newMockFS().Good().Override(func(o *overrideFS) {
+				o.MultiplyFile("data_stream", "foo", 1000)
+			}),
+			valid: false,
+		},
+		{
+			title: "fieldsPerDataStreamLimit exceeded",
+			fsys: newMockFS().Good().WithFiles(
+				newMockFile("data_stream/foo/fields/many-fields.yml").WithContent(generateFields(1500)),
+			),
+			valid: false,
+		},
+		{
+			title: "config template sizeLimit exceeded",
+			fsys: newMockFS().Good().WithFiles(
+				newMockFile("agent/input/stream.yml.hbs").WithSize(6 * spectypes.MegaByte),
+			),
+			valid: false,
+		},
+		{
+			title: "ingest pipeline sizeLimit exceeded",
+			fsys: newMockFS().Good().WithFiles(
+				newMockFile("elasticsearch/ingest_pipeline/pipeline.yml").WithContent("---\n").WithSize(4 * spectypes.MegaByte),
+			),
+			valid: false,
+		},
+		{
+			title: "ignore developer files",
+			fsys: newMockFS().Good().WithFiles(
+				newMockFile("_dev/deploy/docker/entrypoint.sh").WithSize(2048 * spectypes.MegaByte),
+			),
+			valid: true,
+		},
+	}
+
+	for _, c := range cases {
+		t.Run(c.title, func(t *testing.T) {
+			err := ValidateFromFS("test-package", c.fsys)
+			if c.valid {
+				assert.NoError(t, err)
+			} else {
+				assert.Error(t, err)
+			}
+		})
+	}
+}
+
+var _ fs.FS = &mockFS{}
+
+type mockFS struct {
+	root *mockFile
+}
+
+func newMockFS() *mockFS {
+	return &mockFS{root: newMockDir(".")}
+}
+
+func (fs *mockFS) WithFiles(files ...*mockFile) *mockFS {
+	fs.root.WithFiles(files...)
+	return fs
+}
+
+//go:embed testdata/limits/manifest.yml
+var manifestYml string
+
+//go:embed testdata/limits/changelog.yml
+var changelogYml string
+
+//go:embed testdata/limits/data_stream/foo/manifest.yml
+var datastreamManifestYml string
+
+//go:embed testdata/limits/data_stream/foo/fields/base-fields.yml
+var fieldsYml string
+
+func generateFields(n int) string {
+	var buf strings.Builder
+
+	for i := 0; i < n; i++ {
+		buf.WriteString(fmt.Sprintf("- name: generated.foo%d\n", i))
+		buf.WriteString("  type: keyword\n")
+	}
+	return buf.String()
+}
+
+func (fs *mockFS) Good() *mockFS {
+	return fs.WithFiles(
+		newMockFile("manifest.yml").WithContent(manifestYml),
+		newMockFile("changelog.yml").WithContent(changelogYml),
+		newMockFile("docs/README.md").WithContent("## README"),
+		newMockFile("img/kibana-system.png"),
+		newMockFile("img/system.svg"),
+		newMockFile("_dev/deploy/docker/docker-compose.yml").WithContent("version: 2.3"),
+		newMockFile("data_stream/foo/manifest.yml").WithContent(datastreamManifestYml),
+		newMockFile("data_stream/foo/fields/base-fields.yml").WithContent(fieldsYml),
+	)
+}
+
+func (fs *mockFS) Override(overrider func(*overrideFS)) *mockFS {
+	overrider(&overrideFS{fs})
+	return fs
+}
+
+type overrideFS struct {
+	fs *mockFS
+}
+
+func (o *overrideFS) File(name string) *mockFile {
+	f, err := o.fs.root.findFile(name)
+	if err != nil {
+		panic(err)
+	}
+	return f
+}
+
+func (o *overrideFS) MultiplyFile(dir string, name string, times int) {
+	d, err := o.fs.root.findFile(dir)
+	if err != nil {
+		panic(err)
+	}
+
+	f, err := d.findFile(name)
+	if err != nil {
+		panic(err)
+	}
+
+	for i := 0; i < times; i++ {
+		cp := f.Copy()
+		cp.stat.name = fmt.Sprintf("%s%d", f.stat.name, i)
+		d.files = append(d.files, cp)
+	}
+}
+
+func (fs *mockFS) Open(name string) (fs.File, error) {
+	f, err := fs.root.findFile(name)
+	if err != nil {
+		return nil, err
+	}
+	return f.open(), nil
+}
+
+var _ fs.File = &mockFile{}
+var _ fs.ReadDirFile = &mockFile{}
+
+type mockFile struct {
+	stat    mockFileInfo
+	content string
+	reader  io.Reader
+	files   []*mockFile
+}
+
+func newMockFile(name string) *mockFile {
+	return &mockFile{
+		stat: mockFileInfo{
+			name:  name,
+			mode:  0644,
+			isDir: false,
+		},
+	}
+}
+
+func newMockDir(name string) *mockFile {
+	f := newMockFile(name)
+	f.stat.mode = 0755
+	f.stat.isDir = true
+	return f
+}
+
+func (f *mockFile) Copy() *mockFile {
+	cp := mockFile{}
+	cp.stat = f.stat
+	cp.content = f.content
+	for _, src := range f.files {
+		cp.files = append(cp.files, src.Copy())
+	}
+	return &cp
+}
+
+func (f *mockFile) WithContent(content string) *mockFile {
+	if f.stat.isDir {
+		panic("directory cannot have content")
+	}
+	f.content = content
+	f.stat.size = int64(len(content))
+	return f
+}
+
+func (f *mockFile) WithSize(size spectypes.FileSize) *mockFile {
+	f.stat.size = int64(size)
+	return f
+}
+
+func (f *mockFile) WithFiles(files ...*mockFile) *mockFile {
+	if !f.stat.isDir {
+		panic("regular file cannot contain files")
+	}
+	for _, file := range files {
+		f.addFileWithDirs(file)
+	}
+	return f
+}
+
+func (f *mockFile) WithGeneratedFiles(n int, suffix string, size spectypes.FileSize) *mockFile {
+	var files []*mockFile
+	for i := 0; i < n; i++ {
+		files = append(files,
+			newMockFile(fmt.Sprintf("tmp%d%s", i, suffix)).WithSize(size))
+	}
+	f.WithFiles(files...)
+	return f
+}
+
+func (f *mockFile) addFileWithDirs(file *mockFile) {
+	parts := strings.Split(file.stat.name, string(os.PathSeparator))
+	dir := f
+	for i, part := range parts[:len(parts)-1] {
+		d, err := dir.findFile(part)
+		if err == nil {
+			if !d.stat.isDir {
+				panic(strings.Join(parts[:i], string(os.PathSeparator)) + " is not a directory")
+			}
+			dir = d
+		} else {
+			d = newMockDir(part)
+			dir.files = append(dir.files, d)
+			dir = d
+		}
+	}
+	file.stat.name = parts[len(parts)-1]
+	dir.files = append(dir.files, file)
+}
+
+func (f *mockFile) findFile(name string) (*mockFile, error) {
+	if name == "." {
+		return f, nil
+	}
+	name = filepath.Clean(name)
+	parts := strings.SplitN(name, string(os.PathSeparator), 2)
+
+	if len(parts) == 0 {
+		panic("path should not be empty here")
+	}
+
+	var file *mockFile
+	for _, candidate := range f.files {
+		if candidate.stat.name == parts[0] {
+			file = candidate
+			break
+		}
+	}
+
+	if file == nil {
+		return nil, os.ErrNotExist
+	}
+
+	if len(parts) == 2 {
+		return file.findFile(parts[1])
+	}
+	return file, nil
+}
+
+func (f *mockFile) open() *mockFile {
+	var descriptor mockFile
+	descriptor = *f
+	if f.content != "" {
+		descriptor.reader = strings.NewReader(f.content)
+	}
+	return &descriptor
+}
+
+func (f *mockFile) Stat() (fs.FileInfo, error) { return &f.stat, nil }
+func (f *mockFile) Read(d []byte) (int, error) {
+	if f.reader == nil {
+		return 0, io.EOF
+	}
+	return f.reader.Read(d)
+}
+func (f *mockFile) Close() error { return nil }
+
+func (f *mockFile) ReadDir(n int) ([]fs.DirEntry, error) {
+	if !f.stat.isDir {
+		return nil, os.ErrInvalid
+	}
+	var result []fs.DirEntry
+	for i, entry := range f.files {
+		if n > 0 && i >= n {
+			break
+		}
+		result = append(result, &entry.stat)
+	}
+	return result, nil
+}
+
+var _ fs.FileInfo = &mockFileInfo{}
+var _ fs.DirEntry = &mockFileInfo{}
+
+type mockFileInfo struct {
+	name    string
+	size    int64
+	mode    fs.FileMode
+	modTime time.Time
+	isDir   bool
+}
+
+func (fi *mockFileInfo) Info() (fs.FileInfo, error) { return fi, nil }
+func (fi *mockFileInfo) IsDir() bool                { return fi.isDir }
+func (fi *mockFileInfo) Mode() fs.FileMode          { return fi.mode }
+func (fi *mockFileInfo) ModTime() time.Time         { return fi.modTime }
+func (fi *mockFileInfo) Name() string               { return fi.name }
+func (fi *mockFileInfo) Size() int64                { return fi.size }
+func (fi *mockFileInfo) Sys() interface{}           { return nil }
+func (fi *mockFileInfo) Type() fs.FileMode          { return fi.mode.Type() }
diff --git a/code/go/pkg/validator/testdata/limits/changelog.yml b/code/go/pkg/validator/testdata/limits/changelog.yml
new file mode 100644
index 000000000..61f4d434b
--- /dev/null
+++ b/code/go/pkg/validator/testdata/limits/changelog.yml
@@ -0,0 +1,26 @@
+- version: 1.0.0
+  changes:
+    - description: LTS version
+      type: enhancement
+      link: https://github.com/elastic/package-spec/pull/256
+- version: 1.0.0-next
+  changes:
+    - description: release candidate
+      type: enhancement
+      link: https://github.com/elastic/package-spec/pull/172
+- version: 0.1.1
+  changes:
+    - description: Change A
+      type: enhancement
+      link: https://github.com/elastic/package-spec/pull/193
+    - description: Change B
+      type: enhancement
+      link: https://github.com/elastic/package-spec/pull/193
+    - description: Change C
+      type: enhancement
+      link: https://github.com/elastic/package-spec/pull/193
+- version: 0.1.2
+  changes:
+    - description: initial release
+      type: enhancement
+      link: https://github.com/elastic/package-spec/pull/131
diff --git a/code/go/pkg/validator/testdata/limits/data_stream/foo/fields/base-fields.yml b/code/go/pkg/validator/testdata/limits/data_stream/foo/fields/base-fields.yml
new file mode 100644
index 000000000..e6e1d439f
--- /dev/null
+++ b/code/go/pkg/validator/testdata/limits/data_stream/foo/fields/base-fields.yml
@@ -0,0 +1,24 @@
+- name: source
+  title: Source
+  group: 2
+  type: group
+  fields:
+    - name: geo.city_name
+      level: core
+      type: keyword
+      description: City name.
+      ignore_above: 1024
+    - name: geo.location
+      level: core
+      type: geo_point
+      description: Longitude and latitude.
+    - name: geo.region_iso_code
+      level: core
+      type: keyword
+      description: Region ISO code.
+      ignore_above: 1024
+    - name: geo.region_name
+      level: core
+      type: keyword
+      description: Region name.
+      ignore_above: 1024
\ No newline at end of file
diff --git a/code/go/pkg/validator/testdata/limits/data_stream/foo/manifest.yml b/code/go/pkg/validator/testdata/limits/data_stream/foo/manifest.yml
new file mode 100644
index 000000000..7b2aa77a2
--- /dev/null
+++ b/code/go/pkg/validator/testdata/limits/data_stream/foo/manifest.yml
@@ -0,0 +1,35 @@
+title: Nginx access logs
+type: logs
+streams:
+  - input: logfile
+    vars:
+      - name: empty_array
+        type: text
+        title: Empty array
+        multi: true
+        required: false
+        show_user: true
+        default: []
+      - name: paths
+        type: text
+        title: Paths
+        multi: true
+        required: true
+        show_user: true
+        default:
+          - /var/log/nginx/access.log*
+      - name: server_status_path
+        type: text
+        title: Server Status Path
+        multi: false
+        required: true
+        show_user: false
+        default: /server-status
+    title: Nginx access logs
+    description: Collect Nginx access logs
+dataset_is_prefix: true
+elasticsearch.index_template.mappings:
+  a:
+    b: 1
+elasticsearch.index_template.ingest_pipeline.name: foobar
+elasticsearch.privileges.indices: [auto_configure, create_doc, monitor]
diff --git a/code/go/pkg/validator/testdata/limits/manifest.yml b/code/go/pkg/validator/testdata/limits/manifest.yml
new file mode 100644
index 000000000..07d605118
--- /dev/null
+++ b/code/go/pkg/validator/testdata/limits/manifest.yml
@@ -0,0 +1,42 @@
+format_version: 1.0.4
+name: good
+title: Good package
+description: This package is good.
+version: 1.0.0
+conditions:
+  kibana.version: '^7.9.0'
+policy_templates:
+  - name: apache
+    title: Apache logs and metrics
+    description: Collect logs and metrics from Apache instances
+    inputs:
+      - type: apache/metrics
+        title: Collect metrics from Apache instances
+        description: Collecting Apache status metrics
+        multi: false
+        vars:
+          - name: hosts
+            type: url
+            url_allowed_schemes: ['http', 'https']
+            title: Hosts
+            multi: true
+            required: true
+            show_user: true
+            default:
+            - http://127.0.0.1
+owner:
+  github: elastic/foobar
+screenshots:
+  - src: /img/kibana-system.png
+    title: kibana system
+    size: 1220x852
+    type: image/png
+icons:
+  - src: /img/system.svg
+    title: system
+    size: 1000x1000
+    type: image/svg+xml
+# /main is a specific action underneath the monitor privilege. Declaring
+# "monitor/main" limits the provided privilege, "monitor", to only the "main"
+# action.
+elasticsearch.privileges.cluster: [monitor/main]
diff --git a/code/go/pkg/validator/validator_test.go b/code/go/pkg/validator/validator_test.go
index 7551d2173..20d0c2d61 100644
--- a/code/go/pkg/validator/validator_test.go
+++ b/code/go/pkg/validator/validator_test.go
@@ -47,8 +47,8 @@ func TestValidateFile(t *testing.T) {
 		"missing_image_files": {
 			"manifest.yml",
 			[]string{
-				"field screenshots.0.src: relative path is invalid or target doesn't exist",
-				"field icons.0.src: relative path is invalid or target doesn't exist",
+				"field screenshots.0.src: relative path is invalid, target doesn't exist or it exceeds the file size limit",
+				"field icons.0.src: relative path is invalid, target doesn't exist or it exceeds the file size limit",
 			},
 		},
 		"input_template": {},
diff --git a/versions/1/agent/spec.yml b/versions/1/agent/spec.yml
index fa0379e68..dae0db318 100644
--- a/versions/1/agent/spec.yml
+++ b/versions/1/agent/spec.yml
@@ -9,5 +9,6 @@ spec:
     contents:
     - description: Config template file for inputs defined in the policy_templates section of the top level manifest
       type: file
+      sizeLimit: 2MB
       pattern: '^.+.yml.hbs$'
-      required: true
\ No newline at end of file
+      required: true
diff --git a/versions/1/changelog.yml b/versions/1/changelog.yml
index 612ccbc7c..e196282b5 100644
--- a/versions/1/changelog.yml
+++ b/versions/1/changelog.yml
@@ -4,9 +4,6 @@
 ##
 - version: 1.4.2-next
   changes:
-  - description: Prepare for next version
-    type: enhancement
-    link: https://github.com/elastic/package-spec/pull/275
   - description: Add kibana/csp-rule-template asset
     type: enhancement
     link: https://github.com/elastic/package-spec/pull/276
@@ -16,6 +13,9 @@
   - description: Add support for match_only_text field type
     type: enhancement
     link: https://github.com/elastic/package-spec/issues/284
+  - description: Add limits to file sizes and number of files
+    type: enhancement
+    link: https://github.com/elastic/package-spec/pull/278
 - version: 1.4.1
   changes:
   - description: ML model file name now matches the id of the model.
diff --git a/versions/1/data_stream/spec.yml b/versions/1/data_stream/spec.yml
index 20cb381e1..53fd7b390 100644
--- a/versions/1/data_stream/spec.yml
+++ b/versions/1/data_stream/spec.yml
@@ -1,5 +1,6 @@
 spec:
   additionalContents: false
+  totalContentsLimit: 500
   contents:
   - description: Folder containing a single data stream definition
     type: folder
@@ -10,6 +11,7 @@ spec:
     - description: A data stream's manifest file
       type: file
       contentMediaType: "application/x-yaml"
+      sizeLimit: 5MB
       name: "manifest.yml"
       required: true
       $ref: "./manifest.spec.yml"
@@ -80,4 +82,4 @@ spec:
       name: _dev
       required: false
       visibility: private
-      $ref: "./_dev/spec.yml"
\ No newline at end of file
+      $ref: "./_dev/spec.yml"
diff --git a/versions/1/elasticsearch/spec.yml b/versions/1/elasticsearch/spec.yml
index 407361d2a..8ed7900bc 100644
--- a/versions/1/elasticsearch/spec.yml
+++ b/versions/1/elasticsearch/spec.yml
@@ -17,12 +17,14 @@ spec:
       contents:
         - description: Supporting ingest pipeline definitions in YAML
           type: file
+          sizeLimit: 3MB
           pattern: '^.+\.yml$'
           # TODO Determine if special handling of `---` is required (issue: https://github.com/elastic/package-spec/pull/54)
           contentMediaType: "application/x-yaml; require-document-dashes=true"
           required: false
         - description: Supporting ingest pipeline definitions in JSON
           type: file
+          sizeLimit: 3MB
           pattern: '^.+\.json$'
           contentMediaType: "application/json"
           required: false
diff --git a/versions/1/spec.yml b/versions/1/spec.yml
index b9da18c56..8bdcedaab 100644
--- a/versions/1/spec.yml
+++ b/versions/1/spec.yml
@@ -6,10 +6,17 @@
 version: 1.1.1-next
 spec:
   additionalContents: true
+  totalContentsLimit: 65535
+  totalSizeLimit: 250MB
+  sizeLimit: 150MB
+  configurationSizeLimit: 5MB
+  relativePathSizeLimit: 3MB
+  fieldsPerDataStreamLimit: 1024
   contents:
   - description: The main package manifest file
     type: file
     contentMediaType: "application/x-yaml"
+    sizeLimit: 5MB
     name: "manifest.yml"
     required: true
     $ref: "./manifest.spec.yml"