| 65 | } |
| 66 | |
| 67 | func (p *ProcessorJSON) processLog(log *protocol.Log) { |
| 68 | findKey := false |
| 69 | for idx := range log.Contents { |
| 70 | if log.Contents[idx].Key == p.SourceKey { |
| 71 | objectVal := log.Contents[idx].Value |
| 72 | param := ExpandParam{ |
| 73 | sourceKey: p.SourceKey, |
| 74 | log: log, |
| 75 | nowDepth: 0, |
| 76 | maxDepth: p.ExpandDepth, |
| 77 | connector: p.ExpandConnector, |
| 78 | prefix: p.Prefix, |
| 79 | ignoreFirstConnector: p.IgnoreFirstConnector, |
| 80 | expandArray: p.ExpandArray, |
| 81 | } |
| 82 | if p.UseSourceKeyAsPrefix { |
| 83 | param.preKey = p.SourceKey |
| 84 | } |
| 85 | err := jsonparser.ObjectEach([]byte(objectVal), param.ExpandJSONCallBack) |
| 86 | if err != nil { |
| 87 | logger.Warningf(p.context.GetRuntimeContext(), selfmonitor.ProcessorJSONParserAlarm, "parser json error %v", err) |
| 88 | } |
| 89 | if !p.shouldKeepSource(err) { |
| 90 | log.Contents = append(log.Contents[:idx], log.Contents[idx+1:]...) |
| 91 | } |
| 92 | findKey = true |
| 93 | break |
| 94 | } |
| 95 | } |
| 96 | if !findKey && p.NoKeyError { |
| 97 | logger.Warningf(p.context.GetRuntimeContext(), selfmonitor.ProcessorJSONFindAlarm, "cannot find key %v", p.SourceKey) |
| 98 | } |
| 99 | } |
| 100 | |
| 101 | func (p *ProcessorJSON) shouldKeepSource(err error) bool { |
| 102 | return p.KeepSource || (p.KeepSourceIfParseError && err != nil) |