Skip to content

Commit

Permalink
[BUG] Coordinator - Find by id (#1676)
Browse files Browse the repository at this point in the history
## Description of changes

*Summarize the changes made by this PR.*
 - Improvements & Bug fixes
- The coordinator did not respect find by id queries if they were given
without tenant and database, this breaks auth in the FE server when it
tries to resolve a collections tenant and database.
 - New functionality
	 - /
  • Loading branch information
HammadB authored Jan 24, 2024
1 parent 4ca525e commit 425cd38
Showing 1 changed file with 33 additions and 11 deletions.
44 changes: 33 additions & 11 deletions go/coordinator/internal/coordinator/meta.go
Original file line number Diff line number Diff line change
Expand Up @@ -243,17 +243,39 @@ func (mt *MetaTable) GetCollections(ctx context.Context, collectionID types.Uniq
mt.ddLock.RLock()
defer mt.ddLock.RUnlock()

if _, ok := mt.tenantDatabaseCollectionCache[tenantID]; !ok {
log.Error("tenant not found", zap.Any("tenantID", tenantID))
return nil, common.ErrTenantNotFound
}
if _, ok := mt.tenantDatabaseCollectionCache[tenantID][databaseName]; !ok {
return nil, common.ErrDatabaseNotFound
}
collections := make([]*model.Collection, 0, len(mt.tenantDatabaseCollectionCache[tenantID][databaseName]))
for _, collection := range mt.tenantDatabaseCollectionCache[tenantID][databaseName] {
if model.FilterCollection(collection, collectionID, collectionName, collectionTopic) {
collections = append(collections, collection)
// There are three cases
// In the case of getting by id, we do not care about the tenant and database name.
// In the case of getting by name, we need the fully qualified path of the collection which is the tenant/database/name.
// In the case of getting by topic, we need the fully qualified path of the collection which is the tenant/database/topic.
collections := make([]*model.Collection, 0, len(mt.tenantDatabaseCollectionCache))
if collectionID != types.NilUniqueID() {
// Case 1: getting by id
// Due to how the cache is constructed, we iterate over the whole cache to find the collection.
// This is not efficient but it is not a problem for now because the number of collections is small.
// HACK warning. TODO: fix this when we remove the cache.
for _, search_databases := range mt.tenantDatabaseCollectionCache {
for _, search_collections := range search_databases {
for _, collection := range search_collections {
if model.FilterCollection(collection, collectionID, collectionName, collectionTopic) {
collections = append(collections, collection)
}
}
}
}
} else {
// Case 2 & 3: getting by name or topic
// Note: The support for case 3 is not correct here, we shouldn't require the database name and tenant to get by topic.
if _, ok := mt.tenantDatabaseCollectionCache[tenantID]; !ok {
log.Error("tenant not found", zap.Any("tenantID", tenantID))
return nil, common.ErrTenantNotFound
}
if _, ok := mt.tenantDatabaseCollectionCache[tenantID][databaseName]; !ok {
return nil, common.ErrDatabaseNotFound
}
for _, collection := range mt.tenantDatabaseCollectionCache[tenantID][databaseName] {
if model.FilterCollection(collection, collectionID, collectionName, collectionTopic) {
collections = append(collections, collection)
}
}
}
log.Info("meta collections", zap.Any("collections", collections))
Expand Down

0 comments on commit 425cd38

Please sign in to comment.