Fix MCP dial-out mode. (istio#13399)

* Fix MCP dial-out mode.

+ The MCP dial-out mode sends an initial trigger response to trigger proper server/client communication. This is needed under certain scenarios. The original code expected a NACK response to this using a synchronous wait. However, this caused problems as the NACK can be sent *after* the actual resource requests are enqueued in the gRPC stream. This PR fixes the issue by making the handling of the trigger response in-line, as part of regular stream handling.
+ Adding a new dial-out integration tests capturing the basic scenario.
+ Adding a sleep in the Galley integration component, as the component startup is inherently racy. There is a race between setting the os signal event handlers during startup and applicatrion of configuration (and subsequent event trigger). The stop-gap solution is to sleep. The right solution is to go back to the correct ordering model for the startup of Galley.

* Add an explicit name to the trigger collection to avoid collisions.

* Fix lint issues.

* Fix lint issues.

* Remove failing test case.

* Update code coverage.

Author: ozevren

codecov.threshold

@@ -27,6 +27,7 @@ istio.io/istio/pilot/pkg/proxy/envoy/v2=10
 istio.io/istio/pilot/pkg/proxy/envoy/v2/rds.go=25
 istio.io/istio/pilot/pkg/serviceregistry/consul/monitor.go=20
 istio.io/istio/pkg/mcp/creds/watcher.go=100
+istio.io/istio/pkg/mcp/source/client_source.go=90
 istio.io/istio/pkg/mcp/source/source.go=90
 istio.io/istio/security/pkg/nodeagent=15
 istio.io/istio/istioctl/cmd/istioctl/kubeinject.go=41

pkg/appsignals/watcher.go

@@ -64,10 +64,13 @@ func Watch(c chan<- Signal) {
 func Notify(trigger string, signal os.Signal) {
 	handlers.Lock()
 	defer handlers.Unlock()
+	log.Infof("watcher.Notify: (trigger: %q, signal: %v)", trigger, signal)
 	for _, v := range handlers.listeners {
+		log.Debugf("watcher.Notify: Dispatching to listener '%v' (trigger: %q, signal: %v)", v, trigger, signal)
 		select {
 		case v <- Signal{trigger, signal}:
 		default:
+			log.Warnf("watcher.Notify: Signal channel is full (trigger: %q, signal: %v)", trigger, signal)
 		}
 	}
 }

pkg/mcp/sink/client_sink.go

@@ -41,6 +41,7 @@ type Client struct {
 	reporter monitoring.Reporter
 }
 
+// NewClient returns a new instance of Client.
 func NewClient(client mcp.ResourceSourceClient, options *Options) *Client {
 	return &Client{
 		Sink:     New(options),

pkg/mcp/source/client_source.go

@@ -16,21 +16,21 @@ package source
 
 import (
 	"context"
-	"fmt"
 	"io"
 	"time"
 
-	"google.golang.org/grpc/codes"
-
 	"github.com/gogo/status"
+	"google.golang.org/grpc/codes"
 
 	mcp "istio.io/api/mcp/v1alpha1"
+
 	"istio.io/istio/pkg/mcp/monitoring"
 )
 
 var (
 	// try to re-establish the bi-directional grpc stream after this delay.
 	reestablishStreamDelay = time.Second
+	triggerCollection      = "$triggerCollection"
 )
 
 // Client implements the client for the MCP sink service. The client is the
@@ -44,6 +44,7 @@ type Client struct {
 	source   *Source
 }
 
+// NewClient returns a new instance of Client.
 func NewClient(client mcp.ResourceSinkClient, options *Options) *Client {
 	return &Client{
 		source:   New(options),
@@ -59,31 +60,23 @@ var reconnectTestProbe = func() {}
 // trigger response which we expect the server to NACK.
 func (c *Client) sendTriggerResponse(stream Stream) error {
 	trigger := &mcp.Resources{
-		Collection: "", // unimplemented collection
+		Collection: triggerCollection,
 	}
 
 	if err := stream.Send(trigger); err != nil {
 		return status.Errorf(status.Code(err), "could not send trigger request %v", err)
 	}
 
-	msg, err := stream.Recv()
-	if err != nil {
-		return status.Errorf(status.Code(err),
-			"could not receive expected nack response: %v", err)
-	}
-
-	if msg.ErrorDetail == nil {
-		return fmt.Errorf("server should have nacked, did not get an error")
-	}
-	errCode := codes.Code(msg.ErrorDetail.Code)
-	if errCode != codes.Unimplemented {
-		return fmt.Errorf("server should have nacked with code=%v: got %v",
-			codes.Unimplemented, errCode)
-	}
-
 	return nil
 }
 
+// isTriggerResponse checks whether the given RequestResources object is an expected NACK response to a previous
+// trigger message.
+func isTriggerResponse(msg *mcp.RequestResources) bool {
+	return msg.Collection == triggerCollection && msg.ErrorDetail != nil && codes.Code(msg.ErrorDetail.Code) == codes.Unimplemented
+}
+
+// Run implements mcpClient
 func (c *Client) Run(ctx context.Context) {
 	// The first attempt is immediate.
 	retryDelay := time.Nanosecond

pkg/mcp/source/client_source_test.go

@@ -85,7 +85,7 @@ func TestClientSource(t *testing.T) {
 		Resources:  []*mcp.Resource{test.Type2A[0].Resource},
 	})
 
-	h.requestsChan <- test.MakeRequest(false, "", "", codes.Unimplemented)
+	h.requestsChan <- test.MakeRequest(false, triggerCollection, "", codes.Unimplemented)
 	h.requestsChan <- test.MakeRequest(false, test.FakeType0Collection, "", codes.OK)
 	h.requestsChan <- test.MakeRequest(false, test.FakeType1Collection, "", codes.OK)
 	h.requestsChan <- test.MakeRequest(false, test.FakeType2Collection, "", codes.OK)
@@ -124,11 +124,6 @@ func TestClientSource(t *testing.T) {
 	h.requestsChan <- test.MakeRequest(false, "", "", codes.OK)
 	proceed <- true
 
-	<-waiting
-	h.client = true
-	h.requestsChan <- test.MakeRequest(false, "", "", codes.InvalidArgument)
-	proceed <- true
-
 	<-waiting
 	h.setRecvError(errors.New("fake recv error"))
 	h.client = true

pkg/mcp/source/source.go

@@ -399,6 +399,10 @@ func (con *connection) close() {
 }
 
 func (con *connection) processClientRequest(req *mcp.RequestResources) error {
+	if isTriggerResponse(req) {
+		return nil
+	}
+
 	collection := req.Collection
 
 	con.reporter.RecordRequestSize(collection, con.id, internal.ProtoSize(req))

pkg/test/framework/components/galley/galley.go

@@ -51,8 +51,12 @@ type Instance interface {
 	WaitForSnapshot(collection string, validator SnapshotValidatorFunc) error
 }
 
-// Configuration for Galley
+// Config for Galley
 type Config struct {
+
+	// SinkAddress to dial-out to, if set.
+	SinkAddress string
+
 	// MeshConfig to use for this instance.
 	MeshConfig string
 }

pkg/test/framework/components/galley/native.go

@@ -49,7 +49,7 @@ const (
 	meshConfigFile = "meshconfig.yaml"
 )
 
-// NewNativeComponent factory function for the component
+// newNative returns the native implementation of galley.Instance.
 func newNative(ctx resource.Context, cfg Config) (Instance, error) {
 
 	n := &nativeComponent{
@@ -337,6 +337,11 @@ func (c *nativeComponent) restart() error {
 	// Bind to an arbitrary port.
 	a.APIAddress = "tcp://0.0.0.0:0"
 
+	if c.cfg.SinkAddress != "" {
+		a.SinkAddress = c.cfg.SinkAddress
+		a.SinkAuthMode = "NONE"
+	}
+
 	s, err := server.New(a)
 	if err != nil {
 		scopes.Framework.Errorf("Error starting Galley: %v", err)
@@ -347,6 +352,10 @@ func (c *nativeComponent) restart() error {
 
 	go s.Run()
 
+	// TODO: This is due to Galley start-up being racy. We should go back to the "Start" based model where
+	// return from s.Start() guarantees that all the setup is complete.
+	time.Sleep(time.Second)
+
 	c.client = &client{
 		address: fmt.Sprintf("tcp://%s", s.Address().String()),
 	}

pkg/test/framework/components/mcpserver/mcpserver.go

@@ -0,0 +1,52 @@
+// Copyright 2019 Istio Authors
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package mcpserver
+
+import (
+	"testing"
+
+	"istio.io/istio/pkg/mcp/sink"
+	"istio.io/istio/pkg/test/framework/components/environment"
+	"istio.io/istio/pkg/test/framework/resource"
+)
+
+// Instance is a new mcpserver instance. MCP Server is a generic MCP server implementation for testing purposes.
+type Instance interface {
+	Address() string
+	GetCollectionStateOrFail(t *testing.T, collection string) []*sink.Object
+}
+
+// SinkConfig is configuration for the mcpserver for sink mode.
+type SinkConfig struct {
+	Collections []string
+}
+
+// NewSink returns a new instance of MCP Server in Sink mode.
+func NewSink(ctx resource.Context, cfg SinkConfig) (i Instance, err error) {
+	err = resource.UnsupportedEnvironment(ctx.Environment())
+	ctx.Environment().Case(environment.Native, func() {
+		i, err = newSinkNative(ctx, cfg)
+	})
+	return
+}
+
+// NewSinkOrFail returns a new instance of MCP server in Sink mode or fails.
+func NewSinkOrFail(t *testing.T, c resource.Context, cfg SinkConfig) Instance {
+	i, err := NewSink(c, cfg)
+	if err != nil {
+		t.Fatalf("mcpserver.NewOrFail: %v", err)
+	}
+	return i
+}

pkg/test/framework/components/mcpserver/native.go

@@ -0,0 +1,102 @@
+// Copyright 2019 Istio Authors
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package mcpserver
+
+import (
+	"io"
+	"net"
+	"testing"
+	"time"
+
+	"google.golang.org/grpc"
+
+	mcp "istio.io/api/mcp/v1alpha1"
+	"istio.io/istio/pkg/mcp/rate"
+	"istio.io/istio/pkg/mcp/server"
+	"istio.io/istio/pkg/mcp/sink"
+	"istio.io/istio/pkg/mcp/testing/monitoring"
+	"istio.io/istio/pkg/test/framework/resource"
+	"istio.io/istio/pkg/test/scopes"
+)
+
+type native struct {
+	id resource.ID
+	l  net.Listener
+	s  *grpc.Server
+	u  *sink.InMemoryUpdater
+}
+
+var _ Instance = &native{}
+var _ resource.Resource = &native{}
+var _ io.Closer = &native{}
+
+func newSinkNative(ctx resource.Context, cfg SinkConfig) (*native, error) {
+	n := &native{}
+	n.id = ctx.TrackResource(n)
+	u := sink.NewInMemoryUpdater()
+
+	so := sink.Options{
+		ID:                "mcpserver.sink",
+		CollectionOptions: sink.CollectionOptionsFromSlice(cfg.Collections),
+		Updater:           u,
+		Reporter:          monitoring.NewInMemoryStatsContext(),
+	}
+
+	l, err := net.Listen("tcp", "0.0.0.0:0")
+	if err != nil {
+		return nil, err
+	}
+
+	s := grpc.NewServer()
+	srv := sink.NewServer(&so, &sink.ServerOptions{
+		AuthChecker: &server.AllowAllChecker{},
+		RateLimiter: rate.NewRateLimiter(time.Millisecond, 1000).Create(),
+	})
+
+	mcp.RegisterResourceSinkServer(s, srv)
+
+	go func() {
+		if err := s.Serve(l); err != nil {
+			scopes.Framework.Errorf("mcpserver.Serve: %v", err)
+		}
+	}()
+
+	n.l = l
+	n.s = s
+	n.u = u
+
+	return n, nil
+}
+
+// ID implements resource.Resource
+func (n *native) ID() resource.ID {
+	return n.id
+}
+
+// Address implements Instance
+func (n *native) Address() string {
+	return n.l.Addr().String()
+}
+
+// GetCollectionStateOrFail implements Instance
+func (n *native) GetCollectionStateOrFail(t *testing.T, collection string) []*sink.Object {
+	return n.u.Get(collection)
+}
+
+// Close implements io.Closer
+func (n *native) Close() error {
+	n.s.Stop()
+	return nil
+}