agent: rewrite readiness probe (istio#16336)

* pilot-agent: support filter chain inbound listener

Signed-off-by: Yuchen Dai <[email protected]>

* apply and add test

Signed-off-by: Yuchen Dai <[email protected]>

* lint

Signed-off-by: Yuchen Dai <[email protected]>

* improve test

Signed-off-by: Yuchen Dai <[email protected]>

* lint again

Signed-off-by: Yuchen Dai <[email protected]>

* apply and add test

Signed-off-by: Yuchen Dai <[email protected]>

* Address comment: 1.3 pilot agent assumes virtual inbound listener using filter chains

Signed-off-by: Yuchen Dai <[email protected]>

* remove env var

Signed-off-by: Yuchen Dai <[email protected]>

* not check inbound for sidecar proxy

Signed-off-by: Yuchen Dai <[email protected]>

* retore check virtual inbound

Signed-off-by: Yuchen Dai <[email protected]>

* check inbound listener only for router type envoy

Signed-off-by: Yuchen Dai <[email protected]>

Author: lambdai

pilot/cmd/pilot-agent/status/ready/probe.go

@@ -17,22 +17,20 @@ package ready
 import (
 	"fmt"
 
-	"github.com/hashicorp/go-multierror"
-
-	"istio.io/istio/pilot/pkg/model"
-
 	admin "github.com/envoyproxy/go-control-plane/envoy/admin/v2alpha"
+	"github.com/hashicorp/go-multierror"
 
 	"istio.io/istio/pilot/cmd/pilot-agent/status/util"
+	"istio.io/istio/pilot/pkg/model"
 )
 
 // Probe for readiness.
 type Probe struct {
+	ApplicationPorts    []uint16
 	LocalHostAddr       string
+	NodeType            model.NodeType
 	AdminPort           uint16
 	receivedFirstUpdate bool
-	ApplicationPorts    []uint16
-	NodeType            model.NodeType
 }
 
 // Check executes the probe and returns an error if the probe fails.
@@ -42,16 +40,19 @@ func (p *Probe) Check() error {
 		return err
 	}
 
-	// Envoy has received some configuration, make sure that configuration has been received for
-	// all inbound ports.
-	if err := p.checkInboundConfigured(); err != nil {
-		return err
+	// Envoy has received some listener configuration, make sure that configuration has been received for
+	// any of the inbound ports.
+	if p.NodeType == model.Router {
+		if err := p.checkInboundConfigured(); err != nil {
+			return err
+		}
 	}
 
 	return p.checkServerInfo()
 }
 
 // checkApplicationPorts verifies that Envoy has received configuration for all ports exposed by the application container.
+// Notes it is used only by envoy in Router mode.
 func (p *Probe) checkInboundConfigured() error {
 	if len(p.ApplicationPorts) > 0 {
 		listeningPorts, listeners, err := util.GetInboundListeningPorts(p.LocalHostAddr, p.AdminPort, p.NodeType)

pilot/cmd/pilot-agent/status/ready/probe_test.go

@@ -21,15 +21,35 @@ import (
 	"testing"
 
 	admin "github.com/envoyproxy/go-control-plane/envoy/admin/v2alpha"
+	envoyapicore "github.com/envoyproxy/go-control-plane/envoy/api/v2/core"
 	"github.com/gogo/protobuf/jsonpb"
 	"github.com/gogo/protobuf/proto"
 	. "github.com/onsi/gomega"
+
+	"istio.io/istio/pilot/pkg/model"
+	networking "istio.io/istio/pilot/pkg/networking/core/v1alpha3"
 )
 
 var (
 	goodStats      = "cluster_manager.cds.update_success: 1\nlistener_manager.lds.update_success: 1"
 	liveServerInfo = &admin.ServerInfo{State: admin.ServerInfo_LIVE}
 	initServerInfo = &admin.ServerInfo{State: admin.ServerInfo_INITIALIZING}
+	listeners      = admin.Listeners{
+		ListenerStatuses: []*admin.ListenerStatus{
+			{
+				Name: networking.VirtualInboundListenerName,
+				LocalAddress: &envoyapicore.Address{
+					Address: &envoyapicore.Address_SocketAddress{
+						SocketAddress: &envoyapicore.SocketAddress{
+							PortSpecifier: &envoyapicore.SocketAddress_PortValue{
+								PortValue: 15006,
+							},
+						},
+					},
+				},
+			},
+		},
+	}
 )
 
 func TestEnvoyStatsCompleteAndSuccessful(t *testing.T) {
@@ -164,19 +184,53 @@ func TestEnvoyInitializing(t *testing.T) {
 	g.Expect(err).To(HaveOccurred())
 }
 
-func createAndStartServer(statsToReturn string, serverInfo proto.Message) *httptest.Server {
-	mux := http.NewServeMux()
-	mux.HandleFunc("/stats", http.HandlerFunc(func(rw http.ResponseWriter, req *http.Request) {
-		// Send response to be tested
-		rw.Write([]byte(statsToReturn))
-	}))
-	mux.HandleFunc("/server_info", http.HandlerFunc(func(rw http.ResponseWriter, req *http.Request) {
+func TestEnvoyInitializingWithVirtualInboundListener(t *testing.T) {
+	g := NewGomegaWithT(t)
+
+	funcMap := createDefaultFuncMap(goodStats, liveServerInfo)
+
+	funcMap["/listeners"] = func(rw http.ResponseWriter, _ *http.Request) {
 		jsonm := &jsonpb.Marshaler{Indent: "  "}
-		infoJSON, _ := jsonm.MarshalToString(serverInfo)
+		listenerBytes, _ := jsonm.MarshalToString(&listeners)
 
 		// Send response to be tested
-		rw.Write([]byte(infoJSON))
-	}))
+		rw.Write([]byte(listenerBytes))
+	}
+
+	server := createHTTPServer(funcMap)
+	defer server.Close()
+	probe := Probe{AdminPort: 1234, receivedFirstUpdate: true, NodeType: model.SidecarProxy}
+
+	err := probe.Check()
+
+	g.Expect(err).ToNot(HaveOccurred())
+}
+
+func createDefaultFuncMap(statsToReturn string, serverInfo proto.Message) map[string]func(rw http.ResponseWriter, _ *http.Request) {
+	return map[string]func(rw http.ResponseWriter, _ *http.Request){
+
+		"/stats": func(rw http.ResponseWriter, _ *http.Request) {
+			// Send response to be tested
+			rw.Write([]byte(statsToReturn))
+		},
+		"/server_info": func(rw http.ResponseWriter, _ *http.Request) {
+			jsonm := &jsonpb.Marshaler{Indent: "  "}
+			infoJSON, _ := jsonm.MarshalToString(serverInfo)
+
+			// Send response to be tested
+			rw.Write([]byte(infoJSON))
+		},
+	}
+}
+func createAndStartServer(statsToReturn string, serverInfo proto.Message) *httptest.Server {
+	return createHTTPServer(createDefaultFuncMap(statsToReturn, serverInfo))
+}
+
+func createHTTPServer(handlers map[string]func(rw http.ResponseWriter, _ *http.Request)) *httptest.Server {
+	mux := http.NewServeMux()
+	for k, v := range handlers {
+		mux.HandleFunc(k, http.HandlerFunc(v))
+	}
 
 	// Start a local HTTP server
 	server := httptest.NewUnstartedServer(mux)

pilot/cmd/pilot-agent/status/server.go

@@ -63,13 +63,13 @@ type KubeAppProbers map[string]*corev1.HTTPGetAction
 
 // Config for the status server.
 type Config struct {
-	LocalHostAddr    string
-	StatusPort       uint16
-	AdminPort        uint16
 	ApplicationPorts []uint16
+	LocalHostAddr    string
 	// KubeAppHTTPProbers is a json with Kubernetes application HTTP prober config encoded.
 	KubeAppHTTPProbers string
 	NodeType           model.NodeType
+	StatusPort         uint16
+	AdminPort          uint16
 }
 
 // Server provides an endpoint for handling status probes.

pilot/cmd/pilot-agent/status/util/listeners.go

@@ -21,7 +21,7 @@ import (
 
 	admin "github.com/envoyproxy/go-control-plane/envoy/admin/v2alpha"
 	"github.com/gogo/protobuf/jsonpb"
-	multierror "github.com/hashicorp/go-multierror"
+	"github.com/hashicorp/go-multierror"
 
 	"istio.io/istio/pilot/pkg/model"
 )
@@ -69,6 +69,23 @@ func GetInboundListeningPorts(localHostAddr string, adminPort uint16, nodeType m
 	return ports, buf.String(), nil
 }
 
+func HasListenerName(localHostAddr string, adminPort uint16, name string) error {
+	buf, err := doHTTPGet(fmt.Sprintf("http://%s:%d/listeners?format=json", localHostAddr, adminPort))
+	if err != nil {
+		return multierror.Prefix(err, "failed retrieving Envoy listeners:")
+	}
+	listeners := &admin.Listeners{}
+	if err := jsonpb.Unmarshal(buf, listeners); err != nil {
+		return fmt.Errorf("failed parsing Envoy listeners %s: %s", buf, err)
+	}
+	for _, l := range listeners.ListenerStatuses {
+		if l.Name == name {
+			return nil
+		}
+	}
+	return fmt.Errorf("no such listener name %s", name)
+}
+
 func isLocalListener(l string) bool {
 	for _, ipPrefix := range ipPrefixes {
 		// In case of IPv6 address, it always comes in "[]", remove them so HasPrefix would work