代码拉取完成,页面将自动刷新
package actions
import (
"bytes"
"encoding/json"
"fmt"
"strings"
"github.com/elastic/beats/libbeat/common"
"github.com/elastic/beats/libbeat/common/jsontransform"
"github.com/elastic/beats/libbeat/logp"
"github.com/elastic/beats/libbeat/processors"
"github.com/pkg/errors"
)
type decodeJSONFields struct {
fields []string
maxDepth int
overwriteKeys bool
processArray bool
target *string
}
type config struct {
Fields []string `config:"fields"`
MaxDepth int `config:"max_depth" validate:"min=1"`
OverwriteKeys bool `config:"overwrite_keys"`
ProcessArray bool `config:"process_array"`
Target *string `config:"target"`
}
var (
defaultConfig = config{
MaxDepth: 1,
ProcessArray: false,
}
)
var debug = logp.MakeDebug("filters")
func init() {
processors.RegisterPlugin("decode_json_fields",
configChecked(newDecodeJSONFields,
requireFields("fields"),
allowedFields("fields", "max_depth", "overwrite_keys", "process_array", "target", "when")))
}
func newDecodeJSONFields(c common.Config) (processors.Processor, error) {
config := defaultConfig
err := c.Unpack(&config)
if err != nil {
logp.Warn("Error unpacking config for decode_json_fields")
return nil, fmt.Errorf("fail to unpack the decode_json_fields configuration: %s", err)
}
f := decodeJSONFields{fields: config.Fields, maxDepth: config.MaxDepth, overwriteKeys: config.OverwriteKeys, processArray: config.ProcessArray, target: config.Target}
return f, nil
}
func (f decodeJSONFields) Run(event common.MapStr) (common.MapStr, error) {
var errs []string
for _, field := range f.fields {
data, err := event.GetValue(field)
if err != nil && errors.Cause(err) != common.ErrKeyNotFound {
debug("Error trying to GetValue for field : %s in event : %v", field, event)
errs = append(errs, err.Error())
continue
}
text, ok := data.(string)
if ok {
var output interface{}
err := unmarshal(f.maxDepth, []byte(text), &output, f.processArray)
if err != nil {
debug("Error trying to unmarshal %s", event[field])
errs = append(errs, err.Error())
continue
}
if f.target != nil {
if len(*f.target) > 0 {
_, err = event.Put(*f.target, output)
} else {
switch t := output.(type) {
default:
errs = append(errs, errors.New("Error trying to add target to root.").Error())
case map[string]interface{}:
jsontransform.WriteJSONKeys(event, t, f.overwriteKeys, "json_error")
}
}
} else {
_, err = event.Put(field, output)
}
if err != nil {
debug("Error trying to Put value %v for field : %s", output, field)
errs = append(errs, err.Error())
continue
}
}
}
if len(errs) > 0 {
return event, fmt.Errorf(strings.Join(errs, ", "))
}
return event, nil
}
func unmarshal(maxDepth int, text []byte, fields *interface{}, processArray bool) error {
if err := DecodeJSON(text, fields); err != nil {
return err
}
maxDepth--
if maxDepth == 0 {
return nil
}
tryUnmarshal := func(v interface{}) (interface{}, bool) {
str, isString := v.(string)
if !isString {
return v, false
}
var tmp interface{}
err := unmarshal(maxDepth, []byte(str), &tmp, processArray)
if err != nil {
return v, false
}
return tmp, true
}
// try to deep unmarshal fields
switch O := interface{}(*fields).(type) {
case map[string]interface{}:
for k, v := range O {
if decoded, ok := tryUnmarshal(v); ok {
O[k] = decoded
}
}
// We want to process arrays here
case []interface{}:
if !processArray {
break
}
for i, v := range O {
if decoded, ok := tryUnmarshal(v); ok {
O[i] = decoded
}
}
}
return nil
}
func DecodeJSON(text []byte, to *interface{}) error {
dec := json.NewDecoder(bytes.NewReader(text))
dec.UseNumber()
err := dec.Decode(to)
if err != nil {
return err
}
if dec.More() {
return errors.New("multiple json elements found")
}
switch O := interface{}(*to).(type) {
case map[string]interface{}:
jsontransform.TransformNumbers(O)
}
return nil
}
func (f decodeJSONFields) String() string {
return "decode_json_fields=" + strings.Join(f.fields, ", ")
}
此处可能存在不合适展示的内容,页面不予展示。您可通过相关编辑功能自查并修改。
如您确认内容无涉及 不当用语 / 纯广告导流 / 暴力 / 低俗色情 / 侵权 / 盗版 / 虚假 / 无价值内容或违法国家有关法律法规的内容,可点击提交进行申诉,我们将尽快为您处理。