util: fix Host column of show processlist to host_name:client_por…

…t format (#22707)

infoschema/tables_test.go

@@ -463,6 +463,7 @@ func (s *testTableSuite) TestSomeTables(c *C) {
 		ID:      1,
 		User:    "user-1",
 		Host:    "localhost",
+		Port:    "",
 		DB:      "information_schema",
 		Command: byte(1),
 		Digest:  "abc1",
@@ -474,28 +475,44 @@ func (s *testTableSuite) TestSomeTables(c *C) {
 		ID:      2,
 		User:    "user-2",
 		Host:    "localhost",
+		Port:    "",
 		DB:      "test",
 		Command: byte(2),
 		Digest:  "abc2",
 		State:   2,
 		Info:    strings.Repeat("x", 101),
 		StmtCtx: tk.Se.GetSessionVars().StmtCtx,
 	}
+	sm.processInfoMap[3] = &util.ProcessInfo{
+		ID:      3,
+		User:    "user-3",
+		Host:    "127.0.0.1",
+		Port:    "12345",
+		DB:      "test",
+		Command: byte(2),
+		Digest:  "abc3",
+		State:   1,
+		Info:    "check port",
+		StmtCtx: tk.Se.GetSessionVars().StmtCtx,
+	}
 	tk.Se.SetSessionManager(sm)
 	tk.MustQuery("select * from information_schema.PROCESSLIST order by ID;").Sort().Check(
 		testkit.Rows(
 			fmt.Sprintf("1 user-1 localhost information_schema Quit 9223372036 %s %s abc1 0 0 ", "in transaction", "do something"),
 			fmt.Sprintf("2 user-2 localhost test Init DB 9223372036 %s %s abc2 0 0 ", "autocommit", strings.Repeat("x", 101)),
+			fmt.Sprintf("3 user-3 127.0.0.1:12345 test Init DB 9223372036 %s %s abc3 0 0 ", "in transaction", "check port"),
 		))
 	tk.MustQuery("SHOW PROCESSLIST;").Sort().Check(
 		testkit.Rows(
 			fmt.Sprintf("1 user-1 localhost information_schema Quit 9223372036 %s %s", "in transaction", "do something"),
 			fmt.Sprintf("2 user-2 localhost test Init DB 9223372036 %s %s", "autocommit", strings.Repeat("x", 100)),
+			fmt.Sprintf("3 user-3 127.0.0.1:12345 test Init DB 9223372036 %s %s", "in transaction", "check port"),
 		))
 	tk.MustQuery("SHOW FULL PROCESSLIST;").Sort().Check(
 		testkit.Rows(
 			fmt.Sprintf("1 user-1 localhost information_schema Quit 9223372036 %s %s", "in transaction", "do something"),
 			fmt.Sprintf("2 user-2 localhost test Init DB 9223372036 %s %s", "autocommit", strings.Repeat("x", 101)),
+			fmt.Sprintf("3 user-3 127.0.0.1:12345 test Init DB 9223372036 %s %s", "in transaction", "check port"),
 		))
 
 	sm = &mockSessionManager{make(map[uint64]*util.ProcessInfo, 2)}

server/conn.go

@@ -682,13 +682,14 @@ func (cc *clientConn) openSessionAndDoAuth(authData []byte) error {
 	if len(authData) == 0 {
 		hasPassword = "NO"
 	}
-	host, err := cc.PeerHost(hasPassword)
+	host, port, err := cc.PeerHost(hasPassword)
 	if err != nil {
 		return err
 	}
 	if !cc.ctx.Auth(&auth.UserIdentity{Username: cc.user, Hostname: host}, authData, cc.salt) {
 		return errAccessDenied.FastGenByArgs(cc.user, host, hasPassword)
 	}
+	cc.ctx.SetPort(port)
 	if cc.dbname != "" {
 		err = cc.useDB(context.Background(), cc.dbname)
 		if err != nil {
@@ -699,17 +700,16 @@ func (cc *clientConn) openSessionAndDoAuth(authData []byte) error {
 	return nil
 }
 
-func (cc *clientConn) PeerHost(hasPassword string) (host string, err error) {
+func (cc *clientConn) PeerHost(hasPassword string) (host, port string, err error) {
 	if len(cc.peerHost) > 0 {
-		return cc.peerHost, nil
+		return cc.peerHost, "", nil
 	}
 	host = variable.DefHostname
 	if cc.server.isUnixSocket() {
 		cc.peerHost = host
 		return
 	}
 	addr := cc.bufReadConn.RemoteAddr().String()
-	var port string
 	host, port, err = net.SplitHostPort(addr)
 	if err != nil {
 		err = errAccessDenied.GenWithStackByArgs(cc.user, addr, hasPassword)

server/server.go

@@ -344,7 +344,7 @@ func (s *Server) Run() error {
 		err = plugin.ForeachPlugin(plugin.Audit, func(p *plugin.Plugin) error {
 			authPlugin := plugin.DeclareAuditManifest(p.Manifest)
 			if authPlugin.OnConnectionEvent != nil {
-				host, err := clientConn.PeerHost("")
+				host, _, err := clientConn.PeerHost("")
 				if err != nil {
 					logutil.BgLogger().Error("get peer host failed", zap.Error(err))
 					terror.Log(clientConn.Close())

session/session.go

@@ -145,6 +145,7 @@ type Session interface {
 	PrepareTxnCtx(context.Context)
 	// FieldList returns fields list of a table.
 	FieldList(tableName string) (fields []*ast.ResultField, err error)
+	SetPort(port string)
 }
 
 var (
@@ -1232,6 +1233,7 @@ func (s *session) SetProcessInfo(sql string, t time.Time, command byte, maxExecu
 	}
 	pi := util.ProcessInfo{
 		ID:               s.sessionVars.ConnectionID,
+		Port:             s.sessionVars.Port,
 		DB:               s.sessionVars.CurrentDB,
 		Command:          command,
 		Plan:             p,
@@ -2970,3 +2972,7 @@ func (s *session) checkPlacementPolicyBeforeCommit() error {
 	}
 	return err
 }
+
+func (s *session) SetPort(port string) {
+	s.sessionVars.Port = port
+}

sessionctx/variable/session.go

@@ -436,6 +436,9 @@ type SessionVars struct {
 	// User is the user identity with which the session login.
 	User *auth.UserIdentity
 
+	// Port is the port of the connected socket
+	Port string
+
 	// CurrentDB is the default database of this session.
 	CurrentDB string
 

util/processinfo.go

@@ -32,6 +32,7 @@ type ProcessInfo struct {
 	ID               uint64
 	User             string
 	Host             string
+	Port             string
 	DB               string
 	Digest           string
 	Plan             interface{}
@@ -67,10 +68,16 @@ func (pi *ProcessInfo) ToRowForShow(full bool) []interface{} {
 	if len(pi.DB) > 0 {
 		db = pi.DB
 	}
+	var host string
+	if pi.Port != "" {
+		host = fmt.Sprintf("%s:%s", pi.Host, pi.Port)
+	} else {
+		host = pi.Host
+	}
 	return []interface{}{
 		pi.ID,
 		pi.User,
-		pi.Host,
+		host,
 		db,
 		mysql.Command2Str[pi.Command],
 		t,