add kube-json parser

main.go

@@ -397,7 +397,6 @@ func processSource(nsCfg config.NamespaceConfig, t tail.Follower, parser parser.
 		}
 
 		fields, err := parser.ParseString(line)
-		fmt.Println(fields)
 		if err != nil {
 			fmt.Printf("error while parsing line '%s': %s\n", line, err)
 			metrics.parseErrorsTotal.Inc()

parser/jsonparser/jsonparser.go

@@ -3,21 +3,39 @@ package jsonparser
 import (
 	"encoding/json"
 	"fmt"
+	"strings"
+
+	"github.com/tidwall/gjson"
 )
 
 // JsonParser parse a JSON string.
-type JsonParser struct{}
+type JsonParser struct {
+	fnGetActualLog func(string) string
+}
 
-// NewJsonParser returns a new json parser.
 func NewJsonParser() *JsonParser {
 	return &JsonParser{}
 }
 
+func NewKubeJsonParser() *JsonParser {
+	return &JsonParser{
+		fnGetActualLog: func(line string) string {
+			// the actual k8s log is wrapped in log field
+			line = gjson.Get(line, "log").String()
+			return strings.TrimSpace(line)
+		},
+	}
+}
+
 // ParseString implements the Parser interface.
 // The value in the map is not necessarily a string, so it needs to be converted.
 func (j *JsonParser) ParseString(line string) (map[string]string, error) {
 	var parsed map[string]interface{}
-	err := json.Unmarshal([]byte(line), &parsed)
+	actualLogLine := line
+	if j.fnGetActualLog != nil {
+		actualLogLine = j.fnGetActualLog(line)
+	}
+	err := json.Unmarshal([]byte(actualLogLine), &parsed)
 	if err != nil {
 		return nil, fmt.Errorf("json log parsing err: %w", err)
 	}

parser/parser.go

@@ -27,6 +27,8 @@ func NewParser(nsCfg config.NamespaceConfig) Parser {
 		return kubeparser.NewKubeParser(nsCfg.Format)
 	case "kube-cri":
 		return kubeparser.NewKubeCRIParser(nsCfg.Format)
+	case "kube-json":
+		return jsonparser.NewKubeJsonParser()
 	case "istio":
 		return istioparser.NewIstioParser()
 	case "istio-cri":