[#1640] object: Add priority metric based on geo distance

Change-Id: I3a7ea4fc4807392bf50e6ff1389c61367c953074
Signed-off-by: Anton Nikiforov <an.nikiforov@yadro.com>

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)
+	locodeDBPath := nodeconfig.LocodeDBPath(c)
+	parser, err := placement.NewMetricsParser(locodeDBPath, node.LOCODE())
 	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
 
@@ -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 {

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")
+}

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
+}

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)}

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)
+	})
 }