forked from TrueCloudLab/frostfs-node
[#1640] object: Add priority metric based on geo distance
Change-Id: I3a7ea4fc4807392bf50e6ff1389c61367c953074 Signed-off-by: Anton Nikiforov <an.nikiforov@yadro.com>
This commit is contained in:
parent
0712c113de
commit
56d09a9957
8 changed files with 241 additions and 38 deletions
|
@ -247,15 +247,16 @@ func (a *applicationConfiguration) readConfig(c *config.Config) error {
|
|||
// Object
|
||||
|
||||
a.ObjectCfg.tombstoneLifetime = objectconfig.TombstoneLifetime(c)
|
||||
var pm []placement.Metric
|
||||
for _, raw := range objectconfig.Get(c).Priority() {
|
||||
m, err := placement.ParseMetric(raw)
|
||||
locodeDBPath := nodeconfig.LocodeDBPath(c)
|
||||
parser, err := placement.NewMetricsParser(locodeDBPath)
|
||||
if err != nil {
|
||||
return err
|
||||
return fmt.Errorf("metrics parser creation: %w", err)
|
||||
}
|
||||
pm = append(pm, m)
|
||||
m, err := parser.ParseMetrics(objectconfig.Get(c).Priority())
|
||||
if err != nil {
|
||||
return fmt.Errorf("parse metrics: %w", err)
|
||||
}
|
||||
a.ObjectCfg.priorityMetrics = pm
|
||||
a.ObjectCfg.priorityMetrics = m
|
||||
|
||||
// Storage Engine
|
||||
|
||||
|
|
|
@ -217,3 +217,8 @@ func (l PersistentPolicyRulesConfig) NoSync() bool {
|
|||
func CompatibilityMode(c *config.Config) bool {
|
||||
return config.BoolSafe(c.Sub(subsection), "kludge_compatibility_mode")
|
||||
}
|
||||
|
||||
// LocodeDBPath returns path to LOCODE database.
|
||||
func LocodeDBPath(c *config.Config) string {
|
||||
return config.String(c.Sub(subsection), "locode_db_path")
|
||||
}
|
||||
|
|
|
@ -23,6 +23,7 @@ FROSTFS_NODE_ATTRIBUTE_1="UN-LOCODE:RU MSK"
|
|||
FROSTFS_NODE_RELAY=true
|
||||
FROSTFS_NODE_PERSISTENT_SESSIONS_PATH=/sessions
|
||||
FROSTFS_NODE_PERSISTENT_STATE_PATH=/state
|
||||
FROSTFS_NODE_LOCODE_DB_PATH=/path/to/locode/db
|
||||
|
||||
# Tree service section
|
||||
FROSTFS_TREE_ENABLED=true
|
||||
|
|
|
@ -37,7 +37,8 @@
|
|||
},
|
||||
"persistent_state": {
|
||||
"path": "/state"
|
||||
}
|
||||
},
|
||||
"locode_db_path": "/path/to/locode/db"
|
||||
},
|
||||
"grpc": {
|
||||
"0": {
|
||||
|
|
|
@ -36,6 +36,7 @@ node:
|
|||
path: /sessions # path to persistent session tokens file of Storage node (default: in-memory sessions)
|
||||
persistent_state:
|
||||
path: /state # path to persistent state file of Storage node
|
||||
"locode_db_path": "/path/to/locode/db"
|
||||
|
||||
grpc:
|
||||
- endpoint: s01.frostfs.devenv:8080 # endpoint for gRPC server
|
||||
|
|
|
@ -13,7 +13,8 @@ There are some custom types used for brevity:
|
|||
# Structure
|
||||
|
||||
| Section | Description |
|
||||
|------------------------|---------------------------------------------------------------------|
|
||||
|--------------|---------------------------------------------------------|
|
||||
| `node` | [Node parameters](#node-section) |
|
||||
| `logger` | [Logging parameters](#logger-section) |
|
||||
| `pprof` | [PProf configuration](#pprof-section) |
|
||||
| `prometheus` | [Prometheus metrics configuration](#prometheus-section) |
|
||||
|
@ -384,10 +385,11 @@ node:
|
|||
path: /sessions
|
||||
persistent_state:
|
||||
path: /state
|
||||
locode_db_path: "/path/to/locode/db"
|
||||
```
|
||||
|
||||
| Parameter | Type | Default value | Description |
|
||||
|-----------------------|---------------------------------------------------------------|---------------|-------------------------------------------------------------------------|
|
||||
|-----------------------|---------------------------------------------------------------|---------------|-----------------------------------------------------------------------------------------------------|
|
||||
| `key` | `string` | | Path to the binary-encoded private key. |
|
||||
| `wallet` | [Wallet config](#wallet-subsection) | | Wallet configuration. Has no effect if `key` is provided. |
|
||||
| `addresses` | `[]string` | | Addresses advertised in the netmap. |
|
||||
|
@ -395,6 +397,7 @@ node:
|
|||
| `relay` | `bool` | | Enable relay mode. |
|
||||
| `persistent_sessions` | [Persistent sessions config](#persistent_sessions-subsection) | | Persistent session token store configuration. |
|
||||
| `persistent_state` | [Persistent state config](#persistent_state-subsection) | | Persistent state configuration. |
|
||||
| `locode_db_path` | `string` | empty | Path to UN/LOCODE [database](https://git.frostfs.info/TrueCloudLab/frostfs-locode-db/) for FrostFS. |
|
||||
|
||||
## `wallet` subsection
|
||||
N3 wallet configuration.
|
||||
|
|
|
@ -2,24 +2,90 @@ package placement
|
|||
|
||||
import (
|
||||
"errors"
|
||||
"fmt"
|
||||
"maps"
|
||||
"math"
|
||||
"strings"
|
||||
"sync"
|
||||
"sync/atomic"
|
||||
|
||||
locodedb "git.frostfs.info/TrueCloudLab/frostfs-locode-db/pkg/locode/db"
|
||||
locodebolt "git.frostfs.info/TrueCloudLab/frostfs-locode-db/pkg/locode/db/boltdb"
|
||||
"git.frostfs.info/TrueCloudLab/frostfs-sdk-go/netmap"
|
||||
)
|
||||
|
||||
const (
|
||||
attrPrefix = "$attribute:"
|
||||
|
||||
geoDistance = "$geoDistance"
|
||||
)
|
||||
|
||||
type Metric interface {
|
||||
CalculateValue(*netmap.NodeInfo, *netmap.NodeInfo) int
|
||||
}
|
||||
|
||||
func ParseMetric(raw string) (Metric, error) {
|
||||
if attr, found := strings.CutPrefix(raw, attrPrefix); found {
|
||||
return NewAttributeMetric(attr), nil
|
||||
type metricsParser struct {
|
||||
locodeDBPath string
|
||||
locodes map[string]locodedb.Point
|
||||
}
|
||||
|
||||
type MetricParser interface {
|
||||
ParseMetrics([]string) ([]Metric, error)
|
||||
}
|
||||
|
||||
func NewMetricsParser(locodeDBPath string) (MetricParser, error) {
|
||||
return &metricsParser{
|
||||
locodeDBPath: locodeDBPath,
|
||||
}, nil
|
||||
}
|
||||
|
||||
func (p *metricsParser) initLocodes() error {
|
||||
if len(p.locodes) != 0 {
|
||||
return nil
|
||||
}
|
||||
return nil, errors.New("unsupported priority metric")
|
||||
if len(p.locodeDBPath) > 0 {
|
||||
p.locodes = make(map[string]locodedb.Point)
|
||||
locodeDB := locodebolt.New(locodebolt.Prm{
|
||||
Path: p.locodeDBPath,
|
||||
},
|
||||
locodebolt.ReadOnly(),
|
||||
)
|
||||
err := locodeDB.Open()
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
defer locodeDB.Close()
|
||||
err = locodeDB.IterateOverLocodes(func(k string, v locodedb.Point) {
|
||||
p.locodes[k] = v
|
||||
})
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
return nil
|
||||
}
|
||||
return errors.New("set path to locode database")
|
||||
}
|
||||
|
||||
func (p *metricsParser) ParseMetrics(priority []string) ([]Metric, error) {
|
||||
var metrics []Metric
|
||||
for _, raw := range priority {
|
||||
if attr, found := strings.CutPrefix(raw, attrPrefix); found {
|
||||
metrics = append(metrics, NewAttributeMetric(attr))
|
||||
} else if raw == geoDistance {
|
||||
err := p.initLocodes()
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
if len(p.locodes) == 0 {
|
||||
return nil, fmt.Errorf("provide locodes database for metric %s", raw)
|
||||
}
|
||||
m := NewGeoDistanceMetric(p.locodes)
|
||||
metrics = append(metrics, m)
|
||||
} else {
|
||||
return nil, fmt.Errorf("unsupported priority metric %s", raw)
|
||||
}
|
||||
}
|
||||
return metrics, nil
|
||||
}
|
||||
|
||||
// attributeMetric describes priority metric based on attribute.
|
||||
|
@ -41,3 +107,79 @@ func (am *attributeMetric) CalculateValue(from *netmap.NodeInfo, to *netmap.Node
|
|||
func NewAttributeMetric(attr string) Metric {
|
||||
return &attributeMetric{attribute: attr}
|
||||
}
|
||||
|
||||
// geoDistanceMetric describes priority metric based on attribute.
|
||||
type geoDistanceMetric struct {
|
||||
locodes map[string]locodedb.Point
|
||||
distance *atomic.Pointer[map[string]int]
|
||||
mtx sync.Mutex
|
||||
}
|
||||
|
||||
func NewGeoDistanceMetric(locodes map[string]locodedb.Point) Metric {
|
||||
d := atomic.Pointer[map[string]int]{}
|
||||
m := make(map[string]int)
|
||||
d.Store(&m)
|
||||
gm := &geoDistanceMetric{
|
||||
locodes: locodes,
|
||||
distance: &d,
|
||||
}
|
||||
return gm
|
||||
}
|
||||
|
||||
// CalculateValue return distance in kilometers between current node and provided,
|
||||
// if coordinates for provided node found. In other case return math.MaxInt.
|
||||
func (gm *geoDistanceMetric) CalculateValue(from *netmap.NodeInfo, to *netmap.NodeInfo) int {
|
||||
fl := from.LOCODE()
|
||||
tl := to.LOCODE()
|
||||
if fl == tl {
|
||||
return 0
|
||||
}
|
||||
m := gm.distance.Load()
|
||||
if v, ok := (*m)[fl+tl]; ok {
|
||||
return v
|
||||
}
|
||||
return gm.calculateDistance(fl, tl)
|
||||
}
|
||||
|
||||
func (gm *geoDistanceMetric) calculateDistance(from, to string) int {
|
||||
gm.mtx.Lock()
|
||||
defer gm.mtx.Unlock()
|
||||
od := gm.distance.Load()
|
||||
if v, ok := (*od)[from+to]; ok {
|
||||
return v
|
||||
}
|
||||
nd := maps.Clone(*od)
|
||||
var dist int
|
||||
pointFrom, okFrom := gm.locodes[from]
|
||||
pointTo, okTo := gm.locodes[to]
|
||||
if okFrom && okTo {
|
||||
dist = int(distance(pointFrom.Latitude(), pointFrom.Longitude(), pointTo.Latitude(), pointTo.Longitude()))
|
||||
} else {
|
||||
dist = math.MaxInt
|
||||
}
|
||||
nd[from+to] = dist
|
||||
gm.distance.Store(&nd)
|
||||
|
||||
return dist
|
||||
}
|
||||
|
||||
// distance return amount of KM between two points.
|
||||
// Parameters are latitude and longitude of point 1 and 2 in decimal degrees.
|
||||
// Original implementation can be found here https://www.geodatasource.com/developers/go.
|
||||
func distance(lt1 float64, ln1 float64, lt2 float64, ln2 float64) float64 {
|
||||
radLat1 := math.Pi * lt1 / 180
|
||||
radLat2 := math.Pi * lt2 / 180
|
||||
radTheta := math.Pi * (ln1 - ln2) / 180
|
||||
|
||||
dist := math.Sin(radLat1)*math.Sin(radLat2) + math.Cos(radLat1)*math.Cos(radLat2)*math.Cos(radTheta)
|
||||
|
||||
if dist > 1 {
|
||||
dist = 1
|
||||
}
|
||||
|
||||
dist = math.Acos(dist)
|
||||
dist = dist * 180 / math.Pi
|
||||
dist = dist * 60 * 1.1515 * 1.609344
|
||||
|
||||
return dist
|
||||
}
|
||||
|
|
|
@ -601,4 +601,53 @@ func TestTraverserPriorityMetrics(t *testing.T) {
|
|||
next = tr.Next()
|
||||
require.Nil(t, next)
|
||||
})
|
||||
|
||||
t.Run("one rep one geo metric", func(t *testing.T) {
|
||||
t.Skip()
|
||||
selectors := []int{2}
|
||||
replicas := []int{2}
|
||||
|
||||
nodes, cnr := testPlacement(selectors, replicas)
|
||||
|
||||
// Node_0, PK - ip4/0.0.0.0/tcp/0
|
||||
nodes[0][0].SetAttribute("UN-LOCODE", "RU MOW")
|
||||
// Node_1, PK - ip4/0.0.0.0/tcp/1
|
||||
nodes[0][1].SetAttribute("UN-LOCODE", "RU LED")
|
||||
|
||||
sdkNode := testNode(2)
|
||||
sdkNode.SetAttribute("UN-LOCODE", "FI HEL")
|
||||
|
||||
nodesCopy := copyVectors(nodes)
|
||||
|
||||
parser, err := NewMetricsParser("/path/to/locode_db")
|
||||
require.NoError(t, err)
|
||||
m, err := parser.ParseMetrics([]string{geoDistance})
|
||||
require.NoError(t, err)
|
||||
|
||||
tr, err := NewTraverser(context.Background(),
|
||||
ForContainer(cnr),
|
||||
UseBuilder(&testBuilder{
|
||||
vectors: nodesCopy,
|
||||
}),
|
||||
WithoutSuccessTracking(),
|
||||
WithPriorityMetrics(m),
|
||||
WithNodeState(&nodeState{
|
||||
node: &sdkNode,
|
||||
}),
|
||||
)
|
||||
require.NoError(t, err)
|
||||
|
||||
// Without priority metric `$geoDistance` the order will be:
|
||||
// [ {Node_0 RU MOW}, {Node_1 RU LED}]
|
||||
// With priority metric `$geoDistance` the order should be:
|
||||
// [ {Node_1 RU LED}, {Node_0 RU MOW}]
|
||||
next := tr.Next()
|
||||
require.NotNil(t, next)
|
||||
require.Equal(t, 2, len(next))
|
||||
require.Equal(t, "/ip4/0.0.0.0/tcp/1", string(next[0].PublicKey()))
|
||||
require.Equal(t, "/ip4/0.0.0.0/tcp/0", string(next[1].PublicKey()))
|
||||
|
||||
next = tr.Next()
|
||||
require.Nil(t, next)
|
||||
})
|
||||
}
|
||||
|
|
Loading…
Add table
Reference in a new issue