Skip to content

Commit 2840a56

Browse files
henrybear327henrybear327
committed
Remove ModifyTx and ModifyRx of the reverse proxy from the e2e test
Part of the patches to fix #17737 During the development of #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 features targeting L4 reverse proxy. Signed-off-by: Chun-Hung Tseng <[email protected]>
1 parent 42b390f commit 2840a56

File tree

3 files changed

+52
-143
lines changed

3 files changed

+52
-143
lines changed

‎pkg/proxy/server.go

+42-57
Original file line numberDiff line numberDiff line change
@@ -75,28 +75,12 @@ type Server interface {
7575
// LatencyRx returns current receive latency.
7676
LatencyRx() time.Duration
7777

78-
// ModifyTx alters/corrupts/drops "outgoing" packets from the listener
79-
// with the given edit function.
80-
ModifyTx(f func(data []byte) []byte)
81-
// UnmodifyTx removes modify operation on "forwarding".
82-
UnmodifyTx()
83-
84-
// ModifyRx alters/corrupts/drops "incoming" packets to client
85-
// with the given edit function.
86-
ModifyRx(f func(data []byte) []byte)
87-
// UnmodifyRx removes modify operation on "receiving".
88-
UnmodifyRx()
89-
9078
// BlackholeTx drops all "outgoing" packets before "forwarding".
91-
// "BlackholeTx" operation is a wrapper around "ModifyTx" with
92-
// a function that returns empty bytes.
9379
BlackholeTx()
9480
// UnblackholeTx removes blackhole operation on "sending".
9581
UnblackholeTx()
9682

9783
// BlackholeRx drops all "incoming" packets to client.
98-
// "BlackholeRx" operation is a wrapper around "ModifyRx" with
99-
// a function that returns empty bytes.
10084
BlackholeRx()
10185
// UnblackholeRx removes blackhole operation on "receiving".
10286
UnblackholeRx()
@@ -140,11 +124,11 @@ type server struct {
140124
listenerMu sync.RWMutex
141125
listener net.Listener
142126

143-
modifyTxMu sync.RWMutex
144-
modifyTx func(data []byte) []byte
127+
shouldDropTxMu sync.RWMutex
128+
shouldDropTx bool
145129

146-
modifyRxMu sync.RWMutex
147-
modifyRx func(data []byte) []byte
130+
shouldDropRxMu sync.RWMutex
131+
shouldDropRx bool
148132

149133
pauseTxMu sync.Mutex
150134
pauseTxc chan struct{}
@@ -404,20 +388,20 @@ func (s *server) ioCopy(dst io.Writer, src io.Reader, ptype proxyType) {
404388
}
405389
data := buf[:nr1]
406390

407-
// alters/corrupts/drops data
391+
// drops data
408392
switch ptype {
409393
case proxyTx:
410-
s.modifyTxMu.RLock()
411-
if s.modifyTx != nil {
412-
data = s.modifyTx(data)
394+
s.shouldDropTxMu.RLock()
395+
if s.shouldDropTx {
396+
data = nil
413397
}
414-
s.modifyTxMu.RUnlock()
398+
s.shouldDropTxMu.RUnlock()
415399
case proxyRx:
416-
s.modifyRxMu.RLock()
417-
if s.modifyRx != nil {
418-
data = s.modifyRx(data)
400+
s.shouldDropRxMu.RLock()
401+
if s.shouldDropRx {
402+
data = nil
419403
}
420-
s.modifyRxMu.RUnlock()
404+
s.shouldDropRxMu.RUnlock()
421405
default:
422406
panic("unknown proxy type")
423407
}
@@ -691,55 +675,56 @@ func computeLatency(lat, rv time.Duration) time.Duration {
691675
return lat + time.Duration(int64(sign)*mrand.Int63n(rv.Nanoseconds()))
692676
}
693677

694-
func (s *server) ModifyTx(f func([]byte) []byte) {
695-
s.modifyTxMu.Lock()
696-
s.modifyTx = f
697-
s.modifyTxMu.Unlock()
678+
func (s *server) setShouldDropTx() {
679+
s.shouldDropTxMu.Lock()
680+
s.shouldDropTx = true
681+
s.shouldDropTxMu.Unlock()
698682

699683
s.lg.Info(
700-
"modifying tx",
684+
"setShouldDropTx",
701685
zap.String("from", s.From()),
702686
zap.String("to", s.To()),
703687
)
704688
}
705689

706-
func (s *server) UnmodifyTx() {
707-
s.modifyTxMu.Lock()
708-
s.modifyTx = nil
709-
s.modifyTxMu.Unlock()
690+
func (s *server) unsetShouldDropTx() {
691+
s.shouldDropTxMu.Lock()
692+
s.shouldDropTx = false
693+
s.shouldDropTxMu.Unlock()
710694

711695
s.lg.Info(
712-
"unmodifyed tx",
696+
"unsetShouldDropTx",
713697
zap.String("from", s.From()),
714698
zap.String("to", s.To()),
715699
)
716700
}
717701

718-
func (s *server) ModifyRx(f func([]byte) []byte) {
719-
s.modifyRxMu.Lock()
720-
s.modifyRx = f
721-
s.modifyRxMu.Unlock()
702+
func (s *server) setShouldDropRx() {
703+
s.shouldDropRxMu.Lock()
704+
s.shouldDropRx = true
705+
s.shouldDropRxMu.Unlock()
706+
722707
s.lg.Info(
723-
"modifying rx",
724-
zap.String("from", s.To()),
725-
zap.String("to", s.From()),
708+
"setShouldDropRx",
709+
zap.String("from", s.From()),
710+
zap.String("to", s.To()),
726711
)
727712
}
728713

729-
func (s *server) UnmodifyRx() {
730-
s.modifyRxMu.Lock()
731-
s.modifyRx = nil
732-
s.modifyRxMu.Unlock()
714+
func (s *server) unsetShouldDropRx() {
715+
s.shouldDropRxMu.Lock()
716+
s.shouldDropRx = false
717+
s.shouldDropRxMu.Unlock()
733718

734719
s.lg.Info(
735-
"unmodifyed rx",
736-
zap.String("from", s.To()),
737-
zap.String("to", s.From()),
720+
"unsetShouldDropRx",
721+
zap.String("from", s.From()),
722+
zap.String("to", s.To()),
738723
)
739724
}
740725

741726
func (s *server) BlackholeTx() {
742-
s.ModifyTx(func([]byte) []byte { return nil })
727+
s.setShouldDropTx()
743728
s.lg.Info(
744729
"blackholed tx",
745730
zap.String("from", s.From()),
@@ -748,7 +733,7 @@ func (s *server) BlackholeTx() {
748733
}
749734

750735
func (s *server) UnblackholeTx() {
751-
s.UnmodifyTx()
736+
s.unsetShouldDropTx()
752737
s.lg.Info(
753738
"unblackholed tx",
754739
zap.String("from", s.From()),
@@ -757,7 +742,7 @@ func (s *server) UnblackholeTx() {
757742
}
758743

759744
func (s *server) BlackholeRx() {
760-
s.ModifyRx(func([]byte) []byte { return nil })
745+
s.setShouldDropRx()
761746
s.lg.Info(
762747
"blackholed rx",
763748
zap.String("from", s.To()),
@@ -766,7 +751,7 @@ func (s *server) BlackholeRx() {
766751
}
767752

768753
func (s *server) UnblackholeRx() {
769-
s.UnmodifyRx()
754+
s.unsetShouldDropRx()
770755
s.lg.Info(
771756
"unblackholed rx",
772757
zap.String("from", s.To()),

‎pkg/proxy/server_test.go

-77
Original file line numberDiff line numberDiff line change
@@ -175,83 +175,6 @@ func createTLSInfo(lg *zap.Logger, secure bool) transport.TLSInfo {
175175
return transport.TLSInfo{Logger: lg}
176176
}
177177

178-
func TestServer_ModifyTx_corrupt(t *testing.T) {
179-
lg := zaptest.NewLogger(t)
180-
scheme := "unix"
181-
srcAddr, dstAddr := newUnixAddr(), newUnixAddr()
182-
defer func() {
183-
os.RemoveAll(srcAddr)
184-
os.RemoveAll(dstAddr)
185-
}()
186-
ln := listen(t, scheme, dstAddr, transport.TLSInfo{})
187-
defer ln.Close()
188-
189-
p := NewServer(ServerConfig{
190-
Logger: lg,
191-
From: url.URL{Scheme: scheme, Host: srcAddr},
192-
To: url.URL{Scheme: scheme, Host: dstAddr},
193-
})
194-
195-
waitForServer(t, p)
196-
197-
defer p.Close()
198-
199-
p.ModifyTx(func(d []byte) []byte {
200-
d[len(d)/2]++
201-
return d
202-
})
203-
data := []byte("Hello World!")
204-
send(t, data, scheme, srcAddr, transport.TLSInfo{})
205-
if d := receive(t, ln); bytes.Equal(d, data) {
206-
t.Fatalf("expected corrupted data, got %q", string(d))
207-
}
208-
209-
p.UnmodifyTx()
210-
send(t, data, scheme, srcAddr, transport.TLSInfo{})
211-
if d := receive(t, ln); !bytes.Equal(d, data) {
212-
t.Fatalf("expected uncorrupted data, got %q", string(d))
213-
}
214-
}
215-
216-
func TestServer_ModifyTx_packet_loss(t *testing.T) {
217-
lg := zaptest.NewLogger(t)
218-
scheme := "unix"
219-
srcAddr, dstAddr := newUnixAddr(), newUnixAddr()
220-
defer func() {
221-
os.RemoveAll(srcAddr)
222-
os.RemoveAll(dstAddr)
223-
}()
224-
ln := listen(t, scheme, dstAddr, transport.TLSInfo{})
225-
defer ln.Close()
226-
227-
p := NewServer(ServerConfig{
228-
Logger: lg,
229-
From: url.URL{Scheme: scheme, Host: srcAddr},
230-
To: url.URL{Scheme: scheme, Host: dstAddr},
231-
})
232-
233-
waitForServer(t, p)
234-
235-
defer p.Close()
236-
237-
// 50% packet loss
238-
p.ModifyTx(func(d []byte) []byte {
239-
half := len(d) / 2
240-
return d[:half:half]
241-
})
242-
data := []byte("Hello World!")
243-
send(t, data, scheme, srcAddr, transport.TLSInfo{})
244-
if d := receive(t, ln); bytes.Equal(d, data) {
245-
t.Fatalf("expected corrupted data, got %q", string(d))
246-
}
247-
248-
p.UnmodifyTx()
249-
send(t, data, scheme, srcAddr, transport.TLSInfo{})
250-
if d := receive(t, ln); !bytes.Equal(d, data) {
251-
t.Fatalf("expected uncorrupted data, got %q", string(d))
252-
}
253-
}
254-
255178
func TestServer_BlackholeTx(t *testing.T) {
256179
lg := zaptest.NewLogger(t)
257180
scheme := "unix"

‎tests/robustness/failpoint/network.go

+10-9
Original file line numberDiff line numberDiff line change
@@ -193,21 +193,22 @@ func (f dropPeerNetworkFailpoint) Inject(ctx context.Context, t *testing.T, lg *
193193
member := clus.Procs[rand.Int()%len(clus.Procs)]
194194
proxy := member.PeerProxy()
195195

196-
proxy.ModifyRx(f.modifyPacket)
197-
proxy.ModifyTx(f.modifyPacket)
196+
if !f.shouldDropPacket() {
197+
return nil, nil
198+
}
199+
200+
proxy.BlackholeRx()
201+
proxy.BlackholeTx()
198202
lg.Info("Dropping traffic from and to member", zap.String("member", member.Config().Name), zap.Int("probability", f.dropProbabilityPercent))
199203
time.Sleep(f.duration)
200204
lg.Info("Traffic drop removed", zap.String("member", member.Config().Name))
201-
proxy.UnmodifyRx()
202-
proxy.UnmodifyTx()
205+
proxy.UnblackholeRx()
206+
proxy.UnblackholeTx()
203207
return nil, nil
204208
}
205209

206-
func (f dropPeerNetworkFailpoint) modifyPacket(data []byte) []byte {
207-
if rand.Intn(100) < f.dropProbabilityPercent {
208-
return nil
209-
}
210-
return data
210+
func (f dropPeerNetworkFailpoint) shouldDropPacket() bool {
211+
return rand.Intn(100) < f.dropProbabilityPercent
211212
}
212213

213214
func (f dropPeerNetworkFailpoint) Name() string {

0 commit comments

Comments
 (0)