You are viewing a plain text version of this content. The canonical link for it is here.
Posted to notifications@skywalking.apache.org by ha...@apache.org on 2024/03/03 07:01:53 UTC
(skywalking-banyandb) 01/02: Add more cases to cover entity query
This is an automated email from the ASF dual-hosted git repository.
hanahmily pushed a commit to branch measure-query
in repository https://gitbox.apache.org/repos/asf/skywalking-banyandb.git
commit 9c6f4acdd02baa7b904c9ae678f5e0af98a53504
Author: Gao Hongtao <ha...@gmail.com>
AuthorDate: Sun Mar 3 06:59:18 2024 +0000
Add more cases to cover entity query
Signed-off-by: Gao Hongtao <ha...@gmail.com>
---
banyand/measure/introducer.go | 13 +-
banyand/measure/part_metadata.go | 15 ++
banyand/measure/tstable.go | 19 ++
.../testdata/measures/service_latency_minute.json | 47 +++++
test/cases/init.go | 1 +
test/cases/measure/data/input/all_latency.yaml | 26 +++
.../data/testdata/service_latency_minute_data.json | 206 +++++++++++++++++++++
test/cases/measure/data/want/all_latency.yaml | 120 ++++++++++++
test/cases/measure/measure.go | 1 +
9 files changed, 442 insertions(+), 6 deletions(-)
diff --git a/banyand/measure/introducer.go b/banyand/measure/introducer.go
index e3bb2e01..e371d246 100644
--- a/banyand/measure/introducer.go
+++ b/banyand/measure/introducer.go
@@ -146,7 +146,7 @@ func (tst *tsTable) introduceMemPart(nextIntroduction *introduction, epoch uint6
nextSnp := cur.copyAllTo(epoch)
nextSnp.parts = append(nextSnp.parts, next)
nextSnp.creator = snapshotCreatorMemPart
- tst.replaceSnapshot(&nextSnp)
+ tst.replaceSnapshot(&nextSnp, false)
if nextIntroduction.applied != nil {
close(nextIntroduction.applied)
}
@@ -160,8 +160,7 @@ func (tst *tsTable) introduceFlushed(nextIntroduction *flusherIntroduction, epoc
defer cur.decRef()
nextSnp := cur.merge(epoch, nextIntroduction.flushed)
nextSnp.creator = snapshotCreatorFlusher
- tst.replaceSnapshot(&nextSnp)
- tst.persistSnapshot(&nextSnp)
+ tst.replaceSnapshot(&nextSnp, true)
if nextIntroduction.applied != nil {
close(nextIntroduction.applied)
}
@@ -177,20 +176,22 @@ func (tst *tsTable) introduceMerged(nextIntroduction *mergerIntroduction, epoch
nextSnp := cur.remove(epoch, nextIntroduction.merged)
nextSnp.parts = append(nextSnp.parts, nextIntroduction.newPart)
nextSnp.creator = nextIntroduction.creator
- tst.replaceSnapshot(&nextSnp)
- tst.persistSnapshot(&nextSnp)
+ tst.replaceSnapshot(&nextSnp, true)
if nextIntroduction.applied != nil {
close(nextIntroduction.applied)
}
}
-func (tst *tsTable) replaceSnapshot(next *snapshot) {
+func (tst *tsTable) replaceSnapshot(next *snapshot, persisted bool) {
tst.Lock()
defer tst.Unlock()
if tst.snapshot != nil {
tst.snapshot.decRef()
}
tst.snapshot = next
+ if persisted {
+ tst.persistSnapshot(next)
+ }
}
func (tst *tsTable) currentEpoch() uint64 {
diff --git a/banyand/measure/part_metadata.go b/banyand/measure/part_metadata.go
index 9b64c612..7faaaf12 100644
--- a/banyand/measure/part_metadata.go
+++ b/banyand/measure/part_metadata.go
@@ -21,6 +21,8 @@ import (
"encoding/json"
"path/filepath"
+ "github.com/pkg/errors"
+
"github.com/apache/skywalking-banyandb/pkg/fs"
"github.com/apache/skywalking-banyandb/pkg/logger"
)
@@ -45,6 +47,19 @@ func (pm *partMetadata) reset() {
pm.ID = 0
}
+func validatePartMetadata(fileSystem fs.FileSystem, partPath string) error {
+ metadataPath := filepath.Join(partPath, metadataFilename)
+ metadata, err := fileSystem.Read(metadataPath)
+ if err != nil {
+ return errors.WithMessage(err, "cannot read metadata.json")
+ }
+ var pm partMetadata
+ if err := json.Unmarshal(metadata, &pm); err != nil {
+ return errors.WithMessage(err, "cannot parse metadata.json")
+ }
+ return nil
+}
+
func (pm *partMetadata) mustReadMetadata(fileSystem fs.FileSystem, partPath string) {
pm.reset()
diff --git a/banyand/measure/tstable.go b/banyand/measure/tstable.go
index 356c937a..329c469b 100644
--- a/banyand/measure/tstable.go
+++ b/banyand/measure/tstable.go
@@ -72,6 +72,13 @@ func newTSTable(fileSystem fs.FileSystem, rootPath string, p common.Position,
needToDelete = append(needToDelete, ee[i].Name())
continue
}
+ err = validatePartMetadata(fileSystem, filepath.Join(rootPath, ee[i].Name()))
+ if err != nil {
+ l.Info().Err(err).Msg("cannot validate part metadata. skip and delete it")
+ needToDelete = append(needToDelete, ee[i].Name())
+ continue
+ }
+
loadedParts = append(loadedParts, p)
continue
}
@@ -87,6 +94,7 @@ func newTSTable(fileSystem fs.FileSystem, rootPath string, p common.Position,
loadedSnapshots = append(loadedSnapshots, snapshot)
}
for i := range needToDelete {
+ l.Info().Str("path", filepath.Join(rootPath, needToDelete[i])).Msg("delete invalid directory or file")
if err := fileSystem.DeleteFile(filepath.Join(rootPath, needToDelete[i])); err != nil {
l.Warn().Err(err).Str("path", filepath.Join(rootPath, needToDelete[i])).Msg("failed to delete part. Please check manually")
}
@@ -125,6 +133,7 @@ func (tst *tsTable) loadSnapshot(epoch uint64, loadedParts []uint64) {
snp := snapshot{
epoch: epoch,
}
+ needToPersist := false
for _, id := range loadedParts {
var find bool
for j := range parts {
@@ -136,6 +145,13 @@ func (tst *tsTable) loadSnapshot(epoch uint64, loadedParts []uint64) {
if !find {
tst.gc.submitParts(id)
}
+ err := validatePartMetadata(tst.fileSystem, partPath(tst.root, id))
+ if err != nil {
+ tst.l.Info().Err(err).Uint64("id", id).Msg("cannot validate part metadata. skip and delete it")
+ tst.gc.submitParts(id)
+ needToPersist = true
+ continue
+ }
p := mustOpenFilePart(id, tst.root, tst.fileSystem)
p.partMetadata.ID = id
snp.parts = append(snp.parts, newPartWrapper(nil, p))
@@ -150,6 +166,9 @@ func (tst *tsTable) loadSnapshot(epoch uint64, loadedParts []uint64) {
}
snp.incRef()
tst.snapshot = &snp
+ if needToPersist {
+ tst.persistSnapshot(&snp)
+ }
}
func (tst *tsTable) startLoop(cur uint64) {
diff --git a/pkg/test/measure/testdata/measures/service_latency_minute.json b/pkg/test/measure/testdata/measures/service_latency_minute.json
new file mode 100644
index 00000000..46ddf59c
--- /dev/null
+++ b/pkg/test/measure/testdata/measures/service_latency_minute.json
@@ -0,0 +1,47 @@
+{
+ "metadata": {
+ "name": "service_latency_minute",
+ "group": "sw_metric"
+ },
+ "tag_families": [
+ {
+ "name": "default",
+ "tags": [
+ {
+ "name": "id",
+ "type": "TAG_TYPE_STRING"
+ }
+ ]
+ },
+ {
+ "name": "storage_only",
+ "tags": [
+ {
+ "name": "entity_id",
+ "type": "TAG_TYPE_STRING"
+ }
+ ]
+ }
+ ],
+ "fields": [
+ {
+ "name": "total",
+ "field_type": "FIELD_TYPE_INT",
+ "encoding_method": "ENCODING_METHOD_GORILLA",
+ "compression_method": "COMPRESSION_METHOD_ZSTD"
+ },
+ {
+ "name": "value",
+ "field_type": "FIELD_TYPE_INT",
+ "encoding_method": "ENCODING_METHOD_GORILLA",
+ "compression_method": "COMPRESSION_METHOD_ZSTD"
+ }
+ ],
+ "entity": {
+ "tag_names": [
+ "entity_id"
+ ]
+ },
+ "interval": "1m",
+ "updated_at": "2021-04-15T01:30:15.01Z"
+}
\ No newline at end of file
diff --git a/test/cases/init.go b/test/cases/init.go
index 72f66315..e8fff89d 100644
--- a/test/cases/init.go
+++ b/test/cases/init.go
@@ -50,4 +50,5 @@ func Initialize(addr string, now time.Time) {
casesmeasuredata.Write(conn, "service_instance_endpoint_cpm_minute", "sw_metric", "service_instance_endpoint_cpm_minute_data.json", now, interval)
casesmeasuredata.Write(conn, "service_instance_endpoint_cpm_minute", "sw_metric", "service_instance_endpoint_cpm_minute_data1.json", now.Add(10*time.Second), interval)
casesmeasuredata.Write(conn, "service_instance_endpoint_cpm_minute", "sw_metric", "service_instance_endpoint_cpm_minute_data2.json", now.Add(10*time.Minute), interval)
+ casesmeasuredata.Write(conn, "service_latency_minute", "sw_metric", "service_latency_minute_data.json", now, interval)
}
diff --git a/test/cases/measure/data/input/all_latency.yaml b/test/cases/measure/data/input/all_latency.yaml
new file mode 100644
index 00000000..cd48d603
--- /dev/null
+++ b/test/cases/measure/data/input/all_latency.yaml
@@ -0,0 +1,26 @@
+# Licensed to Apache Software Foundation (ASF) under one or more contributor
+# license agreements. See the NOTICE file distributed with
+# this work for additional information regarding copyright
+# ownership. Apache Software Foundation (ASF) licenses this file to you 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.
+
+metadata:
+ name: "service_latency_minute"
+ group: "sw_metric"
+tagProjection:
+ tagFamilies:
+ - name: "storage_only"
+ tags: ["entity_id"]
+fieldProjection:
+ names: ["total", "value"]
diff --git a/test/cases/measure/data/testdata/service_latency_minute_data.json b/test/cases/measure/data/testdata/service_latency_minute_data.json
new file mode 100644
index 00000000..2040c6af
--- /dev/null
+++ b/test/cases/measure/data/testdata/service_latency_minute_data.json
@@ -0,0 +1,206 @@
+[
+ {
+ "tag_families": [
+ {
+ "tags": [
+ {
+ "str": {
+ "value": "svc1"
+ }
+ }
+ ]
+ },
+ {
+ "tags": [
+ {
+ "str": {
+ "value": "entity_1"
+ }
+ }
+ ]
+ }
+ ],
+ "fields": [
+ {
+ "int": {
+ "value": 100
+ }
+ },
+ {
+ "int": {
+ "value": 1
+ }
+ }
+ ]
+ },
+ {
+ "tag_families": [
+ {
+ "tags": [
+ {
+ "str": {
+ "value": "svc1"
+ }
+ }
+ ]
+ },
+ {
+ "tags": [
+ {
+ "str": {
+ "value": "entity_2"
+ }
+ }
+ ]
+ }
+ ],
+ "fields": [
+ {
+ "int": {
+ "value": 100
+ }
+ },
+ {
+ "int": {
+ "value": 2
+ }
+ }
+ ]
+ },
+ {
+ "tag_families": [
+ {
+ "tags": [
+ {
+ "str": {
+ "value": "svc1"
+ }
+ }
+ ]
+ },
+ {
+ "tags": [
+ {
+ "str": {
+ "value": "entity_3"
+ }
+ }
+ ]
+ }
+ ],
+ "fields": [
+ {
+ "int": {
+ "value": 100
+ }
+ },
+ {
+ "int": {
+ "value": 3
+ }
+ }
+ ]
+ },
+ {
+ "tag_families": [
+ {
+ "tags": [
+ {
+ "str": {
+ "value": "svc2"
+ }
+ }
+ ]
+ },
+ {
+ "tags": [
+ {
+ "str": {
+ "value": "entity_4"
+ }
+ }
+ ]
+ }
+ ],
+ "fields": [
+ {
+ "int": {
+ "value": 100
+ }
+ },
+ {
+ "int": {
+ "value": 5
+ }
+ }
+ ]
+ },
+ {
+ "tag_families": [
+ {
+ "tags": [
+ {
+ "str": {
+ "value": "svc2"
+ }
+ }
+ ]
+ },
+ {
+ "tags": [
+ {
+ "str": {
+ "value": "entity_5"
+ }
+ }
+ ]
+ }
+ ],
+ "fields": [
+ {
+ "int": {
+ "value": 50
+ }
+ },
+ {
+ "int": {
+ "value": 4
+ }
+ }
+ ]
+ },
+ {
+ "tag_families": [
+ {
+ "tags": [
+ {
+ "str": {
+ "value": "svc3"
+ }
+ }
+ ]
+ },
+ {
+ "tags": [
+ {
+ "str": {
+ "value": "entity_6"
+ }
+ }
+ ]
+ }
+ ],
+ "fields": [
+ {
+ "int": {
+ "value": 300
+ }
+ },
+ {
+ "int": {
+ "value": 6
+ }
+ }
+ ]
+ }
+]
diff --git a/test/cases/measure/data/want/all_latency.yaml b/test/cases/measure/data/want/all_latency.yaml
new file mode 100644
index 00000000..99515607
--- /dev/null
+++ b/test/cases/measure/data/want/all_latency.yaml
@@ -0,0 +1,120 @@
+# Licensed to Apache Software Foundation (ASF) under one or more contributor
+# license agreements. See the NOTICE file distributed with
+# this work for additional information regarding copyright
+# ownership. Apache Software Foundation (ASF) licenses this file to you 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.
+
+dataPoints:
+- fields:
+ - name: total
+ value:
+ int:
+ value: "100"
+ - name: value
+ value:
+ int:
+ value: "1"
+ tagFamilies:
+ - name: storage_only
+ tags:
+ - key: entity_id
+ value:
+ str:
+ value: entity_1
+ timestamp: "2024-03-03T06:46:00Z"
+- fields:
+ - name: total
+ value:
+ int:
+ value: "100"
+ - name: value
+ value:
+ int:
+ value: "2"
+ tagFamilies:
+ - name: storage_only
+ tags:
+ - key: entity_id
+ value:
+ str:
+ value: entity_2
+ timestamp: "2024-03-03T06:47:00Z"
+- fields:
+ - name: total
+ value:
+ int:
+ value: "100"
+ - name: value
+ value:
+ int:
+ value: "3"
+ tagFamilies:
+ - name: storage_only
+ tags:
+ - key: entity_id
+ value:
+ str:
+ value: entity_3
+ timestamp: "2024-03-03T06:48:00Z"
+- fields:
+ - name: total
+ value:
+ int:
+ value: "100"
+ - name: value
+ value:
+ int:
+ value: "5"
+ tagFamilies:
+ - name: storage_only
+ tags:
+ - key: entity_id
+ value:
+ str:
+ value: entity_4
+ timestamp: "2024-03-03T06:49:00Z"
+- fields:
+ - name: total
+ value:
+ int:
+ value: "50"
+ - name: value
+ value:
+ int:
+ value: "4"
+ tagFamilies:
+ - name: storage_only
+ tags:
+ - key: entity_id
+ value:
+ str:
+ value: entity_5
+ timestamp: "2024-03-03T06:50:00Z"
+- fields:
+ - name: total
+ value:
+ int:
+ value: "300"
+ - name: value
+ value:
+ int:
+ value: "6"
+ tagFamilies:
+ - name: storage_only
+ tags:
+ - key: entity_id
+ value:
+ str:
+ value: entity_6
+ timestamp: "2024-03-03T06:51:00Z"
diff --git a/test/cases/measure/measure.go b/test/cases/measure/measure.go
index 3cc814ac..e0398efc 100644
--- a/test/cases/measure/measure.go
+++ b/test/cases/measure/measure.go
@@ -65,4 +65,5 @@ var _ = g.DescribeTable("Scanning Measures", verify,
g.Entry("In and not In expressions", helpers.Args{Input: "in", Duration: 25 * time.Minute, Offset: -20 * time.Minute}),
g.Entry("float64 value", helpers.Args{Input: "float", Duration: 25 * time.Minute, Offset: -20 * time.Minute}),
g.Entry("float64 aggregation:min", helpers.Args{Input: "float_agg_min", Duration: 25 * time.Minute, Offset: -20 * time.Minute}),
+ g.Entry("all_latency", helpers.Args{Input: "all_latency", Duration: 25 * time.Minute, Offset: -20 * time.Minute}),
)