JakeHillion/storj
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

satellite/satellitedb: simplify select nodes query construction

Browse Source 
Change-Id: I07009b28762d4485929a2a999e8f4be8179bee51
This commit is contained in:
Egon Elbre 2021-10-20 13:54:52 +03:00
parent 3989107031
commit f2d8e97d97
2 changed files with 68 additions and 70 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

72
satellite/overlay/benchmark_test.go
View File

@ -206,6 +206,10 @@ func BenchmarkNodeSelection(b *testing.B) {
OnlineWindow: time.Hour, OnlineWindow: time.Hour,
DistinctIP: true, DistinctIP: true,
MinimumDiskSpace: 0, MinimumDiskSpace: 0,
AsOfSystemTime: overlay.AsOfSystemTimeConfig{
Enabled: true,
DefaultInterval: -time.Microsecond,
},
} }
var excludedIDs []storj.NodeID var excludedIDs []storj.NodeID
@ -279,20 +283,22 @@ func BenchmarkNodeSelection(b *testing.B) {
} }
criteria := &overlay.NodeCriteria{ criteria := &overlay.NodeCriteria{
FreeDisk: 0, FreeDisk: 0,
ExcludedIDs: nil, ExcludedIDs: nil,
ExcludedNetworks: nil, ExcludedNetworks: nil,
MinimumVersion: "v1.0.0", MinimumVersion: "v1.0.0",
OnlineWindow: time.Hour, OnlineWindow: time.Hour,
DistinctIP: false, DistinctIP: false,
AsOfSystemInterval: -time.Microsecond,
} }
excludedCriteria := &overlay.NodeCriteria{ excludedCriteria := &overlay.NodeCriteria{
FreeDisk: 0, FreeDisk: 0,
ExcludedIDs: excludedIDs, ExcludedIDs: excludedIDs,
ExcludedNetworks: excludedNets, ExcludedNetworks: excludedNets,
MinimumVersion: "v1.0.0", MinimumVersion: "v1.0.0",
OnlineWindow: time.Hour, OnlineWindow: time.Hour,
DistinctIP: false, DistinctIP: false,
AsOfSystemInterval: -time.Microsecond,
} }
b.Run("SelectStorageNodes", func(b *testing.B) { b.Run("SelectStorageNodes", func(b *testing.B) {
@ -362,9 +368,10 @@ func BenchmarkNodeSelection(b *testing.B) {
b.Run("FindStorageNodesWithPreference", func(b *testing.B) { b.Run("FindStorageNodesWithPreference", func(b *testing.B) {
for i := 0; i < b.N; i++ { for i := 0; i < b.N; i++ {
selected, err := service.FindStorageNodesWithPreferences(ctx, overlay.FindStorageNodesRequest{ selected, err := service.FindStorageNodesWithPreferences(ctx, overlay.FindStorageNodesRequest{
RequestedCount: SelectCount, RequestedCount: SelectCount,
ExcludedIDs: nil, ExcludedIDs: nil,
MinimumVersion: "v1.0.0", MinimumVersion: "v1.0.0",
AsOfSystemInterval: -time.Microsecond,
}, &nodeSelectionConfig) }, &nodeSelectionConfig)
require.NoError(b, err) require.NoError(b, err)
require.NotEmpty(b, selected) require.NotEmpty(b, selected)
@ -374,9 +381,10 @@ func BenchmarkNodeSelection(b *testing.B) {
b.Run("FindStorageNodesWithPreferenceExclusion", func(b *testing.B) { b.Run("FindStorageNodesWithPreferenceExclusion", func(b *testing.B) {
for i := 0; i < b.N; i++ { for i := 0; i < b.N; i++ {
selected, err := service.FindStorageNodesWithPreferences(ctx, overlay.FindStorageNodesRequest{ selected, err := service.FindStorageNodesWithPreferences(ctx, overlay.FindStorageNodesRequest{
RequestedCount: SelectCount, RequestedCount: SelectCount,
ExcludedIDs: excludedIDs, ExcludedIDs: excludedIDs,
MinimumVersion: "v1.0.0", MinimumVersion: "v1.0.0",
AsOfSystemInterval: -time.Microsecond,
}, &nodeSelectionConfig) }, &nodeSelectionConfig)
require.NoError(b, err) require.NoError(b, err)
require.NotEmpty(b, selected) require.NotEmpty(b, selected)
@ -386,9 +394,10 @@ func BenchmarkNodeSelection(b *testing.B) {
b.Run("FindStorageNodes", func(b *testing.B) { b.Run("FindStorageNodes", func(b *testing.B) {
for i := 0; i < b.N; i++ { for i := 0; i < b.N; i++ {
selected, err := service.FindStorageNodesForUpload(ctx, overlay.FindStorageNodesRequest{ selected, err := service.FindStorageNodesForUpload(ctx, overlay.FindStorageNodesRequest{
RequestedCount: SelectCount, RequestedCount: SelectCount,
ExcludedIDs: nil, ExcludedIDs: nil,
MinimumVersion: "v1.0.0", MinimumVersion: "v1.0.0",
AsOfSystemInterval: -time.Microsecond,
}) })
require.NoError(b, err) require.NoError(b, err)
require.NotEmpty(b, selected) require.NotEmpty(b, selected)
@ -398,9 +407,10 @@ func BenchmarkNodeSelection(b *testing.B) {
b.Run("FindStorageNodesExclusion", func(b *testing.B) { b.Run("FindStorageNodesExclusion", func(b *testing.B) {
for i := 0; i < b.N; i++ { for i := 0; i < b.N; i++ {
selected, err := service.FindStorageNodesForUpload(ctx, overlay.FindStorageNodesRequest{ selected, err := service.FindStorageNodesForUpload(ctx, overlay.FindStorageNodesRequest{
RequestedCount: SelectCount, RequestedCount: SelectCount,
ExcludedIDs: excludedIDs, ExcludedIDs: excludedIDs,
MinimumVersion: "v1.0.0", MinimumVersion: "v1.0.0",
AsOfSystemInterval: -time.Microsecond,
}) })
require.NoError(b, err) require.NoError(b, err)
require.NotEmpty(b, selected) require.NotEmpty(b, selected)
@ -410,9 +420,10 @@ func BenchmarkNodeSelection(b *testing.B) {
b.Run("UploadSelectionCacheGetNodes", func(b *testing.B) { b.Run("UploadSelectionCacheGetNodes", func(b *testing.B) {
for i := 0; i < b.N; i++ { for i := 0; i < b.N; i++ {
selected, err := service.UploadSelectionCache.GetNodes(ctx, overlay.FindStorageNodesRequest{ selected, err := service.UploadSelectionCache.GetNodes(ctx, overlay.FindStorageNodesRequest{
RequestedCount: SelectCount, RequestedCount: SelectCount,
ExcludedIDs: nil, ExcludedIDs: nil,
MinimumVersion: "v1.0.0", MinimumVersion: "v1.0.0",
AsOfSystemInterval: -time.Microsecond,
}) })
require.NoError(b, err) require.NoError(b, err)
require.NotEmpty(b, selected) require.NotEmpty(b, selected)
@ -422,9 +433,10 @@ func BenchmarkNodeSelection(b *testing.B) {
b.Run("UploadSelectionCacheGetNodesExclusion", func(b *testing.B) { b.Run("UploadSelectionCacheGetNodesExclusion", func(b *testing.B) {
for i := 0; i < b.N; i++ { for i := 0; i < b.N; i++ {
selected, err := service.UploadSelectionCache.GetNodes(ctx, overlay.FindStorageNodesRequest{ selected, err := service.UploadSelectionCache.GetNodes(ctx, overlay.FindStorageNodesRequest{
RequestedCount: SelectCount, RequestedCount: SelectCount,
ExcludedIDs: excludedIDs, ExcludedIDs: excludedIDs,
MinimumVersion: "v1.0.0", MinimumVersion: "v1.0.0",
AsOfSystemInterval: -time.Microsecond,
}) })
require.NoError(b, err) require.NoError(b, err)
require.NotEmpty(b, selected) require.NotEmpty(b, selected)

66
satellite/satellitedb/nodeselection.go
View File

@ -104,43 +104,38 @@ func (cache *overlaycache) selectStorageNodesOnce(ctx context.Context, reputable
var reputableNodeQuery, newNodeQuery partialQuery var reputableNodeQuery, newNodeQuery partialQuery
asOf := cache.db.impl.AsOfSystemInterval(criteria.AsOfSystemInterval)
// Note: the true/false at the end of each selection string indicates if the selection is for new nodes or not. // Note: the true/false at the end of each selection string indicates if the selection is for new nodes or not.
// Later, the flag allows us to distinguish if a node is new when scanning the db rows. // Later, the flag allows us to distinguish if a node is new when scanning the db rows.
if !criteria.DistinctIP { if !criteria.DistinctIP {
reputableNodeQuery = partialQuery{ reputableNodeQuery = partialQuery{
selection: `SELECT last_net, id, address, last_ip_port, false FROM nodes ` + asOf, selection: `SELECT last_net, id, address, last_ip_port, false FROM nodes`,
condition: reputableNodesCondition, condition: reputableNodesCondition,
limit: reputableNodeCount, limit: reputableNodeCount,
aostClause: asOf,
} }
newNodeQuery = partialQuery{ newNodeQuery = partialQuery{
selection: `SELECT last_net, id, address, last_ip_port, true FROM nodes ` + asOf, selection: `SELECT last_net, id, address, last_ip_port, true FROM nodes`,
condition: newNodesCondition, condition: newNodesCondition,
limit: newNodeCount, limit: newNodeCount,
aostClause: asOf,
} }
} else { } else {
reputableNodeQuery = partialQuery{ reputableNodeQuery = partialQuery{
selection: `SELECT DISTINCT ON (last_net) last_net, id, address, last_ip_port, false FROM nodes ` + asOf, selection: `SELECT DISTINCT ON (last_net) last_net, id, address, last_ip_port, false FROM nodes`,
condition: reputableNodesCondition, condition: reputableNodesCondition,
distinct: true, distinct: true,
limit: reputableNodeCount, limit: reputableNodeCount,
orderBy: "last_net", orderBy: "last_net",
aostClause: asOf,
} }
newNodeQuery = partialQuery{ newNodeQuery = partialQuery{
selection: `SELECT DISTINCT ON (last_net) last_net, id, address, last_ip_port, true FROM nodes ` + asOf, selection: `SELECT DISTINCT ON (last_net) last_net, id, address, last_ip_port, true FROM nodes`,
condition: newNodesCondition, condition: newNodesCondition,
distinct: true, distinct: true,
limit: newNodeCount, limit: newNodeCount,
orderBy: "last_net", orderBy: "last_net",
aostClause: asOf,
} }
} }
query := unionAll(asOf, newNodeQuery, reputableNodeQuery) query := unionAll(newNodeQuery, reputableNodeQuery)
query.query = cache.db.impl.WrapAsOfSystemInterval(query.query, criteria.AsOfSystemInterval)
rows, err := cache.db.Query(ctx, cache.db.Rebind(query.query), query.args...) rows, err := cache.db.Query(ctx, cache.db.Rebind(query.query), query.args...)
if err != nil { if err != nil {
@ -238,12 +233,11 @@ func nodeSelectionCondition(ctx context.Context, criteria *overlay.NodeCriteria,
// SELECT * FROM ($selection WHERE $condition ORDER BY $orderBy, RANDOM()) filtered ORDER BY RANDOM() LIMIT $limit // SELECT * FROM ($selection WHERE $condition ORDER BY $orderBy, RANDOM()) filtered ORDER BY RANDOM() LIMIT $limit
// //
type partialQuery struct { type partialQuery struct {
selection string selection string
condition condition condition condition
distinct bool distinct bool
orderBy string orderBy string
limit int limit int
aostClause string
} }
// isEmpty returns whether the result for the query is definitely empty. // isEmpty returns whether the result for the query is definitely empty.
@ -274,7 +268,7 @@ func (partial partialQuery) asQuery() query {
fmt.Fprint(&q, " LIMIT ? ") fmt.Fprint(&q, " LIMIT ? ")
args = append(args, partial.limit) args = append(args, partial.limit)
} else { } else {
fmt.Fprint(&q, ") filtered "+partial.aostClause+" ORDER BY RANDOM() LIMIT ?") fmt.Fprint(&q, ") filtered ORDER BY RANDOM() LIMIT ?")
args = append(args, partial.limit) args = append(args, partial.limit)
} }
@ -282,7 +276,7 @@ func (partial partialQuery) asQuery() query {
} }
// unionAll combines multiple partial queries into a single query. // unionAll combines multiple partial queries into a single query.
func unionAll(asOf string, partials ...partialQuery) query { func unionAll(partials ...partialQuery) query {
var queries []string var queries []string
var args []interface{} var args []interface{}
for _, partial := range partials { for _, partial := range partials {
@ -302,16 +296,8 @@ func unionAll(asOf string, partials ...partialQuery) query {
return query{query: queries[0], args: args} return query{query: queries[0], args: args}
} }
union := "(" + strings.Join(queries, ") UNION ALL (") + ")"
if asOf == "" {
return query{
query: union,
args: args,
}
}
return query{ return query{
query: "SELECT * FROM (" + union + ") " + asOf, query: "(" + strings.Join(queries, ") UNION ALL (") + ")",
args: args, args: args,
} }
} }