Storage: Content root storage (#54929)
Co-authored-by: Ryan McKinley <ryantxu@gmail.com>
This commit is contained in:
Artur Wierzbicki
co-authored by
Ryan McKinley
Ryan McKinley
parent
89d94eeab2
commit
e19f36649f
@@ -25,7 +25,13 @@ var (
|
||||
func asNameToFileStorageMap(storages []storageRuntime) map[string]storageRuntime {
|
||||
lookup := make(map[string]storageRuntime)
|
||||
for _, storage := range storages {
|
||||
lookup[storage.Meta().Config.Prefix] = storage
|
||||
isUnderContentRoot := storage.Meta().Config.UnderContentRoot
|
||||
prefix := storage.Meta().Config.Prefix
|
||||
if !isUnderContentRoot {
|
||||
lookup[prefix] = storage
|
||||
} else {
|
||||
lookup[fmt.Sprintf("%s/%s", RootContent, prefix)] = storage
|
||||
}
|
||||
}
|
||||
return lookup
|
||||
}
|
||||
@@ -61,12 +67,44 @@ func (t *nestedTree) getRoot(orgId int64, path string) (storageRuntime, string)
|
||||
rootKey, path := splitFirstSegment(path)
|
||||
root, ok := t.lookup[orgId][rootKey]
|
||||
if ok && root != nil {
|
||||
if root.Meta().Config.Prefix == RootContent && path != "" && path != "/" {
|
||||
mountedKey, nestedPath := splitFirstSegment(path)
|
||||
nestedLookupKey := rootKey + filestorage.Delimiter + mountedKey
|
||||
nestedRoot, nestedOk := t.lookup[orgId][nestedLookupKey]
|
||||
|
||||
if nestedOk && nestedRoot != nil {
|
||||
return nestedRoot, filestorage.Delimiter + nestedPath
|
||||
}
|
||||
|
||||
if orgId != ac.GlobalOrgID {
|
||||
globalRoot, globalOk := t.lookup[ac.GlobalOrgID][nestedLookupKey]
|
||||
if globalOk && globalRoot != nil {
|
||||
return globalRoot, filestorage.Delimiter + nestedPath
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
return root, filestorage.Delimiter + path
|
||||
}
|
||||
|
||||
if orgId != ac.GlobalOrgID {
|
||||
globalRoot, ok := t.lookup[ac.GlobalOrgID][rootKey]
|
||||
if ok && globalRoot != nil {
|
||||
if globalRoot.Meta().Config.Prefix == RootContent && path != "" && path != "/" {
|
||||
mountedKey, nestedPath := splitFirstSegment(path)
|
||||
nestedLookupKey := rootKey + filestorage.Delimiter + mountedKey
|
||||
nestedRoot, nestedOk := t.lookup[orgId][nestedLookupKey]
|
||||
|
||||
if nestedOk && nestedRoot != nil {
|
||||
return nestedRoot, filestorage.Delimiter + nestedPath
|
||||
}
|
||||
|
||||
globalNestedRoot, globalOk := t.lookup[ac.GlobalOrgID][nestedLookupKey]
|
||||
if globalOk && globalNestedRoot != nil {
|
||||
return globalNestedRoot, filestorage.Delimiter + nestedPath
|
||||
}
|
||||
}
|
||||
|
||||
return globalRoot, filestorage.Delimiter + path
|
||||
}
|
||||
}
|
||||
@@ -90,12 +128,22 @@ func (t *nestedTree) GetFile(ctx context.Context, orgId int64, path string) (*fi
|
||||
return file, err
|
||||
}
|
||||
|
||||
func filterStoragesUnderContentRoot(storages []storageRuntime) []storageRuntime {
|
||||
out := make([]storageRuntime, 0)
|
||||
for _, s := range storages {
|
||||
if s.Meta().Config.UnderContentRoot {
|
||||
out = append(out, s)
|
||||
}
|
||||
}
|
||||
return out
|
||||
}
|
||||
|
||||
func (t *nestedTree) getStorages(orgId int64) []storageRuntime {
|
||||
globalStorages := make([]storageRuntime, 0)
|
||||
globalStorages = append(globalStorages, t.rootsByOrgId[ac.GlobalOrgID]...)
|
||||
|
||||
if orgId == ac.GlobalOrgID {
|
||||
return globalStorages
|
||||
return append(make([]storageRuntime, 0), globalStorages...)
|
||||
}
|
||||
|
||||
orgPrefixes := make(map[string]bool)
|
||||
@@ -124,6 +172,7 @@ func (t *nestedTree) ListFolder(ctx context.Context, orgId int64, path string, a
|
||||
|
||||
storages := t.getStorages(orgId)
|
||||
count := len(storages)
|
||||
grafanaStorageLogger.Info("Listing root folder", "path", path, "storageCount", len(storages))
|
||||
|
||||
names := data.NewFieldFromFieldType(data.FieldTypeString, count)
|
||||
title := data.NewFieldFromFieldType(data.FieldTypeString, count)
|
||||
@@ -154,23 +203,38 @@ func (t *nestedTree) ListFolder(ctx context.Context, orgId int64, path string, a
|
||||
return nil, nil // not found (or not ready)
|
||||
}
|
||||
|
||||
var storages []storageRuntime
|
||||
if root.Meta().Config.Prefix == RootContent && (path == "" || path == "/") {
|
||||
storages = filterStoragesUnderContentRoot(t.getStorages(orgId))
|
||||
}
|
||||
grafanaStorageLogger.Info("Listing folder", "path", path, "storageCount", len(storages), "root", root.Meta().Config.Prefix)
|
||||
|
||||
store := root.Store()
|
||||
if store == nil {
|
||||
return nil, fmt.Errorf("store not ready")
|
||||
}
|
||||
|
||||
pathFilter := accessFilter
|
||||
if root.Meta().Config.Prefix == RootContent && len(storages) > 0 {
|
||||
// create a PathFilter that will filter out folders that are "shadowed" by the mounted storages
|
||||
pathFilter = filestorage.NewAndPathFilter(
|
||||
accessFilter,
|
||||
t.createPathFilterForContentRoot(storages),
|
||||
)
|
||||
}
|
||||
|
||||
listResponse, err := store.List(ctx, path, nil, &filestorage.ListOptions{
|
||||
Recursive: false,
|
||||
WithFolders: true,
|
||||
WithFiles: true,
|
||||
Filter: accessFilter,
|
||||
Filter: pathFilter,
|
||||
})
|
||||
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
|
||||
count := len(listResponse.Files)
|
||||
count := len(listResponse.Files) + len(storages)
|
||||
names := data.NewFieldFromFieldType(data.FieldTypeString, count)
|
||||
mtype := data.NewFieldFromFieldType(data.FieldTypeString, count)
|
||||
fsize := data.NewFieldFromFieldType(data.FieldTypeInt64, count)
|
||||
@@ -180,11 +244,22 @@ func (t *nestedTree) ListFolder(ctx context.Context, orgId int64, path string, a
|
||||
fsize.Config = &data.FieldConfig{
|
||||
Unit: "bytes",
|
||||
}
|
||||
for i, f := range listResponse.Files {
|
||||
names.Set(i, f.Name)
|
||||
mtype.Set(i, f.MimeType)
|
||||
fsize.Set(i, f.Size)
|
||||
|
||||
idx := 0
|
||||
for _, s := range storages {
|
||||
names.Set(idx, s.Meta().Config.Prefix)
|
||||
mtype.Set(idx, filestorage.DirectoryMimeType)
|
||||
fsize.Set(idx, int64(0))
|
||||
idx++
|
||||
}
|
||||
|
||||
for _, f := range listResponse.Files {
|
||||
names.Set(idx, f.Name)
|
||||
mtype.Set(idx, f.MimeType)
|
||||
fsize.Set(idx, f.Size)
|
||||
idx++
|
||||
}
|
||||
|
||||
frame := data.NewFrame("", names, mtype, fsize)
|
||||
frame.SetMeta(&data.FrameMeta{
|
||||
Type: data.FrameTypeDirectoryListing,
|
||||
@@ -194,3 +269,17 @@ func (t *nestedTree) ListFolder(ctx context.Context, orgId int64, path string, a
|
||||
})
|
||||
return &StorageListFrame{frame}, nil
|
||||
}
|
||||
|
||||
func (t *nestedTree) createPathFilterForContentRoot(storages []storageRuntime) filestorage.PathFilter {
|
||||
disallowedPrefixes := make([]string, 0)
|
||||
disallowedPaths := make([]string, 0)
|
||||
|
||||
for _, s := range storages {
|
||||
path := filestorage.Delimiter + s.Meta().Config.Prefix
|
||||
disallowedPaths = append(disallowedPaths, path)
|
||||
disallowedPrefixes = append(disallowedPrefixes, path+filestorage.Delimiter)
|
||||
}
|
||||
|
||||
grafanaStorageLogger.Info("Created a path filter for the content root", "disallowedPrefixes", disallowedPrefixes, "disallowedPaths", disallowedPaths)
|
||||
return filestorage.NewPathFilter(nil, nil, disallowedPrefixes, disallowedPaths)
|
||||
}
|
||||
|
||||
Reference in New Issue
Block a user