Skip to content

Commit 9dfb8d8

Browse files
rleungxrleungx
committed
tests: avoid test port conflicts
Signed-off-by: Ryan Leung <rleungx@gmail.com>
1 parent 5c811b2 commit 9dfb8d8

21 files changed

Lines changed: 295 additions & 93 deletions

File tree

‎pkg/mcs/resourcemanager/server/server.go‎

Lines changed: 30 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -133,6 +133,19 @@ func (s *Server) Run() (err error) {
133133
if err = utils.InitClient(s); err != nil {
134134
return err
135135
}
136+
if err = s.initListenerAndUpdateConfig(); err != nil {
137+
return err
138+
}
139+
defer func() {
140+
if err != nil && s.serviceRegister != nil {
141+
if deregisterErr := s.serviceRegister.Deregister(); deregisterErr != nil {
142+
log.Warn("failed to deregister the service", errs.ZapError(deregisterErr))
143+
}
144+
}
145+
if err != nil && s.GetListener() != nil {
146+
_ = s.GetListener().Close()
147+
}
148+
}()
136149

137150
if s.serviceID, s.serviceRegister, err = utils.Register(s, constant.ResourceManagerServiceName); err != nil {
138151
return err
@@ -386,6 +399,23 @@ func (s *Server) GetTLSConfig() *grpcutil.TLSConfig {
386399
return &s.cfg.Security.TLSConfig
387400
}
388401

402+
func (s *Server) initListenerAndUpdateConfig() error {
403+
oldListenAddr := s.cfg.ListenAddr
404+
if err := s.InitListener(s.GetTLSConfig(), s.cfg.GetListenAddr()); err != nil {
405+
return err
406+
}
407+
actualListenAddr := s.GetActualListenAddr()
408+
if actualListenAddr == "" {
409+
return nil
410+
}
411+
s.cfg.ListenAddr = server.ResolveListenAddr(s.cfg.ListenAddr, actualListenAddr)
412+
s.cfg.AdvertiseListenAddr = server.ResolveAdvertiseListenAddr(s.cfg.AdvertiseListenAddr, actualListenAddr)
413+
if s.cfg.Name == "" || s.cfg.Name == oldListenAddr {
414+
s.cfg.Name = s.cfg.AdvertiseListenAddr
415+
}
416+
return nil
417+
}
418+
389419
// GetServingUrls gets service endpoints from the leader in election group.
390420
func (s *Server) GetServingUrls() []string {
391421
return s.participant.GetServingUrls()
@@ -416,9 +446,6 @@ func (s *Server) startServer() (err error) {
416446
rejectMetadataWritesViaGRPC: s.ShouldRejectMetadataWritesViaGRPC(),
417447
}
418448

419-
if err := s.InitListener(s.GetTLSConfig(), s.cfg.GetListenAddr()); err != nil {
420-
return err
421-
}
422449
// Only start the metering writer if a valid metering config is provided.
423450
if len(s.cfg.Metering.Type) > 0 {
424451
s.meteringWriter, err = metering.NewWriter(s.Context(), &s.cfg.Metering, fmt.Sprintf("pd%d", s.participant.ID()))

‎pkg/mcs/router/server/server.go‎

Lines changed: 30 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -120,6 +120,19 @@ func (s *Server) Run() (err error) {
120120
if err = utils.InitClient(s); err != nil {
121121
return err
122122
}
123+
if err = s.initListenerAndUpdateConfig(); err != nil {
124+
return err
125+
}
126+
defer func() {
127+
if err != nil && s.serviceRegister != nil {
128+
if deregisterErr := s.serviceRegister.Deregister(); deregisterErr != nil {
129+
log.Warn("failed to deregister the service", errs.ZapError(deregisterErr))
130+
}
131+
}
132+
if err != nil && s.GetListener() != nil {
133+
_ = s.GetListener().Close()
134+
}
135+
}()
123136

124137
if s.serviceID, s.serviceRegister, err = utils.Register(s, constant.RouterServiceName); err != nil {
125138
return err
@@ -181,6 +194,23 @@ func (s *Server) GetTLSConfig() *grpcutil.TLSConfig {
181194
return &s.cfg.Security.TLSConfig
182195
}
183196

197+
func (s *Server) initListenerAndUpdateConfig() error {
198+
oldListenAddr := s.cfg.ListenAddr
199+
if err := s.InitListener(s.GetTLSConfig(), s.cfg.GetListenAddr()); err != nil {
200+
return err
201+
}
202+
actualListenAddr := s.GetActualListenAddr()
203+
if actualListenAddr == "" {
204+
return nil
205+
}
206+
s.cfg.ListenAddr = server.ResolveListenAddr(s.cfg.ListenAddr, actualListenAddr)
207+
s.cfg.AdvertiseListenAddr = server.ResolveAdvertiseListenAddr(s.cfg.AdvertiseListenAddr, actualListenAddr)
208+
if s.cfg.Name == "" || s.cfg.Name == oldListenAddr {
209+
s.cfg.Name = s.cfg.AdvertiseListenAddr
210+
}
211+
return nil
212+
}
213+
184214
// GetCluster returns the cluster.
185215
func (s *Server) GetCluster() *Cluster {
186216
return s.cluster
@@ -235,9 +265,6 @@ func (s *Server) startServer() (err error) {
235265
bs.ServerMaxProcsGauge.Set(float64(runtime.GOMAXPROCS(0)))
236266

237267
s.service = &Service{Server: s}
238-
if err := s.InitListener(s.GetTLSConfig(), s.cfg.GetListenAddr()); err != nil {
239-
return err
240-
}
241268

242269
serverReadyChan := make(chan struct{})
243270
defer close(serverReadyChan)

‎pkg/mcs/scheduling/server/server.go‎

Lines changed: 30 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -154,6 +154,19 @@ func (s *Server) Run() (err error) {
154154
if err = utils.InitClient(s); err != nil {
155155
return err
156156
}
157+
if err = s.initListenerAndUpdateConfig(); err != nil {
158+
return err
159+
}
160+
defer func() {
161+
if err != nil && s.serviceRegister != nil {
162+
if deregisterErr := s.serviceRegister.Deregister(); deregisterErr != nil {
163+
log.Warn("failed to deregister the service", errs.ZapError(deregisterErr))
164+
}
165+
}
166+
if err != nil && s.GetListener() != nil {
167+
_ = s.GetListener().Close()
168+
}
169+
}()
157170

158171
if s.serviceID, s.serviceRegister, err = utils.Register(s, constant.SchedulingServiceName); err != nil {
159172
return err
@@ -406,6 +419,23 @@ func (s *Server) GetTLSConfig() *grpcutil.TLSConfig {
406419
return &s.cfg.Security.TLSConfig
407420
}
408421

422+
func (s *Server) initListenerAndUpdateConfig() error {
423+
oldListenAddr := s.cfg.ListenAddr
424+
if err := s.InitListener(s.GetTLSConfig(), s.cfg.GetListenAddr()); err != nil {
425+
return err
426+
}
427+
actualListenAddr := s.GetActualListenAddr()
428+
if actualListenAddr == "" {
429+
return nil
430+
}
431+
s.cfg.ListenAddr = server.ResolveListenAddr(s.cfg.ListenAddr, actualListenAddr)
432+
s.cfg.AdvertiseListenAddr = server.ResolveAdvertiseListenAddr(s.cfg.AdvertiseListenAddr, actualListenAddr)
433+
if s.cfg.Name == "" || s.cfg.Name == oldListenAddr {
434+
s.cfg.Name = s.cfg.AdvertiseListenAddr
435+
}
436+
return nil
437+
}
438+
409439
// GetCluster returns the cluster.
410440
func (s *Server) GetCluster() *Cluster {
411441
cluster := s.cluster.Load()
@@ -478,9 +508,6 @@ func (s *Server) startServer() (err error) {
478508
s.service = &Service{Server: s}
479509
s.AddServiceReadyCallback(s.startCluster)
480510
s.AddServiceExitCallback(s.stopCluster)
481-
if err := s.InitListener(s.GetTLSConfig(), s.cfg.GetListenAddr()); err != nil {
482-
return err
483-
}
484511

485512
serverReadyChan := make(chan struct{})
486513
defer close(serverReadyChan)

‎pkg/mcs/server/server.go‎

Lines changed: 56 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -45,6 +45,7 @@ type BaseServer struct {
4545
clientConns sync.Map
4646
secure bool
4747
muxListener net.Listener
48+
listenAddr string
4849
// Callback functions for different stages
4950
// startCallbacks will be called after the server is started.
5051
startCallbacks []func()
@@ -158,14 +159,41 @@ func (bs *BaseServer) InitListener(tlsCfg *grpcutil.TLSConfig, listenAddr string
158159
} else {
159160
bs.muxListener, err = net.Listen(constant.TCPNetworkStr, listenURL.Host)
160161
}
161-
return err
162+
if err != nil {
163+
return err
164+
}
165+
bs.listenAddr = buildActualListenAddr(listenURL, bs.muxListener.Addr())
166+
return nil
162167
}
163168

164169
// GetListener returns the listener.
165170
func (bs *BaseServer) GetListener() net.Listener {
166171
return bs.muxListener
167172
}
168173

174+
// GetActualListenAddr returns the listener address after binding.
175+
func (bs *BaseServer) GetActualListenAddr() string {
176+
return bs.listenAddr
177+
}
178+
179+
// ResolveListenAddr returns actualListenAddr only when listenAddr points at a
180+
// kernel-selected port.
181+
func ResolveListenAddr(listenAddr, actualListenAddr string) string {
182+
if hasZeroPort(listenAddr) {
183+
return actualListenAddr
184+
}
185+
return listenAddr
186+
}
187+
188+
// ResolveAdvertiseListenAddr returns actualListenAddr when advertiseAddr was left
189+
// unspecified or still points at a kernel-selected port.
190+
func ResolveAdvertiseListenAddr(advertiseAddr, actualListenAddr string) string {
191+
if advertiseAddr == "" || hasZeroPort(advertiseAddr) {
192+
return actualListenAddr
193+
}
194+
return advertiseAddr
195+
}
196+
169197
// IsSecure checks if the server enable TLS.
170198
func (bs *BaseServer) IsSecure() bool {
171199
return bs.secure
@@ -186,3 +214,30 @@ func (bs *BaseServer) CloseClientConns() {
186214
return true
187215
})
188216
}
217+
218+
func buildActualListenAddr(listenURL *url.URL, addr net.Addr) string {
219+
host, _, err := net.SplitHostPort(listenURL.Host)
220+
if err != nil || host == "" {
221+
host, _, _ = net.SplitHostPort(addr.String())
222+
}
223+
if ip := net.ParseIP(host); ip != nil && ip.IsUnspecified() {
224+
host = "127.0.0.1"
225+
}
226+
_, port, err := net.SplitHostPort(addr.String())
227+
if err != nil {
228+
return listenURL.String()
229+
}
230+
actualURL := *listenURL
231+
actualURL.Host = net.JoinHostPort(host, port)
232+
return actualURL.String()
233+
}
234+
235+
func hasZeroPort(addr string) bool {
236+
parsed, err := url.Parse(addr)
237+
host := addr
238+
if err == nil && parsed.Host != "" {
239+
host = parsed.Host
240+
}
241+
_, port, err := net.SplitHostPort(host)
242+
return err == nil && port == "0"
243+
}

‎pkg/mcs/server/server_test.go‎

Lines changed: 51 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,51 @@
1+
// Copyright 2026 TiKV Project Authors.
2+
//
3+
// Licensed under the Apache License, Version 2.0 (the "License");
4+
// you may not use this file except in compliance with the License.
5+
// You may obtain a copy of the License at
6+
//
7+
// http://www.apache.org/licenses/LICENSE-2.0
8+
//
9+
// Unless required by applicable law or agreed to in writing, software
10+
// distributed under the License is distributed on an "AS IS" BASIS,
11+
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
12+
// See the License for the specific language governing permissions and
13+
// limitations under the License.
14+
15+
package server
16+
17+
import (
18+
"context"
19+
"net/url"
20+
"testing"
21+
22+
"github.com/stretchr/testify/require"
23+
24+
"github.com/tikv/pd/pkg/utils/grpcutil"
25+
)
26+
27+
func TestInitListenerWithKernelSelectedPort(t *testing.T) {
28+
re := require.New(t)
29+
svr := NewBaseServer(context.Background())
30+
re.NoError(svr.InitListener(&grpcutil.TLSConfig{}, "http://127.0.0.1:0"))
31+
defer svr.GetListener().Close()
32+
33+
actual := svr.GetActualListenAddr()
34+
u, err := url.Parse(actual)
35+
re.NoError(err)
36+
re.Equal("http", u.Scheme)
37+
re.Equal("127.0.0.1", u.Hostname())
38+
re.NotEmpty(u.Port())
39+
re.NotEqual("0", u.Port())
40+
}
41+
42+
func TestResolveAdvertiseListenAddr(t *testing.T) {
43+
actualAddr := "http://127.0.0.1:12345"
44+
require.Equal(t, actualAddr, ResolveListenAddr("http://127.0.0.1:0", actualAddr))
45+
require.Equal(t, "http://127.0.0.1:23456", ResolveListenAddr("http://127.0.0.1:23456", actualAddr))
46+
47+
require.Equal(t, actualAddr, ResolveAdvertiseListenAddr("", actualAddr))
48+
require.Equal(t, actualAddr, ResolveAdvertiseListenAddr("http://127.0.0.1:0", actualAddr))
49+
require.Equal(t, actualAddr, ResolveAdvertiseListenAddr("127.0.0.1:0", actualAddr))
50+
require.Equal(t, "http://127.0.0.1:23456", ResolveAdvertiseListenAddr("http://127.0.0.1:23456", actualAddr))
51+
}

‎pkg/mcs/tso/server/server.go‎

Lines changed: 49 additions & 20 deletions
Original file line numberDiff line numberDiff line change
@@ -165,6 +165,19 @@ func (s *Server) Run() (err error) {
165165
if err = utils.InitClient(s); err != nil {
166166
return err
167167
}
168+
if err = s.initListenerAndUpdateConfig(); err != nil {
169+
return err
170+
}
171+
defer func() {
172+
if err != nil && s.serviceRegister != nil {
173+
if deregisterErr := s.serviceRegister.Deregister(); deregisterErr != nil {
174+
log.Warn("failed to deregister the service", errs.ZapError(deregisterErr))
175+
}
176+
}
177+
if err != nil && s.GetListener() != nil {
178+
_ = s.GetListener().Close()
179+
}
180+
}()
168181

169182
if s.serviceID, s.serviceRegister, err = utils.Register(s, mcs.TSOServiceName); err != nil {
170183
return err
@@ -347,6 +360,24 @@ func (s *Server) GetTLSConfig() *grpcutil.TLSConfig {
347360
return &s.cfg.Security.TLSConfig
348361
}
349362

363+
func (s *Server) initListenerAndUpdateConfig() error {
364+
oldListenAddr := s.cfg.ListenAddr
365+
if err := s.InitListener(s.GetTLSConfig(), s.cfg.ListenAddr); err != nil {
366+
return err
367+
}
368+
actualListenAddr := s.GetActualListenAddr()
369+
if actualListenAddr == "" {
370+
return nil
371+
}
372+
s.cfg.ListenAddr = server.ResolveListenAddr(s.cfg.ListenAddr, actualListenAddr)
373+
s.cfg.AdvertiseListenAddr = server.ResolveAdvertiseListenAddr(s.cfg.AdvertiseListenAddr, actualListenAddr)
374+
if s.cfg.Name == "" || s.cfg.Name == oldListenAddr {
375+
s.cfg.Name = s.cfg.AdvertiseListenAddr
376+
}
377+
s.advertiseListenHost = parseAdvertiseListenHost(s.cfg.AdvertiseListenAddr)
378+
return nil
379+
}
380+
350381
func (s *Server) startServer() (err error) {
351382
clusterID := keypath.ClusterID()
352383
// It may lose accuracy if use float64 to store uint64. So we store the cluster id in label.
@@ -368,10 +399,6 @@ func (s *Server) startServer() (err error) {
368399
s.tsoProtoFactory = &tsoutil.TSOProtoFactory{}
369400
s.service = &Service{Server: s}
370401

371-
if err := s.InitListener(s.GetTLSConfig(), s.cfg.ListenAddr); err != nil {
372-
return err
373-
}
374-
375402
serverReadyChan := make(chan struct{})
376403
defer close(serverReadyChan)
377404
s.serverLoopWg.Add(1)
@@ -390,22 +417,7 @@ func (s *Server) startServer() (err error) {
390417

391418
// CreateServer creates the Server
392419
func CreateServer(ctx context.Context, cfg *Config) *Server {
393-
addr := cfg.GetAdvertiseListenAddr()
394-
parsed, err := url.Parse(addr)
395-
advertiseListenHost := ""
396-
if err != nil {
397-
if _, _, splitErr := net.SplitHostPort(addr); splitErr != nil {
398-
panic(fmt.Sprintf("invalid advertise listen address: %s", addr))
399-
}
400-
advertiseListenHost = addr
401-
} else {
402-
advertiseListenHost = parsed.Host
403-
if advertiseListenHost == "" {
404-
if _, _, splitErr := net.SplitHostPort(addr); splitErr == nil {
405-
advertiseListenHost = addr
406-
}
407-
}
408-
}
420+
advertiseListenHost := parseAdvertiseListenHost(cfg.GetAdvertiseListenAddr())
409421
svr := &Server{
410422
BaseServer: server.NewBaseServer(ctx),
411423
DiagnosticsServer: sysutil.NewDiagnosticsServer(cfg.Log.File.Filename),
@@ -415,6 +427,23 @@ func CreateServer(ctx context.Context, cfg *Config) *Server {
415427
return svr
416428
}
417429

430+
func parseAdvertiseListenHost(addr string) string {
431+
parsed, err := url.Parse(addr)
432+
if err != nil {
433+
if _, _, splitErr := net.SplitHostPort(addr); splitErr != nil {
434+
panic(fmt.Sprintf("invalid advertise listen address: %s", addr))
435+
}
436+
return addr
437+
}
438+
advertiseListenHost := parsed.Host
439+
if advertiseListenHost == "" {
440+
if _, _, splitErr := net.SplitHostPort(addr); splitErr == nil {
441+
advertiseListenHost = addr
442+
}
443+
}
444+
return advertiseListenHost
445+
}
446+
418447
// CreateServerWrapper encapsulates the configuration/log/metrics initialization and create the server
419448
func CreateServerWrapper(cmd *cobra.Command, args []string) {
420449
err := cmd.Flags().Parse(args)

0 commit comments

Comments
 (0)