Repository: VictoriaMetrics/VictoriaMetrics
Pull request #2032 (2022-01-04 07:44:29)

vmstorage: fix query like `{foo=~"bar|"}` return extra timeseries cau…
…se by negative filter transformation malfunction

1. L2749 make kb.B remain the value of comonPrefix instead of tf.prefix
2. L2762 avoid change tf.value from "bar|" to ".+r|"

waldoweng: @valyala please take a look.

Patch:
diff --git a/.github/workflows/check-rebased.sh b/.github/workflows/check-rebased.sh
deleted file mode 100755
index 2c69bcda5..000000000
--- a/.github/workflows/check-rebased.sh
+++ /dev/null
@@ -1,26 +0,0 @@
-#!/usr/bin/env bash
-set -x -e -u
-
-git fetch origin "$GITHUB_BASE_REF"
-
-BASE_ID=$(git rev-parse --verify "origin/$GITHUB_BASE_REF")
-CURRENT_ID=$(git rev-parse --verify "HEAD")
-
-# Oldest commit that's not in the base branch
-FIRST_BRANCH_ID=$(git rev-list "origin/$GITHUB_BASE_REF..HEAD" | tail -n 1)
-
-# Common ancestor (usually just parent of $FIRST_BRANCH_ID)
-# Command will return error 1 if not found anything. So we || true to proceed.
-FORK_POINT_ID=$(git merge-base "$FIRST_BRANCH_ID" "origin/$GITHUB_BASE_REF") || true
-
-if [[ -z "$FORK_POINT_ID" ]]
-then
-  echo "Current branch is not forked from its base branch origin/$GITHUB_BASE_REF"
-  exit 1
-fi
-
-if [[ "$BASE_ID" != "$FORK_POINT_ID" ]]
-then
-  echo "Current branch (at $CURRENT_ID) forked at $FORK_POINT_ID is not $BASE_ID (which is the latest \"$GITHUB_BASE_REF\")"
-  exit 1
-fi
diff --git a/.github/workflows/check-rebased.yml b/.github/workflows/check-rebased.yml
deleted file mode 100644
index d88bf3a55..000000000
--- a/.github/workflows/check-rebased.yml
+++ /dev/null
@@ -1,17 +0,0 @@
-name: check-rebased
-on:
-  pull_request:
-    types: [ synchronize, opened, reopened, edited ] # +edited (for triggering on PR base change)
-  push:
-    branches: [ main ]
-jobs:
-  hook:
-    name: Check whether current branch is based on its base branch
-    runs-on: ubuntu-latest
-    steps:
-      - name: Check out repo @ current branch
-        uses: actions/checkout@v2
-        with:
-          fetch-depth: 1000 # Hopefully current branch is less than 1000 commits from main
-      - name: Run check-rebased.sh
-        run: .github/workflows/check-rebased.sh
diff --git a/app/vmctl/opentsdb.go b/app/vmctl/opentsdb.go
index 2018c9a63..f3532a4ca 100644
--- a/app/vmctl/opentsdb.go
+++ b/app/vmctl/opentsdb.go
@@ -145,7 +145,7 @@ func (op *otsdbProcessor) run(silent, verbose bool) error {
 func (op *otsdbProcessor) do(s queryObj) error {
 	start := s.StartTime - s.Tr.Start
 	end := s.StartTime - s.Tr.End
-	data, err := op.oc.GetData(s.Series, s.Rt, start, end)
+	data, err := op.oc.GetData(s.Series, s.Rt, start, end, op.oc.MsecsTime)
 	if err != nil {
 		return fmt.Errorf("failed to collect data for %v in %v:%v :: %v", s.Series, s.Rt, s.Tr, err)
 	}
diff --git a/app/vmctl/opentsdb/opentsdb.go b/app/vmctl/opentsdb/opentsdb.go
index 06684f039..df446fc43 100644
--- a/app/vmctl/opentsdb/opentsdb.go
+++ b/app/vmctl/opentsdb/opentsdb.go
@@ -46,6 +46,7 @@ type Client struct {
 	Filters    []string
 	Normalize  bool
 	HardTS     int64
+	MsecsTime  bool
 }
 
 // Config contains fields required
@@ -82,9 +83,9 @@ type MetaResults struct {
 // Meta A meta object about a metric
 // only contain the tags/etc. and no data
 type Meta struct {
-	//tsuid  string
 	Metric string            `json:"metric"`
 	Tags   map[string]string `json:"tags"`
+	//tsuid  string
 }
 
 // OtsdbMetric is a single series in OpenTSDB's returned format
@@ -152,7 +153,7 @@ func (c Client) FindSeries(metric string) ([]Meta, error) {
 
 // GetData actually retrieves data for a series at a specified time range
 // e.g. /api/query?start=1&end=200&m=sum:1m-avg-none:system.load5{host=host1}
-func (c Client) GetData(series Meta, rt RetentionMeta, start int64, end int64) (Metric, error) {
+func (c Client) GetData(series Meta, rt RetentionMeta, start int64, end int64, mSecs bool) (Metric, error) {
 	/*
 		First, build our tag string.
 		It's literally just key=value,key=value,...
@@ -195,7 +196,7 @@ func (c Client) GetData(series Meta, rt RetentionMeta, start int64, end int64) (
 		3. bad format of response body
 	*/
 	if resp.StatusCode != 200 {
-		log.Println(fmt.Sprintf("bad response code from OpenTSDB query %v...skipping", resp.StatusCode))
+		log.Println(fmt.Sprintf("bad response code from OpenTSDB query %v for %q...skipping", resp.StatusCode, q))
 		return Metric{}, nil
 	}
 	defer func() { _ = resp.Body.Close() }()
@@ -272,7 +273,11 @@ func (c Client) GetData(series Meta, rt RetentionMeta, start int64, end int64) (
 		then convert the timestamp back to something reasonable.
 	*/
 	for ts, val := range output[0].Dps {
-		data.Timestamps = append(data.Timestamps, ts)
+		if !mSecs {
+			data.Timestamps = append(data.Timestamps, ts*1000)
+		} else {
+			data.Timestamps = append(data.Timestamps, ts)
+		}
 		data.Values = append(data.Values, val)
 	}
 	return data, nil
@@ -283,6 +288,7 @@ func (c Client) GetData(series Meta, rt RetentionMeta, start int64, end int64) (
 func NewClient(cfg Config) (*Client, error) {
 	var retentions []Retention
 	offsetPrint := int64(time.Now().Unix())
+	// convert a number of days to seconds
 	offsetSecs := cfg.Offset * 24 * 60 * 60
 	if cfg.MsecsTime {
 		// 1000000 == Nanoseconds -> Milliseconds difference
@@ -318,6 +324,7 @@ func NewClient(cfg Config) (*Client, error) {
 		Filters:    cfg.Filters,
 		Normalize:  cfg.Normalize,
 		HardTS:     cfg.HardTS,
+		MsecsTime:  cfg.MsecsTime,
 	}
 	return client, nil
 }
diff --git a/app/vmctl/opentsdb/parser.go b/app/vmctl/opentsdb/parser.go
index a46312920..6a1fa8baf 100644
--- a/app/vmctl/opentsdb/parser.go
+++ b/app/vmctl/opentsdb/parser.go
@@ -87,6 +87,34 @@ func convertRetention(retention string, offset int64, msecTime bool) (Retention,
 	if len(chunks) != 3 {
 		return Retention{}, fmt.Errorf("invalid retention string: %q", retention)
 	}
+	queryLengthDuration, err := convertDuration(chunks[2])
+	if err != nil {
+		return Retention{}, fmt.Errorf("invalid ttl (second order) duration string: %q: %s", chunks[2], err)
+	}
+	// set ttl in milliseconds, unless we aren't using millisecond time in OpenTSDB...then use seconds
+	queryLength := queryLengthDuration.Milliseconds()
+	if !msecTime {
+		queryLength = queryLength / 1000
+	}
+	queryRange := queryLength
+	// bump by the offset so we don't look at empty ranges any time offset > ttl
+	queryLength += offset
+
+	// first/second order aggregations for queries defined in chunk 0...
+	aggregates := strings.Split(chunks[0], "-")
+	if len(aggregates) != 3 {
+		return Retention{}, fmt.Errorf("invalid aggregation string: %q", chunks[0])
+	}
+
+	aggTimeDuration, err := convertDuration(aggregates[1])
+	if err != nil {
+		return Retention{}, fmt.Errorf("invalid aggregation time duration string: %q: %s", aggregates[1], err)
+	}
+	aggTime := aggTimeDuration.Milliseconds()
+	if !msecTime {
+		aggTime = aggTime / 1000
+	}
+
 	rowLengthDuration, err := convertDuration(chunks[1])
 	if err != nil {
 		return Retention{}, fmt.Errorf("invalid row length (first order) duration string: %q: %s", chunks[1], err)
@@ -96,27 +124,35 @@ func convertRetention(retention string, offset int64, msecTime bool) (Retention,
 	if !msecTime {
 		rowLength = rowLength / 1000
 	}
-	ttlDuration, err := convertDuration(chunks[2])
-	if err != nil {
-		return Retention{}, fmt.Errorf("invalid ttl (second order) duration string: %q: %s", chunks[2], err)
-	}
-	// set ttl in milliseconds, unless we aren't using millisecond time in OpenTSDB...then use seconds
-	ttl := ttlDuration.Milliseconds()
-	if !msecTime {
-		ttl = ttl / 1000
+
+	var querySize int64
+	/*
+		The idea here is to ensure each individual query sent to OpenTSDB is *at least*
+		large enough to ensure no single query requests essentially 0 data.
+	*/
+	if rowLength > aggTime {
+		/*
+			We'll look at 2x the row size for each query we perform
+			This is a strange function, but the logic works like this:
+			1. we discover the "number" of ranges we should split the time range into
+			   This is found with queryRange / (rowLength * 4)...kind of a percentage query
+			2. we discover the actual size of each "chunk"
+			   This is second division step
+		*/
+		querySize = int64(queryRange / (queryRange / (rowLength * 4)))
+	} else {
+		/*
+			Unless the aggTime (how long a range of data we're requesting per individual point)
+			is greater than the row size. Then we'll need to use that to determine
+			how big each individual query should be
+		*/
+		querySize = int64(queryRange / (queryRange / (aggTime * 4)))
 	}
-	// bump by the offset so we don't look at empty ranges any time offset > ttl
-	ttl += offset
 
 	var timeChunks []TimeRange
 	var i int64
-	for i = offset; i <= ttl; i = i + rowLength {
-		timeChunks = append(timeChunks, TimeRange{Start: i + rowLength, End: i})
-	}
-	// first/second order aggregations for queries defined in chunk 0...
-	aggregates := strings.Split(chunks[0], "-")
-	if len(aggregates) != 3 {
-		return Retention{}, fmt.Errorf("invalid aggregation string: %q", chunks[0])
+	for i = offset; i <= queryLength; i = i + querySize {
+		timeChunks = append(timeChunks, TimeRange{Start: i + querySize, End: i})
 	}
 
 	ret := Retention{FirstOrder: aggregates[0],
diff --git a/docs/CHANGELOG.md b/docs/CHANGELOG.md
index 5a5cd3b90..c51f39403 100644
--- a/docs/CHANGELOG.md
+++ b/docs/CHANGELOG.md
@@ -12,6 +12,7 @@ sort: 15
 
 * BUGFIX: [vmagent](https://docs.victoriametrics.com/vmagent.html): make sure that `vmagent` replicas scrape the same targets at different time offsets when [replication is enabled in vmagent clustering mode](https://docs.victoriametrics.com/vmagent.html#scraping-big-number-of-targets). This guarantees that the [deduplication](https://docs.victoriametrics.com/#deduplication) consistently leaves samples from the same `vmagent` replica.
 * BUGFIX: return the proper response stub from `/api/v1/query_exemplars` handler, which is needed for Grafana v8+. See [this issue](https://github.com/VictoriaMetrics/VictoriaMetrics/issues/1999).
+* BUGFIX: [vmctl](https://docs.victoriametrics.com/vmctl.html): fix a few edge cases and improve migration speed for OpenTSDB importer. See [this pull request](https://github.com/VictoriaMetrics/VictoriaMetrics/pull/2019).
 
 
 ## [v1.71.0](https://github.com/VictoriaMetrics/VictoriaMetrics/releases/tag/v1.71.0)
diff --git a/lib/storage/index_db.go b/lib/storage/index_db.go
index bc959f113..544777646 100644
--- a/lib/storage/index_db.go
+++ b/lib/storage/index_db.go
@@ -2744,10 +2744,9 @@ func (is *indexSearch) getMetricIDsForDateTagFilter(tf *tagFilter, date uint64,
 		// Use global search if date isn't set.
 		kb.B = is.marshalCommonPrefix(kb.B[:0], nsPrefixTagToMetricIDs)
 	}
-	kb.B = append(kb.B, tf.prefix[len(commonPrefix):]...)
 	tfNew := *tf
 	tfNew.isNegative = false // isNegative for the original tf is handled by the caller.
-	tfNew.prefix = kb.B
+	tfNew.prefix = append(kb.B, tf.prefix[len(commonPrefix):]...)
 	metricIDs, loopsCount, err := is.getMetricIDsForTagFilter(&tfNew, maxMetrics, maxLoopsCount)
 	if err != nil {
 		return nil, loopsCount, err
@@ -2760,6 +2759,7 @@ func (is *indexSearch) getMetricIDsForDateTagFilter(tf *tagFilter, date uint64,
 	// This fixes https://github.com/VictoriaMetrics/VictoriaMetrics/issues/1601
 	// See also https://github.com/VictoriaMetrics/VictoriaMetrics/issues/395
 	maxLoopsCount -= loopsCount
+	tfNew = tagFilter{}
 	if err := tfNew.Init(kb.B, tf.key, []byte(".+"), false, true); err != nil {
 		logger.Panicf(`BUG: cannot init tag filter: {%q=~".+"}: %s`, tf.key, err)
 	}


Test patch:
diff --git a/app/vmctl/opentsdb/parser_test.go b/app/vmctl/opentsdb/parser_test.go
index 0aa93b7cd..a5adec540 100644
--- a/app/vmctl/opentsdb/parser_test.go
+++ b/app/vmctl/opentsdb/parser_test.go
@@ -8,7 +8,7 @@ func TestConvertRetention(t *testing.T) {
 	/*
 		2592000 seconds in 30 days
 		3600 in one hour
-		2592000 / 3600 = 720 individual query "ranges" should exist, plus one because time ranges can be weird
+		2592000 / 14400 = 180 individual query "ranges" should exist, plus one because time ranges can be weird
 		First order should == "sum"
 		Second order should == "avg"
 		AggTime should == "1m"
@@ -17,8 +17,8 @@ func TestConvertRetention(t *testing.T) {
 	if err != nil {
 		t.Fatalf("Error parsing valid retention string: %v", err)
 	}
-	if len(res.QueryRanges) != 721 {
-		t.Fatalf("Found %v query ranges. Should have found 720", len(res.QueryRanges))
+	if len(res.QueryRanges) != 181 {
+		t.Fatalf("Found %v query ranges. Should have found 181", len(res.QueryRanges))
 	}
 	if res.FirstOrder != "sum" {
 		t.Fatalf("Incorrect first order aggregation %q. Should have been 'sum'", res.FirstOrder)