kuma

1
package envoyadmin_test
2

3
import (
4
	"context"
5

6
	. "github.com/onsi/ginkgo/v2"
7
	. "github.com/onsi/gomega"
8
	"google.golang.org/grpc"
9

10
	mesh_proto "github.com/kumahq/kuma/api/mesh/v1alpha1"
11
	system_proto "github.com/kumahq/kuma/api/system/v1alpha1"
12
	"github.com/kumahq/kuma/pkg/core/resources/apis/system"
13
	"github.com/kumahq/kuma/pkg/core/resources/manager"
14
	"github.com/kumahq/kuma/pkg/core/resources/model"
15
	core_store "github.com/kumahq/kuma/pkg/core/resources/store"
16
	"github.com/kumahq/kuma/pkg/envoy/admin"
17
	"github.com/kumahq/kuma/pkg/intercp/catalog"
18
	"github.com/kumahq/kuma/pkg/intercp/envoyadmin"
19
	"github.com/kumahq/kuma/pkg/plugins/resources/memory"
20
	"github.com/kumahq/kuma/pkg/test/resources/samples"
21
	"github.com/kumahq/kuma/pkg/test/runtime"
22
)
23

24
var _ = Describe("Forwarding KDS Client", func() {
25
	const thisInstanceID = "instance-1"
26
	const otherInstanceID = "instance-2"
27

28
	var forwardClient *countingForwardClient
29
	var adminClient *runtime.DummyEnvoyAdminClient
30
	var resManager manager.ResourceManager
31

32
	var forwardingClient admin.EnvoyAdminClient
33

34
	dp := samples.DataplaneBackendBuilder().
35
		WithName("east.backend").
36
		Build()
37

38
	BeforeEach(func() {
39
		forwardClient = &countingForwardClient{}
40
		adminClient = &runtime.DummyEnvoyAdminClient{}
41
		resManager = manager.NewResourceManager(memory.NewStore())
42
		cat := catalog.NewConfigCatalog(resManager)
43
		forwardingClient = envoyadmin.NewForwardingEnvoyAdminClient(
44
			resManager,
45
			cat,
46
			thisInstanceID,
47
			func(url string) (mesh_proto.InterCPEnvoyAdminForwardServiceClient, error) {
48
				return forwardClient, nil
49
			},
50
			adminClient,
51
		)
52

53
		_, err := cat.Replace(context.Background(), []catalog.Instance{
54
			{Id: thisInstanceID},
55
			{Id: otherInstanceID},
56
		})
57
		Expect(err).ToNot(HaveOccurred())
58

59
		Expect(samples.MeshDefaultBuilder().Create(resManager)).To(Succeed())
60
		err = resManager.Create(context.Background(), dp, core_store.CreateByKey("east.backend", "default"))
61
		Expect(err).ToNot(HaveOccurred())
62
	})
63

64
	createZoneInsightConnectedToGlobal := func(insight string, globalInstanceID string) {
65
		zoneInsight := system.NewZoneInsightResource()
66
		zoneInsight.Spec.EnvoyAdminStreams = &system_proto.EnvoyAdminStreams{
67
			ConfigDumpGlobalInstanceId: globalInstanceID,
68
			StatsGlobalInstanceId:      globalInstanceID,
69
			ClustersGlobalInstanceId:   globalInstanceID,
70
		}
71
		err := resManager.Create(context.Background(), zoneInsight, core_store.CreateByKey(insight, model.NoMesh))
72
		Expect(err).ToNot(HaveOccurred())
73
	}
74

75
	type testCase struct {
76
		globalInstanceID  string
77
		forwardedRequests int
78
		executedRequests  int
79
	}
80

81
	DescribeTable("when request for config dump is executed",
82
		func(given testCase) {
83
			// given
84
			createZoneInsightConnectedToGlobal("east", given.globalInstanceID)
85

86
			// when
87
			_, err := forwardingClient.ConfigDump(context.Background(), dp)
88

89
			// then
90
			Expect(err).ToNot(HaveOccurred())
91
			Expect(forwardClient.xdsConfigCalled).To(Equal(given.forwardedRequests))
92
			Expect(adminClient.ConfigDumpCalled).To(Equal(given.executedRequests))
93
		},
94
		Entry("should forward request when the zone of dp is connected to other instance of Global CP", testCase{
95
			globalInstanceID:  otherInstanceID,
96
			forwardedRequests: 1,
97
			executedRequests:  0,
98
		}),
99
		Entry("should execute request when the zone of dp is connected to this instance of Global CP", testCase{
100
			globalInstanceID:  thisInstanceID,
101
			forwardedRequests: 0,
102
			executedRequests:  1,
103
		}),
104
	)
105

106
	DescribeTable("when request for stats is executed",
107
		func(given testCase) {
108
			// given
109
			createZoneInsightConnectedToGlobal("east", given.globalInstanceID)
110

111
			// when
112
			_, err := forwardingClient.Stats(context.Background(), dp)
113

114
			// then
115
			Expect(err).ToNot(HaveOccurred())
116
			Expect(forwardClient.statsCalled).To(Equal(given.forwardedRequests))
117
			Expect(adminClient.StatsCalled).To(Equal(given.executedRequests))
118
		},
119
		Entry("should forward request when the zone of dp is connected to other instance of Global CP", testCase{
120
			globalInstanceID:  otherInstanceID,
121
			forwardedRequests: 1,
122
			executedRequests:  0,
123
		}),
124
		Entry("should execute request when the zone of dp is connected to this instance of Global CP", testCase{
125
			globalInstanceID:  thisInstanceID,
126
			forwardedRequests: 0,
127
			executedRequests:  1,
128
		}),
129
	)
130

131
	DescribeTable("when request for clusters is executed",
132
		func(given testCase) {
133
			// given
134
			createZoneInsightConnectedToGlobal("east", given.globalInstanceID)
135

136
			// when
137
			_, err := forwardingClient.Clusters(context.Background(), dp)
138

139
			// then
140
			Expect(err).ToNot(HaveOccurred())
141
			Expect(forwardClient.clustersCalled).To(Equal(given.forwardedRequests))
142
			Expect(adminClient.ClustersCalled).To(Equal(given.executedRequests))
143
		},
144
		Entry("should forward request when the zone of dp is connected to other instance of Global CP", testCase{
145
			globalInstanceID:  otherInstanceID,
146
			forwardedRequests: 1,
147
			executedRequests:  0,
148
		}),
149
		Entry("should execute request when the zone of dp is connected to this instance of Global CP", testCase{
150
			globalInstanceID:  thisInstanceID,
151
			forwardedRequests: 0,
152
			executedRequests:  1,
153
		}),
154
	)
155
})
156

157
type countingForwardClient struct {
158
	xdsConfigCalled int
159
	statsCalled     int
160
	clustersCalled  int
161
}
162

163
var _ mesh_proto.InterCPEnvoyAdminForwardServiceClient = &countingForwardClient{}
164

165
func (c *countingForwardClient) XDSConfig(ctx context.Context, in *mesh_proto.XDSConfigRequest, opts ...grpc.CallOption) (*mesh_proto.XDSConfigResponse, error) {
166
	c.xdsConfigCalled++
167
	return &mesh_proto.XDSConfigResponse{
168
		Result: &mesh_proto.XDSConfigResponse_Config{
169
			Config: []byte("forwarded"),
170
		},
171
	}, nil
172
}
173

174
func (c *countingForwardClient) Stats(ctx context.Context, in *mesh_proto.StatsRequest, opts ...grpc.CallOption) (*mesh_proto.StatsResponse, error) {
175
	c.statsCalled++
176
	return &mesh_proto.StatsResponse{
177
		Result: &mesh_proto.StatsResponse_Stats{
178
			Stats: []byte("forwarded"),
179
		},
180
	}, nil
181
}
182

183
func (c *countingForwardClient) Clusters(ctx context.Context, in *mesh_proto.ClustersRequest, opts ...grpc.CallOption) (*mesh_proto.ClustersResponse, error) {
184
	c.clustersCalled++
185
	return &mesh_proto.ClustersResponse{
186
		Result: &mesh_proto.ClustersResponse_Clusters{
187
			Clusters: []byte("forwarded"),
188
		},
189
	}, nil
190
}
191