istio

1
// Copyright Istio Authors
2
//
3
// Licensed under the Apache License, Version 2.0 (the "License");
4
// you may not use this file except in compliance with the License.
5
// You may obtain a copy of the License at
6
//
7
//     http://www.apache.org/licenses/LICENSE-2.0
8
//
9
// Unless required by applicable law or agreed to in writing, software
10
// distributed under the License is distributed on an "AS IS" BASIS,
11
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
12
// See the License for the specific language governing permissions and
13
// limitations under the License.
14

15
package controller
16

17
import (
18
	"time"
19

20
	corev1 "k8s.io/api/core/v1"
21
	"k8s.io/apimachinery/pkg/runtime/schema"
22

23
	meshconfig "istio.io/api/mesh/v1alpha1"
24
	"istio.io/istio/pilot/pkg/model"
25
	"istio.io/istio/pilot/pkg/serviceregistry/aggregate"
26
	"istio.io/istio/pilot/pkg/serviceregistry/util/xdsfake"
27
	"istio.io/istio/pkg/cluster"
28
	"istio.io/istio/pkg/config/mesh"
29
	kubelib "istio.io/istio/pkg/kube"
30
	"istio.io/istio/pkg/kube/kclient"
31
	"istio.io/istio/pkg/kube/kclient/clienttest"
32
	"istio.io/istio/pkg/kube/namespace"
33
	"istio.io/istio/pkg/queue"
34
	"istio.io/istio/pkg/test"
35
	"istio.io/istio/pkg/test/util/assert"
36
)
37

38
const (
39
	defaultFakeDomainSuffix = "company.com"
40
)
41

42
type FakeControllerOptions struct {
43
	Client            kubelib.Client
44
	CRDs              []schema.GroupVersionResource
45
	NetworksWatcher   mesh.NetworksWatcher
46
	MeshWatcher       mesh.Watcher
47
	ServiceHandler    model.ServiceHandler
48
	ClusterID         cluster.ID
49
	WatchedNamespaces string
50
	DomainSuffix      string
51
	XDSUpdater        model.XDSUpdater
52
	Stop              chan struct{}
53
	SkipRun           bool
54
	ConfigCluster     bool
55
	SystemNamespace   string
56
}
57

58
type FakeController struct {
59
	*Controller
60
	Endpoints *model.EndpointIndex
61
}
62

63
func NewFakeControllerWithOptions(t test.Failer, opts FakeControllerOptions) (*FakeController, *xdsfake.Updater) {
64
	xdsUpdater := opts.XDSUpdater
65
	var endpoints *model.EndpointIndex
66
	if xdsUpdater == nil {
67
		endpoints = model.NewEndpointIndex(model.DisabledCache{})
68
		delegate := model.NewEndpointIndexUpdater(endpoints)
69
		xdsUpdater = xdsfake.NewWithDelegate(delegate)
70
	}
71

72
	domainSuffix := defaultFakeDomainSuffix
73
	if opts.DomainSuffix != "" {
74
		domainSuffix = opts.DomainSuffix
75
	}
76
	if opts.Client == nil {
77
		opts.Client = kubelib.NewFakeClient()
78
	}
79
	if opts.MeshWatcher == nil {
80
		opts.MeshWatcher = mesh.NewFixedWatcher(&meshconfig.MeshConfig{})
81
	}
82
	cleanupStop := false
83
	stop := opts.Stop
84
	if stop == nil {
85
		// If we created the stop, clean it up. Otherwise, caller is responsible
86
		cleanupStop = true
87
		stop = make(chan struct{})
88
	}
89
	f := namespace.NewDiscoveryNamespacesFilter(
90
		kclient.New[*corev1.Namespace](opts.Client),
91
		opts.MeshWatcher,
92
		stop,
93
	)
94
	kubelib.SetObjectFilter(opts.Client, f)
95

96
	meshServiceController := aggregate.NewController(aggregate.Options{MeshHolder: opts.MeshWatcher})
97

98
	options := Options{
99
		DomainSuffix:          domainSuffix,
100
		XDSUpdater:            xdsUpdater,
101
		Metrics:               &model.Environment{},
102
		MeshNetworksWatcher:   opts.NetworksWatcher,
103
		MeshWatcher:           opts.MeshWatcher,
104
		ClusterID:             opts.ClusterID,
105
		MeshServiceController: meshServiceController,
106
		ConfigCluster:         opts.ConfigCluster,
107
		SystemNamespace:       opts.SystemNamespace,
108
	}
109
	c := NewController(opts.Client, options)
110
	meshServiceController.AddRegistry(c)
111

112
	if opts.ServiceHandler != nil {
113
		c.AppendServiceHandler(opts.ServiceHandler)
114
	}
115

116
	t.Cleanup(func() {
117
		c.client.Shutdown()
118
	})
119
	if !opts.SkipRun {
120
		t.Cleanup(func() {
121
			assert.NoError(t, queue.WaitForClose(c.queue, time.Second*5))
122
		})
123
	}
124
	c.stop = stop
125
	if cleanupStop {
126
		t.Cleanup(func() {
127
			close(stop)
128
		})
129
	}
130
	for _, crd := range opts.CRDs {
131
		clienttest.MakeCRD(t, c.client, crd)
132
	}
133
	opts.Client.RunAndWait(c.stop)
134
	var fx *xdsfake.Updater
135
	if x, ok := xdsUpdater.(*xdsfake.Updater); ok {
136
		fx = x
137
	}
138

139
	if !opts.SkipRun {
140
		go c.Run(c.stop)
141
		kubelib.WaitForCacheSync("test", c.stop, c.HasSynced)
142
	}
143

144
	return &FakeController{Controller: c, Endpoints: endpoints}, fx
145
}
146