Skip to content

Commit 8e8c039

Browse files
heanlanheanlan
authored
Add live Flow Visibility via Flow Aggregator gRPC
Add real-time flow visibility to the Antrea UI by streaming flow records directly from the Flow Aggregator. - Backend: add a gRPC client (GRPCFlowStreamSubscriber) that connects to the Flow Aggregator FlowStreamService over server-side TLS and relays records to the browser as Server-Sent Events at /api/v1/flows/stream. The CA certificate is read from a ConfigMap, with an explicit insecureSkipVerify option for dev/test. - Frontend: add a Flow Visibility page with a live Flow List and a D3-based Service Map, backed by a fetch-based SSE client that supports Authorization headers. Includes filtering by namespace, pod, service, IP, flow type, and direction. - Helm chart: add flowAggregator values and a cross-namespace RoleBinding so antrea-ui can read the Flow Aggregator CA ConfigMap. - Vendor the FlowStreamService proto stubs locally under pkg/flowpb to avoid depending on the full antrea.io/antrea module. Signed-off-by: Anlan He <anlan.he@broadcom.com>
1 parent e62a284 commit 8e8c039

41 files changed

Lines changed: 7390 additions & 319 deletions

Some content is hidden

Large Commits have some content hidden by default. Use the searchbox below for content that may be hidden.

‎apis/v1/flow.go‎

Lines changed: 164 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,164 @@
1+
// Copyright 2026 Antrea Authors
2+
//
3+
// Licensed under the Apache License, Version 2.0 (the "License");
4+
// you may not use this file except in compliance with the License.
5+
// You may obtain a copy of the License at
6+
//
7+
// http://www.apache.org/licenses/LICENSE-2.0
8+
//
9+
// Unless required by applicable law or agreed to in writing, software
10+
// distributed under the License is distributed on an "AS IS" BASIS,
11+
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
12+
// See the License for the specific language governing permissions and
13+
// limitations under the License.
14+
15+
package v1
16+
17+
// JSON-serializable flow types for the SSE API, mirroring the protobuf Flow message.
18+
19+
type FlowType int32
20+
21+
const (
22+
FlowTypeUnspecified FlowType = 0
23+
FlowTypeIntraNode FlowType = 1
24+
FlowTypeInterNode FlowType = 2
25+
FlowTypeToExternal FlowType = 3
26+
FlowTypeFromExternal FlowType = 4
27+
)
28+
29+
type NetworkPolicyType int32
30+
31+
const (
32+
NetworkPolicyTypeUnspecified NetworkPolicyType = 0
33+
NetworkPolicyTypeK8s NetworkPolicyType = 1
34+
NetworkPolicyTypeANP NetworkPolicyType = 2
35+
NetworkPolicyTypeACNP NetworkPolicyType = 3
36+
)
37+
38+
type NetworkPolicyRuleAction int32
39+
40+
const (
41+
NetworkPolicyRuleActionNoAction NetworkPolicyRuleAction = 0
42+
NetworkPolicyRuleActionAllow NetworkPolicyRuleAction = 1
43+
NetworkPolicyRuleActionDrop NetworkPolicyRuleAction = 2
44+
NetworkPolicyRuleActionReject NetworkPolicyRuleAction = 3
45+
)
46+
47+
type IPVersion int32
48+
49+
const (
50+
IPVersionUnspecified IPVersion = 0
51+
IPVersionIPv4 IPVersion = 4
52+
IPVersionIPv6 IPVersion = 6
53+
)
54+
55+
type FlowEndReason int32
56+
57+
const (
58+
FlowEndReasonUnspecified FlowEndReason = 0
59+
FlowEndReasonIdleTimeout FlowEndReason = 1
60+
FlowEndReasonActiveTimeout FlowEndReason = 2
61+
FlowEndReasonEndOfFlow FlowEndReason = 3
62+
FlowEndReasonForcedEnd FlowEndReason = 4
63+
FlowEndReasonLackOfResources FlowEndReason = 5
64+
)
65+
66+
type FlowStats struct {
67+
PacketTotalCount uint64 `json:"packetTotalCount"`
68+
PacketDeltaCount uint64 `json:"packetDeltaCount"`
69+
OctetTotalCount uint64 `json:"octetTotalCount"`
70+
OctetDeltaCount uint64 `json:"octetDeltaCount"`
71+
Throughput uint64 `json:"throughput,omitempty"`
72+
}
73+
74+
type FlowTCP struct {
75+
StateName string `json:"stateName"`
76+
}
77+
78+
type FlowTransport struct {
79+
ProtocolNumber uint32 `json:"protocolNumber"`
80+
SourcePort uint32 `json:"sourcePort"`
81+
DestinationPort uint32 `json:"destinationPort"`
82+
TCP *FlowTCP `json:"tcp,omitempty"`
83+
}
84+
85+
type FlowIP struct {
86+
Version IPVersion `json:"version"`
87+
Source string `json:"source"`
88+
Destination string `json:"destination"`
89+
}
90+
91+
type FlowKubernetes struct {
92+
FlowType FlowType `json:"flowType"`
93+
94+
SourcePodNamespace string `json:"sourcePodNamespace"`
95+
SourcePodName string `json:"sourcePodName"`
96+
SourcePodUid string `json:"sourcePodUid"`
97+
SourcePodLabels map[string]string `json:"sourcePodLabels,omitempty"`
98+
99+
SourceNodeName string `json:"sourceNodeName"`
100+
SourceNodeUid string `json:"sourceNodeUid"`
101+
102+
DestinationPodNamespace string `json:"destinationPodNamespace"`
103+
DestinationPodName string `json:"destinationPodName"`
104+
DestinationPodUid string `json:"destinationPodUid"`
105+
DestinationPodLabels map[string]string `json:"destinationPodLabels,omitempty"`
106+
107+
DestinationNodeName string `json:"destinationNodeName"`
108+
DestinationNodeUid string `json:"destinationNodeUid"`
109+
110+
DestinationClusterIp string `json:"destinationClusterIp"`
111+
DestinationServicePort uint32 `json:"destinationServicePort"`
112+
DestinationServicePortName string `json:"destinationServicePortName"`
113+
DestinationServiceUid string `json:"destinationServiceUid"`
114+
115+
IngressNetworkPolicyType NetworkPolicyType `json:"ingressNetworkPolicyType"`
116+
IngressNetworkPolicyNamespace string `json:"ingressNetworkPolicyNamespace"`
117+
IngressNetworkPolicyName string `json:"ingressNetworkPolicyName"`
118+
IngressNetworkPolicyUid string `json:"ingressNetworkPolicyUid"`
119+
IngressNetworkPolicyRuleName string `json:"ingressNetworkPolicyRuleName"`
120+
IngressNetworkPolicyRuleAction NetworkPolicyRuleAction `json:"ingressNetworkPolicyRuleAction"`
121+
122+
EgressNetworkPolicyType NetworkPolicyType `json:"egressNetworkPolicyType"`
123+
EgressNetworkPolicyNamespace string `json:"egressNetworkPolicyNamespace"`
124+
EgressNetworkPolicyName string `json:"egressNetworkPolicyName"`
125+
EgressNetworkPolicyUid string `json:"egressNetworkPolicyUid"`
126+
EgressNetworkPolicyRuleName string `json:"egressNetworkPolicyRuleName"`
127+
EgressNetworkPolicyRuleAction NetworkPolicyRuleAction `json:"egressNetworkPolicyRuleAction"`
128+
129+
EgressName string `json:"egressName,omitempty"`
130+
EgressIp string `json:"egressIp,omitempty"`
131+
EgressNodeName string `json:"egressNodeName,omitempty"`
132+
EgressNodeUid string `json:"egressNodeUid,omitempty"`
133+
EgressUid string `json:"egressUid,omitempty"`
134+
}
135+
136+
type Flow struct {
137+
ID string `json:"id"`
138+
StartTs string `json:"startTs"`
139+
EndTs string `json:"endTs"`
140+
EndReason FlowEndReason `json:"endReason"`
141+
IP FlowIP `json:"ip"`
142+
Transport FlowTransport `json:"transport"`
143+
K8s FlowKubernetes `json:"k8s"`
144+
Stats FlowStats `json:"stats"`
145+
ReverseStats FlowStats `json:"reverseStats"`
146+
}
147+
148+
// FlowStreamEvent carries flow data and/or a dropped count from the stream.
149+
// When Flows is non-empty, the SSE handler emits a "flow" event.
150+
// When DroppedCount is non-zero, the SSE handler emits a "dropped" event.
151+
type FlowStreamEvent struct {
152+
Flows []Flow `json:"flows,omitempty"`
153+
DroppedCount uint64 `json:"droppedCount,omitempty"`
154+
}
155+
156+
// FlowStreamDroppedEvent is the JSON payload for an SSE "dropped" event.
157+
type FlowStreamDroppedEvent struct {
158+
DroppedCount uint64 `json:"droppedCount"`
159+
}
160+
161+
// FlowStreamErrorEvent is the JSON payload for an SSE "error" event.
162+
type FlowStreamErrorEvent struct {
163+
Message string `json:"message"`
164+
}

‎apis/v1/frontend_settings.go‎

Lines changed: 7 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -20,11 +20,16 @@ type FrontendAuthSettings struct {
2020
OIDCProviderName string `json:"oidcProviderName,omitempty"`
2121
}
2222

23+
type FrontendFeatureSettings struct {
24+
FlowVisibilityEnabled bool `json:"flowVisibilityEnabled"`
25+
}
26+
2327
// FrontendSettings are global settings exposed to the frontend, which can be
2428
// used to render some pages appropriately. These settings are not user-specific
2529
// and not confidential (the API for these settings is not protected by any auth
2630
// mechanism).
2731
type FrontendSettings struct {
28-
Version string `json:"version"`
29-
Auth FrontendAuthSettings `json:"auth"`
32+
Version string `json:"version"`
33+
Auth FrontendAuthSettings `json:"auth"`
34+
Features FrontendFeatureSettings `json:"features"`
3035
}

‎build/charts/antrea-ui/README.md‎

Lines changed: 7 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -19,6 +19,7 @@ Kubernetes: `>= 1.16.0-0`
1919
| Key | Type | Default | Description |
2020
|-----|------|---------|-------------|
2121
| affinity | object | `{}` | Affinity for the Antrea UI Pod. |
22+
| antreaNamespace | string | `"kube-system"` | Namespace where Antrea is installed. |
2223
| auth.basic.enable | bool | `true` | Enable password-based authentication. |
2324
| auth.oidc.clientID | string | `""` | Application (client) ID to be used by the Antrea UI server to identify itself to the OIDC provider. |
2425
| auth.oidc.clientIDSecretRef.key | string | `"clientID"` | Name of the key field storing the application (client) ID in the referenced secret. |
@@ -40,6 +41,12 @@ Kubernetes: `>= 1.16.0-0`
4041
| dex.image | object | `{"pullPolicy":"IfNotPresent","repository":"ghcr.io/dexidp/dex","tag":"v2.36.0-distroless"}` | Container image to use for Dex. |
4142
| dex.resources | object | `{}` | Resource requests and limits for the Dex container. |
4243
| extraVolumes | list | `[]` | Additional volumes. |
44+
| flowAggregator.address | string | `"flow-aggregator.flow-aggregator.svc:14740"` | gRPC address (host:port) of the FlowStreamService. |
45+
| flowAggregator.caConfigMap | string | `"flow-aggregator-ca"` | Name of the ConfigMap (in namespace below) containing the CA certificate (key: ca.crt) used to verify the FlowStreamService server certificate. The FlowStreamService uses server-side TLS only (no client authentication). Leave empty to skip server certificate verification (dev/test only). |
46+
| flowAggregator.enabled | bool | `false` | When true, the backend connects to Flow Aggregator's FlowStreamService over gRPC. |
47+
| flowAggregator.insecureSkipVerify | bool | `false` | Disable TLS server certificate verification. Should only be used for development or testing; never enable this in production. |
48+
| flowAggregator.namespace | string | `"flow-aggregator"` | Namespace where the Flow Aggregator is installed. |
49+
| flowAggregator.serverName | string | `""` | Override the TLS server name used for certificate verification. Useful when dialing via kubectl port-forward (loopback address) while the server cert is issued for the in-cluster Service DNS name (e.g. flow-aggregator.flow-aggregator.svc). Leave empty to use the hostname from the address field. |
4350
| frontend.extraVolumeMounts | list | `[]` | Additional volumeMounts. |
4451
| frontend.image | object | `{"pullPolicy":"IfNotPresent","repository":"antrea/antrea-ui-frontend","tag":""}` | Container image to use for the Antrea UI frontend. |
4552
| frontend.port | int | `3000` | Container port on which the frontend will listen. |

‎build/charts/antrea-ui/templates/_backend_conf.tpl‎

Lines changed: 12 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -1,6 +1,7 @@
11
{{- define "antrea-ui.backend.conf" }}
22
addr: ":{{ .Values.backend.port }}"
3-
url: {{ .Values.url }}
3+
url: {{ .Values.url | quote }}
4+
antreaNamespace: {{ .Values.antreaNamespace | quote }}
45
auth:
56
basic:
67
enabled: {{ .Values.auth.basic.enable }}
@@ -12,5 +13,14 @@ auth:
1213
logoutURL: {{ .Values.auth.oidc.logoutURL | quote }}
1314
jwtKeyPath: "/app/jwt-key.pem"
1415
cookieSecure: {{ include "cookieSecure" . }}
15-
logVerbosity: {{ .Values.logVerbosity }}
16+
logVerbosity: {{ .Values.backend.logVerbosity }}
17+
flowAggregator:
18+
enabled: {{ .Values.flowAggregator.enabled }}
19+
address: {{ .Values.flowAggregator.address | quote }}
20+
{{- if .Values.flowAggregator.enabled }}
21+
caConfigMap: {{ .Values.flowAggregator.caConfigMap | quote }}
22+
namespace: {{ .Values.flowAggregator.namespace | default "flow-aggregator" | quote }}
23+
serverName: {{ .Values.flowAggregator.serverName | quote }}
24+
insecureSkipVerify: {{ .Values.flowAggregator.insecureSkipVerify }}
25+
{{- end }}
1626
{{- end }}

‎build/charts/antrea-ui/templates/_nginx_conf.tpl‎

Lines changed: 19 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -27,6 +27,25 @@ server {
2727
proxy_set_header X-Forwarded-For $proxy_add_x_forwarded_for;
2828
proxy_set_header X-Real-IP $remote_addr;
2929

30+
# Flow SSE can be idle for a long time when the Flow Aggregator ring buffer has no matching
31+
# records; nginx's default proxy_read_timeout (~60s) then returns 504 to the browser.
32+
location /api/v1/flows/stream {
33+
proxy_http_version 1.1;
34+
proxy_pass_request_headers on;
35+
proxy_hide_header Access-Control-Allow-Origin;
36+
proxy_set_header Connection '';
37+
proxy_buffering off;
38+
proxy_read_timeout 86400s;
39+
proxy_send_timeout 86400s;
40+
proxy_pass http://127.0.0.1:{{ .Values.backend.port }};
41+
{{- $secure := include "cookieSecure" . -}}
42+
{{- if eq $secure "true" }}
43+
proxy_cookie_flags ~ httponly secure;
44+
{{- else }}
45+
proxy_cookie_flags ~ httponly;
46+
{{- end }}
47+
}
48+
3049
location /api {
3150
proxy_http_version 1.1;
3251
proxy_pass_request_headers on;

‎build/charts/antrea-ui/templates/flow-aggregator-rolebinding.yaml‎

Lines changed: 22 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,22 @@
1+
{{- if .Values.flowAggregator.enabled -}}
2+
kind: RoleBinding
3+
apiVersion: rbac.authorization.k8s.io/v1
4+
metadata:
5+
labels:
6+
app: antrea-ui
7+
name: {{ .Release.Name }}-flow-aggregator-ca-reader
8+
# This RoleBinding is in the flow-aggregator namespace so the antrea-ui
9+
# ServiceAccount can read the flow-aggregator-ca ConfigMap across namespaces.
10+
# It references flow-aggregator-exporter-role which is created by the
11+
# flow-aggregator Helm chart. The flow-aggregator chart must be installed
12+
# before (or alongside) antrea-ui for this RoleBinding to be effective.
13+
namespace: {{ .Values.flowAggregator.namespace | default "flow-aggregator" }}
14+
subjects:
15+
- kind: ServiceAccount
16+
name: antrea-ui
17+
namespace: {{ .Release.Namespace }}
18+
roleRef:
19+
kind: Role
20+
name: flow-aggregator-exporter-role
21+
apiGroup: rbac.authorization.k8s.io
22+
{{- end }}

‎build/charts/antrea-ui/values.yaml‎

Lines changed: 25 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -31,6 +31,31 @@ backend:
3131
# -- Address at which the Antrea UI is accessible. Not required for most configurations.
3232
url: ""
3333

34+
# -- Namespace where Antrea is installed.
35+
antreaNamespace: "kube-system"
36+
37+
# Flow Aggregator gRPC integration (live flow visibility in the UI).
38+
flowAggregator:
39+
# -- When true, the backend connects to Flow Aggregator's FlowStreamService over gRPC.
40+
enabled: false
41+
# -- gRPC address (host:port) of the FlowStreamService.
42+
address: "flow-aggregator.flow-aggregator.svc:14740"
43+
# -- Name of the ConfigMap (in namespace below) containing the CA certificate (key: ca.crt)
44+
# used to verify the FlowStreamService server certificate.
45+
# The FlowStreamService uses server-side TLS only (no client authentication).
46+
# Leave empty to skip server certificate verification (dev/test only).
47+
caConfigMap: flow-aggregator-ca
48+
# -- Namespace where the Flow Aggregator is installed.
49+
namespace: flow-aggregator
50+
# -- Override the TLS server name used for certificate verification. Useful when dialing
51+
# via kubectl port-forward (loopback address) while the server cert is issued for the
52+
# in-cluster Service DNS name (e.g. flow-aggregator.flow-aggregator.svc).
53+
# Leave empty to use the hostname from the address field.
54+
serverName: ""
55+
# -- Disable TLS server certificate verification. Should only be used for development
56+
# or testing; never enable this in production.
57+
insecureSkipVerify: false
58+
3459
security:
3560
# -- (bool) Set the Secure attribute for Antrea UI cookies. The attribute is set by default when HTTPS is
3661
# enabled in Antrea UI (by setting https.enable to true). When using an Ingress to terminate TLS,

‎client/web/antrea-ui/src/App.tsx‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -176,7 +176,7 @@ function App() {
176176
</header>
177177
<div cds-layout="horizontal align:top wrap:none" style={{ height: "100%" }}>
178178
<NavTab />
179-
<div cds-layout="vertical p:md gap:md">
179+
<div cds-layout="vertical p:md gap:md" style={{ minWidth: 0, flex: 1 }}>
180180
<AppErrorProvider>
181181
<WaitForSettings>
182182
<LoginWall>

‎client/web/antrea-ui/src/api/flow-stream.test.ts‎

Lines changed: 38 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,38 @@
1+
/**
2+
* Copyright 2026 Antrea Authors.
3+
*
4+
* Licensed under the Apache License, Version 2.0 (the "License");
5+
* you may not use this file except in compliance with the License.
6+
* You may obtain a copy of the License at
7+
*
8+
* http://www.apache.org/licenses/LICENSE-2.0
9+
*
10+
* Unless required by applicable law or agreed to in writing, software
11+
* distributed under the License is distributed on an "AS IS" BASIS,
12+
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
13+
* See the License for the specific language governing permissions and
14+
* limitations under the License.
15+
*/
16+
17+
import { describe, expect, it } from 'vitest';
18+
import { streamFilterKey, FlowStreamFilter } from './flow-stream';
19+
20+
describe('streamFilterKey', () => {
21+
it('matches for different object instances with the same filter', () => {
22+
const a: FlowStreamFilter = {};
23+
const b: FlowStreamFilter = {};
24+
expect(streamFilterKey(a)).toBe(streamFilterKey(b));
25+
});
26+
27+
it('normalizes array field order', () => {
28+
const a: FlowStreamFilter = { namespaces: ['z', 'a'] };
29+
const b: FlowStreamFilter = { namespaces: ['a', 'z'] };
30+
expect(streamFilterKey(a)).toBe(streamFilterKey(b));
31+
});
32+
33+
it('changes when a filter field changes', () => {
34+
const empty: FlowStreamFilter = {};
35+
const withNs: FlowStreamFilter = { namespaces: ['default'] };
36+
expect(streamFilterKey(empty)).not.toBe(streamFilterKey(withNs));
37+
});
38+
});

0 commit comments

Comments
 (0)