Allow referencing waypoints that are sent as workloads rather than se…

…rvices (istio#1306)

Currently, we are failing if the waypoint is a workload. This is fine
for our typical cases, where we have a Gateway making a
Service+Deployment, so we get a Service + Workloads.

But in other cases, we may use an external waypoint solution, where we
only get an Address in the Gateway status, and we need to map it to
something. In these cases, it may be desireable to support these as
Workloads. For instance, we map a variety of things as Workloads -- not
just Pods:
```
		PodWorkloads,
		WorkloadEntryWorkloads,
		ServiceEntryWorkloads,
		EndpointSliceWorkloads,
		NetworkGatewayWorkloads,
```

So this enables any of those to be used

src/state.rs

@@ -240,13 +240,20 @@ impl ProxyState {
 
     /// Find either a workload or a service by hostname.
     pub fn find_hostname(&self, name: &NamespacedHostname) -> Option<Address> {
-        // Hostnames for services are more common, so lookup service first and fallback
-        // to workload.
-        // We do not looking up workloads by hostname. We could, but we only allow referencing "frontends",
-        // not backends
+        // Hostnames for services are more common, so lookup service first and fallback to workload.
         self.services
             .get_by_namespaced_host(name)
             .map(Address::Service)
+            .or_else(|| {
+                // Slow path: lookup workload by O(n) lookup. This is an uncommon path, so probably not worth
+                // the memory cost to index currently
+                self.workloads
+                    .by_uid
+                    .values()
+                    .find(|w| w.hostname == name.hostname && w.namespace == name.namespace)
+                    .cloned()
+                    .map(Address::Workload)
+            })
     }
 
     fn find_upstream(
@@ -794,10 +801,9 @@ impl DemandProxyState {
                         // adapt to the callers IP family.
                         (us, original_destination_address)
                     }
-                    Some(_) => {
-                        return Err(Error::UnsupportedFeature(
-                            "waypoint must be a service, not a workload".to_string(),
-                        ))
+                    Some(Address::Workload(w)) => {
+                        let us = Some((w, gw_address.hbone_mtls_port, None));
+                        (us, original_destination_address)
                     }
                     None => {
                         return Err(Error::UnknownWaypoint(format!(

tests/namespaced.rs

@@ -325,6 +325,69 @@ mod namespaced {
         Ok(())
     }
 
+    #[tokio::test]
+    async fn service_waypoint_workload_hostname() -> anyhow::Result<()> {
+        let mut manager = setup_netns_test!(Shared);
+
+        let zt = manager.deploy_ztunnel(DEFAULT_NODE).await?;
+
+        let waypoint = manager
+            .workload_builder("waypoint", DEFAULT_NODE)
+            .uncaptured()
+            .mutate_workload(|w| w.hostname = "waypoint.example.com".into())
+            .register()
+            .await?;
+        run_hbone_server(waypoint, "waypoint")?;
+
+        manager
+            .workload_builder("server", DEFAULT_NODE)
+            .waypoint_hostname("waypoint.example.com")
+            .register()
+            .await?;
+        let client = manager
+            .workload_builder("client", DEFAULT_NODE)
+            .register()
+            .await?;
+
+        let server_ip = manager.resolver().resolve("server")?;
+        let waypoint_pod_ip = manager.resolver().resolve("waypoint")?;
+        run_tcp_to_hbone_client(client, manager.resolver(), "server")?;
+
+        let metrics = [
+            (CONNECTIONS_OPENED, 1),
+            (CONNECTIONS_CLOSED, 1),
+            (BYTES_RECV, REQ_SIZE),
+            (BYTES_SENT, HBONE_REQ_SIZE),
+        ];
+        verify_metrics(&zt, &metrics, &source_labels()).await;
+
+        let sent = format!("{REQ_SIZE}");
+        let recv = format!("{HBONE_REQ_SIZE}");
+        let hbone_addr = format!("{server_ip}:8080");
+        let dst_addr = format!("{waypoint_pod_ip}:15008");
+        let want = HashMap::from([
+            ("scope", "access"),
+            ("src.workload", "client"),
+            ("dst.workload", "waypoint"),
+            ("dst.hbone_addr", &hbone_addr),
+            ("dst.addr", &dst_addr),
+            ("bytes_sent", &sent),
+            ("bytes_recv", &recv),
+            ("direction", "outbound"),
+            ("message", "connection complete"),
+            (
+                "src.identity",
+                "spiffe://cluster.local/ns/default/sa/client",
+            ),
+            (
+                "dst.identity",
+                "spiffe://cluster.local/ns/default/sa/waypoint",
+            ),
+        ]);
+        telemetry::testing::assert_contains(want);
+        Ok(())
+    }
+
     #[tokio::test]
     async fn sandwich_waypoint_plain() -> anyhow::Result<()> {
         let mut manager = setup_netns_test!(Shared);