1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
|
package pb
import (
"fmt"
"net"
"strconv"
"strings"
"github.com/seaweedfs/seaweedfs/weed/pb/master_pb"
"github.com/seaweedfs/seaweedfs/weed/util"
)
type ServerAddress string
type ServerAddresses string
type ServerSrvAddress string
func NewServerAddress(host string, port int, grpcPort int) ServerAddress {
if grpcPort == 0 || grpcPort == port+10000 {
return ServerAddress(util.JoinHostPort(host, port))
}
return ServerAddress(util.JoinHostPort(host, port) + "." + strconv.Itoa(grpcPort))
}
func NewServerAddressWithGrpcPort(address string, grpcPort int) ServerAddress {
if grpcPort == 0 {
return ServerAddress(address)
}
_, port, _ := hostAndPort(address)
if uint64(grpcPort) == port+10000 {
return ServerAddress(address)
}
return ServerAddress(address + "." + strconv.Itoa(grpcPort))
}
func NewServerAddressFromDataNode(dn *master_pb.DataNodeInfo) ServerAddress {
// Use Address field if available (new behavior), fall back to Id for backward compatibility
addr := dn.Address
if addr == "" {
addr = dn.Id // backward compatibility: old nodes use ip:port as id
}
return NewServerAddressWithGrpcPort(addr, int(dn.GrpcPort))
}
func NewServerAddressFromLocation(dn *master_pb.Location) ServerAddress {
return NewServerAddressWithGrpcPort(dn.Url, int(dn.GrpcPort))
}
func (sa ServerAddress) String() string {
return sa.ToHttpAddress()
}
func (sa ServerAddress) ToHttpAddress() string {
portsSepIndex := strings.LastIndex(string(sa), ":")
if portsSepIndex < 0 {
return string(sa)
}
if portsSepIndex+1 >= len(sa) {
return string(sa)
}
ports := string(sa[portsSepIndex+1:])
sepIndex := strings.LastIndex(string(ports), ".")
if sepIndex >= 0 {
host := string(sa[0:portsSepIndex])
return net.JoinHostPort(host, ports[0:sepIndex])
}
return string(sa)
}
func (sa ServerAddress) ToGrpcAddress() string {
portsSepIndex := strings.LastIndex(string(sa), ":")
if portsSepIndex < 0 {
return string(sa)
}
if portsSepIndex+1 >= len(sa) {
return string(sa)
}
ports := string(sa[portsSepIndex+1:])
sepIndex := strings.LastIndex(ports, ".")
if sepIndex >= 0 {
host := string(sa[0:portsSepIndex])
return net.JoinHostPort(host, ports[sepIndex+1:])
}
return ServerToGrpcAddress(string(sa))
}
// LookUp may return an error for some records along with successful lookups - make sure you do not
// discard `addresses` even if `err == nil`
func (r ServerSrvAddress) LookUp() (addresses []ServerAddress, err error) {
_, records, lookupErr := net.LookupSRV("", "", string(r))
if lookupErr != nil {
err = fmt.Errorf("lookup SRV address %s: %v", r, lookupErr)
}
for _, srv := range records {
address := fmt.Sprintf("%s:%d", srv.Target, srv.Port)
addresses = append(addresses, ServerAddress(address))
}
return
}
// ToServiceDiscovery expects one of: a comma-separated list of ip:port, like
//
// 10.0.0.1:9999,10.0.0.2:24:9999
//
// OR an SRV Record prepended with 'dnssrv+', like:
//
// dnssrv+_grpc._tcp.master.consul
// dnssrv+_grpc._tcp.headless.default.svc.cluster.local
// dnssrv+seaweed-master.master.consul
func (sa ServerAddresses) ToServiceDiscovery() (sd *ServerDiscovery) {
sd = &ServerDiscovery{}
prefix := "dnssrv+"
if strings.HasPrefix(string(sa), prefix) {
trimmed := strings.TrimPrefix(string(sa), prefix)
srv := ServerSrvAddress(trimmed)
sd.srvRecord = &srv
} else {
sd.list = sa.ToAddresses()
}
return
}
func (sa ServerAddresses) ToAddresses() (addresses []ServerAddress) {
parts := strings.Split(string(sa), ",")
for _, address := range parts {
if address != "" {
addresses = append(addresses, ServerAddress(address))
}
}
return
}
func (sa ServerAddresses) ToAddressMap() (addresses map[string]ServerAddress) {
addresses = make(map[string]ServerAddress)
for _, address := range sa.ToAddresses() {
addresses[string(address)] = address
}
return
}
func (sa ServerAddresses) ToAddressStrings() (addresses []string) {
parts := strings.Split(string(sa), ",")
for _, address := range parts {
addresses = append(addresses, address)
}
return
}
func ToAddressStrings(addresses []ServerAddress) []string {
var strings []string
for _, addr := range addresses {
strings = append(strings, string(addr))
}
return strings
}
func ToAddressStringsFromMap(addresses map[string]ServerAddress) []string {
var strings []string
for _, addr := range addresses {
strings = append(strings, string(addr))
}
return strings
}
func FromAddressStrings(strings []string) []ServerAddress {
var addresses []ServerAddress
for _, addr := range strings {
addresses = append(addresses, ServerAddress(addr))
}
return addresses
}
func ParseUrl(input string) (address ServerAddress, path string, err error) {
if !strings.HasPrefix(input, "http://") {
return "", "", fmt.Errorf("url %s needs prefix 'http://'", input)
}
input = input[7:]
pathSeparatorIndex := strings.Index(input, "/")
hostAndPorts := input
if pathSeparatorIndex > 0 {
path = input[pathSeparatorIndex:]
hostAndPorts = input[0:pathSeparatorIndex]
}
commaSeparatorIndex := strings.Index(input, ":")
if commaSeparatorIndex < 0 {
err = fmt.Errorf("port should be specified in %s", input)
return
}
address = ServerAddress(hostAndPorts)
return
}
|
