LinkPort/internal/cluster/cluster.go

// Package cluster discovers upstream topology without replaying application requests.
// Each accepted client session remains pinned to its selected node.
package cluster

import (
	"context"
	"errors"
	"linkport/internal/model"
	"net"
	"strconv"
	"time"
)

func node(p model.Proxy) model.ClusterNode {
	return model.ClusterNode{Host: p.TargetHost, Port: p.TargetPort}
}
func endpoint(address string, defaultPort int) (model.ClusterNode, bool) {
	h, raw, err := net.SplitHostPort(address)
	if err != nil {
		h = address
		raw = strconv.Itoa(defaultPort)
	}
	port, err := strconv.Atoi(raw)
	n := model.ClusterNode{Host: h, Port: port}
	return n, err == nil && n.Valid()
}

func addNodes(dst []model.ClusterNode, nodes ...model.ClusterNode) []model.ClusterNode {
	for _, n := range nodes {
		if !n.Valid() {
			continue
		}
		found := false
		for i, old := range dst {
			if old.Address() == n.Address() {
				found = true
				if n.Role != "" {
					dst[i].Role = n.Role
				}
				break
			}
		}
		if !found && len(dst) < 64 {
			dst = append(dst, n)
		}
	}
	return dst
}

// Inspect is read-only; ambiguous or privilege-limited discovery remains explicit.
func Inspect(ctx context.Context, p model.Proxy, dial model.DialFunc) (*model.ClusterInfo, error) {
	if !model.ClusterProtocol(p.Type) {
		return nil, errors.New("cluster detection is unsupported for this protocol")
	}
	if err := p.CheckTargetCredential(); err != nil {
		return nil, err
	}
	if p.Outbound.Mode == "k8s" {
		return nil, errors.New("cluster discovery cannot follow nodes through a fixed Kubernetes resource")
	}
	return inspect(ctx, p, dial, probe)
}

// inspect owns traversal, deadlines and topology selection. The protocol
// strategy only observes one candidate and never replays application requests.
func inspect(ctx context.Context, p model.Proxy, dial model.DialFunc, probe probeStrategy) (*model.ClusterInfo, error) {
	ctx, cancel := context.WithTimeout(ctx, 20*time.Second)
	defer cancel()
	queue := addNodes(nil, append([]model.ClusterNode{node(p)}, p.Cluster.Nodes...)...)
	info := &model.ClusterInfo{Type: "unknown", Nodes: queue}
	var reachable *model.ClusterNode
	for i := 0; i < len(queue) && i < 64; i++ {
		if ctx.Err() != nil {
			break
		}
		candidate := p
		candidate.TargetHost, candidate.TargetPort = queue[i].Host, queue[i].Port
		if info.Type == "redis-sentinel" && queue[i].Role == "primary" {
			candidate.Cluster.Mode = "auto"
			candidate.Cluster.SentinelAccountID = ""
		}
		attempt, stop := context.WithTimeout(ctx, 3*time.Second)
		got, err := probe(attempt, candidate, dial)
		stop()
		if err != nil {
			continue
		}
		if got.Selected == nil {
			continue
		}
		if p.Cluster.Name != "" && got.Name != "" && p.Cluster.Name != got.Name {
			continue
		}
		if info.Name != "" && got.Name != "" && info.Name != got.Name {
			continue
		}
		fromSentinel := info.Type == "redis-sentinel" && (got.Type == "redis-replication" || got.Type == "standalone")
		fromReplication := info.Type == "redis-replication" && got.Type == "standalone"
		if !fromSentinel && !fromReplication && p.Cluster.Mode != "" && p.Cluster.Mode != "single" && p.Cluster.Mode != "auto" && p.Cluster.Mode != "failover" && got.Type != "unknown" && got.Type != p.Cluster.Mode {
			continue
		}
		if !fromSentinel && (info.Type == "unknown" || got.Type != "standalone") {
			info.Type, info.Name, info.Warning = got.Type, got.Name, got.Warning
		}
		info.Nodes = addNodes(info.Nodes, got.Nodes...)
		queue = addNodes(queue, got.Nodes...)
		if reachable == nil && got.Selected.Role != "sentinel" {
			copy := *got.Selected
			reachable = &copy
		}
		if got.Selected.Role == "primary" || got.Selected.Role == "router" || got.Selected.Role == "standalone" || got.Selected.Role == "broker" || got.Type == "redis-cluster" {
			info.Selected = got.Selected
			info.Version = got.Version
			return info, nil
		}
	}
	if reachable != nil && info.Type == "unknown" && (p.Cluster.Mode == "auto" || p.Cluster.Mode == "failover" || p.Cluster.Mode == "" || p.Cluster.Mode == "single") {
		info.Selected = reachable
		info.Warning = "服务器未提供足够拓扑信息；当前节点可连接，但无法确认集群类型或主节点。"
		return info, nil
	}
	if len(info.Nodes) > 0 && info.Type != "unknown" {
		if info.Warning == "" {
			info.Warning = "已识别拓扑，但未找到可用主节点；请检查节点地址、权限及集群名称。"
		}
		return info, nil
	}
	return nil, errors.New("cluster discovery failed: no reachable matching node")
}

func Resolve(ctx context.Context, p model.Proxy, dial model.DialFunc) (model.Proxy, *model.ClusterInfo, error) {
	if !p.ClusterEnabled() {
		return p, nil, nil
	}
	info, err := Inspect(ctx, p, dial)
	if err != nil {
		return p, nil, err
	}
	if info.Selected == nil {
		return p, info, errors.New("cluster has no available primary or router")
	}
	p.TargetHost, p.TargetPort = info.Selected.Host, info.Selected.Port
	// Resolved snapshots must not recursively discover topology in protocol helpers.
	p.Cluster = model.ClusterConfig{}
	return p, info, nil
}

type contextDialer struct{ dial model.DialFunc }

func (d contextDialer) DialContext(ctx context.Context, network, address string) (net.Conn, error) {
	return d.dial(ctx, network, address)
}

func singleInfo(p model.Proxy, kind, role string) *model.ClusterInfo {
	n := node(p)
	n.Role = role
	return &model.ClusterInfo{Type: kind, Nodes: []model.ClusterNode{n}, Selected: &n}
}