log/logstorage/ipv6: Support ipv6_range for VictoriaLogs

lib/logstorage/filter_ipv6_range.go

@@ -0,0 +1,144 @@
+package logstorage
+
+import (
+	"fmt"
+	"net"
+
+	"github.com/VictoriaMetrics/VictoriaLogs/lib/prefixfilter"
+)
+
+// filterIPv6Range matches the given ipv6 range [minValue..maxValue].
+//
+// Example LogsQL: `fieldName:ipv6_range(::1, ::2)`
+type filterIPv6Range struct {
+	fieldName string
+	minValue  string
+	maxValue  string
+}
+
+func (fr *filterIPv6Range) String() string {
+	minIP := net.IP([]byte(fr.minValue))
+	maxIP := net.IP([]byte(fr.maxValue))
+	return fmt.Sprintf("%sipv6_range(%s, %s)", quoteFieldNameIfNeeded(fr.fieldName), minIP.String(), maxIP.String())
+}
+
+func (fr *filterIPv6Range) updateNeededFields(pf *prefixfilter.Filter) {
+	pf.AddAllowFilter(fr.fieldName)
+}
+
+func (fr *filterIPv6Range) matchRow(fields []Field) bool {
+	v := getFieldValueByName(fields, fr.fieldName)
+	return matchIPv6Range(v, fr.minValue, fr.maxValue)
+}
+
+func (fr *filterIPv6Range) applyToBlockResult(br *blockResult, bm *bitmap) {
+	minValue := fr.minValue
+	maxValue := fr.maxValue
+
+	if minValue > maxValue {
+		bm.resetBits()
+		return
+	}
+
+	c := br.getColumnByName(fr.fieldName)
+	if c.isConst {
+		v := c.valuesEncoded[0]
+		if !matchIPv6Range(v, minValue, maxValue) {
+			bm.resetBits()
+		}
+		return
+	}
+	if c.isTime {
+		bm.resetBits()
+		return
+	}
+
+	switch c.valueType {
+	case valueTypeString:
+		values := c.getValues(br)
+		bm.forEachSetBit(func(idx int) bool {
+			v := values[idx]
+			return matchIPv6Range(v, minValue, maxValue)
+		})
+	case valueTypeDict:
+		bb := bbPool.Get()
+		for _, v := range c.dictValues {
+			c := byte(0)
+			if matchIPv6Range(v, minValue, maxValue) {
+				c = 1
+			}
+			bb.B = append(bb.B, c)
+		}
+		valuesEncoded := c.getValuesEncoded(br)
+		bm.forEachSetBit(func(idx int) bool {
+			n := valuesEncoded[idx][0]
+			return bb.B[n] == 1
+		})
+		bbPool.Put(bb)
+	default:
+		bm.resetBits()
+	}
+}
+
+func (fr *filterIPv6Range) applyToBlockSearch(bs *blockSearch, bm *bitmap) {
+	fieldName := fr.fieldName
+	minValue := fr.minValue
+	maxValue := fr.maxValue
+
+	if minValue > maxValue {
+		bm.resetBits()
+		return
+	}
+
+	v := bs.getConstColumnValue(fieldName)
+	if v != "" {
+		if !matchIPv6Range(v, minValue, maxValue) {
+			bm.resetBits()
+		}
+		return
+	}
+
+	// Verify whether filter matches other columns
+	ch := bs.getColumnHeader(fieldName)
+	if ch == nil {
+		// Fast path - there are no matching columns.
+		bm.resetBits()
+		return
+	}
+
+	switch ch.valueType {
+	case valueTypeString:
+		matchStringByIPv6Range(bs, ch, bm, minValue, maxValue)
+	case valueTypeDict:
+		matchValuesDictByIPv6Range(bs, ch, bm, minValue, maxValue)
+	default:
+		bm.resetBits()
+	}
+}
+
+func matchValuesDictByIPv6Range(bs *blockSearch, ch *columnHeader, bm *bitmap, minValue, maxValue string) {
+	bb := bbPool.Get()
+	for _, v := range ch.valuesDict.values {
+		c := byte(0)
+		if matchIPv6Range(v, minValue, maxValue) {
+			c = 1
+		}
+		bb.B = append(bb.B, c)
+	}
+	matchEncodedValuesDict(bs, ch, bm, bb.B)
+	bbPool.Put(bb)
+}
+
+func matchStringByIPv6Range(bs *blockSearch, ch *columnHeader, bm *bitmap, minValue, maxValue string) {
+	visitValues(bs, ch, bm, func(v string) bool {
+		return matchIPv6Range(v, minValue, maxValue)
+	})
+}
+
+func matchIPv6Range(s string, minValue, maxValue string) bool {
+	ip, ok := tryParseIPv6(s)
+	if !ok {
+		return false
+	}
+	return ip >= minValue && ip <= maxValue
+}

lib/logstorage/filter_ipv6_range_test.go

@@ -0,0 +1,182 @@
+package logstorage
+
+import (
+	"net"
+	"testing"
+
+	"github.com/VictoriaMetrics/VictoriaMetrics/lib/fs"
+)
+
+func TestMatchIPv6Range(t *testing.T) {
+	t.Parallel()
+
+	f := func(s, minValue, maxValue string, resultExpected bool) {
+		t.Helper()
+		minIP := string(net.ParseIP(minValue).To16())
+		maxIP := string(net.ParseIP(maxValue).To16())
+		result := matchIPv6Range(s, minIP, maxIP)
+		if result != resultExpected {
+			t.Fatalf("unexpected result; got %v; want %v", result, resultExpected)
+		}
+	}
+
+	// Invalid IP
+	f("", "::1", "::2", false)
+	f("123", "::1", "::2", false)
+	f("1.2.3.4", "::1", "::2", false)
+
+	// range mismatch
+	f("::1", "::2", "::3", false)
+	f("2001:db8::1", "2001:db8::2", "2001:db8::3", false)
+
+	// range match
+	f("::1", "::1", "::1", true)
+	f("::1", "::0", "::2", true)
+	f("2001:db8::1", "2001:db8::", "2001:db8::ffff", true)
+}
+
+func TestFilterIPv6Range(t *testing.T) {
+	t.Parallel()
+
+	parseIP := func(s string) string {
+		return string(net.ParseIP(s).To16())
+	}
+
+	t.Run("const-column", func(t *testing.T) {
+		columns := []column{
+			{
+				name: "foo",
+				values: []string{
+					"::1",
+					"::1",
+					"::1",
+				},
+			},
+		}
+
+		// match
+		fr := &filterIPv6Range{
+			fieldName: "foo",
+			minValue:  parseIP("::0"),
+			maxValue:  parseIP("::2"),
+		}
+		testFilterMatchForColumns(t, columns, fr, "foo", []int{0, 1, 2})
+
+		fr = &filterIPv6Range{
+			fieldName: "foo",
+			minValue:  parseIP("::1"),
+			maxValue:  parseIP("::1"),
+		}
+		testFilterMatchForColumns(t, columns, fr, "foo", []int{0, 1, 2})
+
+		// mismatch
+		fr = &filterIPv6Range{
+			fieldName: "foo",
+			minValue:  parseIP("::2"),
+			maxValue:  parseIP("::3"),
+		}
+		testFilterMatchForColumns(t, columns, fr, "foo", nil)
+
+		fr = &filterIPv6Range{
+			fieldName: "non-existing-column",
+			minValue:  parseIP("::0"),
+			maxValue:  parseIP("::ffff"),
+		}
+		testFilterMatchForColumns(t, columns, fr, "foo", nil)
+
+		fr = &filterIPv6Range{
+			fieldName: "foo",
+			minValue:  parseIP("::2"),
+			maxValue:  parseIP("::0"),
+		}
+		testFilterMatchForColumns(t, columns, fr, "foo", nil)
+	})
+
+	t.Run("dict", func(t *testing.T) {
+		columns := []column{
+			{
+				name: "foo",
+				values: []string{
+					"",
+					"::1",
+					"Abc",
+					"2001:db8::1",
+					"10.4",
+					"foo ::1",
+					"::1 bar",
+					"::1",
+				},
+			},
+		}
+
+		// match
+		fr := &filterIPv6Range{
+			fieldName: "foo",
+			minValue:  parseIP("::0"),
+			maxValue:  parseIP("::2"),
+		}
+		testFilterMatchForColumns(t, columns, fr, "foo", []int{1, 7})
+
+		fr = &filterIPv6Range{
+			fieldName: "foo",
+			minValue:  parseIP("2001:db8::"),
+			maxValue:  parseIP("2001:db8::ffff"),
+		}
+		testFilterMatchForColumns(t, columns, fr, "foo", []int{3})
+
+		// mismatch
+		fr = &filterIPv6Range{
+			fieldName: "foo",
+			minValue:  parseIP("::3"),
+			maxValue:  parseIP("::4"),
+		}
+		testFilterMatchForColumns(t, columns, fr, "foo", nil)
+	})
+
+	t.Run("strings", func(t *testing.T) {
+		columns := []column{
+			{
+				name: "foo",
+				values: []string{
+					"A FOO",
+					"a 10",
+					"::1",
+					"20",
+					"15.5",
+					"-5",
+					"a fooBaR",
+					"a ::1 dfff",
+					"a ТЕСТЙЦУК НГКШ ",
+					"a !!,23.(!1)",
+					"2001:db8::1",
+				},
+			},
+		}
+
+		// match
+		fr := &filterIPv6Range{
+			fieldName: "foo",
+			minValue:  parseIP("::0"),
+			maxValue:  parseIP("::2"),
+		}
+		testFilterMatchForColumns(t, columns, fr, "foo", []int{2})
+
+		fr = &filterIPv6Range{
+			fieldName: "foo",
+			minValue:  parseIP("2001:db8::"),
+			maxValue:  parseIP("2001:db8::ffff"),
+		}
+		testFilterMatchForColumns(t, columns, fr, "foo", []int{10})
+
+		// mismatch
+		fr = &filterIPv6Range{
+			fieldName: "foo",
+			minValue:  parseIP("::3"),
+			maxValue:  parseIP("::4"),
+		}
+		testFilterMatchForColumns(t, columns, fr, "foo", nil)
+	})
+
+	// Remove the remaining data files for the test
+	fs.MustRemoveDir(t.Name())
+}