Merge pull request #226 from jpinsonneau/334

NETOBSERV-334 concurrent map iteration & write fix

pkg/config/config.go

@@ -24,8 +24,6 @@ import (
 	"github.com/sirupsen/logrus"
 )
 
-type GenericMap map[string]interface{}
-
 var (
 	Opt        = Options{}
 	PipeLine   []Stage

pkg/config/generic_map.go

@@ -0,0 +1,31 @@
+/*
+ * Copyright (C) 2022 IBM, Inc.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ *
+ */
+
+package config
+
+type GenericMap map[string]interface{}
+
+// Copy will create a flat copy of GenericMap
+func (m GenericMap) Copy() GenericMap {
+	result := GenericMap{}
+
+	for k, v := range m {
+		result[k] = v
+	}
+
+	return result
+}

pkg/pipeline/write/write_loki.go

@@ -109,8 +109,11 @@ func buildLokiConfig(c *api.WriteLoki) (loki.Config, error) {
 }
 
 func (l *Loki) ProcessRecord(record config.GenericMap) error {
+	// copy record before process to avoid alteration on parallel stages
+	recordCopy := record.Copy()
+
 	// Get timestamp from record (default: TimeFlowStart)
-	timestamp := l.extractTimestamp(record)
+	timestamp := l.extractTimestamp(recordCopy)
 
 	labels := model.LabelSet{}
 
@@ -119,15 +122,15 @@ func (l *Loki) ProcessRecord(record config.GenericMap) error {
 		labels[k] = v
 	}
 
-	l.addNonStaticLabels(record, labels)
+	l.addNonStaticLabels(recordCopy, labels)
 
 	// Remove labels and configured ignore list from record
 	ignoreList := append(l.apiConfig.IgnoreList, l.apiConfig.Labels...)
 	for _, label := range ignoreList {
-		delete(record, label)
+		delete(recordCopy, label)
 	}
 
-	js, err := jsonIter.ConfigCompatibleWithStandardLibrary.Marshal(record)
+	js, err := jsonIter.ConfigCompatibleWithStandardLibrary.Marshal(recordCopy)
 	if err != nil {
 		return err
 	}