NETOBSERV-2186: Set sampling in flows

Author: jotak

pkg/pipeline/transform/transform_filter.go

@@ -190,13 +190,22 @@ func applyPredicates(entry config.GenericMap, rule predicatesRule) bool {
 			return false
 		}
 	}
+	if rule.sampling > 0 {
+		storeSampling(entry, int(rule.sampling))
+	}
 	return true
 }
 
 func sample(entry config.GenericMap, rules []*api.SamplingCondition) bool {
 	for _, r := range rules {
 		if isRemoveEntrySatisfied(entry, r.Rules) {
-			return rollSampling(r.Value)
+			if rollSampling(r.Value) {
+				if r.Value > 0 {
+					storeSampling(entry, int(r.Value))
+				}
+				return true
+			}
+			return false
 		}
 	}
 	return true
@@ -206,6 +215,20 @@ func rollSampling(value uint16) bool {
 	return value == 0 || (rndgen.Intn(int(value)) == 0)
 }
 
+func storeSampling(entry config.GenericMap, value int) {
+	// TODO: make "Sampling" string configurable, and configured from the operator.
+	if current, found := entry["Sampling"]; found {
+		if cast, err := utils.ConvertToInt(current); err != nil {
+			if cast == 0 {
+				cast = 1
+			}
+			entry["Sampling"] = cast * value
+		}
+	} else {
+		entry["Sampling"] = value
+	}
+}
+
 // NewTransformFilter create a new filter transform
 func NewTransformFilter(params config.StageParam) (Transformer, error) {
 	tlog.Debugf("entering NewTransformFilter")