implement last import time

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())
+}

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)
+		}
 	})
 }

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 {