Merge pull request kubernetes-sigs#269 from cheftako/agent-log

Instrumenting connection issues on Proxy Server

Author: k8s-ci-robot

cmd/server/app/options/options.go

@@ -70,6 +70,13 @@ type ProxyRunOptions struct {
 	// backend within the destCIDR. if it still can't find any backend,
 	// it will use the default backend manager to choose a random backend.
 	ProxyStrategies string
+
+	// This controls if we attempt to push onto a "full" transfer channel.
+	// However checking that the transfer channel is full is not safe.
+	// It violates our race condition checking. Adding locks around a potentially
+	// blocking call has its own problems, so it cannot easily be made race condition safe.
+	// The check is an "unlocked" read but is still use at your own peril.
+	WarnOnChannelLimit bool
 }
 
 func (o *ProxyRunOptions) Flags() *pflag.FlagSet {
@@ -100,6 +107,7 @@ func (o *ProxyRunOptions) Flags() *pflag.FlagSet {
 	flags.IntVar(&o.KubeconfigBurst, "kubeconfig-burst", o.KubeconfigBurst, "Maximum client burst (proxy server uses this client to authenticate agent tokens).")
 	flags.StringVar(&o.AuthenticationAudience, "authentication-audience", o.AuthenticationAudience, "Expected agent's token authentication audience (used with agent-namespace, agent-service-account, kubeconfig).")
 	flags.StringVar(&o.ProxyStrategies, "proxy-strategies", o.ProxyStrategies, "The list of proxy strategies used by the server to pick a backend/tunnel, available strategies are: default, destHost.")
+	flags.BoolVar(&o.WarnOnChannelLimit, "warn-on-channel-limit", o.WarnOnChannelLimit, "Turns on a warning if the system is going to push to a full channel. The check involves an unsafe read.")
 	return flags
 }
 
@@ -130,6 +138,7 @@ func (o *ProxyRunOptions) Print() {
 	klog.V(1).Infof("KubeconfigQPS set to %f.\n", o.KubeconfigQPS)
 	klog.V(1).Infof("KubeconfigBurst set to %d.\n", o.KubeconfigBurst)
 	klog.V(1).Infof("ProxyStrategies set to %q.\n", o.ProxyStrategies)
+	klog.V(1).Infof("WarnOnChannelLimit set to %t.\n", o.WarnOnChannelLimit)
 }
 
 func (o *ProxyRunOptions) Validate() error {
@@ -290,6 +299,7 @@ func NewProxyRunOptions() *ProxyRunOptions {
 		KubeconfigBurst:           0,
 		AuthenticationAudience:    "",
 		ProxyStrategies:           "default",
+		WarnOnChannelLimit:        false,
 	}
 	return &o
 }

cmd/server/app/server.go

@@ -92,7 +92,7 @@ func (p *Proxy) run(o *options.ProxyRunOptions) error {
 	if err != nil {
 		return err
 	}
-	server := server.NewProxyServer(o.ServerID, ps, int(o.ServerCount), authOpt)
+	server := server.NewProxyServer(o.ServerID, ps, int(o.ServerCount), authOpt, o.WarnOnChannelLimit)
 
 	frontendStop, err := p.runFrontendServer(ctx, o, server)
 	if err != nil {

pkg/server/server.go

@@ -40,6 +40,8 @@ import (
 	"sigs.k8s.io/apiserver-network-proxy/proto/header"
 )
 
+const xfrChannelSize = 10
+
 type key int
 
 type ProxyClientConnection struct {
@@ -136,6 +138,9 @@ type ProxyServer struct {
 	serverID    string // unique ID of this server
 	serverCount int    // Number of proxy server instances, should be 1 unless it is a HA server.
 
+	// Allows a special debug flag which warns if we write to a full transfer channel
+	warnOnChannelLimit bool
+
 	// agent authentication
 	AgentAuthenticationOptions *AgentTokenAuthenticationOptions
 
@@ -321,7 +326,7 @@ func (s *ProxyServer) getFrontendsForBackendConn(agentID string, backend Backend
 }
 
 // NewProxyServer creates a new ProxyServer instance
-func NewProxyServer(serverID string, proxyStrategies []ProxyStrategy, serverCount int, agentAuthenticationOptions *AgentTokenAuthenticationOptions) *ProxyServer {
+func NewProxyServer(serverID string, proxyStrategies []ProxyStrategy, serverCount int, agentAuthenticationOptions *AgentTokenAuthenticationOptions, warnOnChannelLimit bool) *ProxyServer {
 	var bms []BackendManager
 	for _, ps := range proxyStrategies {
 		switch ps {
@@ -343,9 +348,10 @@ func NewProxyServer(serverID string, proxyStrategies []ProxyStrategy, serverCoun
 		serverCount:                serverCount,
 		BackendManagers:            bms,
 		AgentAuthenticationOptions: agentAuthenticationOptions,
-		// use the first backendmanager as the Readiness Manager
-		Readiness:       bms[0],
-		proxyStrategies: proxyStrategies,
+		// use the first backend-manager as the Readiness Manager
+		Readiness:          bms[0],
+		proxyStrategies:    proxyStrategies,
+		warnOnChannelLimit: warnOnChannelLimit,
 	}
 }
 
@@ -361,12 +367,13 @@ func (s *ProxyServer) Proxy(stream client.ProxyService_ProxyServer) error {
 	userAgent := md.Get(header.UserAgent)
 	klog.V(2).InfoS("proxy request from client", "userAgent", userAgent)
 
-	recvCh := make(chan *client.Packet, 10)
+	recvCh := make(chan *client.Packet, xfrChannelSize)
 	stopCh := make(chan error)
 
 	go s.serveRecvFrontend(stream, recvCh)
 
 	defer func() {
+		klog.V(2).InfoS("Receive channel on Proxy is stopping", "userAgent", userAgent, "serverID", s.serverID)
 		close(recvCh)
 	}()
 
@@ -375,6 +382,7 @@ func (s *ProxyServer) Proxy(stream client.ProxyService_ProxyServer) error {
 		for {
 			in, err := stream.Recv()
 			if err == io.EOF {
+				klog.V(2).InfoS("Stream closed on Proxy", "userAgent", userAgent, "serverID", s.serverID)
 				close(stopCh)
 				return
 			}
@@ -384,6 +392,9 @@ func (s *ProxyServer) Proxy(stream client.ProxyService_ProxyServer) error {
 				return
 			}
 
+			if s.warnOnChannelLimit && len(recvCh) >= xfrChannelSize {
+				klog.V(2).InfoS("Receive channel on Proxy is full", "userAgent", userAgent, "serverID", s.serverID)
+			}
 			recvCh <- in
 		}
 	}()
@@ -410,13 +421,15 @@ func (s *ProxyServer) serveRecvFrontend(stream client.ProxyService_ProxyServer,
 			// a new connection to the address.
 			backend, err = s.getBackend(pkt.GetDialRequest().Address)
 			if err != nil {
-				klog.ErrorS(err, "Failed to get a backend")
+				klog.ErrorS(err, "Failed to get a backend", "serverID", s.serverID)
 
 				resp := &client.Packet{
 					Type:    client.PacketType_DIAL_RSP,
 					Payload: &client.Packet_DialResponse{DialResponse: &client.DialResponse{Error: err.Error()}},
 				}
-				stream.Send(resp)
+				if err := stream.Send(resp); err != nil {
+					klog.V(5).Infoln("Failed to send DIAL_RSP for no backend", "error", err, "serverID", s.serverID)
+				}
 				// The Dial is failing; no reason to keep this goroutine.
 				return
 			}
@@ -430,29 +443,30 @@ func (s *ProxyServer) serveRecvFrontend(stream client.ProxyService_ProxyServer,
 					backend:   backend,
 				})
 			if err := backend.Send(pkt); err != nil {
-				klog.ErrorS(err, "DIAL_REQ to Backend failed")
+				klog.ErrorS(err, "DIAL_REQ to Backend failed", "serverID", s.serverID)
 			}
 			klog.V(5).Infoln("DIAL_REQ sent to backend") // got this. but backend didn't receive anything.
 
 		case client.PacketType_CLOSE_REQ:
 			connID := pkt.GetCloseRequest().ConnectID
 			klog.V(5).InfoS("Received CLOSE_REQ", "connectionID", connID)
 			if backend == nil {
-				klog.V(2).InfoS("Backend has not been initialized for requested connection. Client should send a Dial Request first", "connectionID", connID)
+				klog.V(2).InfoS("Backend has not been initialized for requested connection. Client should send a Dial Request first",
+					"serverID", s.serverID, "connectionID", connID)
 				continue
 			}
 			if err := backend.Send(pkt); err != nil {
 				// TODO: retry with other backends connecting to this agent.
-				klog.ErrorS(err, "CLOSE_REQ to Backend failed")
+				klog.ErrorS(err, "CLOSE_REQ to Backend failed", "serverID", s.serverID, "connectionID", connID)
 			}
-			klog.V(5).Infoln("CLOSE_REQ sent to backend")
+			klog.V(5).Infoln("CLOSE_REQ sent to backend", "serverID", s.serverID, "connectionID", connID)
 
 		case client.PacketType_DIAL_CLS:
 			random := pkt.GetCloseDial().Random
-			klog.V(5).InfoS("Received DIAL_CLOSE", "random", random)
+			klog.V(5).InfoS("Received DIAL_CLOSE", "serverID", s.serverID, "random", random)
 			// Currently not worrying about backend as we do not have an established connection,
 			s.PendingDial.Remove(random)
-			klog.V(5).Infoln("Removing pending dial request", "random", random)
+			klog.V(5).Infoln("Removing pending dial request", "serverID", s.serverID, "random", random)
 
 		case client.PacketType_DATA:
 			connID := pkt.GetData().ConnectID
@@ -470,17 +484,18 @@ func (s *ProxyServer) serveRecvFrontend(stream client.ProxyService_ProxyServer,
 			}
 			if err := backend.Send(pkt); err != nil {
 				// TODO: retry with other backends connecting to this agent.
-				klog.ErrorS(err, "DATA to Backend failed")
+				klog.ErrorS(err, "DATA to Backend failed", "serverID", s.serverID, "connectionID", connID)
 				continue
 			}
 			klog.V(5).Infoln("DATA sent to Backend")
 
 		default:
-			klog.V(5).InfoS("Ignore packet coming from frontend", "type", pkt.Type)
+			klog.V(5).InfoS("Ignore packet coming from frontend",
+				"type", pkt.Type, "serverID", s.serverID, "connectionID", firstConnID)
 		}
 	}
 
-	klog.V(5).InfoS("Close streaming", "connectionID", firstConnID)
+	klog.V(5).InfoS("Close streaming", "serverID", s.serverID, "connectionID", firstConnID)
 
 	pkt := &client.Packet{
 		Type: client.PacketType_CLOSE_REQ,
@@ -496,7 +511,7 @@ func (s *ProxyServer) serveRecvFrontend(stream client.ProxyService_ProxyServer,
 		return
 	}
 	if err := backend.Send(pkt); err != nil {
-		klog.ErrorS(err, "CLOSE_REQ to Backend failed")
+		klog.ErrorS(err, "CLOSE_REQ to Backend failed", "serverID", s.serverID)
 	}
 }
 
@@ -617,24 +632,26 @@ func (s *ProxyServer) Connect(stream agent.AgentService_ConnectServer) error {
 
 	if s.AgentAuthenticationOptions.Enabled {
 		if err := s.authenticateAgentViaToken(stream.Context()); err != nil {
-			klog.ErrorS(err, "Client authentication failed")
+			klog.ErrorS(err, "Client authentication failed", "agentID", agentID)
 			return err
 		}
 	}
 
 	h := metadata.Pairs(header.ServerID, s.serverID, header.ServerCount, strconv.Itoa(s.serverCount))
 	if err := stream.SendHeader(h); err != nil {
+		klog.ErrorS(err, "Failed to send server count back to agent", "agentID", agentID)
 		return err
 	}
 
 	backend := s.addBackend(agentID, stream)
 	defer s.removeBackend(agentID, stream)
 
-	recvCh := make(chan *client.Packet, 10)
+	recvCh := make(chan *client.Packet, xfrChannelSize)
 
 	go s.serveRecvBackend(backend, stream, agentID, recvCh)
 
 	defer func() {
+		klog.V(2).InfoS("Receive channel on Connect is stopping", "agentID", agentID, "serverID", s.serverID)
 		close(recvCh)
 	}()
 
@@ -643,6 +660,7 @@ func (s *ProxyServer) Connect(stream agent.AgentService_ConnectServer) error {
 		for {
 			in, err := stream.Recv()
 			if err == io.EOF {
+				klog.V(2).InfoS("Stream closed on Connect", "agentID", agentID, "serverID", s.serverID)
 				close(stopCh)
 				return
 			}
@@ -652,6 +670,9 @@ func (s *ProxyServer) Connect(stream agent.AgentService_ConnectServer) error {
 				return
 			}
 
+			if s.warnOnChannelLimit && len(recvCh) >= xfrChannelSize {
+				klog.V(2).InfoS("Receive channel on Connect is full", "agentID", agentID, "serverID", s.serverID)
+			}
 			recvCh <- in
 		}
 	}()
@@ -666,7 +687,8 @@ func (s *ProxyServer) serveRecvBackend(backend Backend, stream agent.AgentServic
 		// TODO(#126): Frontends in PendingDial state that have not been added to the
 		//             list of frontends should also be closed.
 		frontends, _ := s.getFrontendsForBackendConn(agentID, backend)
-		klog.V(3).InfoS("Close frontends connected to agent", "count", len(frontends), "agentID", agentID)
+		klog.V(3).InfoS("Close frontends connected to agent",
+			"serverID", s.serverID, "count", len(frontends), "agentID", agentID)
 
 		for _, frontend := range frontends {
 			s.removeFrontend(agentID, frontend.connectID)
@@ -678,7 +700,7 @@ func (s *ProxyServer) serveRecvBackend(backend Backend, stream agent.AgentServic
 			}
 			pkt.GetCloseResponse().ConnectID = frontend.connectID
 			if err := frontend.send(pkt); err != nil {
-				klog.ErrorS(err, "CLOSE_RSP to frontend failed")
+				klog.ErrorS(err, "CLOSE_RSP to frontend failed", "serverID", s.serverID, "agentID", agentID)
 			}
 		}
 	}()
@@ -690,17 +712,18 @@ func (s *ProxyServer) serveRecvBackend(backend Backend, stream agent.AgentServic
 			klog.V(5).InfoS("Received DIAL_RSP", "random", resp.Random, "agentID", agentID, "connectionID", resp.ConnectID)
 
 			if frontend, ok := s.PendingDial.Get(resp.Random); !ok {
-				klog.V(5).Infoln("DIAL_RSP not recognized; dropped")
+				klog.V(2).Infoln("DIAL_RSP not recognized; dropped", "random", resp.Random, "agentID", agentID, "connectionID", resp.ConnectID)
 			} else {
 				dialErr := false
 				if resp.Error != "" {
-					klog.ErrorS(errors.New(resp.Error), "DIAL_RSP contains failure")
+					klog.ErrorS(errors.New(resp.Error), "DIAL_RSP contains failure", "random", resp.Random, "agentID", agentID, "connectionID", resp.ConnectID)
 					dialErr = true
 				}
 				err := frontend.send(pkt)
 				s.PendingDial.Remove(resp.Random)
 				if err != nil {
-					klog.ErrorS(err, "DIAL_RSP send to frontend stream failure")
+					klog.ErrorS(err, "DIAL_RSP send to frontend stream failure",
+						"random", resp.Random, "serverID", s.serverID, "agentID", agentID, "connectionID", resp.ConnectID)
 					dialErr = true
 				}
 				// Avoid adding the frontend if there was an error dialing the destination
@@ -719,35 +742,35 @@ func (s *ProxyServer) serveRecvBackend(backend Backend, stream agent.AgentServic
 			klog.V(5).InfoS("Received data from agent", "bytes", len(resp.Data), "agentID", agentID, "connectionID", resp.ConnectID)
 			frontend, err := s.getFrontend(agentID, resp.ConnectID)
 			if err != nil {
-				klog.ErrorS(err, "could not get frontend client", "connectionID", resp.ConnectID)
+				klog.ErrorS(err, "could not get frontend client", "serverID", s.serverID, "agentID", agentID, "connectionID", resp.ConnectID)
 				break
 			}
 			if err := frontend.send(pkt); err != nil {
-				klog.ErrorS(err, "send to client stream failure")
+				klog.ErrorS(err, "send to client stream failure", "serverID", s.serverID, "agentID", agentID, "connectionID", resp.ConnectID)
 			} else {
 				klog.V(5).InfoS("DATA sent to frontend")
 			}
 
 		case client.PacketType_CLOSE_RSP:
 			resp := pkt.GetCloseResponse()
-			klog.V(5).InfoS("Received CLOSE_RSP", "connectionID", resp.ConnectID)
+			klog.V(5).InfoS("Received CLOSE_RSP", "serverID", s.serverID, "agentID", agentID, "connectionID", resp.ConnectID)
 			frontend, err := s.getFrontend(agentID, resp.ConnectID)
 			if err != nil {
-				klog.ErrorS(err, "could not get frontend client", "connectionID", resp.ConnectID)
+				klog.ErrorS(err, "could not get frontend client", "serverID", s.serverID, "agentID", agentID, "connectionID", resp.ConnectID)
 				break
 			}
 			if err := frontend.send(pkt); err != nil {
 				// Normal when frontend closes it.
-				klog.ErrorS(err, "CLOSE_RSP send to client stream error", "connectionID", resp.ConnectID)
+				klog.ErrorS(err, "CLOSE_RSP send to client stream error", "serverID", s.serverID, "agentID", agentID, "connectionID", resp.ConnectID)
 			} else {
 				klog.V(5).Infoln("CLOSE_RSP sent to frontend", "connectionID", resp.ConnectID)
 			}
 			s.removeFrontend(agentID, resp.ConnectID)
 			klog.V(5).InfoS("Close streaming", "agentID", agentID, "connectionID", resp.ConnectID)
 
 		default:
-			klog.V(2).InfoS("Unrecognized packet", "packet", pkt)
+			klog.V(2).InfoS("Unrecognized packet", "packet", pkt, "serverID", s.serverID, "agentID", agentID)
 		}
 	}
-	klog.V(5).InfoS("Close backend of agent", "backend", stream, "agentID", agentID)
+	klog.V(5).InfoS("Close backend of agent", "backend", stream, "serverID", s.serverID, "agentID", agentID)
 }

pkg/server/server_test.go

@@ -164,7 +164,9 @@ func TestAgentTokenAuthenticationErrorsToken(t *testing.T) {
 				KubernetesClient:    kcs,
 				AgentNamespace:      tc.wantNamespace,
 				AgentServiceAccount: tc.wantServiceAccount,
-			})
+			},
+				false,
+			)
 
 			err := p.Connect(conn)
 			if tc.wantError {
@@ -187,15 +189,15 @@ func TestAddRemoveFrontends(t *testing.T) {
 	agent2ConnID2 := new(ProxyClientConnection)
 	agent3ConnID1 := new(ProxyClientConnection)
 
-	p := NewProxyServer("", []ProxyStrategy{ProxyStrategyDefault}, 1, nil)
+	p := NewProxyServer("", []ProxyStrategy{ProxyStrategyDefault}, 1, nil, false)
 	p.addFrontend("agent1", int64(1), agent1ConnID1)
 	p.removeFrontend("agent1", int64(1))
 	expectedFrontends := make(map[string]map[int64]*ProxyClientConnection)
 	if e, a := expectedFrontends, p.frontends; !reflect.DeepEqual(e, a) {
 		t.Errorf("expected %v, got %v", e, a)
 	}
 
-	p = NewProxyServer("", []ProxyStrategy{ProxyStrategyDefault}, 1, nil)
+	p = NewProxyServer("", []ProxyStrategy{ProxyStrategyDefault}, 1, nil, false)
 	p.addFrontend("agent1", int64(1), agent1ConnID1)
 	p.addFrontend("agent1", int64(2), agent1ConnID2)
 	p.addFrontend("agent2", int64(1), agent2ConnID1)

tests/proxy_test.go

@@ -356,7 +356,7 @@ func runGRPCProxyServerWithServerCount(serverCount int) (proxy, *server.ProxySer
 	var err error
 	var lis, lis2 net.Listener
 
-	server := server.NewProxyServer(uuid.New().String(), []server.ProxyStrategy{server.ProxyStrategyDefault}, serverCount, &server.AgentTokenAuthenticationOptions{})
+	server := server.NewProxyServer(uuid.New().String(), []server.ProxyStrategy{server.ProxyStrategyDefault}, serverCount, &server.AgentTokenAuthenticationOptions{}, false)
 	grpcServer := grpc.NewServer()
 	agentServer := grpc.NewServer()
 	cleanup := func() {
@@ -395,7 +395,7 @@ func runGRPCProxyServerWithServerCount(serverCount int) (proxy, *server.ProxySer
 func runHTTPConnProxyServer() (proxy, func(), error) {
 	ctx := context.Background()
 	var proxy proxy
-	s := server.NewProxyServer(uuid.New().String(), []server.ProxyStrategy{server.ProxyStrategyDefault}, 0, &server.AgentTokenAuthenticationOptions{})
+	s := server.NewProxyServer(uuid.New().String(), []server.ProxyStrategy{server.ProxyStrategyDefault}, 0, &server.AgentTokenAuthenticationOptions{}, false)
 	agentServer := grpc.NewServer()
 
 	agentproto.RegisterAgentServiceServer(agentServer, s)