Skip to content

Commit 98789cb

Browse files
ZerpetZerpet
committed
Include protocol information in Service ports
By setting this field, we are able to take full advantage of all Istio traffic monitoring features and rich metrics, we must hint Istio what's the underlying protocol. [Source here](https://istio.io/latest/docs/ops/configuration/traffic-management/protocol-selection/). According to Kubernetes docs, protocol that are not registered in IANA registry as well-known ports, must use a suffix, like my.company.com/my-protocol. Surprisingly, STOMP does not have a well-known port assigned. https://www.iana.org/assignments/service-names-port-numbers/service-names-port-numbers.xhtml Signed-off-by: Aitor Perez Cedres <acedres@vmware.com>
1 parent 5c77366 commit 98789cb

File tree

3 files changed

+342
-352
lines changed

3 files changed

+342
-352
lines changed

‎controllers/rabbitmqcluster_controller_test.go‎

Lines changed: 18 additions & 14 deletions
Original file line numberDiff line numberDiff line change
@@ -421,8 +421,9 @@ var _ = Describe("RabbitmqClusterController", func() {
421421
}, 3).Should(HaveKeyWithValue("test-key", "test-value"))
422422

423423
// verify that SuccessfulUpdate event is recorded for the service
424-
Expect(aggregateEventMsgs(ctx, cluster, "SuccessfulUpdate")).To(
425-
ContainSubstring("updated resource %s of Type *v1.Service", cluster.ChildResourceName("")))
424+
Eventually(func() string {
425+
return aggregateEventMsgs(ctx, cluster, "SuccessfulUpdate")
426+
}, 5).Should(ContainSubstring("updated resource %s of Type *v1.Service", cluster.ChildResourceName("")))
426427
})
427428

428429
It("the CPU and memory requirements are updated", func() {
@@ -1091,22 +1092,25 @@ var _ = Describe("RabbitmqClusterController", func() {
10911092
Expect(svc.Spec.Type).To(Equal(corev1.ServiceTypeClusterIP))
10921093
Expect(svc.Spec.Ports).To(ConsistOf(
10931094
corev1.ServicePort{
1094-
Name: "amqp",
1095-
Port: 5672,
1096-
Protocol: corev1.ProtocolTCP,
1097-
TargetPort: amqpTargetPort,
1095+
Name: "amqp",
1096+
Port: 5672,
1097+
Protocol: corev1.ProtocolTCP,
1098+
TargetPort: amqpTargetPort,
1099+
AppProtocol: pointer.String("amqp"),
10981100
},
10991101
corev1.ServicePort{
1100-
Name: "management",
1101-
Port: 15672,
1102-
Protocol: corev1.ProtocolTCP,
1103-
TargetPort: managementTargetPort,
1102+
Name: "management",
1103+
Port: 15672,
1104+
Protocol: corev1.ProtocolTCP,
1105+
TargetPort: managementTargetPort,
1106+
AppProtocol: pointer.String("http"),
11041107
},
11051108
corev1.ServicePort{
1106-
Name: "prometheus",
1107-
Port: 15692,
1108-
Protocol: corev1.ProtocolTCP,
1109-
TargetPort: prometheusTargetPort,
1109+
Name: "prometheus",
1110+
Port: 15692,
1111+
Protocol: corev1.ProtocolTCP,
1112+
TargetPort: prometheusTargetPort,
1113+
AppProtocol: pointer.String("prometheus.io/metrics"),
11101114
},
11111115
corev1.ServicePort{
11121116
Protocol: corev1.ProtocolTCP,

‎internal/resource/service.go‎

Lines changed: 99 additions & 152 deletions
Original file line numberDiff line numberDiff line change
@@ -12,6 +12,7 @@ package resource
1212
import (
1313
"encoding/json"
1414
"fmt"
15+
"k8s.io/utils/pointer"
1516

1617
"sigs.k8s.io/controller-runtime/pkg/client"
1718

@@ -110,172 +111,114 @@ func applySvcOverride(svc *corev1.Service, override *rabbitmqv1beta1.Service) er
110111
return nil
111112
}
112113

113-
func (builder *ServiceBuilder) generateServicePortsMapOnlyTLSListeners() map[string]corev1.ServicePort {
114-
servicePortsMap := map[string]corev1.ServicePort{
115-
"amqps": {
116-
Protocol: corev1.ProtocolTCP,
117-
Port: 5671,
118-
TargetPort: intstr.FromInt(5671),
119-
Name: "amqps",
120-
},
121-
"management-tls": {
122-
Protocol: corev1.ProtocolTCP,
123-
Port: 15671,
124-
TargetPort: intstr.FromInt(15671),
125-
Name: "management-tls",
126-
},
127-
"prometheus-tls": {
128-
Protocol: corev1.ProtocolTCP,
129-
Port: 15691,
130-
TargetPort: intstr.FromInt(15691),
131-
Name: "prometheus-tls",
132-
},
133-
}
134-
135-
if builder.Instance.AdditionalPluginEnabled("rabbitmq_stomp") {
136-
servicePortsMap["stomps"] = corev1.ServicePort{
137-
Protocol: corev1.ProtocolTCP,
138-
Port: 61614,
139-
Name: "stomps",
140-
TargetPort: intstr.FromInt(61614),
141-
}
142-
}
143-
if builder.Instance.AdditionalPluginEnabled("rabbitmq_mqtt") {
144-
servicePortsMap["mqtts"] = corev1.ServicePort{
145-
Protocol: corev1.ProtocolTCP,
146-
Port: 8883,
147-
Name: "mqtts",
148-
TargetPort: intstr.FromInt(8883),
114+
func (builder *ServiceBuilder) generateServicePortsMap() map[string]corev1.ServicePort {
115+
servicePortsMap := make(map[string]corev1.ServicePort, 7)
116+
117+
if builder.Instance.DisableNonTLSListeners() == false {
118+
servicePortsMap["amqp"] = corev1.ServicePort{
119+
Protocol: corev1.ProtocolTCP,
120+
Port: 5672,
121+
TargetPort: intstr.FromInt(5672),
122+
Name: "amqp",
123+
AppProtocol: pointer.String("amqp"),
149124
}
150-
}
151125

152-
if builder.Instance.StreamNeeded() {
153-
servicePortsMap["streams"] = corev1.ServicePort{
154-
Protocol: corev1.ProtocolTCP,
155-
Port: 5551,
156-
Name: "streams",
157-
TargetPort: intstr.FromInt(5551),
126+
servicePortsMap["management"] = corev1.ServicePort{
127+
Protocol: corev1.ProtocolTCP,
128+
Port: 15672,
129+
TargetPort: intstr.FromInt(15672),
130+
Name: "management",
131+
AppProtocol: pointer.String("http"),
158132
}
159-
}
160133

161-
if builder.Instance.MutualTLSEnabled() {
162-
if builder.Instance.AdditionalPluginEnabled("rabbitmq_web_stomp") {
163-
servicePortsMap["web-stomp-tls"] = corev1.ServicePort{
164-
Protocol: corev1.ProtocolTCP,
165-
Port: 15673,
166-
Name: "web-stomp-tls",
167-
TargetPort: intstr.FromInt(15673),
134+
if builder.Instance.AdditionalPluginEnabled("rabbitmq_mqtt") {
135+
servicePortsMap["mqtt"] = corev1.ServicePort{
136+
Protocol: corev1.ProtocolTCP,
137+
Port: 1883,
138+
TargetPort: intstr.FromInt(1883),
139+
Name: "mqtt",
140+
AppProtocol: pointer.String("mqtt"),
168141
}
169142
}
170143
if builder.Instance.AdditionalPluginEnabled("rabbitmq_web_mqtt") {
171-
servicePortsMap["web-mqtt-tls"] = corev1.ServicePort{
172-
Protocol: corev1.ProtocolTCP,
173-
Port: 15676,
174-
Name: "web-mqtt-tls",
175-
TargetPort: intstr.FromInt(15676),
144+
servicePortsMap["web-mqtt"] = corev1.ServicePort{
145+
Protocol: corev1.ProtocolTCP,
146+
Port: 15675,
147+
TargetPort: intstr.FromInt(15675),
148+
Name: "web-mqtt",
149+
AppProtocol: pointer.String("http"),
176150
}
177151
}
178-
}
179-
return servicePortsMap
180-
}
181-
182-
func (builder *ServiceBuilder) generateServicePortsMap() map[string]corev1.ServicePort {
183-
if builder.Instance.DisableNonTLSListeners() {
184-
return builder.generateServicePortsMapOnlyTLSListeners()
185-
}
186-
187-
servicePortsMap := map[string]corev1.ServicePort{
188-
"amqp": {
189-
Protocol: corev1.ProtocolTCP,
190-
Port: 5672,
191-
TargetPort: intstr.FromInt(5672),
192-
Name: "amqp",
193-
},
194-
"management": {
195-
Protocol: corev1.ProtocolTCP,
196-
Port: 15672,
197-
TargetPort: intstr.FromInt(15672),
198-
Name: "management",
199-
},
200-
}
201-
202-
if builder.Instance.AdditionalPluginEnabled("rabbitmq_mqtt") {
203-
servicePortsMap["mqtt"] = corev1.ServicePort{
204-
Protocol: corev1.ProtocolTCP,
205-
Port: 1883,
206-
TargetPort: intstr.FromInt(1883),
207-
Name: "mqtt",
208-
}
209-
}
210-
if builder.Instance.AdditionalPluginEnabled("rabbitmq_web_mqtt") {
211-
servicePortsMap["web-mqtt"] = corev1.ServicePort{
212-
Protocol: corev1.ProtocolTCP,
213-
Port: 15675,
214-
TargetPort: intstr.FromInt(15675),
215-
Name: "web-mqtt",
216-
}
217-
}
218-
if builder.Instance.AdditionalPluginEnabled("rabbitmq_stomp") {
219-
servicePortsMap["stomp"] = corev1.ServicePort{
220-
Protocol: corev1.ProtocolTCP,
221-
Port: 61613,
222-
TargetPort: intstr.FromInt(61613),
223-
Name: "stomp",
152+
if builder.Instance.AdditionalPluginEnabled("rabbitmq_stomp") {
153+
servicePortsMap["stomp"] = corev1.ServicePort{
154+
Protocol: corev1.ProtocolTCP,
155+
Port: 61613,
156+
TargetPort: intstr.FromInt(61613),
157+
Name: "stomp",
158+
AppProtocol: pointer.String("stomp.github.io/stomp"),
159+
}
224160
}
225-
}
226-
if builder.Instance.AdditionalPluginEnabled("rabbitmq_web_stomp") {
227-
servicePortsMap["web-stomp"] = corev1.ServicePort{
228-
Protocol: corev1.ProtocolTCP,
229-
Port: 15674,
230-
TargetPort: intstr.FromInt(15674),
231-
Name: "web-stomp",
161+
if builder.Instance.AdditionalPluginEnabled("rabbitmq_web_stomp") {
162+
servicePortsMap["web-stomp"] = corev1.ServicePort{
163+
Protocol: corev1.ProtocolTCP,
164+
Port: 15674,
165+
TargetPort: intstr.FromInt(15674),
166+
Name: "web-stomp",
167+
AppProtocol: pointer.String("http"),
168+
}
232169
}
233-
}
234170

235-
if builder.Instance.StreamNeeded() {
236-
servicePortsMap["stream"] = corev1.ServicePort{
237-
Protocol: corev1.ProtocolTCP,
238-
Port: 5552,
239-
TargetPort: intstr.FromInt(5552),
240-
Name: "stream",
171+
if builder.Instance.StreamNeeded() {
172+
servicePortsMap["stream"] = corev1.ServicePort{
173+
Protocol: corev1.ProtocolTCP,
174+
Port: 5552,
175+
TargetPort: intstr.FromInt(5552),
176+
Name: "stream",
177+
AppProtocol: pointer.String("rabbitmq.com/stream"),
178+
}
241179
}
242180
}
243181

244182
if builder.Instance.TLSEnabled() {
245183
servicePortsMap["amqps"] = corev1.ServicePort{
246-
Protocol: corev1.ProtocolTCP,
247-
Port: 5671,
248-
TargetPort: intstr.FromInt(5671),
249-
Name: "amqps",
184+
Protocol: corev1.ProtocolTCP,
185+
Port: 5671,
186+
TargetPort: intstr.FromInt(5671),
187+
Name: "amqps",
188+
AppProtocol: pointer.String("amqps"),
250189
}
251190
servicePortsMap["management-tls"] = corev1.ServicePort{
252-
Protocol: corev1.ProtocolTCP,
253-
Port: 15671,
254-
TargetPort: intstr.FromInt(15671),
255-
Name: "management-tls",
191+
Protocol: corev1.ProtocolTCP,
192+
Port: 15671,
193+
TargetPort: intstr.FromInt(15671),
194+
Name: "management-tls",
195+
AppProtocol: pointer.String("https"),
256196
}
257197
if builder.Instance.AdditionalPluginEnabled("rabbitmq_stomp") {
258198
servicePortsMap["stomps"] = corev1.ServicePort{
259-
Protocol: corev1.ProtocolTCP,
260-
Port: 61614,
261-
Name: "stomps",
262-
TargetPort: intstr.FromInt(61614),
199+
Protocol: corev1.ProtocolTCP,
200+
Port: 61614,
201+
Name: "stomps",
202+
TargetPort: intstr.FromInt(61614),
203+
AppProtocol: pointer.String("stomp.github.io/stomp-tls"),
263204
}
264205
}
265206
if builder.Instance.AdditionalPluginEnabled("rabbitmq_mqtt") {
266207
servicePortsMap["mqtts"] = corev1.ServicePort{
267-
Protocol: corev1.ProtocolTCP,
268-
Port: 8883,
269-
Name: "mqtts",
270-
TargetPort: intstr.FromInt(8883),
208+
Protocol: corev1.ProtocolTCP,
209+
Port: 8883,
210+
Name: "mqtts",
211+
TargetPort: intstr.FromInt(8883),
212+
AppProtocol: pointer.String("mqtts"),
271213
}
272214
}
273215
if builder.Instance.StreamNeeded() {
274216
servicePortsMap["streams"] = corev1.ServicePort{
275-
Protocol: corev1.ProtocolTCP,
276-
Port: 5551,
277-
Name: "streams",
278-
TargetPort: intstr.FromInt(5551),
217+
Protocol: corev1.ProtocolTCP,
218+
Port: 5551,
219+
Name: "streams",
220+
TargetPort: intstr.FromInt(5551),
221+
AppProtocol: pointer.String("rabbitmq.com/stream-tls"),
279222
}
280223
}
281224

@@ -285,35 +228,39 @@ func (builder *ServiceBuilder) generateServicePortsMap() map[string]corev1.Servi
285228
// doubling the number of nodes showing up in Grafana because the
286229
// 'instance' label consists of "<host>:<port>".
287230
servicePortsMap["prometheus-tls"] = corev1.ServicePort{
288-
Protocol: corev1.ProtocolTCP,
289-
Port: 15691,
290-
TargetPort: intstr.FromInt(15691),
291-
Name: "prometheus-tls",
231+
Protocol: corev1.ProtocolTCP,
232+
Port: 15691,
233+
TargetPort: intstr.FromInt(15691),
234+
Name: "prometheus-tls",
235+
AppProtocol: pointer.String("prometheus.io/metric-tls"),
292236
}
293237
} else {
294238
servicePortsMap["prometheus"] = corev1.ServicePort{
295-
Protocol: corev1.ProtocolTCP,
296-
Port: 15692,
297-
TargetPort: intstr.FromInt(15692),
298-
Name: "prometheus",
239+
Protocol: corev1.ProtocolTCP,
240+
Port: 15692,
241+
TargetPort: intstr.FromInt(15692),
242+
Name: "prometheus",
243+
AppProtocol: pointer.String("prometheus.io/metrics"),
299244
}
300245
}
301246

302247
if builder.Instance.MutualTLSEnabled() {
303248
if builder.Instance.AdditionalPluginEnabled("rabbitmq_web_stomp") {
304249
servicePortsMap["stomps"] = corev1.ServicePort{
305-
Protocol: corev1.ProtocolTCP,
306-
Port: 15673,
307-
Name: "web-stomp-tls",
308-
TargetPort: intstr.FromInt(15673),
250+
Protocol: corev1.ProtocolTCP,
251+
Port: 15673,
252+
Name: "web-stomp-tls",
253+
TargetPort: intstr.FromInt(15673),
254+
AppProtocol: pointer.String("https"),
309255
}
310256
}
311257
if builder.Instance.AdditionalPluginEnabled("rabbitmq_web_mqtt") {
312258
servicePortsMap["mqtts"] = corev1.ServicePort{
313-
Protocol: corev1.ProtocolTCP,
314-
Port: 15676,
315-
Name: "web-mqtt-tls",
316-
TargetPort: intstr.FromInt(15676),
259+
Protocol: corev1.ProtocolTCP,
260+
Port: 15676,
261+
Name: "web-mqtt-tls",
262+
TargetPort: intstr.FromInt(15676),
263+
AppProtocol: pointer.String("https"),
317264
}
318265
}
319266
}

0 commit comments

Comments
 (0)