feat(operators): support subset checking for in and notin (#1555)

* feat(operators): support subset checking for in and notin

Signed-off-by: Arsh Sharma <arshsharma461@gmail.com>

* feat(operators): fixed NotIn function

Signed-off-by: Arsh Sharma <arshsharma461@gmail.com>

pkg/engine/variables/operator/in.go

@@ -103,6 +103,80 @@ func keyExistsInArray(key string, value interface{}, log logr.Logger) (invalidTy
 	return false, false
 }
 
+func (in InHandler) validateValueWithStringSetPattern(key []string, value interface{}) (keyExists bool) {
+	invalidType, keyExists := setExistsInArray(key, value, in.log, false)
+	if invalidType {
+		in.log.Info("expected type []string", "value", value, "type", fmt.Sprintf("%T", value))
+		return false
+	}
+
+	return keyExists
+}
+
+// setExistsInArray checks if the key is a subset of value
+// The value can be a string, an array of strings, or a JSON format
+// array of strings (e.g. ["val1", "val2", "val3"].
+// notIn argument is set to true when we want to check if key is NOT a subset of value
+func setExistsInArray(key []string, value interface{}, log logr.Logger, notIn bool) (invalidType bool, keyExists bool) {
+	switch valuesAvailable := value.(type) {
+
+	case []interface{}:
+		var valueSlice []string
+		for _, val := range valuesAvailable {
+			v, ok := val.(string)
+			if !ok {
+				return true, false
+			}
+			valueSlice = append(valueSlice, v)
+		}
+		if notIn {
+			return false, checkInSubsetForNotIn(key, valueSlice)
+		}
+		return false, checkInSubset(key, valueSlice)
+
+	case string:
+
+		if len(key) == 1 && key[0] == valuesAvailable {
+			return false, true
+		}
+
+		var arr []string
+		if err := json.Unmarshal([]byte(valuesAvailable), &arr); err != nil {
+			log.Error(err, "failed to unmarshal value to JSON string array", "key", key, "value", value)
+			return true, false
+		}
+		if notIn {
+			return false, checkInSubsetForNotIn(key, arr)
+		}
+		return false, checkInSubset(key, arr)
+
+	default:
+		return true, false
+	}
+}
+
+// checkInSubset checks if ALL values of S1 are in S2
+func checkInSubset(key []string, value []string) bool {
+	set := make(map[string]int)
+
+	for _, val := range value {
+		set[val]++
+	}
+
+	for _, val := range key {
+		count, found := set[val]
+		if !found {
+			return false
+		} else if count < 1 {
+			return false
+		} else {
+			set[val] = count - 1
+		}
+	}
+
+	return true
+}
+
 func (in InHandler) validateValueWithBoolPattern(_ bool, _ interface{}) bool {
 	return false
 }

pkg/engine/variables/operator/notin.go

@@ -57,6 +57,34 @@ func (nin NotInHandler) validateValueWithStringPattern(key string, value interfa
 	return !keyExists
 }
 
+func (nin NotInHandler) validateValueWithStringSetPattern(key []string, value interface{}) bool {
+	invalidType, keyExists := setExistsInArray(key, value, nin.log, true)
+	if invalidType {
+		nin.log.Info("expected type []string", "value", value, "type", fmt.Sprintf("%T", value))
+		return false
+	}
+
+	return !keyExists
+}
+
+// checkInSubsetForNotIn checks if ANY of the values of S1 is in S2
+func checkInSubsetForNotIn(key []string, value []string) bool {
+	set := make(map[string]int)
+
+	for _, val := range value {
+		set[val]++
+	}
+
+	for _, val := range key {
+		_, found := set[val]
+		if found {
+			return true
+		}
+	}
+
+	return false
+}
+
 func (nin NotInHandler) validateValueWithBoolPattern(_ bool, _ interface{}) bool {
 	return false
 }