Merge remote-tracking branch 'origin/release-1.1' into r11tomv2

Gopkg.toml

@@ -111,7 +111,7 @@ ignored = [
 
 [[constraint]]
   name = "github.com/envoyproxy/go-control-plane"
-  version = "^0.6.1"
+  version = "^0.6.2"
 
 [[constraint]]
   name = "github.com/fluent/fluent-logger-golang"

Makefile

@@ -173,7 +173,7 @@ export ISTIO_ENVOY_RELEASE_DIR ?= ${OUT_DIR}/${GOOS}_${GOARCH}/release
 export ISTIO_ENVOY_RELEASE_NAME ?= envoy-${ISTIO_ENVOY_VERSION}
 export ISTIO_ENVOY_RELEASE_PATH ?= ${ISTIO_ENVOY_RELEASE_DIR}/${ISTIO_ENVOY_RELEASE_NAME}
 
-GO_VERSION_REQUIRED:=1.9
+GO_VERSION_REQUIRED:=1.10
 
 HUB?=istio
 ifeq ($(HUB),)
@@ -276,7 +276,7 @@ depend.diff: $(ISTIO_OUT)
 # Used by CI for automatic go code generation and generates a git diff of the generated files against HEAD.
 go.generate.diff: $(ISTIO_OUT)
 	git diff HEAD > $(ISTIO_OUT)/before_go_generate.diff
-	-go generate ./... 
+	-go generate ./...
 	git diff HEAD > $(ISTIO_OUT)/after_go_generate.diff
 	diff $(ISTIO_OUT)/before_go_generate.diff $(ISTIO_OUT)/after_go_generate.diff
 

galley/cmd/galley/cmd/root.go

@@ -138,6 +138,13 @@ func GetRootCmd(args []string, printf, fatalf shared.FormatFn) *cobra.Command {
 		"The access list yaml file that contains the allowd mTLS peer ids.")
 	rootCmd.PersistentFlags().StringVar(&serverArgs.ConfigPath, "configPath", serverArgs.ConfigPath,
 		"Istio config file path")
+	rootCmd.PersistentFlags().StringVar(&serverArgs.MeshConfigFile, "meshConfigFile", serverArgs.MeshConfigFile,
+		"Path to the mesh config file")
+	rootCmd.PersistentFlags().StringVar(&serverArgs.DomainSuffix, "domain", serverArgs.DomainSuffix,
+		"DNS domain suffix")
+	rootCmd.PersistentFlags().BoolVar(&serverArgs.DisableResourceReadyCheck, "disableResourceReadyCheck", serverArgs.DisableResourceReadyCheck,
+		"Disable resource readiness checks. This allows Galley to start if not all resource types are supported")
+
 	serverArgs.IntrospectionOptions.AttachCobraFlags(rootCmd)
 
 	//validation config

galley/pkg/fs/fssource.go

@@ -26,6 +26,8 @@ import (
 	"github.com/howeyc/fsnotify"
 	"k8s.io/apimachinery/pkg/apis/meta/v1/unstructured"
 
+	"istio.io/istio/galley/pkg/kube/converter"
+
 	"istio.io/istio/pkg/log"
 
 	"istio.io/istio/galley/pkg/kube"
@@ -41,34 +43,37 @@ var supportedExtensions = map[string]bool{
 }
 var scope = log.RegisterScope("fs", "File system source debugging", 0)
 
-//fsSource is source implementation for filesystem.
+// fsSource is source implementation for filesystem.
 type fsSource struct {
-	//Config File Path
+	// configuration for the converters.
+	config *converter.Config
+
+	// Config File Path
 	root string
 
 	donec chan struct{}
 
 	mu sync.RWMutex
 
-	//map to store namespace/name : shas
-	shas map[string][sha1.Size]byte
+	// map to store namespace/name : shas
+	shas map[resource.FullName][sha1.Size]byte
 
 	ch chan resource.Event
 
 	// map to store kind : bool to indicate whether we need to deal with the resource or not
 	kinds map[string]bool
 
-	//map to store filename: []{namespace/name,kind} to indicate whether the resources has been deleted from one file
+	// map to store filename: []{namespace/name,kind} to indicate whether the resources has been deleted from one file
 	fileResorceKeys map[string][]*fileResourceKey
 
-	//fsresource version
+	// fsresource version
 	version int64
 
 	watcher *fsnotify.Watcher
 }
 
-func (s *fsSource) readFiles(root string) map[string]*istioResource {
-	results := map[string]*istioResource{}
+func (s *fsSource) readFiles(root string) map[resource.FullName]*istioResource {
+	results := map[resource.FullName]*istioResource{}
 
 	err := filepath.Walk(root, func(path string, info os.FileInfo, err error) error {
 		if err != nil {
@@ -92,8 +97,8 @@ func (s *fsSource) readFiles(root string) map[string]*istioResource {
 	return results
 }
 
-func (s *fsSource) readFile(path string, info os.FileInfo, initial bool) map[string]*istioResource {
-	result := map[string]*istioResource{}
+func (s *fsSource) readFile(path string, info os.FileInfo, initial bool) map[resource.FullName]*istioResource {
+	result := map[resource.FullName]*istioResource{}
 	if mode := info.Mode() & os.ModeType; !supportedExtensions[filepath.Ext(path)] || (mode != 0 && mode != os.ModeSymlink) {
 		return nil
 	}
@@ -119,7 +124,7 @@ func (s *fsSource) readFile(path string, info os.FileInfo, initial bool) map[str
 }
 
 //process delete part of the resources in a file
-func (s *fsSource) processPartialDelete(fileName string, newData *map[string]*istioResource) {
+func (s *fsSource) processPartialDelete(fileName string, newData *map[resource.FullName]*istioResource) {
 	s.mu.Lock()
 	defer s.mu.Unlock()
 	if fileResorceKeys, ok := s.fileResorceKeys[fileName]; ok {
@@ -142,7 +147,7 @@ func (s *fsSource) processPartialDelete(fileName string, newData *map[string]*is
 
 }
 
-func (s *fsSource) processAddOrUpdate(fileName string, newData *map[string]*istioResource) {
+func (s *fsSource) processAddOrUpdate(fileName string, newData *map[resource.FullName]*istioResource) {
 	s.mu.Lock()
 	defer s.mu.Unlock()
 	// need versionUpdated as sometimes when fswatcher fires events, there is actually no change on the file content
@@ -199,6 +204,7 @@ func (s *fsSource) initialCheck() {
 	newData := s.readFiles(s.root)
 	s.mu.Lock()
 	defer s.mu.Unlock()
+	s.ch <- resource.Event{Kind: resource.FullSync}
 	for k, r := range newData {
 		s.process(resource.Added, k, "", r)
 		s.shas[k] = r.sha
@@ -211,7 +217,7 @@ func (s *fsSource) Stop() {
 	s.watcher.Close()
 }
 
-func (s *fsSource) process(eventKind resource.EventKind, key, resourceKind string, r *istioResource) {
+func (s *fsSource) process(eventKind resource.EventKind, key resource.FullName, resourceKind string, r *istioResource) {
 	var u *unstructured.Unstructured
 	var spec kube.ResourceSpec
 	var kind string
@@ -229,7 +235,8 @@ func (s *fsSource) process(eventKind resource.EventKind, key, resourceKind strin
 			break
 		}
 	}
-	source.ProcessEvent(spec, eventKind, key, fmt.Sprintf("v%d", s.version), u, s.ch)
+
+	source.ProcessEvent(s.config, spec, eventKind, key, fmt.Sprintf("v%d", s.version), u, s.ch)
 }
 
 // Start implements runtime.Source
@@ -295,17 +302,18 @@ func (s *fsSource) Start() (chan resource.Event, error) {
 }
 
 // New returns a File System implementation of runtime.Source.
-func New(root string) (runtime.Source, error) {
-	return newFsSource(root, kube_meta.Types.All())
+func New(root string, config *converter.Config) (runtime.Source, error) {
+	return newFsSource(root, config, kube_meta.Types.All())
 }
 
-func newFsSource(root string, specs []kube.ResourceSpec) (runtime.Source, error) {
+func newFsSource(root string, config *converter.Config, specs []kube.ResourceSpec) (runtime.Source, error) {
 	fs := &fsSource{
+		config:          config,
 		root:            root,
 		kinds:           map[string]bool{},
 		fileResorceKeys: map[string][]*fileResourceKey{},
 		donec:           make(chan struct{}),
-		shas:            map[string][sha1.Size]byte{},
+		shas:            map[resource.FullName][sha1.Size]byte{},
 		version:         0,
 	}
 	for _, spec := range specs {

galley/pkg/fs/fssource_test.go

@@ -23,7 +23,11 @@ import (
 	"testing"
 	"time"
 
+	"istio.io/istio/galley/pkg/kube/converter"
+	"istio.io/istio/galley/pkg/meshconfig"
+	"istio.io/istio/galley/pkg/runtime"
 	"istio.io/istio/galley/pkg/runtime/resource"
+	sn "istio.io/istio/pkg/mcp/snapshot"
 )
 
 var mixerYAML = `
@@ -157,7 +161,7 @@ func TestNewSource(t *testing.T) {
 	}
 	fst.testSetup(t)
 	defer fst.testTeardown(t)
-	s, err := New(fst.rootPath)
+	s, err := New(fst.rootPath, &converter.Config{Mesh: meshconfig.NewInMemory()})
 	if err != nil {
 		t.Fatalf("Unexpected error found: %v", err)
 	}
@@ -172,7 +176,7 @@ func TestFsSource_InitialScan(t *testing.T) {
 	}
 	fst.testSetup(t)
 	defer fst.testTeardown(t)
-	s, err := New(fst.rootPath)
+	s, err := New(fst.rootPath, &converter.Config{Mesh: meshconfig.NewInMemory()})
 	if err != nil {
 		t.Fatalf("Unexpected error: %v", err)
 	}
@@ -186,7 +190,7 @@ func TestFsSource_InitialScan(t *testing.T) {
 		t.Fatalf("Unexpected error: %v", err)
 	}
 
-	log := logChannelOutput(ch, 1)
+	log := logChannelOutput(ch, 2)
 	expected := strings.TrimSpace(`
 	[Event](Added: [VKey](type.googleapis.com/istio.networking.v1alpha3.VirtualService:route-for-myapp @v0))`)
 	if log != expected {
@@ -202,7 +206,7 @@ func TestFsSource_AddFile(t *testing.T) {
 	}
 	fst.testSetup(t)
 	defer fst.testTeardown(t)
-	s, err := New(fst.rootPath)
+	s, err := New(fst.rootPath, &converter.Config{Mesh: meshconfig.NewInMemory()})
 	if err != nil {
 		t.Fatalf("Unexpected error: %v", err)
 	}
@@ -222,7 +226,7 @@ func TestFsSource_AddFile(t *testing.T) {
 		t.Fatalf("Unexpected error: %v", err)
 	}
 
-	log := logChannelOutput(ch, 1)
+	log := logChannelOutput(ch, 2)
 	expected := strings.TrimSpace(`
 	[Event](Added: [VKey](type.googleapis.com/istio.networking.v1alpha3.VirtualService:route-for-myapp @v1))`)
 
@@ -240,7 +244,7 @@ func TestFsSource_DeleteFile(t *testing.T) {
 	}
 	fst.testSetup(t)
 	defer fst.testTeardown(t)
-	s, err := New(fst.rootPath)
+	s, err := New(fst.rootPath, &converter.Config{Mesh: meshconfig.NewInMemory()})
 	if err != nil {
 		t.Fatalf("Unexpected error: %v", err)
 	}
@@ -258,7 +262,7 @@ func TestFsSource_DeleteFile(t *testing.T) {
 		t.Fatalf("Unexpected error: %v", err)
 	}
 
-	log := logChannelOutput(ch, 2)
+	log := logChannelOutput(ch, 3)
 	expected := strings.TrimSpace(`
 	[Event](Deleted: [VKey](type.googleapis.com/istio.networking.v1alpha3.VirtualService:route-for-myapp @v0))`)
 
@@ -275,7 +279,7 @@ func TestFsSource_ModifyFile(t *testing.T) {
 	}
 	fst.testSetup(t)
 	defer fst.testTeardown(t)
-	s, err := New(fst.rootPath)
+	s, err := New(fst.rootPath, &converter.Config{Mesh: meshconfig.NewInMemory()})
 	if err != nil {
 		t.Fatalf("Unexpected error: %v", err)
 	}
@@ -292,7 +296,7 @@ func TestFsSource_ModifyFile(t *testing.T) {
 	if err != nil {
 		t.Fatalf("Unexpected error: %v", err)
 	}
-	log := logChannelOutput(ch, 3)
+	log := logChannelOutput(ch, 4)
 	expected := strings.TrimSpace(`
 	[Event](Added: [VKey](type.googleapis.com/istio.networking.v1alpha3.VirtualService:route-for-myapp-changed @v1))`)
 	if log != expected {
@@ -308,7 +312,7 @@ func TestFsSource_DeletePartResorceInFile(t *testing.T) {
 	}
 	fst.testSetup(t)
 	defer fst.testTeardown(t)
-	s, err := New(fst.rootPath)
+	s, err := New(fst.rootPath, &converter.Config{Mesh: meshconfig.NewInMemory()})
 	if err != nil {
 		t.Fatalf("Unexpected error: %v", err)
 	}
@@ -339,6 +343,43 @@ func TestFsSource_DeletePartResorceInFile(t *testing.T) {
 	s.Stop()
 }
 
+func TestFsSource_publishEvent(t *testing.T) {
+	fst = &fsTestSourceState{
+		ConfigFiles: map[string][]byte{"virtual_service.yml": []byte(virtualServiceYAML)},
+	}
+	fst.testSetup(t)
+	defer fst.testTeardown(t)
+	s, err := New(fst.rootPath, &converter.Config{Mesh: meshconfig.NewInMemory()})
+	if err != nil {
+		t.Fatalf("Unexpected error: %v", err)
+	}
+
+	if s == nil {
+		t.Fatal("Expected non nil source")
+	}
+	d := runtime.NewInMemoryDistributor()
+	cfg := &runtime.Config{Mesh: meshconfig.NewInMemory()}
+	processor := runtime.NewProcessor(s, d, cfg)
+	err = processor.Start()
+	if err != nil {
+		t.Fatalf("Unexpected error: %v", err)
+	}
+	ch := make(chan bool)
+	listenerAction := func(sp sn.Snapshot) {
+		ch <- true
+	}
+	cancel := make(chan bool)
+	defer func() { close(cancel) }()
+	go d.ListenChanges(cancel, listenerAction)
+	select {
+	case <-ch:
+		return
+	case <-time.After(5 * time.Second):
+		t.Fatal("The snapshot should have been set")
+	}
+	processor.Stop()
+}
+
 //Only log the last event out
 func logChannelOutput(ch chan resource.Event, count int) string {
 	var result resource.Event