diff --git a/libbeat/beat/event.go b/libbeat/beat/event.go
index 54fc3e27cc39..dffb827603aa 100644
--- a/libbeat/beat/event.go
+++ b/libbeat/beat/event.go
@@ -19,6 +19,7 @@ package beat
 
 import (
 	"errors"
+	"fmt"
 	"strings"
 	"time"
 
@@ -26,12 +27,22 @@ import (
 	"github.com/elastic/elastic-agent-libs/mapstr"
 )
 
+type updateMode bool
+
+var (
+	updateModeOverwrite   updateMode = true
+	updateModeNoOverwrite updateMode = false
+)
+
 // FlagField fields used to keep information or errors when events are parsed.
 const FlagField = "log.flags"
 
 const (
-	timestampFieldKey = "@timestamp"
-	metadataFieldKey  = "@metadata"
+	TimestampFieldKey = "@timestamp"
+	MetadataFieldKey  = "@metadata"
+	ErrorFieldKey     = "error"
+	metadataKeyPrefix = MetadataFieldKey + "."
+	metadataKeyOffset = len(metadataKeyPrefix)
 )
 
 // Event is the common event format shared by all beats.
@@ -47,28 +58,44 @@ type Event struct {
 }
 
 var (
-	errNoTimestamp = errors.New("value is no timestamp")
-	errNoMapStr    = errors.New("value is no map[string]interface{} type")
+	ErrValueNotTimestamp = errors.New("value is not a timestamp")
+	ErrValueNotMapStr    = errors.New("value is not `mapstr.M` or `map[string]interface{}` type")
+	ErrAlterMetadataKey  = fmt.Errorf("deleting/replacing %q key is not supported", MetadataFieldKey)
+	ErrMetadataAccess    = fmt.Errorf("accessing %q key directly is not supported, try nested keys", MetadataFieldKey)
+	ErrDeleteTimestamp   = fmt.Errorf("deleting %q key is not supported", TimestampFieldKey)
 )
 
 // SetID overwrites the "id" field in the events metadata.
 // If Meta is nil, a new Meta dictionary is created.
 func (e *Event) SetID(id string) {
-	if e.Meta == nil {
-		e.Meta = mapstr.M{}
-	}
-	e.Meta["_id"] = id
+	_, _ = e.PutValue(metadataKeyPrefix+"_id", id)
 }
 
+// GetValue gets a value from the event. If the key does not exist then an error
+// is returned.
+//
+// Use `@timestamp` key for getting the event timestamp.
+// Use `@metadata.*` keys for getting the event metadata fields.
+// If `@metadata` key is used then `ErrMetadataAccess` is returned.
 func (e *Event) GetValue(key string) (interface{}, error) {
-	if key == timestampFieldKey {
+	if key == TimestampFieldKey {
 		return e.Timestamp, nil
-	} else if subKey, ok := metadataKey(key); ok {
-		if subKey == "" || e.Meta == nil {
-			return e.Meta, nil
+	}
+	if key == MetadataFieldKey {
+		return nil, ErrMetadataAccess
+	}
+
+	if subKey, ok := e.metadataSubKey(key); ok {
+		if e.Meta == nil {
+			return nil, mapstr.ErrKeyNotFound
 		}
 		return e.Meta.GetValue(subKey)
 	}
+
+	if e.Fields == nil {
+		return nil, mapstr.ErrKeyNotFound
+	}
+
 	return e.Fields.GetValue(key)
 }
 
@@ -92,7 +119,7 @@ func (e *Event) Clone() *Event {
 // `DeepUpdateNoOverwrite` is a version of this function that does not
 // overwrite existing values.
 func (e *Event) DeepUpdate(d mapstr.M) {
-	e.deepUpdate(d, true)
+	e.deepUpdate(d, updateModeOverwrite)
 }
 
 // DeepUpdateNoOverwrite recursively copies the key-value pairs from `d` to various properties of the event.
@@ -103,31 +130,34 @@ func (e *Event) DeepUpdate(d mapstr.M) {
 // via `DeepUpdateNoOverwrite`.
 // `DeepUpdate` is a version of this function that overwrites existing values.
 func (e *Event) DeepUpdateNoOverwrite(d mapstr.M) {
-	e.deepUpdate(d, false)
+	e.deepUpdate(d, updateModeNoOverwrite)
 }
 
-func (e *Event) deepUpdate(d mapstr.M, overwrite bool) {
+func (e *Event) deepUpdate(d mapstr.M, mode updateMode) {
 	if len(d) == 0 {
 		return
 	}
 
 	// It's supported to update the timestamp using this function.
 	// However, we must handle it separately since it's a separate field of the event.
-	timestampValue, timestampExists := d[timestampFieldKey]
+	timestampValue, timestampExists := d[TimestampFieldKey]
 	if timestampExists {
-		if overwrite {
-			_ = e.setTimestamp(timestampValue)
+		if mode == updateModeOverwrite {
+			_, _ = e.setTimestamp(timestampValue)
 		}
 
 		// Temporary delete it from the update map,
 		// so we can do `e.Fields.DeepUpdate(d)` or
 		// `e.Fields.DeepUpdateNoOverwrite(d)` later
-		delete(d, timestampFieldKey)
+		delete(d, TimestampFieldKey)
+		defer func() {
+			d[TimestampFieldKey] = timestampValue
+		}()
 	}
 
 	// It's supported to update the metadata using this function.
 	// However, we must handle it separately since it's a separate field of the event.
-	metaValue, metaExists := d[metadataFieldKey]
+	metaValue, metaExists := d[MetadataFieldKey]
 	if metaExists {
 		var metaUpdate mapstr.M
 
@@ -142,9 +172,10 @@ func (e *Event) deepUpdate(d mapstr.M, overwrite bool) {
 			if e.Meta == nil {
 				e.Meta = mapstr.M{}
 			}
-			if overwrite {
+			switch mode {
+			case updateModeOverwrite:
 				e.Meta.DeepUpdate(metaUpdate)
-			} else {
+			case updateModeNoOverwrite:
 				e.Meta.DeepUpdateNoOverwrite(metaUpdate)
 			}
 		}
@@ -152,19 +183,12 @@ func (e *Event) deepUpdate(d mapstr.M, overwrite bool) {
 		// Temporary delete it from the update map,
 		// so we can do `e.Fields.DeepUpdate(d)` or
 		// `e.Fields.DeepUpdateNoOverwrite(d)` later
-		delete(d, metadataFieldKey)
+		delete(d, MetadataFieldKey)
+		defer func() {
+			d[MetadataFieldKey] = metaValue
+		}()
 	}
 
-	// At the end we revert all changes we made to the update map
-	defer func() {
-		if timestampExists {
-			d[timestampFieldKey] = timestampValue
-		}
-		if metaExists {
-			d[metadataFieldKey] = metaValue
-		}
-	}()
-
 	if len(d) == 0 {
 		return
 	}
@@ -173,90 +197,150 @@ func (e *Event) deepUpdate(d mapstr.M, overwrite bool) {
 		e.Fields = mapstr.M{}
 	}
 
-	if overwrite {
+	switch mode {
+	case updateModeOverwrite:
 		e.Fields.DeepUpdate(d)
-	} else {
+	case updateModeNoOverwrite:
 		e.Fields.DeepUpdateNoOverwrite(d)
 	}
 }
 
-func (e *Event) setTimestamp(v interface{}) error {
+func (e *Event) setTimestamp(v interface{}) (interface{}, error) {
+	// to satisfy the PutValue interface, this function
+	// must return the overwritten value
+	prevValue := e.Timestamp
+
 	switch ts := v.(type) {
 	case time.Time:
 		e.Timestamp = ts
+		return prevValue, nil
 	case common.Time:
 		e.Timestamp = time.Time(ts)
+		return prevValue, nil
 	default:
-		return errNoTimestamp
+		return nil, ErrValueNotTimestamp
 	}
-
-	return nil
 }
 
+// Put associates the specified value with the specified key. If the event
+// previously contained a mapping for the key, the old value is replaced and
+// returned. The key can be expressed in dot-notation (e.g. x.y) to put a value
+// into a nested map.
+//
+// If you need insert keys containing dots then you must use bracket notation
+// to insert values (e.g. m[key] = value).
+//
+// Use `@timestamp` key for setting the event timestamp.
+// Use `@metadata.*` keys for setting the event metadata fields.
+// If `@metadata` key is used then `ErrAlterMetadataKey` is returned.
 func (e *Event) PutValue(key string, v interface{}) (interface{}, error) {
-	if key == timestampFieldKey {
-		err := e.setTimestamp(v)
-		return nil, err
-	} else if subKey, ok := metadataKey(key); ok {
-		if subKey == "" {
-			switch meta := v.(type) {
-			case mapstr.M:
-				e.Meta = meta
-			case map[string]interface{}:
-				e.Meta = meta
-			default:
-				return nil, errNoMapStr
-			}
-		} else if e.Meta == nil {
+	if key == TimestampFieldKey {
+		return e.setTimestamp(v)
+	}
+	if key == MetadataFieldKey {
+		return nil, ErrAlterMetadataKey
+	}
+
+	if subKey, ok := e.metadataSubKey(key); ok {
+		if e.Meta == nil {
 			e.Meta = mapstr.M{}
 		}
+
 		return e.Meta.Put(subKey, v)
 	}
 
+	if e.Fields == nil {
+		e.Fields = mapstr.M{}
+	}
+
 	return e.Fields.Put(key, v)
 }
 
+// Delete deletes the given key from the event.
+//
+// Use `@metadata.*` keys for deleting the event metadata fields.
+// If `@metadata` key is used then `ErrAlterMetadataKey` is returned.
+// If `@timestamp` key is used then `ErrDeleteTimestamp` is returned.
 func (e *Event) Delete(key string) error {
-	if subKey, ok := metadataKey(key); ok {
-		if subKey == "" {
-			e.Meta = nil
-			return nil
-		}
+	if key == TimestampFieldKey {
+		return ErrDeleteTimestamp
+	}
+	if key == MetadataFieldKey {
+		return ErrAlterMetadataKey
+	}
+	if subKey, ok := e.metadataSubKey(key); ok {
 		if e.Meta == nil {
-			return nil
+			return mapstr.ErrKeyNotFound
 		}
 		return e.Meta.Delete(subKey)
 	}
+
+	if e.Fields == nil {
+		return mapstr.ErrKeyNotFound
+	}
 	return e.Fields.Delete(key)
 }
 
-func metadataKey(key string) (string, bool) {
-	if !strings.HasPrefix(key, metadataFieldKey) {
+func (e *Event) metadataSubKey(key string) (string, bool) {
+	if !strings.HasPrefix(key, metadataKeyPrefix) {
 		return "", false
 	}
 
-	subKey := key[len(metadataFieldKey):]
+	subKey := key[metadataKeyOffset:]
 	if subKey == "" {
-		return "", true
-	}
-	if subKey[0] == '.' {
-		return subKey[1:], true
+		return "", false
 	}
-	return "", false
+	return subKey, true
 }
 
 // SetErrorWithOption sets the event error field with the message when the addErrKey is set to true.
 // If you want to include the data and field you can pass them as parameters and will be appended into the
 // error as fields with the corresponding name.
 func (e *Event) SetErrorWithOption(message string, addErrKey bool, data string, field string) {
-	if addErrKey {
-		errorField := mapstr.M{"message": message, "type": "json"}
-		if data != "" {
-			errorField["data"] = data
-		}
-		if field != "" {
-			errorField["field"] = field
+	if !addErrKey {
+		return
+	}
+
+	errorField := mapstr.M{"message": message, "type": "json"}
+	if data != "" {
+		errorField["data"] = data
+	}
+	if field != "" {
+		errorField["field"] = field
+	}
+	e.Fields[ErrorFieldKey] = errorField
+}
+
+// String returns a string representation of the event.
+func (e *Event) String() string {
+	m := mapstr.M{
+		TimestampFieldKey: e.Timestamp,
+		MetadataFieldKey:  mapstr.M{},
+	}
+	if e.Meta != nil {
+		m[MetadataFieldKey] = e.Meta
+	}
+	m.DeepUpdate(e.Fields)
+	return m.String()
+}
+
+// HasKey returns true if the key exist. If an error occurs then false is
+// returned with a non-nil error.
+func (e *Event) HasKey(key string) (bool, error) {
+	if key == TimestampFieldKey || key == MetadataFieldKey {
+		return true, nil
+	}
+
+	if subKey, ok := e.metadataSubKey(key); ok {
+		if e.Meta == nil {
+			return false, nil
 		}
-		e.Fields["error"] = errorField
+		return e.Meta.HasKey(subKey)
 	}
+
+	if e.Fields == nil {
+		return false, nil
+	}
+
+	return e.Fields.HasKey(key)
 }
diff --git a/libbeat/beat/event_test.go b/libbeat/beat/event_test.go
index cd165a3c4593..24ffb87dfa3e 100644
--- a/libbeat/beat/event_test.go
+++ b/libbeat/beat/event_test.go
@@ -18,351 +18,569 @@
 package beat
 
 import (
-	"crypto/rand"
 	"testing"
 	"time"
 
-	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
 
 	"github.com/elastic/elastic-agent-libs/mapstr"
 )
 
-const (
-	propSize = 1024 * 2014 * 10
-)
-
-var largeProp string
-
-func init() {
-	b := make([]byte, propSize)
-	_, _ = rand.Read(b)
-	largeProp = string(b)
-}
-
-func newEmptyEvent() *Event {
-	return &Event{Fields: mapstr.M{}}
-}
-
-func newEvent(e mapstr.M) *Event {
-	n := &mapstr.M{
-		"Fields": mapstr.M{
-			"large_prop": largeProp,
-		},
-	}
-	n.DeepUpdate(e)
-	var ts time.Time
-	var meta mapstr.M
-	var fields mapstr.M
-	var private mapstr.M
-
-	v, ex := (*n)["Timestamp"]
-	if ex {
-		ts = v.(time.Time)
+func TestEvent(t *testing.T) {
+	metadataNestedNestedMap := mapstr.M{
+		"metaLevel2Value": "metavalue3",
 	}
-	v, ex = (*n)["Meta"]
-	if ex {
-		meta = v.(mapstr.M)
+	metadataNestedMap := mapstr.M{
+		"metaLevel1Map": metadataNestedNestedMap,
 	}
-	v, ex = (*n)["Fields"]
-	if ex {
-		fields = v.(mapstr.M)
+
+	fieldsNestedNestedMap := mapstr.M{
+		"fieldsLevel2Value": "fieldsvalue3",
 	}
-	v, ex = (*n)["Private"]
-	if ex {
-		private = v.(mapstr.M)
+	fieldsNestedMap := mapstr.M{
+		"fieldsLevel1Map": fieldsNestedNestedMap,
 	}
-	return &Event{
-		Timestamp: ts,
-		Meta:      meta,
-		Fields:    fields,
-		Private:   private,
+
+	metaUntouchedMap := mapstr.M{}
+	fieldsUntouchedMap := mapstr.M{}
+
+	event := &Event{
+		Timestamp: time.Now(),
+		Meta: mapstr.M{
+			"a.b":             "c",
+			"metaLevel0Map":   metadataNestedMap,
+			"metaLevel0Value": "metavalue1",
+			// these keys should never be edited by the tests
+			// to verify that existing keys remain
+			"metaLevel0Value2": "untouched",
+			"metaUntouchedMap": metaUntouchedMap,
+		},
+		Fields: mapstr.M{
+			"a.b":               "c",
+			"fieldsLevel0Map":   fieldsNestedMap,
+			"fieldsLevel0Value": "fieldsvalue1",
+			// these keys should never be edited by the tests
+			// to verify that existing keys remain
+			"fieldsLevel0Value2": "untouched",
+			"fieldsUntouchedMap": fieldsUntouchedMap,
+		},
 	}
-}
 
-func BenchmarkTestEventPutGetTimestamp(b *testing.B) {
-	evt := newEmptyEvent()
-	ts := time.Now()
+	t.Run("empty", func(t *testing.T) {
+		t.Run("Delete", func(t *testing.T) {
+			event := &Event{}
+			require.NotPanics(t, func() {
+				err := event.Delete(metadataKeyPrefix + "some")
+				require.Error(t, err)
+				require.ErrorIs(t, err, mapstr.ErrKeyNotFound)
+				err = event.Delete("some")
+				require.Error(t, err)
+				require.ErrorIs(t, err, mapstr.ErrKeyNotFound)
+			})
+		})
 
-	evt.PutValue("@timestamp", ts)
+		t.Run("HasKey", func(t *testing.T) {
+			event := &Event{}
+			require.NotPanics(t, func() {
+				has, err := event.HasKey(metadataKeyPrefix + "some")
+				require.NoError(t, err)
+				require.False(t, has)
+				has, err = event.HasKey("some")
+				require.NoError(t, err)
+				require.False(t, has)
+			})
+		})
 
-	v, err := evt.GetValue("@timestamp")
-	if err != nil {
-		b.Fatal(err)
-	}
+		t.Run("GetValue", func(t *testing.T) {
+			event := &Event{}
+			require.NotPanics(t, func() {
+				_, err := event.GetValue(metadataKeyPrefix + "some")
+				require.Error(t, err)
+				require.ErrorIs(t, err, mapstr.ErrKeyNotFound)
+				_, err = event.GetValue("some")
+				require.Error(t, err)
+				require.ErrorIs(t, err, mapstr.ErrKeyNotFound)
+			})
+		})
 
-	assert.Equal(b, ts, v)
-	assert.Equal(b, ts, evt.Timestamp)
+		t.Run("PutValue", func(t *testing.T) {
+			event := &Event{}
+			require.NotPanics(t, func() {
+				prev, err := event.PutValue(metadataKeyPrefix+"some", "value")
+				require.NoError(t, err)
+				require.Nil(t, prev)
+				prev, err = event.PutValue("some", "value")
+				require.NoError(t, err)
+				require.Nil(t, prev)
+			})
+		})
 
-	// The @timestamp is not written into Fields.
-	assert.Nil(b, evt.Fields["@timestamp"])
-}
+		t.Run("DeepUpdate", func(t *testing.T) {
+			event := &Event{}
+			require.NotPanics(t, func() {
+				event.DeepUpdate(mapstr.M{
+					MetadataFieldKey: mapstr.M{"key": "value"},
+					"key":            "value",
+				})
+			})
+		})
 
-func BenchmarkTestDeepUpdate(b *testing.B) {
-	ts := time.Now()
-
-	cases := []struct {
-		name      string
-		event     *Event
-		update    mapstr.M
-		overwrite bool
-		expected  *Event
-	}{
-		{
-			name:     "does nothing if no update",
-			event:    newEvent(mapstr.M{}),
-			update:   mapstr.M{},
-			expected: newEvent(mapstr.M{}),
-		},
-		{
-			name:  "updates timestamp",
-			event: newEvent(mapstr.M{}),
-			update: mapstr.M{
-				timestampFieldKey: ts,
-			},
-			overwrite: true,
-			expected: &Event{
-				Timestamp: ts,
-				Fields: mapstr.M{
-					"large_prop": largeProp,
-				},
+		t.Run("String", func(t *testing.T) {
+			event := &Event{}
+			require.NotPanics(t, func() {
+				s := event.String()
+				require.Equal(t, `{"@metadata":{},"@timestamp":"0001-01-01T00:00:00Z"}`, s)
+			})
+		})
+	})
+
+	t.Run("Get", func(t *testing.T) {
+		cases := []struct {
+			name   string
+			key    string
+			exp    interface{}
+			expErr error
+		}{
+			{
+				name: TimestampFieldKey,
+				key:  TimestampFieldKey,
+				exp:  event.Timestamp,
 			},
-		},
-		{
-			name: "does not overwrite timestamp",
-			event: newEvent(mapstr.M{
-				"Timestamp": ts,
-			}),
-			update: mapstr.M{
-				timestampFieldKey: time.Now().Add(time.Hour),
-			},
-			overwrite: false,
-			expected: &Event{
-				Timestamp: ts,
-				Fields: mapstr.M{
-					"large_prop": largeProp,
-				},
+			{
+				name:   "no acess to metadata key",
+				key:    MetadataFieldKey,
+				expErr: ErrMetadataAccess,
 			},
-		},
-		{
-			name:  "initializes metadata if nil",
-			event: newEvent(mapstr.M{}),
-			update: mapstr.M{
-				metadataFieldKey: mapstr.M{
-					"first":  "new",
-					"second": 42,
-				},
+			{
+				name:   "non-existing metadata sub-key",
+				key:    metadataKeyPrefix + "none",
+				expErr: mapstr.ErrKeyNotFound,
 			},
-			expected: &Event{
-				Meta: mapstr.M{
-					"first":  "new",
-					"second": 42,
-				},
-				Fields: mapstr.M{
-					"large_prop": largeProp,
-				},
+			{
+				name: "a value type from metadata",
+				key:  metadataKeyPrefix + "metaLevel0Value",
+				exp:  "metavalue1",
 			},
-		},
-		{
-			name: "updates metadata but does not overwrite",
-			event: newEvent(mapstr.M{
-				"Meta": mapstr.M{
-					"first": "initial",
-				},
-			}),
-			update: mapstr.M{
-				metadataFieldKey: mapstr.M{
-					"first":  "new",
-					"second": 42,
-				},
+			{
+				name: "a root-level dot-key from metadata",
+				key:  metadataKeyPrefix + "a.b",
+				exp:  "c",
 			},
-			overwrite: false,
-			expected: &Event{
-				Meta: mapstr.M{
-					"first":  "initial",
-					"second": 42,
-				},
-				Fields: mapstr.M{
-					"large_prop": largeProp,
-				},
+			{
+				name: "a nested map from metadata",
+				key:  metadataKeyPrefix + "metaLevel0Map",
+				exp:  metadataNestedMap,
 			},
-		},
-		{
-			name: "updates metadata and overwrites",
-			event: newEvent(mapstr.M{
-				"Meta": mapstr.M{
-					"first": "initial",
-				},
-			}),
-			update: mapstr.M{
-				metadataFieldKey: mapstr.M{
-					"first":  "new",
-					"second": 42,
-				},
+			{
+				name:   "non-existing field key",
+				key:    "none",
+				expErr: mapstr.ErrKeyNotFound,
 			},
-			overwrite: true,
-			expected: &Event{
-				Meta: mapstr.M{
-					"first":  "new",
-					"second": 42,
-				},
-				Fields: mapstr.M{
-					"large_prop": largeProp,
-				},
+			{
+				name: "a value type from fields",
+				key:  "fieldsLevel0Value",
+				exp:  "fieldsvalue1",
 			},
-		},
-		{
-			name: "updates fields but does not overwrite",
-			event: newEvent(mapstr.M{
-				"Fields": mapstr.M{
-					"first": "initial",
-				},
-			}),
-			update: mapstr.M{
-				"first":  "new",
-				"second": 42,
+			{
+				name: "a root-level dot-key from fields",
+				key:  "a.b",
+				exp:  "c",
 			},
-			overwrite: false,
-			expected: &Event{
-				Fields: mapstr.M{
-					"first":      "initial",
-					"second":     42,
-					"large_prop": largeProp,
-				},
+			{
+				name: "a nested map from fields",
+				key:  "fieldsLevel0Map",
+				exp:  fieldsNestedMap,
 			},
-		},
-		{
-			name: "updates metadata and overwrites",
-			event: newEvent(mapstr.M{
-				"Fields": mapstr.M{
-					"first": "initial",
-				},
-			}),
-			update: mapstr.M{
-				"first":  "new",
-				"second": 42,
+		}
+
+		for _, tc := range cases {
+			t.Run(tc.name, func(t *testing.T) {
+				val, err := event.GetValue(tc.key)
+				if tc.expErr != nil {
+					require.Error(t, err)
+					require.Nil(t, val)
+					require.ErrorIs(t, err, tc.expErr)
+					return
+				}
+				require.NoError(t, err)
+				require.Equal(t, tc.exp, val)
+			})
+		}
+	})
+
+	t.Run("Delete", func(t *testing.T) {
+		cases := []struct {
+			name   string
+			key    string
+			exp    interface{}
+			expErr error
+		}{
+			{
+				name:   TimestampFieldKey,
+				key:    TimestampFieldKey,
+				expErr: ErrDeleteTimestamp,
 			},
-			overwrite: true,
-			expected: &Event{
-				Fields: mapstr.M{
-					"first":      "new",
-					"second":     42,
-					"large_prop": largeProp,
-				},
+			{
+				name:   "no acess to metadata key",
+				key:    MetadataFieldKey,
+				expErr: ErrAlterMetadataKey,
 			},
-		},
-		{
-			name:  "initializes fields if nil",
-			event: newEvent(mapstr.M{}),
-			update: mapstr.M{
-				"first":  "new",
-				"second": 42,
-			},
-			expected: &Event{
-				Fields: mapstr.M{
-					"first":      "new",
-					"second":     42,
-					"large_prop": largeProp,
-				},
+			{
+				name:   "non-existing metadata sub key",
+				key:    metadataKeyPrefix + "none",
+				expErr: mapstr.ErrKeyNotFound,
 			},
-		},
-	}
-
-	for _, tc := range cases {
-		b.Run(tc.name, func(b *testing.B) {
-			tc.event.deepUpdate(tc.update, tc.overwrite)
-			assert.Equal(b, tc.expected.Timestamp, tc.event.Timestamp)
-			assert.Equal(b, tc.expected.Fields, tc.event.Fields)
-			assert.Equal(b, tc.expected.Meta, tc.event.Meta)
-		})
-	}
-}
-
-func BenchmarkTestEventMetadata(b *testing.B) {
-	const id = "123"
-	newMeta := func() mapstr.M { return mapstr.M{"_id": id} }
-
-	b.Run("put", func(b *testing.B) {
-		evt := newEmptyEvent()
-		meta := newMeta()
-
-		evt.PutValue("@metadata", meta)
-
-		assert.Equal(b, meta, evt.Meta)
-		assert.Empty(b, evt.Fields)
-	})
-
-	b.Run("get", func(b *testing.B) {
-		evt := newEmptyEvent()
-		evt.Meta = newMeta()
-
-		meta, err := evt.GetValue("@metadata")
-
-		assert.NoError(b, err)
-		assert.Equal(b, evt.Meta, meta)
+			{
+				name: "a value type from metadata",
+				key:  metadataKeyPrefix + "metaLevel0Value",
+				exp:  "metavalue1",
+			},
+			{
+				name: "a root-level dot-key from metadata",
+				key:  metadataKeyPrefix + "a.b",
+				exp:  "c",
+			},
+			{
+				name: "a nested map from metadata",
+				key:  metadataKeyPrefix + "metaLevel0Map",
+				exp:  metadataNestedMap,
+			},
+			{
+				name:   "non-existing field key",
+				key:    "none",
+				expErr: mapstr.ErrKeyNotFound,
+			},
+			{
+				name: "a value type from fields",
+				key:  "fieldsLevel0Value",
+				exp:  "fieldsvalue1",
+			},
+			{
+				name: "a root-level dot-key from fields",
+				key:  "a.b",
+				exp:  "c",
+			},
+			{
+				name: "a nested map from fields",
+				key:  "fieldsLevel0Map",
+				exp:  fieldsNestedMap,
+			},
+		}
+
+		for _, tc := range cases {
+			t.Run(tc.name, func(t *testing.T) {
+				event := event.Clone()
+				err := event.Delete(tc.key)
+				if tc.expErr != nil {
+					require.Error(t, err)
+					require.ErrorIs(t, err, tc.expErr)
+					return
+				}
+				require.NoError(t, err)
+				_, err = event.GetValue(tc.key)
+				require.Error(t, err)
+				require.ErrorIs(t, err, mapstr.ErrKeyNotFound)
+			})
+		}
 	})
 
-	b.Run("put sub-key", func(b *testing.B) {
-		evt := newEmptyEvent()
-
-		evt.PutValue("@metadata._id", id)
-
-		assert.Equal(b, newMeta(), evt.Meta)
-		assert.Empty(b, evt.Fields)
-	})
+	t.Run("PutValue", func(t *testing.T) {
+		newTs := time.Now().Add(time.Hour)
+		cases := []struct {
+			name    string
+			key     string
+			val     interface{}
+			expPrev interface{}
+			expErr  error
+		}{
+			{
+				name:    "timestamp",
+				key:     TimestampFieldKey,
+				val:     newTs,
+				expPrev: event.Timestamp,
+			},
+			{
+				name:   "incorrect type for timestamp",
+				key:    TimestampFieldKey,
+				val:    "wrong",
+				expErr: ErrValueNotTimestamp,
+			},
+			{
+				name:   "no acess to metadata key",
+				key:    MetadataFieldKey,
+				expErr: ErrAlterMetadataKey,
+			},
+			{
+				name:    "non-existing metadata key",
+				key:     metadataKeyPrefix + "none",
+				expPrev: nil,
+			},
+			{
+				name:    "a value type from metadata",
+				key:     metadataKeyPrefix + "metaLevel0Value",
+				val:     "some",
+				expPrev: "metavalue1",
+			},
+			{
+				name:    "a root-level dot-key from metadata",
+				key:     metadataKeyPrefix + "a.b",
+				val:     "d",
+				expPrev: "c",
+			},
+			{
+				name:    "a nested map from metadata",
+				key:     metadataKeyPrefix + "metaLevel0Map",
+				val:     "some",
+				expPrev: metadataNestedMap,
+			},
+			{
+				name:    "non-existing field key",
+				key:     "none",
+				val:     "some",
+				expPrev: nil,
+			},
+			{
+				name:    "a value type from fields",
+				key:     "fieldsLevel0Value",
+				val:     "some",
+				expPrev: "fieldsvalue1",
+			},
+			{
+				name:    "a root-level dot-key from fields",
+				key:     "a.b",
+				val:     "d",
+				expPrev: "c",
+			},
+			{
+				name:    "a nested map from fields",
+				key:     "fieldsLevel0Map",
+				val:     "some",
+				expPrev: fieldsNestedMap,
+			},
+		}
+
+		for _, tc := range cases {
+			t.Run(tc.name, func(t *testing.T) {
+				event := event.Clone()
+				prevVal, err := event.PutValue(tc.key, tc.val)
+				if tc.expErr != nil {
+					require.Error(t, err)
+					require.ErrorIs(t, err, tc.expErr)
+					require.Nil(t, prevVal)
+					return
+				}
+				require.NoError(t, err)
+				require.Equal(t, tc.expPrev, prevVal)
+				actual, err := event.GetValue(tc.key)
+				require.NoError(t, err)
+				require.Equal(t, tc.val, actual)
+				has, err := event.HasKey(tc.key)
+				require.NoError(t, err)
+				require.True(t, has)
+			})
+		}
+
+		t.Run("type conflict", func(t *testing.T) {
+			event := &Event{
+				Meta: mapstr.M{
+					"a": 9,
+					"c": 10,
+				},
+				Fields: mapstr.M{
+					"a": 9,
+					"c": 10,
+				},
+			}
+
+			_, err := event.PutValue("a.c", 10)
+			require.Error(t, err)
+			require.Equal(t, "expected map but type is int", err.Error())
+			_, err = event.PutValue("a.value", 9)
+			require.Error(t, err)
+			require.Equal(t, "expected map but type is int", err.Error())
+		})
 
-	b.Run("get sub-key", func(b *testing.B) {
-		evt := newEmptyEvent()
-		evt.Meta = newMeta()
+		t.Run("hierarchy", func(t *testing.T) {
+			event := &Event{
+				Fields: mapstr.M{
+					"a.b": 1,
+				},
+			}
+			err := event.Delete("a.b")
+			require.NoError(t, err)
+
+			prev, err := event.PutValue("a.b.c", 1)
+			require.NoError(t, err)
+			require.Nil(t, prev)
+
+			expFields := mapstr.M{
+				"a": mapstr.M{
+					"b": mapstr.M{
+						"c": 1,
+					},
+				},
+			}
 
-		v, err := evt.GetValue("@metadata._id")
+			require.Equal(t, expFields, event.Fields)
+		})
 
-		assert.NoError(b, err)
-		assert.Equal(b, id, v)
+		t.Run("SetID", func(t *testing.T) {
+			event := &Event{}
+			event.SetID("unique")
+			require.Equal(t, "unique", event.Meta["_id"])
+		})
 	})
 
-	b.Run("delete", func(b *testing.B) {
-		evt := newEmptyEvent()
-		evt.Meta = newMeta()
-
-		err := evt.Delete("@metadata")
-
-		assert.NoError(b, err)
-		assert.Nil(b, evt.Meta)
+	t.Run("SetErrorWithOption", func(t *testing.T) {
+		cloned := event.Clone()
+		cloned.SetErrorWithOption("message", false, "data", "field")
+		require.Equal(t, event, cloned)
+		expEvent := cloned.Clone()
+		expEvent.Fields[ErrorFieldKey] = mapstr.M{
+			"message": "message",
+			"field":   "field",
+			"data":    "data",
+			"type":    "json",
+		}
+		cloned.SetErrorWithOption("message", true, "data", "field")
+		require.Equal(t, expEvent, cloned)
 	})
 
-	b.Run("delete sub-key", func(b *testing.B) {
-		evt := newEmptyEvent()
-		evt.Meta = newMeta()
-
-		err := evt.Delete("@metadata._id")
-
-		assert.NoError(b, err)
-		assert.Empty(b, evt.Meta)
-	})
+	t.Run("DeepUpdate", func(t *testing.T) {
+		newTs := time.Now().Add(time.Hour)
+		update := map[string]interface{}{
+			TimestampFieldKey: newTs,
+			MetadataFieldKey: map[string]interface{}{
+				"metaLevel0Map": mapstr.M{ // mix types on purpose, should support both
+					"metaLevel1Map": map[string]interface{}{
+						"new1": "newmetavalue1",
+					},
+				},
+				"metaLevel0Value": "metareplaced1",
+				"new2":            "newmetavalue2",
+			},
+			"fieldsLevel0Map": map[string]interface{}{
+				"fieldsLevel1Map": mapstr.M{
+					"new3": "newfieldsvalue1",
+				},
+				"newmap": map[string]interface{}{
+					"new4": "newfieldsvalue2",
+				},
+			},
+			"fieldsLevel0Value": "fieldsreplaced1",
+		}
+
+		t.Run("empty", func(t *testing.T) {
+			cloned := event.Clone()
+			cloned.DeepUpdate(nil)
+			require.Equal(t, event.Meta, cloned.Meta)
+			require.Equal(t, event.Fields, cloned.Fields)
+		})
 
-	b.Run("setID", func(b *testing.B) {
-		evt := newEmptyEvent()
+		t.Run("overwrite", func(t *testing.T) {
+			event := event.Clone()
+			event.DeepUpdate(update)
 
-		evt.SetID(id)
+			expEvent := &Event{
+				Timestamp: newTs,
+				Meta: mapstr.M{
+					"a.b": "c",
+					"metaLevel0Map": mapstr.M{
+						"metaLevel1Map": mapstr.M{
+							"metaLevel2Value": "metavalue3",
+							"new1":            "newmetavalue1",
+						},
+					},
+					"metaLevel0Value":  "metareplaced1",
+					"metaLevel0Value2": "untouched",
+					"new2":             "newmetavalue2",
+					"metaUntouchedMap": metaUntouchedMap,
+				},
+				Fields: mapstr.M{
+					"a.b": "c",
+					"fieldsLevel0Map": mapstr.M{
+						"fieldsLevel1Map": mapstr.M{
+							"fieldsLevel2Value": "fieldsvalue3",
+							"new3":              "newfieldsvalue1",
+						},
+						"newmap": mapstr.M{
+							"new4": "newfieldsvalue2",
+						},
+					},
+					"fieldsLevel0Value":  "fieldsreplaced1",
+					"fieldsLevel0Value2": "untouched",
+					"fieldsUntouchedMap": fieldsUntouchedMap,
+				},
+			}
 
-		assert.Equal(b, newMeta(), evt.Meta)
-	})
+			require.Equal(t, expEvent.Timestamp, event.Timestamp)
+			require.Equal(t, expEvent.Meta, event.Meta)
+			require.Equal(t, expEvent.Fields, event.Fields)
+		})
 
-	b.Run("put non-metadata", func(b *testing.B) {
-		evt := newEmptyEvent()
+		t.Run("no overwrite", func(t *testing.T) {
+			cloned := event.Clone()
+			cloned.DeepUpdateNoOverwrite(update)
 
-		evt.PutValue("@metadataSpecial", id)
+			expEvent := &Event{
+				// should have the original/non-overwritten timestamp value
+				Timestamp: event.Timestamp,
+				Meta: mapstr.M{
+					"a.b": "c",
+					"metaLevel0Map": mapstr.M{
+						"metaLevel1Map": mapstr.M{
+							"metaLevel2Value": "metavalue3",
+							"new1":            "newmetavalue1",
+						},
+					},
+					"metaLevel0Value":  "metavalue1",
+					"metaLevel0Value2": "untouched",
+					"new2":             "newmetavalue2",
+					"metaUntouchedMap": metaUntouchedMap,
+				},
+				Fields: mapstr.M{
+					"a.b": "c",
+					"fieldsLevel0Map": mapstr.M{
+						"fieldsLevel1Map": mapstr.M{
+							"fieldsLevel2Value": "fieldsvalue3",
+							"new3":              "newfieldsvalue1",
+						},
+						"newmap": mapstr.M{
+							"new4": "newfieldsvalue2",
+						},
+					},
+					"fieldsLevel0Value":  "fieldsvalue1",
+					"fieldsLevel0Value2": "untouched",
+					"fieldsUntouchedMap": fieldsUntouchedMap,
+				},
+			}
 
-		assert.Equal(b, mapstr.M{"@metadataSpecial": id}, evt.Fields)
+			require.Equal(t, expEvent.Timestamp, cloned.Timestamp)
+			require.Equal(t, expEvent.Meta, cloned.Meta)
+			require.Equal(t, expEvent.Fields, cloned.Fields)
+		})
 	})
 
-	b.Run("delete non-metadata", func(b *testing.B) {
-		evt := newEmptyEvent()
-		evt.Meta = newMeta()
+	t.Run("String", func(t *testing.T) {
+		ts := time.Now().Add(time.Hour)
+		event := &Event{
+			Timestamp: ts,
+			Meta: mapstr.M{
+				"metakey": "metavalue",
+			},
+			Fields: mapstr.M{
+				"key": "value",
+			},
+		}
 
-		err := evt.Delete("@metadataSpecial")
+		exp := mapstr.M{
+			TimestampFieldKey: ts,
+			MetadataFieldKey: mapstr.M{
+				"metakey": "metavalue",
+			},
+			"key": "value",
+		}
 
-		assert.Error(b, err)
-		assert.Equal(b, newMeta(), evt.Meta)
+		require.Equal(t, exp.String(), event.String())
 	})
 }
diff --git a/libbeat/processors/add_id/add_id_test.go b/libbeat/processors/add_id/add_id_test.go
index 18effb85205b..e3c5c410bf17 100644
--- a/libbeat/processors/add_id/add_id_test.go
+++ b/libbeat/processors/add_id/add_id_test.go
@@ -61,8 +61,8 @@ func TestNonDefaultTargetField(t *testing.T) {
 	assert.NotEmpty(t, v)
 
 	v, err = newEvent.GetValue("@metadata._id")
-	assert.NoError(t, err)
-	assert.Empty(t, v)
+	assert.Error(t, err)
+	assert.ErrorIs(t, err, mapstr.ErrKeyNotFound)
 }
 
 func TestNonDefaultMetadataTarget(t *testing.T) {