Support separate volume server ID independent of RPC bind address (#7609)

* pb: add id field to Heartbeat message for stable volume server identification

This adds an 'id' field to the Heartbeat protobuf message that allows
volume servers to identify themselves independently of their IP:port address.

Ref: https://github.com/seaweedfs/seaweedfs/issues/7487

* storage: add Id field to Store struct

Add Id field to Store struct and include it in CollectHeartbeat().
The Id field provides a stable volume server identity independent of IP:port.

Ref: https://github.com/seaweedfs/seaweedfs/issues/7487

* topology: support id-based DataNode identification

Update GetOrCreateDataNode to accept an id parameter for stable node
identification. When id is provided, the DataNode can maintain its identity
even when its IP address changes (e.g., in Kubernetes pod reschedules).

For backward compatibility:
- If id is provided, use it as the node ID
- If id is empty, fall back to ip:port

Ref: https://github.com/seaweedfs/seaweedfs/issues/7487

* volume: add -id flag for stable volume server identity

Add -id command line flag to volume server that allows specifying a stable
identifier independent of the IP address. This is useful for Kubernetes
deployments with hostPath volumes where pods can be rescheduled to different
nodes while the persisted data remains on the original node.

Usage: weed volume -id=node-1 -ip=10.0.0.1 ...

If -id is not specified, it defaults to ip:port for backward compatibility.

Fixes https://github.com/seaweedfs/seaweedfs/issues/7487

* server: add -volume.id flag to weed server command

Support the -volume.id flag in the all-in-one 'weed server' command,
consistent with the standalone 'weed volume' command.

Usage: weed server -volume.id=node-1 ...

Ref: https://github.com/seaweedfs/seaweedfs/issues/7487

* topology: add test for id-based DataNode identification

Test the key scenarios:
1. Create DataNode with explicit id
2. Same id with different IP returns same DataNode (K8s reschedule)
3. IP/PublicUrl are updated when node reconnects with new address
4. Different id creates new DataNode
5. Empty id falls back to ip:port (backward compatibility)

Ref: https://github.com/seaweedfs/seaweedfs/issues/7487

* pb: add address field to DataNodeInfo for proper node addressing

Previously, DataNodeInfo.Id was used as the node address, which worked
when Id was always ip:port. Now that Id can be an explicit string,
we need a separate Address field for connection purposes.

Changes:
- Add 'address' field to DataNodeInfo protobuf message
- Update ToDataNodeInfo() to populate the address field
- Update NewServerAddressFromDataNode() to use Address (with Id fallback)
- Fix LookupEcVolume to use dn.Url() instead of dn.Id()

Ref: https://github.com/seaweedfs/seaweedfs/issues/7487

* fix: trim whitespace from volume server id and fix test

- Trim whitespace from -id flag to treat ' ' as empty
- Fix store_load_balancing_test.go to include id parameter in NewStore call

Ref: https://github.com/seaweedfs/seaweedfs/issues/7487

* refactor: extract GetVolumeServerId to util package

Move the volume server ID determination logic to a shared utility function
to avoid code duplication between volume.go and rack.go.

Ref: https://github.com/seaweedfs/seaweedfs/issues/7487

* fix: improve transition logic for legacy nodes

- Use exact ip:port match instead of net.SplitHostPort heuristic
- Update GrpcPort and PublicUrl during transition for consistency
- Remove unused net import

Ref: https://github.com/seaweedfs/seaweedfs/issues/7487

* fix: add id normalization and address change logging

- Normalize id parameter at function boundary (trim whitespace)
- Log when DataNode IP:Port changes (helps debug K8s pod rescheduling)

Ref: https://github.com/seaweedfs/seaweedfs/issues/7487

1 files changed, 63 insertions, 6 deletions

diff --git a/weed/topology/rack.go b/weed/topology/rack.go
index f526cd84d..1e5c8b632 100644
--- a/weed/topology/rack.go
+++ b/weed/topology/rack.go
@@ -5,6 +5,7 @@ import (
 	"strings"
 	"time"
 
+	"github.com/seaweedfs/seaweedfs/weed/glog"
 	"github.com/seaweedfs/seaweedfs/weed/pb/master_pb"
 	"github.com/seaweedfs/seaweedfs/weed/storage/types"
 	"github.com/seaweedfs/seaweedfs/weed/util"
@@ -34,17 +35,73 @@ func (r *Rack) FindDataNode(ip string, port int) *DataNode {
 	}
 	return nil
 }
-func (r *Rack) GetOrCreateDataNode(ip string, port int, grpcPort int, publicUrl string, maxVolumeCounts map[string]uint32) *DataNode {
+
+// FindDataNodeById finds a DataNode by its ID using O(1) map lookup
+func (r *Rack) FindDataNodeById(id string) *DataNode {
+	r.RLock()
+	defer r.RUnlock()
+	if c, ok := r.children[NodeId(id)]; ok {
+		return c.(*DataNode)
+	}
+	return nil
+}
+
+func (r *Rack) GetOrCreateDataNode(ip string, port int, grpcPort int, publicUrl string, id string, maxVolumeCounts map[string]uint32) *DataNode {
 	r.Lock()
 	defer r.Unlock()
-	for _, c := range r.children {
+
+	// Normalize the id parameter (trim whitespace)
+	id = strings.TrimSpace(id)
+
+	// Determine the node ID: use provided id, or fall back to ip:port for backward compatibility
+	nodeId := util.GetVolumeServerId(id, ip, port)
+
+	// First, try to find by node ID using O(1) map lookup (stable identity)
+	if c, ok := r.children[NodeId(nodeId)]; ok {
 		dn := c.(*DataNode)
-		if dn.MatchLocation(ip, port) {
-			dn.LastSeen = time.Now().Unix()
-			return dn
+		// Log if IP or Port changed (e.g., pod rescheduled in K8s)
+		if dn.Ip != ip || dn.Port != port {
+			glog.V(0).Infof("DataNode %s address changed from %s:%d to %s:%d", nodeId, dn.Ip, dn.Port, ip, port)
 		}
+		// Update the IP/Port in case they changed
+		dn.Ip = ip
+		dn.Port = port
+		dn.GrpcPort = grpcPort
+		dn.PublicUrl = publicUrl
+		dn.LastSeen = time.Now().Unix()
+		return dn
 	}
-	dn := NewDataNode(util.JoinHostPort(ip, port))
+
+	// For backward compatibility: if explicit id was provided, also check by ip:port
+	// to handle transition from old (ip:port) to new (explicit id) behavior
+	ipPortId := util.JoinHostPort(ip, port)
+	if nodeId != ipPortId {
+		for oldId, c := range r.children {
+			dn := c.(*DataNode)
+			if dn.MatchLocation(ip, port) {
+				// Only transition if the oldId exactly matches ip:port (legacy identification).
+				// If oldId is different, this is a node with an explicit id that happens to
+				// reuse the same ip:port - don't incorrectly merge them.
+				if string(oldId) != ipPortId {
+					glog.Warningf("Volume server with id %s has ip:port %s which is used by node %s", nodeId, ipPortId, oldId)
+					continue
+				}
+				// Found a legacy node identified by ip:port, transition it to use the new explicit id
+				glog.V(0).Infof("Volume server %s transitioning id from %s to %s", dn.Url(), oldId, nodeId)
+				// Re-key the node in the children map with the new id
+				delete(r.children, oldId)
+				dn.id = NodeId(nodeId)
+				r.children[NodeId(nodeId)] = dn
+				// Update connection info in case they changed
+				dn.GrpcPort = grpcPort
+				dn.PublicUrl = publicUrl
+				dn.LastSeen = time.Now().Unix()
+				return dn
+			}
+		}
+	}
+
+	dn := NewDataNode(nodeId)
 	dn.Ip = ip
 	dn.Port = port
 	dn.GrpcPort = grpcPort