- watches k8s resource for existing pods, so that pods that have already been started can be moved in or out of the ambient mesh.
- sends UDS events to ztunnel via a socket whenever a pod is enabled for ambient mesh (whether from CNI plugin or node watcher), instructing ztunnel to create the "tube" socket.

		// finishes as soon as watch is initialized (which is generally orders of
		// magnitude faster then the watch request itself). This means that Handle()
		// call finishes much faster and for performance reasons we want to reduce
		// the number of running goroutines - so we run the shorter thing in a
		// dedicated goroutine and the actual watch handler in the main one.
		go func() {
			defer func() {

						},
					},
					Disabled: wrapperspb.Bool(true),
				},
			},
		},
	})

	env := model.NewEnvironment()
	env.ServiceDiscovery = serviceDiscovery
	env.ConfigStore = configStore
	env.Watcher = mesh.NewFixedWatcher(meshConfig)

	pushContext := model.NewPushContext()
	env.Init()
	pushContext.InitContext(env, nil, nil)
	env.SetPushContext(pushContext)

	return env
}

var (

func (c *Controller) onNetworkChange() {
	// the network for endpoints are computed when we process the events; this will fix the cache
	// NOTE: this must run before the other network watcher handler that creates a force push
	if err := c.syncPods(); err != nil {
		log.Errorf("one or more errors force-syncing pods: %v", err)
	}
	if err := c.endpoints.initializeNamespace(metav1.NamespaceAll, true); err != nil {

		filter kclient.Filter
		want   kubetypes.InformerOptions
	}{
		{
			name: "watch pods in the foo namespace",
			gvr:  gvr.Pod,
			filter: kclient.Filter{
				Namespace: "foo",
			},
			want: kubetypes.InformerOptions{
				Namespace: "foo",
				Cluster:   c.ClusterID(),
			},
		},
		{
			name: "watch pods in the InformerWatchNamespace",
			gvr:  gvr.Pod,
			want: kubetypes.InformerOptions{

// Controller is a collection of synchronized resource watchers
// Caches are thread-safe
type Controller struct {
	opts Options

	client kubelib.Client

	queue queue.Instance

	namespaces kclient.Client[*v1.Namespace]
	services   kclient.Client[*v1.Service]

	endpoints *endpointSliceController

	// Used to watch node accessible from remote cluster.

		t.Run(tc.name, func(t *testing.T) {
			matcher := PolicyMatcherFor(tc.workloadNamespace, tc.workloadLabels, tc.isWaypoint)
			if got := policies.GetJwtPoliciesForWorkload(matcher); !reflect.DeepEqual(tc.wantRequestAuthn, got) {
				t.Fatalf("want %+v\n, but got %+v\n", printConfigs(tc.wantRequestAuthn), printConfigs(got))
			}
			if got := policies.GetPeerAuthenticationsForWorkload(matcher); !reflect.DeepEqual(tc.wantPeerAuthn, got) {

}

const maxDurationSecondsForLog = 5

func (sys *IAMSys) periodicRoutines(ctx context.Context, baseInterval time.Duration) {
	// Watch for IAM config changes for iamStorageWatcher.
	watcher, isWatcher := sys.store.IAMStorageAPI.(iamStorageWatcher)
	if isWatcher {
		go func() {
			ch := watcher.watch(ctx, iamConfigPrefix)
			for event := range ch {
				if err := sys.loadWatchedEvent(ctx, event); err != nil {
					// we simply log errors

	}
	m := mesh.DefaultMeshConfig()

	m.ExtensionProviders = append(m.ExtensionProviders, jsonTextProvider, textFormattersProvider, jsonFormattersProvider)

	environment := &Environment{
		ConfigStore: store,
		Watcher:     mesh.NewFixedWatcher(m),
	}
	telemetries := getTelemetries(environment)

	ctx := NewPushContext()
	ctx.Mesh = m
	return telemetries, ctx
}

		// annotation is already set, nothing to do
		return claim, nil
	}

	// The volume from method args can be pointing to watcher cache. We must not
	// modify these, therefore create a copy.
	claimClone := claim.DeepCopy()
	// TODO: remove the beta storage provisioner anno after the deprecation period
	logger := klog.FromContext(ctx)