libbeat: Refactor error handling in schema.Apply() (#7335)

`schema.Apply()` returned a `schema.Errors` object as error, with this
object is not easy to check if there was really an error, specially when
optional fields are involved. This leads to lots of cases where these
errors were being ignored.

This change introduces a series of changes to have better control on the
errors happening during the application of an schema.

The interface is kept as `event, err := schema.Apply(data)`, but now the
error returned is always a plain error in case of one or more errors
happened, or nil if no error happened.

`schema.ApplyTo()` returns the list of errors, and can be used where
multiple schemas are applied for the same event.

`Apply()` and `ApplyTo()` can now receive a list of options, these
options are functions that can be used to filter the errors returned or
to access the unfiltered list of errors.  This allows different
combinations of resulting errors in case of optional or required fields.
It also allows to add additional checks on all the errors happened, even
on the ones ignored for the final result. Three options are added by now:
* `AllRequired` to fail if any non-optional field is missing, this
mimics the current behaviour, and is set by default if no other option
is set.
* `FailOnRequired` to fail if any required field is missing. A new
`Required` option is also added to be able to declare required fields.
* `NotFoundKeys(cb func([]string))` is an option builder that receives
a function, this function is called with the list of missing keys, so
additional custom checks or tests can be done.

For lists of errors, the `multierror` library is used now, some custom
errors have been added to help on previous modifications and to improve
some error messages. Errors on structured data contain now the full
paths of the fields.

CHANGELOG-developer.asciidoc

@@ -33,3 +33,4 @@ The list below covers the major changes between 6.3.0 and master only.
 ==== Added
 
 - Libbeat provides a global registry for beats developer that allow to register and retrieve plugin. {pull}7392[7392]
+- Added more options to control required and optional fields in schema.Apply(), error returned is a plain nil if no error happened {pull}7335[7335]

CHANGELOG.asciidoc

@@ -170,6 +170,7 @@ https://github.com/elastic/beats/compare/v6.2.3...master[Check the HEAD diff]
 - Add ability to define input configuration as stringified JSON for autodiscover. {pull}7372[7372]
 - Add processor definition support for hints builder {pull}7386[7386]
 - Add support to disable html escaping in outputs. {pull}7445[7445]
+- Refactor error handing in schema.Apply(). {pull}7335[7335]
 - Add additional types to kubernetes metadata {pull}7457[7457]
 
 *Auditbeat*

libbeat/autodiscover/providers/jolokia/discovery.go

@@ -58,14 +58,14 @@ import (
 // Message contains the information of a Jolokia Discovery message
 var messageSchema = s.Schema{
 	"agent": s.Object{
-		"id":      c.Str("agent_id"),
-		"version": c.Str("agent_version", s.Optional),
+		"id":      c.Str("agent_id", s.Required),
+		"version": c.Str("agent_version"),
 	},
-	"secured": c.Bool("secured", s.Optional),
+	"secured": c.Bool("secured"),
 	"server": s.Object{
-		"product": c.Str("server_product", s.Optional),
-		"vendor":  c.Str("server_vendor", s.Optional),
-		"version": c.Str("server_version", s.Optional),
+		"product": c.Str("server_product"),
+		"vendor":  c.Str("server_vendor"),
+		"version": c.Str("server_version"),
 	},
 	"url": c.Str("url"),
 }
@@ -245,7 +245,11 @@ func (d *Discovery) sendProbe(config InterfaceConfig) {
 					logp.Err(err.Error())
 					continue
 				}
-				message, _ := messageSchema.Apply(m)
+				message, err := messageSchema.Apply(m, s.FailOnRequired)
+				if err != nil {
+					logp.Err(err.Error())
+					continue
+				}
 				d.update(config, message)
 			}
 		}()

libbeat/common/schema/error.go

@@ -17,37 +17,76 @@
 
 package schema
 
-import "fmt"
-
-const (
-	RequiredType ErrorType = iota
-	OptionalType ErrorType = iota
+import (
+	"fmt"
 )
 
-type ErrorType int
+// KeyError is an error with a field key
+type KeyError interface {
+	Key() string
+	SetKey(k string)
+}
+
+type errorKey struct {
+	key string
+}
+
+// Key returns the value of the field key
+func (k *errorKey) Key() string {
+	return k.key
+}
+
+// SetKey sets the value of the field key
+func (k *errorKey) SetKey(v string) {
+	k.key = v
+}
+
+// KeyNotFoundError is an error happening when a field key is not found
+type KeyNotFoundError struct {
+	errorKey
+
+	Err      error
+	Optional bool
+	Required bool
+}
 
-type Error struct {
-	key       string
-	message   string
-	errorType ErrorType
+// NewKeyNotFoundError builds a KeyNotFoundError
+func NewKeyNotFoundError(key string) *KeyNotFoundError {
+	var e KeyNotFoundError
+	e.SetKey(key)
+	return &e
 }
 
-func NewError(key string, message string) *Error {
-	return &Error{
-		key:       key,
-		message:   message,
-		errorType: RequiredType,
+// Error returns the error message of a KeyNotFoundError
+func (err *KeyNotFoundError) Error() string {
+	msg := fmt.Sprintf("key `%s` not found", err.Key())
+	if err.Err != nil {
+		msg += ": " + err.Err.Error()
 	}
+	return msg
 }
 
-func (err *Error) SetType(errorType ErrorType) {
-	err.errorType = errorType
+// WrongFormatError is an error happening when a field format is incorrect
+type WrongFormatError struct {
+	errorKey
+
+	Msg string
 }
 
-func (err *Error) IsType(errorType ErrorType) bool {
-	return err.errorType == errorType
+// NewWrongFormatError builds a new WrongFormatError
+func NewWrongFormatError(key string, msg string) *WrongFormatError {
+	e := WrongFormatError{
+		Msg: msg,
+	}
+	e.SetKey(key)
+	return &e
 }
 
-func (err *Error) Error() string {
-	return fmt.Sprintf("Missing field: %s, Error: %s", err.key, err.message)
+// Error returns the error message of a WrongFormatError
+func (err *WrongFormatError) Error() string {
+	msg := fmt.Sprintf("wrong format in `%s`", err.Key())
+	if err.Msg != "" {
+		msg += ": " + err.Msg
+	}
+	return msg
 }