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

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

cmd/frostfs-node/config.go

@@ -15,6 +15,7 @@ import (
 	"syscall"
 	"time"
 
+	locodebolt "git.frostfs.info/TrueCloudLab/frostfs-locode-db/pkg/locode/db/boltdb"
 	"git.frostfs.info/TrueCloudLab/frostfs-node/cmd/frostfs-node/config"
 	apiclientconfig "git.frostfs.info/TrueCloudLab/frostfs-node/cmd/frostfs-node/config/apiclient"
 	"git.frostfs.info/TrueCloudLab/frostfs-node/cmd/frostfs-node/config/audit"
@@ -205,7 +206,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, db *locodebolt.DB, node netmap.NodeInfo) error {
 	if a._read {
 		err := c.Reload()
 		if err != nil {
@@ -237,7 +238,7 @@ func (a *applicationConfiguration) readConfig(c *config.Config) error {
 	a.ObjectCfg.tombstoneLifetime = objectconfig.TombstoneLifetime(c)
 	var pm []placement.Metric
 	for _, raw := range objectconfig.Get(c).Priority() {
-		m, err := placement.ParseMetric(raw)
+		m, err := placement.ParseMetric(raw, db, node.LOCODE())
 		if err != nil {
 			return err
 		}
@@ -393,6 +394,7 @@ type internals struct {
 	// is node under maintenance
 	isMaintenance atomic.Bool
 	audit         *atomic.Bool
+	locodeDB      *locodebolt.DB
 
 	sdNotify bool
 }
@@ -695,7 +697,19 @@ func initCfg(appCfg *config.Config) *cfg {
 	}
 	initLocalNodeInfo(c, key, netAddr, attrs)
 
-	err := c.readConfig(appCfg)
+	var locodeDB *locodebolt.DB
+	locodeDBPath := nodeconfig.LocodeDBPath(appCfg)
+	if len(locodeDBPath) > 0 {
+		locodeDB = locodebolt.New(locodebolt.Prm{
+			Path: locodeDBPath,
+		},
+			locodebolt.ReadOnly(),
+		)
+		err := locodeDB.Open()
+		fatalOnErr(err)
+	}
+
+	err := c.readConfig(appCfg, locodeDB, c.cfgNodeInfo.localInfo)
 	if err != nil {
 		panic(fmt.Errorf("config reading: %w", err))
 	}
@@ -717,7 +731,7 @@ func initCfg(appCfg *config.Config) *cfg {
 		}))
 	}
 
-	c.internals = initInternals(appCfg, log)
+	c.internals = initInternals(appCfg, log, locodeDB)
 
 	c.cfgAccounting = cfgAccounting{
 		scriptHash: contractsconfig.Balance(appCfg),
@@ -754,7 +768,7 @@ func initLocalNodeInfo(c *cfg, key *keys.PrivateKey, netAddr network.AddressGrou
 	c.cfgNodeInfo.localInfo.SetStatus(netmap.Offline)
 }
 
-func initInternals(appCfg *config.Config, log *logger.Logger) internals {
+func initInternals(appCfg *config.Config, log *logger.Logger, db *locodebolt.DB) internals {
 	var healthStatus atomic.Int32
 	healthStatus.Store(int32(control.HealthStatus_HEALTH_STATUS_UNDEFINED))
 
@@ -770,6 +784,7 @@ func initInternals(appCfg *config.Config, log *logger.Logger) internals {
 		healthStatus: &healthStatus,
 		sdNotify:     initSdNotify(appCfg),
 		audit:        &auditRequests,
+		locodeDB:     db,
 	}
 }
 
@@ -1449,7 +1464,7 @@ func (c *cfg) reloadAppConfig() error {
 	unlock := c.LockAppConfigExclusive()
 	defer unlock()
 
-	return c.readConfig(c.appCfg)
+	return c.readConfig(c.appCfg, c.internals.locodeDB, 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")
+}

cmd/frostfs-node/main.go

@@ -172,6 +172,10 @@ func wait(c *cfg) {
 
 	close(c.internalErr)
 	drain.Wait()
+
+	if c.locodeDB != nil {
+		_ = c.locodeDB.Close()
+	}
 }
 
 func (c *cfg) onShutdown(f func()) {

pkg/services/object_manager/placement/metrics.go

@@ -1,25 +1,34 @@
 package placement
 
 import (
-	"errors"
+	"fmt"
+	"math"
 	"strings"
+	"sync"
 
+	"git.frostfs.info/TrueCloudLab/frostfs-locode-db/pkg/locode"
+	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
+	CalculateValue(*netmap.NodeInfo, *netmap.NodeInfo) (int, error)
 }
 
-func ParseMetric(raw string) (Metric, error) {
+func ParseMetric(raw string, locodeDB *locodebolt.DB, nodeLocode string) (Metric, error) {
 	if attr, found := strings.CutPrefix(raw, attrPrefix); found {
 		return NewAttributeMetric(attr), nil
+	} else if raw == geoDistance && len(nodeLocode) > 0 && locodeDB != nil {
+		return NewGeoDistanceMetric(locodeDB, nodeLocode)
 	}
-	return nil, errors.New("unsupported priority metric")
+	return nil, fmt.Errorf("unsupported priority metric %s", raw)
 }
 
 // attributeMetric describes priority metric based on attribute.
@@ -29,15 +38,100 @@ type attributeMetric struct {
 
 // CalculateValue return [0] if from and to contains attribute attributeMetric.attribute and
 // the value of attribute is the same. In other case return [1].
-func (am *attributeMetric) CalculateValue(from *netmap.NodeInfo, to *netmap.NodeInfo) int {
+func (am *attributeMetric) CalculateValue(from *netmap.NodeInfo, to *netmap.NodeInfo) (int, error) {
 	fromAttr := from.Attribute(am.attribute)
 	toAttr := to.Attribute(am.attribute)
 	if len(fromAttr) > 0 && len(toAttr) > 0 && fromAttr == toAttr {
-		return 0
+		return 0, nil
 	}
-	return 1
+	return 1, nil
 }
 
 func NewAttributeMetric(attr string) Metric {
 	return &attributeMetric{attribute: attr}
 }
+
+// geoDistanceMetric describes priority metric based on attribute.
+type geoDistanceMetric struct {
+	locodeDB *locodebolt.DB
+	distance map[string]int
+	mtx      sync.Mutex
+	lat      float64
+	long     float64
+}
+
+func NewGeoDistanceMetric(locodeDB *locodebolt.DB, nodeLocode string) (Metric, error) {
+	point, err := getPoint(locodeDB, nodeLocode)
+	if err != nil {
+		return nil, fmt.Errorf("geo point for locode %s: %w", nodeLocode, err)
+	}
+	gm := &geoDistanceMetric{
+		locodeDB: locodeDB,
+		lat:      point.Latitude(),
+		long:     point.Longitude(),
+		distance: make(map[string]int),
+	}
+	return gm, nil
+}
+
+// CalculateValue return [0] if from and to contains attribute attributeMetric.attribute and
+// the value of attribute is the same. In other case return [1].
+func (gm *geoDistanceMetric) CalculateValue(_ *netmap.NodeInfo, to *netmap.NodeInfo) (int, error) {
+	tl := to.LOCODE()
+	if v, ok := gm.distance[tl]; ok {
+		return v, nil
+	}
+	return gm.calculateDistance(tl)
+}
+
+func (gm *geoDistanceMetric) calculateDistance(to string) (int, error) {
+	gm.mtx.Lock()
+	defer gm.mtx.Unlock()
+	if v, ok := gm.distance[to]; ok {
+		return v, nil
+	}
+
+	pointTo, err := getPoint(gm.locodeDB, to)
+	if err != nil {
+		return 0, fmt.Errorf("geo point for locode %s: %w", to, err)
+	}
+
+	gm.distance[to] = int(distance(gm.lat, gm.long, pointTo.Latitude(), pointTo.Longitude()))
+	return gm.distance[to], nil
+}
+
+func getPoint(db *locodebolt.DB, raw string) (*locodedb.Point, error) {
+	lc, err := locode.FromString(raw)
+	if err != nil {
+		return nil, fmt.Errorf("invalid locode value: %w", err)
+	}
+	key, err := locodedb.NewKey(*lc)
+	if err != nil {
+		return nil, fmt.Errorf("create key from locode: %w", err)
+	}
+	record, err := db.Get(*key)
+	if err != nil {
+		return nil, fmt.Errorf("could not get locode record from DB: %w", err)
+	}
+	return record.GeoPoint(), nil
+}
+
+// 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

@@ -121,7 +121,10 @@ func NewTraverser(ctx context.Context, opts ...Option) (*Traverser, error) {
 		}
 		rem = []int{-1, -1}
 
-		sortedVector := sortVector(cfg, unsortedVector)
+		sortedVector, err := sortVector(cfg, unsortedVector)
+		if err != nil {
+			return nil, fmt.Errorf("could not sort nodes by metrics: %w", err)
+		}
 		ns = [][]netmap.NodeInfo{sortedVector, regularVector}
 	} else if cfg.flatSuccess != nil {
 		ns = flatNodes(ns)
@@ -186,14 +189,18 @@ type nodeMetrics struct {
 	metrics []int
 }
 
-func sortVector(cfg *cfg, unsortedVector []netmap.NodeInfo) []netmap.NodeInfo {
+func sortVector(cfg *cfg, unsortedVector []netmap.NodeInfo) ([]netmap.NodeInfo, error) {
 	nm := make([]nodeMetrics, len(unsortedVector))
 	node := cfg.nodeState.LocalNodeInfo()
 
+	var err error
 	for i := range unsortedVector {
 		m := make([]int, len(cfg.metrics))
 		for j, pm := range cfg.metrics {
-			m[j] = pm.CalculateValue(node, &unsortedVector[i])
+			m[j], err = pm.CalculateValue(node, &unsortedVector[i])
+			if err != nil {
+				return nil, err
+			}
 		}
 		nm[i] = nodeMetrics{
 			index:   i,
@@ -207,7 +214,7 @@ func sortVector(cfg *cfg, unsortedVector []netmap.NodeInfo) []netmap.NodeInfo {
 	for i := range unsortedVector {
 		sortedVector[i] = unsortedVector[nm[i].index]
 	}
-	return sortedVector
+	return sortedVector, nil
 }
 
 // Node is a descriptor of storage node with information required for intra-container communication.

pkg/services/object_manager/placement/traverser_test.go

@@ -6,6 +6,7 @@ import (
 	"strconv"
 	"testing"
 
+	locodebolt "git.frostfs.info/TrueCloudLab/frostfs-locode-db/pkg/locode/db/boltdb"
 	netmapcore "git.frostfs.info/TrueCloudLab/frostfs-node/pkg/core/netmap"
 	"git.frostfs.info/TrueCloudLab/frostfs-node/pkg/network"
 	"git.frostfs.info/TrueCloudLab/frostfs-sdk-go/container"
@@ -601,4 +602,58 @@ 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)
+
+		locodeDB := locodebolt.New(locodebolt.Prm{
+			Path: "/path/to/locode_db",
+		},
+			locodebolt.ReadOnly(),
+		)
+		require.NoError(t, locodeDB.Open())
+		m, err := NewGeoDistanceMetric(locodeDB, "FI HEL")
+		require.NoError(t, err)
+
+		tr, err := NewTraverser(context.Background(),
+			ForContainer(cnr),
+			UseBuilder(&testBuilder{
+				vectors: nodesCopy,
+			}),
+			WithoutSuccessTracking(),
+			WithPriorityMetrics([]Metric{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)
+		require.NoError(t, locodeDB.Close())
+	})
 }