a.serviceEntryWorkloadBuilder(MeshConfig, AuthorizationPolicies, PeerAuths, Waypoints, Namespaces),
		krt.WithName("ServiceEntryWorkloads"),
	)
	Workloads := krt.JoinCollection([]krt.Collection[model.WorkloadInfo]{PodWorkloads, WorkloadEntryWorkloads, ServiceEntryWorkloads}, krt.WithName("Workloads"))
	return Workloads
}

func (a *index) workloadEntryWorkloadBuilder(
	MeshConfig krt.Singleton[MeshConfig],

	// TODO: we should be able to handle multiple IPs or a hostname
	if len(key.Addresses) == 0 {
		return nil
	}
	workloads := a.workloads.ByOwningWaypoint.Lookup(networkAddress{
		network: key.Network,
		ip:      key.Addresses[0],
	})
	workloads = model.SortWorkloadsByCreationTime(workloads)
	return workloads
}

func (a *index) AdditionalPodSubscriptions(
	proxy *model.Proxy,
	allAddresses sets.String,

  istioctl ztunnel-config workloads --file ztunnel-config.json

  # Retrieve workload summary for a specific namespace
  istioctl ztunnel-config workloads <ztunnel-name[.namespace]> --workloads-namespace foo
`,
		Aliases: []string{"w", "workloads"},
		Args:    common.validateArgs,
		RunE: runConfigDump(ctx, common, func(cw *ztunnelDump.ConfigWriter) error {

	req, err := http.NewRequest("GET", "/metrics/resources", nil)
	if err != nil {
		t.Fatal(err)
	}
	h.ServeHTTP(r, req)

	expected := `# HELP kube_pod_resource_limit [STABLE] Resources limit for workloads on the cluster, broken down by pod. This shows the resource usage the scheduler and kubelet expect per pod for resources along with the unit for the resource if any.
# TYPE kube_pod_resource_limit gauge

		ip: "10.10.10.30", port: 9090,
		metaNetwork: "vm",
	}
	// VM always sees itself directly
	vm.Expect(vm, "10.10.10.30:9090")

	workloads := []*workload{pod, vm}

	var kubeObjects []runtime.Object
	var configObjects []config.Config
	for _, w := range workloads {
		_, objs := w.kubeObjects()
		kubeObjects = append(kubeObjects, objs...)
		configObjects = append(configObjects, w.configs()...)
	}

}

func verifyThatRequestWasTunneled(target echo.Instance, expectedSourceIPs []corev1.PodIP, expectedPath string) error {
	workloads, err := target.Workloads()
	if err != nil {
		return fmt.Errorf("failed to get workloads of %s: %s", target.ServiceName(), err)
	}
	var logs strings.Builder
	for _, w := range workloads {
		workloadLogs, err := w.Logs()
		if err != nil {

		// Nothing to do.
		return o.ProxyProtocolVersion
	}
	return 0
}

func (o *CallOptions) numWorkloads() int {
	if o.To == nil {
		return 0
	}
	workloads, err := o.To.Workloads()
	if err != nil {
		return 0
	}
	return len(workloads)
}

func (o *CallOptions) fillConnectionParams() {
	// Overrides connection parameters for scheme.
	switch o.Scheme {
	case scheme.DNS:

					WorkloadType:      workloadapi.WorkloadType_POD,
					WorkloadName:      "name0",
					ClusterId:         testC,
				},
			},
		},
	}})

	// test code path where service entry selects workloads via `ServiceEntry.workloadSelector`
	s.addPods(t, "140.140.0.10", "pod1", "sa1", map[string]string{"app": "a"}, nil, true, corev1.PodRunning)
	s.assertEvent(t, s.podXdsName("pod1"))

Most of the API is fairly straight forward.
However, one interesting aspect is how these policies associate with workloads.
Istio's AuthorizationPolicy has label selectors.
However, we intentionally do not send those as part of the Workload API, in order to keep the size low.

The obvious solution to this is to put the list of selected workloads into the policy itself.
However, this means anytime a workload changes (often), we need to update the policy.

    // This could be a single Pod, a VM instance, etc.
    Workload workload = 1;
    // Service represents a service - a group of workloads that can be accessed together.
    Service service = 2;
  }
}

// Service represents a service - a group of workloads that can be accessed together.
// The xds primary key is "namespace/hostname".