Skip to content

Commit 1b0e033

Browse files
meethighermeethigher
committed
feat: 调整muxclient与muxserver的日志，更容易排查问题
1 parent 3aa2f74 commit 1b0e033

File tree

6 files changed

+89
-45
lines changed

6 files changed

+89
-45
lines changed

‎src/main/java/top/meethigher/proxy/tcp/mux/Mux.java‎

Lines changed: 5 additions & 5 deletions
Original file line numberDiff line numberDiff line change
@@ -4,7 +4,7 @@
44
import io.vertx.core.buffer.Buffer;
55
import org.slf4j.Logger;
66
import org.slf4j.LoggerFactory;
7-
import top.meethigher.proxy.NetAddress;
7+
import top.meethigher.proxy.tcp.mux.model.MuxConfiguration;
88
import top.meethigher.proxy.tcp.tunnel.codec.TunnelMessageCodec;
99

1010
import javax.crypto.SecretKey;
@@ -47,8 +47,8 @@ public Mux(Vertx vertx, String secret) {
4747
/**
4848
* 将host与port通过英文冒号连接，返回加密base64串(无换行)
4949
*/
50-
public Buffer aesBase64Encode(NetAddress netAddress) {
51-
String addr = netAddress.toString();
50+
public Buffer aesBase64Encode(MuxConfiguration configuration) {
51+
String addr = configuration.toString();
5252
SecretKey key = restoreKey(secret.getBytes(StandardCharsets.UTF_8));
5353
String encryptedAddr = encryptToBase64(addr.getBytes(StandardCharsets.UTF_8), key);
5454
return TunnelMessageCodec.encode(type, encryptedAddr.getBytes(StandardCharsets.UTF_8));
@@ -57,13 +57,13 @@ public Buffer aesBase64Encode(NetAddress netAddress) {
5757
/**
5858
* 将加密内容还原
5959
*/
60-
public NetAddress aesBase64Decode(Buffer buffer) {
60+
public MuxConfiguration aesBase64Decode(Buffer buffer) {
6161
TunnelMessageCodec.DecodedMessage decode = TunnelMessageCodec.decode(buffer);
6262
String encryptedAddr = new String(decode.body, StandardCharsets.UTF_8);
6363
SecretKey key = restoreKey(secret.getBytes(StandardCharsets.UTF_8));
6464
String addr = new String(decryptFromBase64(encryptedAddr, key),
6565
StandardCharsets.UTF_8);
66-
return NetAddress.parse(addr);
66+
return MuxConfiguration.parse(addr);
6767
}
6868

6969

‎src/main/java/top/meethigher/proxy/tcp/mux/ReverseTcpProxyMuxClient.java‎

Lines changed: 21 additions & 17 deletions
Original file line numberDiff line numberDiff line change
@@ -9,7 +9,9 @@
99
import org.slf4j.Logger;
1010
import org.slf4j.LoggerFactory;
1111
import top.meethigher.proxy.NetAddress;
12+
import top.meethigher.proxy.tcp.mux.model.MuxConfiguration;
1213
import top.meethigher.proxy.tcp.mux.model.MuxNetAddress;
14+
import top.meethigher.proxy.tcp.tunnel.utils.IdGenerator;
1315

1416
import java.util.ArrayList;
1517
import java.util.List;
@@ -57,41 +59,43 @@ public ReverseTcpProxyMuxClient(Vertx vertx, String secret, Map<MuxNetAddress, N
5759

5860
protected void handleConnect(NetSocket src, MuxNetAddress localServer, NetAddress backendServer) {
5961
src.pause();
60-
log.debug("{}: source {} -- {} connected", localServer.getName(), src.localAddress(), src.remoteAddress());
61-
src.exceptionHandler(e -> log.error("{}: source {} -- {} exception occurred", localServer.getName(), src.localAddress(), src.remoteAddress(), e))
62-
.closeHandler(v -> log.debug("{}: source {} -- {} closed", localServer.getName(), src.localAddress(), src.remoteAddress()));
62+
final int sessionId = IdGenerator.nextId();
63+
log.debug("{}: sessionId {}, source {} -- {} connected", localServer.getName(), sessionId, src.localAddress(), src.remoteAddress());
64+
src.exceptionHandler(e -> log.error("{}: sessionId {}, source {} -- {} exception occurred", localServer.getName(), sessionId, src.localAddress(), src.remoteAddress(), e))
65+
.closeHandler(v -> log.debug("{}: sessionId {}, source {} -- {} closed", localServer.getName(), sessionId, src.localAddress(), src.remoteAddress()));
6366
netClient.connect(muxServerAddress.getPort(), muxServerAddress.getHost())
6467
.onFailure(e -> {
65-
log.error("{}: failed to connect to {}", localServer.getName(), muxServerAddress, e);
68+
log.error("{}: sessionId {}, failed to connect to {}", localServer.getName(), sessionId, muxServerAddress, e);
6669
src.close();
6770
})
6871
.onSuccess(dst -> {
6972
dst.pause();
70-
log.debug("{}: target {} -- {} connected", localServer.getName(), dst.localAddress(), dst.remoteAddress());
71-
dst.exceptionHandler(e -> log.error("{}: target {} -- {} exception occurred", localServer.getName(), dst.localAddress(), dst.remoteAddress(), e))
72-
.closeHandler(v -> log.debug("{}: target {} -- {} closed", localServer.getName(), dst.localAddress(), dst.remoteAddress()));
73+
log.debug("{}: sessionId {}, target {} -- {} connected", localServer.getName(), sessionId, dst.localAddress(), dst.remoteAddress());
74+
dst.exceptionHandler(e -> log.error("{}: sessionId {}, target {} -- {} exception occurred", localServer.getName(), sessionId, dst.localAddress(), dst.remoteAddress(), e))
75+
.closeHandler(v -> log.debug("{}: sessionId {}, target {} -- {} closed", localServer.getName(), sessionId, dst.localAddress(), dst.remoteAddress()));
7376
Handler<Void> writeSuccessHandler = t -> {
7477
// https://github.com/meethigher/tcp-reverse-proxy/issues/12
7578
// 将日志记录详细，便于排查问题
7679
src.pipeTo(dst)
77-
.onSuccess(v -> log.debug("{}: source {} -- {} pipe to target {} -- {} succeeded",
78-
localServer.getName(), src.localAddress(), src.remoteAddress(), dst.localAddress(), dst.remoteAddress()))
79-
.onFailure(e -> log.error("{}: source {} -- {} pipe to target {} -- {} failed",
80-
localServer.getName(), src.localAddress(), src.remoteAddress(), dst.localAddress(), dst.remoteAddress(), e));
80+
.onSuccess(v -> log.debug("{}: sessionId {}, source {} -- {} pipe to target {} -- {} succeeded",
81+
localServer.getName(), sessionId, src.localAddress(), src.remoteAddress(), dst.localAddress(), dst.remoteAddress()))
82+
.onFailure(e -> log.error("{}: sessionId {}, source {} -- {} pipe to target {} -- {} failed",
83+
localServer.getName(), sessionId, src.localAddress(), src.remoteAddress(), dst.localAddress(), dst.remoteAddress(), e));
8184
dst.pipeTo(src)
82-
.onSuccess(v -> log.debug("{}: target {} -- {} pipe to source {} -- {} succeeded",
83-
localServer.getName(), dst.localAddress(), dst.remoteAddress(), src.localAddress(), src.remoteAddress()))
84-
.onFailure(e -> log.error("{}: target {} -- {} pipe to source {} -- {} failed",
85-
localServer.getName(), dst.localAddress(), dst.remoteAddress(), src.localAddress(), src.remoteAddress(), e));
86-
log.debug("{}: source {} -- {} bound to target {} -- {} with backend server {}",
85+
.onSuccess(v -> log.debug("{}: sessionId {}, target {} -- {} pipe to source {} -- {} succeeded",
86+
localServer.getName(), sessionId, dst.localAddress(), dst.remoteAddress(), src.localAddress(), src.remoteAddress()))
87+
.onFailure(e -> log.error("{}: sessionId {}, target {} -- {} pipe to source {} -- {} failed",
88+
localServer.getName(), sessionId, dst.localAddress(), dst.remoteAddress(), src.localAddress(), src.remoteAddress(), e));
89+
log.debug("{}: sessionId {}, source {} -- {} bound to target {} -- {} with backend server {}",
8790
localServer.getName(),
91+
sessionId,
8892
src.localAddress(), src.remoteAddress(),
8993
dst.localAddress(), dst.remoteAddress(),
9094
backendServer);
9195
src.resume();
9296
dst.resume();
9397
};
94-
dst.write(this.aesBase64Encode(backendServer))
98+
dst.write(this.aesBase64Encode(new MuxConfiguration(localServer.getName(), sessionId, backendServer)))
9599
.onSuccess(writeSuccessHandler)
96100
.onFailure(e -> {
97101
dst.close();

‎src/main/java/top/meethigher/proxy/tcp/mux/ReverseTcpProxyMuxServer.java‎

Lines changed: 24 additions & 20 deletions
Original file line numberDiff line numberDiff line change
@@ -7,6 +7,7 @@
77
import org.slf4j.Logger;
88
import org.slf4j.LoggerFactory;
99
import top.meethigher.proxy.NetAddress;
10+
import top.meethigher.proxy.tcp.mux.model.MuxConfiguration;
1011

1112
import java.util.concurrent.ThreadLocalRandom;
1213

@@ -38,11 +39,6 @@ protected ReverseTcpProxyMuxServer(Vertx vertx, String secret, NetServer netServ
3839

3940
protected void handleConnect(NetSocket src) {
4041
src.pause();
41-
log.debug("source {} -- {} connected", src.localAddress(), src.remoteAddress());
42-
// 由于内部都是使用pipe来进行数据传输，所以exceptionHandler肯定是都重新注册过了，参考{@code io.vertx.core.streams.impl.PipeImpl.PipeImpl }
43-
// 但如果还没进入pipe前，连接出现异常，那么就会触发此处的exceptionHandler。https://github.com/meethigher/tcp-reverse-proxy/issues/18
44-
src.exceptionHandler(e -> log.error("source {} -- {} exception occurred", src.localAddress(), src.remoteAddress(), e))
45-
.closeHandler(v -> log.debug("source {} -- {} closed", src.localAddress(), src.remoteAddress()));
4642
src.handler(new MuxMessageParser(muxMsg -> this.bindMuxConnections(src, muxMsg), src));
4743
src.resume();
4844
}
@@ -52,26 +48,32 @@ protected void handleConnect(NetSocket src) {
5248
*/
5349
protected void bindMuxConnections(NetSocket src, MuxMessageParser.MuxMessage muxMsg) {
5450
src.pause();
55-
NetAddress backend = aesBase64Decode(muxMsg.backendServerBuf);
56-
if (backend == null) {
51+
MuxConfiguration cfg = aesBase64Decode(muxMsg.backendServerBuf);
52+
if (cfg == null) {
5753
log.warn("source {} -- {} exception occurred: failed to parsing the backendServer address from encrypted content:{}",
5854
src.localAddress(), src.remoteAddress(),
5955
muxMsg.backendServerBuf);
6056
src.close();
6157
return;
6258
}
59+
NetAddress backend = cfg.backendServer;
60+
log.debug("{}: sessionId {}, source {} -- {} connected", cfg.name, cfg.sessionId, src.localAddress(), src.remoteAddress());
61+
// 由于内部都是使用pipe来进行数据传输，所以exceptionHandler肯定是都重新注册过了，参考{@code io.vertx.core.streams.impl.PipeImpl.PipeImpl }
62+
// 但如果还没进入pipe前，连接出现异常，那么就会触发此处的exceptionHandler。https://github.com/meethigher/tcp-reverse-proxy/issues/18
63+
src.exceptionHandler(e -> log.error("{}: sessionId {}, source {} -- {} exception occurred", cfg.name, cfg.sessionId, src.localAddress(), src.remoteAddress(), e))
64+
.closeHandler(v -> log.debug("{}: sessionId {}, source {} -- {} closed", cfg.name, cfg.sessionId, src.localAddress(), src.remoteAddress()));
6365
netClient.connect(backend.getPort(), backend.getHost())
6466
.onFailure(e -> {
65-
log.error("source {} -- {} failed to connect to {}", src.localAddress(), src.remoteAddress(), backend, e);
67+
log.error("{}: sessionId {}, source {} -- {} failed to connect to {}", cfg.name, cfg.sessionId, src.localAddress(), src.remoteAddress(), backend, e);
6668
src.close();
6769
})
6870
.onSuccess(dst -> {
6971
dst.pause();
70-
log.debug("target {} -- {} connected", dst.localAddress(), dst.remoteAddress());
72+
log.debug("{}: sessionId {}, target {} -- {} connected", cfg.name, cfg.sessionId, dst.localAddress(), dst.remoteAddress());
7173
// 由于内部都是使用pipe来进行数据传输，所以exceptionHandler肯定是都重新注册过了，参考{@code io.vertx.core.streams.impl.PipeImpl.PipeImpl }
7274
// 但如果还没进入pipe前，连接出现异常，那么就会触发此处的exceptionHandler。https://github.com/meethigher/tcp-reverse-proxy/issues/18
73-
dst.exceptionHandler(e -> log.error("target {} -- {} exception occurred", dst.localAddress(), dst.remoteAddress(), e))
74-
.closeHandler(v -> log.debug("target {} -- {} closed", dst.localAddress(), dst.remoteAddress()));
75+
dst.exceptionHandler(e -> log.error("{}: sessionId {}, target {} -- {} exception occurred", cfg.name, cfg.sessionId, dst.localAddress(), dst.remoteAddress(), e))
76+
.closeHandler(v -> log.debug("{}: sessionId {}, target {} -- {} closed", cfg.name, cfg.sessionId, dst.localAddress(), dst.remoteAddress()));
7577
/**
7678
* 不能使用write的成功与否判断链路是否正常，但是可以通过write.onSuccess保证顺序写入。
7779
* 测试中发现，即便链路异常，返回仍然是true 参考 https://github.com/meethigher/bug-test/blob/vertx-network-disconnect/src/main/java/top/meethigher/BugTest.java
@@ -83,16 +85,18 @@ protected void bindMuxConnections(NetSocket src, MuxMessageParser.MuxMessage mux
8385
// https://github.com/meethigher/tcp-reverse-proxy/issues/12
8486
// 将日志记录详细，便于排查问题
8587
src.pipeTo(dst)
86-
.onSuccess(v -> log.debug("source {} -- {} pipe to target {} -- {} succeeded",
87-
src.localAddress(), src.remoteAddress(), dst.localAddress(), dst.remoteAddress()))
88-
.onFailure(e -> log.error("source {} -- {} pipe to target {} -- {} failed",
89-
src.localAddress(), src.remoteAddress(), dst.localAddress(), dst.remoteAddress(), e));
88+
.onSuccess(v -> log.debug("{}: sessionId {}, source {} -- {} pipe to target {} -- {} succeeded",
89+
cfg.name, cfg.sessionId, src.localAddress(), src.remoteAddress(), dst.localAddress(), dst.remoteAddress()))
90+
.onFailure(e -> log.error("{}: sessionId {}, source {} -- {} pipe to target {} -- {} failed",
91+
cfg.name, cfg.sessionId, src.localAddress(), src.remoteAddress(), dst.localAddress(), dst.remoteAddress(), e));
9092
dst.pipeTo(src)
91-
.onSuccess(v -> log.debug("target {} -- {} pipe to source {} -- {} succeeded",
92-
dst.localAddress(), dst.remoteAddress(), src.localAddress(), src.remoteAddress()))
93-
.onFailure(e -> log.error("target {} -- {} pipe to source {} -- {} failed",
94-
dst.localAddress(), dst.remoteAddress(), src.localAddress(), src.remoteAddress(), e));
95-
log.debug("source {} -- {} bound to target {} -- {}", src.localAddress(), src.remoteAddress(),
93+
.onSuccess(v -> log.debug("{}: sessionId {}, target {} -- {} pipe to source {} -- {} succeeded",
94+
cfg.name, cfg.sessionId, dst.localAddress(), dst.remoteAddress(), src.localAddress(), src.remoteAddress()))
95+
.onFailure(e -> log.error("{}: sessionId {}, target {} -- {} pipe to source {} -- {} failed",
96+
cfg.name, cfg.sessionId, dst.localAddress(), dst.remoteAddress(), src.localAddress(), src.remoteAddress(), e));
97+
log.debug("{}: sessionId {}, source {} -- {} bound to target {} -- {}",
98+
cfg.name, cfg.sessionId,
99+
src.localAddress(), src.remoteAddress(),
96100
dst.localAddress(), dst.remoteAddress());
97101
src.resume();
98102
dst.resume();

‎src/main/java/top/meethigher/proxy/tcp/mux/model/MuxConfiguration.java‎

Lines changed: 34 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,34 @@
1+
package top.meethigher.proxy.tcp.mux.model;
2+
3+
import top.meethigher.proxy.NetAddress;
4+
5+
public class MuxConfiguration {
6+
7+
public final String name;
8+
public final int sessionId;
9+
public final NetAddress backendServer;
10+
11+
12+
public MuxConfiguration(String name, int sessionId, NetAddress backendServer) {
13+
this.name = name;
14+
this.sessionId = sessionId;
15+
this.backendServer = backendServer;
16+
}
17+
18+
@Override
19+
public String toString() {
20+
return name + "," + sessionId + "," + backendServer.toString();
21+
}
22+
23+
public static MuxConfiguration parse(String configuration) {
24+
MuxConfiguration muxConfiguration = null;
25+
try {
26+
if (configuration.contains(",")) {
27+
String[] arr = configuration.split(",");
28+
muxConfiguration = new MuxConfiguration(arr[0], Integer.parseInt(arr[1]), NetAddress.parse(arr[2]));
29+
}
30+
} catch (Exception ignore) {
31+
}
32+
return muxConfiguration;
33+
}
34+
}

‎src/test/java/top/meethigher/proxy/tcp/mux/MuxMessageParserTest.java‎

Lines changed: 2 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -7,6 +7,7 @@
77
import io.vertx.core.parsetools.RecordParser;
88
import org.junit.Test;
99
import top.meethigher.proxy.NetAddress;
10+
import top.meethigher.proxy.tcp.mux.model.MuxConfiguration;
1011

1112
import java.util.concurrent.locks.LockSupport;
1213

@@ -60,7 +61,7 @@ public void name() {
6061
NetAddress netAddress1 = new NetAddress("127.0.0.1", 8080);
6162

6263

63-
Buffer buffer1 = mux.aesBase64Encode(netAddress1);
64+
Buffer buffer1 = mux.aesBase64Encode(new MuxConfiguration("test", 1, netAddress1));
6465

6566

6667
Buffer buffer2 = Buffer.buffer("halo wode");

‎src/test/java/top/meethigher/proxy/tcp/mux/MuxTest.java‎

Lines changed: 3 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -4,6 +4,7 @@
44
import io.vertx.core.buffer.Buffer;
55
import org.junit.Test;
66
import top.meethigher.proxy.NetAddress;
7+
import top.meethigher.proxy.tcp.mux.model.MuxConfiguration;
78

89
import java.util.ArrayList;
910
import java.util.List;
@@ -23,9 +24,9 @@ public void name() {
2324
list.add(netAddress1);
2425
list.add(netAddress2);
2526

26-
Buffer encode = mux.aesBase64Encode(netAddress1);
27+
Buffer encode = mux.aesBase64Encode(new MuxConfiguration("http",1,netAddress1));
2728

28-
NetAddress decode = mux.aesBase64Decode(encode);
29+
MuxConfiguration decode = mux.aesBase64Decode(encode);
2930
System.out.println(decode);
3031
}
3132
}

0 commit comments

Comments
 (0)