From 98d690063c71a178caaffc32a99f2fa7b9b0c631 Mon Sep 17 00:00:00 2001 From: Anton Nikiforov Date: Fri, 7 Feb 2025 17:09:08 +0300 Subject: [PATCH] [#1640] object: Add priority metric based on geo distance Signed-off-by: Anton Nikiforov --- cmd/frostfs-node/config.go | 27 ++++- cmd/frostfs-node/config/node/config.go | 5 + cmd/frostfs-node/main.go | 4 + .../object_manager/placement/metrics.go | 108 ++++++++++++++++-- .../object_manager/placement/traverser.go | 15 ++- .../placement/traverser_test.go | 55 +++++++++ 6 files changed, 197 insertions(+), 17 deletions(-) diff --git a/cmd/frostfs-node/config.go b/cmd/frostfs-node/config.go index bfe73dc85..d251586fd 100644 --- a/cmd/frostfs-node/config.go +++ b/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 { 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/cmd/frostfs-node/main.go b/cmd/frostfs-node/main.go index b53bd8823..546b2fd32 100644 --- a/cmd/frostfs-node/main.go +++ b/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()) { diff --git a/pkg/services/object_manager/placement/metrics.go b/pkg/services/object_manager/placement/metrics.go index 45e6df339..b20b0553f 100644 --- a/pkg/services/object_manager/placement/metrics.go +++ b/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 +} diff --git a/pkg/services/object_manager/placement/traverser.go b/pkg/services/object_manager/placement/traverser.go index efa4a5b06..4cc543858 100644 --- a/pkg/services/object_manager/placement/traverser.go +++ b/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. diff --git a/pkg/services/object_manager/placement/traverser_test.go b/pkg/services/object_manager/placement/traverser_test.go index 9c825bf19..4655d2550 100644 --- a/pkg/services/object_manager/placement/traverser_test.go +++ b/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()) + }) }