Repository: cilium/cilium
Pull request #11595 (2020-05-19 10:08:41)

bpf: use per-cpu scratch space from xdp context to store meta data
Moving to per-CPU scratch space instead avoids cache-misses and allows
to keep the meta buffer always hot in cache. Only really adjust the
pkt meta data space if we push up the stack to signal tc layer that we
handled the service xlation. This improves the performance on my local
testing by roughly +1.2Mpps. We don't perform an expensive clearing of
the scratch space upon entering XDP as usage here is all self-contained
and across tail calls we only read out what we populated beforehand.

Signed-off-by: Daniel Borkmann <daniel@iogearbox.net>


borkmann: test-me-please
borkmann: (ci failures due to quay being down)

Patch:
diff --git a/Documentation/.gitignore b/Documentation/.gitignore
index 5f4085d633..e3a79aafdb 100644
--- a/Documentation/.gitignore
+++ b/Documentation/.gitignore
@@ -1,2 +1,5 @@
 _build
 _api
+_preview
+Pipfile
+Pipfile.lock
diff --git a/Documentation/Makefile b/Documentation/Makefile
index 1b7153297e..de3add4c6d 100644
--- a/Documentation/Makefile
+++ b/Documentation/Makefile
@@ -9,7 +9,7 @@ include ../Makefile.quiet
 default: html
 
 clean:
-	-$(QUIET)rm -rf _build
+	-$(QUIET)rm -rf _build _preview Pipfile Pipfile.lock
 
 builder-image: Dockerfile requirements.txt
 	$(QUIET)tar c requirements.txt Dockerfile \
@@ -61,3 +61,13 @@ run-server: stop-server
 
 stop-server:
 	-$(QUIET)$(CONTAINER_ENGINE) container rm --force --volumes docs-cilium
+
+pipenv:
+	@which pipenv > /dev/null || (echo "Please install pipenv: https://pipenv.pypa.io/en/latest/#install-pipenv-today"; exit 1)
+
+pipenv-install: pipenv
+	pipenv --rm || true
+	pipenv install --three -r requirements.txt
+
+live-preview: pipenv
+	pipenv run sphinx-autobuild -B $(SPHINX_OPTS) -i *.swp -Q . _preview
diff --git a/Documentation/cmdref/cilium-agent.md b/Documentation/cmdref/cilium-agent.md
index 26d308b406..557062ad96 100644
--- a/Documentation/cmdref/cilium-agent.md
+++ b/Documentation/cmdref/cilium-agent.md
@@ -61,6 +61,7 @@ cilium-agent [flags]
       --enable-endpoint-routes                        Use per endpoint routes instead of routing via cilium_host
       --enable-external-ips                           Enable k8s service externalIPs feature (requires enabling enable-node-port) (default true)
       --enable-health-checking                        Enable connectivity health checking (default true)
+      --enable-host-firewall                          Enable host network policies
       --enable-host-port                              Enable k8s hostPort mapping feature (requires enabling enable-node-port) (default true)
       --enable-host-reachable-services                Enable reachability of services for host applications (beta)
       --enable-hubble                                 Enable hubble server
diff --git a/Documentation/requirements.txt b/Documentation/requirements.txt
index 553868ea3b..e787d782d4 100644
--- a/Documentation/requirements.txt
+++ b/Documentation/requirements.txt
@@ -11,7 +11,7 @@ MarkupSafe==1.1.1
 pyenchant==2.0.0
 Pygments==2.4.2
 pytz==2018.7
-PyYAML==4.2b1
+PyYAML==5.3.1
 requests==2.20.0
 six==1.11.0
 snowballstemmer==1.2.1
diff --git a/Documentation/spelling_wordlist.txt b/Documentation/spelling_wordlist.txt
index afbc97d82d..409a03cac2 100644
--- a/Documentation/spelling_wordlist.txt
+++ b/Documentation/spelling_wordlist.txt
@@ -1,3 +1,4 @@
+Admin
 Alemayhu
 Ansible
 Apiserver
@@ -7,6 +8,7 @@ Backporting
 Balancer
 Bertin
 Blanco
+Blog
 Bookinfo
 Borkmann
 Brenden
@@ -22,6 +24,8 @@ Chewbacca
 Chun
 Cloudflare
 Clusterwide
+Committer
+Committers
 DDoS
 DMA'ed
 DNS
@@ -99,6 +103,7 @@ O'Reilly
 Orchestrators
 Pepelnjak
 Pfaff
+Plugins
 Podcasts
 Polytechnique
 Portmap
@@ -138,6 +143,7 @@ Terraform
 Tierney
 Toolchain
 Tracepoints
+Tu
 Twilio
 Userspace
 VM
@@ -157,6 +163,8 @@ Yan
 Yunsong
 Zannoni
 Zhou
+Zookeeper
+admin
 allocator
 allocators
 amd
@@ -165,6 +173,8 @@ api
 apiKeys
 apiVersion
 apiserver
+app
+apps
 arithmetics
 artii
 asm
@@ -186,6 +196,7 @@ bcc
 behaviour
 benefitting
 bgpd
+blog
 bnxt
 bookinfo
 boolean
@@ -213,6 +224,8 @@ clustermesh
 cni
 cnp
 codebase
+committer
+committers
 conf
 config
 conntrack
@@ -248,6 +261,7 @@ dereference
 deserialization
 dev
 devel
+diff
 disassembly
 discoverable
 distro
@@ -255,6 +269,7 @@ distros
 dns
 dockerhub
 dports
+droid
 dryRun
 dst
 dwarfris
@@ -269,6 +284,7 @@ endpointmanager
 etcd
 ethernet
 ethtool
+extendable
 extensibility
 externalIPs
 fallback
@@ -278,6 +294,8 @@ filenames
 filesystem
 filesystems
 firewalling
+followup
+foobar
 foundational
 fqdn
 fromCIDR
@@ -300,6 +318,7 @@ gluecon
 gobpf
 gocheck
 golang
+grep
 hairpinned
 hardcode
 hardcoded
@@ -322,13 +341,16 @@ hypervisors
 iam
 ie
 incrementing
+indices
 ingressing
 init
+inline
 inlined
 inlines
 inlining
 inodes
 integrations
+internet
 io
 ip
 ipam
@@ -448,6 +470,7 @@ nftables
 nginx
 nodeX
 observability
+offline
 onData
 onwards
 opcodes
@@ -455,6 +478,7 @@ openSUSE
 openapi
 openssl
 orchestrator
+org
 pahole
 parserfactory
 parsers
@@ -462,7 +486,10 @@ pc
 pcap
 perf
 pipelining
+playbook
 pluggable
+plugin
+plugins
 pmdabcc
 png
 podcasts
@@ -490,6 +517,7 @@ recv
 recvmsg
 refactoring
 regenerations
+regex
 regexes
 relocations
 reparsing
@@ -554,6 +582,7 @@ sysctl
 sysctls
 systemd
 systemtap
+tarball
 tc
 tcp
 tcpdump
@@ -581,6 +610,7 @@ ubuntu
 udp
 uint
 uninline
+uninstall
 unix
 unmanaged
 untriaged
@@ -624,6 +654,7 @@ www
 xdp
 xfrm
 xml
+xor
 xoring
 xwing
 yaml
diff --git a/Makefile b/Makefile
index 56e26f5766..abe3cdf255 100644
--- a/Makefile
+++ b/Makefile
@@ -596,6 +596,9 @@ update-authors:
 render-docs:
 	$(MAKE) -C Documentation html run-server
 
+render-docs-live-preview:
+	$(MAKE) -C Documentation pipenv-install live-preview
+
 manpages:
 	-rm -r man
 	mkdir -p man
diff --git a/bpf/Makefile b/bpf/Makefile
index 844f737ce4..a53820c9ab 100644
--- a/bpf/Makefile
+++ b/bpf/Makefile
@@ -100,6 +100,8 @@ bpf_overlay.o: bpf_overlay.ll
 	$(QUIET) ${LLC} ${LLC_FLAGS} -filetype=obj -o $@ $(patsubst %.o,%.ll,$@)
 
 NETDEV_OPTIONS = $(LXC_OPTIONS) \
+	-DENABLE_IPV4:-DENABLE_IPV6:-DENABLE_HOST_FIREWALL \
+	-DENABLE_IPV4:-DENABLE_IPV6:-DENABLE_HOST_SERVICES_UDP:-DENABLE_HOST_SERVICES_TCP:-DENABLE_HOST_FIREWALL \
 	-DENABLE_IPV4:-DENABLE_IPV6:-DENABLE_HOST_SERVICES_UDP:-DENABLE_HOST_SERVICES_TCP:-DENABLE_EXTERNAL_IP:-DENABLE_DSR \
 	-DENABLE_IPV4:-DENABLE_IPV6:-DENABLE_IPSEC:-DENABLE_HOST_SERVICES_UDP:-DENABLE_HOST_SERVICES_TCP:-DENABLE_EXTERNAL_IP:-DENABLE_DSR \
 	-DENABLE_IPV4:-DENABLE_IPV6:-DENABLE_HOST_SERVICES_UDP:-DENABLE_HOST_SERVICES_TCP:-DENABLE_NODEPORT:-DENABLE_EXTERNAL_IP:-DENABLE_HOSTPORT \
@@ -107,10 +109,12 @@ NETDEV_OPTIONS = $(LXC_OPTIONS) \
 	-DENABLE_IPV4:-DENABLE_IPV6:-DENABLE_IPSEC:-DENABLE_HOST_SERVICES_UDP:-DENABLE_HOST_SERVICES_TCP:-DENABLE_NODEPORT:-DENABLE_EXTERNAL_IP:-DENABLE_MASQUERADE \
 	-DENABLE_IPV4:-DENABLE_IPV6:-DENABLE_HOST_SERVICES_UDP:-DENABLE_HOST_SERVICES_TCP:-DENABLE_NODEPORT:-DENABLE_EXTERNAL_IP:-DENABLE_DSR \
 	-DENABLE_IPV4:-DENABLE_IPV6:-DENABLE_HOST_SERVICES_UDP:-DENABLE_HOST_SERVICES_TCP:-DENABLE_NODEPORT:-DENABLE_EXTERNAL_IP:-DENABLE_DSR:-DENABLE_DSR_HYBRID \
+	-DENABLE_IPV4:-DENABLE_IPV6:-DENABLE_HOST_SERVICES_UDP:-DENABLE_HOST_SERVICES_TCP:-DENABLE_NODEPORT:-DENABLE_EXTERNAL_IP:-DENABLE_DSR:-DENABLE_DSR_HYBRID:-DENABLE_HOST_FIREWALL \
 	-DENABLE_IPV4:-DENABLE_IPV6:-DENABLE_HOST_SERVICES_UDP:-DENABLE_HOST_SERVICES_TCP:-DENABLE_NODEPORT:-DENABLE_NODEPORT_ACCELERATION:-DENABLE_EXTERNAL_IP:-DENABLE_DSR:-DENABLE_DSR_HYBRID:-DENABLE_PREFILTER \
-	-DENABLE_IPV4:-DENABLE_IPV6:-DENABLE_HOST_SERVICES_UDP:-DENABLE_HOST_SERVICES_TCP:-DENABLE_NODEPORT:-DENABLE_NODEPORT_ACCELERATION:-DENABLE_EXTERNAL_IP:-DENABLE_DSR:-DENABLE_DSR_HYBRID:-DENABLE_PREFILTER:-DENABLE_SESSION_AFFINITY
+	-DENABLE_IPV4:-DENABLE_IPV6:-DENABLE_HOST_SERVICES_UDP:-DENABLE_HOST_SERVICES_TCP:-DENABLE_NODEPORT:-DENABLE_NODEPORT_ACCELERATION:-DENABLE_EXTERNAL_IP:-DENABLE_DSR:-DENABLE_DSR_HYBRID:-DENABLE_PREFILTER:-DENABLE_HOST_FIREWALL \
+	-DENABLE_IPV4:-DENABLE_IPV6:-DENABLE_HOST_SERVICES_UDP:-DENABLE_HOST_SERVICES_TCP:-DENABLE_NODEPORT:-DENABLE_NODEPORT_ACCELERATION:-DENABLE_EXTERNAL_IP:-DENABLE_DSR:-DENABLE_DSR_HYBRID:-DENABLE_PREFILTER:-DENABLE_SESSION_AFFINITY:-DENABLE_HOST_FIREWALL
 
-MAX_NETDEV_OPTIONS = $(MAX_LXC_OPTIONS)
+MAX_NETDEV_OPTIONS = $(MAX_LXC_OPTIONS) -DENABLE_HOST_FIREWALL
 
 bpf_host.ll: bpf_host.c $(LIB)
 	$(QUIET) set -e; \
diff --git a/bpf/bpf_host.c b/bpf/bpf_host.c
index 8e016f92bc..dcd469b5eb 100644
--- a/bpf/bpf_host.c
+++ b/bpf/bpf_host.c
@@ -25,6 +25,10 @@
 /* CB_PROXY_MAGIC overlaps with CB_ENCRYPT_MAGIC */
 #define ENCRYPT_OR_PROXY_MAGIC 0
 
+#ifdef POD_ENDPOINT
+# undef ENABLE_HOST_FIREWALL
+#endif
+
 #include "lib/utils.h"
 #include "lib/common.h"
 #include "lib/arp.h"
@@ -45,6 +49,10 @@
 #include "lib/lb.h"
 #include "lib/nodeport.h"
 #include "lib/eps.h"
+#ifdef ENABLE_HOST_FIREWALL
+# include "lib/policy.h"
+# include "lib/policy_log.h"
+#endif
 
 #if defined(ENABLE_IPV4) || defined(ENABLE_IPV6)
 static __always_inline int rewrite_dmac_to_host(struct __ctx_buff *ctx,
@@ -82,6 +90,28 @@ derive_src_id(const union v6addr *node_ip, struct ipv6hdr *ip6, __u32 *identity)
 	return 0;
 }
 
+# ifdef ENABLE_HOST_FIREWALL
+static __always_inline __u32
+ipcache_lookup_srcid6(struct __ctx_buff *ctx)
+{
+	struct remote_endpoint_info *info = NULL;
+	void *data, *data_end;
+	struct ipv6hdr *ip6;
+	__u32 srcid = 0;
+
+	if (!revalidate_data(ctx, &data, &data_end, &ip6))
+		return DROP_INVALID;
+
+	info = lookup_ip6_remote_endpoint((union v6addr *) &ip6->saddr);
+	if (info != NULL)
+		srcid = info->sec_label;
+	cilium_dbg(ctx, info ? DBG_IP_ID_MAP_SUCCEED6 : DBG_IP_ID_MAP_FAILED6,
+		   ip6->saddr.s6_addr32[3], srcid);
+
+	return srcid;
+}
+# endif /* ENABLE_HOST_FIREWALL */
+
 static __always_inline __u32
 resolve_srcid_ipv6(struct __ctx_buff *ctx, __u32 srcid_from_proxy,
 		   bool from_host)
@@ -129,6 +159,171 @@ resolve_srcid_ipv6(struct __ctx_buff *ctx, __u32 srcid_from_proxy,
 	return src_id;
 }
 
+#ifdef ENABLE_HOST_FIREWALL
+static __always_inline int
+ipv6_host_policy_egress(struct __ctx_buff *ctx, __u32 srcID)
+{
+	int ret, verdict, l3_off = ETH_HLEN, l4_off, hdrlen;
+	struct ct_state ct_state_new = {}, ct_state = {};
+	__u8 policy_match_type = POLICY_MATCH_NONE;
+	struct remote_endpoint_info *info;
+	struct ipv6_ct_tuple tuple = {};
+	__u32 dstID = 0, monitor = 0;
+	union v6addr orig_dip;
+	void *data, *data_end;
+	struct ipv6hdr *ip6;
+
+	/* Only enforce host policies for packets from host IPs. */
+	if (srcID != HOST_ID)
+		return CTX_ACT_OK;
+
+	if (!revalidate_data(ctx, &data, &data_end, &ip6))
+		return DROP_INVALID;
+
+	/* Lookup connection in conntrack map. */
+	tuple.nexthdr = ip6->nexthdr;
+	ipv6_addr_copy(&tuple.daddr, (union v6addr *) &ip6->daddr);
+	ipv6_addr_copy(&orig_dip, (union v6addr *) &ip6->daddr);
+	hdrlen = ipv6_hdrlen(ctx, ETH_HLEN, &tuple.nexthdr);
+	if (hdrlen < 0)
+		return hdrlen;
+	l4_off = l3_off + hdrlen;
+	ret = ct_lookup6(get_ct_map6(&tuple), &tuple, ctx, l4_off, CT_EGRESS,
+			 &ct_state, &monitor);
+	if (ret < 0)
+		return ret;
+
+	/* Retrieve destination identity. */
+	info = lookup_ip6_remote_endpoint(&orig_dip);
+	if (info != NULL && info->sec_label)
+		dstID = info->sec_label;
+	cilium_dbg(ctx, info ? DBG_IP_ID_MAP_SUCCEED6 : DBG_IP_ID_MAP_FAILED6,
+		   orig_dip.p4, dstID);
+
+	/* Perform policy lookup. */
+	verdict = policy_can_egress6(ctx, &tuple, srcID, dstID,
+				     &policy_match_type);
+
+	/* Reply traffic and related are allowed regardless of policy verdict. */
+	if (ret != CT_REPLY && ret != CT_RELATED && verdict < 0) {
+		send_policy_verdict_notify(ctx, dstID, tuple.dport,
+					   tuple.nexthdr, POLICY_EGRESS, 1,
+					   verdict, policy_match_type);
+		return verdict;
+	}
+
+	switch (ret) {
+	case CT_NEW:
+		send_policy_verdict_notify(ctx, dstID, tuple.dport,
+					   tuple.nexthdr, POLICY_EGRESS, 1,
+					   verdict, policy_match_type);
+		ct_state_new.src_sec_id = HOST_ID;
+		ret = ct_create6(get_ct_map6(&tuple), &CT_MAP_ANY6, &tuple,
+				 ctx, CT_EGRESS, &ct_state_new, verdict > 0);
+		if (IS_ERR(ret))
+			return ret;
+		break;
+
+	case CT_ESTABLISHED:
+	case CT_RELATED:
+	case CT_REPLY:
+		break;
+
+	default:
+		return DROP_UNKNOWN_CT;
+	}
+
+	return CTX_ACT_OK;
+}
+
+static __always_inline int
+ipv6_host_policy_ingress(struct __ctx_buff *ctx, __u32 *srcID)
+{
+	struct ct_state ct_state_new = {}, ct_state = {};
+	__u8 policy_match_type = POLICY_MATCH_NONE;
+	__u32 monitor = 0, dstID = WORLD_ID;
+	struct remote_endpoint_info *info;
+	int ret, verdict, l4_off, hdrlen;
+	struct ipv6_ct_tuple tuple = {};
+	union v6addr orig_sip;
+	void *data, *data_end;
+	struct ipv6hdr *ip6;
+
+	if (!revalidate_data(ctx, &data, &data_end, &ip6))
+		return DROP_INVALID;
+
+	/* Retrieve destination identity. */
+	ipv6_addr_copy(&tuple.daddr, (union v6addr *) &ip6->daddr);
+	info = lookup_ip6_remote_endpoint(&tuple.daddr);
+	if (info != NULL && info->sec_label)
+		dstID = info->sec_label;
+	cilium_dbg(ctx, info ? DBG_IP_ID_MAP_SUCCEED6 : DBG_IP_ID_MAP_FAILED6,
+		   tuple.daddr.p4, dstID);
+
+	/* Only enforce host policies for packets to host IPs. */
+	if (dstID != HOST_ID)
+		return CTX_ACT_OK;
+
+	/* Lookup connection in conntrack map. */
+	tuple.nexthdr = ip6->nexthdr;
+	ipv6_addr_copy(&tuple.saddr, (union v6addr *) &ip6->saddr);
+	ipv6_addr_copy(&orig_sip, (union v6addr *) &ip6->saddr);
+	hdrlen = ipv6_hdrlen(ctx, ETH_HLEN, &tuple.nexthdr);
+	if (hdrlen < 0)
+		return hdrlen;
+	l4_off = ETH_HLEN + hdrlen;
+	ret = ct_lookup6(get_ct_map6(&tuple), &tuple, ctx, l4_off, CT_INGRESS,
+			 &ct_state, &monitor);
+	if (ret < 0)
+		return ret;
+
+	/* Retrieve source identity. */
+	info = lookup_ip6_remote_endpoint(&orig_sip);
+	if (info != NULL && info->sec_label)
+		*srcID = info->sec_label;
+	cilium_dbg(ctx, info ? DBG_IP_ID_MAP_SUCCEED6 : DBG_IP_ID_MAP_FAILED6,
+		   orig_sip.p4, *srcID);
+
+	/* Perform policy lookup */
+	verdict = policy_can_access_ingress(ctx, *srcID, dstID, tuple.dport,
+					    tuple.nexthdr, false,
+					    &policy_match_type);
+
+	/* Reply traffic and related are allowed regardless of policy verdict. */
+	if (ret != CT_REPLY && ret != CT_RELATED && verdict < 0) {
+		send_policy_verdict_notify(ctx, *srcID, tuple.dport,
+					   tuple.nexthdr, POLICY_INGRESS, 1,
+					   verdict, policy_match_type);
+		return verdict;
+	}
+
+	switch (ret) {
+	case CT_NEW:
+		send_policy_verdict_notify(ctx, *srcID, tuple.dport,
+					   tuple.nexthdr, POLICY_INGRESS, 1,
+					   verdict, policy_match_type);
+
+		/* Create new entry for connection in conntrack map. */
+		ct_state_new.src_sec_id = *srcID;
+		ct_state_new.node_port = ct_state.node_port;
+		ret = ct_create6(get_ct_map6(&tuple), &CT_MAP_ANY6, &tuple,
+				 ctx, CT_INGRESS, &ct_state_new, verdict > 0);
+		if (IS_ERR(ret))
+			return ret;
+
+	case CT_ESTABLISHED:
+	case CT_RELATED:
+	case CT_REPLY:
+		break;
+
+	default:
+		return DROP_UNKNOWN_CT;
+	}
+
+	return CTX_ACT_OK;
+}
+#endif /* ENABLE_HOST_FIREWALL */
+
 static __always_inline int
 handle_ipv6(struct __ctx_buff *ctx, __u32 secctx, bool from_host)
 {
@@ -137,6 +332,8 @@ handle_ipv6(struct __ctx_buff *ctx, __u32 secctx, bool from_host)
 	struct ipv6hdr *ip6;
 	union v6addr *dst;
 	int ret, l3_off = ETH_HLEN, hdrlen;
+	__u32 __maybe_unused remoteID = WORLD_ID;
+	bool skip_redirect = false;
 	struct endpoint_info *ep;
 	__u8 nexthdr;
 
@@ -153,7 +350,11 @@ handle_ipv6(struct __ctx_buff *ctx, __u32 secctx, bool from_host)
 		}
 #if defined(ENCAP_IFINDEX) || defined(NO_REDIRECT)
 		/* See IPv4 case for NO_REDIRECT comments */
+# ifdef ENABLE_HOST_FIREWALL
+		skip_redirect = true;
+# else
 		return CTX_ACT_OK;
+# endif
 #endif /* ENCAP_IFINDEX || NO_REDIRECT */
 		/* Verifier workaround: modified ctx access. */
 		if (!revalidate_data(ctx, &data, &data_end, &ip6))
@@ -167,14 +368,14 @@ handle_ipv6(struct __ctx_buff *ctx, __u32 secctx, bool from_host)
 		return hdrlen;
 
 #ifdef HANDLE_NS
-	if (unlikely(nexthdr == IPPROTO_ICMPV6)) {
+	if (!skip_redirect && unlikely(nexthdr == IPPROTO_ICMPV6)) {
 		ret = icmp6_handle(ctx, ETH_HLEN, ip6, METRIC_INGRESS);
 		if (IS_ERR(ret))
 			return ret;
 	}
 #endif
 
-	if (from_host) {
+	if (from_host && !skip_redirect) {
 		/* If we are attached to cilium_host at egress, this will
 		 * rewrite the destination mac address to the MAC of cilium_net */
 		ret = rewrite_dmac_to_host(ctx, secctx);
@@ -186,6 +387,24 @@ handle_ipv6(struct __ctx_buff *ctx, __u32 secctx, bool from_host)
 			return DROP_INVALID;
 	}
 
+#ifdef ENABLE_HOST_FIREWALL
+	if (from_host)
+		ret = ipv6_host_policy_egress(ctx, secctx);
+	else
+		ret = ipv6_host_policy_ingress(ctx, &remoteID);
+
+	/* Needed to pass 128k complexity limit. Measured on 4.9 verifier:
+	 * 173251 insns before, 148991 after. */
+	relax_verifier();
+	if (IS_ERR(ret))
+		return ret;
+	if (skip_redirect)
+		return CTX_ACT_OK;
+
+	if (!revalidate_data(ctx, &data, &data_end, &ip6))
+		return DROP_INVALID;
+#endif /* ENABLE_HOST_FIREWALL */
+
 	/* Lookup IPv4 address in list of local endpoints */
 	if ((ep = lookup_ip6_endpoint(ip6)) != NULL) {
 		/* Let through packets to the node-ip so they are
@@ -262,9 +481,11 @@ tail_handle_ipv6(struct __ctx_buff *ctx, bool from_host)
 	ctx_store_meta(ctx, CB_SRC_IDENTITY, 0);
 
 	ret = handle_ipv6(ctx, proxy_identity, from_host);
-	if (IS_ERR(ret))
+	if (IS_ERR(ret)) {
+		relax_verifier();
 		return send_drop_notify_error(ctx, proxy_identity, ret,
 					      CTX_ACT_DROP, METRIC_INGRESS);
+	}
 	return ret;
 }
 
@@ -282,6 +503,28 @@ int tail_handle_ipv6_from_netdev(struct __ctx_buff *ctx)
 #endif /* ENABLE_IPV6 */
 
 #ifdef ENABLE_IPV4
+# ifdef ENABLE_HOST_FIREWALL
+static __always_inline __u32
+ipcache_lookup_srcid4(struct __ctx_buff *ctx)
+{
+	struct remote_endpoint_info *info = NULL;
+	void *data, *data_end;
+	struct iphdr *ip4;
+	__u32 srcid = 0;
+
+	if (!revalidate_data(ctx, &data, &data_end, &ip4))
+		return DROP_INVALID;
+
+	info = lookup_ip4_remote_endpoint(ip4->saddr);
+	if (info != NULL)
+		srcid = info->sec_label;
+	cilium_dbg(ctx, info ? DBG_IP_ID_MAP_SUCCEED4 : DBG_IP_ID_MAP_FAILED4,
+		   ip4->saddr, srcid);
+
+	return srcid;
+}
+# endif /* ENABLE_HOST_FIREWALL */
+
 static __always_inline __u32
 resolve_srcid_ipv4(struct __ctx_buff *ctx, __u32 srcid_from_proxy,
 		   bool from_host)
@@ -333,14 +576,181 @@ resolve_srcid_ipv4(struct __ctx_buff *ctx, __u32 srcid_from_proxy,
 	return src_id;
 }
 
+#ifdef ENABLE_HOST_FIREWALL
+static __always_inline int
+ipv4_host_policy_egress(struct __ctx_buff *ctx, __u32 srcID)
+{
+	struct ct_state ct_state_new = {}, ct_state = {};
+	int ret, verdict, l4_off, l3_off = ETH_HLEN;
+	__u8 policy_match_type = POLICY_MATCH_NONE;
+	struct remote_endpoint_info *info;
+	struct ipv4_ct_tuple tuple = {};
+	__u32 dstID = 0, monitor = 0;
+	void *data, *data_end;
+	struct iphdr *ip4;
+
+	/* Only enforce host policies for packets from host IPs. */
+	if (srcID != HOST_ID)
+		return CTX_ACT_OK;
+
+	if (!revalidate_data(ctx, &data, &data_end, &ip4))
+		return DROP_INVALID;
+
+	/* Lookup connection in conntrack map. */
+	tuple.nexthdr = ip4->protocol;
+	tuple.daddr = ip4->daddr;
+	tuple.saddr = ip4->saddr;
+	l4_off = l3_off + ipv4_hdrlen(ip4);
+	ret = ct_lookup4(get_ct_map4(&tuple), &tuple, ctx, l4_off, CT_EGRESS,
+			 &ct_state, &monitor);
+	if (ret < 0)
+		return ret;
+
+	/* Retrieve destination identity. */
+	info = lookup_ip4_remote_endpoint(ip4->daddr);
+	if (info != NULL && info->sec_label)
+		dstID = info->sec_label;
+	cilium_dbg(ctx, info ? DBG_IP_ID_MAP_SUCCEED4 : DBG_IP_ID_MAP_FAILED4,
+		   ip4->daddr, dstID);
+
+	/* Perform policy lookup. */
+	verdict = policy_can_egress4(ctx, &tuple, srcID, dstID,
+				     &policy_match_type);
+
+	/* Reply traffic and related are allowed regardless of policy verdict. */
+	if (ret != CT_REPLY && ret != CT_RELATED && verdict < 0) {
+		send_policy_verdict_notify(ctx, dstID, tuple.dport,
+					   tuple.nexthdr, POLICY_EGRESS, 0,
+					   verdict, policy_match_type);
+		return verdict;
+	}
+
+	switch (ret) {
+	case CT_NEW:
+		send_policy_verdict_notify(ctx, dstID, tuple.dport,
+					   tuple.nexthdr, POLICY_EGRESS, 0,
+					   verdict, policy_match_type);
+		ct_state_new.src_sec_id = HOST_ID;
+		ret = ct_create4(get_ct_map4(&tuple), &CT_MAP_ANY4, &tuple,
+				 ctx, CT_EGRESS, &ct_state_new, verdict > 0);
+		if (IS_ERR(ret))
+			return ret;
+		break;
+
+	case CT_ESTABLISHED:
+	case CT_RELATED:
+	case CT_REPLY:
+		break;
+
+	default:
+		return DROP_UNKNOWN_CT;
+	}
+
+	return CTX_ACT_OK;
+}
+
+static __always_inline int
+ipv4_host_policy_ingress(struct __ctx_buff *ctx, __u32 *srcID)
+{
+	struct ct_state ct_state_new = {}, ct_state = {};
+	int ret, verdict, l4_off, l3_off = ETH_HLEN;
+	__u8 policy_match_type = POLICY_MATCH_NONE;
+	__u32 monitor = 0, dstID = WORLD_ID;
+	struct remote_endpoint_info *info;
+	struct ipv4_ct_tuple tuple = {};
+	bool is_untracked_fragment = false;
+	void *data, *data_end;
+	struct iphdr *ip4;
+
+	if (!revalidate_data(ctx, &data, &data_end, &ip4))
+		return DROP_INVALID;
+
+	/* Retrieve destination identity. */
+	info = lookup_ip4_remote_endpoint(ip4->daddr);
+	if (info != NULL && info->sec_label)
+		dstID = info->sec_label;
+	cilium_dbg(ctx, info ? DBG_IP_ID_MAP_SUCCEED4 : DBG_IP_ID_MAP_FAILED4,
+		   ip4->daddr, dstID);
+
+	/* Only enforce host policies for packets to host IPs. */
+	if (dstID != HOST_ID)
+		return CTX_ACT_OK;
+
+	/* Lookup connection in conntrack map. */
+	tuple.nexthdr = ip4->protocol;
+	tuple.daddr = ip4->daddr;
+	tuple.saddr = ip4->saddr;
+	l4_off = l3_off + ipv4_hdrlen(ip4);
+#ifndef IPV4_FRAGMENTS
+	/* Indicate that this is a datagram fragment for which we cannot
+	 * retrieve L4 ports. Do not set flag if we support fragmentation.
+	 */
+	is_untracked_fragment = ipv4_is_fragment(ip4);
+#endif
+	ret = ct_lookup4(get_ct_map4(&tuple), &tuple, ctx, l4_off, CT_INGRESS,
+			 &ct_state, &monitor);
+	if (ret < 0)
+		return ret;
+
+	/* Retrieve source identity. */
+	info = lookup_ip4_remote_endpoint(ip4->saddr);
+	if (info != NULL && info->sec_label)
+		*srcID = info->sec_label;
+	cilium_dbg(ctx, info ? DBG_IP_ID_MAP_SUCCEED4 : DBG_IP_ID_MAP_FAILED4,
+		   ip4->saddr, *srcID);
+
+	/* Perform policy lookup */
+	verdict = policy_can_access_ingress(ctx, *srcID, dstID, tuple.dport,
+					    tuple.nexthdr,
+					    is_untracked_fragment,
+					    &policy_match_type);
+
+	/* Reply traffic and related are allowed regardless of policy verdict. */
+	if (ret != CT_REPLY && ret != CT_RELATED && verdict < 0) {
+		send_policy_verdict_notify(ctx, *srcID, tuple.dport,
+					   tuple.nexthdr, POLICY_INGRESS, 0,
+					   verdict, policy_match_type);
+		return verdict;
+	}
+
+	switch (ret) {
+	case CT_NEW:
+		send_policy_verdict_notify(ctx, *srcID, tuple.dport,
+					   tuple.nexthdr, POLICY_INGRESS, 0,
+					   verdict, policy_match_type);
+
+		/* Create new entry for connection in conntrack map. */
+		ct_state_new.src_sec_id = *srcID;
+		ct_state_new.node_port = ct_state.node_port;
+		ret = ct_create4(get_ct_map4(&tuple), &CT_MAP_ANY4, &tuple,
+				 ctx, CT_INGRESS, &ct_state_new, verdict > 0);
+		if (IS_ERR(ret))
+			return ret;
+
+	case CT_ESTABLISHED:
+	case CT_RELATED:
+	case CT_REPLY:
+		break;
+
+	default:
+		return DROP_UNKNOWN_CT;
+	}
+
+	return CTX_ACT_OK;
+}
+#endif /* ENABLE_HOST_FIREWALL */
+
 static __always_inline int
 handle_ipv4(struct __ctx_buff *ctx, __u32 secctx, bool from_host)
 {
 	struct remote_endpoint_info *info = NULL;
+	__u32 __maybe_unused remoteID = 0;
 	struct ipv4_ct_tuple tuple = {};
+	bool skip_redirect = false;
 	struct endpoint_info *ep;
 	void *data, *data_end;
 	struct iphdr *ip4;
+	int ret;
 
 	if (!revalidate_data(ctx, &data, &data_end, &ip4))
 		return DROP_INVALID;
@@ -349,7 +759,7 @@ handle_ipv4(struct __ctx_buff *ctx, __u32 secctx, bool from_host)
 	if (!from_host) {
 		if (ctx_get_xfer(ctx) != XFER_PKT_NO_SVC &&
 		    !bpf_skip_nodeport(ctx)) {
-			int ret = nodeport_lb4(ctx, secctx);
+			ret = nodeport_lb4(ctx, secctx);
 			if (ret < 0)
 				return ret;
 		}
@@ -359,7 +769,11 @@ handle_ipv4(struct __ctx_buff *ctx, __u32 secctx, bool from_host)
 		 * This makes a second reply from the endpoint to be MASQUERADEd
 		 * or to be DROPed by k8s's "--ctstate INVALID -j DROP"
 		 * depending via which interface it was inputed. */
+# ifdef ENABLE_HOST_FIREWALL
+		skip_redirect = true;
+# else
 		return CTX_ACT_OK;
+#endif /* ENABLE_HOST_FIREWALL */
 #endif /* ENCAP_IFINDEX || NO_REDIRECT */
 		/* Verifier workaround: modified ctx access. */
 		if (!revalidate_data(ctx, &data, &data_end, &ip4))
@@ -369,9 +783,7 @@ handle_ipv4(struct __ctx_buff *ctx, __u32 secctx, bool from_host)
 
 	tuple.nexthdr = ip4->protocol;
 
-	if (from_host) {
-		int ret;
-
+	if (from_host && !skip_redirect) {
 		/* If we are attached to cilium_host at egress, this will
 		 * rewrite the destination mac address to the MAC of cilium_net */
 		ret = rewrite_dmac_to_host(ctx, secctx);
@@ -383,6 +795,22 @@ handle_ipv4(struct __ctx_buff *ctx, __u32 secctx, bool from_host)
 			return DROP_INVALID;
 	}
 
+#ifdef ENABLE_HOST_FIREWALL
+	if (from_host)
+		/* We're on the egress path of cilium_host. */
+		ret = ipv4_host_policy_egress(ctx, secctx);
+	else
+		/* We're on the ingress path of the native device. */
+		ret = ipv4_host_policy_ingress(ctx, &remoteID);
+	if (IS_ERR(ret))
+		return ret;
+	if (skip_redirect)
+		return CTX_ACT_OK;
+
+	if (!revalidate_data(ctx, &data, &data_end, &ip4))
+		return DROP_INVALID;
+#endif /* ENABLE_HOST_FIREWALL */
+
 	/* Lookup IPv4 address in list of local endpoints and host IPs */
 	if ((ep = lookup_ip4_endpoint(ip4)) != NULL) {
 		/* Let through packets to the node-ip so they are
@@ -402,9 +830,9 @@ handle_ipv4(struct __ctx_buff *ctx, __u32 secctx, bool from_host)
 #ifdef ENCAP_IFINDEX
 	info = ipcache_lookup4(&IPCACHE_MAP, ip4->daddr, V4_CACHE_KEY_LEN);
 	if (info != NULL && info->tunnel_endpoint != 0) {
-		int ret = encap_and_redirect_with_nodeid(ctx, info->tunnel_endpoint,
-							 info->key,
-							 secctx, TRACE_PAYLOAD_LEN);
+		ret = encap_and_redirect_with_nodeid(ctx, info->tunnel_endpoint,
+						     info->key, secctx,
+						     TRACE_PAYLOAD_LEN);
 
 		if (ret == IPSEC_ENDPOINT)
 			return CTX_ACT_OK;
@@ -413,7 +841,6 @@ handle_ipv4(struct __ctx_buff *ctx, __u32 secctx, bool from_host)
 	} else {
 		/* IPv4 lookup key: daddr & IPV4_MASK */
 		struct endpoint_key key = {};
-		int ret;
 
 		key.ip4 = ip4->daddr & IPV4_MASK;
 		key.family = ENDPOINT_KEY_IPV4;
@@ -469,9 +896,11 @@ tail_handle_ipv4(struct __ctx_buff *ctx, bool from_host)
 	ctx_store_meta(ctx, CB_SRC_IDENTITY, 0);
 
 	ret = handle_ipv4(ctx, proxy_identity, from_host);
-	if (IS_ERR(ret))
+	if (IS_ERR(ret)) {
+		relax_verifier();
 		return send_drop_notify_error(ctx, proxy_identity,
 					      ret, CTX_ACT_DROP, METRIC_INGRESS);
+	}
 	return ret;
 }
 
@@ -716,8 +1145,13 @@ do_netdev(struct __ctx_buff *ctx, __u16 proto, bool from_host)
 		                              CTX_ACT_OK, METRIC_INGRESS);
 #endif
 	default:
+#ifdef ENABLE_HOST_FIREWALL
+		ret = send_drop_notify_error(ctx, identity, DROP_UNKNOWN_L3,
+					     CTX_ACT_DROP, METRIC_INGRESS);
+#else
 		/* Pass unknown traffic to the stack */
 		ret = CTX_ACT_OK;
+#endif /* ENABLE_HOST_FIREWALL */
 	}
 
 	return ret;
@@ -736,10 +1170,17 @@ handle_netdev(struct __ctx_buff *ctx, bool from_host)
 	__u16 proto;
 
 	if (!validate_ethertype(ctx, &proto)) {
+#ifdef ENABLE_HOST_FIREWALL
+		int ret = DROP_UNSUPPORTED_L2;
+
+		return send_drop_notify(ctx, SECLABEL, WORLD_ID, 0, ret,
+					CTX_ACT_DROP, METRIC_EGRESS);
+#else
 		send_trace_notify(ctx, TRACE_TO_STACK, HOST_ID, 0, 0, 0,
 				  REASON_FORWARDED, 0);
 		/* Pass unknown traffic to the stack */
 		return CTX_ACT_OK;
+#endif /* ENABLE_HOST_FIREWALL */
 	}
 
 	return do_netdev(ctx, proto, from_host);
@@ -760,19 +1201,65 @@ int from_host(struct __ctx_buff *ctx)
 __section("to-netdev")
 int to_netdev(struct __ctx_buff *ctx __maybe_unused)
 {
-	/* Cannot compile the section out entriely, test/bpf/verifier-test.sh
-	 * workaround.
-	 */
+	__u32 __maybe_unused remoteID = WORLD_ID, srcID = 0;
+	__u16 __maybe_unused proto = 0;
 	int ret = CTX_ACT_OK;
+
 #if defined(ENABLE_NODEPORT) && \
 	(!defined(ENABLE_DSR) || \
 	 (defined(ENABLE_DSR) && defined(ENABLE_DSR_HYBRID)))
-	if ((ctx->mark & MARK_MAGIC_SNAT_DONE) == MARK_MAGIC_SNAT_DONE)
-		return CTX_ACT_OK;
-	ret = nodeport_nat_fwd(ctx, false);
-	if (IS_ERR(ret))
-		return send_drop_notify_error(ctx, 0, ret, CTX_ACT_DROP, METRIC_EGRESS);
+	if ((ctx->mark & MARK_MAGIC_SNAT_DONE) != MARK_MAGIC_SNAT_DONE) {
+		ret = nodeport_nat_fwd(ctx, false);
+		if (IS_ERR(ret))
+			return send_drop_notify_error(ctx, 0, ret,
+						      CTX_ACT_DROP,
+						      METRIC_EGRESS);
+	}
 #endif
+
+#ifdef ENABLE_HOST_FIREWALL
+	if (!proto && !validate_ethertype(ctx, &proto)) {
+		ret = DROP_UNSUPPORTED_L2;
+		goto out;
+	}
+
+	policy_clear_mark(ctx);
+
+	switch (proto) {
+# if defined ENABLE_ARP_PASSTHROUGH || defined ENABLE_ARP_RESPONDER
+	case bpf_htons(ETH_P_ARP):
+		ret = CTX_ACT_OK;
+		break;
+# endif
+# ifdef ENABLE_IPV6
+	case bpf_htons(ETH_P_IPV6):
+		/* to-netdev is attached to the egress path of the native
+		 * device. */
+		srcID = ipcache_lookup_srcid6(ctx);
+		ret = ipv6_host_policy_egress(ctx, srcID);
+		break;
+# endif
+# ifdef ENABLE_IPV4
+	case bpf_htons(ETH_P_IP):
+		/* to-netdev is attached to the egress path of the native
+		 * device. */
+		srcID = ipcache_lookup_srcid4(ctx);
+		ret = ipv4_host_policy_egress(ctx, srcID);
+		break;
+# endif
+	default:
+		ret = DROP_UNKNOWN_L3;
+		break;
+	}
+
+out:
+	if (IS_ERR(ret))
+		return send_drop_notify_error(ctx, srcID, ret, CTX_ACT_DROP,
+					      METRIC_EGRESS);
+#else
+	ret = CTX_ACT_OK;
+#endif /* ENABLE_HOST_FIREWALL */
+
 	return ret;
 }
 
@@ -780,14 +1267,15 @@ __section("to-host")
 int to_host(struct __ctx_buff *ctx)
 {
 	__u32 magic = ctx_load_meta(ctx, ENCRYPT_OR_PROXY_MAGIC);
+	__u16 __maybe_unused proto = 0;
 	int ret = CTX_ACT_OK;
-	__u32 src_label = 0;
 	bool traced = false;
+	__u32 srcID = 0;
 
 	if ((magic & MARK_MAGIC_HOST_MASK) == MARK_MAGIC_ENCRYPT) {
 		ctx->mark = magic; // CB_ENCRYPT_MAGIC
-		src_label = ctx_load_meta(ctx, CB_ENCRYPT_IDENTITY);
-		set_identity_mark(ctx, src_label);
+		srcID = ctx_load_meta(ctx, CB_ENCRYPT_IDENTITY);
+		set_identity_mark(ctx, srcID);
 	} else if ((magic & 0xFFFF) == MARK_MAGIC_TO_PROXY) {
 		/* Upper 16 bits may carry proxy port number */
 		__be16 port = magic >> 16;
@@ -800,9 +1288,46 @@ int to_host(struct __ctx_buff *ctx)
 	}
 
 	if (!traced)
-		send_trace_notify(ctx, TRACE_TO_STACK, src_label, 0, 0,
+		send_trace_notify(ctx, TRACE_TO_STACK, srcID, 0, 0,
 				  CILIUM_IFINDEX, ret, 0);
 
+#ifdef ENABLE_HOST_FIREWALL
+	if (!validate_ethertype(ctx, &proto)) {
+		ret = DROP_UNSUPPORTED_L2;
+		goto out;
+	}
+
+	policy_clear_mark(ctx);
+
+	switch (proto) {
+# if defined ENABLE_ARP_PASSTHROUGH || defined ENABLE_ARP_RESPONDER
+	case bpf_htons(ETH_P_ARP):
+		ret = CTX_ACT_OK;
+		break;
+# endif
+# ifdef ENABLE_IPV6
+	case bpf_htons(ETH_P_IPV6):
+		ret = ipv6_host_policy_ingress(ctx, &srcID);
+		break;
+# endif
+# ifdef ENABLE_IPV4
+	case bpf_htons(ETH_P_IP):
+		ret = ipv4_host_policy_ingress(ctx, &srcID);
+		break;
+# endif
+	default:
+		ret = DROP_UNKNOWN_L3;
+		break;
+	}
+
+out:
+	if (IS_ERR(ret))
+		return send_drop_notify_error(ctx, srcID, ret, CTX_ACT_DROP,
+					      METRIC_INGRESS);
+#else
+	ret = CTX_ACT_OK;
+#endif /* ENABLE_HOST_FIREWALL */
+
 	return ret;
 }
 
diff --git a/bpf/bpf_lxc.c b/bpf/bpf_lxc.c
index d33b7421d5..a2a0772ff5 100644
--- a/bpf/bpf_lxc.c
+++ b/bpf/bpf_lxc.c
@@ -4,8 +4,8 @@
 #include <bpf/ctx/skb.h>
 #include <bpf/api.h>
 
-#include <node_config.h>
 #include <ep_config.h>
+#include <node_config.h>
 
 #include <linux/icmpv6.h>
 
@@ -171,7 +171,8 @@ skip_service_lookup:
 	/* If the packet is in the establishing direction and it's destined
 	 * within the cluster, it must match policy or be dropped. If it's
 	 * bound for the host/outside, perform the CIDR policy check. */
-	verdict = policy_can_egress6(ctx, tuple, *dstID, &policy_match_type);
+	verdict = policy_can_egress6(ctx, tuple, SECLABEL, *dstID,
+				     &policy_match_type);
 	if (ret != CT_REPLY && ret != CT_RELATED && verdict < 0) {
 		send_policy_verdict_notify(ctx, *dstID, tuple->dport, tuple->nexthdr,
 						POLICY_EGRESS, 1, verdict, policy_match_type);
@@ -525,7 +526,8 @@ skip_service_lookup:
 	/* If the packet is in the establishing direction and it's destined
 	 * within the cluster, it must match policy or be dropped. If it's
 	 * bound for the host/outside, perform the CIDR policy check. */
-	verdict = policy_can_egress4(ctx, &tuple, *dstID, &policy_match_type);
+	verdict = policy_can_egress4(ctx, &tuple, SECLABEL, *dstID,
+				     &policy_match_type);
 
 	if (ret != CT_REPLY && ret != CT_RELATED && verdict < 0) {
 		send_policy_verdict_notify(ctx, *dstID, tuple.dport, tuple.nexthdr,
@@ -866,8 +868,9 @@ ipv6_policy(struct __ctx_buff *ctx, int ifindex, __u32 src_label, __u8 *reason,
 			return ret2;
 	}
 
-	verdict = policy_can_access_ingress(ctx, src_label, tuple.dport,
-			tuple.nexthdr, false, &policy_match_type);
+	verdict = policy_can_access_ingress(ctx, src_label, SECLABEL,
+					    tuple.dport, tuple.nexthdr, false,
+					    &policy_match_type);
 
 	/* Reply packets and related packets are allowed, all others must be
 	 * permitted by policy */
@@ -1084,8 +1087,8 @@ ipv4_policy(struct __ctx_buff *ctx, int ifindex, __u32 src_label, __u8 *reason,
 			return ret2;
 	}
 
-	verdict = policy_can_access_ingress(ctx, src_label, tuple.dport,
-					    tuple.nexthdr,
+	verdict = policy_can_access_ingress(ctx, src_label, SECLABEL,
+					    tuple.dport, tuple.nexthdr,
 					    is_untracked_fragment,
 					    &policy_match_type);
 
diff --git a/bpf/bpf_xdp.c b/bpf/bpf_xdp.c
index 5508d4b5d5..effff7b6f5 100644
--- a/bpf/bpf_xdp.c
+++ b/bpf/bpf_xdp.c
@@ -71,7 +71,6 @@ struct bpf_elf_map __section_maps CIDR6_LMAP_NAME = {
 static __always_inline __maybe_unused int
 bpf_xdp_exit(struct __ctx_buff *ctx, const int verdict)
 {
-	/* Undo meta data, so GRO can perform natural aggregation. */
 	if (verdict == CTX_ACT_OK) {
 		__u32 meta_xfer = ctx_load_meta(ctx, XFER_MARKER);
 
@@ -79,10 +78,13 @@ bpf_xdp_exit(struct __ctx_buff *ctx, const int verdict)
 		 * does not break packet trains in GRO.
 		 */
 		if (meta_xfer) {
-			ctx_adjust_meta(ctx, META_PIVOT - sizeof(__u32));
-			ctx_store_meta(ctx, 0, meta_xfer);
-		} else {
-			ctx_adjust_meta(ctx, META_PIVOT);
+			if (!ctx_adjust_meta(ctx, -(int)sizeof(meta_xfer))) {
+				__u32 *data_meta = ctx_data_meta(ctx);
+				__u32 *data = ctx_data(ctx);
+
+				if (!ctx_no_room(data_meta + 1, data))
+					data_meta[0] = meta_xfer;
+			}
 		}
 	}
 
@@ -224,8 +226,6 @@ static __always_inline int check_filters(struct __ctx_buff *ctx)
 
 	if (!validate_ethertype(ctx, &proto))
 		return CTX_ACT_OK;
-	if (ctx_adjust_meta(ctx, -META_PIVOT))
-		return CTX_ACT_OK;
 
 	ctx_store_meta(ctx, XFER_MARKER, 0);
 	bpf_skip_nodeport_clear(ctx);
diff --git a/bpf/include/bpf/api.h b/bpf/include/bpf/api.h
index eac5faa91e..aa93db023c 100644
--- a/bpf/include/bpf/api.h
+++ b/bpf/include/bpf/api.h
@@ -16,23 +16,6 @@
 #include "verifier.h"
 #include "tailcall.h"
 #include "errno.h"
-
-#define PIN_NONE		0
-#define PIN_OBJECT_NS		1
-#define PIN_GLOBAL_NS		2
-
-struct bpf_elf_map {
-	__u32 type;
-	__u32 size_key;
-	__u32 size_value;
-	__u32 max_elem;
-	__u32 flags;
-	__u32 id;
-	__u32 pinning;
-#ifdef SOCKMAP
-	__u32 inner_id;
-	__u32 inner_idx;
-#endif
-};
+#include "loader.h"
 
 #endif /* __BPF_API__ */
diff --git a/bpf/include/bpf/compiler.h b/bpf/include/bpf/compiler.h
index 263640bb8e..52328fafe0 100644
--- a/bpf/include/bpf/compiler.h
+++ b/bpf/include/bpf/compiler.h
@@ -44,6 +44,10 @@
 # define unlikely(X)		__builtin_expect(!!(X), 0)
 #endif
 
+#ifndef always_succeeds		/* Mainly for documentation purpose. */
+# define always_succeeds(X)	likely(X)
+#endif
+
 #undef __always_inline		/* stddef.h defines its own */
 #define __always_inline		inline __attribute__((always_inline))
 
diff --git a/bpf/include/bpf/ctx/xdp.h b/bpf/include/bpf/ctx/xdp.h
index 80de26699c..2ac0ea2207 100644
--- a/bpf/include/bpf/ctx/xdp.h
+++ b/bpf/include/bpf/ctx/xdp.h
@@ -12,6 +12,8 @@
 #include "common.h"
 #include "../helpers_xdp.h"
 #include "../builtins.h"
+#include "../section.h"
+#include "../loader.h"
 
 #define CTX_ACT_OK			XDP_PASS
 #define CTX_ACT_DROP			XDP_DROP
@@ -270,31 +272,33 @@ ctx_full_len(const struct xdp_md *ctx)
 	return ctx_data_end(ctx) - ctx_data(ctx);
 }
 
+struct bpf_elf_map __section_maps cilium_xdp_scratch = {
+	.type		= BPF_MAP_TYPE_PERCPU_ARRAY,
+	.size_key	= sizeof(int),
+	.size_value	= META_PIVOT,
+	.pinning	= PIN_GLOBAL_NS,
+	.max_elem	= 1,
+};
+
 static __always_inline __maybe_unused void
-ctx_store_meta(struct xdp_md *ctx, const __u64 off, __u32 datum)
+ctx_store_meta(struct xdp_md *ctx __maybe_unused, const __u64 off, __u32 datum)
 {
-	__u32 *data_meta = ctx_data_meta(ctx);
-	void *data = ctx_data(ctx);
+	__u32 zero = 0, *data_meta = map_lookup_elem(&cilium_xdp_scratch, &zero);
 
-	if (!ctx_no_room(data_meta + off + 1, data)) {
+	if (always_succeeds(data_meta))
 		data_meta[off] = datum;
-	} else {
-		build_bug_on((off + 1) * sizeof(__u32) > META_PIVOT);
-	}
+	build_bug_on((off + 1) * sizeof(__u32) > META_PIVOT);
 }
 
 static __always_inline __maybe_unused __u32
-ctx_load_meta(const struct xdp_md *ctx, const __u64 off)
+ctx_load_meta(const struct xdp_md *ctx __maybe_unused, const __u64 off)
 {
-	__u32 *data_meta = ctx_data_meta(ctx);
-	void *data = ctx_data(ctx);
+	__u32 zero = 0, *data_meta = map_lookup_elem(&cilium_xdp_scratch, &zero);
 
-	if (!ctx_no_room(data_meta + off + 1, data)) {
+	if (always_succeeds(data_meta))
 		return data_meta[off];
-	} else {
-		build_bug_on((off + 1) * sizeof(__u32) > META_PIVOT);
-		return 0;
-	}
+	build_bug_on((off + 1) * sizeof(__u32) > META_PIVOT);
+	return 0;
 }
 
 static __always_inline __maybe_unused __u32
diff --git a/bpf/include/bpf/loader.h b/bpf/include/bpf/loader.h
new file mode 100644
index 0000000000..a81eed5d8a
--- /dev/null
+++ b/bpf/include/bpf/loader.h
@@ -0,0 +1,27 @@
+/* SPDX-License-Identifier: GPL-2.0 */
+/* Copyright (C) 2016-2020 Authors of Cilium */
+
+#ifndef __BPF_LOADER__
+#define __BPF_LOADER__
+
+#include <linux/types.h>
+
+#define PIN_NONE		0
+#define PIN_OBJECT_NS		1
+#define PIN_GLOBAL_NS		2
+
+struct bpf_elf_map {
+	__u32 type;
+	__u32 size_key;
+	__u32 size_value;
+	__u32 max_elem;
+	__u32 flags;
+	__u32 id;
+	__u32 pinning;
+#ifdef SOCKMAP
+	__u32 inner_id;
+	__u32 inner_idx;
+#endif
+};
+
+#endif /* __BPF_LOADER__ */
diff --git a/bpf/lib/policy.h b/bpf/lib/policy.h
index fa7a37b452..3c95459fc5 100644
--- a/bpf/lib/policy.h
+++ b/bpf/lib/policy.h
@@ -79,13 +79,13 @@ account(struct __ctx_buff *ctx, struct policy_entry *policy)
 }
 
 static __always_inline int
-__policy_can_access(const void *map, struct __ctx_buff *ctx, __u32 identity,
-		    __u16 dport, __u8 proto, int dir,
+__policy_can_access(const void *map, struct __ctx_buff *ctx, __u32 localID,
+		    __u32 remoteID, __u16 dport, __u8 proto, int dir,
 		    bool is_untracked_fragment, __u8 *match_type)
 {
 	struct policy_entry *policy;
 	struct policy_key key = {
-		.sec_label = identity,
+		.sec_label = remoteID,
 		.dport = dport,
 		.protocol = proto,
 		.egress = !dir,
@@ -119,7 +119,7 @@ __policy_can_access(const void *map, struct __ctx_buff *ctx, __u32 identity,
 		/* Start with L3/L4 lookup. */
 		policy = map_lookup_elem(map, &key);
 		if (likely(policy)) {
-			cilium_dbg3(ctx, DBG_L4_CREATE, identity, SECLABEL,
+			cilium_dbg3(ctx, DBG_L4_CREATE, remoteID, localID,
 				    dport << 16 | proto);
 
 			account(ctx, policy);
@@ -135,7 +135,7 @@ __policy_can_access(const void *map, struct __ctx_buff *ctx, __u32 identity,
 			*match_type = POLICY_MATCH_L4_ONLY;
 			return policy->proxy_port;
 		}
-		key.sec_label = identity;
+		key.sec_label = remoteID;
 	}
 
 	/* If L4 policy check misses, fall back to L3. */
@@ -168,10 +168,11 @@ __policy_can_access(const void *map, struct __ctx_buff *ctx, __u32 identity,
 
 /**
  * Determine whether the policy allows this traffic on ingress.
- * @arg ctx			Packet to allow or deny
- * @arg src_identity		Source security identity for this packet
- * @arg dport			Destination port of this packet
- * @arg proto			L3 Protocol of this packet
+ * @arg ctx		Packet to allow or deny
+ * @arg srcID		Source security identity for this packet
+ * @arg dstID		Destination security identity for this packet
+ * @arg dport		Destination port of this packet
+ * @arg proto		L3 Protocol of this packet
  * @arg is_untracked_fragment	True if packet is a TCP/UDP datagram fragment
  *				AND IPv4 fragment tracking is disabled
  * @arg match_type		Pointer to store layers used for policy match
@@ -182,19 +183,19 @@ __policy_can_access(const void *map, struct __ctx_buff *ctx, __u32 identity,
  *   - Negative error code if the packet should be dropped
  */
 static __always_inline int
-policy_can_access_ingress(struct __ctx_buff *ctx, __u32 src_identity,
+policy_can_access_ingress(struct __ctx_buff *ctx, __u32 srcID, __u32 dstID,
 			  __u16 dport, __u8 proto, bool is_untracked_fragment,
 			  __u8 *match_type)
 {
 	int ret;
 
-	ret = __policy_can_access(&POLICY_MAP, ctx, src_identity, dport,
+	ret = __policy_can_access(&POLICY_MAP, ctx, dstID, srcID, dport,
 				  proto, CT_INGRESS, is_untracked_fragment,
 				  match_type);
 	if (ret >= CTX_ACT_OK)
 		return ret;
 
-	cilium_dbg(ctx, DBG_POLICY_DENIED, src_identity, SECLABEL);
+	cilium_dbg(ctx, DBG_POLICY_DENIED, srcID, dstID);
 
 #ifdef POLICY_AUDIT_MODE
 	if (IS_ERR(ret)) {
@@ -216,8 +217,8 @@ static __always_inline bool is_encap(__u16 dport, __u8 proto)
 #endif
 
 static __always_inline int
-policy_can_egress(struct __ctx_buff *ctx, __u32 identity, __u16 dport, __u8 proto,
-		  __u8 *match_type)
+policy_can_egress(struct __ctx_buff *ctx, __u32 srcID, __u32 dstID,
+		  __u16 dport, __u8 proto, __u8 *match_type)
 {
 	int ret;
 
@@ -225,11 +226,11 @@ policy_can_egress(struct __ctx_buff *ctx, __u32 identity, __u16 dport, __u8 prot
 	if (is_encap(dport, proto))
 		return DROP_ENCAP_PROHIBITED;
 #endif
-	ret = __policy_can_access(&POLICY_MAP, ctx, identity, dport, proto,
+	ret = __policy_can_access(&POLICY_MAP, ctx, srcID, dstID, dport, proto,
 				  CT_EGRESS, false, match_type);
 	if (ret >= 0)
 		return ret;
-	cilium_dbg(ctx, DBG_POLICY_DENIED, SECLABEL, identity);
+	cilium_dbg(ctx, DBG_POLICY_DENIED, srcID, dstID);
 #ifdef POLICY_AUDIT_MODE
 	if (IS_ERR(ret)) {
 		ret = CTX_ACT_OK;
@@ -240,16 +241,20 @@ policy_can_egress(struct __ctx_buff *ctx, __u32 identity, __u16 dport, __u8 prot
 
 static __always_inline int policy_can_egress6(struct __ctx_buff *ctx,
 					      const struct ipv6_ct_tuple *tuple,
-					      __u32 identity, __u8 *match_type)
+					      __u32 srcID, __u32 dstID,
+					      __u8 *match_type)
 {
-	return policy_can_egress(ctx, identity, tuple->dport, tuple->nexthdr, match_type);
+	return policy_can_egress(ctx, srcID, dstID, tuple->dport,
+				 tuple->nexthdr, match_type);
 }
 
 static __always_inline int policy_can_egress4(struct __ctx_buff *ctx,
 					      const struct ipv4_ct_tuple *tuple,
-					      __u32 identity, __u8 *match_type)
+					      __u32 srcID, __u32 dstID,
+					      __u8 *match_type)
 {
-	return policy_can_egress(ctx, identity, tuple->dport, tuple->nexthdr, match_type);
+	return policy_can_egress(ctx, srcID, dstID, tuple->dport,
+				 tuple->nexthdr, match_type);
 }
 
 /**
diff --git a/bpf/node_config.h b/bpf/node_config.h
index 7e4b362a7d..f9312501c4 100644
--- a/bpf/node_config.h
+++ b/bpf/node_config.h
@@ -103,10 +103,11 @@ DEFINE_IPV6(HOST_IP, 0xbe, 0xef, 0x0, 0x0, 0x0, 0x0, 0x0, 0x1, 0x0, 0x0, 0xa, 0x
 #ifndef SKIP_DEBUG
 #define LB_DEBUG
 #endif
+#define POLICY_VERDICT_NOTIFY
 #define MONITOR_AGGREGATION 5
 #define MTU 1500
 #define EPHEMERAL_MIN 32768
-#ifdef ENABLE_NODEPORT
+#if defined(ENABLE_NODEPORT) || defined(ENABLE_HOST_FIREWALL)
 #define CT_MAP_TCP6 test_cilium_ct_tcp6_65535
 #define CT_MAP_ANY6 test_cilium_ct_any6_65535
 #define CT_MAP_TCP4 test_cilium_ct_tcp4_65535
@@ -115,7 +116,7 @@ DEFINE_IPV6(HOST_IP, 0xbe, 0xef, 0x0, 0x0, 0x0, 0x0, 0x0, 0x1, 0x0, 0x0, 0xa, 0x
 #define CT_MAP_SIZE_ANY 4096
 #define CONNTRACK
 #define CONNTRACK_ACCOUNTING
-#endif /* ENABLE_NODEPORT */
+#endif /* ENABLE_NODEPORT || ENABLE_HOST_FIREWALL */
 
 #ifdef ENABLE_NODEPORT
 #ifdef ENABLE_IPV4
diff --git a/daemon/cmd/daemon_main.go b/daemon/cmd/daemon_main.go
index 56df480939..98304ab277 100644
--- a/daemon/cmd/daemon_main.go
+++ b/daemon/cmd/daemon_main.go
@@ -526,6 +526,9 @@ func init() {
 	flags.Bool(option.EnableSessionAffinity, false, "Enable support for service session affinity")
 	option.BindEnv(option.EnableSessionAffinity)
 
+	flags.Bool(option.EnableHostFirewall, false, "Enable host network policies")
+	option.BindEnv(option.EnableHostFirewall)
+
 	flags.String(option.LibDir, defaults.LibraryPath, "Directory path to store runtime build environment")
 	option.BindEnv(option.LibDir)
 
@@ -1099,6 +1102,7 @@ func initEnv(cmd *cobra.Command) {
 		option.Config.EncryptInterface = link
 	}
 
+	checkHostFirewallWithEgressLB()
 	initKubeProxyReplacementOptions()
 	if option.Config.EnableNodePort {
 		if err := node.InitNodePortAddrs(option.Config.Devices); err != nil {
@@ -1132,6 +1136,17 @@ func initEnv(cmd *cobra.Command) {
 		}
 	}
 
+	if option.Config.EnableHostFirewall && len(option.Config.Devices) == 0 {
+		device, err := linuxdatapath.NodeDeviceNameWithDefaultRoute()
+		if err != nil {
+			msg := "Host firewall's external facing device could not be determined. Use --%s to specify."
+			log.WithError(err).Fatalf(msg, option.Device)
+		}
+		log.WithField(logfields.Interface, device).
+			Info("Using auto-derived device for host firewall")
+		option.Config.Devices = []string{device}
+	}
+
 	// If there is one device specified, use it to derive better default
 	// allocation prefixes
 	node.InitDefaultPrefix(option.Config.Devices)
@@ -1553,6 +1568,30 @@ func initClockSourceOption() {
 	}
 }
 
+func checkHostFirewallWithEgressLB() {
+	// Egress LB is enabled in datapath under condition:
+	// ENABLE_SERVICES && (!ENABLE_HOST_SERVICES_FULL || \
+	//                    (ENABLE_EXTERNAL_IP && !BPF_HAVE_NETNS_COOKIE))
+	// We can't enable both egress LB and host firewall on kernels <4.14 due to
+	// the verifier complexity limit, at 96k instructions.
+	var netnsCookieSupport bool
+	pm := probes.NewProbeManager()
+	h1, h2 := pm.GetHelpers("cgroup_sock_addr"), pm.GetHelpers("cgroup_sock")
+	if _, ok := h1["bpf_get_netns_cookie"]; ok {
+		if _, ok := h2["bpf_get_netns_cookie"]; ok {
+			netnsCookieSupport = true
+		}
+	}
+	egressLBEnabled := !option.Config.DisableK8sServices &&
+		(!option.Config.EnableHostServicesTCP || !option.Config.EnableHostServicesUDP ||
+			(option.Config.EnableExternalIPs && !netnsCookieSupport))
+	if option.Config.EnableHostFirewall && egressLBEnabled {
+		log.Warn("Enabling both BPF-based east-west load balancing and the host firewall isn't supported yet. Disabling east-west load balancing.")
+		option.Config.DisableK8sServices = true
+		option.Config.KubeProxyReplacement = option.KubeProxyReplacementDisabled
+	}
+}
+
 func initKubeProxyReplacementOptions() {
 	if option.Config.KubeProxyReplacement != option.KubeProxyReplacementStrict &&
 		option.Config.KubeProxyReplacement != option.KubeProxyReplacementPartial &&
diff --git a/examples/policies/l3/host/host-policy.json b/examples/policies/l3/host/host-policy.json
new file mode 100644
index 0000000000..9377ea30fe
--- /dev/null
+++ b/examples/policies/l3/host/host-policy.json
@@ -0,0 +1,41 @@
+[
+  {
+    "nodeSelector": {},
+    "ingress": [
+      {
+        "toPorts": [
+          {
+            "ports": [
+              {"port": "6443", "protocol": "TCP"},
+              {"port": "22", "protocol": "TCP"}
+            ]
+          }
+        ]
+      }
+    ],
+    "egress": [
+      {
+        "toPorts": [
+          {
+            "ports": [
+              {"port": "4240", "protocol": "TCP"},
+              {"port": "8080", "protocol": "TCP"}
+            ]
+          }
+        ]
+      },
+      {
+        "toCIDR": [
+          "8.8.0.0/16"
+        ],
+        "toPorts": [
+          {
+            "ports": [
+              {"port": "53", "protocol": "UDP"}
+            ]
+          }
+        ]
+      }
+    ]
+  }
+]
diff --git a/operator/api.go b/operator/api.go
index 4c78f0cdf8..ade1f337a2 100644
--- a/operator/api.go
+++ b/operator/api.go
@@ -17,10 +17,13 @@ package main
 import (
 	"context"
 	"fmt"
+	"net"
 	"net/http"
+	"syscall"
 
 	"github.com/cilium/cilium/pkg/k8s"
 	"github.com/cilium/cilium/pkg/kvstore"
+	"golang.org/x/sys/unix"
 )
 
 // startServer starts an api server listening on the given address.
@@ -49,8 +52,14 @@ func startServer(shutdownSignal <-chan struct{}, allSystemsGo <-chan struct{}, a
 		srv := &http.Server{Addr: addr}
 		errCh := make(chan error, 1)
 
+		lc := net.ListenConfig{Control: setsockoptReuseAddrAndPort}
+		ln, err := lc.Listen(context.Background(), "tcp", addr)
+		if err != nil {
+			log.WithError(err).Fatalf("Unable to listen on %s for healthz apiserver", addr)
+		}
+
 		go func() {
-			err := srv.ListenAndServe()
+			err := srv.Serve(ln)
 			if err != nil {
 				errCh <- err
 				errs <- err
@@ -117,3 +126,24 @@ func checkStatus() error {
 
 	return nil
 }
+
+// setsockoptReuseAddrAndPort sets SO_REUSEADDR and SO_REUSEPORT
+func setsockoptReuseAddrAndPort(network, address string, c syscall.RawConn) error {
+	var soerr error
+	if err := c.Control(func(su uintptr) {
+		s := int(su)
+		// Allow reuse of recently-used addresses. This socket option is
+		// set by default on listeners in Go's net package, see
+		// net setDefaultListenerSockopts
+		soerr = unix.SetsockoptInt(s, unix.SOL_SOCKET, unix.SO_REUSEADDR, 1)
+		if soerr != nil {
+			return
+		}
+		// Allow reuse of recently-used ports. This gives the agent a
+		// better change to re-bind upon restarts.
+		soerr = unix.SetsockoptInt(s, unix.SOL_SOCKET, unix.SO_REUSEPORT, 1)
+	}); err != nil {
+		return err
+	}
+	return soerr
+}
diff --git a/operator/main.go b/operator/main.go
index 2982393366..522a8d936b 100644
--- a/operator/main.go
+++ b/operator/main.go
@@ -256,7 +256,6 @@ func runOperator(cmd *cobra.Command) {
 								shortCutK8sCache: &sc,
 								k8sCache:         &k8sSvcCache,
 							}
-							break
 						case errors.IsNotFound(err):
 							scopedLog.Error("Service not found in k8s")
 						default:
@@ -327,5 +326,4 @@ func runOperator(cmd *cobra.Command) {
 	<-shutdownSignal
 	// graceful exit
 	log.Info("Received termination signal. Shutting down")
-	return
 }
diff --git a/pkg/datapath/linux/config/config.go b/pkg/datapath/linux/config/config.go
index 546bf67d82..57c96545ab 100644
--- a/pkg/datapath/linux/config/config.go
+++ b/pkg/datapath/linux/config/config.go
@@ -261,6 +261,10 @@ func (h *HeaderfileWriter) WriteNodeConfig(w io.Writer, cfg *datapath.LocalNodeC
 		cDefinesMap["NODEPORT_PORT_MAX_NAT"] = "65535"
 	}
 
+	if option.Config.EnableHostFirewall {
+		cDefinesMap["ENABLE_HOST_FIREWALL"] = "1"
+	}
+
 	if option.Config.EncryptInterface != "" {
 		link, err := netlink.LinkByName(option.Config.EncryptInterface)
 		if err == nil {
diff --git a/pkg/datapath/loader/loader.go b/pkg/datapath/loader/loader.go
index 2f16f165d5..d73934fbab 100644
--- a/pkg/datapath/loader/loader.go
+++ b/pkg/datapath/loader/loader.go
@@ -234,7 +234,7 @@ func (l *Loader) reloadHostDatapath(ctx context.Context, ep datapath.Endpoint, o
 		interfaceNames = append(interfaceNames, device)
 		symbols = append(symbols, symbolFromHostNetdevEp)
 		directions = append(directions, dirIngress)
-		if option.Config.EnableNodePort {
+		if option.Config.EnableNodePort || option.Config.EnableHostFirewall {
 			interfaceNames = append(interfaceNames, device)
 			symbols = append(symbols, symbolToHostNetdevEp)
 			directions = append(directions, dirEgress)
diff --git a/pkg/hubble/api/v1/const.go b/pkg/hubble/api/v1/const.go
index 368c36633d..2dd72a9ee0 100644
--- a/pkg/hubble/api/v1/const.go
+++ b/pkg/hubble/api/v1/const.go
@@ -1,4 +1,5 @@
 // Copyright 2019 Authors of Hubble
+// Copyright 2020 Authors of Cilium
 //
 // Licensed under the Apache License, Version 2.0 (the "License");
 // you may not use this file except in compliance with the License.
@@ -17,7 +18,4 @@ package v1
 const (
 	// ObserverServiceName is the name of the observer service for the grpc health check
 	ObserverServiceName = "hubble.server.Observer"
-
-	// K8sNamespaceTag is the label tag which denotes the namespace.
-	K8sNamespaceTag = "k8s:io.kubernetes.pod.namespace"
 )
diff --git a/pkg/hubble/api/v1/endpoint.go b/pkg/hubble/api/v1/endpoint.go
deleted file mode 100644
index af1ca9efff..0000000000
--- a/pkg/hubble/api/v1/endpoint.go
+++ /dev/null
@@ -1,225 +0,0 @@
-// Copyright 2019 Authors of Hubble
-//
-// Licensed under the Apache License, Version 2.0 (the "License");
-// you may not use this file except in compliance with the License.
-// You may obtain a copy of the License at
-//
-//     http://www.apache.org/licenses/LICENSE-2.0
-//
-// Unless required by applicable law or agreed to in writing, software
-// distributed under the License is distributed on an "AS IS" BASIS,
-// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
-// See the License for the specific language governing permissions and
-// limitations under the License.
-
-package v1
-
-import (
-	"net"
-)
-
-// EndpointsHandler defines an interface for interacting with Cilium endpoints.
-type EndpointsHandler interface {
-	SyncEndpoints([]*Endpoint)
-	UpdateEndpoint(*Endpoint)
-	FindEPs(epID uint64, ns, pod string) []Endpoint
-	GetEndpoint(ip net.IP) (endpoint *Endpoint, ok bool)
-	DeleteEndpoint(*Endpoint)
-	GetEndpointByContainerID(id string) (*Endpoint, bool)
-	GetEndpointByPodName(namespace string, name string) (*Endpoint, bool)
-}
-
-// EqualsByID compares if the receiver's endpoint has the same ID, PodName and
-// PodNamespace.
-func (e *Endpoint) EqualsByID(o *Endpoint) bool {
-	if o == nil {
-		return false
-	}
-	return (e.ID == o.ID && e.PodName == "" && e.PodNamespace == "") ||
-		e.ID == o.ID &&
-			e.PodName == o.PodName &&
-			e.PodNamespace == o.PodNamespace
-}
-
-// DeepCopy returns a deep copy of this endpoint.
-func (e *Endpoint) DeepCopy() *Endpoint {
-	result := *e
-	if e.ContainerIDs != nil {
-		result.ContainerIDs = make([]string, len(e.ContainerIDs))
-		copy(result.ContainerIDs, e.ContainerIDs)
-	}
-	if e.IPv4 != nil {
-		result.IPv4 = make(net.IP, len(e.IPv4))
-		copy(result.IPv4, e.IPv4)
-	}
-	if e.IPv6 != nil {
-		result.IPv6 = make(net.IP, len(e.IPv6))
-		copy(result.IPv6, e.IPv6)
-	}
-	if e.Labels != nil {
-		result.Labels = make([]string, len(e.Labels))
-		copy(result.Labels, e.Labels)
-	}
-	return &result
-}
-
-// SyncEndpoints adds the given list of endpoints to the internal endpoint
-// slice.
-func (es *Endpoints) SyncEndpoints(newEps []*Endpoint) {
-	if len(newEps) == 0 {
-		return
-	}
-	es.mutex.Lock()
-	defer es.mutex.Unlock()
-	// Add the endpoint to the list of endpoints.
-	for _, updatedEp := range newEps {
-		es.updateEndpoint(updatedEp)
-	}
-	// some endpoints were deleted, remove them
-	if len(es.eps) != len(newEps) {
-		for _, ep := range es.eps {
-			found := false
-			for _, newEp := range newEps {
-				if newEp.EqualsByID(ep) {
-					found = true
-					break
-				}
-			}
-			if !found {
-				es.deleteEndpoint(ep)
-			}
-		}
-	}
-}
-
-// FindEPs returns all the EPs that have the given epID or the given namespace
-// or the given podName (running in the given namespace).
-func (es *Endpoints) FindEPs(epID uint64, namespace string, podName string) []Endpoint {
-	var eps []Endpoint
-	es.mutex.RLock()
-	defer es.mutex.RUnlock()
-	for _, ep := range es.eps {
-		// If is the endpoint ID we are looking for
-		if (epID != 0 && ep.ID == epID) ||
-			// The pod name is the one we are looking for
-			(podName != "" && (ep.PodName == podName && ep.PodNamespace == namespace)) ||
-			// The pod namespace is in the same namespace we are looking for
-			(podName == "" && ep.PodNamespace == namespace) {
-
-			eps = append(eps, *ep)
-		}
-	}
-
-	return eps
-}
-
-// setFrom sets all fields from the given endpoint 'o' in receiver's endpoint.
-func (e *Endpoint) setFrom(o *Endpoint) {
-	if o.ContainerIDs != nil {
-		e.ContainerIDs = o.ContainerIDs
-	}
-	if o.ID != 0 {
-		e.ID = o.ID
-	}
-	if o.IPv4 != nil {
-		e.IPv4 = o.IPv4
-	}
-	if o.IPv6 != nil {
-		e.IPv6 = o.IPv6
-	}
-	if len(o.Labels) != 0 {
-		e.Labels = o.Labels
-	}
-	if o.PodName != "" {
-		e.PodName = o.PodName
-	}
-	if o.PodNamespace != "" {
-		e.PodNamespace = o.PodNamespace
-	}
-}
-
-func (es *Endpoints) updateEndpoint(updateEp *Endpoint) {
-	for _, ep := range es.eps {
-		// Update endpoint if the ID is the same *and* the podName and
-		// podNamespace do not exist, otherwise check if the given updateEp
-		// equals to ep.
-		if ep.EqualsByID(updateEp) {
-			ep.setFrom(updateEp)
-			return
-		}
-	}
-	// If we haven't found it, then we need to add it to the list of
-	// endpoints
-	es.eps = append(es.eps, updateEp)
-}
-
-// UpdateEndpoint updates the given endpoint if already exists in the slice of
-// endpoints. If the endpoint does not exists, it is appended to the slice of
-// endpoints.
-func (es *Endpoints) UpdateEndpoint(updateEp *Endpoint) {
-	es.mutex.Lock()
-	defer es.mutex.Unlock()
-	es.updateEndpoint(updateEp)
-}
-
-// GetEndpoint returns the endpoint that has the given ip.
-func (es *Endpoints) GetEndpoint(ip net.IP) (endpoint *Endpoint, ok bool) {
-	es.mutex.RLock()
-	defer es.mutex.RUnlock()
-	for _, ep := range es.eps {
-		if ep.IPv4.Equal(ip) || ep.IPv6.Equal(ip) {
-			return ep.DeepCopy(), true
-		}
-	}
-	return
-}
-
-// DeleteEndpoint deletes the given endpoint if present in the endpoints slice.
-func (es *Endpoints) DeleteEndpoint(del *Endpoint) {
-	es.mutex.Lock()
-	defer es.mutex.Unlock()
-	es.deleteEndpoint(del)
-}
-
-func (es *Endpoints) deleteEndpoint(del *Endpoint) {
-	for i, ep := range es.eps {
-		if ep.EqualsByID(del) {
-			// deleting without preserving order avoids doing a new allocation
-			es.eps[i] = es.eps[len(es.eps)-1]
-			es.eps[len(es.eps)-1] = nil // avoid memory leak
-			es.eps = es.eps[:len(es.eps)-1]
-			break
-		}
-	}
-}
-
-// GetEndpointInfo returns the endpoint info that has the given ip.
-func (es *Endpoints) GetEndpointInfo(ip net.IP) (endpoint EndpointInfo, ok bool) {
-	return es.GetEndpoint(ip)
-}
-
-// GetEndpointByContainerID returns the endpoint that has the given container ID.
-func (es *Endpoints) GetEndpointByContainerID(id string) (*Endpoint, bool) {
-	es.mutex.RLock()
-	defer es.mutex.RUnlock()
-	for _, ep := range es.eps {
-		for _, containerID := range ep.ContainerIDs {
-			if id == containerID {
-				return ep.DeepCopy(), true
-			}
-		}
-	}
-	return nil, false
-}
-
-// GetEndpointByPodName returns the endpoint with the given pod name.
-func (es *Endpoints) GetEndpointByPodName(namespace string, name string) (*Endpoint, bool) {
-	es.mutex.RLock()
-	defer es.mutex.RUnlock()
-	for _, ep := range es.eps {
-		if ep.PodNamespace == namespace && ep.PodName == name {
-			return ep.DeepCopy(), true
-		}
-	}
-	return nil, false
-}
diff --git a/pkg/hubble/api/v1/http.go b/pkg/hubble/api/v1/http.go
deleted file mode 100644
index 687e2d9ddb..0000000000
--- a/pkg/hubble/api/v1/http.go
+++ /dev/null
@@ -1,29 +0,0 @@
-// Copyright 2019 Authors of Hubble
-//
-// Licensed under the Apache License, Version 2.0 (the "License");
-// you may not use this file except in compliance with the License.
-// You may obtain a copy of the License at
-//
-//     http://www.apache.org/licenses/LICENSE-2.0
-//
-// Unless required by applicable law or agreed to in writing, software
-// distributed under the License is distributed on an "AS IS" BASIS,
-// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
-// See the License for the specific language governing permissions and
-// limitations under the License.
-
-package v1
-
-import (
-	pb "github.com/cilium/cilium/api/v1/flow"
-)
-
-// LooseCompareHTTP returns true if both HTTP flows are loosely identical. This
-// means that the following fields must match:
-//  - Code
-//  - Method
-//  - Url
-//  - Protocol
-func LooseCompareHTTP(a, b *pb.HTTP) bool {
-	return a.Code == b.Code && a.Method == b.Method && a.Url == b.Url && a.Protocol == b.Protocol
-}
diff --git a/pkg/hubble/api/v1/interface.go b/pkg/hubble/api/v1/interface.go
index a21953ec19..fc7ba34927 100644
--- a/pkg/hubble/api/v1/interface.go
+++ b/pkg/hubble/api/v1/interface.go
@@ -1,4 +1,5 @@
 // Copyright 2019 Authors of Hubble
+// Copyright 2020 Authors of Cilium
 //
 // Licensed under the Apache License, Version 2.0 (the "License");
 // you may not use this file except in compliance with the License.
@@ -15,7 +16,7 @@
 package v1
 
 import (
-	pb "github.com/cilium/cilium/api/v1/flow"
+	flowpb "github.com/cilium/cilium/api/v1/flow"
 	"github.com/cilium/cilium/pkg/identity"
 
 	"github.com/golang/protobuf/proto"
@@ -26,29 +27,29 @@ import (
 type Flow interface {
 	proto.Message
 	GetTime() *timestamp.Timestamp
-	GetVerdict() pb.Verdict
+	GetVerdict() flowpb.Verdict
 	GetDropReason() uint32
-	GetEthernet() *pb.Ethernet
-	GetIP() *pb.IP
-	GetL4() *pb.Layer4
-	GetSource() *pb.Endpoint
-	GetDestination() *pb.Endpoint
-	GetType() pb.FlowType
+	GetEthernet() *flowpb.Ethernet
+	GetIP() *flowpb.IP
+	GetL4() *flowpb.Layer4
+	GetSource() *flowpb.Endpoint
+	GetDestination() *flowpb.Endpoint
+	GetType() flowpb.FlowType
 	GetNodeName() string
 	GetSourceNames() []string
 	GetDestinationNames() []string
-	GetL7() *pb.Layer7
+	GetL7() *flowpb.Layer7
 	GetReply() bool
-	GetEventType() *pb.CiliumEventType
-	GetSourceService() *pb.Service
-	GetDestinationService() *pb.Service
-	GetTrafficDirection() pb.TrafficDirection
+	GetEventType() *flowpb.CiliumEventType
+	GetSourceService() *flowpb.Service
+	GetDestinationService() *flowpb.Service
+	GetTrafficDirection() flowpb.TrafficDirection
 	GetPolicyMatchType() uint32
 	GetSummary() string
 }
 
 // This ensures that the protobuf definition implements the interface
-var _ Flow = &pb.Flow{}
+var _ Flow = &flowpb.Flow{}
 
 // EndpointInfo defines readable fields of a Cilium endpoint.
 type EndpointInfo interface {
diff --git a/pkg/hubble/api/v1/kafka.go b/pkg/hubble/api/v1/kafka.go
deleted file mode 100644
index 8045e0985a..0000000000
--- a/pkg/hubble/api/v1/kafka.go
+++ /dev/null
@@ -1,28 +0,0 @@
-// Copyright 2019 Authors of Hubble
-//
-// Licensed under the Apache License, Version 2.0 (the "License");
-// you may not use this file except in compliance with the License.
-// You may obtain a copy of the License at
-//
-//     http://www.apache.org/licenses/LICENSE-2.0
-//
-// Unless required by applicable law or agreed to in writing, software
-// distributed under the License is distributed on an "AS IS" BASIS,
-// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
-// See the License for the specific language governing permissions and
-// limitations under the License.
-
-package v1
-
-import (
-	pb "github.com/cilium/cilium/api/v1/flow"
-)
-
-// CompareKafka returns true if both Kafka flows are identical
-func CompareKafka(a, b *pb.Kafka) bool {
-	return a.ErrorCode == b.ErrorCode &&
-		a.ApiVersion == b.ApiVersion &&
-		a.ApiKey == b.ApiKey &&
-		a.CorrelationId == b.CorrelationId &&
-		a.Topic == b.Topic
-}
diff --git a/pkg/hubble/api/v1/types.go b/pkg/hubble/api/v1/types.go
index fe01a965d5..37816c7d9f 100644
--- a/pkg/hubble/api/v1/types.go
+++ b/pkg/hubble/api/v1/types.go
@@ -19,7 +19,6 @@ import (
 
 	pb "github.com/cilium/cilium/api/v1/flow"
 	"github.com/cilium/cilium/pkg/identity"
-	"github.com/cilium/cilium/pkg/lock"
 
 	"github.com/golang/protobuf/ptypes/timestamp"
 )
@@ -80,16 +79,3 @@ func (ev *Event) GetFlow() Flow {
 	}
 	return nil
 }
-
-// Endpoints is a slice of endpoints and their cached dns queries protected by a mutex.
-type Endpoints struct {
-	mutex lock.RWMutex
-	eps   []*Endpoint
-}
-
-// NewEndpoints returns a new *Endpoints.
-func NewEndpoints() *Endpoints {
-	return &Endpoints{
-		eps: []*Endpoint{},
-	}
-}
diff --git a/pkg/hubble/filters/event_type.go b/pkg/hubble/filters/event_type.go
index c9f53b4179..939ccaee98 100644
--- a/pkg/hubble/filters/event_type.go
+++ b/pkg/hubble/filters/event_type.go
@@ -17,11 +17,11 @@ package filters
 import (
 	"context"
 
-	pb "github.com/cilium/cilium/api/v1/flow"
+	flowpb "github.com/cilium/cilium/api/v1/flow"
 	v1 "github.com/cilium/cilium/pkg/hubble/api/v1"
 )
 
-func filterByEventType(types []*pb.EventTypeFilter) FilterFunc {
+func filterByEventType(types []*flowpb.EventTypeFilter) FilterFunc {
 	return func(ev *v1.Event) bool {
 		event := ev.GetFlow().GetEventType()
 		if event == nil {
@@ -48,7 +48,7 @@ func filterByEventType(types []*pb.EventTypeFilter) FilterFunc {
 type EventTypeFilter struct{}
 
 // OnBuildFilter builds an event type filter
-func (e *EventTypeFilter) OnBuildFilter(ctx context.Context, ff *pb.FlowFilter) ([]FilterFunc, error) {
+func (e *EventTypeFilter) OnBuildFilter(ctx context.Context, ff *flowpb.FlowFilter) ([]FilterFunc, error) {
 	var fs []FilterFunc
 
 	types := ff.GetEventType()
diff --git a/pkg/hubble/filters/filters.go b/pkg/hubble/filters/filters.go
index 0b7f4bfefa..6a0e475d66 100644
--- a/pkg/hubble/filters/filters.go
+++ b/pkg/hubble/filters/filters.go
@@ -17,7 +17,7 @@ package filters
 import (
 	"context"
 
-	pb "github.com/cilium/cilium/api/v1/flow"
+	flowpb "github.com/cilium/cilium/api/v1/flow"
 	v1 "github.com/cilium/cilium/pkg/hubble/api/v1"
 )
 
@@ -76,14 +76,14 @@ func (fs FilterFuncs) MatchNone(ev *v1.Event) bool {
 
 // OnBuildFilter is invoked while building a flow filter
 type OnBuildFilter interface {
-	OnBuildFilter(context.Context, *pb.FlowFilter) ([]FilterFunc, error)
+	OnBuildFilter(context.Context, *flowpb.FlowFilter) ([]FilterFunc, error)
 }
 
 // OnBuildFilterFunc implements OnBuildFilter for a single function
-type OnBuildFilterFunc func(context.Context, *pb.FlowFilter) ([]FilterFunc, error)
+type OnBuildFilterFunc func(context.Context, *flowpb.FlowFilter) ([]FilterFunc, error)
 
 // OnBuildFilter is invoked while building a flow filter
-func (f OnBuildFilterFunc) OnBuildFilter(ctx context.Context, flow *pb.FlowFilter) ([]FilterFunc, error) {
+func (f OnBuildFilterFunc) OnBuildFilter(ctx context.Context, flow *flowpb.FlowFilter) ([]FilterFunc, error) {
 	return f(ctx, flow)
 }
 
@@ -91,7 +91,7 @@ func (f OnBuildFilterFunc) OnBuildFilter(ctx context.Context, flow *pb.FlowFilte
 // - the FilterFunc to be used to filter packets based on the requested
 //   FlowFilter;
 // - an error in case something went wrong.
-func BuildFilter(ctx context.Context, ff *pb.FlowFilter, auxFilters []OnBuildFilter) (FilterFuncs, error) {
+func BuildFilter(ctx context.Context, ff *flowpb.FlowFilter, auxFilters []OnBuildFilter) (FilterFuncs, error) {
 	var fs []FilterFunc
 
 	for _, f := range auxFilters {
@@ -112,7 +112,7 @@ func BuildFilter(ctx context.Context, ff *pb.FlowFilter, auxFilters []OnBuildFil
 // - the FilterFunc to be used to filter packets based on the requested
 //   FlowFilter;
 // - an error in case something went wrong.
-func BuildFilterList(ctx context.Context, ff []*pb.FlowFilter, auxFilters []OnBuildFilter) (FilterFuncs, error) {
+func BuildFilterList(ctx context.Context, ff []*flowpb.FlowFilter, auxFilters []OnBuildFilter) (FilterFuncs, error) {
 	filterList := make([]FilterFunc, 0, len(ff))
 
 	for _, flowFilter := range ff {
diff --git a/pkg/hubble/filters/fqdn.go b/pkg/hubble/filters/fqdn.go
index a4c174f780..7030c0a59d 100644
--- a/pkg/hubble/filters/fqdn.go
+++ b/pkg/hubble/filters/fqdn.go
@@ -20,7 +20,7 @@ import (
 	"regexp"
 	"strings"
 
-	pb "github.com/cilium/cilium/api/v1/flow"
+	flowpb "github.com/cilium/cilium/api/v1/flow"
 	v1 "github.com/cilium/cilium/pkg/hubble/api/v1"
 )
 
@@ -114,7 +114,7 @@ func filterByDNSQueries(queryPatterns []string) (FilterFunc, error) {
 type FQDNFilter struct{}
 
 // OnBuildFilter builds a FQDN filter
-func (f *FQDNFilter) OnBuildFilter(ctx context.Context, ff *pb.FlowFilter) ([]FilterFunc, error) {
+func (f *FQDNFilter) OnBuildFilter(ctx context.Context, ff *flowpb.FlowFilter) ([]FilterFunc, error) {
 	var fs []FilterFunc
 
 	if ff.GetSourceFqdn() != nil {
diff --git a/pkg/hubble/filters/http.go b/pkg/hubble/filters/http.go
index 810dd86b6b..7e22bf61a4 100644
--- a/pkg/hubble/filters/http.go
+++ b/pkg/hubble/filters/http.go
@@ -21,12 +21,12 @@ import (
 	"regexp"
 	"strings"
 
-	pb "github.com/cilium/cilium/api/v1/flow"
+	flowpb "github.com/cilium/cilium/api/v1/flow"
 	v1 "github.com/cilium/cilium/pkg/hubble/api/v1"
 	"github.com/cilium/cilium/pkg/monitor/api"
 )
 
-func httpMatchCompatibleEventFilter(types []*pb.EventTypeFilter) bool {
+func httpMatchCompatibleEventFilter(types []*flowpb.EventTypeFilter) bool {
 	if len(types) == 0 {
 		return true
 	}
@@ -86,7 +86,7 @@ func filterByHTTPStatusCode(statusCodePrefixes []string) (FilterFunc, error) {
 type HTTPFilter struct{}
 
 // OnBuildFilter builds a HTTP filter
-func (h *HTTPFilter) OnBuildFilter(ctx context.Context, ff *pb.FlowFilter) ([]FilterFunc, error) {
+func (h *HTTPFilter) OnBuildFilter(ctx context.Context, ff *flowpb.FlowFilter) ([]FilterFunc, error) {
 	var fs []FilterFunc
 
 	if ff.GetHttpStatusCode() != nil {
diff --git a/pkg/hubble/filters/identitty.go b/pkg/hubble/filters/identity.go
similarity index 86%
rename from pkg/hubble/filters/identitty.go
rename to pkg/hubble/filters/identity.go
index 03f59fbbbc..16700a971c 100644
--- a/pkg/hubble/filters/identitty.go
+++ b/pkg/hubble/filters/identity.go
@@ -17,19 +17,19 @@ package filters
 import (
 	"context"
 
-	pb "github.com/cilium/cilium/api/v1/flow"
+	flowpb "github.com/cilium/cilium/api/v1/flow"
 	v1 "github.com/cilium/cilium/pkg/hubble/api/v1"
 )
 
-func sourceEndpoint(ev *v1.Event) *pb.Endpoint {
+func sourceEndpoint(ev *v1.Event) *flowpb.Endpoint {
 	return ev.GetFlow().GetSource()
 }
 
-func destinationEndpoint(ev *v1.Event) *pb.Endpoint {
+func destinationEndpoint(ev *v1.Event) *flowpb.Endpoint {
 	return ev.GetFlow().GetDestination()
 }
 
-func filterByIdentity(identities []uint32, getEndpoint func(*v1.Event) *pb.Endpoint) FilterFunc {
+func filterByIdentity(identities []uint32, getEndpoint func(*v1.Event) *flowpb.Endpoint) FilterFunc {
 	return func(ev *v1.Event) bool {
 		if endpoint := getEndpoint(ev); endpoint != nil {
 			for _, i := range identities {
@@ -46,7 +46,7 @@ func filterByIdentity(identities []uint32, getEndpoint func(*v1.Event) *pb.Endpo
 type IdentityFilter struct{}
 
 // OnBuildFilter builds a security identity filter
-func (i *IdentityFilter) OnBuildFilter(ctx context.Context, ff *pb.FlowFilter) ([]FilterFunc, error) {
+func (i *IdentityFilter) OnBuildFilter(ctx context.Context, ff *flowpb.FlowFilter) ([]FilterFunc, error) {
 	var fs []FilterFunc
 
 	if ff.GetSourceIdentity() != nil {
diff --git a/pkg/hubble/filters/ip.go b/pkg/hubble/filters/ip.go
index 380505772d..6c90bbb6d1 100644
--- a/pkg/hubble/filters/ip.go
+++ b/pkg/hubble/filters/ip.go
@@ -19,7 +19,7 @@ import (
 	"fmt"
 	"net"
 
-	pb "github.com/cilium/cilium/api/v1/flow"
+	flowpb "github.com/cilium/cilium/api/v1/flow"
 	v1 "github.com/cilium/cilium/pkg/hubble/api/v1"
 )
 
@@ -59,7 +59,7 @@ func filterByIPs(ips []string, getIP func(*v1.Event) string) (FilterFunc, error)
 type IPFilter struct{}
 
 // OnBuildFilter builds an IP address filter
-func (f *IPFilter) OnBuildFilter(ctx context.Context, ff *pb.FlowFilter) ([]FilterFunc, error) {
+func (f *IPFilter) OnBuildFilter(ctx context.Context, ff *flowpb.FlowFilter) ([]FilterFunc, error) {
 	var fs []FilterFunc
 
 	if ff.GetSourceIp() != nil {
diff --git a/pkg/hubble/filters/k8s.go b/pkg/hubble/filters/k8s.go
index 740f6295ed..696ba71ef6 100644
--- a/pkg/hubble/filters/k8s.go
+++ b/pkg/hubble/filters/k8s.go
@@ -19,7 +19,7 @@ import (
 	"fmt"
 	"strings"
 
-	pb "github.com/cilium/cilium/api/v1/flow"
+	flowpb "github.com/cilium/cilium/api/v1/flow"
 	v1 "github.com/cilium/cilium/pkg/hubble/api/v1"
 	"github.com/cilium/cilium/pkg/hubble/k8s"
 )
@@ -75,7 +75,7 @@ func filterByNamespacedName(names []string, getName func(*v1.Event) (ns, name st
 type PodFilter struct{}
 
 // OnBuildFilter builds a Kubernetes pod name filter
-func (p *PodFilter) OnBuildFilter(ctx context.Context, ff *pb.FlowFilter) ([]FilterFunc, error) {
+func (p *PodFilter) OnBuildFilter(ctx context.Context, ff *flowpb.FlowFilter) ([]FilterFunc, error) {
 	var fs []FilterFunc
 
 	if ff.GetSourcePod() != nil {
@@ -101,7 +101,7 @@ func (p *PodFilter) OnBuildFilter(ctx context.Context, ff *pb.FlowFilter) ([]Fil
 type ServiceFilter struct{}
 
 // OnBuildFilter builds a Kubernetes service name filter
-func (s *ServiceFilter) OnBuildFilter(ctx context.Context, ff *pb.FlowFilter) ([]FilterFunc, error) {
+func (s *ServiceFilter) OnBuildFilter(ctx context.Context, ff *flowpb.FlowFilter) ([]FilterFunc, error) {
 	var fs []FilterFunc
 
 	if ff.GetSourceService() != nil {
diff --git a/pkg/hubble/filters/labels.go b/pkg/hubble/filters/labels.go
index 9ce3c1372b..fc5abf402a 100644
--- a/pkg/hubble/filters/labels.go
+++ b/pkg/hubble/filters/labels.go
@@ -19,7 +19,7 @@ import (
 	"fmt"
 	"regexp"
 
-	pb "github.com/cilium/cilium/api/v1/flow"
+	flowpb "github.com/cilium/cilium/api/v1/flow"
 	v1 "github.com/cilium/cilium/pkg/hubble/api/v1"
 	k8sLabels "github.com/cilium/cilium/pkg/k8s/slim/k8s/apis/labels"
 	ciliumLabels "github.com/cilium/cilium/pkg/labels"
@@ -82,7 +82,7 @@ func FilterByLabelSelectors(labelSelectors []string, getLabels func(*v1.Event) k
 type LabelsFilter struct{}
 
 // OnBuildFilter builds a labels filter
-func (l *LabelsFilter) OnBuildFilter(ctx context.Context, ff *pb.FlowFilter) ([]FilterFunc, error) {
+func (l *LabelsFilter) OnBuildFilter(ctx context.Context, ff *flowpb.FlowFilter) ([]FilterFunc, error) {
 	var fs []FilterFunc
 
 	if ff.GetSourceLabel() != nil {
diff --git a/pkg/hubble/filters/port.go b/pkg/hubble/filters/port.go
index 037b2a2975..829a202196 100644
--- a/pkg/hubble/filters/port.go
+++ b/pkg/hubble/filters/port.go
@@ -19,7 +19,7 @@ import (
 	"fmt"
 	"strconv"
 
-	pb "github.com/cilium/cilium/api/v1/flow"
+	flowpb "github.com/cilium/cilium/api/v1/flow"
 	v1 "github.com/cilium/cilium/pkg/hubble/api/v1"
 )
 
@@ -71,7 +71,7 @@ func filterByPort(portStrs []string, getPort func(*v1.Event) (port uint16, ok bo
 type PortFilter struct{}
 
 // OnBuildFilter builds a L4 port filter
-func (p *PortFilter) OnBuildFilter(ctx context.Context, ff *pb.FlowFilter) ([]FilterFunc, error) {
+func (p *PortFilter) OnBuildFilter(ctx context.Context, ff *flowpb.FlowFilter) ([]FilterFunc, error) {
 	var fs []FilterFunc
 
 	if ff.GetSourcePort() != nil {
diff --git a/pkg/hubble/filters/protocol.go b/pkg/hubble/filters/protocol.go
index b31589a3c2..fa36dd8048 100644
--- a/pkg/hubble/filters/protocol.go
+++ b/pkg/hubble/filters/protocol.go
@@ -19,7 +19,7 @@ import (
 	"fmt"
 	"strings"
 
-	pb "github.com/cilium/cilium/api/v1/flow"
+	flowpb "github.com/cilium/cilium/api/v1/flow"
 	v1 "github.com/cilium/cilium/pkg/hubble/api/v1"
 )
 
@@ -90,7 +90,7 @@ func filterByProtocol(protocols []string) (FilterFunc, error) {
 type ProtocolFilter struct{}
 
 // OnBuildFilter builds a L4 protocol filter
-func (p *ProtocolFilter) OnBuildFilter(ctx context.Context, ff *pb.FlowFilter) ([]FilterFunc, error) {
+func (p *ProtocolFilter) OnBuildFilter(ctx context.Context, ff *flowpb.FlowFilter) ([]FilterFunc, error) {
 	var fs []FilterFunc
 
 	if ff.GetProtocol() != nil {
diff --git a/pkg/hubble/filters/reply.go b/pkg/hubble/filters/reply.go
index 83a76f941c..fef7723830 100644
--- a/pkg/hubble/filters/reply.go
+++ b/pkg/hubble/filters/reply.go
@@ -17,7 +17,7 @@ package filters
 import (
 	"context"
 
-	pb "github.com/cilium/cilium/api/v1/flow"
+	flowpb "github.com/cilium/cilium/api/v1/flow"
 	v1 "github.com/cilium/cilium/pkg/hubble/api/v1"
 )
 
@@ -43,7 +43,7 @@ func filterByReplyField(replyParams []bool) FilterFunc {
 type ReplyFilter struct{}
 
 // OnBuildFilter builds a reply filter
-func (r *ReplyFilter) OnBuildFilter(ctx context.Context, ff *pb.FlowFilter) ([]FilterFunc, error) {
+func (r *ReplyFilter) OnBuildFilter(ctx context.Context, ff *flowpb.FlowFilter) ([]FilterFunc, error) {
 	var fs []FilterFunc
 
 	if ff.GetReply() != nil {
diff --git a/pkg/hubble/filters/verdict.go b/pkg/hubble/filters/verdict.go
index 64894df757..83e0875be8 100644
--- a/pkg/hubble/filters/verdict.go
+++ b/pkg/hubble/filters/verdict.go
@@ -17,11 +17,11 @@ package filters
 import (
 	"context"
 
-	pb "github.com/cilium/cilium/api/v1/flow"
+	flowpb "github.com/cilium/cilium/api/v1/flow"
 	v1 "github.com/cilium/cilium/pkg/hubble/api/v1"
 )
 
-func filterByVerdicts(vs []pb.Verdict) FilterFunc {
+func filterByVerdicts(vs []flowpb.Verdict) FilterFunc {
 	return func(ev *v1.Event) bool {
 		flow := ev.GetFlow()
 		if flow == nil {
@@ -41,7 +41,7 @@ func filterByVerdicts(vs []pb.Verdict) FilterFunc {
 type VerdictFilter struct{}
 
 // OnBuildFilter builds a forwarding verdict filter
-func (v *VerdictFilter) OnBuildFilter(ctx context.Context, ff *pb.FlowFilter) ([]FilterFunc, error) {
+func (v *VerdictFilter) OnBuildFilter(ctx context.Context, ff *flowpb.FlowFilter) ([]FilterFunc, error) {
 	var fs []FilterFunc
 
 	if ff.GetVerdict() != nil {
diff --git a/pkg/hubble/k8s/utils.go b/pkg/hubble/k8s/utils.go
index 5f4df230a7..e60fa58422 100644
--- a/pkg/hubble/k8s/utils.go
+++ b/pkg/hubble/k8s/utils.go
@@ -32,18 +32,3 @@ func ParseNamespaceName(namespaceName string) (string, string) {
 		return "default", ns
 	}
 }
-
-// ParseNamespaceNames returns the object's namespace and name. If namespace is
-// not specified, the namespace "default" is returned.
-func ParseNamespaceNames(namespaceNames []string) ([]string, []string) {
-	pods := make([]string, 0, len(namespaceNames))
-	nss := make([]string, 0, len(namespaceNames))
-
-	for _, namespaceName := range namespaceNames {
-		ns, pod := ParseNamespaceName(namespaceName)
-		nss = append(nss, ns)
-		pods = append(pods, pod)
-	}
-
-	return nss, pods
-}
diff --git a/pkg/hubble/metrics/api/api.go b/pkg/hubble/metrics/api/api.go
index 154588a4a2..b220541a51 100644
--- a/pkg/hubble/metrics/api/api.go
+++ b/pkg/hubble/metrics/api/api.go
@@ -17,7 +17,7 @@ package api
 import (
 	"strings"
 
-	"github.com/cilium/cilium/pkg/hubble/api/v1"
+	v1 "github.com/cilium/cilium/pkg/hubble/api/v1"
 
 	"github.com/prometheus/client_golang/prometheus"
 )
diff --git a/pkg/hubble/metrics/api/context.go b/pkg/hubble/metrics/api/context.go
index 92b2b99c24..467ec32e77 100644
--- a/pkg/hubble/metrics/api/context.go
+++ b/pkg/hubble/metrics/api/context.go
@@ -20,7 +20,7 @@ import (
 	"sort"
 	"strings"
 
-	"github.com/cilium/cilium/pkg/hubble/api/v1"
+	v1 "github.com/cilium/cilium/pkg/hubble/api/v1"
 )
 
 // ContextIdentifier describes the identification method of a transmission or
diff --git a/pkg/hubble/metrics/dns/handler.go b/pkg/hubble/metrics/dns/handler.go
index 9472fb60c6..b420e14baa 100644
--- a/pkg/hubble/metrics/dns/handler.go
+++ b/pkg/hubble/metrics/dns/handler.go
@@ -18,8 +18,8 @@ import (
 	"fmt"
 	"strings"
 
-	pb "github.com/cilium/cilium/api/v1/flow"
-	"github.com/cilium/cilium/pkg/hubble/api/v1"
+	flowpb "github.com/cilium/cilium/api/v1/flow"
+	v1 "github.com/cilium/cilium/pkg/hubble/api/v1"
 	"github.com/cilium/cilium/pkg/hubble/metrics/api"
 
 	"github.com/prometheus/client_golang/prometheus"
@@ -121,7 +121,7 @@ func (d *dnsHandler) ProcessFlow(flow v1.Flow) {
 	}
 	labels = append(labels, labelValues...)
 
-	if flow.GetVerdict() == pb.Verdict_DROPPED {
+	if flow.GetVerdict() == flowpb.Verdict_DROPPED {
 		d.queries.WithLabelValues(labels...).Inc()
 		labels[0] = "Policy denied"
 		d.responses.WithLabelValues(labels...).Inc()
diff --git a/pkg/hubble/metrics/drop/handler.go b/pkg/hubble/metrics/drop/handler.go
index 2b74268463..0223c7407c 100644
--- a/pkg/hubble/metrics/drop/handler.go
+++ b/pkg/hubble/metrics/drop/handler.go
@@ -15,8 +15,8 @@
 package drop
 
 import (
-	pb "github.com/cilium/cilium/api/v1/flow"
-	"github.com/cilium/cilium/pkg/hubble/api/v1"
+	flowpb "github.com/cilium/cilium/api/v1/flow"
+	v1 "github.com/cilium/cilium/pkg/hubble/api/v1"
 	"github.com/cilium/cilium/pkg/hubble/metrics/api"
 	monitorAPI "github.com/cilium/cilium/pkg/monitor/api"
 
@@ -53,7 +53,7 @@ func (d *dropHandler) Status() string {
 }
 
 func (d *dropHandler) ProcessFlow(flow v1.Flow) {
-	if flow.GetVerdict() != pb.Verdict_DROPPED {
+	if flow.GetVerdict() != flowpb.Verdict_DROPPED {
 		return
 	}
 
diff --git a/pkg/hubble/metrics/http/handler.go b/pkg/hubble/metrics/http/handler.go
index 78bf782660..1de203df76 100644
--- a/pkg/hubble/metrics/http/handler.go
+++ b/pkg/hubble/metrics/http/handler.go
@@ -18,8 +18,8 @@ import (
 	"strconv"
 	"time"
 
-	pb "github.com/cilium/cilium/api/v1/flow"
-	"github.com/cilium/cilium/pkg/hubble/api/v1"
+	flowpb "github.com/cilium/cilium/api/v1/flow"
+	v1 "github.com/cilium/cilium/pkg/hubble/api/v1"
 	"github.com/cilium/cilium/pkg/hubble/metrics/api"
 
 	"github.com/prometheus/client_golang/prometheus"
@@ -76,9 +76,9 @@ func (h *httpHandler) ProcessFlow(flow v1.Flow) {
 		return
 	}
 	labelValues := h.context.GetLabelValues(flow)
-	if l7.Type == pb.L7FlowType_REQUEST {
+	if l7.Type == flowpb.L7FlowType_REQUEST {
 		h.requests.WithLabelValues(append(labelValues, http.Method, http.Protocol)...).Inc()
-	} else if l7.Type == pb.L7FlowType_RESPONSE {
+	} else if l7.Type == flowpb.L7FlowType_RESPONSE {
 		status := strconv.Itoa(int(http.Code))
 		h.responses.WithLabelValues(append(labelValues, status, http.Method)...).Inc()
 		h.duration.WithLabelValues(append(labelValues, http.Method)...).Observe(float64(l7.LatencyNs) / float64(time.Second))
diff --git a/pkg/hubble/metrics/icmp/handler.go b/pkg/hubble/metrics/icmp/handler.go
index 4621e9729b..5fef1bdec5 100644
--- a/pkg/hubble/metrics/icmp/handler.go
+++ b/pkg/hubble/metrics/icmp/handler.go
@@ -15,7 +15,7 @@
 package icmp
 
 import (
-	"github.com/cilium/cilium/pkg/hubble/api/v1"
+	v1 "github.com/cilium/cilium/pkg/hubble/api/v1"
 	"github.com/cilium/cilium/pkg/hubble/metrics/api"
 
 	"github.com/google/gopacket/layers"
diff --git a/pkg/hubble/metrics/port-distribution/handler.go b/pkg/hubble/metrics/port-distribution/handler.go
index 3c7c5d7a56..fae5234c36 100644
--- a/pkg/hubble/metrics/port-distribution/handler.go
+++ b/pkg/hubble/metrics/port-distribution/handler.go
@@ -17,8 +17,8 @@ package portdistribution
 import (
 	"fmt"
 
-	pb "github.com/cilium/cilium/api/v1/flow"
-	"github.com/cilium/cilium/pkg/hubble/api/v1"
+	flowpb "github.com/cilium/cilium/api/v1/flow"
+	v1 "github.com/cilium/cilium/pkg/hubble/api/v1"
 	"github.com/cilium/cilium/pkg/hubble/metrics/api"
 
 	"github.com/prometheus/client_golang/prometheus"
@@ -54,7 +54,7 @@ func (h *portDistributionHandler) Status() string {
 }
 
 func (h *portDistributionHandler) ProcessFlow(flow v1.Flow) {
-	if flow.GetVerdict() != pb.Verdict_FORWARDED || flow.GetL4() == nil || flow.GetReply() {
+	if flow.GetVerdict() != flowpb.Verdict_FORWARDED || flow.GetL4() == nil || flow.GetReply() {
 		return
 	}
 
diff --git a/pkg/hubble/metrics/tcp/handler.go b/pkg/hubble/metrics/tcp/handler.go
index a97c9ebcd0..6a77db3d30 100644
--- a/pkg/hubble/metrics/tcp/handler.go
+++ b/pkg/hubble/metrics/tcp/handler.go
@@ -15,8 +15,8 @@
 package tcp
 
 import (
-	pb "github.com/cilium/cilium/api/v1/flow"
-	"github.com/cilium/cilium/pkg/hubble/api/v1"
+	flowpb "github.com/cilium/cilium/api/v1/flow"
+	v1 "github.com/cilium/cilium/pkg/hubble/api/v1"
 	"github.com/cilium/cilium/pkg/hubble/metrics/api"
 
 	"github.com/prometheus/client_golang/prometheus"
@@ -52,7 +52,7 @@ func (h *tcpHandler) Status() string {
 }
 
 func (h *tcpHandler) ProcessFlow(flow v1.Flow) {
-	if flow.GetVerdict() != pb.Verdict_FORWARDED || flow.GetL4() == nil {
+	if flow.GetVerdict() != flowpb.Verdict_FORWARDED || flow.GetL4() == nil {
 		return
 	}
 
diff --git a/pkg/hubble/observer/local_observer.go b/pkg/hubble/observer/local_observer.go
index 780e538d29..7c4f6c50da 100644
--- a/pkg/hubble/observer/local_observer.go
+++ b/pkg/hubble/observer/local_observer.go
@@ -23,7 +23,7 @@ import (
 	"strings"
 	"time"
 
-	pb "github.com/cilium/cilium/api/v1/flow"
+	flowpb "github.com/cilium/cilium/api/v1/flow"
 	observerpb "github.com/cilium/cilium/api/v1/observer"
 	v1 "github.com/cilium/cilium/pkg/hubble/api/v1"
 	"github.com/cilium/cilium/pkg/hubble/container"
@@ -49,9 +49,9 @@ type GRPCServer interface {
 	// Start starts the server and blocks.
 	Start()
 	// GetEventsChannel returns the channel to push monitor events to.
-	GetEventsChannel() chan *pb.Payload
+	GetEventsChannel() chan *flowpb.Payload
 	// SetEventsChannel sets the events channel. For unit testing only.
-	SetEventsChannel(chan *pb.Payload)
+	SetEventsChannel(chan *flowpb.Payload)
 	///GetRingBuffer returns the underlying ring buffer to parsed events.
 	GetRingBuffer() *container.Ring
 	// GetStopped returns a channel that gets closed at the end of the
@@ -71,7 +71,7 @@ type LocalObserverServer struct {
 
 	// events is the channel used by the writer(s) to send the flow data
 	// into the observer server.
-	events chan *pb.Payload
+	events chan *flowpb.Payload
 
 	// stopped is mostly used in unit tests to signalize when the events
 	// channel is empty, once it's closed.
@@ -82,7 +82,7 @@ type LocalObserverServer struct {
 	// channel to receive events from observer server.
 	eventschan chan *observerpb.GetFlowsResponse
 
-	// payloadParser decodes pb.Payload into pb.Flow
+	// payloadParser decodes flowpb.Payload into flowpb.Flow
 	payloadParser *parser.Parser
 
 	opts observeroption.Options
@@ -116,7 +116,7 @@ func NewLocalServer(
 	s := &LocalObserverServer{
 		log:           logger,
 		ring:          container.NewRing(opts.MaxFlows),
-		events:        make(chan *pb.Payload, opts.MonitorBuffer),
+		events:        make(chan *flowpb.Payload, opts.MonitorBuffer),
 		stopped:       make(chan struct{}),
 		eventschan:    make(chan *observerpb.GetFlowsResponse, 100), // option here?
 		payloadParser: payloadParser,
@@ -182,13 +182,13 @@ nextEvent:
 	close(s.GetStopped())
 }
 
-// GetEventsChannel returns the event channel to receive pb.Payload events.
-func (s *LocalObserverServer) GetEventsChannel() chan *pb.Payload {
+// GetEventsChannel returns the event channel to receive flowpb.Payload events.
+func (s *LocalObserverServer) GetEventsChannel() chan *flowpb.Payload {
 	return s.events
 }
 
 // SetEventsChannel implements GRPCServer.SetEventsChannel.
-func (s *LocalObserverServer) SetEventsChannel(events chan *pb.Payload) {
+func (s *LocalObserverServer) SetEventsChannel(events chan *flowpb.Payload) {
 	s.events = events
 }
 
@@ -268,7 +268,7 @@ func (s *LocalObserverServer) GetFlows(
 			"buffer_size":     ring.Cap(),
 			"whitelist":       logFilters(req.Whitelist),
 			"blacklist":       logFilters(req.Blacklist),
-			"took":            time.Now().Sub(start),
+			"took":            time.Since(start),
 		}).Debug("GetFlows finished")
 	}()
 
@@ -325,7 +325,7 @@ func getUntil(req *observerpb.GetFlowsRequest, defaultTime *timestamp.Timestamp)
 	return ptypes.Timestamp(until)
 }
 
-func logFilters(filters []*pb.FlowFilter) string {
+func logFilters(filters []*flowpb.FlowFilter) string {
 	var s []string
 	for _, f := range filters {
 		s = append(s, f.String())
@@ -333,9 +333,9 @@ func logFilters(filters []*pb.FlowFilter) string {
 	return "{" + strings.Join(s, ",") + "}"
 }
 
-func decodeFlow(payloadParser *parser.Parser, pl *pb.Payload) (*pb.Flow, error) {
+func decodeFlow(payloadParser *parser.Parser, pl *flowpb.Payload) (*flowpb.Flow, error) {
 	// TODO: Pool these instead of allocating new flows each time.
-	f := &pb.Flow{}
+	f := &flowpb.Flow{}
 	err := payloadParser.Decode(pl, f)
 	if err != nil {
 		return nil, err
@@ -388,7 +388,7 @@ func newFlowsReader(r *container.RingReader, req *observerpb.GetFlowsRequest, lo
 }
 
 // Next returns the next flow that matches the request criterias.
-func (r *flowsReader) Next(ctx context.Context) (*pb.Flow, error) {
+func (r *flowsReader) Next(ctx context.Context) (*flowpb.Flow, error) {
 	for {
 		select {
 		case <-ctx.Done():
@@ -427,7 +427,7 @@ func (r *flowsReader) Next(ctx context.Context) (*pb.Flow, error) {
 				continue
 			}
 		}
-		flow, ok := e.Event.(*pb.Flow)
+		flow, ok := e.Event.(*flowpb.Flow)
 		if ok && filters.Apply(r.whitelist, r.blacklist, e) {
 			r.flowsCount++
 			return flow, nil
@@ -469,7 +469,7 @@ func newRingReader(ring *container.Ring, req *observerpb.GetFlowsRequest, whitel
 		} else if err != nil {
 			return nil, err
 		}
-		_, ok := e.Event.(*pb.Flow)
+		_, ok := e.Event.(*flowpb.Flow)
 		if !ok || !filters.Apply(whitelist, blacklist, e) {
 			continue
 		}
diff --git a/pkg/hubble/parser/getters/getters.go b/pkg/hubble/parser/getters/getters.go
index 4d4d4e1554..818d7ccd4e 100644
--- a/pkg/hubble/parser/getters/getters.go
+++ b/pkg/hubble/parser/getters/getters.go
@@ -17,7 +17,7 @@ package getters
 import (
 	"net"
 
-	pb "github.com/cilium/cilium/api/v1/flow"
+	flowpb "github.com/cilium/cilium/api/v1/flow"
 	"github.com/cilium/cilium/api/v1/models"
 	v1 "github.com/cilium/cilium/pkg/hubble/api/v1"
 	"github.com/cilium/cilium/pkg/hubble/ipcache"
@@ -50,5 +50,5 @@ type IPGetter interface {
 
 // ServiceGetter fetches service metadata.
 type ServiceGetter interface {
-	GetServiceByAddr(ip net.IP, port uint16) (service pb.Service, ok bool)
+	GetServiceByAddr(ip net.IP, port uint16) (service flowpb.Service, ok bool)
 }
diff --git a/pkg/hubble/relay/server.go b/pkg/hubble/relay/server.go
index 0a36468bf4..0405c07e46 100644
--- a/pkg/hubble/relay/server.go
+++ b/pkg/hubble/relay/server.go
@@ -30,6 +30,7 @@ import (
 	"google.golang.org/grpc"
 	"google.golang.org/grpc/health"
 	healthpb "google.golang.org/grpc/health/grpc_health_v1"
+	"google.golang.org/grpc/reflection"
 )
 
 type hubblePeer struct {
@@ -82,6 +83,7 @@ func (s *Server) Serve() error {
 		return fmt.Errorf("failed to listen on tcp socket %s: %v", s.opts.ListenAddress, err)
 	}
 	go s.syncPeers()
+	reflection.Register(s.server)
 	return s.server.Serve(socket)
 }
 
diff --git a/pkg/k8s/apis/cilium.io/v2/client/register.go b/pkg/k8s/apis/cilium.io/v2/client/register.go
index 437d7fe243..dfd2559ccf 100644
--- a/pkg/k8s/apis/cilium.io/v2/client/register.go
+++ b/pkg/k8s/apis/cilium.io/v2/client/register.go
@@ -54,7 +54,7 @@ const (
 
 	// CustomResourceDefinitionSchemaVersion is semver-conformant version of CRD schema
 	// Used to determine if CRD needs to be updated in cluster
-	CustomResourceDefinitionSchemaVersion = "1.18"
+	CustomResourceDefinitionSchemaVersion = "1.19"
 
 	// CustomResourceDefinitionSchemaVersionKey is key to label which holds the CRD schema version
 	CustomResourceDefinitionSchemaVersionKey = "io.cilium.k8s.crd.schema.version"
@@ -1442,15 +1442,20 @@ var (
 	Rule = apiextensionsv1beta1.JSONSchemaProps{
 		Type: "object",
 		Description: "Rule is a policy rule which must be applied to all endpoints which match " +
-			"the labels contained in the endpointSelector\n\nEach rule is split into an " +
-			"ingress section which contains all rules applicable at ingress, and an egress " +
-			"section applicable at egress. For rule types such as `L4Rule` and `CIDR` which " +
-			"can be applied at both ingress and egress, both ingress and egress side have to " +
-			"either specifically allow the connection or one side has to be omitted.\n\n" +
-			"Either ingress, egress, or both can be provided. If both ingress and egress are " +
-			"omitted, the rule has no effect.",
-		Required: []string{
-			"endpointSelector",
+			"the labels contained in the endpointSelector or all nodes which match the labels " +
+			"contained in the nodeSelector.\n\nEach rule is split into an ingress section " +
+			"which contains all rules applicable at ingress, and an egress section applicable " +
+			"at egress. For rule types such as `L4Rule` and `CIDR` which can be applied at both " +
+			"ingress and egress, both ingress and egress side have to either specifically allow " +
+			"the connection or one side has to be omitted.\n\nEither ingress, egress, or both " +
+			"can be provided. If both ingress and egress are omitted, the rule has no effect.",
+		OneOf: []apiextensionsv1beta1.JSONSchemaProps{
+			{
+				Required: []string{"endpointSelector"},
+			},
+			{
+				Required: []string{"nodeSelector"},
+			},
 		},
 		Properties: map[string]apiextensionsv1beta1.JSONSchemaProps{
 			"Description": {
@@ -1468,6 +1473,7 @@ var (
 				},
 			},
 			"endpointSelector": EndpointSelector,
+			"nodeSelector":     EndpointSelector,
 			"ingress": {
 				Description: "Ingress is a list of IngressRule which are enforced at ingress. " +
 					"If omitted or empty, this rule does not apply at ingress.",
@@ -1534,9 +1540,14 @@ func init() {
 
 	ruleProps := Rule.Properties["endpointSelector"]
 	ruleProps.Description = "EndpointSelector selects all endpoints which should be subject " +
-		"to this rule. Cannot be empty."
+		"to this rule. Cannot be empty if nodeSelector is empty."
 	Rule.Properties["endpointSelector"] = ruleProps
 
+	ruleProps = Rule.Properties["nodeSelector"]
+	ruleProps.Description = "NodeSelector selects all nodes which should be subject " +
+		"to this rule. Cannot be empty if endpointSelector is empty."
+	Rule.Properties["nodeSelector"] = ruleProps
+
 	serviceProps := Service.Properties["k8sServiceSelector"]
 	serviceProps.Description = "K8sServiceSelector selects services by k8s labels. " +
 		"Not supported yet"
diff --git a/pkg/option/config.go b/pkg/option/config.go
index af2bc979ed..3863e8becb 100644
--- a/pkg/option/config.go
+++ b/pkg/option/config.go
@@ -221,6 +221,9 @@ const (
 	// LabelPrefixFile is the valid label prefixes file path
 	LabelPrefixFile = "label-prefix-file"
 
+	// EnableHostFirewall enables network policies for the host
+	EnableHostFirewall = "enable-host-firewall"
+
 	// EnableHostPort enables HostPort forwarding implemented by Cilium in BPF
 	EnableHostPort = "enable-host-port"
 
@@ -1627,6 +1630,9 @@ type DaemonConfig struct {
 	// EnableExternalIPs enables implementation of k8s services with externalIPs in datapath
 	EnableExternalIPs bool
 
+	// EnableHostFirewall enables network policies for the host
+	EnableHostFirewall bool
+
 	// K8sEnableEndpointSlice enables k8s endpoint slice feature that is used
 	// in kubernetes.
 	K8sEnableK8sEndpointSlice bool
@@ -2156,6 +2162,7 @@ func (c *DaemonConfig) Populate() {
 	c.EnableAutoProtectNodePortRange = viper.GetBool(EnableAutoProtectNodePortRange)
 	c.KubeProxyReplacement = viper.GetString(KubeProxyReplacement)
 	c.EnableSessionAffinity = viper.GetBool(EnableSessionAffinity)
+	c.EnableHostFirewall = viper.GetBool(EnableHostFirewall)
 	c.EncryptInterface = viper.GetString(EncryptInterface)
 	c.EncryptNode = viper.GetBool(EncryptNode)
 	c.EnvoyLogPath = viper.GetString(EnvoyLog)
diff --git a/pkg/policy/api/rule.go b/pkg/policy/api/rule.go
index 91b28209c9..2838d0d27a 100644
--- a/pkg/policy/api/rule.go
+++ b/pkg/policy/api/rule.go
@@ -15,7 +15,9 @@
 package api
 
 import (
+	"bytes"
 	"context"
+	"encoding/json"
 
 	"github.com/cilium/cilium/pkg/labels"
 )
@@ -34,8 +36,18 @@ import (
 // +deepequal-gen:private-method=true
 type Rule struct {
 	// EndpointSelector selects all endpoints which should be subject to
-	// this rule. Cannot be empty.
-	EndpointSelector EndpointSelector `json:"endpointSelector"`
+	// this rule. EndpointSelector and NodeSelector cannot be both empty and
+	// are mutually exclusive.
+	//
+	// +optional
+	EndpointSelector EndpointSelector `json:"endpointSelector,omitempty"`
+
+	// NodeSelector selects all nodes which should be subject to this rule.
+	// EndpointSelector and NodeSelector cannot be both empty and are mutually
+	// exclusive.
+	//
+	// +optional
+	NodeSelector EndpointSelector `json:"nodeSelector,omitempty"`
 
 	// Ingress is a list of IngressRule which are enforced at ingress.
 	// If omitted or empty, this rule does not apply at ingress.
@@ -65,6 +77,70 @@ type Rule struct {
 	Description string `json:"description,omitempty"`
 }
 
+// MarshalJSON returns the JSON encoding of Rule r. We need to overwrite it to
+// enforce omitempty on the EndpointSelector nested structures.
+func (r *Rule) MarshalJSON() ([]byte, error) {
+	var buf bytes.Buffer
+	buf.WriteByte('{')
+
+	if r.EndpointSelector.LabelSelector != nil {
+		jsonValue, err := json.Marshal(r.EndpointSelector)
+		if err != nil {
+			return nil, err
+		}
+		buf.Write([]byte(`"endpointSelector":`))
+		buf.Write(jsonValue)
+	}
+
+	if r.NodeSelector.LabelSelector != nil {
+		jsonValue, err := json.Marshal(r.NodeSelector)
+		if err != nil {
+			return nil, err
+		}
+		buf.Write([]byte(`"nodeSelector":`))
+		buf.Write(jsonValue)
+	}
+
+	if r.Ingress != nil {
+		jsonValue, err := json.Marshal(r.Ingress)
+		if err != nil {
+			return nil, err
+		}
+		buf.Write([]byte(`,"ingress":`))
+		buf.Write(jsonValue)
+	}
+
+	if r.Egress != nil {
+		jsonValue, err := json.Marshal(r.Egress)
+		if err != nil {
+			return nil, err
+		}
+		buf.Write([]byte(`,"egress":`))
+		buf.Write(jsonValue)
+	}
+
+	if r.Labels != nil {
+		jsonValue, err := json.Marshal(r.Labels)
+		if err != nil {
+			return nil, err
+		}
+		buf.Write([]byte(`,"labels":`))
+		buf.Write(jsonValue)
+	}
+
+	if r.Description != "" {
+		jsonValue, err := json.Marshal(r.Description)
+		if err != nil {
+			return nil, err
+		}
+		buf.Write([]byte(`,"description":`))
+		buf.Write(jsonValue)
+	}
+
+	buf.WriteByte('}')
+	return buf.Bytes(), nil
+}
+
 func (r *Rule) DeepEqual(o *Rule) bool {
 	switch {
 	case (r == nil) != (o == nil):
diff --git a/pkg/policy/api/rule_validation.go b/pkg/policy/api/rule_validation.go
index bc6352c0f4..75ed731313 100644
--- a/pkg/policy/api/rule_validation.go
+++ b/pkg/policy/api/rule_validation.go
@@ -46,24 +46,47 @@ func (r Rule) Sanitize() error {
 		}
 	}
 
-	if r.EndpointSelector.LabelSelector == nil {
-		return fmt.Errorf("rule cannot have nil EndpointSelector")
+	if r.EndpointSelector.LabelSelector == nil && r.NodeSelector.LabelSelector == nil {
+		return fmt.Errorf("rule must have one of EndpointSelector or NodeSelector")
+	}
+	if r.EndpointSelector.LabelSelector != nil && r.NodeSelector.LabelSelector != nil {
+		return fmt.Errorf("rule cannot have both EndpointSelector and NodeSelector")
 	}
 
-	if err := r.EndpointSelector.sanitize(); err != nil {
-		return err
+	if r.EndpointSelector.LabelSelector != nil {
+		if err := r.EndpointSelector.sanitize(); err != nil {
+			return err
+		}
+	}
+
+	var hostPolicy bool
+	if r.NodeSelector.LabelSelector != nil {
+		if err := r.NodeSelector.sanitize(); err != nil {
+			return err
+		}
+		hostPolicy = true
 	}
 
 	for i := range r.Ingress {
 		if err := r.Ingress[i].sanitize(); err != nil {
 			return err
 		}
+		if hostPolicy {
+			if len(countL7Rules(r.Ingress[i].ToPorts)) > 0 {
+				return fmt.Errorf("host policies do not support L7 rules yet")
+			}
+		}
 	}
 
 	for i := range r.Egress {
 		if err := r.Egress[i].sanitize(); err != nil {
 			return err
 		}
+		if hostPolicy {
+			if len(countL7Rules(r.Egress[i].ToPorts)) > 0 {
+				return fmt.Errorf("host policies do not support L7 rules yet")
+			}
+		}
 	}
 
 	return nil
diff --git a/pkg/policy/api/zz_generated.deepcopy.go b/pkg/policy/api/zz_generated.deepcopy.go
index 08fa6fd0d0..ae3246a9c1 100644
--- a/pkg/policy/api/zz_generated.deepcopy.go
+++ b/pkg/policy/api/zz_generated.deepcopy.go
@@ -644,6 +644,7 @@ func (in PortRuleL7) DeepCopy() PortRuleL7 {
 func (in *Rule) DeepCopyInto(out *Rule) {
 	*out = *in
 	in.EndpointSelector.DeepCopyInto(&out.EndpointSelector)
+	in.NodeSelector.DeepCopyInto(&out.NodeSelector)
 	if in.Ingress != nil {
 		in, out := &in.Ingress, &out.Ingress
 		*out = make([]IngressRule, len(*in))
diff --git a/pkg/policy/api/zz_generated.deepequal.go b/pkg/policy/api/zz_generated.deepequal.go
index 2d68978522..e1dd1646d7 100644
--- a/pkg/policy/api/zz_generated.deepequal.go
+++ b/pkg/policy/api/zz_generated.deepequal.go
@@ -894,6 +894,10 @@ func (in *Rule) deepEqual(other *Rule) bool {
 		return false
 	}
 
+	if !in.NodeSelector.DeepEqual(&other.NodeSelector) {
+		return false
+	}
+
 	if ((in.Ingress != nil) && (other.Ingress != nil)) || ((in.Ingress == nil) != (other.Ingress == nil)) {
 		in, other := &in.Ingress, &other.Ingress
 		if other == nil {
diff --git a/pkg/policy/identifier.go b/pkg/policy/identifier.go
index de5a0f8316..4470045688 100644
--- a/pkg/policy/identifier.go
+++ b/pkg/policy/identifier.go
@@ -25,10 +25,12 @@ import (
 // * a node-local ID stored as a uint16
 // * a security identity
 // * a means of incrementing its policy revision
+// * a means of checking if it represents a node or a pod.
 type Endpoint interface {
 	GetID16() uint16
 	GetSecurityIdentity() (*identity.Identity, error)
 	PolicyRevisionBumpEvent(rev uint64)
+	IsHost() bool
 }
 
 // EndpointSet is used to be able to group together a given set of Endpoints
diff --git a/pkg/policy/repository.go b/pkg/policy/repository.go
index 7c15c9a2a2..7a3d14d798 100644
--- a/pkg/policy/repository.go
+++ b/pkg/policy/repository.go
@@ -495,7 +495,7 @@ func (p *Repository) GetRulesMatching(lbls labels.LabelArray) (ingressMatch bool
 	ingressMatch = false
 	egressMatch = false
 	for _, r := range p.rules {
-		rulesMatch := r.EndpointSelector.Matches(lbls)
+		rulesMatch := r.getSelector().Matches(lbls)
 		if rulesMatch {
 			if len(r.Ingress) > 0 {
 				ingressMatch = true
@@ -522,6 +522,11 @@ func (p *Repository) getMatchingRules(securityIdentity *identity.Identity) (ingr
 	ingressMatch = false
 	egressMatch = false
 	for _, r := range p.rules {
+		isNode := securityIdentity.ID == identity.ReservedIdentityHost
+		selectsNode := r.NodeSelector.LabelSelector != nil
+		if selectsNode != isNode {
+			continue
+		}
 		if ruleMatches := r.matches(securityIdentity); ruleMatches {
 			// Don't need to update whether ingressMatch is true if it already
 			// has been determined to be true - allows us to not have to check
diff --git a/pkg/policy/rule.go b/pkg/policy/rule.go
index d6bb516525..51e5730f65 100644
--- a/pkg/policy/rule.go
+++ b/pkg/policy/rule.go
@@ -57,6 +57,13 @@ func (r *rule) String() string {
 	return fmt.Sprintf("%v", r.EndpointSelector)
 }
 
+func (r *rule) getSelector() *api.EndpointSelector {
+	if r.NodeSelector.LabelSelector != nil {
+		return &r.NodeSelector
+	}
+	return &r.EndpointSelector
+}
+
 func (epd *PerSelectorPolicy) appendL7WildcardRule(ctx *SearchContext) *PerSelectorPolicy {
 	// Wildcard rule only needs to be appended if some rules already exist
 	switch {
@@ -424,7 +431,7 @@ func (state *traceState) unSelectRule(ctx *SearchContext, labels labels.LabelArr
 // as requirements form conjunctions across all rules.
 func (r *rule) resolveIngressPolicy(policyCtx PolicyContext, ctx *SearchContext, state *traceState, result L4PolicyMap, requirements []slim_metav1.LabelSelectorRequirement) (L4PolicyMap, error) {
 	if !ctx.rulesSelect {
-		if !r.EndpointSelector.Matches(ctx.To) {
+		if !r.getSelector().Matches(ctx.To) {
 			state.unSelectRule(ctx, ctx.To, r)
 			return nil, nil
 		}
@@ -478,7 +485,7 @@ func mergeCIDR(ctx *SearchContext, dir string, ipRules []api.CIDR, ruleLabels la
 func (r *rule) resolveCIDRPolicy(ctx *SearchContext, state *traceState, result *CIDRPolicy) *CIDRPolicy {
 	// Don't select rule if it doesn't apply to the given context.
 	if !ctx.rulesSelect {
-		if !r.EndpointSelector.Matches(ctx.To) {
+		if !r.getSelector().Matches(ctx.To) {
 			state.unSelectRule(ctx, ctx.To, r)
 			return nil
 		}
@@ -538,8 +545,13 @@ func (r *rule) matches(securityIdentity *identity.Identity) bool {
 	if ruleMatches, cached := r.metadata.IdentitySelected[securityIdentity.ID]; cached {
 		return ruleMatches
 	}
+	isNode := securityIdentity.ID == identity.ReservedIdentityHost
+	if (r.NodeSelector.LabelSelector != nil) != isNode {
+		r.metadata.IdentitySelected[securityIdentity.ID] = false
+		return ruleMatches
+	}
 	// Fall back to costly matching.
-	if ruleMatches = r.EndpointSelector.Matches(securityIdentity.LabelArray); ruleMatches {
+	if ruleMatches = r.getSelector().Matches(securityIdentity.LabelArray); ruleMatches {
 		// Update cache so we don't have to do costly matching again.
 		r.metadata.IdentitySelected[securityIdentity.ID] = true
 	} else {
@@ -661,7 +673,7 @@ func mergeEgressPortProto(policyCtx PolicyContext, ctx *SearchContext, endpoints
 
 func (r *rule) resolveEgressPolicy(policyCtx PolicyContext, ctx *SearchContext, state *traceState, result L4PolicyMap, requirements []slim_metav1.LabelSelectorRequirement) (L4PolicyMap, error) {
 	if !ctx.rulesSelect {
-		if !r.EndpointSelector.Matches(ctx.From) {
+		if !r.getSelector().Matches(ctx.From) {
 			state.unSelectRule(ctx, ctx.From, r)
 			return nil, nil
 		}
diff --git a/pkg/policy/rules.go b/pkg/policy/rules.go
index fccad1236a..ab5ca656dd 100644
--- a/pkg/policy/rules.go
+++ b/pkg/policy/rules.go
@@ -39,7 +39,7 @@ func (rules ruleSlice) resolveL4IngressPolicy(policyCtx PolicyContext, ctx *Sear
 	// each FromEndpoints for all ingress rules. This ensures that FromRequires
 	// is taken into account when evaluating policy at L4.
 	for _, r := range rules {
-		if ctx.rulesSelect || r.EndpointSelector.Matches(ctx.To) {
+		if ctx.rulesSelect || r.getSelector().Matches(ctx.To) {
 			matchedRules = append(matchedRules, r)
 			for _, ingressRule := range r.Ingress {
 				for _, requirement := range ingressRule.FromRequires {
@@ -87,7 +87,7 @@ func (rules ruleSlice) resolveL4EgressPolicy(policyCtx PolicyContext, ctx *Searc
 	// ToEndpoints for all egress rules. This ensures that ToRequires is
 	// taken into account when evaluating policy at L4.
 	for _, r := range rules {
-		if ctx.rulesSelect || r.EndpointSelector.Matches(ctx.From) {
+		if ctx.rulesSelect || r.getSelector().Matches(ctx.From) {
 			matchedRules = append(matchedRules, r)
 			for _, egressRule := range r.Egress {
 				for _, requirement := range egressRule.ToRequires {
@@ -157,6 +157,10 @@ func (rules ruleSlice) updateEndpointsCaches(ep Endpoint) (bool, error) {
 	}
 	endpointSelected := false
 	for _, r := range rules {
+		// NodeSelector can only match nodes, EndpointSelector only pods.
+		if (r.NodeSelector.LabelSelector != nil) != ep.IsHost() {
+			continue
+		}
 		// Update the matches cache of each rule, and note if
 		// the ep is selected by any of them.
 		if ruleMatches := r.matches(securityIdentity); ruleMatches {


Test patch:
diff --git a/Documentation/contributing/testing/documentation.rst b/Documentation/contributing/testing/documentation.rst
index 8d0884d1af..a0c958e0b5 100644
--- a/Documentation/contributing/testing/documentation.rst
+++ b/Documentation/contributing/testing/documentation.rst
@@ -7,7 +7,18 @@
 Documentation
 =============
 
-Whenever making changes to Cilium documentation you should check that you did not introduce any new warnings or errors, and also check that your changes look as you intended.  To do this you can build the docs:
+First, start a local document server that automatically refreshes when you save files for
+real-time preview. After installing `pipenv <https://pipenv.pypa.io/en/latest/#install-pipenv-today>`_,
+run:
+
+::
+
+    $ make render-docs-live-preview
+
+and preview the documentation at http://localhost:8000/ as you make changes. After making changes to
+Cilium documentation you should check that you did not introduce any new warnings or errors, and also
+check that your changes look as you intended one last time before opening a pull request. To do this
+you can build the docs:
 
 ::
 
@@ -16,7 +27,3 @@ Whenever making changes to Cilium documentation you should check that you did no
 This generates documentation files and starts a web server using a Docker container. You can
 view the updated documentation by opening either ``Documentation/_build/html/index.html`` or
 http://localhost:9081 in a browser.
-
-.. note:: ``make render-docs`` is relatively slow since it performs syntax and spelling checks.
-          You can run ``make render-docs SKIP_LINT=1`` to render the documentation without performing
-          these checks while you iterate on updating the documentation.
diff --git a/pkg/hubble/api/v1/endpoint_test.go b/pkg/hubble/api/v1/endpoint_test.go
deleted file mode 100644
index 74db21d1b0..0000000000
--- a/pkg/hubble/api/v1/endpoint_test.go
+++ /dev/null
@@ -1,712 +0,0 @@
-// Copyright 2019 Authors of Hubble
-//
-// Licensed under the Apache License, Version 2.0 (the "License");
-// you may not use this file except in compliance with the License.
-// You may obtain a copy of the License at
-//
-//     http://www.apache.org/licenses/LICENSE-2.0
-//
-// Unless required by applicable law or agreed to in writing, software
-// distributed under the License is distributed on an "AS IS" BASIS,
-// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
-// See the License for the specific language governing permissions and
-// limitations under the License.
-
-// +build !privileged_tests
-
-package v1
-
-import (
-	"net"
-	"reflect"
-	"testing"
-
-	"github.com/cilium/cilium/pkg/lock"
-
-	"github.com/stretchr/testify/assert"
-)
-
-func TestEndpoint_EqualsByID(t *testing.T) {
-	type fields struct {
-		ContainerID  []string
-		ID           uint64
-		IPv4         net.IP
-		IPv6         net.IP
-		PodName      string
-		PodNamespace string
-	}
-	tests := []struct {
-		name   string
-		fields fields
-		arg    *Endpoint
-		want   bool
-	}{
-		{
-			name:   "<nil> endpoint",
-			fields: fields{},
-			arg:    nil,
-			want:   false,
-		}, {
-			name: "compare by a same ID and all other fields different should be considered equal",
-			fields: fields{
-				ContainerID:  []string{"foo"},
-				ID:           1,
-				IPv4:         net.ParseIP("2.2.2.2"),
-				PodName:      "",
-				PodNamespace: "",
-			},
-			arg: &Endpoint{
-				ContainerIDs: []string{"bar"},
-				ID:           1,
-				IPv4:         net.ParseIP("1.1.1.1"),
-				PodName:      "",
-				PodNamespace: "",
-			},
-			want: true,
-		},
-		{
-			name: "compare by a same ID, but different pod name should be considered different",
-			fields: fields{
-				ContainerID:  []string{"foo"},
-				ID:           1,
-				IPv4:         net.ParseIP("2.2.2.2"),
-				PodName:      "pod-bar",
-				PodNamespace: "",
-			},
-			arg: &Endpoint{
-				ContainerIDs: []string{"bar"},
-				ID:           1,
-				IPv4:         net.ParseIP("1.1.1.1"),
-				PodName:      "pod-foo",
-				PodNamespace: "",
-			},
-			want: false,
-		},
-		{
-			name: "compare by a same ID, but different namespace should be considered different",
-			fields: fields{
-				ContainerID:  []string{"foo"},
-				ID:           1,
-				IPv4:         net.ParseIP("2.2.2.2"),
-				PodName:      "pod-bar",
-				PodNamespace: "kube-system",
-			},
-			arg: &Endpoint{
-				ContainerIDs: []string{"bar"},
-				ID:           1,
-				IPv4:         net.ParseIP("1.1.1.1"),
-				PodName:      "pod-bar",
-				PodNamespace: "cilium",
-			},
-			want: false,
-		},
-		{
-			name: "compare by a same ID where podname and podnamespace are empty should be considered equal",
-			fields: fields{
-				ContainerID:  []string{"foo"},
-				ID:           1,
-				IPv4:         net.ParseIP("2.2.2.2"),
-				PodName:      "",
-				PodNamespace: "",
-			},
-			arg: &Endpoint{
-				ContainerIDs: []string{"bar"},
-				ID:           1,
-				IPv4:         net.ParseIP("1.1.1.1"),
-				PodName:      "foo",
-				PodNamespace: "default",
-			},
-			want: true,
-		},
-	}
-	for _, tt := range tests {
-		t.Run(tt.name, func(t *testing.T) {
-			e := &Endpoint{
-				ContainerIDs: tt.fields.ContainerID,
-				ID:           tt.fields.ID,
-				IPv4:         tt.fields.IPv4,
-				IPv6:         tt.fields.IPv6,
-				PodName:      tt.fields.PodName,
-				PodNamespace: tt.fields.PodNamespace,
-			}
-			if got := e.EqualsByID(tt.arg); got != tt.want {
-				t.Errorf("Equals() = %v, want %v", got, tt.want)
-			}
-		})
-	}
-}
-
-func TestEndpoint_setFrom(t *testing.T) {
-	type fields struct {
-		ContainerIDs []string
-		ID           uint64
-		IPv4         net.IP
-		IPv6         net.IP
-		PodName      string
-		PodNamespace string
-		Labels       []string
-	}
-	type args struct {
-		o *Endpoint
-	}
-	tests := []struct {
-		name   string
-		fields fields
-		args   args
-		want   *Endpoint
-	}{
-		{
-			name: "all fields are copied",
-			fields: fields{
-				ContainerIDs: nil,
-				ID:           0,
-				IPv4:         nil,
-				IPv6:         nil,
-				PodName:      "",
-				PodNamespace: "",
-				Labels:       nil,
-			},
-			args: args{
-				o: &Endpoint{
-					ContainerIDs: []string{"foo"},
-					ID:           1,
-					IPv4:         net.ParseIP("1.1.1.1"),
-					IPv6:         net.ParseIP("fd00::"),
-					PodName:      "pod-bar",
-					PodNamespace: "cilium",
-					Labels:       []string{"a", "b"},
-				},
-			},
-			want: &Endpoint{
-				ContainerIDs: []string{"foo"},
-				ID:           1,
-				IPv4:         net.ParseIP("1.1.1.1"),
-				IPv6:         net.ParseIP("fd00::"),
-				PodName:      "pod-bar",
-				PodNamespace: "cilium",
-				Labels:       []string{"a", "b"},
-			},
-		},
-	}
-	for _, tt := range tests {
-		t.Run(tt.name, func(t *testing.T) {
-			e := &Endpoint{
-				ContainerIDs: tt.fields.ContainerIDs,
-				ID:           tt.fields.ID,
-				IPv4:         tt.fields.IPv4,
-				IPv6:         tt.fields.IPv6,
-				PodName:      tt.fields.PodName,
-				PodNamespace: tt.fields.PodNamespace,
-			}
-			e.setFrom(tt.args.o)
-			if !reflect.DeepEqual(e, tt.want) {
-				t.Errorf("setFrom() got = %v, want %v", e, tt.want)
-			}
-		})
-	}
-}
-
-func TestEndpoints_SyncEndpoints(t *testing.T) {
-	es := &Endpoints{
-		mutex: lock.RWMutex{},
-		eps:   []*Endpoint{},
-	}
-
-	eps := []*Endpoint{
-		{
-			ContainerIDs: []string{"313c63b8b164a19ec0fe42cd86c4159f3276ba8a415d77f340817fcfee2cb479"},
-			ID:           1,
-			IPv4:         net.ParseIP("1.1.1.1").To4(),
-			IPv6:         net.ParseIP("fd00::1").To16(),
-			PodName:      "foo",
-			PodNamespace: "default",
-		},
-		{
-			ContainerIDs: []string{"313c63b8b164a19ec0fe42cd86c4159f3276ba8a415d77f340817fcfee2cb471"},
-			ID:           2,
-			IPv4:         net.ParseIP("1.1.1.2").To4(),
-			IPv6:         net.ParseIP("fd00::2").To16(),
-			PodName:      "bar",
-			PodNamespace: "default",
-		},
-		{
-			ContainerIDs: []string{"313c63b8b164a19ec0fe42cd86c4159f3276ba8a415d77f340817fcfee2cb471"},
-			ID:           3,
-			IPv4:         net.ParseIP("1.1.1.3").To4(),
-			IPv6:         net.ParseIP("fd00::3").To16(),
-			PodName:      "bar",
-			PodNamespace: "kube-system",
-		},
-	}
-
-	endpointsWanted := []*Endpoint{
-		{
-			ContainerIDs: []string{"313c63b8b164a19ec0fe42cd86c4159f3276ba8a415d77f340817fcfee2cb479"},
-			ID:           1,
-			IPv4:         net.ParseIP("1.1.1.1").To4(),
-			IPv6:         net.ParseIP("fd00::1").To16(),
-			PodName:      "foo",
-			PodNamespace: "default",
-		},
-		{
-			ContainerIDs: []string{"313c63b8b164a19ec0fe42cd86c4159f3276ba8a415d77f340817fcfee2cb471"},
-			ID:           2,
-			IPv4:         net.ParseIP("1.1.1.2").To4(),
-			IPv6:         net.ParseIP("fd00::2").To16(),
-			PodName:      "bar",
-			PodNamespace: "default",
-		},
-		{
-			ContainerIDs: []string{"313c63b8b164a19ec0fe42cd86c4159f3276ba8a415d77f340817fcfee2cb471"},
-			ID:           3,
-			IPv4:         net.ParseIP("1.1.1.3").To4(),
-			IPv6:         net.ParseIP("fd00::3").To16(),
-			PodName:      "bar",
-			PodNamespace: "kube-system",
-		},
-	}
-
-	// add 2 new endpoints
-	es.SyncEndpoints(eps[0:2])
-
-	es.mutex.RLock()
-	// check if the endpoints were added
-	assert.EqualValues(t, endpointsWanted[0:2], es.eps)
-	es.mutex.RUnlock()
-
-	// Add only the first endpoint,
-	es.SyncEndpoints(eps[0:1])
-
-	es.mutex.Lock()
-	assert.EqualValues(t, endpointsWanted[0:1], es.eps)
-	es.mutex.Unlock()
-
-	// Re-add all endpoints
-	es.SyncEndpoints(endpointsWanted)
-	es.mutex.RLock()
-	// check if the endpoints were added
-	assert.EqualValues(t, endpointsWanted, es.eps)
-	es.mutex.RUnlock()
-}
-
-func TestEndpoints_FindEPs(t *testing.T) {
-	type args struct {
-		epID      uint64
-		namespace string
-		podName   string
-	}
-	tests := []struct {
-		name string
-		eps  []*Endpoint
-		args args
-		want []Endpoint
-	}{
-		{
-			name: "return all eps in a particular namespace",
-			eps: []*Endpoint{
-				{
-					ContainerIDs: []string{"313c63b8b164a19ec0fe42cd86c4159f3276ba8a415d77f340817fcfee2cb479"},
-					ID:           1,
-					IPv4:         net.ParseIP("1.1.1.1").To4(),
-					IPv6:         net.ParseIP("fd00::1").To16(),
-					PodName:      "foo",
-					PodNamespace: "default",
-				},
-				{
-					ContainerIDs: []string{"313c63b8b164a19ec0fe42cd86c4159f3276ba8a415d77f340817fcfee2cb471"},
-					ID:           2,
-					IPv4:         net.ParseIP("1.1.1.2").To4(),
-					IPv6:         net.ParseIP("fd00::2").To16(),
-					PodName:      "bar",
-					PodNamespace: "default",
-				},
-				{
-					ContainerIDs: []string{"313c63b8b164a19ec0fe42cd86c4159f3276ba8a415d77f340817fcfee2cb473"},
-					ID:           3,
-					IPv4:         net.ParseIP("1.1.1.3").To4(),
-					IPv6:         net.ParseIP("fd00::3").To16(),
-					PodName:      "bar",
-					PodNamespace: "kube-system",
-				},
-			},
-			args: args{
-				namespace: "default",
-			},
-			want: []Endpoint{
-				{
-					ContainerIDs: []string{"313c63b8b164a19ec0fe42cd86c4159f3276ba8a415d77f340817fcfee2cb479"},
-					ID:           1,
-					IPv4:         net.ParseIP("1.1.1.1").To4(),
-					IPv6:         net.ParseIP("fd00::1").To16(),
-					PodName:      "foo",
-					PodNamespace: "default",
-				},
-				{
-					ContainerIDs: []string{"313c63b8b164a19ec0fe42cd86c4159f3276ba8a415d77f340817fcfee2cb471"},
-					ID:           2,
-					IPv4:         net.ParseIP("1.1.1.2").To4(),
-					IPv6:         net.ParseIP("fd00::2").To16(),
-					PodName:      "bar",
-					PodNamespace: "default",
-				},
-			},
-		},
-		{
-			name: "return the ep of a pod",
-			eps: []*Endpoint{
-				{
-					ContainerIDs: []string{"313c63b8b164a19ec0fe42cd86c4159f3276ba8a415d77f340817fcfee2cb479"},
-					ID:           1,
-					IPv4:         net.ParseIP("1.1.1.1").To4(),
-					IPv6:         net.ParseIP("fd00::1").To16(),
-					PodName:      "foo",
-					PodNamespace: "default",
-				},
-				{
-					ContainerIDs: []string{"313c63b8b164a19ec0fe42cd86c4159f3276ba8a415d77f340817fcfee2cb471"},
-					ID:           2,
-					IPv4:         net.ParseIP("1.1.1.2").To4(),
-					IPv6:         net.ParseIP("fd00::2").To16(),
-					PodName:      "bar",
-					PodNamespace: "default",
-				},
-				{
-					ContainerIDs: []string{"313c63b8b164a19ec0fe42cd86c4159f3276ba8a415d77f340817fcfee2cb473"},
-					ID:           3,
-					IPv4:         net.ParseIP("1.1.1.3").To4(),
-					IPv6:         net.ParseIP("fd00::3").To16(),
-					PodName:      "bar",
-					PodNamespace: "kube-system",
-				},
-			},
-			args: args{
-				podName:   "bar",
-				namespace: "kube-system",
-			},
-			want: []Endpoint{
-				{
-					ContainerIDs: []string{"313c63b8b164a19ec0fe42cd86c4159f3276ba8a415d77f340817fcfee2cb473"},
-					ID:           3,
-					IPv4:         net.ParseIP("1.1.1.3").To4(),
-					IPv6:         net.ParseIP("fd00::3").To16(),
-					PodName:      "bar",
-					PodNamespace: "kube-system",
-				},
-			},
-		},
-		{
-			name: "return eps with the given pod name and namespace",
-			eps: []*Endpoint{
-				{
-					ContainerIDs: []string{"313c63b8b164a19ec0fe42cd86c4159f3276ba8a415d77f340817fcfee2cb479"},
-					ID:           1,
-					IPv4:         net.ParseIP("1.1.1.1").To4(),
-					IPv6:         net.ParseIP("fd00::1").To16(),
-					PodName:      "foo",
-					PodNamespace: "default",
-				},
-				{
-					ContainerIDs: []string{"313c63b8b164a19ec0fe42cd86c4159f3276ba8a415d77f340817fcfee2cb471"},
-					ID:           2,
-					IPv4:         net.ParseIP("1.1.1.2").To4(),
-					IPv6:         net.ParseIP("fd00::2").To16(),
-					PodName:      "bar",
-					PodNamespace: "default",
-				},
-				{
-					ContainerIDs: []string{"313c63b8b164a19ec0fe42cd86c4159f3276ba8a415d77f340817fcfee2cb473"},
-					ID:           3,
-					IPv4:         net.ParseIP("1.1.1.3").To4(),
-					IPv6:         net.ParseIP("fd00::3").To16(),
-					PodName:      "bar",
-					PodNamespace: "kube-system",
-				},
-			},
-			args: args{
-				epID: 2,
-			},
-			want: []Endpoint{
-				{
-					ContainerIDs: []string{"313c63b8b164a19ec0fe42cd86c4159f3276ba8a415d77f340817fcfee2cb471"},
-					ID:           2,
-					IPv4:         net.ParseIP("1.1.1.2").To4(),
-					IPv6:         net.ParseIP("fd00::2").To16(),
-					PodName:      "bar",
-					PodNamespace: "default",
-				},
-			},
-		},
-	}
-	for _, tt := range tests {
-		t.Run(tt.name, func(t *testing.T) {
-			es := &Endpoints{eps: tt.eps}
-			if got := es.FindEPs(tt.args.epID, tt.args.namespace, tt.args.podName); !reflect.DeepEqual(got, tt.want) {
-				t.Errorf("FindEPs() = %v, want %v", got, tt.want)
-			}
-		})
-	}
-
-	// Test that we can modify the endpoint without disrupting the original
-	// endpoint
-	epWant := Endpoint{
-		ContainerIDs: []string{"313c63b8b164a19ec0fe42cd86c4159f3276ba8a415d77f340817fcfee2cb479"},
-		ID:           1,
-		IPv4:         net.ParseIP("1.1.1.1").To4(),
-		IPv6:         net.ParseIP("fd00::1").To16(),
-		PodName:      "foo",
-		PodNamespace: "default",
-	}
-	eps := []*Endpoint{&epWant}
-	es := &Endpoints{
-		mutex: lock.RWMutex{},
-		eps:   eps,
-	}
-	gotEps := es.FindEPs(1, "", "")
-	assert.Len(t, gotEps, 1)
-	assert.Equal(t, gotEps[0], epWant, 1)
-	gotEps[0].ContainerIDs = append(gotEps[0].ContainerIDs, "foo")
-
-	epWantModified := Endpoint{
-		ContainerIDs: []string{"313c63b8b164a19ec0fe42cd86c4159f3276ba8a415d77f340817fcfee2cb479", "foo"},
-		ID:           1,
-		IPv4:         net.ParseIP("1.1.1.1").To4(),
-		IPv6:         net.ParseIP("fd00::1").To16(),
-		PodName:      "foo",
-		PodNamespace: "default",
-	}
-
-	assert.NotEqual(t, gotEps[0], epWant, 1)
-	assert.Equal(t, gotEps[0], epWantModified, 1)
-}
-
-func TestEndpoints_GetEndpoint(t *testing.T) {
-	type args struct {
-		ip net.IP
-	}
-	tests := []struct {
-		name         string
-		eps          []*Endpoint
-		args         args
-		wantEndpoint *Endpoint
-		wantOk       bool
-	}{
-		{
-			name: "found",
-			eps: []*Endpoint{
-				{
-					ID:   15,
-					IPv4: net.ParseIP("1.1.1.1"),
-				},
-			},
-			args: args{
-				ip: net.ParseIP("1.1.1.1"),
-			},
-			wantEndpoint: &Endpoint{
-				ID:   15,
-				IPv4: net.ParseIP("1.1.1.1"),
-			},
-			wantOk: true,
-		},
-		{
-			name: "not found",
-			eps: []*Endpoint{
-				{
-					ID:   15,
-					IPv4: net.ParseIP("1.1.1.1"),
-				},
-			},
-			args: args{
-				ip: net.ParseIP("1.1.1.2"),
-			},
-			wantEndpoint: nil,
-			wantOk:       false,
-		},
-	}
-	for _, tt := range tests {
-		t.Run(tt.name, func(t *testing.T) {
-			es := &Endpoints{eps: tt.eps}
-			gotEndpoint, gotOk := es.GetEndpoint(tt.args.ip)
-			if !reflect.DeepEqual(gotEndpoint, tt.wantEndpoint) {
-				t.Errorf("GetEndpoint() gotEndpoint = %v, want %v", gotEndpoint, tt.wantEndpoint)
-			}
-			if gotOk != tt.wantOk {
-				t.Errorf("GetEndpoint() gotOk = %v, want %v", gotOk, tt.wantOk)
-			}
-		})
-	}
-}
-
-func TestEndpoints_DeleteEndpoint(t *testing.T) {
-	tests := []struct {
-		name    string
-		eps     []*Endpoint
-		del     *Endpoint
-		wantEps []*Endpoint
-	}{
-		{
-			name:    "delete non-existing endpoint",
-			eps:     []*Endpoint{{ID: 15}},
-			del:     &Endpoint{ID: 12},
-			wantEps: []*Endpoint{{ID: 15}},
-		}, {
-			name:    "delete first endpoint",
-			eps:     []*Endpoint{{ID: 15}, {ID: 20}},
-			del:     &Endpoint{ID: 15},
-			wantEps: []*Endpoint{{ID: 20}},
-		}, {
-			name:    "delete last endpoint",
-			eps:     []*Endpoint{{ID: 15}, {ID: 20}},
-			del:     &Endpoint{ID: 20},
-			wantEps: []*Endpoint{{ID: 15}},
-		},
-	}
-	for _, tt := range tests {
-		t.Run(tt.name, func(t *testing.T) {
-			es := &Endpoints{eps: tt.eps}
-			es.DeleteEndpoint(tt.del)
-			assert.Equal(t, tt.wantEps, es.eps)
-		})
-	}
-}
-
-func TestEndpoints_GetEndpointByContainerID(t *testing.T) {
-	type args struct {
-		id string
-	}
-	tests := []struct {
-		name         string
-		eps          []*Endpoint
-		args         args
-		wantEndpoint *Endpoint
-		wantOk       bool
-	}{
-		{
-			name: "found",
-			eps: []*Endpoint{
-				{
-					ID:           15,
-					ContainerIDs: []string{"c0", "c1"},
-				},
-			},
-			args: args{
-				id: "c1",
-			},
-			wantEndpoint: &Endpoint{
-				ID:           15,
-				ContainerIDs: []string{"c0", "c1"},
-			},
-			wantOk: true,
-		},
-		{
-			name: "not found",
-			eps: []*Endpoint{
-				{
-					ID:           15,
-					ContainerIDs: []string{"c0", "c1"},
-				},
-			},
-			args: args{
-				id: "c2",
-			},
-			wantEndpoint: nil,
-			wantOk:       false,
-		},
-	}
-	for _, tt := range tests {
-		t.Run(tt.name, func(t *testing.T) {
-			es := Endpoints{eps: tt.eps}
-			gotEndpoint, gotOk := es.GetEndpointByContainerID(tt.args.id)
-			assert.Equal(t, tt.wantEndpoint, gotEndpoint)
-			assert.Equal(t, tt.wantOk, gotOk)
-		})
-	}
-}
-
-func TestEndpoint_Copy(t *testing.T) {
-	ep := &Endpoint{
-		ContainerIDs: nil,
-		ID:           0,
-		IPv4:         nil,
-		IPv6:         nil,
-		PodName:      "",
-		PodNamespace: "",
-		Labels:       nil,
-	}
-	cp := ep.DeepCopy()
-	assert.Equal(t, ep, cp)
-
-	ep = &Endpoint{
-		ContainerIDs: []string{"c1", "c2"},
-		ID:           3,
-		IPv4:         net.ParseIP("1.1.1.1"),
-		IPv6:         net.ParseIP("ffff:ffff:ffff:ffff:ffff:ffff:ffff:ffff"),
-		PodName:      "pod1",
-		PodNamespace: "ns1",
-		Labels:       []string{"a=b", "c=d"},
-	}
-	cp1 := ep.DeepCopy()
-	cp2 := ep.DeepCopy()
-	assert.Equal(t, ep, cp2)
-	assert.Equal(t, cp1, cp2)
-	cp1.ContainerIDs = []string{"c3", "c4"}
-	cp1.ID = 4
-	cp1.IPv4 = net.ParseIP("2.2.2.2")
-	cp1.IPv4 = net.ParseIP("eeee:eeee:eeee:eeee:eeee:eeee:eeee:eeee")
-	cp1.PodName = "pod1"
-	cp1.PodNamespace = "ns1"
-	cp1.Labels = []string{"e=f"}
-	assert.Equal(t, ep, cp2)
-	assert.NotEqual(t, cp1, cp2)
-}
-
-func TestEndpoints_GetEndpointByPodName(t *testing.T) {
-	type args struct {
-		namespace string
-		name      string
-	}
-	tests := []struct {
-		name         string
-		eps          []*Endpoint
-		args         args
-		wantEndpoint *Endpoint
-		wantOk       bool
-	}{
-		{
-			name: "found",
-			eps: []*Endpoint{
-				{PodNamespace: "ns1", PodName: "pod1"},
-				{PodNamespace: "ns2", PodName: "pod2"},
-			},
-			args: args{
-				namespace: "ns2",
-				name:      "pod2",
-			},
-			wantEndpoint: &Endpoint{
-				PodNamespace: "ns2",
-				PodName:      "pod2",
-			},
-			wantOk: true,
-		},
-		{
-			name: "not found",
-			eps: []*Endpoint{
-				{PodNamespace: "ns1", PodName: "pod1"},
-				{PodNamespace: "ns2", PodName: "pod2"},
-			},
-			args: args{
-				namespace: "ns3",
-				name:      "pod3",
-			},
-			wantEndpoint: nil,
-			wantOk:       false,
-		},
-	}
-	for _, tt := range tests {
-		t.Run(tt.name, func(t *testing.T) {
-			es := Endpoints{eps: tt.eps}
-			gotEndpoint, gotOk := es.GetEndpointByPodName(tt.args.namespace, tt.args.name)
-			assert.Equal(t, tt.wantEndpoint, gotEndpoint)
-			assert.Equal(t, tt.wantOk, gotOk)
-		})
-	}
-}
diff --git a/pkg/hubble/api/v1/http_test.go b/pkg/hubble/api/v1/http_test.go
deleted file mode 100644
index 022762d6a6..0000000000
--- a/pkg/hubble/api/v1/http_test.go
+++ /dev/null
@@ -1,52 +0,0 @@
-// Copyright 2019 Authors of Hubble
-//
-// Licensed under the Apache License, Version 2.0 (the "License");
-// you may not use this file except in compliance with the License.
-// You may obtain a copy of the License at
-//
-//     http://www.apache.org/licenses/LICENSE-2.0
-//
-// Unless required by applicable law or agreed to in writing, software
-// distributed under the License is distributed on an "AS IS" BASIS,
-// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
-// See the License for the specific language governing permissions and
-// limitations under the License.
-
-// +build !privileged_tests
-
-package v1
-
-import (
-	"testing"
-
-	pb "github.com/cilium/cilium/api/v1/flow"
-
-	"github.com/stretchr/testify/assert"
-)
-
-func TestCompareHTTP(t *testing.T) {
-	type testDef struct {
-		a      *pb.HTTP
-		b      *pb.HTTP
-		result bool
-	}
-
-	matrix := []testDef{
-		{a: &pb.HTTP{}, b: &pb.HTTP{}, result: true},
-		{a: &pb.HTTP{Code: 10}, b: &pb.HTTP{Code: 10}, result: true},
-		{a: &pb.HTTP{Code: 10}, b: &pb.HTTP{Code: 20}, result: false},
-		{a: &pb.HTTP{Method: "GET"}, b: &pb.HTTP{Method: "GET"}, result: true},
-		{a: &pb.HTTP{Method: "GET"}, b: &pb.HTTP{Method: "POST"}, result: false},
-		{a: &pb.HTTP{Method: "GET", Url: "/path"}, b: &pb.HTTP{Method: "GET", Url: "/path"}, result: true},
-		{a: &pb.HTTP{Method: "GET", Url: "/path"}, b: &pb.HTTP{Method: "GET", Url: "/other"}, result: false},
-		{a: &pb.HTTP{Protocol: "HTTP"}, b: &pb.HTTP{Protocol: "HTTP"}, result: true},
-		{a: &pb.HTTP{Protocol: "HTTP"}, b: &pb.HTTP{Protocol: "HTTP/2"}, result: false},
-		// HTTP header must be ignored in comparison
-		{a: &pb.HTTP{Method: "GET", Headers: []*pb.HTTPHeader{{Key: "foo", Value: "value"}}}, b: &pb.HTTP{Method: "GET"}, result: true},
-	}
-
-	for _, test := range matrix {
-		assert.EqualValues(t, LooseCompareHTTP(test.a, test.b), test.result)
-	}
-
-}
diff --git a/pkg/hubble/api/v1/kafka_test.go b/pkg/hubble/api/v1/kafka_test.go
deleted file mode 100644
index 0734848e95..0000000000
--- a/pkg/hubble/api/v1/kafka_test.go
+++ /dev/null
@@ -1,52 +0,0 @@
-// Copyright 2019 Authors of Hubble
-//
-// Licensed under the Apache License, Version 2.0 (the "License");
-// you may not use this file except in compliance with the License.
-// You may obtain a copy of the License at
-//
-//     http://www.apache.org/licenses/LICENSE-2.0
-//
-// Unless required by applicable law or agreed to in writing, software
-// distributed under the License is distributed on an "AS IS" BASIS,
-// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
-// See the License for the specific language governing permissions and
-// limitations under the License.
-
-// +build !privileged_tests
-
-package v1
-
-import (
-	"testing"
-
-	pb "github.com/cilium/cilium/api/v1/flow"
-
-	"github.com/stretchr/testify/assert"
-)
-
-func TestCompareKafka(t *testing.T) {
-	type testDef struct {
-		a      *pb.Kafka
-		b      *pb.Kafka
-		result bool
-	}
-
-	matrix := []testDef{
-		{a: &pb.Kafka{}, b: &pb.Kafka{}, result: true},
-		{a: &pb.Kafka{ErrorCode: 10}, b: &pb.Kafka{ErrorCode: 10}, result: true},
-		{a: &pb.Kafka{ErrorCode: 10}, b: &pb.Kafka{ErrorCode: 20}, result: false},
-		{a: &pb.Kafka{ErrorCode: 10, ApiKey: "foo"}, b: &pb.Kafka{ErrorCode: 10, ApiKey: "foo"}, result: true},
-		{a: &pb.Kafka{ErrorCode: 10, ApiKey: "foo"}, b: &pb.Kafka{ErrorCode: 10, ApiKey: "bar"}, result: false},
-		{a: &pb.Kafka{ApiVersion: 10}, b: &pb.Kafka{ApiVersion: 10}, result: true},
-		{a: &pb.Kafka{ApiVersion: 10}, b: &pb.Kafka{ApiVersion: 20}, result: false},
-		{a: &pb.Kafka{CorrelationId: 10}, b: &pb.Kafka{CorrelationId: 10}, result: true},
-		{a: &pb.Kafka{CorrelationId: 10}, b: &pb.Kafka{CorrelationId: 20}, result: false},
-		{a: &pb.Kafka{ErrorCode: 10, Topic: "foo"}, b: &pb.Kafka{ErrorCode: 10, Topic: "foo"}, result: true},
-		{a: &pb.Kafka{ErrorCode: 10, Topic: "foo"}, b: &pb.Kafka{ErrorCode: 10, Topic: "bar"}, result: false},
-	}
-
-	for _, test := range matrix {
-		assert.EqualValues(t, CompareKafka(test.a, test.b), test.result)
-	}
-
-}
diff --git a/pkg/hubble/container/ring_reader_test.go b/pkg/hubble/container/ring_reader_test.go
index 5e3304f170..fe4281063b 100644
--- a/pkg/hubble/container/ring_reader_test.go
+++ b/pkg/hubble/container/ring_reader_test.go
@@ -23,7 +23,7 @@ import (
 	"testing"
 	"time"
 
-	"github.com/cilium/cilium/pkg/hubble/api/v1"
+	v1 "github.com/cilium/cilium/pkg/hubble/api/v1"
 
 	"github.com/golang/protobuf/ptypes/timestamp"
 	"github.com/stretchr/testify/assert"
diff --git a/pkg/hubble/filters/filters_test.go b/pkg/hubble/filters/filters_test.go
index 0757539295..0a3fa38551 100644
--- a/pkg/hubble/filters/filters_test.go
+++ b/pkg/hubble/filters/filters_test.go
@@ -20,7 +20,7 @@ import (
 	"context"
 	"testing"
 
-	pb "github.com/cilium/cilium/api/v1/flow"
+	flowpb "github.com/cilium/cilium/api/v1/flow"
 	v1 "github.com/cilium/cilium/pkg/hubble/api/v1"
 	"github.com/stretchr/testify/assert"
 )
@@ -84,40 +84,40 @@ func TestMatch(t *testing.T) {
 
 type testFilterTrue struct{}
 
-func (t *testFilterTrue) OnBuildFilter(_ context.Context, ff *pb.FlowFilter) ([]FilterFunc, error) {
+func (t *testFilterTrue) OnBuildFilter(_ context.Context, ff *flowpb.FlowFilter) ([]FilterFunc, error) {
 	return []FilterFunc{func(ev *v1.Event) bool { return true }}, nil
 }
 
 type testFilterFalse struct{}
 
-func (t *testFilterFalse) OnBuildFilter(_ context.Context, ff *pb.FlowFilter) ([]FilterFunc, error) {
+func (t *testFilterFalse) OnBuildFilter(_ context.Context, ff *flowpb.FlowFilter) ([]FilterFunc, error) {
 	return []FilterFunc{func(ev *v1.Event) bool { return false }}, nil
 }
 
 func TestOnBuildFilter(t *testing.T) {
 	fl, err := BuildFilterList(context.Background(),
-		[]*pb.FlowFilter{{SourceIdentity: []uint32{1, 2, 3}}}, // true
-		[]OnBuildFilter{&testFilterTrue{}})                    // true
+		[]*flowpb.FlowFilter{{SourceIdentity: []uint32{1, 2, 3}}}, // true
+		[]OnBuildFilter{&testFilterTrue{}})                        // true
 	assert.NoError(t, err)
-	assert.Equal(t, true, fl.MatchAll(&v1.Event{Event: &pb.Flow{
-		Source: &pb.Endpoint{Identity: 3},
+	assert.Equal(t, true, fl.MatchAll(&v1.Event{Event: &flowpb.Flow{
+		Source: &flowpb.Endpoint{Identity: 3},
 	}}))
 
 	fl, err = BuildFilterList(context.Background(),
-		[]*pb.FlowFilter{{SourceIdentity: []uint32{1, 2, 3}}}, // true
-		[]OnBuildFilter{&testFilterFalse{}})                   // false
+		[]*flowpb.FlowFilter{{SourceIdentity: []uint32{1, 2, 3}}}, // true
+		[]OnBuildFilter{&testFilterFalse{}})                       // false
 	assert.NoError(t, err)
-	assert.Equal(t, false, fl.MatchAll(&v1.Event{Event: &pb.Flow{
-		Source: &pb.Endpoint{Identity: 3},
+	assert.Equal(t, false, fl.MatchAll(&v1.Event{Event: &flowpb.Flow{
+		Source: &flowpb.Endpoint{Identity: 3},
 	}}))
 
 	fl, err = BuildFilterList(context.Background(),
-		[]*pb.FlowFilter{{SourceIdentity: []uint32{1, 2, 3}}}, // true
+		[]*flowpb.FlowFilter{{SourceIdentity: []uint32{1, 2, 3}}}, // true
 		[]OnBuildFilter{
 			&testFilterFalse{}, // false
 			&testFilterTrue{}}) // true
 	assert.NoError(t, err)
-	assert.Equal(t, false, fl.MatchAll(&v1.Event{Event: &pb.Flow{
-		Source: &pb.Endpoint{Identity: 3},
+	assert.Equal(t, false, fl.MatchAll(&v1.Event{Event: &flowpb.Flow{
+		Source: &flowpb.Endpoint{Identity: 3},
 	}}))
 }
diff --git a/pkg/hubble/filters/fqdn_test.go b/pkg/hubble/filters/fqdn_test.go
index 2ff2b9bbc3..a867a3f5bb 100644
--- a/pkg/hubble/filters/fqdn_test.go
+++ b/pkg/hubble/filters/fqdn_test.go
@@ -20,7 +20,7 @@ import (
 	"context"
 	"testing"
 
-	pb "github.com/cilium/cilium/api/v1/flow"
+	flowpb "github.com/cilium/cilium/api/v1/flow"
 	v1 "github.com/cilium/cilium/pkg/hubble/api/v1"
 
 	"github.com/stretchr/testify/assert"
@@ -28,7 +28,7 @@ import (
 
 func TestFQDNFilter(t *testing.T) {
 	type args struct {
-		f  []*pb.FlowFilter
+		f  []*flowpb.FlowFilter
 		ev []*v1.Event
 	}
 	tests := []struct {
@@ -40,14 +40,14 @@ func TestFQDNFilter(t *testing.T) {
 		{
 			name: "source fqdn",
 			args: args{
-				f: []*pb.FlowFilter{
+				f: []*flowpb.FlowFilter{
 					{SourceFqdn: []string{"cilium.io", "ebpf.io"}},
 				},
 				ev: []*v1.Event{
-					{Event: &pb.Flow{SourceNames: []string{"cilium.io"}}},
-					{Event: &pb.Flow{SourceNames: []string{"ebpf.io"}}},
-					{Event: &pb.Flow{DestinationNames: []string{"cilium.io"}}},
-					{Event: &pb.Flow{DestinationNames: []string{"ebpf.io"}}},
+					{Event: &flowpb.Flow{SourceNames: []string{"cilium.io"}}},
+					{Event: &flowpb.Flow{SourceNames: []string{"ebpf.io"}}},
+					{Event: &flowpb.Flow{DestinationNames: []string{"cilium.io"}}},
+					{Event: &flowpb.Flow{DestinationNames: []string{"ebpf.io"}}},
 				},
 			},
 			want: []bool{
@@ -60,14 +60,14 @@ func TestFQDNFilter(t *testing.T) {
 		{
 			name: "destination fqdn",
 			args: args{
-				f: []*pb.FlowFilter{
+				f: []*flowpb.FlowFilter{
 					{DestinationFqdn: []string{"cilium.io", "ebpf.io"}},
 				},
 				ev: []*v1.Event{
-					{Event: &pb.Flow{SourceNames: []string{"cilium.io"}}},
-					{Event: &pb.Flow{SourceNames: []string{"ebpf.io"}}},
-					{Event: &pb.Flow{DestinationNames: []string{"cilium.io"}}},
-					{Event: &pb.Flow{DestinationNames: []string{"ebpf.io"}}},
+					{Event: &flowpb.Flow{SourceNames: []string{"cilium.io"}}},
+					{Event: &flowpb.Flow{SourceNames: []string{"ebpf.io"}}},
+					{Event: &flowpb.Flow{DestinationNames: []string{"cilium.io"}}},
+					{Event: &flowpb.Flow{DestinationNames: []string{"ebpf.io"}}},
 				},
 			},
 			want: []bool{
@@ -80,22 +80,22 @@ func TestFQDNFilter(t *testing.T) {
 		{
 			name: "source and destination fqdn",
 			args: args{
-				f: []*pb.FlowFilter{
+				f: []*flowpb.FlowFilter{
 					{
 						SourceFqdn:      []string{"cilium.io", "docs.cilium.io"},
 						DestinationFqdn: []string{"ebpf.io"},
 					},
 				},
 				ev: []*v1.Event{
-					{Event: &pb.Flow{
+					{Event: &flowpb.Flow{
 						SourceNames:      []string{"cilium.io"},
 						DestinationNames: []string{"ebpf.io"},
 					}},
-					{Event: &pb.Flow{
+					{Event: &flowpb.Flow{
 						SourceNames:      []string{"ebpf.io"},
 						DestinationNames: []string{"cilium.io"},
 					}},
-					{Event: &pb.Flow{
+					{Event: &flowpb.Flow{
 						SourceNames:      []string{"deathstar.empire.svc.cluster.local", "docs.cilium.io"},
 						DestinationNames: []string{"ebpf.io"},
 					}},
@@ -110,26 +110,26 @@ func TestFQDNFilter(t *testing.T) {
 		{
 			name: "source or destination fqdn",
 			args: args{
-				f: []*pb.FlowFilter{
+				f: []*flowpb.FlowFilter{
 					{SourceFqdn: []string{"cilium.io", "docs.cilium.io"}},
 					{DestinationFqdn: []string{"ebpf.io"}},
 				},
 				ev: []*v1.Event{
-					{Event: &pb.Flow{
+					{Event: &flowpb.Flow{
 						SourceNames:      []string{"cilium.io"},
 						DestinationNames: []string{"ebpf.io"},
 					}},
-					{Event: &pb.Flow{
+					{Event: &flowpb.Flow{
 						SourceNames:      []string{"ebpf.io"},
 						DestinationNames: []string{"cilium.io"},
 					}},
-					{Event: &pb.Flow{
+					{Event: &flowpb.Flow{
 						SourceNames: []string{"deathstar.empire.svc.cluster.local", "docs.cilium.io"},
 					}},
-					{Event: &pb.Flow{
+					{Event: &flowpb.Flow{
 						DestinationNames: []string{"ebpf.io"},
 					}},
-					{Event: &pb.Flow{
+					{Event: &flowpb.Flow{
 						SourceNames:      []string{"deathstar.empire.svc.cluster.local", "docs.cilium.io"},
 						DestinationNames: []string{"ebpf.io"},
 					}},
@@ -146,16 +146,16 @@ func TestFQDNFilter(t *testing.T) {
 		{
 			name: "invalid data",
 			args: args{
-				f: []*pb.FlowFilter{
+				f: []*flowpb.FlowFilter{
 					{SourceFqdn: []string{"cilium.io."}},
 				},
 				ev: []*v1.Event{
 					nil,
 					{},
-					{Event: &pb.Flow{}},
-					{Event: &pb.Flow{SourceNames: []string{"cilium.io."}}}, // should not have trailing dot
-					{Event: &pb.Flow{SourceNames: []string{"www.cilium.io"}}},
-					{Event: &pb.Flow{SourceNames: []string{""}}},
+					{Event: &flowpb.Flow{}},
+					{Event: &flowpb.Flow{SourceNames: []string{"cilium.io."}}}, // should not have trailing dot
+					{Event: &flowpb.Flow{SourceNames: []string{"www.cilium.io"}}},
+					{Event: &flowpb.Flow{SourceNames: []string{""}}},
 				},
 			},
 			want: []bool{
@@ -170,7 +170,7 @@ func TestFQDNFilter(t *testing.T) {
 		{
 			name: "invalid source fqdn filter",
 			args: args{
-				f: []*pb.FlowFilter{
+				f: []*flowpb.FlowFilter{
 					{SourceFqdn: []string{""}},
 				},
 			},
@@ -179,7 +179,7 @@ func TestFQDNFilter(t *testing.T) {
 		{
 			name: "invalid destination fqdn filter",
 			args: args{
-				f: []*pb.FlowFilter{
+				f: []*flowpb.FlowFilter{
 					{DestinationFqdn: []string{"."}},
 				},
 			},
@@ -188,18 +188,18 @@ func TestFQDNFilter(t *testing.T) {
 		{
 			name: "wildcard filters",
 			args: args{
-				f: []*pb.FlowFilter{
+				f: []*flowpb.FlowFilter{
 					{SourceFqdn: []string{"*.cilium.io", "*.org."}},
 					{DestinationFqdn: []string{"*"}},
 				},
 				ev: []*v1.Event{
-					{Event: &pb.Flow{SourceNames: []string{"www.cilium.io"}}},
-					{Event: &pb.Flow{SourceNames: []string{"multiple.domains.org"}}},
-					{Event: &pb.Flow{SourceNames: []string{"cilium.io"}}},
-					{Event: &pb.Flow{SourceNames: []string{"tiefighter", "empire.org"}}},
-					{Event: &pb.Flow{DestinationNames: []string{}}},
-					{Event: &pb.Flow{DestinationNames: []string{"anything.really"}}},
-					{Event: &pb.Flow{DestinationNames: []string{""}}},
+					{Event: &flowpb.Flow{SourceNames: []string{"www.cilium.io"}}},
+					{Event: &flowpb.Flow{SourceNames: []string{"multiple.domains.org"}}},
+					{Event: &flowpb.Flow{SourceNames: []string{"cilium.io"}}},
+					{Event: &flowpb.Flow{SourceNames: []string{"tiefighter", "empire.org"}}},
+					{Event: &flowpb.Flow{DestinationNames: []string{}}},
+					{Event: &flowpb.Flow{DestinationNames: []string{"anything.really"}}},
+					{Event: &flowpb.Flow{DestinationNames: []string{""}}},
 				},
 			},
 			want: []bool{
@@ -231,7 +231,7 @@ func TestFQDNFilter(t *testing.T) {
 
 func Test_filterByDNSQuery(t *testing.T) {
 	type args struct {
-		f  []*pb.FlowFilter
+		f  []*flowpb.FlowFilter
 		ev *v1.Event
 	}
 	tests := []struct {
@@ -243,8 +243,8 @@ func Test_filterByDNSQuery(t *testing.T) {
 		{
 			name: "not-dns",
 			args: args{
-				f:  []*pb.FlowFilter{{DnsQuery: []string{".*"}}},
-				ev: &v1.Event{Event: &pb.Flow{}},
+				f:  []*flowpb.FlowFilter{{DnsQuery: []string{".*"}}},
+				ev: &v1.Event{Event: &flowpb.Flow{}},
 			},
 			wantErr: false,
 			want:    false,
@@ -252,18 +252,18 @@ func Test_filterByDNSQuery(t *testing.T) {
 		{
 			name: "invalid-regex",
 			args: args{
-				f: []*pb.FlowFilter{{DnsQuery: []string{"*"}}},
+				f: []*flowpb.FlowFilter{{DnsQuery: []string{"*"}}},
 			},
 			wantErr: true,
 		},
 		{
 			name: "positive",
 			args: args{
-				f: []*pb.FlowFilter{{DnsQuery: []string{".*\\.com$", ".*\\.io"}}},
-				ev: &v1.Event{Event: &pb.Flow{
-					L7: &pb.Layer7{
-						Record: &pb.Layer7_Dns{
-							Dns: &pb.DNS{
+				f: []*flowpb.FlowFilter{{DnsQuery: []string{".*\\.com$", ".*\\.io"}}},
+				ev: &v1.Event{Event: &flowpb.Flow{
+					L7: &flowpb.Layer7{
+						Record: &flowpb.Layer7_Dns{
+							Dns: &flowpb.DNS{
 								Query: "cilium.io",
 							},
 						},
@@ -275,11 +275,11 @@ func Test_filterByDNSQuery(t *testing.T) {
 		{
 			name: "positive",
 			args: args{
-				f: []*pb.FlowFilter{{DnsQuery: []string{".*\\.com$", ".*\\.io"}}},
-				ev: &v1.Event{Event: &pb.Flow{
-					L7: &pb.Layer7{
-						Record: &pb.Layer7_Dns{
-							Dns: &pb.DNS{
+				f: []*flowpb.FlowFilter{{DnsQuery: []string{".*\\.com$", ".*\\.io"}}},
+				ev: &v1.Event{Event: &flowpb.Flow{
+					L7: &flowpb.Layer7{
+						Record: &flowpb.Layer7_Dns{
+							Dns: &flowpb.DNS{
 								Query: "cilium.io",
 							},
 						},
@@ -292,11 +292,11 @@ func Test_filterByDNSQuery(t *testing.T) {
 		{
 			name: "negative",
 			args: args{
-				f: []*pb.FlowFilter{{DnsQuery: []string{".*\\.com$", ".*\\.net"}}},
-				ev: &v1.Event{Event: &pb.Flow{
-					L7: &pb.Layer7{
-						Record: &pb.Layer7_Dns{
-							Dns: &pb.DNS{
+				f: []*flowpb.FlowFilter{{DnsQuery: []string{".*\\.com$", ".*\\.net"}}},
+				ev: &v1.Event{Event: &flowpb.Flow{
+					L7: &flowpb.Layer7{
+						Record: &flowpb.Layer7_Dns{
+							Dns: &flowpb.DNS{
 								Query: "cilium.io",
 							},
 						},
diff --git a/pkg/hubble/filters/http_test.go b/pkg/hubble/filters/http_test.go
index 8d456714a9..cf4530d460 100644
--- a/pkg/hubble/filters/http_test.go
+++ b/pkg/hubble/filters/http_test.go
@@ -20,20 +20,20 @@ import (
 	"context"
 	"testing"
 
-	pb "github.com/cilium/cilium/api/v1/flow"
+	flowpb "github.com/cilium/cilium/api/v1/flow"
 	v1 "github.com/cilium/cilium/pkg/hubble/api/v1"
 	"github.com/cilium/cilium/pkg/monitor/api"
 )
 
 func TestHttpStatusCodeFilter(t *testing.T) {
-	httpFlow := func(http *pb.HTTP) *v1.Event {
+	httpFlow := func(http *flowpb.HTTP) *v1.Event {
 		return &v1.Event{
-			Event: &pb.Flow{
-				EventType: &pb.CiliumEventType{
+			Event: &flowpb.Flow{
+				EventType: &flowpb.CiliumEventType{
 					Type: api.MessageTypeAccessLog,
 				},
-				L7: &pb.Layer7{
-					Record: &pb.Layer7_Http{
+				L7: &flowpb.Layer7{
+					Record: &flowpb.Layer7_Http{
 						Http: http,
 					},
 				}},
@@ -41,7 +41,7 @@ func TestHttpStatusCodeFilter(t *testing.T) {
 	}
 
 	type args struct {
-		f  []*pb.FlowFilter
+		f  []*flowpb.FlowFilter
 		ev []*v1.Event
 	}
 	tests := []struct {
@@ -53,17 +53,17 @@ func TestHttpStatusCodeFilter(t *testing.T) {
 		{
 			name: "status code full",
 			args: args{
-				f: []*pb.FlowFilter{
+				f: []*flowpb.FlowFilter{
 					{
 						HttpStatusCode: []string{"200", "302"},
-						EventType:      []*pb.EventTypeFilter{{Type: api.MessageTypeAccessLog}},
+						EventType:      []*flowpb.EventTypeFilter{{Type: api.MessageTypeAccessLog}},
 					},
 				},
 				ev: []*v1.Event{
-					httpFlow(&pb.HTTP{Code: 200}),
-					httpFlow(&pb.HTTP{Code: 302}),
-					httpFlow(&pb.HTTP{Code: 404}),
-					httpFlow(&pb.HTTP{Code: 500}),
+					httpFlow(&flowpb.HTTP{Code: 200}),
+					httpFlow(&flowpb.HTTP{Code: 302}),
+					httpFlow(&flowpb.HTTP{Code: 404}),
+					httpFlow(&flowpb.HTTP{Code: 500}),
 				},
 			},
 			want: []bool{
@@ -77,22 +77,22 @@ func TestHttpStatusCodeFilter(t *testing.T) {
 		{
 			name: "status code prefix",
 			args: args{
-				f: []*pb.FlowFilter{
+				f: []*flowpb.FlowFilter{
 					{
 						HttpStatusCode: []string{"40+", "5+"},
-						EventType:      []*pb.EventTypeFilter{{Type: api.MessageTypeAccessLog}},
+						EventType:      []*flowpb.EventTypeFilter{{Type: api.MessageTypeAccessLog}},
 					},
 				},
 				ev: []*v1.Event{
-					httpFlow(&pb.HTTP{Code: 302}),
-					httpFlow(&pb.HTTP{Code: 400}),
-					httpFlow(&pb.HTTP{Code: 404}),
-					httpFlow(&pb.HTTP{Code: 410}),
-					httpFlow(&pb.HTTP{Code: 004}),
-					httpFlow(&pb.HTTP{Code: 500}),
-					httpFlow(&pb.HTTP{Code: 501}),
-					httpFlow(&pb.HTTP{Code: 510}),
-					httpFlow(&pb.HTTP{Code: 050}),
+					httpFlow(&flowpb.HTTP{Code: 302}),
+					httpFlow(&flowpb.HTTP{Code: 400}),
+					httpFlow(&flowpb.HTTP{Code: 404}),
+					httpFlow(&flowpb.HTTP{Code: 410}),
+					httpFlow(&flowpb.HTTP{Code: 004}),
+					httpFlow(&flowpb.HTTP{Code: 500}),
+					httpFlow(&flowpb.HTTP{Code: 501}),
+					httpFlow(&flowpb.HTTP{Code: 510}),
+					httpFlow(&flowpb.HTTP{Code: 050}),
 				},
 			},
 			want: []bool{
@@ -111,16 +111,16 @@ func TestHttpStatusCodeFilter(t *testing.T) {
 		{
 			name: "invalid data",
 			args: args{
-				f: []*pb.FlowFilter{
+				f: []*flowpb.FlowFilter{
 					{
 						HttpStatusCode: []string{"200"},
-						EventType:      []*pb.EventTypeFilter{{Type: api.MessageTypeAccessLog}},
+						EventType:      []*flowpb.EventTypeFilter{{Type: api.MessageTypeAccessLog}},
 					},
 				},
 				ev: []*v1.Event{
-					{Event: &pb.Flow{}},
-					httpFlow(&pb.HTTP{}),
-					httpFlow(&pb.HTTP{Code: 777}),
+					{Event: &flowpb.Flow{}},
+					httpFlow(&flowpb.HTTP{}),
+					httpFlow(&flowpb.HTTP{Code: 777}),
 				},
 			},
 			want: []bool{
@@ -133,10 +133,10 @@ func TestHttpStatusCodeFilter(t *testing.T) {
 		{
 			name: "invalid empty filter",
 			args: args{
-				f: []*pb.FlowFilter{
+				f: []*flowpb.FlowFilter{
 					{
 						HttpStatusCode: []string{""},
-						EventType:      []*pb.EventTypeFilter{{Type: api.MessageTypeAccessLog}},
+						EventType:      []*flowpb.EventTypeFilter{{Type: api.MessageTypeAccessLog}},
 					},
 				},
 			},
@@ -145,10 +145,10 @@ func TestHttpStatusCodeFilter(t *testing.T) {
 		{
 			name: "invalid catch-all prefix",
 			args: args{
-				f: []*pb.FlowFilter{
+				f: []*flowpb.FlowFilter{
 					{
 						HttpStatusCode: []string{"+"},
-						EventType:      []*pb.EventTypeFilter{{Type: api.MessageTypeAccessLog}},
+						EventType:      []*flowpb.EventTypeFilter{{Type: api.MessageTypeAccessLog}},
 					},
 				},
 			},
@@ -157,10 +157,10 @@ func TestHttpStatusCodeFilter(t *testing.T) {
 		{
 			name: "invalid status code",
 			args: args{
-				f: []*pb.FlowFilter{
+				f: []*flowpb.FlowFilter{
 					{
 						HttpStatusCode: []string{"909"},
-						EventType:      []*pb.EventTypeFilter{{Type: api.MessageTypeAccessLog}},
+						EventType:      []*flowpb.EventTypeFilter{{Type: api.MessageTypeAccessLog}},
 					},
 				},
 			},
@@ -169,10 +169,10 @@ func TestHttpStatusCodeFilter(t *testing.T) {
 		{
 			name: "invalid status code prefix",
 			args: args{
-				f: []*pb.FlowFilter{
+				f: []*flowpb.FlowFilter{
 					{
 						HttpStatusCode: []string{"3++"},
-						EventType:      []*pb.EventTypeFilter{{Type: api.MessageTypeAccessLog}},
+						EventType:      []*flowpb.EventTypeFilter{{Type: api.MessageTypeAccessLog}},
 					},
 				},
 			},
@@ -181,10 +181,10 @@ func TestHttpStatusCodeFilter(t *testing.T) {
 		{
 			name: "invalid status code prefix",
 			args: args{
-				f: []*pb.FlowFilter{
+				f: []*flowpb.FlowFilter{
 					{
 						HttpStatusCode: []string{"3+0"},
-						EventType:      []*pb.EventTypeFilter{{Type: api.MessageTypeAccessLog}},
+						EventType:      []*flowpb.EventTypeFilter{{Type: api.MessageTypeAccessLog}},
 					},
 				},
 			},
@@ -193,14 +193,14 @@ func TestHttpStatusCodeFilter(t *testing.T) {
 		{
 			name: "empty event type filter",
 			args: args{
-				f: []*pb.FlowFilter{
+				f: []*flowpb.FlowFilter{
 					{
 						HttpStatusCode: []string{"200"},
-						EventType:      []*pb.EventTypeFilter{},
+						EventType:      []*flowpb.EventTypeFilter{},
 					},
 				},
 				ev: []*v1.Event{
-					httpFlow(&pb.HTTP{Code: 200}),
+					httpFlow(&flowpb.HTTP{Code: 200}),
 				},
 			},
 			want: []bool{
@@ -211,17 +211,17 @@ func TestHttpStatusCodeFilter(t *testing.T) {
 		{
 			name: "compatible event type filter",
 			args: args{
-				f: []*pb.FlowFilter{
+				f: []*flowpb.FlowFilter{
 					{
 						HttpStatusCode: []string{"200"},
-						EventType: []*pb.EventTypeFilter{
+						EventType: []*flowpb.EventTypeFilter{
 							{Type: api.MessageTypeAccessLog},
 							{Type: api.MessageTypeTrace},
 						},
 					},
 				},
 				ev: []*v1.Event{
-					httpFlow(&pb.HTTP{Code: 200}),
+					httpFlow(&flowpb.HTTP{Code: 200}),
 				},
 			},
 			want: []bool{
diff --git a/pkg/hubble/filters/identitty_test.go b/pkg/hubble/filters/identity_test.go
similarity index 79%
rename from pkg/hubble/filters/identitty_test.go
rename to pkg/hubble/filters/identity_test.go
index 007553d7fe..e1ed40975c 100644
--- a/pkg/hubble/filters/identitty_test.go
+++ b/pkg/hubble/filters/identity_test.go
@@ -20,7 +20,7 @@ import (
 	"context"
 	"testing"
 
-	pb "github.com/cilium/cilium/api/v1/flow"
+	flowpb "github.com/cilium/cilium/api/v1/flow"
 	v1 "github.com/cilium/cilium/pkg/hubble/api/v1"
 
 	"github.com/stretchr/testify/assert"
@@ -28,7 +28,7 @@ import (
 
 func TestIdentityFilter(t *testing.T) {
 	type args struct {
-		f  []*pb.FlowFilter
+		f  []*flowpb.FlowFilter
 		ev *v1.Event
 	}
 	tests := []struct {
@@ -39,7 +39,7 @@ func TestIdentityFilter(t *testing.T) {
 		{
 			name: "source-nil",
 			args: args{
-				f: []*pb.FlowFilter{{
+				f: []*flowpb.FlowFilter{{
 					SourceIdentity: []uint32{1},
 				}},
 				ev: nil,
@@ -49,7 +49,7 @@ func TestIdentityFilter(t *testing.T) {
 		{
 			name: "destination-nil",
 			args: args{
-				f: []*pb.FlowFilter{{
+				f: []*flowpb.FlowFilter{{
 					DestinationIdentity: []uint32{1},
 				}},
 				ev: nil,
@@ -59,11 +59,11 @@ func TestIdentityFilter(t *testing.T) {
 		{
 			name: "source-positive",
 			args: args{
-				f: []*pb.FlowFilter{{
+				f: []*flowpb.FlowFilter{{
 					SourceIdentity: []uint32{1, 2, 3},
 				}},
-				ev: &v1.Event{Event: &pb.Flow{
-					Source: &pb.Endpoint{Identity: 3},
+				ev: &v1.Event{Event: &flowpb.Flow{
+					Source: &flowpb.Endpoint{Identity: 3},
 				}},
 			},
 			want: true,
@@ -71,11 +71,11 @@ func TestIdentityFilter(t *testing.T) {
 		{
 			name: "source-negative",
 			args: args{
-				f: []*pb.FlowFilter{{
+				f: []*flowpb.FlowFilter{{
 					SourceIdentity: []uint32{1, 2, 3},
 				}},
-				ev: &v1.Event{Event: &pb.Flow{
-					Source: &pb.Endpoint{Identity: 4},
+				ev: &v1.Event{Event: &flowpb.Flow{
+					Source: &flowpb.Endpoint{Identity: 4},
 				}},
 			},
 			want: false,
@@ -83,11 +83,11 @@ func TestIdentityFilter(t *testing.T) {
 		{
 			name: "destination-negative",
 			args: args{
-				f: []*pb.FlowFilter{{
+				f: []*flowpb.FlowFilter{{
 					DestinationIdentity: []uint32{1, 2, 3},
 				}},
-				ev: &v1.Event{Event: &pb.Flow{
-					Destination: &pb.Endpoint{Identity: 5},
+				ev: &v1.Event{Event: &flowpb.Flow{
+					Destination: &flowpb.Endpoint{Identity: 5},
 				}},
 			},
 			want: false,
diff --git a/pkg/hubble/filters/ip_test.go b/pkg/hubble/filters/ip_test.go
index 60c63a5067..a91aed7093 100644
--- a/pkg/hubble/filters/ip_test.go
+++ b/pkg/hubble/filters/ip_test.go
@@ -20,13 +20,13 @@ import (
 	"context"
 	"testing"
 
-	pb "github.com/cilium/cilium/api/v1/flow"
+	flowpb "github.com/cilium/cilium/api/v1/flow"
 	v1 "github.com/cilium/cilium/pkg/hubble/api/v1"
 )
 
 func TestIPFilter(t *testing.T) {
 	type args struct {
-		f  []*pb.FlowFilter
+		f  []*flowpb.FlowFilter
 		ev []*v1.Event
 	}
 	tests := []struct {
@@ -38,14 +38,14 @@ func TestIPFilter(t *testing.T) {
 		{
 			name: "source ip",
 			args: args{
-				f: []*pb.FlowFilter{
+				f: []*flowpb.FlowFilter{
 					{SourceIp: []string{"1.1.1.1", "f00d::a10:0:0:9195"}},
 				},
 				ev: []*v1.Event{
-					{Event: &pb.Flow{IP: &pb.IP{Source: "1.1.1.1", Destination: "2.2.2.2"}}},
-					{Event: &pb.Flow{IP: &pb.IP{Source: "2.2.2.2", Destination: "1.1.1.1"}}},
-					{Event: &pb.Flow{IP: &pb.IP{Source: "f00d::a10:0:0:9195", Destination: "ff02::1:ff00:b3e5"}}},
-					{Event: &pb.Flow{IP: &pb.IP{Source: "ff02::1:ff00:b3e5", Destination: "f00d::a10:0:0:9195"}}},
+					{Event: &flowpb.Flow{IP: &flowpb.IP{Source: "1.1.1.1", Destination: "2.2.2.2"}}},
+					{Event: &flowpb.Flow{IP: &flowpb.IP{Source: "2.2.2.2", Destination: "1.1.1.1"}}},
+					{Event: &flowpb.Flow{IP: &flowpb.IP{Source: "f00d::a10:0:0:9195", Destination: "ff02::1:ff00:b3e5"}}},
+					{Event: &flowpb.Flow{IP: &flowpb.IP{Source: "ff02::1:ff00:b3e5", Destination: "f00d::a10:0:0:9195"}}},
 				},
 			},
 			want: []bool{
@@ -58,14 +58,14 @@ func TestIPFilter(t *testing.T) {
 		{
 			name: "destination ip",
 			args: args{
-				f: []*pb.FlowFilter{
+				f: []*flowpb.FlowFilter{
 					{DestinationIp: []string{"1.1.1.1", "f00d::a10:0:0:9195"}},
 				},
 				ev: []*v1.Event{
-					{Event: &pb.Flow{IP: &pb.IP{Source: "1.1.1.1", Destination: "2.2.2.2"}}},
-					{Event: &pb.Flow{IP: &pb.IP{Source: "2.2.2.2", Destination: "1.1.1.1"}}},
-					{Event: &pb.Flow{IP: &pb.IP{Source: "f00d::a10:0:0:9195", Destination: "ff02::1:ff00:b3e5"}}},
-					{Event: &pb.Flow{IP: &pb.IP{Source: "ff02::1:ff00:b3e5", Destination: "f00d::a10:0:0:9195"}}},
+					{Event: &flowpb.Flow{IP: &flowpb.IP{Source: "1.1.1.1", Destination: "2.2.2.2"}}},
+					{Event: &flowpb.Flow{IP: &flowpb.IP{Source: "2.2.2.2", Destination: "1.1.1.1"}}},
+					{Event: &flowpb.Flow{IP: &flowpb.IP{Source: "f00d::a10:0:0:9195", Destination: "ff02::1:ff00:b3e5"}}},
+					{Event: &flowpb.Flow{IP: &flowpb.IP{Source: "ff02::1:ff00:b3e5", Destination: "f00d::a10:0:0:9195"}}},
 				},
 			},
 			want: []bool{
@@ -78,17 +78,17 @@ func TestIPFilter(t *testing.T) {
 		{
 			name: "source and destination ip",
 			args: args{
-				f: []*pb.FlowFilter{
+				f: []*flowpb.FlowFilter{
 					{
 						SourceIp:      []string{"1.1.1.1", "f00d::a10:0:0:9195"},
 						DestinationIp: []string{"2.2.2.2", "ff02::1:ff00:b3e5"},
 					},
 				},
 				ev: []*v1.Event{
-					{Event: &pb.Flow{IP: &pb.IP{Source: "1.1.1.1", Destination: "2.2.2.2"}}},
-					{Event: &pb.Flow{IP: &pb.IP{Source: "2.2.2.2", Destination: "1.1.1.1"}}},
-					{Event: &pb.Flow{IP: &pb.IP{Source: "f00d::a10:0:0:9195", Destination: "ff02::1:ff00:b3e5"}}},
-					{Event: &pb.Flow{IP: &pb.IP{Source: "ff02::1:ff00:b3e5", Destination: "f00d::a10:0:0:9195"}}},
+					{Event: &flowpb.Flow{IP: &flowpb.IP{Source: "1.1.1.1", Destination: "2.2.2.2"}}},
+					{Event: &flowpb.Flow{IP: &flowpb.IP{Source: "2.2.2.2", Destination: "1.1.1.1"}}},
+					{Event: &flowpb.Flow{IP: &flowpb.IP{Source: "f00d::a10:0:0:9195", Destination: "ff02::1:ff00:b3e5"}}},
+					{Event: &flowpb.Flow{IP: &flowpb.IP{Source: "ff02::1:ff00:b3e5", Destination: "f00d::a10:0:0:9195"}}},
 				},
 			},
 			want: []bool{
@@ -101,15 +101,15 @@ func TestIPFilter(t *testing.T) {
 		{
 			name: "source or destination ip",
 			args: args{
-				f: []*pb.FlowFilter{
+				f: []*flowpb.FlowFilter{
 					{SourceIp: []string{"1.1.1.1"}},
 					{DestinationIp: []string{"2.2.2.2"}},
 				},
 				ev: []*v1.Event{
-					{Event: &pb.Flow{IP: &pb.IP{Source: "1.1.1.1", Destination: "2.2.2.2"}}},
-					{Event: &pb.Flow{IP: &pb.IP{Source: "2.2.2.2", Destination: "1.1.1.1"}}},
-					{Event: &pb.Flow{IP: &pb.IP{Source: "1.1.1.1", Destination: "1.1.1.1"}}},
-					{Event: &pb.Flow{IP: &pb.IP{Source: "2.2.2.2", Destination: "2.2.2.2"}}},
+					{Event: &flowpb.Flow{IP: &flowpb.IP{Source: "1.1.1.1", Destination: "2.2.2.2"}}},
+					{Event: &flowpb.Flow{IP: &flowpb.IP{Source: "2.2.2.2", Destination: "1.1.1.1"}}},
+					{Event: &flowpb.Flow{IP: &flowpb.IP{Source: "1.1.1.1", Destination: "1.1.1.1"}}},
+					{Event: &flowpb.Flow{IP: &flowpb.IP{Source: "2.2.2.2", Destination: "2.2.2.2"}}},
 				},
 			},
 			want: []bool{
@@ -122,14 +122,14 @@ func TestIPFilter(t *testing.T) {
 		{
 			name: "invalid data",
 			args: args{
-				f: []*pb.FlowFilter{
+				f: []*flowpb.FlowFilter{
 					{SourceIp: []string{"1.1.1.1"}},
 				},
 				ev: []*v1.Event{
 					nil,
 					{},
-					{Event: &pb.Flow{}},
-					{Event: &pb.Flow{IP: &pb.IP{Source: ""}}},
+					{Event: &flowpb.Flow{}},
+					{Event: &flowpb.Flow{IP: &flowpb.IP{Source: ""}}},
 				},
 			},
 			want: []bool{
@@ -142,7 +142,7 @@ func TestIPFilter(t *testing.T) {
 		{
 			name: "invalid source ip filter",
 			args: args{
-				f: []*pb.FlowFilter{
+				f: []*flowpb.FlowFilter{
 					{SourceIp: []string{"320.320.320.320"}},
 				},
 			},
@@ -151,7 +151,7 @@ func TestIPFilter(t *testing.T) {
 		{
 			name: "invalid destination ip filter",
 			args: args{
-				f: []*pb.FlowFilter{
+				f: []*flowpb.FlowFilter{
 					{DestinationIp: []string{""}},
 				},
 			},
diff --git a/pkg/hubble/filters/k8s_test.go b/pkg/hubble/filters/k8s_test.go
index 0dad68fc3c..d4d726c1dc 100644
--- a/pkg/hubble/filters/k8s_test.go
+++ b/pkg/hubble/filters/k8s_test.go
@@ -20,13 +20,13 @@ import (
 	"context"
 	"testing"
 
-	pb "github.com/cilium/cilium/api/v1/flow"
+	flowpb "github.com/cilium/cilium/api/v1/flow"
 	v1 "github.com/cilium/cilium/pkg/hubble/api/v1"
 )
 
 func TestPodFilter(t *testing.T) {
 	type args struct {
-		f  []*pb.FlowFilter
+		f  []*flowpb.FlowFilter
 		ev []*v1.Event
 	}
 	tests := []struct {
@@ -38,14 +38,14 @@ func TestPodFilter(t *testing.T) {
 		{
 			name: "source pod",
 			args: args{
-				f: []*pb.FlowFilter{
+				f: []*flowpb.FlowFilter{
 					{SourcePod: []string{"xwing", "default/tiefighter"}},
 				},
 				ev: []*v1.Event{
-					{Event: &pb.Flow{Source: &pb.Endpoint{Namespace: "default", PodName: "xwing"}}},
-					{Event: &pb.Flow{Source: &pb.Endpoint{Namespace: "default", PodName: "tiefighter"}}},
-					{Event: &pb.Flow{Source: &pb.Endpoint{Namespace: "kube-system", PodName: "xwing"}}},
-					{Event: &pb.Flow{Destination: &pb.Endpoint{Namespace: "default", PodName: "xwing"}}},
+					{Event: &flowpb.Flow{Source: &flowpb.Endpoint{Namespace: "default", PodName: "xwing"}}},
+					{Event: &flowpb.Flow{Source: &flowpb.Endpoint{Namespace: "default", PodName: "tiefighter"}}},
+					{Event: &flowpb.Flow{Source: &flowpb.Endpoint{Namespace: "kube-system", PodName: "xwing"}}},
+					{Event: &flowpb.Flow{Destination: &flowpb.Endpoint{Namespace: "default", PodName: "xwing"}}},
 				},
 			},
 			want: []bool{
@@ -58,14 +58,14 @@ func TestPodFilter(t *testing.T) {
 		{
 			name: "destination pod",
 			args: args{
-				f: []*pb.FlowFilter{
+				f: []*flowpb.FlowFilter{
 					{DestinationPod: []string{"xwing", "default/tiefighter"}},
 				},
 				ev: []*v1.Event{
-					{Event: &pb.Flow{Destination: &pb.Endpoint{Namespace: "default", PodName: "xwing"}}},
-					{Event: &pb.Flow{Destination: &pb.Endpoint{Namespace: "default", PodName: "tiefighter"}}},
-					{Event: &pb.Flow{Destination: &pb.Endpoint{Namespace: "kube-system", PodName: "xwing"}}},
-					{Event: &pb.Flow{Source: &pb.Endpoint{Namespace: "default", PodName: "xwing"}}},
+					{Event: &flowpb.Flow{Destination: &flowpb.Endpoint{Namespace: "default", PodName: "xwing"}}},
+					{Event: &flowpb.Flow{Destination: &flowpb.Endpoint{Namespace: "default", PodName: "tiefighter"}}},
+					{Event: &flowpb.Flow{Destination: &flowpb.Endpoint{Namespace: "kube-system", PodName: "xwing"}}},
+					{Event: &flowpb.Flow{Source: &flowpb.Endpoint{Namespace: "default", PodName: "xwing"}}},
 				},
 			},
 			want: []bool{
@@ -78,28 +78,28 @@ func TestPodFilter(t *testing.T) {
 		{
 			name: "source and destination pod",
 			args: args{
-				f: []*pb.FlowFilter{
+				f: []*flowpb.FlowFilter{
 					{
 						SourcePod:      []string{"xwing", "tiefighter"},
 						DestinationPod: []string{"deathstar"},
 					},
 				},
 				ev: []*v1.Event{
-					{Event: &pb.Flow{
-						Source:      &pb.Endpoint{Namespace: "default", PodName: "xwing"},
-						Destination: &pb.Endpoint{Namespace: "default", PodName: "deathstar"},
+					{Event: &flowpb.Flow{
+						Source:      &flowpb.Endpoint{Namespace: "default", PodName: "xwing"},
+						Destination: &flowpb.Endpoint{Namespace: "default", PodName: "deathstar"},
 					}},
-					{Event: &pb.Flow{
-						Source:      &pb.Endpoint{Namespace: "default", PodName: "tiefighter"},
-						Destination: &pb.Endpoint{Namespace: "default", PodName: "deathstar"},
+					{Event: &flowpb.Flow{
+						Source:      &flowpb.Endpoint{Namespace: "default", PodName: "tiefighter"},
+						Destination: &flowpb.Endpoint{Namespace: "default", PodName: "deathstar"},
 					}},
-					{Event: &pb.Flow{
-						Source:      &pb.Endpoint{Namespace: "default", PodName: "deathstar"},
-						Destination: &pb.Endpoint{Namespace: "default", PodName: "xwing"},
+					{Event: &flowpb.Flow{
+						Source:      &flowpb.Endpoint{Namespace: "default", PodName: "deathstar"},
+						Destination: &flowpb.Endpoint{Namespace: "default", PodName: "xwing"},
 					}},
-					{Event: &pb.Flow{
-						Source:      &pb.Endpoint{Namespace: "default", PodName: "tiefighter"},
-						Destination: &pb.Endpoint{Namespace: "default", PodName: "xwing"},
+					{Event: &flowpb.Flow{
+						Source:      &flowpb.Endpoint{Namespace: "default", PodName: "tiefighter"},
+						Destination: &flowpb.Endpoint{Namespace: "default", PodName: "xwing"},
 					}},
 				},
 			},
@@ -113,26 +113,26 @@ func TestPodFilter(t *testing.T) {
 		{
 			name: "source or destination pod",
 			args: args{
-				f: []*pb.FlowFilter{
+				f: []*flowpb.FlowFilter{
 					{SourcePod: []string{"xwing"}},
 					{DestinationPod: []string{"deathstar"}},
 				},
 				ev: []*v1.Event{
-					{Event: &pb.Flow{
-						Source:      &pb.Endpoint{Namespace: "default", PodName: "xwing"},
-						Destination: &pb.Endpoint{Namespace: "default", PodName: "deathstar"},
+					{Event: &flowpb.Flow{
+						Source:      &flowpb.Endpoint{Namespace: "default", PodName: "xwing"},
+						Destination: &flowpb.Endpoint{Namespace: "default", PodName: "deathstar"},
 					}},
-					{Event: &pb.Flow{
-						Source:      &pb.Endpoint{Namespace: "default", PodName: "tiefighter"},
-						Destination: &pb.Endpoint{Namespace: "default", PodName: "deathstar"},
+					{Event: &flowpb.Flow{
+						Source:      &flowpb.Endpoint{Namespace: "default", PodName: "tiefighter"},
+						Destination: &flowpb.Endpoint{Namespace: "default", PodName: "deathstar"},
 					}},
-					{Event: &pb.Flow{
-						Source:      &pb.Endpoint{Namespace: "default", PodName: "deathstar"},
-						Destination: &pb.Endpoint{Namespace: "default", PodName: "xwing"},
+					{Event: &flowpb.Flow{
+						Source:      &flowpb.Endpoint{Namespace: "default", PodName: "deathstar"},
+						Destination: &flowpb.Endpoint{Namespace: "default", PodName: "xwing"},
 					}},
-					{Event: &pb.Flow{
-						Source:      &pb.Endpoint{Namespace: "default", PodName: "tiefighter"},
-						Destination: &pb.Endpoint{Namespace: "default", PodName: "xwing"},
+					{Event: &flowpb.Flow{
+						Source:      &flowpb.Endpoint{Namespace: "default", PodName: "tiefighter"},
+						Destination: &flowpb.Endpoint{Namespace: "default", PodName: "xwing"},
 					}},
 				},
 			},
@@ -146,26 +146,26 @@ func TestPodFilter(t *testing.T) {
 		{
 			name: "namespace filter",
 			args: args{
-				f: []*pb.FlowFilter{
+				f: []*flowpb.FlowFilter{
 					{SourcePod: []string{"kube-system/"}},
 					{DestinationPod: []string{"kube-system/"}},
 				},
 				ev: []*v1.Event{
-					{Event: &pb.Flow{
-						Source:      &pb.Endpoint{Namespace: "kube-system", PodName: "coredns"},
-						Destination: &pb.Endpoint{Namespace: "kube-system", PodName: "kube-proxy"},
+					{Event: &flowpb.Flow{
+						Source:      &flowpb.Endpoint{Namespace: "kube-system", PodName: "coredns"},
+						Destination: &flowpb.Endpoint{Namespace: "kube-system", PodName: "kube-proxy"},
 					}},
-					{Event: &pb.Flow{
-						Source:      &pb.Endpoint{Namespace: "default", PodName: "tiefighter"},
-						Destination: &pb.Endpoint{Namespace: "kube-system", PodName: "coredns"},
+					{Event: &flowpb.Flow{
+						Source:      &flowpb.Endpoint{Namespace: "default", PodName: "tiefighter"},
+						Destination: &flowpb.Endpoint{Namespace: "kube-system", PodName: "coredns"},
 					}},
-					{Event: &pb.Flow{
-						Source:      &pb.Endpoint{Namespace: "kube-system", PodName: "coredns"},
-						Destination: &pb.Endpoint{Namespace: "default", PodName: "xwing"},
+					{Event: &flowpb.Flow{
+						Source:      &flowpb.Endpoint{Namespace: "kube-system", PodName: "coredns"},
+						Destination: &flowpb.Endpoint{Namespace: "default", PodName: "xwing"},
 					}},
-					{Event: &pb.Flow{
-						Source:      &pb.Endpoint{Namespace: "default", PodName: "tiefighter"},
-						Destination: &pb.Endpoint{Namespace: "default", PodName: "xwing"},
+					{Event: &flowpb.Flow{
+						Source:      &flowpb.Endpoint{Namespace: "default", PodName: "tiefighter"},
+						Destination: &flowpb.Endpoint{Namespace: "default", PodName: "xwing"},
 					}},
 				},
 			},
@@ -179,24 +179,24 @@ func TestPodFilter(t *testing.T) {
 		{
 			name: "prefix filter",
 			args: args{
-				f: []*pb.FlowFilter{
+				f: []*flowpb.FlowFilter{
 					{SourcePod: []string{"xwing", "kube-system/coredns-"}},
 				},
 				ev: []*v1.Event{
-					{Event: &pb.Flow{
-						Source: &pb.Endpoint{Namespace: "default", PodName: "xwing"},
+					{Event: &flowpb.Flow{
+						Source: &flowpb.Endpoint{Namespace: "default", PodName: "xwing"},
 					}},
-					{Event: &pb.Flow{
-						Source: &pb.Endpoint{Namespace: "default", PodName: "xwing-t-65b"},
+					{Event: &flowpb.Flow{
+						Source: &flowpb.Endpoint{Namespace: "default", PodName: "xwing-t-65b"},
 					}},
-					{Event: &pb.Flow{
-						Source: &pb.Endpoint{Namespace: "kube-system", PodName: "coredns-12345"},
+					{Event: &flowpb.Flow{
+						Source: &flowpb.Endpoint{Namespace: "kube-system", PodName: "coredns-12345"},
 					}},
-					{Event: &pb.Flow{
-						Source: &pb.Endpoint{Namespace: "kube-system", PodName: "-coredns-12345"},
+					{Event: &flowpb.Flow{
+						Source: &flowpb.Endpoint{Namespace: "kube-system", PodName: "-coredns-12345"},
 					}},
-					{Event: &pb.Flow{
-						Source: &pb.Endpoint{Namespace: "default", PodName: "tiefighter"},
+					{Event: &flowpb.Flow{
+						Source: &flowpb.Endpoint{Namespace: "default", PodName: "tiefighter"},
 					}},
 				},
 			},
@@ -211,14 +211,14 @@ func TestPodFilter(t *testing.T) {
 		{
 			name: "invalid data",
 			args: args{
-				f: []*pb.FlowFilter{
+				f: []*flowpb.FlowFilter{
 					{SourcePod: []string{"xwing"}},
 				},
 				ev: []*v1.Event{
 					nil,
 					{},
-					{Event: &pb.Flow{}},
-					{Event: &pb.Flow{Source: &pb.Endpoint{Namespace: "", PodName: "xwing"}}},
+					{Event: &flowpb.Flow{}},
+					{Event: &flowpb.Flow{Source: &flowpb.Endpoint{Namespace: "", PodName: "xwing"}}},
 				},
 			},
 			want: []bool{
@@ -231,7 +231,7 @@ func TestPodFilter(t *testing.T) {
 		{
 			name: "invalid source pod filter",
 			args: args{
-				f: []*pb.FlowFilter{
+				f: []*flowpb.FlowFilter{
 					{SourcePod: []string{""}},
 				},
 			},
@@ -240,7 +240,7 @@ func TestPodFilter(t *testing.T) {
 		{
 			name: "invalid destination pod filter",
 			args: args{
-				f: []*pb.FlowFilter{
+				f: []*flowpb.FlowFilter{
 					{DestinationPod: []string{""}},
 				},
 			},
@@ -265,7 +265,7 @@ func TestPodFilter(t *testing.T) {
 
 func TestServiceFilter(t *testing.T) {
 	type args struct {
-		f  []*pb.FlowFilter
+		f  []*flowpb.FlowFilter
 		ev []*v1.Event
 	}
 	tests := []struct {
@@ -277,14 +277,14 @@ func TestServiceFilter(t *testing.T) {
 		{
 			name: "source service",
 			args: args{
-				f: []*pb.FlowFilter{
+				f: []*flowpb.FlowFilter{
 					{SourceService: []string{"deathstar", "kube-system/kube-dns"}},
 				},
 				ev: []*v1.Event{
-					{Event: &pb.Flow{SourceService: &pb.Service{Namespace: "default", Name: "xwing"}}},
-					{Event: &pb.Flow{SourceService: &pb.Service{Namespace: "default", Name: "deathstar"}}},
-					{Event: &pb.Flow{SourceService: &pb.Service{Namespace: "kube-system", Name: "kube-dns"}}},
-					{Event: &pb.Flow{DestinationService: &pb.Service{Namespace: "kube-system", Name: "deathstar"}}},
+					{Event: &flowpb.Flow{SourceService: &flowpb.Service{Namespace: "default", Name: "xwing"}}},
+					{Event: &flowpb.Flow{SourceService: &flowpb.Service{Namespace: "default", Name: "deathstar"}}},
+					{Event: &flowpb.Flow{SourceService: &flowpb.Service{Namespace: "kube-system", Name: "kube-dns"}}},
+					{Event: &flowpb.Flow{DestinationService: &flowpb.Service{Namespace: "kube-system", Name: "deathstar"}}},
 				},
 			},
 			want: []bool{
@@ -297,14 +297,14 @@ func TestServiceFilter(t *testing.T) {
 		{
 			name: "destination service",
 			args: args{
-				f: []*pb.FlowFilter{
+				f: []*flowpb.FlowFilter{
 					{DestinationService: []string{"default/", "kube-system/kube-"}},
 				},
 				ev: []*v1.Event{
-					{Event: &pb.Flow{DestinationService: &pb.Service{Namespace: "default", Name: "xwing"}}},
-					{Event: &pb.Flow{DestinationService: &pb.Service{Namespace: "default", Name: "deathstar"}}},
-					{Event: &pb.Flow{DestinationService: &pb.Service{Namespace: "kube-system", Name: "kube-dns"}}},
-					{Event: &pb.Flow{SourceService: &pb.Service{Namespace: "kube-system", Name: "deathstar"}}},
+					{Event: &flowpb.Flow{DestinationService: &flowpb.Service{Namespace: "default", Name: "xwing"}}},
+					{Event: &flowpb.Flow{DestinationService: &flowpb.Service{Namespace: "default", Name: "deathstar"}}},
+					{Event: &flowpb.Flow{DestinationService: &flowpb.Service{Namespace: "kube-system", Name: "kube-dns"}}},
+					{Event: &flowpb.Flow{SourceService: &flowpb.Service{Namespace: "kube-system", Name: "deathstar"}}},
 				},
 			},
 			want: []bool{
diff --git a/pkg/hubble/filters/labels_test.go b/pkg/hubble/filters/labels_test.go
index 716ffaf03e..1c244d48b8 100644
--- a/pkg/hubble/filters/labels_test.go
+++ b/pkg/hubble/filters/labels_test.go
@@ -21,13 +21,13 @@ import (
 	"reflect"
 	"testing"
 
-	pb "github.com/cilium/cilium/api/v1/flow"
+	flowpb "github.com/cilium/cilium/api/v1/flow"
 	v1 "github.com/cilium/cilium/pkg/hubble/api/v1"
 )
 
 func TestLabelSelectorFilter(t *testing.T) {
 	type args struct {
-		f  []*pb.FlowFilter
+		f  []*flowpb.FlowFilter
 		ev []*v1.Event
 	}
 	tests := []struct {
@@ -39,32 +39,32 @@ func TestLabelSelectorFilter(t *testing.T) {
 		{
 			name: "label filter without value",
 			args: args{
-				f: []*pb.FlowFilter{{SourceLabel: []string{"label1", "label2"}}},
+				f: []*flowpb.FlowFilter{{SourceLabel: []string{"label1", "label2"}}},
 				ev: []*v1.Event{
 					{
-						Event: &pb.Flow{
-							Source: &pb.Endpoint{
+						Event: &flowpb.Flow{
+							Source: &flowpb.Endpoint{
 								Labels: []string{"label1"},
 							},
 						},
 					},
 					{
-						Event: &pb.Flow{
-							Source: &pb.Endpoint{
+						Event: &flowpb.Flow{
+							Source: &flowpb.Endpoint{
 								Labels: []string{"label1=val1"},
 							},
 						},
 					},
 					{
-						Event: &pb.Flow{
-							Source: &pb.Endpoint{
+						Event: &flowpb.Flow{
+							Source: &flowpb.Endpoint{
 								Labels: []string{"label2", "label3", "label4=val4"},
 							},
 						},
 					},
 					{
-						Event: &pb.Flow{
-							Source: &pb.Endpoint{
+						Event: &flowpb.Flow{
+							Source: &flowpb.Endpoint{
 								Labels: []string{"label3"},
 							},
 						},
@@ -81,60 +81,60 @@ func TestLabelSelectorFilter(t *testing.T) {
 		{
 			name: "label filter with value",
 			args: args{
-				f: []*pb.FlowFilter{{SourceLabel: []string{"label1=val1", "label2=val2"}}},
+				f: []*flowpb.FlowFilter{{SourceLabel: []string{"label1=val1", "label2=val2"}}},
 				ev: []*v1.Event{
 					{
-						Event: &pb.Flow{
-							Source: &pb.Endpoint{
+						Event: &flowpb.Flow{
+							Source: &flowpb.Endpoint{
 								Labels: []string{"label1"},
 							},
 						},
 					},
 					{
-						Event: &pb.Flow{
-							Source: &pb.Endpoint{
+						Event: &flowpb.Flow{
+							Source: &flowpb.Endpoint{
 								Labels: []string{"label1=val1"},
 							},
 						},
 					},
 					{
-						Event: &pb.Flow{
-							Source: &pb.Endpoint{
+						Event: &flowpb.Flow{
+							Source: &flowpb.Endpoint{
 								Labels: []string{"label1=val2", "label2=val1", "label3"},
 							},
 						},
 					},
 					{
-						Event: &pb.Flow{
-							Source: &pb.Endpoint{
+						Event: &flowpb.Flow{
+							Source: &flowpb.Endpoint{
 								Labels: []string{"label2=val2", "label3"},
 							},
 						},
 					},
 					{
-						Event: &pb.Flow{
-							Source: &pb.Endpoint{
+						Event: &flowpb.Flow{
+							Source: &flowpb.Endpoint{
 								Labels: []string{"label3=val1"},
 							},
 						},
 					},
 					{
-						Event: &pb.Flow{
-							Source: &pb.Endpoint{
+						Event: &flowpb.Flow{
+							Source: &flowpb.Endpoint{
 								Labels: []string{""},
 							},
 						},
 					},
 					{
-						Event: &pb.Flow{
-							Source: &pb.Endpoint{
+						Event: &flowpb.Flow{
+							Source: &flowpb.Endpoint{
 								Labels: nil,
 							},
 						},
 					},
 					{
-						Event: &pb.Flow{
-							Source: &pb.Endpoint{
+						Event: &flowpb.Flow{
+							Source: &flowpb.Endpoint{
 								Labels: []string{"label1=val1=toomuch"},
 							},
 						},
@@ -155,39 +155,39 @@ func TestLabelSelectorFilter(t *testing.T) {
 		{
 			name: "complex label label filter",
 			args: args{
-				f: []*pb.FlowFilter{{SourceLabel: []string{"label1 in (val1, val2), label3 notin ()"}}},
+				f: []*flowpb.FlowFilter{{SourceLabel: []string{"label1 in (val1, val2), label3 notin ()"}}},
 				ev: []*v1.Event{
 					{
-						Event: &pb.Flow{
-							Source: &pb.Endpoint{
+						Event: &flowpb.Flow{
+							Source: &flowpb.Endpoint{
 								Labels: []string{"label1"},
 							},
 						},
 					},
 					{
-						Event: &pb.Flow{
-							Source: &pb.Endpoint{
+						Event: &flowpb.Flow{
+							Source: &flowpb.Endpoint{
 								Labels: []string{"label1=val1"},
 							},
 						},
 					},
 					{
-						Event: &pb.Flow{
-							Source: &pb.Endpoint{
+						Event: &flowpb.Flow{
+							Source: &flowpb.Endpoint{
 								Labels: []string{"label1=val2", "label2=val1", "label3=val3"},
 							},
 						},
 					},
 					{
-						Event: &pb.Flow{
-							Source: &pb.Endpoint{
+						Event: &flowpb.Flow{
+							Source: &flowpb.Endpoint{
 								Labels: []string{"label2=val2", "label3"},
 							},
 						},
 					},
 					{
-						Event: &pb.Flow{
-							Source: &pb.Endpoint{
+						Event: &flowpb.Flow{
+							Source: &flowpb.Endpoint{
 								Labels: []string{"label1=val1", "label3=val3"},
 							},
 						},
@@ -205,7 +205,7 @@ func TestLabelSelectorFilter(t *testing.T) {
 		{
 			name: "source and destination label filter",
 			args: args{
-				f: []*pb.FlowFilter{
+				f: []*flowpb.FlowFilter{
 					{
 						SourceLabel:      []string{"src1, src2=val2"},
 						DestinationLabel: []string{"dst1, dst2=val2"},
@@ -213,45 +213,45 @@ func TestLabelSelectorFilter(t *testing.T) {
 				},
 				ev: []*v1.Event{
 					{
-						Event: &pb.Flow{
-							Source: &pb.Endpoint{
+						Event: &flowpb.Flow{
+							Source: &flowpb.Endpoint{
 								Labels: []string{"src1", "src2=val2"},
 							},
-							Destination: &pb.Endpoint{
+							Destination: &flowpb.Endpoint{
 								Labels: []string{"dst1", "dst2=val2"},
 							},
 						},
 					},
 					{
-						Event: &pb.Flow{
-							Source: &pb.Endpoint{
+						Event: &flowpb.Flow{
+							Source: &flowpb.Endpoint{
 								Labels: []string{"label1=val1"},
 							},
 						},
 					},
 					{
-						Event: &pb.Flow{
-							Destination: &pb.Endpoint{
+						Event: &flowpb.Flow{
+							Destination: &flowpb.Endpoint{
 								Labels: []string{"dst1", "dst2=val2"},
 							},
 						},
 					},
 					{
-						Event: &pb.Flow{
-							Source: &pb.Endpoint{
+						Event: &flowpb.Flow{
+							Source: &flowpb.Endpoint{
 								Labels: []string{"dst1", "dst2=val2"},
 							},
-							Destination: &pb.Endpoint{
+							Destination: &flowpb.Endpoint{
 								Labels: []string{"src1", "src2=val2"},
 							},
 						},
 					},
 					{
-						Event: &pb.Flow{
-							Source: &pb.Endpoint{
+						Event: &flowpb.Flow{
+							Source: &flowpb.Endpoint{
 								Labels: []string{"src1"},
 							},
-							Destination: &pb.Endpoint{
+							Destination: &flowpb.Endpoint{
 								Labels: []string{"dst1"},
 							},
 						},
@@ -269,29 +269,29 @@ func TestLabelSelectorFilter(t *testing.T) {
 		{
 			name: "matchall filter",
 			args: args{
-				f: []*pb.FlowFilter{
+				f: []*flowpb.FlowFilter{
 					{
 						SourceLabel: []string{""},
 					},
 				},
 				ev: []*v1.Event{
 					{
-						Event: &pb.Flow{
-							Source: &pb.Endpoint{
+						Event: &flowpb.Flow{
+							Source: &flowpb.Endpoint{
 								Labels: []string{"src1", "src2=val2"},
 							},
 						},
 					},
 					{
-						Event: &pb.Flow{
-							Source: &pb.Endpoint{
+						Event: &flowpb.Flow{
+							Source: &flowpb.Endpoint{
 								Labels: nil,
 							},
 						},
 					},
 					{
-						Event: &pb.Flow{
-							Source: &pb.Endpoint{
+						Event: &flowpb.Flow{
+							Source: &flowpb.Endpoint{
 								Labels: []string{""},
 							},
 						},
@@ -307,36 +307,36 @@ func TestLabelSelectorFilter(t *testing.T) {
 		{
 			name: "cilium fixed prefix filters",
 			args: args{
-				f: []*pb.FlowFilter{
+				f: []*flowpb.FlowFilter{
 					{
 						SourceLabel: []string{"k8s:app=bar", "foo", "reserved:host"},
 					},
 				},
 				ev: []*v1.Event{
 					{
-						Event: &pb.Flow{
-							Source: &pb.Endpoint{
+						Event: &flowpb.Flow{
+							Source: &flowpb.Endpoint{
 								Labels: []string{"k8s:app=bar"},
 							},
 						},
 					},
 					{
-						Event: &pb.Flow{
-							Source: &pb.Endpoint{
+						Event: &flowpb.Flow{
+							Source: &flowpb.Endpoint{
 								Labels: []string{"k8s:foo=baz"},
 							},
 						},
 					},
 					{
-						Event: &pb.Flow{
-							Source: &pb.Endpoint{
+						Event: &flowpb.Flow{
+							Source: &flowpb.Endpoint{
 								Labels: []string{"k8s.app=bar"},
 							},
 						},
 					},
 					{
-						Event: &pb.Flow{
-							Source: &pb.Endpoint{
+						Event: &flowpb.Flow{
+							Source: &flowpb.Endpoint{
 								Labels: []string{"container:foo=bar", "reserved:host"},
 							},
 						},
@@ -353,29 +353,29 @@ func TestLabelSelectorFilter(t *testing.T) {
 		{
 			name: "cilium any prefix filters",
 			args: args{
-				f: []*pb.FlowFilter{
+				f: []*flowpb.FlowFilter{
 					{
 						SourceLabel: []string{"any:key"},
 					},
 				},
 				ev: []*v1.Event{
 					{
-						Event: &pb.Flow{
-							Source: &pb.Endpoint{
+						Event: &flowpb.Flow{
+							Source: &flowpb.Endpoint{
 								Labels: []string{"key"},
 							},
 						},
 					},
 					{
-						Event: &pb.Flow{
-							Source: &pb.Endpoint{
+						Event: &flowpb.Flow{
+							Source: &flowpb.Endpoint{
 								Labels: []string{"reserved:key"},
 							},
 						},
 					},
 					{
-						Event: &pb.Flow{
-							Source: &pb.Endpoint{
+						Event: &flowpb.Flow{
+							Source: &flowpb.Endpoint{
 								Labels: []string{"any.key"},
 							},
 						},
@@ -391,7 +391,7 @@ func TestLabelSelectorFilter(t *testing.T) {
 		{
 			name: "invalid source filter",
 			args: args{
-				f: []*pb.FlowFilter{
+				f: []*flowpb.FlowFilter{
 					{
 						SourceLabel: []string{"()"},
 					},
@@ -402,7 +402,7 @@ func TestLabelSelectorFilter(t *testing.T) {
 		{
 			name: "invalid destination filter",
 			args: args{
-				f: []*pb.FlowFilter{
+				f: []*flowpb.FlowFilter{
 					{
 						DestinationLabel: []string{"="},
 					},
diff --git a/pkg/hubble/filters/port_test.go b/pkg/hubble/filters/port_test.go
index 987a9101d9..fe115d68e3 100644
--- a/pkg/hubble/filters/port_test.go
+++ b/pkg/hubble/filters/port_test.go
@@ -20,13 +20,13 @@ import (
 	"context"
 	"testing"
 
-	pb "github.com/cilium/cilium/api/v1/flow"
+	flowpb "github.com/cilium/cilium/api/v1/flow"
 	v1 "github.com/cilium/cilium/pkg/hubble/api/v1"
 )
 
 func TestPortFilter(t *testing.T) {
 	type args struct {
-		f  []*pb.FlowFilter
+		f  []*flowpb.FlowFilter
 		ev *v1.Event
 	}
 	tests := []struct {
@@ -38,12 +38,12 @@ func TestPortFilter(t *testing.T) {
 		{
 			name: "udp",
 			args: args{
-				f: []*pb.FlowFilter{{
+				f: []*flowpb.FlowFilter{{
 					SourcePort:      []string{"12345"},
 					DestinationPort: []string{"53"},
 				}},
-				ev: &v1.Event{Event: &pb.Flow{
-					L4: &pb.Layer4{Protocol: &pb.Layer4_UDP{UDP: &pb.UDP{
+				ev: &v1.Event{Event: &flowpb.Flow{
+					L4: &flowpb.Layer4{Protocol: &flowpb.Layer4_UDP{UDP: &flowpb.UDP{
 						SourcePort:      12345,
 						DestinationPort: 53,
 					}}},
@@ -54,12 +54,12 @@ func TestPortFilter(t *testing.T) {
 		{
 			name: "tcp",
 			args: args{
-				f: []*pb.FlowFilter{{
+				f: []*flowpb.FlowFilter{{
 					SourcePort:      []string{"32320"},
 					DestinationPort: []string{"80"},
 				}},
-				ev: &v1.Event{Event: &pb.Flow{
-					L4: &pb.Layer4{Protocol: &pb.Layer4_TCP{TCP: &pb.TCP{
+				ev: &v1.Event{Event: &flowpb.Flow{
+					L4: &flowpb.Layer4{Protocol: &flowpb.Layer4_TCP{TCP: &flowpb.TCP{
 						SourcePort:      32320,
 						DestinationPort: 80,
 					}}},
@@ -70,11 +70,11 @@ func TestPortFilter(t *testing.T) {
 		{
 			name: "wrong direction",
 			args: args{
-				f: []*pb.FlowFilter{{
+				f: []*flowpb.FlowFilter{{
 					DestinationPort: []string{"80"},
 				}},
-				ev: &v1.Event{Event: &pb.Flow{
-					L4: &pb.Layer4{Protocol: &pb.Layer4_TCP{TCP: &pb.TCP{
+				ev: &v1.Event{Event: &flowpb.Flow{
+					L4: &flowpb.Layer4{Protocol: &flowpb.Layer4_TCP{TCP: &flowpb.TCP{
 						SourcePort:      80,
 						DestinationPort: 32320,
 					}}},
@@ -85,11 +85,11 @@ func TestPortFilter(t *testing.T) {
 		{
 			name: "no port",
 			args: args{
-				f: []*pb.FlowFilter{{
+				f: []*flowpb.FlowFilter{{
 					DestinationPort: []string{"0"},
 				}},
-				ev: &v1.Event{Event: &pb.Flow{
-					L4: &pb.Layer4{Protocol: &pb.Layer4_ICMPv4{ICMPv4: &pb.ICMPv4{}}},
+				ev: &v1.Event{Event: &flowpb.Flow{
+					L4: &flowpb.Layer4{Protocol: &flowpb.Layer4_ICMPv4{ICMPv4: &flowpb.ICMPv4{}}},
 				}},
 			},
 			want: false,
@@ -97,7 +97,7 @@ func TestPortFilter(t *testing.T) {
 		{
 			name: "invalid port",
 			args: args{
-				f: []*pb.FlowFilter{{SourcePort: []string{"999999"}}},
+				f: []*flowpb.FlowFilter{{SourcePort: []string{"999999"}}},
 			},
 			wantErr: true,
 		},
diff --git a/pkg/hubble/filters/protocol_test.go b/pkg/hubble/filters/protocol_test.go
index 0ada5b3bf6..356d8e122e 100644
--- a/pkg/hubble/filters/protocol_test.go
+++ b/pkg/hubble/filters/protocol_test.go
@@ -20,13 +20,13 @@ import (
 	"context"
 	"testing"
 
-	pb "github.com/cilium/cilium/api/v1/flow"
+	flowpb "github.com/cilium/cilium/api/v1/flow"
 	v1 "github.com/cilium/cilium/pkg/hubble/api/v1"
 )
 
 func TestFlowProtocolFilter(t *testing.T) {
 	type args struct {
-		f  []*pb.FlowFilter
+		f  []*flowpb.FlowFilter
 		ev *v1.Event
 	}
 	tests := []struct {
@@ -38,9 +38,9 @@ func TestFlowProtocolFilter(t *testing.T) {
 		{
 			name: "udp",
 			args: args{
-				f: []*pb.FlowFilter{{Protocol: []string{"udp"}}},
-				ev: &v1.Event{Event: &pb.Flow{
-					L4: &pb.Layer4{Protocol: &pb.Layer4_UDP{UDP: &pb.UDP{}}},
+				f: []*flowpb.FlowFilter{{Protocol: []string{"udp"}}},
+				ev: &v1.Event{Event: &flowpb.Flow{
+					L4: &flowpb.Layer4{Protocol: &flowpb.Layer4_UDP{UDP: &flowpb.UDP{}}},
 				}},
 			},
 			want: true,
@@ -48,10 +48,10 @@ func TestFlowProtocolFilter(t *testing.T) {
 		{
 			name: "http",
 			args: args{
-				f: []*pb.FlowFilter{{Protocol: []string{"http"}}},
-				ev: &v1.Event{Event: &pb.Flow{
-					L4: &pb.Layer4{Protocol: &pb.Layer4_TCP{TCP: &pb.TCP{}}},
-					L7: &pb.Layer7{Record: &pb.Layer7_Http{Http: &pb.HTTP{}}},
+				f: []*flowpb.FlowFilter{{Protocol: []string{"http"}}},
+				ev: &v1.Event{Event: &flowpb.Flow{
+					L4: &flowpb.Layer4{Protocol: &flowpb.Layer4_TCP{TCP: &flowpb.TCP{}}},
+					L7: &flowpb.Layer7{Record: &flowpb.Layer7_Http{Http: &flowpb.HTTP{}}},
 				}},
 			},
 			want: true,
@@ -59,9 +59,9 @@ func TestFlowProtocolFilter(t *testing.T) {
 		{
 			name: "icmp (v4)",
 			args: args{
-				f: []*pb.FlowFilter{{Protocol: []string{"icmp"}}},
-				ev: &v1.Event{Event: &pb.Flow{
-					L4: &pb.Layer4{Protocol: &pb.Layer4_ICMPv4{ICMPv4: &pb.ICMPv4{}}},
+				f: []*flowpb.FlowFilter{{Protocol: []string{"icmp"}}},
+				ev: &v1.Event{Event: &flowpb.Flow{
+					L4: &flowpb.Layer4{Protocol: &flowpb.Layer4_ICMPv4{ICMPv4: &flowpb.ICMPv4{}}},
 				}},
 			},
 			want: true,
@@ -69,9 +69,9 @@ func TestFlowProtocolFilter(t *testing.T) {
 		{
 			name: "icmp (v6)",
 			args: args{
-				f: []*pb.FlowFilter{{Protocol: []string{"icmp"}}},
-				ev: &v1.Event{Event: &pb.Flow{
-					L4: &pb.Layer4{Protocol: &pb.Layer4_ICMPv6{ICMPv6: &pb.ICMPv6{}}},
+				f: []*flowpb.FlowFilter{{Protocol: []string{"icmp"}}},
+				ev: &v1.Event{Event: &flowpb.Flow{
+					L4: &flowpb.Layer4{Protocol: &flowpb.Layer4_ICMPv6{ICMPv6: &flowpb.ICMPv6{}}},
 				}},
 			},
 			want: true,
@@ -79,9 +79,9 @@ func TestFlowProtocolFilter(t *testing.T) {
 		{
 			name: "multiple protocols",
 			args: args{
-				f: []*pb.FlowFilter{{Protocol: []string{"tcp", "kafka"}}},
-				ev: &v1.Event{Event: &pb.Flow{
-					L4: &pb.Layer4{Protocol: &pb.Layer4_TCP{TCP: &pb.TCP{}}},
+				f: []*flowpb.FlowFilter{{Protocol: []string{"tcp", "kafka"}}},
+				ev: &v1.Event{Event: &flowpb.Flow{
+					L4: &flowpb.Layer4{Protocol: &flowpb.Layer4_TCP{TCP: &flowpb.TCP{}}},
 				}},
 			},
 			want: true,
@@ -89,7 +89,7 @@ func TestFlowProtocolFilter(t *testing.T) {
 		{
 			name: "invalid protocols",
 			args: args{
-				f: []*pb.FlowFilter{{Protocol: []string{"not a protocol"}}},
+				f: []*flowpb.FlowFilter{{Protocol: []string{"not a protocol"}}},
 			},
 			wantErr: true,
 		},
diff --git a/pkg/hubble/filters/reply_test.go b/pkg/hubble/filters/reply_test.go
index 845438b186..e552897dc4 100644
--- a/pkg/hubble/filters/reply_test.go
+++ b/pkg/hubble/filters/reply_test.go
@@ -20,13 +20,13 @@ import (
 	"context"
 	"testing"
 
-	pb "github.com/cilium/cilium/api/v1/flow"
+	flowpb "github.com/cilium/cilium/api/v1/flow"
 	v1 "github.com/cilium/cilium/pkg/hubble/api/v1"
 )
 
 func Test_filterByReplyField(t *testing.T) {
 	type args struct {
-		f  []*pb.FlowFilter
+		f  []*flowpb.FlowFilter
 		ev *v1.Event
 	}
 	tests := []struct {
@@ -38,7 +38,7 @@ func Test_filterByReplyField(t *testing.T) {
 		{
 			name: "nil flow",
 			args: args{
-				f:  []*pb.FlowFilter{{Reply: []bool{true}}},
+				f:  []*flowpb.FlowFilter{{Reply: []bool{true}}},
 				ev: &v1.Event{},
 			},
 			want: false,
@@ -46,48 +46,48 @@ func Test_filterByReplyField(t *testing.T) {
 		{
 			name: "empty-param",
 			args: args{
-				f:  []*pb.FlowFilter{{Reply: []bool{}}},
-				ev: &v1.Event{Event: &pb.Flow{Reply: true}},
+				f:  []*flowpb.FlowFilter{{Reply: []bool{}}},
+				ev: &v1.Event{Event: &flowpb.Flow{Reply: true}},
 			},
 			want: true,
 		},
 		{
 			name: "empty-param-2",
 			args: args{
-				f:  []*pb.FlowFilter{{Reply: []bool{}}},
-				ev: &v1.Event{Event: &pb.Flow{Reply: false}},
+				f:  []*flowpb.FlowFilter{{Reply: []bool{}}},
+				ev: &v1.Event{Event: &flowpb.Flow{Reply: false}},
 			},
 			want: true,
 		},
 		{
 			name: "no-reply",
 			args: args{
-				f:  []*pb.FlowFilter{{Reply: []bool{false}}},
-				ev: &v1.Event{Event: &pb.Flow{Reply: false}},
+				f:  []*flowpb.FlowFilter{{Reply: []bool{false}}},
+				ev: &v1.Event{Event: &flowpb.Flow{Reply: false}},
 			},
 			want: true,
 		},
 		{
 			name: "reply",
 			args: args{
-				f:  []*pb.FlowFilter{{Reply: []bool{true}}},
-				ev: &v1.Event{Event: &pb.Flow{Reply: true}},
+				f:  []*flowpb.FlowFilter{{Reply: []bool{true}}},
+				ev: &v1.Event{Event: &flowpb.Flow{Reply: true}},
 			},
 			want: true,
 		},
 		{
 			name: "no-match",
 			args: args{
-				f:  []*pb.FlowFilter{{Reply: []bool{true}}},
-				ev: &v1.Event{Event: &pb.Flow{Reply: false}},
+				f:  []*flowpb.FlowFilter{{Reply: []bool{true}}},
+				ev: &v1.Event{Event: &flowpb.Flow{Reply: false}},
 			},
 			want: false,
 		},
 		{
 			name: "no-match-2",
 			args: args{
-				f:  []*pb.FlowFilter{{Reply: []bool{false}}},
-				ev: &v1.Event{Event: &pb.Flow{Reply: true}},
+				f:  []*flowpb.FlowFilter{{Reply: []bool{false}}},
+				ev: &v1.Event{Event: &flowpb.Flow{Reply: true}},
 			},
 			want: false,
 		},
diff --git a/pkg/hubble/filters/verdict_test.go b/pkg/hubble/filters/verdict_test.go
index 8c86957439..782ae0d5f8 100644
--- a/pkg/hubble/filters/verdict_test.go
+++ b/pkg/hubble/filters/verdict_test.go
@@ -19,7 +19,7 @@ package filters
 import (
 	"testing"
 
-	pb "github.com/cilium/cilium/api/v1/flow"
+	flowpb "github.com/cilium/cilium/api/v1/flow"
 	v1 "github.com/cilium/cilium/pkg/hubble/api/v1"
 
 	"github.com/stretchr/testify/assert"
@@ -27,10 +27,10 @@ import (
 
 func TestVerdictFilter(t *testing.T) {
 	ev := &v1.Event{
-		Event: &pb.Flow{
-			Verdict: pb.Verdict_FORWARDED,
+		Event: &flowpb.Flow{
+			Verdict: flowpb.Verdict_FORWARDED,
 		},
 	}
-	assert.True(t, filterByVerdicts([]pb.Verdict{pb.Verdict_FORWARDED})(ev))
-	assert.False(t, filterByVerdicts([]pb.Verdict{pb.Verdict_DROPPED})(ev))
+	assert.True(t, filterByVerdicts([]flowpb.Verdict{flowpb.Verdict_FORWARDED})(ev))
+	assert.False(t, filterByVerdicts([]flowpb.Verdict{flowpb.Verdict_DROPPED})(ev))
 }
diff --git a/pkg/hubble/k8s/utils_test.go b/pkg/hubble/k8s/utils_test.go
index 78641cfbab..b23815892e 100644
--- a/pkg/hubble/k8s/utils_test.go
+++ b/pkg/hubble/k8s/utils_test.go
@@ -17,7 +17,6 @@
 package k8s
 
 import (
-	"reflect"
 	"testing"
 )
 
@@ -72,49 +71,3 @@ func TestParseNamespaceName(t *testing.T) {
 		})
 	}
 }
-
-func TestParseNamespaceNames(t *testing.T) {
-	type args struct {
-		namespaceNames []string
-	}
-	tests := []struct {
-		name      string
-		args      args
-		wantNS    []string
-		wantNames []string
-	}{
-		{
-			args: args{
-				namespaceNames: []string{
-					"default/pod-1",
-					"default/",
-					"pod-1",
-					"",
-				},
-			},
-			wantNS: []string{
-				"default",
-				"default",
-				"default",
-				"",
-			},
-			wantNames: []string{
-				"pod-1",
-				"",
-				"pod-1",
-				"",
-			},
-		},
-	}
-	for _, tt := range tests {
-		t.Run(tt.name, func(t *testing.T) {
-			gotNS, gotNames := ParseNamespaceNames(tt.args.namespaceNames)
-			if !reflect.DeepEqual(gotNS, tt.wantNS) {
-				t.Errorf("ParseNamespaceNames() gotNS = %v, wantNS %v", gotNS, tt.wantNS)
-			}
-			if !reflect.DeepEqual(gotNames, tt.wantNames) {
-				t.Errorf("ParseNamespaceNames() gotPod = %v, wantName %v", gotNames, tt.wantNames)
-			}
-		})
-	}
-}
diff --git a/pkg/hubble/metrics/api/registry_test.go b/pkg/hubble/metrics/api/registry_test.go
index 8496cbf872..7ef9e3c0d1 100644
--- a/pkg/hubble/metrics/api/registry_test.go
+++ b/pkg/hubble/metrics/api/registry_test.go
@@ -19,7 +19,7 @@ package api
 import (
 	"testing"
 
-	"github.com/cilium/cilium/pkg/hubble/api/v1"
+	v1 "github.com/cilium/cilium/pkg/hubble/api/v1"
 
 	"github.com/prometheus/client_golang/prometheus"
 	"github.com/stretchr/testify/assert"
diff --git a/pkg/hubble/parser/threefour/parser_test.go b/pkg/hubble/parser/threefour/parser_test.go
index dc35e1b15a..23a150720b 100644
--- a/pkg/hubble/parser/threefour/parser_test.go
+++ b/pkg/hubble/parser/threefour/parser_test.go
@@ -24,7 +24,7 @@ import (
 	"reflect"
 	"testing"
 
-	pb "github.com/cilium/cilium/api/v1/flow"
+	flowpb "github.com/cilium/cilium/api/v1/flow"
 	"github.com/cilium/cilium/api/v1/models"
 	"github.com/cilium/cilium/pkg/byteorder"
 	v1 "github.com/cilium/cilium/pkg/hubble/api/v1"
@@ -85,15 +85,15 @@ func TestL34Decode(t *testing.T) {
 		},
 	}
 	serviceGetter := &testutils.FakeServiceGetter{
-		OnGetServiceByAddr: func(ip net.IP, port uint16) (service pb.Service, ok bool) {
+		OnGetServiceByAddr: func(ip net.IP, port uint16) (service flowpb.Service, ok bool) {
 			if ip.Equal(net.ParseIP("192.168.33.11")) && (port == 6443) {
-				return pb.Service{
+				return flowpb.Service{
 					Name:      "service-1234",
 					Namespace: "remote",
 				}, true
 			}
 			if ip.Equal(net.ParseIP("10.16.236.178")) && (port == 54222) {
-				return pb.Service{
+				return flowpb.Service{
 					Name:      "service-4321",
 					Namespace: "default",
 				}, true
@@ -110,9 +110,9 @@ func TestL34Decode(t *testing.T) {
 	parser, err := New(endpointGetter, identityCache, dnsGetter, ipGetter, serviceGetter)
 	require.NoError(t, err)
 
-	f := &pb.Flow{}
-	p := &pb.Payload{
-		Type:     pb.EventType_EventSample,
+	f := &flowpb.Flow{}
+	p := &flowpb.Payload{
+		Type:     flowpb.EventType_EventSample,
 		Time:     timestamp,
 		Data:     d,
 		HostName: nodeName,
@@ -138,8 +138,8 @@ func TestL34Decode(t *testing.T) {
 
 	assert.Equal(t, int32(api.MessageTypeTrace), f.GetEventType().GetType())
 	assert.Equal(t, int32(api.TraceFromHost), f.GetEventType().GetSubType())
-	assert.Equal(t, pb.Verdict_FORWARDED, f.GetVerdict())
-	assert.Equal(t, &pb.TCPFlags{ACK: true}, f.L4.GetTCP().GetFlags())
+	assert.Equal(t, flowpb.Verdict_FORWARDED, f.GetVerdict())
+	assert.Equal(t, &flowpb.TCPFlags{ACK: true}, f.L4.GetTCP().GetFlags())
 	assert.Equal(t, nodeName, f.GetNodeName())
 
 	// ICMP packet so no ports until that support is merged into master
@@ -181,8 +181,8 @@ func TestL34Decode(t *testing.T) {
 	parser, err = New(endpointGetter, identityCache, dnsGetter, ipGetter, serviceGetter)
 	require.NoError(t, err)
 
-	p = &pb.Payload{
-		Type:     pb.EventType_EventSample,
+	p = &flowpb.Payload{
+		Type:     flowpb.EventType_EventSample,
 		Time:     timestamp,
 		Data:     d2,
 		HostName: nodeName,
@@ -193,7 +193,7 @@ func TestL34Decode(t *testing.T) {
 	// second packet is ICMPv6 and the flags should be totally wiped out
 	assert.Equal(t, []string(nil), f.GetSourceNames())
 	assert.Equal(t, "ff02::1:ff00:b3e5", f.GetIP().GetSource())
-	assert.Equal(t, &pb.ICMPv6{Type: 135}, f.L4.GetICMPv6())
+	assert.Equal(t, &flowpb.ICMPv6{Type: 135}, f.L4.GetICMPv6())
 	assert.Equal(t, "", f.GetSource().GetPodName())
 	assert.Equal(t, "", f.GetSource().GetNamespace())
 
@@ -204,8 +204,8 @@ func TestL34Decode(t *testing.T) {
 
 	assert.Equal(t, int32(api.MessageTypeTrace), f.GetEventType().GetType())
 	assert.Equal(t, int32(api.TraceFromLxc), f.GetEventType().GetSubType())
-	assert.Equal(t, pb.Verdict_FORWARDED, f.GetVerdict())
-	assert.Equal(t, (*pb.TCPFlags)(nil), f.L4.GetTCP().GetFlags())
+	assert.Equal(t, flowpb.Verdict_FORWARDED, f.GetVerdict())
+	assert.Equal(t, (*flowpb.TCPFlags)(nil), f.L4.GetTCP().GetFlags())
 	assert.Equal(t, nodeName, f.GetNodeName())
 }
 
@@ -230,9 +230,9 @@ func BenchmarkL34Decode(b *testing.B) {
 	parser, err := New(endpointGetter, identityCache, dnsGetter, ipGetter, serviceGetter)
 	require.NoError(b, err)
 
-	f := &pb.Flow{}
-	p := &pb.Payload{
-		Type:     pb.EventType_EventSample,
+	f := &flowpb.Flow{}
+	p := &flowpb.Payload{
+		Type:     flowpb.EventType_EventSample,
 		Time:     timestamp,
 		Data:     d,
 		HostName: nodeName,
@@ -281,8 +281,8 @@ func TestDecodeTraceNotify(t *testing.T) {
 	parser, err := New(&testutils.NoopEndpointGetter, identityGetter, &testutils.NoopDNSGetter, &testutils.NoopIPGetter, &testutils.NoopServiceGetter)
 	require.NoError(t, err)
 
-	f := &pb.Flow{}
-	err = parser.Decode(&pb.Payload{Data: buf.Bytes()}, f)
+	f := &flowpb.Flow{}
+	err = parser.Decode(&flowpb.Payload{Data: buf.Bytes()}, f)
 	require.NoError(t, err)
 	assert.Equal(t, []string{"src=label"}, f.GetSource().GetLabels())
 	assert.Equal(t, []string{"dst=label"}, f.GetDestination().GetLabels())
@@ -326,8 +326,8 @@ func TestDecodeDropNotify(t *testing.T) {
 	parser, err := New(&testutils.NoopEndpointGetter, identityGetter, &testutils.NoopDNSGetter, &testutils.NoopIPGetter, &testutils.NoopServiceGetter)
 	require.NoError(t, err)
 
-	f := &pb.Flow{}
-	err = parser.Decode(&pb.Payload{Data: buf.Bytes()}, f)
+	f := &flowpb.Flow{}
+	err = parser.Decode(&flowpb.Payload{Data: buf.Bytes()}, f)
 	require.NoError(t, err)
 	assert.Equal(t, []string{"src=label"}, f.GetSource().GetLabels())
 	assert.Equal(t, []string{"dst=label"}, f.GetDestination().GetLabels())
@@ -361,14 +361,14 @@ func TestDecodePolicyVerdictNotify(t *testing.T) {
 	data, err := testutils.CreateL3L4Payload(pvn)
 	require.NoError(t, err)
 
-	f := &pb.Flow{}
-	err = parser.Decode(&pb.Payload{Data: data}, f)
+	f := &flowpb.Flow{}
+	err = parser.Decode(&flowpb.Payload{Data: data}, f)
 	require.NoError(t, err)
 
 	assert.Equal(t, int32(api.MessageTypePolicyVerdict), f.GetEventType().GetType())
-	assert.Equal(t, pb.TrafficDirection_EGRESS, f.GetTrafficDirection())
+	assert.Equal(t, flowpb.TrafficDirection_EGRESS, f.GetTrafficDirection())
 	assert.Equal(t, uint32(api.PolicyMatchL3L4), f.GetPolicyMatchType())
-	assert.Equal(t, pb.Verdict_FORWARDED, f.GetVerdict())
+	assert.Equal(t, flowpb.Verdict_FORWARDED, f.GetVerdict())
 	assert.Equal(t, []string{"dst=label"}, f.GetDestination().GetLabels())
 
 	// PolicyVerdictNotify for dropped flow
@@ -384,13 +384,13 @@ func TestDecodePolicyVerdictNotify(t *testing.T) {
 	require.NoError(t, err)
 
 	f.Reset()
-	err = parser.Decode(&pb.Payload{Data: data}, f)
+	err = parser.Decode(&flowpb.Payload{Data: data}, f)
 	require.NoError(t, err)
 
 	assert.Equal(t, int32(api.MessageTypePolicyVerdict), f.GetEventType().GetType())
-	assert.Equal(t, pb.TrafficDirection_INGRESS, f.GetTrafficDirection())
+	assert.Equal(t, flowpb.TrafficDirection_INGRESS, f.GetTrafficDirection())
 	assert.Equal(t, uint32(151), f.GetDropReason())
-	assert.Equal(t, pb.Verdict_DROPPED, f.GetVerdict())
+	assert.Equal(t, flowpb.Verdict_DROPPED, f.GetVerdict())
 	assert.Equal(t, []string{"dst=label"}, f.GetSource().GetLabels())
 }
 
@@ -406,8 +406,8 @@ func TestDecodeDropReason(t *testing.T) {
 	parser, err := New(nil, nil, nil, nil, nil)
 	require.NoError(t, err)
 
-	f := &pb.Flow{}
-	err = parser.Decode(&pb.Payload{Data: data}, f)
+	f := &flowpb.Flow{}
+	err = parser.Decode(&flowpb.Payload{Data: data}, f)
 	require.NoError(t, err)
 
 	assert.Equal(t, uint32(reason), f.GetDropReason())
@@ -430,8 +430,8 @@ func TestDecodeLocalIdentity(t *testing.T) {
 	parser, err := New(nil, identityGetter, nil, nil, nil)
 	require.NoError(t, err)
 
-	f := &pb.Flow{}
-	err = parser.Decode(&pb.Payload{Data: data}, f)
+	f := &flowpb.Flow{}
+	err = parser.Decode(&flowpb.Payload{Data: data}, f)
 	require.NoError(t, err)
 
 	assert.Equal(t, []string{"cidr:1.2.3.4/12", "some=label"}, f.GetSource().GetLabels())
@@ -456,7 +456,7 @@ func TestDecodeTrafficDirection(t *testing.T) {
 
 	parser, err := New(endpointGetter, nil, nil, nil, nil)
 	require.NoError(t, err)
-	parseFlow := func(event interface{}, srcIPv4, dstIPv4 net.IP) *pb.Flow {
+	parseFlow := func(event interface{}, srcIPv4, dstIPv4 net.IP) *flowpb.Flow {
 		data, err := testutils.CreateL3L4Payload(event,
 			&layers.Ethernet{
 				SrcMAC:       net.HardwareAddr{1, 2, 3, 4, 5, 6},
@@ -465,8 +465,8 @@ func TestDecodeTrafficDirection(t *testing.T) {
 			},
 			&layers.IPv4{SrcIP: srcIPv4, DstIP: dstIPv4})
 		require.NoError(t, err)
-		f := &pb.Flow{}
-		err = parser.Decode(&pb.Payload{Data: data}, f)
+		f := &flowpb.Flow{}
+		err = parser.Decode(&flowpb.Payload{Data: data}, f)
 		require.NoError(t, err)
 		return f
 	}
@@ -476,7 +476,7 @@ func TestDecodeTrafficDirection(t *testing.T) {
 		Type: byte(api.MessageTypeDrop),
 	}
 	f := parseFlow(dn, localIP, remoteIP)
-	assert.Equal(t, pb.TrafficDirection_TRAFFIC_DIRECTION_UNKNOWN, f.GetTrafficDirection())
+	assert.Equal(t, flowpb.TrafficDirection_TRAFFIC_DIRECTION_UNKNOWN, f.GetTrafficDirection())
 	assert.Equal(t, uint32(localEP), f.GetSource().GetID())
 
 	// DROP Egress
@@ -485,7 +485,7 @@ func TestDecodeTrafficDirection(t *testing.T) {
 		Source: localEP,
 	}
 	f = parseFlow(dn, localIP, remoteIP)
-	assert.Equal(t, pb.TrafficDirection_EGRESS, f.GetTrafficDirection())
+	assert.Equal(t, flowpb.TrafficDirection_EGRESS, f.GetTrafficDirection())
 	assert.Equal(t, uint32(localEP), f.GetSource().GetID())
 
 	// DROP Ingress
@@ -494,7 +494,7 @@ func TestDecodeTrafficDirection(t *testing.T) {
 		Source: localEP,
 	}
 	f = parseFlow(dn, remoteIP, localIP)
-	assert.Equal(t, pb.TrafficDirection_INGRESS, f.GetTrafficDirection())
+	assert.Equal(t, flowpb.TrafficDirection_INGRESS, f.GetTrafficDirection())
 	assert.Equal(t, uint32(localEP), f.GetDestination().GetID())
 
 	// TRACE_TO_LXC at unknown endpoint
@@ -503,7 +503,7 @@ func TestDecodeTrafficDirection(t *testing.T) {
 		ObsPoint: api.TraceToLxc,
 	}
 	f = parseFlow(tn, localIP, remoteIP)
-	assert.Equal(t, pb.TrafficDirection_TRAFFIC_DIRECTION_UNKNOWN, f.GetTrafficDirection())
+	assert.Equal(t, flowpb.TrafficDirection_TRAFFIC_DIRECTION_UNKNOWN, f.GetTrafficDirection())
 	assert.Equal(t, uint32(localEP), f.GetSource().GetID())
 
 	// TRACE_TO_LXC Egress
@@ -513,7 +513,7 @@ func TestDecodeTrafficDirection(t *testing.T) {
 		ObsPoint: api.TraceToLxc,
 	}
 	f = parseFlow(tn, localIP, remoteIP)
-	assert.Equal(t, pb.TrafficDirection_EGRESS, f.GetTrafficDirection())
+	assert.Equal(t, flowpb.TrafficDirection_EGRESS, f.GetTrafficDirection())
 	assert.Equal(t, uint32(localEP), f.GetSource().GetID())
 
 	// TRACE_TO_LXC Egress, reversed by CT_REPLY
@@ -524,7 +524,7 @@ func TestDecodeTrafficDirection(t *testing.T) {
 		Reason:   monitor.TraceReasonCtReply,
 	}
 	f = parseFlow(tn, localIP, remoteIP)
-	assert.Equal(t, pb.TrafficDirection_INGRESS, f.GetTrafficDirection())
+	assert.Equal(t, flowpb.TrafficDirection_INGRESS, f.GetTrafficDirection())
 	assert.Equal(t, uint32(localEP), f.GetSource().GetID())
 
 	// TRACE_TO_HOST Ingress
@@ -534,7 +534,7 @@ func TestDecodeTrafficDirection(t *testing.T) {
 		ObsPoint: api.TraceToHost,
 	}
 	f = parseFlow(tn, remoteIP, localIP)
-	assert.Equal(t, pb.TrafficDirection_INGRESS, f.GetTrafficDirection())
+	assert.Equal(t, flowpb.TrafficDirection_INGRESS, f.GetTrafficDirection())
 	assert.Equal(t, uint32(localEP), f.GetDestination().GetID())
 
 	// TRACE_TO_HOST Ingress, reversed by CT_REPLY
@@ -545,7 +545,7 @@ func TestDecodeTrafficDirection(t *testing.T) {
 		Reason:   monitor.TraceReasonCtReply,
 	}
 	f = parseFlow(tn, remoteIP, localIP)
-	assert.Equal(t, pb.TrafficDirection_EGRESS, f.GetTrafficDirection())
+	assert.Equal(t, flowpb.TrafficDirection_EGRESS, f.GetTrafficDirection())
 	assert.Equal(t, uint32(localEP), f.GetDestination().GetID())
 
 	// TRACE_FROM_LXC (traffic direction not supported)
@@ -555,7 +555,7 @@ func TestDecodeTrafficDirection(t *testing.T) {
 		ObsPoint: api.TraceFromLxc,
 	}
 	f = parseFlow(tn, localIP, remoteIP)
-	assert.Equal(t, pb.TrafficDirection_TRAFFIC_DIRECTION_UNKNOWN, f.GetTrafficDirection())
+	assert.Equal(t, flowpb.TrafficDirection_TRAFFIC_DIRECTION_UNKNOWN, f.GetTrafficDirection())
 	assert.Equal(t, uint32(localEP), f.GetSource().GetID())
 
 	// PolicyVerdictNotify Egress
@@ -565,7 +565,7 @@ func TestDecodeTrafficDirection(t *testing.T) {
 		Flags:  api.PolicyEgress,
 	}
 	f = parseFlow(pvn, localIP, remoteIP)
-	assert.Equal(t, pb.TrafficDirection_EGRESS, f.GetTrafficDirection())
+	assert.Equal(t, flowpb.TrafficDirection_EGRESS, f.GetTrafficDirection())
 	assert.Equal(t, uint32(localEP), f.GetSource().GetID())
 
 	// PolicyVerdictNotify Ingress
@@ -575,7 +575,7 @@ func TestDecodeTrafficDirection(t *testing.T) {
 		Flags:  api.PolicyIngress,
 	}
 	f = parseFlow(pvn, remoteIP, localIP)
-	assert.Equal(t, pb.TrafficDirection_INGRESS, f.GetTrafficDirection())
+	assert.Equal(t, flowpb.TrafficDirection_INGRESS, f.GetTrafficDirection())
 	assert.Equal(t, uint32(localEP), f.GetDestination().GetID())
 }
 
@@ -627,7 +627,7 @@ func Test_filterCidrLabels(t *testing.T) {
 }
 
 func TestTraceNotifyOriginalIP(t *testing.T) {
-	f := &pb.Flow{}
+	f := &flowpb.Flow{}
 	parser, err := New(&testutils.NoopEndpointGetter, nil, &testutils.NoopDNSGetter, &testutils.NoopIPGetter, &testutils.NoopServiceGetter)
 	require.NoError(t, err)
 
@@ -647,7 +647,7 @@ func TestTraceNotifyOriginalIP(t *testing.T) {
 	data, err := testutils.CreateL3L4Payload(v0, &eth, &ip, &layers.TCP{})
 	require.NoError(t, err)
 
-	err = parser.Decode(&pb.Payload{Data: data}, f)
+	err = parser.Decode(&flowpb.Payload{Data: data}, f)
 	require.NoError(t, err)
 	assert.Equal(t, f.IP.Source, "2.2.2.2")
 
@@ -660,7 +660,7 @@ func TestTraceNotifyOriginalIP(t *testing.T) {
 	}
 	data, err = testutils.CreateL3L4Payload(v1, &eth, &ip, &layers.TCP{})
 	require.NoError(t, err)
-	err = parser.Decode(&pb.Payload{Data: data}, f)
+	err = parser.Decode(&flowpb.Payload{Data: data}, f)
 	require.NoError(t, err)
 	assert.Equal(t, f.IP.Source, "1.1.1.1")
 }
@@ -691,8 +691,8 @@ func TestICMP(t *testing.T) {
 	}
 	v4data, err := testutils.CreateL3L4Payload(message, &eth, &ip, &icmpv4)
 	require.NoError(t, err)
-	v4flow := &pb.Flow{}
-	err = parser.Decode(&pb.Payload{Data: v4data}, v4flow)
+	v4flow := &flowpb.Flow{}
+	err = parser.Decode(&flowpb.Payload{Data: v4data}, v4flow)
 	require.NoError(t, err)
 	assert.Equal(t, uint32(1), v4flow.GetL4().GetICMPv4().Type)
 	assert.Equal(t, uint32(2), v4flow.GetL4().GetICMPv4().Code)
@@ -714,15 +714,15 @@ func TestICMP(t *testing.T) {
 	}
 	v6data, err := testutils.CreateL3L4Payload(message, &ethv6, &ipv6, &icmpv6)
 	require.NoError(t, err)
-	v6flow := &pb.Flow{}
-	err = parser.Decode(&pb.Payload{Data: v6data}, v6flow)
+	v6flow := &flowpb.Flow{}
+	err = parser.Decode(&flowpb.Payload{Data: v6data}, v6flow)
 	require.NoError(t, err)
 	assert.Equal(t, uint32(3), v6flow.GetL4().GetICMPv6().Type)
 	assert.Equal(t, uint32(4), v6flow.GetL4().GetICMPv6().Code)
 }
 
 func TestTraceNotifyLocalEndpoint(t *testing.T) {
-	f := &pb.Flow{}
+	f := &flowpb.Flow{}
 
 	ep := &v1.Endpoint{
 		ID:           1234,
@@ -760,7 +760,7 @@ func TestTraceNotifyLocalEndpoint(t *testing.T) {
 	data, err := testutils.CreateL3L4Payload(v0, &eth, &ip, &layers.TCP{})
 	require.NoError(t, err)
 
-	err = parser.Decode(&pb.Payload{Data: data}, f)
+	err = parser.Decode(&flowpb.Payload{Data: data}, f)
 	require.NoError(t, err)
 
 	assert.Equal(t, uint32(ep.ID), f.Source.ID)
diff --git a/pkg/hubble/testutils/fake.go b/pkg/hubble/testutils/fake.go
index 01795f0896..db21fcc56e 100644
--- a/pkg/hubble/testutils/fake.go
+++ b/pkg/hubble/testutils/fake.go
@@ -161,134 +161,6 @@ var NoopIdentityGetter = FakeIdentityGetter{
 	},
 }
 
-// FakeEndpointsHandler implements EndpointsHandler interface for unit testing.
-type FakeEndpointsHandler struct {
-	FakeSyncEndpoints            func([]*v1.Endpoint)
-	FakeUpdateEndpoint           func(*v1.Endpoint)
-	FakeDeleteEndpoint           func(*v1.Endpoint)
-	FakeFindEPs                  func(epID uint64, ns, pod string) []v1.Endpoint
-	FakeGetEndpoint              func(ip net.IP) (endpoint *v1.Endpoint, ok bool)
-	FakeGetEndpointByContainerID func(id string) (endpoint *v1.Endpoint, ok bool)
-	FakeGetEndpointByPodName     func(namespace string, name string) (*v1.Endpoint, bool)
-}
-
-// SyncEndpoints calls FakeSyncEndpoints.
-func (f *FakeEndpointsHandler) SyncEndpoints(eps []*v1.Endpoint) {
-	if f.FakeSyncEndpoints != nil {
-		f.FakeSyncEndpoints(eps)
-		return
-	}
-	panic("SyncEndpoints([]*v1.Endpoint) should not have been called since it was not defined")
-}
-
-// UpdateEndpoint calls FakeUpdateEndpoint.
-func (f *FakeEndpointsHandler) UpdateEndpoint(ep *v1.Endpoint) {
-	if f.FakeUpdateEndpoint != nil {
-		f.FakeUpdateEndpoint(ep)
-		return
-	}
-	panic("UpdateEndpoint(*v1.Endpoint) should not have been called since it was not defined")
-}
-
-// DeleteEndpoint calls FakeDeleteEndpoint.
-func (f *FakeEndpointsHandler) DeleteEndpoint(ep *v1.Endpoint) {
-	if f.FakeDeleteEndpoint != nil {
-		f.FakeDeleteEndpoint(ep)
-		return
-	}
-	panic("DeleteEndpoint(*v1.Endpoint) should not have been called since it was not defined")
-}
-
-// FindEPs calls FakeFindEPs.
-func (f *FakeEndpointsHandler) FindEPs(epID uint64, ns, pod string) []v1.Endpoint {
-	if f.FakeFindEPs != nil {
-		return f.FakeFindEPs(epID, ns, pod)
-	}
-	panic(" FindEPs(epID uint64, ns, pod string) should not have been called since it was not defined")
-}
-
-// GetEndpoint calls FakeGetEndpoint.
-func (f *FakeEndpointsHandler) GetEndpoint(ip net.IP) (ep *v1.Endpoint, ok bool) {
-	if f.FakeGetEndpoint != nil {
-		return f.FakeGetEndpoint(ip)
-	}
-	panic("GetEndpoint(ip net.IP) (ep *v1.Endpoint, ok bool) should not have been called since it was not defined")
-}
-
-// GetEndpointByContainerID calls FakeGetEndpointByContainerID.
-func (f *FakeEndpointsHandler) GetEndpointByContainerID(id string) (ep *v1.Endpoint, ok bool) {
-	if f.FakeGetEndpointByContainerID != nil {
-		return f.FakeGetEndpointByContainerID(id)
-	}
-	panic("GetEndpointByContainerID(id string) (ep *v1.Endpoint, ok bool) should not have been called since it was not defined")
-}
-
-// GetEndpointByPodName calls FakeGetEndpointByPodName.
-func (f *FakeEndpointsHandler) GetEndpointByPodName(namespace string, name string) (ep *v1.Endpoint, ok bool) {
-	if f.FakeGetEndpointByPodName != nil {
-		return f.FakeGetEndpointByPodName(namespace, name)
-	}
-	panic("GetEndpointByPodName(namespace string, name string) (ep *v1.Endpoint, ok bool) should not have been called since it was not defined")
-}
-
-// FakeCiliumClient implements CliliumClient interface for unit testing.
-type FakeCiliumClient struct {
-	FakeEndpointList    func() ([]*models.Endpoint, error)
-	FakeGetEndpoint     func(uint64) (*models.Endpoint, error)
-	FakeGetIdentity     func(uint64) (*models.Identity, error)
-	FakeGetFqdnCache    func() ([]*models.DNSLookup, error)
-	FakeGetIPCache      func() ([]*models.IPListEntry, error)
-	FakeGetServiceCache func() ([]*models.Service, error)
-}
-
-// EndpointList calls FakeEndpointList.
-func (c *FakeCiliumClient) EndpointList() ([]*models.Endpoint, error) {
-	if c.FakeEndpointList != nil {
-		return c.FakeEndpointList()
-	}
-	panic("EndpointList() should not have been called since it was not defined")
-}
-
-// GetEndpoint calls FakeGetEndpoint.
-func (c *FakeCiliumClient) GetEndpoint(id uint64) (*models.Endpoint, error) {
-	if c.FakeGetEndpoint != nil {
-		return c.FakeGetEndpoint(id)
-	}
-	panic("GetEndpoint(uint64) should not have been called since it was not defined")
-}
-
-// GetIdentity calls FakeGetIdentity.
-func (c *FakeCiliumClient) GetIdentity(id uint64) (*models.Identity, error) {
-	if c.FakeGetIdentity != nil {
-		return c.FakeGetIdentity(id)
-	}
-	panic("GetIdentity(uint64) should not have been called since it was not defined")
-}
-
-// GetFqdnCache calls FakeGetFqdnCache.
-func (c *FakeCiliumClient) GetFqdnCache() ([]*models.DNSLookup, error) {
-	if c.FakeGetFqdnCache != nil {
-		return c.FakeGetFqdnCache()
-	}
-	panic("GetFqdnCache() should not have been called since it was not defined")
-}
-
-// GetIPCache calls FakeGetIPCache.
-func (c *FakeCiliumClient) GetIPCache() ([]*models.IPListEntry, error) {
-	if c.FakeGetIPCache != nil {
-		return c.FakeGetIPCache()
-	}
-	panic("GetIPCache() should not have been called since it was not defined")
-}
-
-// GetServiceCache calls FakeGetServiceCache.
-func (c *FakeCiliumClient) GetServiceCache() ([]*models.Service, error) {
-	if c.FakeGetServiceCache != nil {
-		return c.FakeGetServiceCache()
-	}
-	panic("GetServiceCache() should not have been called since it was not defined")
-}
-
 // FakeFlow implements v1.Flow for unit tests. All interface methods
 // return values exposed in the fields.
 type FakeFlow struct {
diff --git a/pkg/policy/api/rule_test.go b/pkg/policy/api/rule_test.go
new file mode 100644
index 0000000000..398df8caf7
--- /dev/null
+++ b/pkg/policy/api/rule_test.go
@@ -0,0 +1,49 @@
+// Copyright 2020 Authors of Cilium
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+// +build !privileged_tests
+
+package api
+
+import (
+	"encoding/json"
+
+	. "gopkg.in/check.v1"
+)
+
+func checkMarshalUnmarshal(c *C, r *Rule) {
+	jsonData, err := json.Marshal(r)
+	c.Assert(err, IsNil)
+
+	newRule := Rule{}
+	err = json.Unmarshal(jsonData, &newRule)
+	c.Assert(err, IsNil)
+
+	c.Check(newRule.EndpointSelector.LabelSelector == nil, Equals, r.EndpointSelector.LabelSelector == nil)
+	c.Check(newRule.NodeSelector.LabelSelector == nil, Equals, r.NodeSelector.LabelSelector == nil)
+}
+
+// This test ensures that the NodeSelector and EndpointSelector fields are kept
+// empty when the rule is marshalled/unmarshalled.
+func (s *PolicyAPITestSuite) TestJSONMarshalling(c *C) {
+	validEndpointRule := Rule{
+		EndpointSelector: WildcardEndpointSelector,
+	}
+	checkMarshalUnmarshal(c, &validEndpointRule)
+
+	validNodeRule := Rule{
+		NodeSelector: WildcardEndpointSelector,
+	}
+	checkMarshalUnmarshal(c, &validNodeRule)
+}
diff --git a/pkg/policy/api/rule_validation_test.go b/pkg/policy/api/rule_validation_test.go
index d421b2a970..09d1f46554 100644
--- a/pkg/policy/api/rule_validation_test.go
+++ b/pkg/policy/api/rule_validation_test.go
@@ -421,6 +421,62 @@ func (s *PolicyAPITestSuite) TestL7Rules(c *C) {
 	c.Assert(err, Not(IsNil))
 }
 
+// This test ensures that host policies with L7 rules are rejected.
+func (s *PolicyAPITestSuite) TestL7RulesWithNodeSelector(c *C) {
+	invalidL7RuleIngress := Rule{
+		NodeSelector: WildcardEndpointSelector,
+		Ingress: []IngressRule{
+			{
+				FromEndpoints: []EndpointSelector{WildcardEndpointSelector},
+				ToPorts: []PortRule{{
+					Ports: []PortProtocol{
+						{Port: "80", Protocol: ProtoTCP},
+					},
+					Rules: &L7Rules{
+						HTTP: []PortRuleHTTP{
+							{Method: "PUT", Path: "/"},
+						},
+					},
+				}},
+			},
+		},
+	}
+	err := invalidL7RuleIngress.Sanitize()
+	c.Assert(err.Error(), Equals, "host policies do not support L7 rules yet")
+
+	invalidL7RuleEgress := Rule{
+		NodeSelector: WildcardEndpointSelector,
+		Egress: []EgressRule{
+			{
+				ToEndpoints: []EndpointSelector{WildcardEndpointSelector},
+				ToPorts: []PortRule{{
+					Ports: []PortProtocol{
+						{Port: "53", Protocol: ProtoUDP},
+					},
+					Rules: &L7Rules{
+						DNS: []PortRuleDNS{
+							{MatchName: "domain.com"},
+						},
+					},
+				}},
+			},
+		},
+	}
+	err = invalidL7RuleEgress.Sanitize()
+	c.Assert(err.Error(), Equals, "host policies do not support L7 rules yet")
+
+	validL7RuleIngress := Rule{
+		NodeSelector: WildcardEndpointSelector,
+		Ingress: []IngressRule{
+			{
+				FromEndpoints: []EndpointSelector{WildcardEndpointSelector},
+			},
+		},
+	}
+	err = validL7RuleIngress.Sanitize()
+	c.Assert(err, IsNil)
+}
+
 func (s *PolicyAPITestSuite) TestInvalidEndpointSelectors(c *C) {
 
 	// Operator in MatchExpressions is invalid, so sanitization should fail.
@@ -497,6 +553,41 @@ func (s *PolicyAPITestSuite) TestInvalidEndpointSelectors(c *C) {
 
 }
 
+func (s *PolicyAPITestSuite) TestNodeSelector(c *C) {
+	// Operator in MatchExpressions is invalid, so sanitization should fail.
+	labelSel := &slim_metav1.LabelSelector{
+		MatchLabels: map[string]string{
+			"any.foo": "bar",
+			"k8s.baz": "alice",
+		},
+		MatchExpressions: []slim_metav1.LabelSelectorRequirement{
+			{
+				Key:      "any.foo",
+				Operator: "asdfasdfasdf",
+				Values:   []string{"default"},
+			},
+		},
+	}
+	invalidSel := NewESFromK8sLabelSelector(labels.LabelSourceK8sKeyPrefix, labelSel)
+	invalidNodeSelectorRule := Rule{
+		NodeSelector: invalidSel,
+	}
+	err := invalidNodeSelectorRule.Sanitize()
+	c.Assert(err.Error(), Equals,
+		"invalid label selector: matchExpressions[0].operator: Invalid value: \"asdfasdfasdf\": not a valid selector operator")
+
+	invalidRuleBothSelectors := Rule{
+		EndpointSelector: WildcardEndpointSelector,
+		NodeSelector:     WildcardEndpointSelector,
+	}
+	err = invalidRuleBothSelectors.Sanitize()
+	c.Assert(err.Error(), Equals, "rule cannot have both EndpointSelector and NodeSelector")
+
+	invalidRuleNoSelector := Rule{}
+	err = invalidRuleNoSelector.Sanitize()
+	c.Assert(err.Error(), Equals, "rule must have one of EndpointSelector or NodeSelector")
+}
+
 func (s *PolicyAPITestSuite) TestTooManyPortsRule(c *C) {
 
 	var portProtocols []PortProtocol
diff --git a/pkg/policy/identifier_test.go b/pkg/policy/identifier_test.go
index 7060711a59..e74233b09b 100644
--- a/pkg/policy/identifier_test.go
+++ b/pkg/policy/identifier_test.go
@@ -32,6 +32,10 @@ func (d *DummyEndpoint) GetID16() uint16 {
 	return 0
 }
 
+func (d *DummyEndpoint) IsHost() bool {
+	return false
+}
+
 func (d *DummyEndpoint) GetSecurityIdentity() (*identity.Identity, error) {
 	return nil, nil
 }
diff --git a/pkg/policy/resolve_test.go b/pkg/policy/resolve_test.go
index aa083350b1..b5f7dfdf9c 100644
--- a/pkg/policy/resolve_test.go
+++ b/pkg/policy/resolve_test.go
@@ -54,6 +54,10 @@ func (d *dummyEndpoint) GetID16() uint16 {
 	return d.ID
 }
 
+func (d *dummyEndpoint) IsHost() bool {
+	return false
+}
+
 func (d *dummyEndpoint) GetSecurityIdentity() (*identity.Identity, error) {
 	return d.SecurityIdentity, nil
 }
diff --git a/pkg/policy/rule_test.go b/pkg/policy/rule_test.go
index 9b77cd3692..a02b0f7b42 100644
--- a/pkg/policy/rule_test.go
+++ b/pkg/policy/rule_test.go
@@ -2241,16 +2241,27 @@ func (ds *PolicyTestSuite) TestL3L4L7Merge(c *C) {
 }
 
 func (ds *PolicyTestSuite) TestMatches(c *C) {
-	repo := parseAndAddRules(c, api.Rules{&api.Rule{
-		EndpointSelector: endpointSelectorA,
-		Ingress: []api.IngressRule{
-			{
-				FromEndpoints: []api.EndpointSelector{endpointSelectorC},
+	repo := parseAndAddRules(c, api.Rules{
+		&api.Rule{
+			EndpointSelector: endpointSelectorA,
+			Ingress: []api.IngressRule{
+				{
+					FromEndpoints: []api.EndpointSelector{endpointSelectorC},
+				},
 			},
 		},
-	}})
+		&api.Rule{
+			NodeSelector: endpointSelectorA,
+			Ingress: []api.IngressRule{
+				{
+					FromEndpoints: []api.EndpointSelector{endpointSelectorC},
+				},
+			},
+		},
+	})
 
-	addedRule := repo.rules[0]
+	epRule := repo.rules[0]
+	hostRule := repo.rules[1]
 
 	selectedEpLabels := labels.ParseSelectLabel("id=a")
 	selectedIdentity := identity2.NewIdentity(54321, labels.Labels{selectedEpLabels.Key: selectedEpLabels})
@@ -2258,23 +2269,41 @@ func (ds *PolicyTestSuite) TestMatches(c *C) {
 	notSelectedEpLabels := labels.ParseSelectLabel("id=b")
 	notSelectedIdentity := identity2.NewIdentity(9876, labels.Labels{notSelectedEpLabels.Key: notSelectedEpLabels})
 
+	hostLabels := labels.Labels{selectedEpLabels.Key: selectedEpLabels}
+	hostLabels.MergeLabels(labels.LabelHost)
+	hostIdentity := identity2.NewIdentity(identity.ReservedIdentityHost, hostLabels)
+
 	// notSelectedEndpoint is not selected by rule, so we it shouldn't be added
 	// to EndpointsSelected.
-	c.Assert(addedRule.matches(notSelectedIdentity), Equals, false)
-	c.Assert(addedRule.metadata.IdentitySelected, checker.DeepEquals, map[identity.NumericIdentity]bool{notSelectedIdentity.ID: false})
+	c.Assert(epRule.matches(notSelectedIdentity), Equals, false)
+	c.Assert(epRule.metadata.IdentitySelected, checker.DeepEquals, map[identity.NumericIdentity]bool{notSelectedIdentity.ID: false})
 
 	// selectedEndpoint is selected by rule, so we it should be added to
 	// EndpointsSelected.
-	c.Assert(addedRule.matches(selectedIdentity), Equals, true)
-	c.Assert(addedRule.metadata.IdentitySelected, checker.DeepEquals, map[identity.NumericIdentity]bool{selectedIdentity.ID: true, notSelectedIdentity.ID: false})
+	c.Assert(epRule.matches(selectedIdentity), Equals, true)
+	c.Assert(epRule.metadata.IdentitySelected, checker.DeepEquals, map[identity.NumericIdentity]bool{selectedIdentity.ID: true, notSelectedIdentity.ID: false})
 
 	// Test again to check for caching working correctly.
-	c.Assert(addedRule.matches(selectedIdentity), Equals, true)
-	c.Assert(addedRule.metadata.IdentitySelected, checker.DeepEquals, map[identity.NumericIdentity]bool{selectedIdentity.ID: true, notSelectedIdentity.ID: false})
+	c.Assert(epRule.matches(selectedIdentity), Equals, true)
+	c.Assert(epRule.metadata.IdentitySelected, checker.DeepEquals, map[identity.NumericIdentity]bool{selectedIdentity.ID: true, notSelectedIdentity.ID: false})
 
 	// Possible scenario where an endpoint is deleted, and soon after another
 	// endpoint is added with the same ID, but with a different identity. Matching
 	// needs to handle this case correctly.
-	c.Assert(addedRule.matches(notSelectedIdentity), Equals, false)
-	c.Assert(addedRule.metadata.IdentitySelected, checker.DeepEquals, map[identity.NumericIdentity]bool{selectedIdentity.ID: true, notSelectedIdentity.ID: false})
+	c.Assert(epRule.matches(notSelectedIdentity), Equals, false)
+	c.Assert(epRule.metadata.IdentitySelected, checker.DeepEquals, map[identity.NumericIdentity]bool{selectedIdentity.ID: true, notSelectedIdentity.ID: false})
+
+	// host endpoint is not selected by rule, so we it shouldn't be added to EndpointsSelected.
+	c.Assert(epRule.matches(hostIdentity), Equals, false)
+	c.Assert(epRule.metadata.IdentitySelected, checker.DeepEquals,
+		map[identity.NumericIdentity]bool{selectedIdentity.ID: true, notSelectedIdentity.ID: false, hostIdentity.ID: false})
+
+	// selectedEndpoint is not selected by rule, so we it shouldn't be added to EndpointsSelected.
+	c.Assert(hostRule.matches(selectedIdentity), Equals, false)
+	c.Assert(hostRule.metadata.IdentitySelected, checker.DeepEquals, map[identity.NumericIdentity]bool{selectedIdentity.ID: false})
+
+	// host endpoint is selected by rule, so we it should be added to EndpointsSelected.
+	c.Assert(hostRule.matches(hostIdentity), Equals, true)
+	c.Assert(hostRule.metadata.IdentitySelected, checker.DeepEquals,
+		map[identity.NumericIdentity]bool{selectedIdentity.ID: false, hostIdentity.ID: true})
 }
diff --git a/test/helpers/utils.go b/test/helpers/utils.go
index 02d30f92b6..e33954cada 100644
--- a/test/helpers/utils.go
+++ b/test/helpers/utils.go
@@ -531,6 +531,11 @@ func DoesNotExistNodeWithoutCilium() bool {
 	return !ExistNodeWithoutCilium()
 }
 
+// DoesNotSupportMetalLB tells whether our CI can run MetalLB
+func DoesNotSupportMetalLB() bool {
+	return true
+}
+
 // GetNodeWithoutCilium returns a name of a node which does not run cilium.
 func GetNodeWithoutCilium() string {
 	return os.Getenv("NO_CILIUM_ON_NODE")
diff --git a/test/k8sT/Services.go b/test/k8sT/Services.go
index b64751cf6b..d080e1d5e6 100644
--- a/test/k8sT/Services.go
+++ b/test/k8sT/Services.go
@@ -1198,7 +1198,7 @@ var _ = Describe("K8sServicesTest", func() {
 						doRequestsFromThirdHostWithLocalPort(svc2URL, 1, false, 64002)
 					})
 
-					SkipContextIf(helpers.DoesNotExistNodeWithoutCilium, "Tests with MetalLB", func() {
+					SkipContextIf(helpers.DoesNotSupportMetalLB, "Tests with MetalLB, GH#10763", func() {
 						var (
 							metalLB string
 						)