diff --git a/cmd/frostfs-node/config.go b/cmd/frostfs-node/config.go index 33980b3e3..17b22f072 100644 --- a/cmd/frostfs-node/config.go +++ b/cmd/frostfs-node/config.go @@ -208,7 +208,7 @@ type subStorageCfg struct { // readConfig fills applicationConfiguration with raw configuration values // not modifying them. -func (a *applicationConfiguration) readConfig(c *config.Config) error { +func (a *applicationConfiguration) readConfig(c *config.Config, node netmap.NodeInfo) error { if a._read { err := c.Reload() if err != nil { @@ -246,15 +246,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) - if err != nil { - return err - } - pm = append(pm, m) + locodeDBPath := nodeconfig.LocodeDBPath(c) + parser, err := placement.NewMetricsParser(locodeDBPath, node.LOCODE()) + if err != nil { + return fmt.Errorf("metrics parser creation: %w", err) } - a.ObjectCfg.priorityMetrics = pm + m, err := parser.ParseMetrics(objectconfig.Get(c).Priority()) + if err != nil { + return fmt.Errorf("parse metrics: %w", err) + } + a.ObjectCfg.priorityMetrics = m // Storage Engine @@ -718,7 +719,7 @@ func initCfg(appCfg *config.Config) *cfg { } initLocalNodeInfo(c, key, netAddr, attrs) - err := c.readConfig(appCfg) + err := c.readConfig(appCfg, c.cfgNodeInfo.localInfo) if err != nil { panic(fmt.Errorf("config reading: %w", err)) } @@ -1451,7 +1452,7 @@ func (c *cfg) reloadAppConfig() error { unlock := c.LockAppConfigExclusive() defer unlock() - return c.readConfig(c.appCfg) + return c.readConfig(c.appCfg, c.cfgNodeInfo.localInfo) } func (c *cfg) createTombstoneSource() *tombstone.ExpirationChecker { diff --git a/cmd/frostfs-node/config/node/config.go b/cmd/frostfs-node/config/node/config.go index 969d77396..fa865ff83 100644 --- a/cmd/frostfs-node/config/node/config.go +++ b/cmd/frostfs-node/config/node/config.go @@ -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).Sub("locode"), "db_path") +} diff --git a/pkg/services/object_manager/placement/metrics.go b/pkg/services/object_manager/placement/metrics.go index 45e6df339..687365b24 100644 --- a/pkg/services/object_manager/placement/metrics.go +++ b/pkg/services/object_manager/placement/metrics.go @@ -2,24 +2,99 @@ 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 + nodeLocode string +} + +type MetricParser interface { + ParseMetrics([]string) ([]Metric, error) +} + +func NewMetricsParser(locodeDBPath string, nodeLocode string) (MetricParser, error) { + return &metricsParser{ + locodeDBPath: locodeDBPath, + nodeLocode: nodeLocode, + }, 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) + } + if len(p.nodeLocode) == 0 { + return nil, errors.New("set locode for node") + } + point, ok := p.locodes[p.nodeLocode] + if !ok { + return nil, fmt.Errorf("not found geo coordinates for locode %s", raw) + } + m := NewGeoDistanceMetric(p.locodes, point.Latitude(), point.Longitude()) + 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 +116,74 @@ 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 + lat float64 + long float64 +} + +func NewGeoDistanceMetric(locodes map[string]locodedb.Point, lat float64, long float64) Metric { + d := atomic.Pointer[map[string]int]{} + m := make(map[string]int) + d.Store(&m) + gm := &geoDistanceMetric{ + locodes: locodes, + lat: lat, + long: long, + 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(_ *netmap.NodeInfo, to *netmap.NodeInfo) int { + tl := to.LOCODE() + m := gm.distance.Load() + if v, ok := (*m)[tl]; ok { + return v + } + return gm.calculateDistance(tl) +} + +func (gm *geoDistanceMetric) calculateDistance(to string) int { + gm.mtx.Lock() + defer gm.mtx.Unlock() + od := gm.distance.Load() + if v, ok := (*od)[to]; ok { + return v + } + nd := maps.Clone(*od) + pointTo, ok := gm.locodes[to] + if ok { + nd[to] = int(distance(gm.lat, gm.long, pointTo.Latitude(), pointTo.Longitude())) + } else { + nd[to] = math.MaxInt + } + gm.distance.Store(&nd) + return nd[to] +} + +// distance return amount of KM between two points. +// Parameters are latitude and longitude of point 1 and 2 in decimal degrees. +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 +} diff --git a/pkg/services/object_manager/placement/traverser.go b/pkg/services/object_manager/placement/traverser.go index a3f9af959..596c92b58 100644 --- a/pkg/services/object_manager/placement/traverser.go +++ b/pkg/services/object_manager/placement/traverser.go @@ -120,9 +120,7 @@ func NewTraverser(ctx context.Context, opts ...Option) (*Traverser, error) { regularVector = append(regularVector, ns[i][pivot:]...) } rem = []int{-1, -1} - - sortedVector := sortVector(cfg, unsortedVector) - ns = [][]netmap.NodeInfo{sortedVector, regularVector} + ns = [][]netmap.NodeInfo{sortVector(cfg, unsortedVector), regularVector} } else if cfg.flatSuccess != nil { ns = flatNodes(ns) rem = []int{int(*cfg.flatSuccess)} diff --git a/pkg/services/object_manager/placement/traverser_test.go b/pkg/services/object_manager/placement/traverser_test.go index 9c825bf19..3e66279f0 100644 --- a/pkg/services/object_manager/placement/traverser_test.go +++ b/pkg/services/object_manager/placement/traverser_test.go @@ -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", sdkNode.LOCODE()) + 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) + }) }