kuma

1
package graceful
2

3
import (
4
	"fmt"
5
	"io"
6
	"net/http"
7
	"strconv"
8
	"time"
9

10
	"github.com/gruntwork-io/terratest/modules/k8s"
11
	. "github.com/onsi/ginkgo/v2"
12
	. "github.com/onsi/gomega"
13
	"github.com/pkg/errors"
14

15
	core_mesh "github.com/kumahq/kuma/pkg/core/resources/apis/mesh"
16
	"github.com/kumahq/kuma/pkg/util/channels"
17
	. "github.com/kumahq/kuma/test/framework"
18
	"github.com/kumahq/kuma/test/framework/deployments/testserver"
19
	"github.com/kumahq/kuma/test/framework/envs/kubernetes"
20
)
21

22
func Graceful() {
23
	if Config.Arch == "arm64" {
24
		return // K3D loadbalancer required for this test seems to not work with K3D
25
	}
26
	if Config.IPV6 {
27
		return // K3D cannot handle loadbalancer for IPV6
28
	}
29

30
	const name = "graceful"
31
	const namespace = "graceful"
32
	const mesh = "graceful"
33

34
	// Set up a gateway to be able to send requests constantly.
35
	// The alternative was to exec to the container, but this introduces a latency of getting into container for every curl
36
	gatewayInstnace := func(replicas int) string {
37
		return fmt.Sprintf(`
38
---
39
apiVersion: kuma.io/v1alpha1
40
kind: MeshGatewayInstance
41
metadata:
42
  name: edge-gateway
43
  namespace: graceful
44
  annotations:
45
    kuma.io/mesh: graceful
46
spec:
47
  replicas: %d
48
  serviceType: LoadBalancer
49
  tags:
50
    kuma.io/service: edge-gateway
51
`, replicas)
52
	}
53

54
	gateway := `
55
---
56
apiVersion: kuma.io/v1alpha1
57
kind: MeshGateway
58
metadata:
59
  name: edge-gateway
60
  namespace: graceful
61
mesh: graceful
62
spec:
63
  selectors:
64
  - match:
65
      kuma.io/service: edge-gateway
66
  conf:
67
    listeners:
68
    - port: 8080
69
      protocol: HTTP
70
---
71
apiVersion: kuma.io/v1alpha1
72
kind: MeshGatewayRoute
73
metadata:
74
  name: edge-gateway
75
  namespace: graceful
76
mesh: graceful
77
spec:
78
  selectors:
79
  - match:
80
      kuma.io/service: edge-gateway
81
  conf:
82
    http:
83
      rules:
84
      - matches:
85
        - path:
86
            match: PREFIX
87
            value: /
88
        backends:
89
        - destination:
90
            kuma.io/service: graceful_graceful_svc_80
91
`
92

93
	var gatewayIP string
94

95
	httpClient := http.Client{
96
		Timeout: 5 * time.Second,
97
	}
98

99
	BeforeAll(func() {
100
		err := NewClusterSetup().
101
			Install(MeshKubernetes(mesh)).
102
			Install(NamespaceWithSidecarInjection(namespace)).
103
			Install(YamlK8s(gateway)).
104
			Install(YamlK8s(gatewayInstnace(1))).
105
			Install(testserver.Install(
106
				testserver.WithNamespace(namespace),
107
				testserver.WithMesh(mesh),
108
				testserver.WithName(name),
109
			)).
110
			Setup(kubernetes.Cluster)
111
		Expect(err).To(Succeed())
112

113
		Eventually(func(g Gomega) {
114
			out, err := k8s.RunKubectlAndGetOutputE(
115
				kubernetes.Cluster.GetTesting(),
116
				kubernetes.Cluster.GetKubectlOptions(namespace),
117
				"get", "service", "edge-gateway", "-ojsonpath={.status.loadBalancer.ingress[0].ip}",
118
			)
119
			g.Expect(err).ToNot(HaveOccurred())
120
			g.Expect(out).ToNot(BeEmpty())
121
			gatewayIP = out
122
		}, "60s", "1s").Should(Succeed(), "could not get a LoadBalancer IP of the Gateway")
123

124
		// remove retries to avoid covering failed request
125
		Expect(DeleteMeshResources(kubernetes.Cluster, mesh, core_mesh.RetryResourceTypeDescriptor)).To(Succeed())
126
	})
127

128
	E2EAfterAll(func() {
129
		Expect(kubernetes.Cluster.TriggerDeleteNamespace(namespace)).To(Succeed())
130
		Expect(kubernetes.Cluster.DeleteMesh(mesh)).To(Succeed())
131
	})
132

133
	requestThroughGateway := func() error {
134
		resp, err := httpClient.Get("http://" + gatewayIP + ":8080")
135
		if err != nil {
136
			return err
137
		}
138
		defer resp.Body.Close()
139
		if resp.StatusCode != 200 {
140
			bytes, err := io.ReadAll(resp.Body)
141
			if err != nil {
142
				return err
143
			}
144
			return errors.Errorf("status code: %d body: %s", resp.StatusCode, string(bytes))
145
		}
146
		_, err = io.Copy(io.Discard, resp.Body)
147
		return err
148
	}
149

150
	type testCase struct {
151
		deploymentName string
152
		scaleFn        func(int) error
153
	}
154

155
	DescribeTable("should not drop a request when scaling up and down",
156
		func(given testCase) {
157
			// given constant traffic between client and server
158
			Eventually(requestThroughGateway, "30s", "1s").Should(Succeed())
159
			var failedErr error
160
			closeCh := make(chan struct{})
161
			defer close(closeCh)
162
			go func() {
163
				for {
164
					if channels.IsClosed(closeCh) {
165
						return
166
					}
167
					if err := requestThroughGateway(); err != nil {
168
						failedErr = err
169
						return
170
					}
171
					// add a slight delay to not overwhelm completely the host running this test and leave more resources to other tests running in parallel.
172
					time.Sleep(50 * time.Millisecond)
173
				}
174
			}()
175

176
			// when
177
			Expect(given.scaleFn(2)).To(Succeed())
178

179
			// then
180
			Eventually(func(g Gomega) {
181
				g.Expect(WaitNumPods(namespace, 2, given.deploymentName)(kubernetes.Cluster)).To(Succeed())
182
				g.Expect(WaitPodsAvailable(namespace, given.deploymentName)(kubernetes.Cluster)).To(Succeed())
183
			}, "30s", "1s").Should(Succeed())
184
			Expect(failedErr).ToNot(HaveOccurred())
185

186
			// when
187
			Expect(given.scaleFn(1)).To(Succeed())
188

189
			// then
190
			Eventually(func(g Gomega) {
191
				g.Expect(WaitNumPods(namespace, 1, given.deploymentName)(kubernetes.Cluster)).To(Succeed())
192
			}, "60s", "1s").Should(Succeed())
193

194
			Expect(failedErr).ToNot(HaveOccurred())
195
		},
196
		Entry("a service", testCase{
197
			deploymentName: name,
198
			scaleFn: func(replicas int) error {
199
				return k8s.RunKubectlE(
200
					kubernetes.Cluster.GetTesting(),
201
					kubernetes.Cluster.GetKubectlOptions(namespace),
202
					"scale", "deployment", name, "--replicas", strconv.Itoa(replicas),
203
				)
204
			},
205
		}),
206
		Entry("a gateway", testCase{
207
			deploymentName: "edge-gateway",
208
			scaleFn: func(replicas int) error {
209
				return kubernetes.Cluster.Install(YamlK8s(gatewayInstnace(replicas)))
210
			},
211
		}),
212
	)
213
}
214