fix(feg): Diameter closed connection recovery

Signed-off-by: Evgeniy Makeev <evgeniym@fb.com>

feg/gateway/diameter/connection.go

@@ -18,6 +18,7 @@ import (
 	"net"
 	"strings"
 	"sync"
+	"time"
 
 	"github.com/fiorix/go-diameter/v4/diam"
 	"github.com/fiorix/go-diameter/v4/diam/avp"
@@ -33,6 +34,9 @@ type messageTypeEnum uint8
 const (
 	requestMessage messageTypeEnum = 1
 	answerMessage  messageTypeEnum = 2
+
+	connectionRecoveryInterval = time.Second
+	connectionRecoveryattempts = 6
 )
 
 // Connection is representing a diameter connection that you can
@@ -155,16 +159,50 @@ func (c *Connection) getDiamConnection() (diam.Conn, *smpeer.Metadata, error) {
 		return nil, nil, errors.New("Could not obtain metadata from connection")
 	}
 	c.conn, c.metadata = conn, metadata
+	if cn, ok := conn.(diam.CloseNotifier); ok && cn != nil {
+		go c.connCloseNotify(cn.CloseNotify(), conn)
+	} else {
+		glog.Errorf("new diam conn (%T) %s -> %s is not CloseNotifier", conn, localAddr, c.server.Addr)
+	}
 	return conn, metadata, nil
 }
 
+func (c *Connection) connCloseNotify(cnc <-chan struct{}, conn diam.Conn) {
+	<-cnc // wait for close notifier
+	if glog.V(2) {
+		remoteAddress, localAddress := "nil", "nil"
+		if conn != nil {
+			if conn.LocalAddr() != nil {
+				localAddress = conn.LocalAddr().String()
+			}
+			if conn.LocalAddr() != nil {
+				remoteAddress = conn.RemoteAddr().String()
+			}
+		}
+		glog.Infof("notified of %s->%s connection closure", localAddress, remoteAddress)
+	}
+	if c != nil && conn != nil && c.destroyConnection(conn) {
+		// if connection was closed not by connection management functions, recover it
+		retryWaitTime := connectionRecoveryInterval
+		for i := 0; i < connectionRecoveryattempts; i++ {
+			_, _, err := c.getDiamConnection()
+			if err == nil {
+				break
+			}
+			time.Sleep(retryWaitTime)
+			retryWaitTime *= 2
+		}
+	}
+}
+
 // destroyConnection closes a bad connection. If the connection
 // passed is the same as the one stored in the locked connection, it is nullified.
 // If the passed diam connection is not the same, this probably means another go routine
 // already created a new connection - just try to close it and return.
-func (c *Connection) destroyConnection(conn diam.Conn) {
+// destroyConnection returns true if connection matches and was closed
+func (c *Connection) destroyConnection(conn diam.Conn) bool {
 	if conn == nil {
-		return
+		return false
 	}
 	c.mutex.Lock()
 	match := conn == c.conn
@@ -191,6 +229,7 @@ func (c *Connection) destroyConnection(conn diam.Conn) {
 		}
 	}
 	conn.Close()
+	return match
 }
 
 // cleanupConnection is similar to destroyConnection, but it closes & cleans up connection unconditionally