public/grafana
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

implement tags in kv

Browse Source
This commit is contained in:
Serge Zaitsev
2025-12-03 17:14:50 +01:00
parent 3db55d5135
commit 3205914a05
6 changed files with 108 additions and 364 deletions
Download Patch File Download Diff File Expand all files Collapse all files
pkg/registry/apps/annotation/kv.go
+78 -340
View File
@@ -5,7 +5,9 @@ import (
"context"
"encoding/json"
"fmt"
"slices"
"strconv"
"strings"
"github.com/dgraph-io/badger/v4"
"github.com/google/uuid"
@@ -27,12 +29,11 @@ func NewKVStore(dbdir string) (Store, error) {
func (kv *kvStore) Close() error { return kv.db.Close() }
// TODO: namespace!!!!
func keyUUID(id string) []byte { return []byte("a:uuid:" + id) }
func keyTime(t int64, id string) []byte { return []byte(fmt.Sprintf("a:time:%10d:%s", t/1000, id)) }
func keyDash(d string, id string) []byte { return []byte("a:dash:" + d + ":" + id) }
// func keyTag(tag, id string) []byte { return []byte("a:tag:" + tag + ":" + id) }
// func keyTagGlobal(tag string) []byte { return []byte("a:tags:" + tag) }
func keyTag(tag, id string) []byte { return []byte("a:tag:" + tag + ":" + id) }
func (kv *kvStore) Get(ctx context.Context, namespace, name string) (*annotationV0.Annotation, error) {
var result *annotationV0.Annotation
@@ -135,23 +136,50 @@ func (kv *kvStore) Create(ctx context.Context, a *annotationV0.Annotation) (*ann
return nil, err
}
}
// for _, t := range a.Tags {
// put(keyTag(t, a.UUID), []byte{})
// put(keyTagGlobal(t), []byte{})
// }
// return nil
for _, t := range a.Spec.Tags {
if err := kv.put(keyTag(t, a.Name), []byte{}); err != nil {
return nil, err
}
}
return a, nil
}
func (kv *kvStore) Update(ctx context.Context, a *annotationV0.Annotation) (*annotationV0.Annotation, error) {
func (kv *kvStore) Update(ctx context.Context, a *annotationV0.Annotation) error {
b, err := json.Marshal(a.Spec)
if err != nil {
return nil, err
return err
}
if err := kv.put(keyUUID(a.Name), b); err != nil {
return nil, err
}
return nil, nil
return kv.db.Update(func(txn *badger.Txn) error {
old, err := kv.load(txn, a.Name)
if err != nil {
return err
}
slices.Sort(a.Spec.Tags)
slices.Sort(old.Spec.Tags)
tagsChanged := false
if len(a.Spec.Tags) != len(old.Spec.Tags) {
tagsChanged = true
} else {
for i := range a.Spec.Tags {
if a.Spec.Tags[i] != old.Spec.Tags[i] {
tagsChanged = true
break
}
}
}
if tagsChanged {
for _, t := range old.Spec.Tags {
txn.Delete(keyTag(t, a.Name))
}
for _, t := range a.Spec.Tags {
txn.Set(keyTag(t, a.Name), []byte{})
}
}
return txn.Set(keyUUID(a.Name), b)
})
}
func (kv *kvStore) Delete(ctx context.Context, namespace, name string) error {
@@ -171,338 +199,48 @@ func (kv *kvStore) Delete(ctx context.Context, namespace, name string) error {
return err
}
}
// for _, t := range a.Tags {
// del(keyTag(t, a.UUID))
// }
for _, t := range a.Spec.Tags {
txn.Delete(keyTag(t, a.Name))
}
return nil
})
}
// import (
// "bytes"
// "encoding/json"
// "flag"
// "fmt"
// "math/rand"
// "os"
// "strconv"
// "strings"
// "time"
func (kv *kvStore) Tags(ctx context.Context, namespace string, opts TagListOptions) ([]Tag, error) {
tagCounts := make(map[string]int64)
// "github.com/dgraph-io/badger/v4"
// "github.com/google/uuid"
// )
prefix := []byte("a:tag:")
err := kv.db.View(func(txn *badger.Txn) error {
it := txn.NewIterator(badger.DefaultIteratorOptions)
for it.Seek(prefix); it.ValidForPrefix(prefix); it.Next() {
k := it.Item().Key()
fmt.Println("tag key", string(k))
parts := bytes.Split(k, []byte(":"))
if len(parts) < 4 {
continue
}
tag := string(parts[2])
if opts.Prefix == "" || strings.HasPrefix(tag, opts.Prefix) {
tagCounts[tag]++
}
}
it.Close()
return nil
})
if err != nil {
return nil, err
}
// type Annotation struct {
// UUID string `json:"uuid"`
// DashboardUID string `json:"dashboard_uid"`
// Time int64 `json:"time"`
// TimeEnd int64 `json:"time_end"`
// Text string `json:"text"`
// Tags []string `json:"tags"`
// Metadata string `json:"metadata"`
// UpdatedAt int64 `json:"updated_at"`
// }
tags := make([]Tag, 0, len(tagCounts))
for name, count := range tagCounts {
tags = append(tags, Tag{Name: name, Count: count})
}
// func main() {
// dbdir := flag.String("db", "./kvdb", "")
// flag.Parse()
// opts := badger.DefaultOptions(*dbdir)
// d, err := badger.Open(opts)
// if err != nil {
// panic(err)
// }
// db = d
// defer db.Close()
// TODO: sort tags by count or name?
// if len(os.Args) < 2 {
// fmt.Println("cmd required")
// return
// }
if opts.Limit > 0 && len(tags) > opts.Limit {
tags = tags[:opts.Limit]
}
// switch os.Args[1] {
// case "insert":
// insertCmd(os.Args[2:])
// case "insert-random":
// insertRandomCmd(os.Args[2:])
// case "query":
// queryCmd(os.Args[2:])
// case "list-tags":
// listTagsCmd()
// case "delete-dashboard":
// deleteDashboardCmd(os.Args[2:])
// case "delete-older":
// deleteOlderCmd(os.Args[2:])
// default:
// fmt.Println("unknown cmd")
// }
// }
// func put(k, v []byte) error {
// return db.Update(func(txn *badger.Txn) error { return txn.Set(k, v) })
// }
// func del(k []byte) error {
// return db.Update(func(txn *badger.Txn) error { return txn.Delete(k) })
// }
// func keyUUID(id string) []byte {
// return []byte("a:uuid:" + id)
// }
// func keyTime(t int64, id string) []byte {
// return []byte(fmt.Sprintf("a:time:%020d:%s", t, id))
// }
// func keyDash(d string, id string) []byte {
// return []byte("a:dash:" + d + ":" + id)
// }
// func keyTag(tag, id string) []byte {
// return []byte("a:tag:" + tag + ":" + id)
// }
// func keyTagGlobal(tag string) []byte {
// return []byte("a:tags:" + tag)
// }
// func insert(a Annotation) error {
// }
// func deleteUUID(id string) error {
// var a Annotation
// err := db.View(func(txn *badger.Txn) error {
// item, e := txn.Get(keyUUID(id))
// if e != nil {
// return e
// }
// return item.Value(func(v []byte) error {
// return json.Unmarshal(v, &a)
// })
// })
// if err != nil {
// return err
// }
// del(keyUUID(id))
// del(keyTime(a.Time, a.UUID))
// del(keyDash(a.DashboardUID, a.UUID))
// for _, t := range a.Tags {
// del(keyTag(t, a.UUID))
// }
// return nil
// }
// func insertCmd(args []string) {
// fs := flag.NewFlagSet("insert", 0)
// text := fs.String("text", "", "")
// tags := fs.String("tags", "", "")
// dash := fs.String("dashboard", "", "")
// t1 := fs.Int64("time", 0, "")
// t2 := fs.Int64("timeend", 0, "")
// fs.Parse(args)
// a := Annotation{
// UUID: uuid.NewString(),
// DashboardUID: *dash,
// Time: *t1,
// TimeEnd: *t2,
// Text: *text,
// Tags: strings.Split(*tags, ","),
// UpdatedAt: time.Now().UnixMilli(),
// }
// insert(a)
// fmt.Println(a.UUID)
// }
// func insertRandomCmd(args []string) {
// fs := flag.NewFlagSet("insert-random", 0)
// n := fs.Int("n", 1000, "")
// fs.Parse(args)
// start := time.Now()
// for i := 0; i < *n; i++ {
// id := uuid.NewString()
// t := time.Now().UnixMilli() - rand.Int63n(1000000000)
// a := Annotation{
// UUID: id,
// DashboardUID: fmt.Sprintf("dash-%d", rand.Intn(10)),
// Time: t,
// TimeEnd: t + rand.Int63n(100000),
// Text: "text",
// Tags: []string{fmt.Sprintf("tag-%d", rand.Intn(20))},
// UpdatedAt: time.Now().UnixMilli(),
// }
// insert(a)
// }
// fmt.Println("duration_ms", time.Since(start).Milliseconds())
// }
// func parseAgo(s string) int64 {
// if s == "" {
// return 0
// }
// now := time.Now().UnixMilli()
// last := s[len(s)-1]
// num := s[:len(s)-1]
// switch last {
// case 's', 'm', 'h', 'd', 'w':
// v, err := strconv.ParseInt(num, 10, 64)
// if err == nil {
// switch last {
// case 's':
// return now - v*1000
// case 'm':
// return now - v*60*1000
// case 'h':
// return now - v*60*60*1000
// case 'd':
// return now - v*24*60*60*1000
// case 'w':
// return now - v*7*24*60*60*1000
// }
// }
// }
// v, err := strconv.ParseInt(s, 10, 64)
// if err == nil {
// return v
// }
// return 0
// }
// func queryCmd(args []string) {
// fs := flag.NewFlagSet("query", 0)
// fromS := fs.String("from", "", "")
// toS := fs.String("to", "", "")
// dash := fs.String("dashboard", "", "")
// tag := fs.String("tag", "", "")
// fs.Parse(args)
// from := parseAgo(*fromS)
// to := parseAgo(*toS)
// if to == 0 {
// to = time.Now().UnixMilli()
// }
// start := time.Now()
// result := []Annotation{}
// if *tag != "" {
// prefix := []byte("a:tag:" + *tag + ":")
// db.View(func(txn *badger.Txn) error {
// it := txn.NewIterator(badger.DefaultIteratorOptions)
// for it.Seek(prefix); it.ValidForPrefix(prefix); it.Next() {
// parts := bytes.Split(it.Item().Key(), []byte(":"))
// id := string(parts[len(parts)-1])
// a, err := loadUUID(txn, id)
// if err == nil && a.TimeEnd >= from && a.Time <= to {
// if *dash == "" || a.DashboardUID == *dash {
// result = append(result, a)
// }
// }
// }
// it.Close()
// return nil
// })
// } else {
// prefix := []byte("a:time:")
// fromKey := []byte(fmt.Sprintf("a:time:%020d:", from))
// db.View(func(txn *badger.Txn) error {
// it := txn.NewIterator(badger.DefaultIteratorOptions)
// for it.Seek(fromKey); it.ValidForPrefix(prefix); it.Next() {
// k := it.Item().Key()
// parts := bytes.Split(k, []byte(":"))
// t, _ := strconv.ParseInt(string(parts[2]), 10, 64)
// if t > to {
// break
// }
// id := string(parts[3])
// a, err := loadUUID(txn, id)
// if err == nil && a.TimeEnd >= from {
// if *dash == "" || a.DashboardUID == *dash {
// result = append(result, a)
// }
// }
// }
// it.Close()
// return nil
// })
// }
// for _, a := range result {
// b, _ := json.Marshal(a)
// fmt.Println(string(b))
// }
// fmt.Println("duration_ms", time.Since(start).Milliseconds())
// }
// func loadUUID(txn *badger.Txn, id string) (Annotation, error) {
// var a Annotation
// item, e := txn.Get(keyUUID(id))
// if e != nil {
// return a, e
// }
// item.Value(func(v []byte) error {
// return json.Unmarshal(v, &a)
// })
// return a, nil
// }
// func listTagsCmd() {
// prefix := []byte("a:tags:")
// db.View(func(txn *badger.Txn) error {
// it := txn.NewIterator(badger.DefaultIteratorOptions)
// for it.Seek(prefix); it.ValidForPrefix(prefix); it.Next() {
// k := it.Item().Key()
// fmt.Println(string(k[len(prefix):]))
// }
// it.Close()
// return nil
// })
// }
// func deleteDashboardCmd(args []string) {
// fs := flag.NewFlagSet("delete-dashboard", 0)
// dash := fs.String("dashboard", "", "")
// fs.Parse(args)
// prefix := []byte("a:dash:" + *dash + ":")
// db.View(func(txn *badger.Txn) error {
// it := txn.NewIterator(badger.DefaultIteratorOptions)
// for it.Seek(prefix); it.ValidForPrefix(prefix); it.Next() {
// parts := bytes.Split(it.Item().Key(), []byte(":"))
// id := string(parts[len(parts)-1])
// deleteUUID(id)
// }
// it.Close()
// return nil
// })
// }
// func deleteOlderCmd(args []string) {
// fs := flag.NewFlagSet("delete-older", 0)
// limit := fs.Int64("ts", 0, "")
// fs.Parse(args)
// prefix := []byte("a:time:")
// db.View(func(txn *badger.Txn) error {
// it := txn.NewIterator(badger.DefaultIteratorOptions)
// for it.Seek(prefix); it.ValidForPrefix(prefix); it.Next() {
// k := it.Item().Key()
// parts := bytes.Split(k, []byte(":"))
// t, _ := strconv.ParseInt(string(parts[2]), 10, 64)
// if t < *limit {
// id := string(parts[3])
// deleteUUID(id)
// }
// }
// it.Close()
// return nil
// })
// }
return tags, nil
}
pkg/registry/apps/annotation/memory_store.go
+7 -3
View File
@@ -98,20 +98,20 @@ func (m *memoryStore) Create(ctx context.Context, anno *annotationV0.Annotation)
return created, nil
}
func (m *memoryStore) Update(ctx context.Context, anno *annotationV0.Annotation) (*annotationV0.Annotation, error) {
func (m *memoryStore) Update(ctx context.Context, anno *annotationV0.Annotation) error {
m.mu.Lock()
defer m.mu.Unlock()
key := anno.Namespace + "/" + anno.Name
if _, exists := m.data[key]; !exists {
return nil, fmt.Errorf("annotation not found")
return fmt.Errorf("annotation not found")
}
updated := anno.DeepCopy()
m.data[key] = updated
return updated, nil
return nil
}
func (m *memoryStore) Delete(ctx context.Context, namespace, name string) error {
@@ -156,3 +156,7 @@ func (m *memoryStore) ListTags(ctx context.Context, namespace string, opts TagLi
return tags, nil
}
func (m *memoryStore) Tags(ctx context.Context, namespace string, opts TagListOptions) ([]Tag, error) {
return nil, fmt.Errorf("not implemented")
}
pkg/registry/apps/annotation/register.go
+9 -4
View File
@@ -60,14 +60,14 @@ func RegisterAppInstaller(
if err != nil {
return nil, err
}
sqlAdapter := NewSQLAdapter(service, cleaner, mapper, cfg)
// sqlAdapter := NewSQLAdapter(service, cleaner, mapper, cfg)
installer.legacy = &restStorage{
// store: sqlAdapter,
store: kvAdapter,
mapper: mapper,
}
// Create the tags handler using the sqlAdapter as TagProvider
tagHandler = newTagsHandler(sqlAdapter)
tagHandler = newTagsHandler(installer.legacy.store)
}
provider := simple.NewAppProvider(apis.LocalManifest(), nil, annotationapp.New)
@@ -159,6 +159,11 @@ func (s *restStorage) ConvertToTable(ctx context.Context, object runtime.Object,
return s.tableConverter.ConvertToTable(ctx, object, tableOptions)
}
// TODO: hierarchy of list options (where do tags fit in - seems to be mutually exclusive with dashboard query?)
// 1. No options: list all for namespace
// 2. By dashboard: list all annotations for namespace+dashboard (i.e. remove by dashboard)
// 3. By panel: list all annotations for namespace+dashboard+panel (i.e. remove by dashboard+panel)
// 4. By time range: list all annotations for namespace in time range
func (s *restStorage) List(ctx context.Context, options *internalversion.ListOptions) (runtime.Object, error) {
namespace := request.NamespaceValue(ctx)
@@ -281,8 +286,8 @@ func (s *restStorage) Update(ctx context.Context,
}
}
// TODO: validate that only name/tags are modified
_, err = s.store.Update(ctx, newObj)
return nil, false, err
err = s.store.Update(ctx, newObj)
return obj, false, err
}
func (s *restStorage) Delete(ctx context.Context, name string, deleteValidation rest.ValidateObjectFunc, options *metav1.DeleteOptions) (runtime.Object, bool, error) {
pkg/registry/apps/annotation/sql_adapter.go
+4 -4
View File
@@ -125,20 +125,20 @@ func (a *sqlAdapter) Create(ctx context.Context, anno *annotationV0.Annotation)
return created, nil
}
func (a *sqlAdapter) Update(ctx context.Context, anno *annotationV0.Annotation) (*annotationV0.Annotation, error) {
func (a *sqlAdapter) Update(ctx context.Context, anno *annotationV0.Annotation) error {
orgID, err := namespaceToOrgID(ctx, anno.Namespace)
if err != nil {
return nil, err
return err
}
item := a.fromK8sResource(anno)
item.OrgID = orgID
if err := a.repo.Update(ctx, item); err != nil {
return nil, err
return err
}
return anno, nil
return nil
}
func (a *sqlAdapter) Delete(ctx context.Context, namespace, name string) error {
pkg/registry/apps/annotation/storage.go
+8 -11
View File
@@ -9,9 +9,12 @@ import (
type Store interface {
Get(ctx context.Context, namespace, name string) (*annotationV0.Annotation, error)
List(ctx context.Context, namespace string, opts ListOptions) (*AnnotationList, error)
// TODO: return id only?
Create(ctx context.Context, annotation *annotationV0.Annotation) (*annotationV0.Annotation, error)
Update(ctx context.Context, annotation *annotationV0.Annotation) (*annotationV0.Annotation, error)
Update(ctx context.Context, annotation *annotationV0.Annotation) error
Delete(ctx context.Context, namespace, name string) error
Tags(ctx context.Context, namespace string, opts TagListOptions) ([]Tag, error)
}
type ListOptions struct {
@@ -22,26 +25,20 @@ type ListOptions struct {
Limit int64
Continue string
}
type AnnotationList struct {
Items []annotationV0.Annotation
Continue string
}
type LifecycleManager interface {
Cleanup(ctx context.Context) (int64, error)
}
type TagProvider interface {
ListTags(ctx context.Context, namespace string, opts TagListOptions) ([]Tag, error)
}
type TagListOptions struct {
Prefix string
Limit int
}
type Tag struct {
Name string
Count int64
}
type LifecycleManager interface {
Cleanup(ctx context.Context) (int64, error)
}
pkg/registry/apps/annotation/tags_handler.go
+2 -2
View File
@@ -17,14 +17,14 @@ type tagItem struct {
Count int64 `json:"count"`
}
func newTagsHandler(tagProvider TagProvider) func(ctx context.Context, writer app.CustomRouteResponseWriter, request *app.CustomRouteRequest) error {
func newTagsHandler(tagProvider Store) func(ctx context.Context, writer app.CustomRouteResponseWriter, request *app.CustomRouteRequest) error {
return func(ctx context.Context, writer app.CustomRouteResponseWriter, request *app.CustomRouteRequest) error {
fmt.Println("Handling /tags request")
namespace := request.ResourceIdentifier.Namespace
if namespace == "" {
namespace = "default"
}
tags, err := tagProvider.ListTags(ctx, namespace, TagListOptions{})
tags, err := tagProvider.Tags(ctx, namespace, TagListOptions{})
if err != nil {
return err
}