Emit port forward event only once at setup instead than upon each con…

…nection. Include k8s cluster info in event.

api/proto/teleport/legacy/types/events/events.proto

@@ -1617,6 +1617,13 @@ message PortForward {
 
   // Addr is a target port forwarding address
   string Addr = 5 [(gogoproto.jsontag) = "addr"];
+
+  // Kubernetes has information about a kubernetes cluster, if applicable.
+  KubernetesClusterMetadata Kubernetes = 6 [
+    (gogoproto.nullable) = false,
+    (gogoproto.embed) = true,
+    (gogoproto.jsontag) = ""
+  ];
 }
 
 // X11Forward is emitted when a user requests X11 protocol forwarding

lib/kube/proxy/forwarder.go

@@ -1772,7 +1772,15 @@ func (f *Forwarder) portForward(authCtx *authContext, w http.ResponseWriter, req
 		return nil, trace.Wrap(err)
 	}
 
+	onForwardConnection := func(addr string, success bool) {
+		if !success {
+			f.log.Warn("Failed to establish connection to %q.", addr)
+		}
+		// TODO: Consider emit audit log events. Here we have the target pod port and an actual success value.
+	}
+
 	onPortForward := func(addr string, success bool) {
+		// NOTE: Kubectl doesn't send the target port initially so `addr` will have port 0.
 		if !sess.isLocalKubernetesCluster {
 			return
 		}
@@ -1791,7 +1799,10 @@ func (f *Forwarder) portForward(authCtx *authContext, w http.ResponseWriter, req
 			Status: apievents.Status{
 				Success: success,
 			},
+			KubernetesClusterMetadata: sess.eventClusterMeta(req),
 		}
+		// NOTE: Success will always be true as there is nothing that can fail at this point.
+		// TODO: Move this to onForwardConnection / rename. Keeping for now to avoid orphan const code.
 		if !success {
 			portForward.Code = events.PortForwardFailureCode
 		}
@@ -1802,15 +1813,16 @@ func (f *Forwarder) portForward(authCtx *authContext, w http.ResponseWriter, req
 
 	q := req.URL.Query()
 	request := portForwardRequest{
-		podNamespace:       p.ByName("podNamespace"),
-		podName:            p.ByName("podName"),
-		ports:              q["ports"],
-		context:            ctx,
-		httpRequest:        req,
-		httpResponseWriter: w,
-		onPortForward:      onPortForward,
-		targetDialer:       dialer,
-		pingPeriod:         f.cfg.ConnPingPeriod,
+		podNamespace:        p.ByName("podNamespace"),
+		podName:             p.ByName("podName"),
+		ports:               q["ports"],
+		context:             ctx,
+		httpRequest:         req,
+		httpResponseWriter:  w,
+		onPortForward:       onPortForward,
+		onForwardConnection: onForwardConnection,
+		targetDialer:        dialer,
+		pingPeriod:          f.cfg.ConnPingPeriod,
 	}
 	f.log.Debugf("Starting %v.", request)
 	err = runPortForwarding(request)
@@ -2186,7 +2198,6 @@ func (f *Forwarder) getSPDYExecutor(sess *clusterSession, req *http.Request) (re
 }
 
 func (f *Forwarder) getPortForwardDialer(sess *clusterSession, req *http.Request) (httpstream.Dialer, error) {
-
 	wsDialer, err := f.getWebsocketDialer(sess, req)
 	if err != nil {
 		return nil, trace.Wrap(err)

lib/kube/proxy/portforward_spdy.go

@@ -37,15 +37,16 @@ import (
 
 // portForwardRequest is a request that specifies port forwarding
 type portForwardRequest struct {
-	podNamespace       string
-	podName            string
-	ports              []string
-	httpRequest        *http.Request
-	httpResponseWriter http.ResponseWriter
-	onPortForward      portForwardCallback
-	context            context.Context
-	targetDialer       httpstream.Dialer
-	pingPeriod         time.Duration
+	podNamespace        string
+	podName             string
+	ports               []string
+	httpRequest         *http.Request
+	httpResponseWriter  http.ResponseWriter
+	onPortForward       portForwardCallback
+	onForwardConnection portForwardCallback
+	context             context.Context
+	targetDialer        httpstream.Dialer
+	pingPeriod          time.Duration
 }
 
 func (p portForwardRequest) String() string {
@@ -179,7 +180,7 @@ func (h *portForwardProxy) forwardStreamPair(p *httpStreamPair, remotePort int64
 
 	// read and write from the error stream
 	targetErrorStream, err := h.targetConn.CreateStream(headers)
-	h.onPortForward(net.JoinHostPort(h.podName, port), err == nil /* success */)
+	h.onForwardConnection(net.JoinHostPort(h.podName, port), err == nil /* success */)
 	if err != nil {
 		err := trace.ConnectionProblem(err, "error creating error stream for port %d", remotePort)
 		p.sendErr(err)
@@ -294,6 +295,7 @@ func (h *portForwardProxy) requestID(stream httpstream.Stream) (string, error) {
 // when the httpstream.Connection is closed.
 func (h *portForwardProxy) run() {
 	h.Debugf("Waiting for port forward streams.")
+	h.onPortForward(net.JoinHostPort(h.podName, "0"), true)
 	for {
 		select {
 		case <-h.context.Done():

lib/kube/proxy/portforward_test.go

@@ -282,8 +282,9 @@ func TestPortForwardProxy_run_connsClosed(t *testing.T) {
 	targetConn := newfakeSPDYConnection()
 	h := &portForwardProxy{
 		portForwardRequest: portForwardRequest{
-			context:       context.Background(),
-			onPortForward: func(addr string, success bool) {},
+			context:             context.Background(),
+			onPortForward:       func(addr string, success bool) {},
+			onForwardConnection: func(addr string, success bool) {},
 		},
 		Entry:                 logger,
 		sourceConn:            sourceConn,

lib/kube/proxy/portforward_websocket.go

@@ -23,6 +23,7 @@ import (
 	"encoding/binary"
 	"fmt"
 	"io"
+	"net"
 	"net/http"
 	"strings"
 	"sync"
@@ -142,11 +143,12 @@ func runPortForwardingWebSocket(req portForwardRequest) error {
 	}
 
 	h := &websocketPortforwardHandler{
-		conn:          conn,
-		streamPairs:   streamPairs,
-		podName:       req.podName,
-		targetConn:    targetConn,
-		onPortForward: req.onPortForward,
+		conn:                conn,
+		streamPairs:         streamPairs,
+		podName:             req.podName,
+		targetConn:          targetConn,
+		onPortForward:       req.onPortForward,
+		onForwardConnection: req.onForwardConnection,
 		FieldLogger: logrus.WithFields(logrus.Fields{
 			teleport.ComponentKey: teleport.Component(teleport.ComponentProxyKube),
 			events.RemoteAddr:     req.httpRequest.RemoteAddr,
@@ -207,11 +209,12 @@ func (w *websocketChannelPair) sendErr(err error) {
 // websocketPortforwardHandler is capable of processing a single port forward
 // request over a websocket connection
 type websocketPortforwardHandler struct {
-	conn          *wsstream.Conn
-	streamPairs   []*websocketChannelPair
-	podName       string
-	targetConn    httpstream.Connection
-	onPortForward portForwardCallback
+	conn                *wsstream.Conn
+	streamPairs         []*websocketChannelPair
+	podName             string
+	targetConn          httpstream.Connection
+	onPortForward       portForwardCallback
+	onForwardConnection portForwardCallback
 	logrus.FieldLogger
 	context context.Context
 }
@@ -223,6 +226,7 @@ func (h *websocketPortforwardHandler) run() {
 	wg := sync.WaitGroup{}
 	wg.Add(len(h.streamPairs))
 
+	h.onPortForward(net.JoinHostPort(h.podName, "0"), true)
 	for _, pair := range h.streamPairs {
 		p := pair
 		go func() {
@@ -251,7 +255,7 @@ func (h *websocketPortforwardHandler) forwardStreamPair(p *websocketChannelPair)
 
 	// read and write from the error stream
 	targetErrorStream, err := h.targetConn.CreateStream(headers)
-	h.onPortForward(fmt.Sprintf("%v:%v", h.podName, p.port), err == nil /* success */)
+	h.onForwardConnection(net.JoinHostPort(h.podName, fmt.Sprint(p.port)), err == nil /* success */)
 	if err != nil {
 		p.sendErr(err)
 		return