kuma

1
package trafficlog
2

3
import (
4
	"bytes"
5
	"strconv"
6
	"strings"
7
	"text/template"
8
	"time"
9

10
	. "github.com/onsi/ginkgo/v2"
11
	. "github.com/onsi/gomega"
12

13
	. "github.com/kumahq/kuma/test/framework"
14
	"github.com/kumahq/kuma/test/framework/client"
15
	"github.com/kumahq/kuma/test/framework/deployments/democlient"
16
	"github.com/kumahq/kuma/test/framework/deployments/testserver"
17
	"github.com/kumahq/kuma/test/framework/envs/kubernetes"
18
)
19

20
func TCPLogging() {
21
	Describe("TrafficLog Logging to TCP", func() {
22
		meshName := "trafficlog-tcp-logging"
23
		trafficNamespace := "trafficlog-tcp-logging"
24
		tcpSinkNamespace := "tcp-sink-namespace"
25
		tcpSinkAppName := "tcp-sink"
26
		testServer := "test-server"
27

28
		args := struct {
29
			MeshName           string
30
			TrafficNamespace   string
31
			TcpSinkNamespace   string
32
			TcpSinkAppName     string
33
			KumaUniversalImage string
34
		}{
35
			meshName, trafficNamespace, tcpSinkNamespace,
36
			tcpSinkAppName, Config.GetUniversalImage(),
37
		}
38

39
		loggingBackend := &bytes.Buffer{}
40
		tmpl := template.Must(template.New("loggingBackend").Parse(`
41
apiVersion: kuma.io/v1alpha1
42
kind: Mesh
43
metadata:
44
  name: {{ .MeshName }}
45
spec:
46
  logging:
47
    defaultBackend: netcat
48
    backends:
49
      - name: netcat
50
        format: '%START_TIME(%s)%,%KUMA_SOURCE_SERVICE%,%KUMA_DESTINATION_SERVICE%'
51
        type: tcp
52
        conf:
53
          address: {{.TcpSinkAppName}}.{{.TcpSinkNamespace}}.svc.cluster.local:9999
54
`))
55
		Expect(tmpl.Execute(loggingBackend, &args)).To(Succeed())
56

57
		trafficLog := &bytes.Buffer{}
58
		tmpl = template.Must(template.New("trafficLog").Parse(`
59
apiVersion: kuma.io/v1alpha1
60
kind: TrafficLog
61
mesh: {{ .MeshName }}
62
metadata:
63
  name: all-traffic
64
spec:
65
  sources:
66
    - match:
67
        kuma.io/service: "*"
68
  destinations:
69
    - match:
70
        kuma.io/service: "*"
71
`))
72
		Expect(tmpl.Execute(trafficLog, &args)).To(Succeed())
73

74
		tcpSink := &bytes.Buffer{}
75
		tmpl = template.Must(template.New("tcpSink").Parse(`
76
apiVersion: v1
77
kind: Service
78
metadata:
79
  name: {{.TcpSinkAppName}}
80
  namespace: {{.TcpSinkNamespace}}
81
spec:
82
  ports:
83
    - port: 9999
84
      name: netcat
85
  selector:
86
    app: {{.TcpSinkAppName}}
87
---
88
apiVersion: apps/v1
89
kind: Deployment
90
metadata:
91
  name: {{.TcpSinkAppName}}
92
  namespace: {{.TcpSinkNamespace}}
93
  labels:
94
    app: {{.TcpSinkAppName}}
95
spec:
96
  selector:
97
    matchLabels:
98
      app: {{.TcpSinkAppName}}
99
  template:
100
    metadata:
101
      labels:
102
        app: {{.TcpSinkAppName}}
103
    spec:
104
      containers:
105
        - name: {{.TcpSinkAppName}}
106
          image: {{.KumaUniversalImage}}
107
          command: ["/bin/bash"]
108
          args: ["-c", "/bin/netcat -lk -p 9999 > /nc.out"]
109
          imagePullPolicy: IfNotPresent
110
          ports:
111
            - containerPort: 9999
112
`))
113

114
		Expect(tmpl.Execute(tcpSink, &args)).To(Succeed())
115

116
		BeforeAll(func() {
117
			err := NewClusterSetup().
118
				Install(YamlK8s(loggingBackend.String())).
119
				Install(YamlK8s(trafficLog.String())).
120
				Install(NamespaceWithSidecarInjection(trafficNamespace)).
121
				Install(Namespace(tcpSinkNamespace)).
122
				Install(democlient.Install(democlient.WithNamespace(trafficNamespace), democlient.WithMesh(meshName))).
123
				Install(YamlK8s(tcpSink.String())).
124
				Install(WaitPodsAvailable(tcpSinkNamespace, tcpSinkAppName)).
125
				Install(testserver.Install(
126
					testserver.WithNamespace(trafficNamespace),
127
					testserver.WithMesh(meshName),
128
					testserver.WithName(testServer))).
129
				Install(TrafficRouteKubernetes(meshName)).
130
				Install(TrafficPermissionKubernetes(meshName)).
131
				Setup(kubernetes.Cluster)
132
			Expect(err).ToNot(HaveOccurred())
133
		})
134

135
		E2EAfterAll(func() {
136
			Expect(kubernetes.Cluster.TriggerDeleteNamespace(trafficNamespace)).To(Succeed())
137
			Expect(kubernetes.Cluster.TriggerDeleteNamespace(tcpSinkNamespace)).To(Succeed())
138
			Expect(kubernetes.Cluster.DeleteMesh(meshName)).To(Succeed())
139
		})
140

141
		It("should send a traffic log to TCP port", func() {
142
			var startTimeStr, src, dst string
143
			var err error
144
			var stdout string
145
			tcpSinkPodName, err := PodNameOfApp(kubernetes.Cluster, tcpSinkAppName, tcpSinkNamespace)
146
			Expect(err).ToNot(HaveOccurred())
147
			Eventually(func(g Gomega) {
148
				_, err := client.CollectEchoResponse(
149
					kubernetes.Cluster, AppModeDemoClient, "test-server",
150
					client.FromKubernetesPod(trafficNamespace, "demo-client"),
151
				)
152
				g.Expect(err).ToNot(HaveOccurred())
153

154
				stdout, _, err = kubernetes.Cluster.Exec(tcpSinkNamespace, tcpSinkPodName,
155
					"tcp-sink", "head", "-1", "/nc.out")
156
				g.Expect(err).ToNot(HaveOccurred())
157
				parts := strings.Split(stdout, ",")
158
				g.Expect(parts).To(HaveLen(3))
159
				startTimeStr, src, dst = parts[0], parts[1], parts[2]
160
			}).Should(Succeed())
161
			startTimeInt, err := strconv.Atoi(startTimeStr)
162
			Expect(err).ToNot(HaveOccurred())
163
			startTime := time.Unix(int64(startTimeInt), 0)
164

165
			// Just testing that it is a timestamp, not accuracy. If it's
166
			// an int that would represent Unix time within an hour of now
167
			// it's probably a timestamp substitution.
168
			Expect(startTime).To(BeTemporally("~", time.Now(), time.Hour))
169

170
			Expect(src).To(Equal("demo-client_trafficlog-tcp-logging_svc"))
171
			Expect(strings.TrimSpace(dst)).To(Equal("test-server_trafficlog-tcp-logging_svc_80"))
172
		})
173
	})
174
}
175