kuma

1
package delegated
2

3
import (
4
	"fmt"
5

6
	. "github.com/onsi/ginkgo/v2"
7
	. "github.com/onsi/gomega"
8

9
	"github.com/kumahq/kuma/pkg/plugins/policies/meshcircuitbreaker/api/v1alpha1"
10
	meshretry_api "github.com/kumahq/kuma/pkg/plugins/policies/meshretry/api/v1alpha1"
11
	"github.com/kumahq/kuma/test/framework"
12
	"github.com/kumahq/kuma/test/framework/client"
13
	"github.com/kumahq/kuma/test/framework/envs/kubernetes"
14
)
15

16
func CircuitBreaker(config *Config) func() {
17
	GinkgoHelper()
18

19
	return func() {
20
		BeforeAll(func() {
21
			Expect(framework.DeleteMeshPolicyOrError(
22
				kubernetes.Cluster,
23
				v1alpha1.MeshCircuitBreakerResourceTypeDescriptor,
24
				fmt.Sprintf("mesh-circuit-breaker-all-%s", config.Mesh),
25
			)).To(Succeed())
26
			Expect(framework.DeleteMeshPolicyOrError(
27
				kubernetes.Cluster,
28
				meshretry_api.MeshRetryResourceTypeDescriptor,
29
				fmt.Sprintf("mesh-retry-all-%s", config.Mesh),
30
			)).To(Succeed())
31
		})
32

33
		framework.E2EAfterEach(func() {
34
			Expect(framework.DeleteMeshResources(
35
				kubernetes.Cluster,
36
				config.Mesh,
37
				v1alpha1.MeshCircuitBreakerResourceTypeDescriptor,
38
			)).To(Succeed())
39
		})
40

41
		DescribeTable("should configure circuit breaker limits and outlier"+
42
			" detectors for connections", func(yaml string) {
43
			// given no MeshCircuitBreaker
44
			mcbs, err := kubernetes.Cluster.GetKumactlOptions().
45
				KumactlList("meshcircuitbreakers", config.Mesh)
46
			Expect(err).ToNot(HaveOccurred())
47
			Expect(mcbs).To(BeEmpty())
48

49
			Eventually(func() ([]client.FailureResponse, error) {
50
				return client.CollectResponsesAndFailures(
51
					kubernetes.Cluster,
52
					"demo-client",
53
					fmt.Sprintf("http://%s/test-server", config.KicIP),
54
					client.FromKubernetesPod(config.NamespaceOutsideMesh, "demo-client"),
55
					client.WithNumberOfRequests(10),
56
				)
57
			}, "30s", "1s").Should(And(
58
				HaveLen(10),
59
				HaveEach(HaveField("ResponseCode", 200)),
60
			))
61

62
			// when
63
			Expect(kubernetes.Cluster.Install(framework.YamlK8s(yaml))).To(Succeed())
64

65
			// then
66
			Eventually(func(g Gomega) ([]client.FailureResponse, error) {
67
				return client.CollectResponsesAndFailures(
68
					kubernetes.Cluster,
69
					"demo-client",
70
					fmt.Sprintf("http://%s/test-server", config.KicIP),
71
					client.FromKubernetesPod(config.NamespaceOutsideMesh, "demo-client"),
72
					client.WithNumberOfRequests(10),
73
					// increase processing time of a request to increase
74
					// a probability of triggering maxPendingRequest limit
75
					client.WithHeader("x-set-response-delay-ms", "1000"),
76
					client.WithoutRetries(),
77
				)
78
			}, "30s", "1s").Should(And(
79
				HaveLen(10),
80
				ContainElement(HaveField("ResponseCode", 503)),
81
			))
82
		},
83
			Entry("outbound circuit breaker", fmt.Sprintf(`
84
apiVersion: kuma.io/v1alpha1
85
kind: MeshCircuitBreaker
86
metadata:
87
  name: mcb-outbound
88
  namespace: %s
89
  labels:
90
    kuma.io/mesh: %s
91
spec:
92
  targetRef:
93
    kind: Mesh
94
  to:
95
    - targetRef:
96
        kind: Mesh
97
      default:
98
        connectionLimits:
99
          maxConnectionPools: 1
100
          maxConnections: 1
101
          maxPendingRequests: 1
102
          maxRequests: 1
103
          maxRetries: 1
104
`, config.CpNamespace, config.Mesh)),
105
			Entry("inbound circuit breaker", fmt.Sprintf(`
106
apiVersion: kuma.io/v1alpha1
107
kind: MeshCircuitBreaker
108
metadata:
109
  name: mcb-inbound
110
  namespace: %s
111
  labels:
112
    kuma.io/mesh: %s
113
spec:
114
  targetRef:
115
    kind: Mesh
116
  from:
117
    - targetRef:
118
        kind: Mesh
119
      default:
120
        connectionLimits:
121
          maxConnectionPools: 1
122
          maxConnections: 1
123
          maxPendingRequests: 1
124
          maxRequests: 1
125
          maxRetries: 1
126
`, config.CpNamespace, config.Mesh)),
127
		)
128
	}
129
}
130