antctl observe command

build/charts/antrea/templates/antctl/clusterrole.yaml

@@ -84,6 +84,14 @@ rules:
       - pods
     verbs:
       - list
+  # "antctl observe" tunnels to the Flow Aggregator Pod through the API server's /portforward
+  # subresource when it has no direct network route to the Pod (see pkg/antctl/raw/observe).
+  - apiGroups:
+      - ""
+    resources:
+      - pods/portforward
+    verbs:
+      - create
   - apiGroups:
       - ""
     resources:
@@ -95,10 +103,19 @@ rules:
       - apps
     resources:
       - daemonsets
-      - deployments
       - replicasets
     verbs:
       - list
+  - apiGroups:
+      - apps
+    resources:
+      - deployments
+    verbs:
+      # "get" is needed by "antctl observe --flow-aggregator <namespace>/<deployment-name>" (see
+      # findFlowAggregatorByName in pkg/antctl/raw/observe/discovery.go), which resolves a specific
+      # instance by Deployment name rather than by listing and filtering.
+      - get
+      - list
   - apiGroups:
       - ""
     resources:
@@ -108,5 +125,8 @@ rules:
       - antrea-ca
       - antrea-ipsec-ca
       - antrea-cluster-identity
+      # Read by "antctl observe" to verify the Flow Aggregator's TLS certificate (see
+      # pkg/antctl/raw/observe/connect.go).
+      - flow-aggregator-ca
     verbs:
       - get

build/yamls/antrea-aks.yml

@@ -5415,6 +5415,14 @@ rules:
       - pods
     verbs:
       - list
+  # "antctl observe" tunnels to the Flow Aggregator Pod through the API server's /portforward
+  # subresource when it has no direct network route to the Pod (see pkg/antctl/raw/observe).
+  - apiGroups:
+      - ""
+    resources:
+      - pods/portforward
+    verbs:
+      - create
   - apiGroups:
       - ""
     resources:
@@ -5426,10 +5434,19 @@ rules:
       - apps
     resources:
       - daemonsets
-      - deployments
       - replicasets
     verbs:
       - list
+  - apiGroups:
+      - apps
+    resources:
+      - deployments
+    verbs:
+      # "get" is needed by "antctl observe --flow-aggregator <namespace>/<deployment-name>" (see
+      # findFlowAggregatorByName in pkg/antctl/raw/observe/discovery.go), which resolves a specific
+      # instance by Deployment name rather than by listing and filtering.
+      - get
+      - list
   - apiGroups:
       - ""
     resources:
@@ -5439,6 +5456,9 @@ rules:
       - antrea-ca
       - antrea-ipsec-ca
       - antrea-cluster-identity
+      # Read by "antctl observe" to verify the Flow Aggregator's TLS certificate (see
+      # pkg/antctl/raw/observe/connect.go).
+      - flow-aggregator-ca
     verbs:
       - get
 ---

build/yamls/antrea-eks.yml

@@ -5411,6 +5411,14 @@ rules:
       - pods
     verbs:
       - list
+  # "antctl observe" tunnels to the Flow Aggregator Pod through the API server's /portforward
+  # subresource when it has no direct network route to the Pod (see pkg/antctl/raw/observe).
+  - apiGroups:
+      - ""
+    resources:
+      - pods/portforward
+    verbs:
+      - create
   - apiGroups:
       - ""
     resources:
@@ -5422,10 +5430,19 @@ rules:
       - apps
     resources:
       - daemonsets
-      - deployments
       - replicasets
     verbs:
       - list
+  - apiGroups:
+      - apps
+    resources:
+      - deployments
+    verbs:
+      # "get" is needed by "antctl observe --flow-aggregator <namespace>/<deployment-name>" (see
+      # findFlowAggregatorByName in pkg/antctl/raw/observe/discovery.go), which resolves a specific
+      # instance by Deployment name rather than by listing and filtering.
+      - get
+      - list
   - apiGroups:
       - ""
     resources:
@@ -5435,6 +5452,9 @@ rules:
       - antrea-ca
       - antrea-ipsec-ca
       - antrea-cluster-identity
+      # Read by "antctl observe" to verify the Flow Aggregator's TLS certificate (see
+      # pkg/antctl/raw/observe/connect.go).
+      - flow-aggregator-ca
     verbs:
       - get
 ---

build/yamls/antrea-gke.yml

@@ -5402,6 +5402,14 @@ rules:
       - pods
     verbs:
       - list
+  # "antctl observe" tunnels to the Flow Aggregator Pod through the API server's /portforward
+  # subresource when it has no direct network route to the Pod (see pkg/antctl/raw/observe).
+  - apiGroups:
+      - ""
+    resources:
+      - pods/portforward
+    verbs:
+      - create
   - apiGroups:
       - ""
     resources:
@@ -5413,10 +5421,19 @@ rules:
       - apps
     resources:
       - daemonsets
-      - deployments
       - replicasets
     verbs:
       - list
+  - apiGroups:
+      - apps
+    resources:
+      - deployments
+    verbs:
+      # "get" is needed by "antctl observe --flow-aggregator <namespace>/<deployment-name>" (see
+      # findFlowAggregatorByName in pkg/antctl/raw/observe/discovery.go), which resolves a specific
+      # instance by Deployment name rather than by listing and filtering.
+      - get
+      - list
   - apiGroups:
       - ""
     resources:
@@ -5426,6 +5443,9 @@ rules:
       - antrea-ca
       - antrea-ipsec-ca
       - antrea-cluster-identity
+      # Read by "antctl observe" to verify the Flow Aggregator's TLS certificate (see
+      # pkg/antctl/raw/observe/connect.go).
+      - flow-aggregator-ca
     verbs:
       - get
 ---

build/yamls/antrea-ipsec.yml

@@ -5415,6 +5415,14 @@ rules:
       - pods
     verbs:
       - list
+  # "antctl observe" tunnels to the Flow Aggregator Pod through the API server's /portforward
+  # subresource when it has no direct network route to the Pod (see pkg/antctl/raw/observe).
+  - apiGroups:
+      - ""
+    resources:
+      - pods/portforward
+    verbs:
+      - create
   - apiGroups:
       - ""
     resources:
@@ -5426,10 +5434,19 @@ rules:
       - apps
     resources:
       - daemonsets
-      - deployments
       - replicasets
     verbs:
       - list
+  - apiGroups:
+      - apps
+    resources:
+      - deployments
+    verbs:
+      # "get" is needed by "antctl observe --flow-aggregator <namespace>/<deployment-name>" (see
+      # findFlowAggregatorByName in pkg/antctl/raw/observe/discovery.go), which resolves a specific
+      # instance by Deployment name rather than by listing and filtering.
+      - get
+      - list
   - apiGroups:
       - ""
     resources:
@@ -5439,6 +5456,9 @@ rules:
       - antrea-ca
       - antrea-ipsec-ca
       - antrea-cluster-identity
+      # Read by "antctl observe" to verify the Flow Aggregator's TLS certificate (see
+      # pkg/antctl/raw/observe/connect.go).
+      - flow-aggregator-ca
     verbs:
       - get
 ---

build/yamls/antrea.yml

@@ -5402,6 +5402,14 @@ rules:
       - pods
     verbs:
       - list
+  # "antctl observe" tunnels to the Flow Aggregator Pod through the API server's /portforward
+  # subresource when it has no direct network route to the Pod (see pkg/antctl/raw/observe).
+  - apiGroups:
+      - ""
+    resources:
+      - pods/portforward
+    verbs:
+      - create
   - apiGroups:
       - ""
     resources:
@@ -5413,10 +5421,19 @@ rules:
       - apps
     resources:
       - daemonsets
-      - deployments
       - replicasets
     verbs:
       - list
+  - apiGroups:
+      - apps
+    resources:
+      - deployments
+    verbs:
+      # "get" is needed by "antctl observe --flow-aggregator <namespace>/<deployment-name>" (see
+      # findFlowAggregatorByName in pkg/antctl/raw/observe/discovery.go), which resolves a specific
+      # instance by Deployment name rather than by listing and filtering.
+      - get
+      - list
   - apiGroups:
       - ""
     resources:
@@ -5426,6 +5443,9 @@ rules:
       - antrea-ca
       - antrea-ipsec-ca
       - antrea-cluster-identity
+      # Read by "antctl observe" to verify the Flow Aggregator's TLS certificate (see
+      # pkg/antctl/raw/observe/connect.go).
+      - flow-aggregator-ca
     verbs:
       - get
 ---

ci/kind/test-e2e-kind.sh

@@ -435,7 +435,8 @@ function run_test {
 
   if $flow_visibility; then
       timeout="60m"
-      flow_visibility_args="-run=^(TestFlowExporter|TestFlowAggregator) --flow-visibility --flow-visibility-protocol=$flow_visibility_protocol"
+      #flow_visibility_args="-run=^(TestFlowExporter|TestFlowAggregator) --flow-visibility --flow-visibility-protocol=$flow_visibility_protocol"
+      flow_visibility_args="-run=^(TestFlowExporter|TestFlowAggregator|TestAntctlObserve) --flow-visibility --flow-visibility-protocol=$flow_visibility_protocol"
       # This is needed so that the FlowAggregator is already configured to mount the Secrets
       # necessary for (m)TLS testing. The Secret names must match the ones expected by the e2e tests.
       coverage_flag=""
@@ -444,9 +445,12 @@ function run_test {
       fi
       flow_visibility_manifest_default_args=("--extra-helm-values" "flowCollector.tls.clientSecretName=ipfix-client-cert,flowCollector.tls.caSecretName=ipfix-server-ca")
       $FLOWAGGREGATOR_YML_CMD "${flow_visibility_manifest_default_args[@]}" ${coverage_flag} | docker exec -i kind-control-plane dd of=/root/flow-aggregator.yml
-      # We are generating flow-aggregators for two separate namespaces so we can test forwarding connection details to different destinations
-      $FLOWAGGREGATOR_YML_CMD "${flow_visibility_manifest_default_args[@]}" -n 'flow-aggregator-1' ${coverage_flag} | docker exec -i kind-control-plane dd of=/root/flow-aggregator-1.yml
-      $FLOWAGGREGATOR_YML_CMD "${flow_visibility_manifest_default_args[@]}" -n 'flow-aggregator-2' ${coverage_flag} | docker exec -i kind-control-plane dd of=/root/flow-aggregator-2.yml
+      # We are generating flow-aggregators for two separate namespaces so we can test forwarding connection details to different destinations.
+      # Both use a distinct --release-name: cluster-scoped RBAC object names (ClusterRole/ClusterRoleBinding) are derived from the Helm
+      # release name, not the Namespace, so leaving any two instances (including the single, unnumbered instance above) at the same
+      # release name would make the later manifest overwrite the earlier one's ClusterRoleBinding subjects on apply.
+      $FLOWAGGREGATOR_YML_CMD "${flow_visibility_manifest_default_args[@]}" -n 'flow-aggregator-1' --release-name 'flow-aggregator-1' ${coverage_flag} | docker exec -i kind-control-plane dd of=/root/flow-aggregator-1.yml
+      $FLOWAGGREGATOR_YML_CMD "${flow_visibility_manifest_default_args[@]}" -n 'flow-aggregator-2' --release-name 'flow-aggregator-2' ${coverage_flag} | docker exec -i kind-control-plane dd of=/root/flow-aggregator-2.yml
 
       $HELM template "$FLOW_VISIBILITY_CHART"  | docker exec -i kind-control-plane dd of=/root/flow-visibility.yml
       $HELM template "$FLOW_VISIBILITY_CHART" --set "secureConnection.enable=true" | docker exec -i kind-control-plane dd of=/root/flow-visibility-tls.yml

cmd/flow-aggregator/flow-aggregator.go

@@ -51,7 +51,7 @@ func run(configFile string) error {
 
 	log.StartLogFileNumberMonitor(stopCh)
 
-	k8sClient, err := createK8sClient()
+	k8sClient, restConfig, err := createK8sClient()
 	if err != nil {
 		return fmt.Errorf("error when creating K8s client: %w", err)
 	}
@@ -73,6 +73,7 @@ func run(configFile string) error {
 
 	flowAggregator, err := aggregator.NewFlowAggregator(
 		k8sClient,
+		restConfig,
 		clusterUUID,
 		podStore,
 		nodeStore,
@@ -116,14 +117,14 @@ func run(configFile string) error {
 	return nil
 }
 
-func createK8sClient() (kubernetes.Interface, error) {
+func createK8sClient() (kubernetes.Interface, *rest.Config, error) {
 	config, err := rest.InClusterConfig()
 	if err != nil {
-		return nil, err
+		return nil, nil, err
 	}
 	k8sClient, err := kubernetes.NewForConfig(config)
 	if err != nil {
-		return nil, err
+		return nil, nil, err
 	}
-	return k8sClient, nil
+	return k8sClient, config, nil
 }