parse last line

elastic · Aug 13, 2021 · 5541d2d · 5541d2d
1 parent 85cd464
commit 5541d2d
Show file tree

Hide file tree

Showing 3 changed files with 78 additions and 49 deletions.
diff --git a/libbeat/dashboards/decode.go b/libbeat/dashboards/decode.go
@@ -47,25 +47,31 @@ func DecodeExported(exported []byte) []byte {
 		line, err := r.ReadBytes('\n')
 		if err != nil {
 			if err == io.EOF {
-				return result
+				return append(result, decodeLine(line)...)
 			}
 			return exported
 		}
-		o := common.MapStr{}
-		err = json.Unmarshal(line, &o)
+		result = append(result, decodeLine(line)...)
+	}
+}
+
+func decodeLine(line []byte) []byte {
+	o := common.MapStr{}
+	err := json.Unmarshal(line, &o)
+	if err != nil {
+		return line
+	}
+	var result []byte
+	for _, key := range responseToDecode {
+		// All fields are optional, so errors are not caught
+		err := decodeValue(o, key)
 		if err != nil {
-			continue
-		}
-		for _, key := range responseToDecode {
-			// All fields are optional, so errors are not caught
-			err := decodeValue(o, key)
-			if err != nil {
-				logger := logp.NewLogger("dashboards")
-				logger.Debugf("Error while decoding dashboard objects: %+v", err)
-			}
-			result = append(result, []byte(o.String())...)
+			logger := logp.NewLogger("dashboards")
+			logger.Debugf("Error while decoding dashboard objects: %+v", err)
 		}
+		result = append(result, []byte(o.String())...)
 	}
+	return result
 }
 
 func decodeValue(data common.MapStr, key string) error {

diff --git a/libbeat/dashboards/modify_json.go b/libbeat/dashboards/modify_json.go
@@ -181,40 +181,49 @@ func ReplaceIndexInDashboardObject(index string, content []byte) []byte {
 		line, err := r.ReadBytes('\n')
 		if err != nil {
 			if err == io.EOF {
-				return result
+				return append(result, replaceInNDJSON(logger, index, line)...)
 			}
 			logger.Error("Error reading bytes from raw dashboard object: %+v", err)
 			return content
 		}
-		objectMap := make(map[string]interface{}, 0)
-		err = json.Unmarshal(line, &objectMap)
-		if err != nil {
-			result = append(result, append(line, newline...)...)
-			continue
-		}
+		result = append(result, replaceInNDJSON(logger, index, line)...)
+	}
+}
 
-		attributes, ok := objectMap["attributes"].(map[string]interface{})
-		if !ok {
-			result = append(result, append(line, newline...)...)
-			continue
-		}
+func replaceInNDJSON(logger *logp.Logger, index string, line []byte) []byte {
+	if len(line) == 0 {
+		return line
+	}
 
-		if kibanaSavedObject, ok := attributes["kibanaSavedObjectMeta"].(map[string]interface{}); ok {
-			attributes["kibanaSavedObjectMeta"] = ReplaceIndexInSavedObject(logger, index, kibanaSavedObject)
-		}
+	objectMap := make(map[string]interface{}, 0)
+	err := json.Unmarshal(line, &objectMap)
+	if err != nil {
+		logger.Errorf("Failed to convert bytes to map[string]interface: %+v", err)
+		return line
+	}
 
-		if visState, ok := attributes["visState"].(string); ok {
-			attributes["visState"] = ReplaceIndexInVisState(logger, index, visState)
-		}
+	attributes, ok := objectMap["attributes"].(map[string]interface{})
+	if !ok {
+		logger.Errorf("Object does not have attributes key")
+		return line
+	}
 
-		b, err := json.Marshal(objectMap)
-		if err != nil {
-			logger.Error("Error marshaling modified dashboard: %+v", err)
-			result = append(result, append(line, newline...)...)
-		}
+	if kibanaSavedObject, ok := attributes["kibanaSavedObjectMeta"].(map[string]interface{}); ok {
+		attributes["kibanaSavedObjectMeta"] = ReplaceIndexInSavedObject(logger, index, kibanaSavedObject)
+	}
 
-		result = append(result, append(b, newline...)...)
+	if visState, ok := attributes["visState"].(string); ok {
+		attributes["visState"] = ReplaceIndexInVisState(logger, index, visState)
 	}
+
+	b, err := json.Marshal(objectMap)
+	if err != nil {
+		logger.Error("Error marshaling modified dashboard: %+v", err)
+		return line
+	}
+
+	return append(b, newline...)
+
 }
 
 // ReplaceStringInDashboard replaces a string field in a dashboard

diff --git a/libbeat/kibana/dashboard.go b/libbeat/kibana/dashboard.go
@@ -35,23 +35,37 @@ func RemoveIndexPattern(data []byte) ([]byte, error) {
 		line, err := r.ReadBytes('\n')
 		if err != nil {
 			if err == io.EOF {
-				return result, nil
+				res, removeErr := removeLineIfIndexPattern(line)
+				if removeErr != nil {
+					return data, removeErr
+				}
+				return append(result, res...), nil
 			}
 			return data, err
 		}
 
-		var r common.MapStr
-		// Full struct need to not loose any data
-		err = json.Unmarshal(line, &r)
+		res, err := removeLineIfIndexPattern(line)
 		if err != nil {
-			return nil, err
-		}
-		v, err := r.GetValue("type")
-		if err != nil {
-			return nil, fmt.Errorf("type key not found or not string")
-		}
-		if v != "index-pattern" {
-			result = append(result, line...)
+			return data, err
 		}
+		result = append(result, res...)
+	}
+}
+
+func removeLineIfIndexPattern(line []byte) ([]byte, error) {
+	var r common.MapStr
+	// Full struct need to not loose any data
+	err := json.Unmarshal(line, &r)
+	if err != nil {
+		return nil, err
+	}
+	v, err := r.GetValue("type")
+	if err != nil {
+		return nil, fmt.Errorf("type key not found or not string")
 	}
+	if v != "index-pattern" {
+		return line, nil
+	}
+
+	return nil, nil
 }