diff --git a/pkg/storage/unified/resource/internalstore.go b/pkg/storage/unified/resource/internalstore.go index 31bd27f2294..306075ef7c8 100644 --- a/pkg/storage/unified/resource/internalstore.go +++ b/pkg/storage/unified/resource/internalstore.go @@ -2,36 +2,40 @@ package resource import ( "context" - "encoding/json" "fmt" + "io" "iter" - "time" + "strings" "github.com/grafana/grafana/pkg/apimachinery/validation" ) const ( - metadataSection = "unified/metadata" + internalSection = "unified/internal" ) type internalStore struct { kv KV } -type MetadataKey struct { - Namespace string - Group string - Resource string +type InternalKey struct { + Namespace string + Group string + Resource string + Subsection string } -func (k MetadataKey) String() string { - return fmt.Sprintf("%s/%s/%s", k.Group, k.Resource, k.Namespace) +func (k InternalKey) String() string { + return fmt.Sprintf("%s/%s/%s/%s/%s", strings.ToLower(k.Subsection), k.Group, k.Resource, k.Namespace) } -func (k MetadataKey) Validate() error { +func (k InternalKey) Validate() error { if k.Namespace == "" { return NewValidationError("namespace", k.Namespace, ErrNamespaceRequired) } + if k.Subsection == "" { + return NewValidationError("Subsection", k.Subsection, "Subsection is required") + } if err := validation.IsValidGroup(k.Group); err != nil { return NewValidationError("group", k.Group, err[0]) } @@ -41,11 +45,17 @@ func (k MetadataKey) Validate() error { return nil } -type Metadata struct { - Namespace string `json:"namespace"` - Group string `json:"group"` - Resource string `json:"resource"` - LastImportTime time.Time `json:"lastImportTime"` +func parseInternalKey(key string) (InternalKey, error) { + parts := strings.Split(key, "/") + if len(parts) != 5 { + return InternalKey{}, fmt.Errorf("invalid internal key: %s", key) + } + return InternalKey{ + Subsection: parts[0], + Group: parts[1], + Resource: parts[2], + Namespace: parts[3], + }, nil } func newInternalStore(kv KV) *internalStore { @@ -54,67 +64,141 @@ func newInternalStore(kv KV) *internalStore { } } -func (d *internalStore) Get(ctx context.Context, key MetadataKey) (Metadata, error) { +type InternalData struct { + Namespace string + Group string + Resource string + Subsection string + Value string +} + +func (d *internalStore) Get(ctx context.Context, key InternalKey) (InternalData, error) { if err := key.Validate(); err != nil { - return Metadata{}, fmt.Errorf("invalid metadata key: %w", err) + return InternalData{}, fmt.Errorf("invalid internal key: %w", err) } - return d.get(ctx, key.String()) -} - -func (d *internalStore) get(ctx context.Context, key string) (Metadata, error) { - reader, err := d.kv.Get(ctx, metadataSection, key) - if err != nil { - return Metadata{}, err - } + reader, err := d.kv.Get(ctx, internalSection, key.String()) defer func() { _ = reader.Close() }() - var metadata Metadata - if err = json.NewDecoder(reader).Decode(&metadata); err != nil { - return Metadata{}, err + if err != nil { + return InternalData{}, err } - return metadata, nil + + value, err := io.ReadAll(reader) + if err != nil { + return InternalData{}, err + } + + return InternalData{ + Namespace: key.Namespace, + Group: key.Group, + Resource: key.Resource, + Subsection: key.Subsection, + Value: string(value), + }, nil } -func (d *internalStore) GetAll(ctx context.Context) iter.Seq2[Metadata, error] { +func (d *internalStore) BatchGet(ctx context.Context, keys []InternalKey) iter.Seq2[InternalData, error] { + return func(yield func(InternalData, error) bool) { + for _, key := range keys { + if err := key.Validate(); err != nil { + yield(InternalData{}, fmt.Errorf("invalid internal key %s: %w", key.String(), err)) + return + } + } + + // Process keys in batches. Uses same batch size as datastore.go + for i := 0; i < len(keys); i += dataBatchSize { + end := i + dataBatchSize + if end > len(keys) { + end = len(keys) + } + batch := keys[i:end] + + stringKeys := make([]string, len(batch)) + for j, key := range batch { + stringKeys[j] = key.String() + } + + for kv, err := range d.kv.BatchGet(ctx, internalSection, stringKeys) { + if err != nil { + yield(InternalData{}, err) + return + } + + key, err := parseInternalKey(kv.Key) + if err != nil { + yield(InternalData{}, err) + return + } + + value, err := io.ReadAll(kv.Value) + if err != nil { + yield(InternalData{}, err) + return + } + + if !yield(InternalData{ + Namespace: key.Namespace, + Group: key.Group, + Resource: key.Resource, + Subsection: key.Subsection, + Value: string(value), + }, nil) { + return + } + } + } + } +} + +func (d *internalStore) GetSubsection(ctx context.Context, Subsection string) iter.Seq2[InternalKey, error] { opts := ListOptions{ Sort: SortOrderAsc, - StartKey: "", + StartKey: Subsection, } - return func(yield func(Metadata, error) bool) { - for metadataKey, err := range d.kv.Keys(ctx, metadataSection, opts) { + return func(yield func(InternalKey, error) bool) { + for key, err := range d.kv.Keys(ctx, internalSection, opts) { if err != nil { - yield(Metadata{}, err) + yield(InternalKey{}, err) return } - metadata, err := d.get(ctx, metadataKey) - if !yield(metadata, err) { + internalKey, err := parseInternalKey(key) + if err != nil { + yield(InternalKey{}, err) + return + } + + if !yield(internalKey, nil) { return } } } } -func (d *internalStore) Save(ctx context.Context, metadata Metadata) error { - metadataKey := MetadataKey{ - Namespace: metadata.Namespace, - Group: metadata.Group, - Resource: metadata.Resource, +func (d *internalStore) Save(ctx context.Context, key InternalKey, value string) error { + if err := key.Validate(); err != nil { + return fmt.Errorf("invalid internal key: %w", err) } - if err := metadataKey.Validate(); err != nil { - return fmt.Errorf("invalid metadataKey key: %w", err) - } - - writer, err := d.kv.Save(ctx, metadataSection, metadataKey.String()) + writer, err := d.kv.Save(ctx, internalSection, key.String()) if err != nil { return err } - encoder := json.NewEncoder(writer) - if err := encoder.Encode(metadata); err != nil { + + _, err = io.WriteString(writer, value) + if err != nil { _ = writer.Close() return err } return writer.Close() } + +func (d *internalStore) Delete(ctx context.Context, key InternalKey) error { + if err := key.Validate(); err != nil { + return fmt.Errorf("invalid internal key: %w", err) + } + + return d.kv.Delete(ctx, internalSection, key.String()) +} diff --git a/pkg/storage/unified/resource/internalstore_test.go b/pkg/storage/unified/resource/internalstore_test.go index 400c1d86bb7..ffb28b13c72 100644 --- a/pkg/storage/unified/resource/internalstore_test.go +++ b/pkg/storage/unified/resource/internalstore_test.go @@ -4,7 +4,6 @@ import ( "context" "errors" "testing" - "time" "github.com/stretchr/testify/assert" "github.com/stretchr/testify/require" @@ -25,82 +24,108 @@ func TestNewInternalStore(t *testing.T) { assert.NotNil(t, store.kv) } -func TestInternalStore_MetadataKey_String(t *testing.T) { +func TestInternalStore_InternalKey_String(t *testing.T) { tests := []struct { name string - metadataKey MetadataKey + internalKey InternalKey expected string }{ { - name: "basic event key", - metadataKey: MetadataKey{ - Namespace: "default", - Group: "apps", - Resource: "resource", + name: "basic internal key", + internalKey: InternalKey{ + Namespace: "default", + Group: "apps", + Resource: "resource", + Subsection: "lastimporttime", }, - expected: "apps/resource/default", + expected: "lastimporttime/apps/resource/default", }, { - name: "empty namespace", - metadataKey: MetadataKey{ - Namespace: "", - Group: "apps", - Resource: "resource", + name: "subsection should be lowercased", + internalKey: InternalKey{ + Namespace: "default", + Group: "apps", + Resource: "resource", + Subsection: "LastImportTime", }, - expected: "apps/resource/", + expected: "lastimporttime/apps/resource/default", }, } for _, tt := range tests { t.Run(tt.name, func(t *testing.T) { - result := tt.metadataKey.String() + result := tt.internalKey.String() assert.Equal(t, tt.expected, result) }) } } -func TestInternalStore_MetadataKey_Validate(t *testing.T) { +func TestInternalStore_InternalKey_Validate(t *testing.T) { tests := []struct { name string - key MetadataKey + key InternalKey error error }{ { name: "valid key", - key: MetadataKey{ - Namespace: "default", - Group: "apps", - Resource: "resource", + key: InternalKey{ + Namespace: "default", + Group: "apps", + Resource: "resource", + Subsection: "lastimporttime", + }, + error: nil, + }, + { + name: "valid key no value", + key: InternalKey{ + Namespace: "default", + Group: "apps", + Resource: "resource", + Subsection: "lastimporttime", }, error: nil, }, { name: "empty namespace", - key: MetadataKey{ - Namespace: "", - Group: "apps", - Resource: "resource", + key: InternalKey{ + Namespace: "", + Group: "apps", + Resource: "resource", + Subsection: "lastimporttime", }, error: errors.New("namespace '' is invalid: namespace is required"), }, { name: "empty group", - key: MetadataKey{ - Namespace: "default", - Group: "", - Resource: "resource", + key: InternalKey{ + Namespace: "default", + Group: "", + Resource: "resource", + Subsection: "lastimporttime", }, error: errors.New("group '' is invalid: group is too short"), }, { name: "empty resource", - key: MetadataKey{ - Namespace: "default", - Group: "apps", - Resource: "", + key: InternalKey{ + Namespace: "default", + Group: "apps", + Resource: "", + Subsection: "lastimporttime", }, error: errors.New("resource '' is invalid: resource is too short"), }, + { + name: "empty subsection", + key: InternalKey{ + Namespace: "default", + Group: "apps", + Resource: "resource", + Subsection: "", + }, + error: errors.New("Subsection '' is invalid: Subsection is required"), + }, } for _, tt := range tests { @@ -120,73 +145,102 @@ func TestInternalStore(t *testing.T) { t.Run("Save and Get", func(t *testing.T) { ctx := context.Background() store := setupTestInternalStore(t) - metadata := Metadata{ - Namespace: "default", - Group: "apps", - Resource: "resource", - LastImportTime: time.Now().Truncate(time.Microsecond), + key := InternalKey{ + Namespace: "default", + Group: "apps", + Resource: "resource", + Subsection: "lastimporttime", } - err := store.Save(ctx, metadata) + err := store.Save(ctx, key, "1") require.NoError(t, err) - metadataKey := MetadataKey{ - Namespace: "default", - Group: "apps", - Resource: "resource", - } - - retrievedMetadata, err := store.Get(ctx, metadataKey) + value, err := store.Get(ctx, key) require.NoError(t, err) - assert.Equal(t, metadata, retrievedMetadata) + assert.Equal(t, "1", value) }) - t.Run("GetAll", func(t *testing.T) { + t.Run("GetSubsection and BatchGet", func(t *testing.T) { ctx := context.Background() store := setupTestInternalStore(t) - metadatas := []Metadata{ + things := []struct { + key InternalKey + value string + }{ { - Namespace: "stacks-1", - Group: "apps", - Resource: "resource", - LastImportTime: time.Now().Add(-2 * time.Minute).Truncate(time.Microsecond), + key: InternalKey{ + Namespace: "stacks-1", + Group: "apps", + Resource: "resource", + Subsection: "lastimporttime", + }, + value: "foo1", }, { - Namespace: "stacks-2", - Group: "apps", - Resource: "resource", - LastImportTime: time.Now().Add(-7 * time.Minute).Truncate(time.Microsecond), + key: InternalKey{ + Namespace: "stacks-2", + Group: "apps", + Resource: "resource", + Subsection: "lastimporttime", + }, + value: "foo2", }, { - Namespace: "stacks-3", - Group: "apps", - Resource: "resource", - LastImportTime: time.Now().Add(-3 * time.Minute).Truncate(time.Microsecond), + key: InternalKey{ + Namespace: "stacks-3", + Group: "apps", + Resource: "resource", + Subsection: "lastimporttime", + }, + value: "foo3", }, { - Namespace: "stacks-4", - Group: "apps", - Resource: "resource", - LastImportTime: time.Now().Add(-24 * time.Minute).Truncate(time.Microsecond), + key: InternalKey{ + Namespace: "stacks-4", + Group: "apps", + Resource: "resource", + Subsection: "lastimporttime", + }, + value: "foo4", }, { - Namespace: "stacks-5", - Group: "apps", - Resource: "resource", - LastImportTime: time.Now().Add(-44 * time.Minute).Truncate(time.Microsecond), + key: InternalKey{ + Namespace: "stacks-5", + Group: "apps", + Resource: "resource", + Subsection: "lastimporttime", + }, + value: "foo5", }, } - for _, metadata := range metadatas { - err := store.Save(ctx, metadata) + for _, thing := range things { + err := store.Save(ctx, thing.key, thing.value) require.NoError(t, err) } var i int - for metadata, err := range store.GetAll(ctx) { + for thing, err := range store.GetSubsection(ctx, "lastimporttime") { require.NoError(t, err) - require.Equal(t, metadatas[i], metadata) + require.Equal(t, things[i].key, thing) + data, err := store.Get(ctx, things[i].key) + require.NoError(t, err) + require.Equal(t, data.Value, things[i].value) i++ } + + keys := make([]InternalKey, len(things)) + for _, thing := range things { + keys = append(keys, thing.key) + } + var j int + for data, err := range store.BatchGet(ctx, keys) { + require.NoError(t, err) + require.Equal(t, things[j].key.Namespace, data.Namespace) + require.Equal(t, things[j].key.Group, data.Group) + require.Equal(t, things[j].key.Resource, data.Resource) + require.Equal(t, things[j].key.Subsection, data.Subsection) + require.Equal(t, things[j].value, data.Value) + } }) } diff --git a/pkg/storage/unified/resource/storage_backend.go b/pkg/storage/unified/resource/storage_backend.go index 9c64a773c94..9b9b3b320c7 100644 --- a/pkg/storage/unified/resource/storage_backend.go +++ b/pkg/storage/unified/resource/storage_backend.go @@ -1235,27 +1235,35 @@ func (k *kvStorageBackend) GetResourceStats(ctx context.Context, nsr NamespacedR return k.dataStore.GetResourceStats(ctx, nsr.Namespace, minCount) } +const lastImportTimeSubsection = "lastimporttime" + func (k *kvStorageBackend) GetResourceLastImportTimes(ctx context.Context) iter.Seq2[ResourceLastImportTime, error] { return func(yield func(ResourceLastImportTime, error) bool) { - for metadata, err := range k.internalStore.GetAll(ctx) { + for key, err := range k.internalStore.GetSubsection(ctx, lastImportTimeSubsection) { if err != nil { yield(ResourceLastImportTime{}, err) return } - if metadata.LastImportTime.IsZero() { - continue + data, err := k.internalStore.Get(ctx, key) + if err != nil { + yield(ResourceLastImportTime{}, err) + return } - // TODO clear LastImportTime from metadata if > lastImportTimeMaxAge? + value, err := time.Parse(time.RFC3339, data.Value) + if err != nil { + yield(ResourceLastImportTime{}, err) + return + } if !yield(ResourceLastImportTime{ NamespacedResource: NamespacedResource{ - Namespace: metadata.Namespace, - Group: metadata.Group, - Resource: metadata.Resource, + Namespace: data.Namespace, + Group: data.Group, + Resource: data.Resource, }, - LastImportTime: metadata.LastImportTime, + LastImportTime: value, }, nil) { return } @@ -1264,27 +1272,14 @@ func (k *kvStorageBackend) GetResourceLastImportTimes(ctx context.Context) iter. } func (k *kvStorageBackend) updateLastImportTime(ctx context.Context, key *resourcepb.ResourceKey, now time.Time) error { - metadata, err := k.internalStore.Get(ctx, MetadataKey{ - Namespace: key.Namespace, - Group: key.Group, - Resource: key.Resource, - }) - - if err != nil && !errors.Is(err, ErrNotFound) { - k.log.Error("Error retrieving metadata for namespace %s: %s", key.Namespace, err) - return err + dataKey := InternalKey{ + Namespace: key.Namespace, + Group: key.Group, + Resource: key.Resource, + Subsection: lastImportTimeSubsection, } - if errors.Is(err, ErrNotFound) { - metadata = Metadata{ - Namespace: key.Namespace, - Group: key.Group, - Resource: key.Resource, - } - } - - metadata.LastImportTime = now.UTC() - return k.internalStore.Save(ctx, metadata) + return k.internalStore.Save(ctx, dataKey, now.UTC().Format(time.RFC3339)) } func (k *kvStorageBackend) ProcessBulk(ctx context.Context, setting BulkSettings, iter BulkRequestIterator) *resourcepb.BulkResponse {