UnifiedStorage: Include RV when fieldSelectors are processed in the backend (#115110)

pkg/storage/unified/resource/fieldSelector.go

@@ -0,0 +1,53 @@
+package resource
+
+import (
+	"context"
+
+	"github.com/grafana/grafana/pkg/storage/unified/resourcepb"
+)
+
+// Some list queries can be calculated with simple reads or search index
+func (s *server) tryFieldSelector(ctx context.Context, req *resourcepb.ListRequest) *resourcepb.ListResponse {
+	if req.Source != resourcepb.ListRequest_STORE || req.Options.Key.Namespace == "" {
+		return nil
+	}
+
+	var names []string
+	for _, v := range req.Options.Fields {
+		if v.Key == "metadata.name" && v.Operator == `=` {
+			names = v.Values
+			continue
+		}
+
+		// TODO: support other field selectors
+	}
+
+	// The required names
+	if len(names) > 0 {
+		read := &resourcepb.ReadRequest{
+			Key:             req.Options.Key,
+			ResourceVersion: req.ResourceVersion,
+		}
+		rsp := &resourcepb.ListResponse{
+			ResourceVersion: 1, // TODO, search result should include when it was indexed
+		}
+		for _, name := range names {
+			read.Key.Name = name
+			found, err := s.Read(ctx, read)
+			if err != nil {
+				return &resourcepb.ListResponse{Error: AsErrorResult(err)}
+			}
+			if len(found.Value) > 0 {
+				rsp.Items = append(rsp.Items, &resourcepb.ResourceWrapper{
+					Value:           found.Value,
+					ResourceVersion: found.ResourceVersion,
+				})
+				if found.ResourceVersion > rsp.ResourceVersion {
+					rsp.ResourceVersion = found.ResourceVersion
+				}
+			}
+		}
+		return rsp
+	}
+	return nil
+}

pkg/storage/unified/resource/server.go

@@ -1039,7 +1039,7 @@ func (s *server) List(ctx context.Context, req *resourcepb.ListRequest) (*resour
 	}
 
 	// Fast path for getting single value in a list
-	if rsp := s.tryFastPathList(ctx, req); rsp != nil {
+	if rsp := s.tryFieldSelector(ctx, req); rsp != nil {
 		return rsp, nil
 	}
 
@@ -1137,40 +1137,6 @@ func (s *server) List(ctx context.Context, req *resourcepb.ListRequest) (*resour
 	return rsp, err
 }
 
-// Some list queries can be calculated with simple reads
-func (s *server) tryFastPathList(ctx context.Context, req *resourcepb.ListRequest) *resourcepb.ListResponse {
-	if req.Source != resourcepb.ListRequest_STORE || req.Options.Key.Namespace == "" {
-		return nil
-	}
-
-	for _, v := range req.Options.Fields {
-		if v.Key == "metadata.name" && v.Operator == `=` {
-			if len(v.Values) == 1 {
-				read := &resourcepb.ReadRequest{
-					Key:             req.Options.Key,
-					ResourceVersion: req.ResourceVersion,
-				}
-				read.Key.Name = v.Values[0]
-				found, err := s.Read(ctx, read)
-				if err != nil {
-					return &resourcepb.ListResponse{Error: AsErrorResult(err)}
-				}
-
-				// Return a value when it exists
-				rsp := &resourcepb.ListResponse{}
-				if len(found.Value) > 0 {
-					rsp.Items = []*resourcepb.ResourceWrapper{{
-						Value:           found.Value,
-						ResourceVersion: found.ResourceVersion,
-					}}
-				}
-				return rsp
-			}
-		}
-	}
-	return nil
-}
-
 // isTrashItemAuthorized checks if the user has access to the trash item.
 func (s *server) isTrashItemAuthorized(ctx context.Context, iter ListIterator, trashChecker claims.ItemChecker) bool {
 	user, ok := claims.AuthInfoFrom(ctx)

pkg/tests/apis/folder/folders_test.go

@@ -1879,6 +1879,14 @@ func TestIntegrationMoveNestedFolderToRootK8S(t *testing.T) {
 	require.Equal(t, http.StatusOK, get.Response.StatusCode)
 	require.Equal(t, "f2", get.Result.UID)
 	require.Equal(t, "", get.Result.ParentUID)
+
+	// Check that we can get the same folder using metadata.name selector
+	results, err := client.Resource.List(context.Background(), metav1.ListOptions{
+		FieldSelector: "metadata.name=" + get.Result.UID,
+	})
+	require.NoError(t, err)
+	require.Len(t, results.Items, 1)
+	require.Equal(t, "f2", results.Items[0].GetName())
 }
 
 // Test deleting nested folders ensures postorder deletion