mirror of
https://github.com/etcd-io/etcd.git
synced 2024-09-27 06:25:44 +00:00
Remove PauseAccept of the reverse proxy from the e2e test
Part of the patches to fix https://github.com/etcd-io/etcd/issues/17737 During the development of https://github.com/etcd-io/etcd/pull/17938, we agreed that during the transition to L7 forward proxy, unused features and features targeting L4 reverse proxy will be dropped. This feature falls under the unused feature. Signed-off-by: Chun-Hung Tseng <henrybear327@gmail.com>
This commit is contained in:
parent
1a08fb2809
commit
3279fb522e
@ -59,11 +59,6 @@ type Server interface {
|
|||||||
// Close closes listener and transport.
|
// Close closes listener and transport.
|
||||||
Close() error
|
Close() error
|
||||||
|
|
||||||
// PauseAccept stops accepting new connections.
|
|
||||||
PauseAccept()
|
|
||||||
// UnpauseAccept removes pause operation on accepting new connections.
|
|
||||||
UnpauseAccept()
|
|
||||||
|
|
||||||
// DelayAccept adds latency ± random variable to accepting
|
// DelayAccept adds latency ± random variable to accepting
|
||||||
// new incoming connections.
|
// new incoming connections.
|
||||||
DelayAccept(latency, rv time.Duration)
|
DelayAccept(latency, rv time.Duration)
|
||||||
@ -115,16 +110,6 @@ type Server interface {
|
|||||||
// UnblackholeRx removes blackhole operation on "receiving".
|
// UnblackholeRx removes blackhole operation on "receiving".
|
||||||
UnblackholeRx()
|
UnblackholeRx()
|
||||||
|
|
||||||
// PauseTx stops "forwarding" packets; "outgoing" traffic blocks.
|
|
||||||
PauseTx()
|
|
||||||
// UnpauseTx removes "forwarding" pause operation.
|
|
||||||
UnpauseTx()
|
|
||||||
|
|
||||||
// PauseRx stops "receiving" packets; "incoming" traffic blocks.
|
|
||||||
PauseRx()
|
|
||||||
// UnpauseRx removes "receiving" pause operation.
|
|
||||||
UnpauseRx()
|
|
||||||
|
|
||||||
// ResetListener closes and restarts listener.
|
// ResetListener closes and restarts listener.
|
||||||
ResetListener() error
|
ResetListener() error
|
||||||
}
|
}
|
||||||
@ -164,9 +149,6 @@ type server struct {
|
|||||||
listenerMu sync.RWMutex
|
listenerMu sync.RWMutex
|
||||||
listener net.Listener
|
listener net.Listener
|
||||||
|
|
||||||
pauseAcceptMu sync.Mutex
|
|
||||||
pauseAcceptc chan struct{}
|
|
||||||
|
|
||||||
latencyAcceptMu sync.RWMutex
|
latencyAcceptMu sync.RWMutex
|
||||||
latencyAccept time.Duration
|
latencyAccept time.Duration
|
||||||
|
|
||||||
@ -208,9 +190,8 @@ func NewServer(cfg ServerConfig) Server {
|
|||||||
donec: make(chan struct{}),
|
donec: make(chan struct{}),
|
||||||
errc: make(chan error, 16),
|
errc: make(chan error, 16),
|
||||||
|
|
||||||
pauseAcceptc: make(chan struct{}),
|
pauseTxc: make(chan struct{}),
|
||||||
pauseTxc: make(chan struct{}),
|
pauseRxc: make(chan struct{}),
|
||||||
pauseRxc: make(chan struct{}),
|
|
||||||
}
|
}
|
||||||
|
|
||||||
_, fromPort, err := net.SplitHostPort(cfg.From.Host)
|
_, fromPort, err := net.SplitHostPort(cfg.From.Host)
|
||||||
@ -233,7 +214,6 @@ func NewServer(cfg ServerConfig) Server {
|
|||||||
s.retryInterval = defaultRetryInterval
|
s.retryInterval = defaultRetryInterval
|
||||||
}
|
}
|
||||||
|
|
||||||
close(s.pauseAcceptc)
|
|
||||||
close(s.pauseTxc)
|
close(s.pauseTxc)
|
||||||
close(s.pauseRxc)
|
close(s.pauseRxc)
|
||||||
|
|
||||||
@ -290,15 +270,6 @@ func (s *server) listenAndServe() {
|
|||||||
close(s.readyc)
|
close(s.readyc)
|
||||||
|
|
||||||
for {
|
for {
|
||||||
s.pauseAcceptMu.Lock()
|
|
||||||
pausec := s.pauseAcceptc
|
|
||||||
s.pauseAcceptMu.Unlock()
|
|
||||||
select {
|
|
||||||
case <-pausec:
|
|
||||||
case <-s.donec:
|
|
||||||
return
|
|
||||||
}
|
|
||||||
|
|
||||||
s.latencyAcceptMu.RLock()
|
s.latencyAcceptMu.RLock()
|
||||||
lat := s.latencyAccept
|
lat := s.latencyAccept
|
||||||
s.latencyAcceptMu.RUnlock()
|
s.latencyAcceptMu.RUnlock()
|
||||||
@ -645,37 +616,6 @@ func (s *server) Close() (err error) {
|
|||||||
return err
|
return err
|
||||||
}
|
}
|
||||||
|
|
||||||
func (s *server) PauseAccept() {
|
|
||||||
s.pauseAcceptMu.Lock()
|
|
||||||
s.pauseAcceptc = make(chan struct{})
|
|
||||||
s.pauseAcceptMu.Unlock()
|
|
||||||
|
|
||||||
s.lg.Info(
|
|
||||||
"paused accept",
|
|
||||||
zap.String("from", s.From()),
|
|
||||||
zap.String("to", s.To()),
|
|
||||||
)
|
|
||||||
}
|
|
||||||
|
|
||||||
func (s *server) UnpauseAccept() {
|
|
||||||
s.pauseAcceptMu.Lock()
|
|
||||||
select {
|
|
||||||
case <-s.pauseAcceptc: // already unpaused
|
|
||||||
case <-s.donec:
|
|
||||||
s.pauseAcceptMu.Unlock()
|
|
||||||
return
|
|
||||||
default:
|
|
||||||
close(s.pauseAcceptc)
|
|
||||||
}
|
|
||||||
s.pauseAcceptMu.Unlock()
|
|
||||||
|
|
||||||
s.lg.Info(
|
|
||||||
"unpaused accept",
|
|
||||||
zap.String("from", s.From()),
|
|
||||||
zap.String("to", s.To()),
|
|
||||||
)
|
|
||||||
}
|
|
||||||
|
|
||||||
func (s *server) DelayAccept(latency, rv time.Duration) {
|
func (s *server) DelayAccept(latency, rv time.Duration) {
|
||||||
if latency <= 0 {
|
if latency <= 0 {
|
||||||
return
|
return
|
||||||
|
|||||||
@ -234,55 +234,6 @@ func testServerDelayAccept(t *testing.T, secure bool) {
|
|||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
func TestServer_PauseTx(t *testing.T) {
|
|
||||||
lg := zaptest.NewLogger(t)
|
|
||||||
scheme := "unix"
|
|
||||||
srcAddr, dstAddr := newUnixAddr(), newUnixAddr()
|
|
||||||
defer func() {
|
|
||||||
os.RemoveAll(srcAddr)
|
|
||||||
os.RemoveAll(dstAddr)
|
|
||||||
}()
|
|
||||||
ln := listen(t, scheme, dstAddr, transport.TLSInfo{})
|
|
||||||
defer ln.Close()
|
|
||||||
|
|
||||||
p := NewServer(ServerConfig{
|
|
||||||
Logger: lg,
|
|
||||||
From: url.URL{Scheme: scheme, Host: srcAddr},
|
|
||||||
To: url.URL{Scheme: scheme, Host: dstAddr},
|
|
||||||
})
|
|
||||||
|
|
||||||
waitForServer(t, p)
|
|
||||||
|
|
||||||
defer p.Close()
|
|
||||||
|
|
||||||
p.PauseTx()
|
|
||||||
|
|
||||||
data := []byte("Hello World!")
|
|
||||||
send(t, data, scheme, srcAddr, transport.TLSInfo{})
|
|
||||||
|
|
||||||
recvc := make(chan []byte, 1)
|
|
||||||
go func() {
|
|
||||||
recvc <- receive(t, ln)
|
|
||||||
}()
|
|
||||||
|
|
||||||
select {
|
|
||||||
case d := <-recvc:
|
|
||||||
t.Fatalf("received unexpected data %q during pause", string(d))
|
|
||||||
case <-time.After(200 * time.Millisecond):
|
|
||||||
}
|
|
||||||
|
|
||||||
p.UnpauseTx()
|
|
||||||
|
|
||||||
select {
|
|
||||||
case d := <-recvc:
|
|
||||||
if !bytes.Equal(data, d) {
|
|
||||||
t.Fatalf("expected %q, got %q", string(data), string(d))
|
|
||||||
}
|
|
||||||
case <-time.After(2 * time.Second):
|
|
||||||
t.Fatal("took too long to receive after unpause")
|
|
||||||
}
|
|
||||||
}
|
|
||||||
|
|
||||||
func TestServer_ModifyTx_corrupt(t *testing.T) {
|
func TestServer_ModifyTx_corrupt(t *testing.T) {
|
||||||
lg := zaptest.NewLogger(t)
|
lg := zaptest.NewLogger(t)
|
||||||
scheme := "unix"
|
scheme := "unix"
|
||||||
|
|||||||
Loading…
x
Reference in New Issue
Block a user