New Message processor

[TomTheBear]

New message processor using expr instead of gval

Can be integrated anywhere to check:

• drop messages by name, message type or condition
• rename messages by name or condition
• move entries between tags, meta and fields based on a condition
• add tag, meta or field entries based on a condition
• delete tag, meta or field entries based on a condition
• normalize units (metrics only)
• change unit prefixes and consequently the metric value based on a condition (metrics only)

It accepts a json.RawMessage as configuration and parses it. Therefore it can be integrated into other configs:

type myConfig struct {
  Myfield string `json:"myfield"`
  MessageProcessor json.RawMessage `json:"process_messages"`
}

For initialization:

p, err := mp.NewMessageProcessor()
if err != nil {
  return fmt.Errorf("initialization of message processor failed: %v", err.Error())
}
err = p.FromConfigJSON(config.MessageProcessor)
if err != nil {
  return fmt.Errorf("failed parsing JSON for message processor: %v", err.Error())
}

Usage:

drop, err := p.ProcessMessage(msg)
if err == nil && !drop {
  // do something with m
}

Rules can also be added and removed at runtime using the various Add/Remove functions of p (in the above examples).

Plan is to integrate it in every component that sends or receives messages to filter them based on the configuration.

• Only accept events through an receiver
• Send messages with different units per sink
• Rename only for PrometheusSink (as an example)