code: def task_node_filter(task_names):
      def _task_node_filter():
          if task_names == []:
              return lambda _: True
  
          tasks = StateApiClient().list(
              StateResource.TASKS,
              options=ListApiOptions(),
              raise_on_missing_output=False,
          )
          filtered_tasks = list(
              filter(lambda task: task.name in task_names, tasks)
          )
          nodes_with_filtered_tasks = {
              task.node_id for task in filtered_tasks
          }
  
          def _filter_fn(node):
              return node["NodeID"] in nodes_with_filtered_tasks
  
          return _filter_fn
  
      return _task_node_filter
  
file_path: ray/release/nightly_tests/setup_chaos.py
imports: import argparse
  from ray.util.state.api import StateApiClient
  from ray.util.state.common import ListApiOptions, StateResource
  import ray
  from ray._private.test_utils import get_and_run_resource_killer, NodeKillerActor, WorkerKillerActor
label_desc: Use image node_id 3 for calling a global function with example usage: task_node_filter(task_names) and returns: _task_node_filter, _filter_fn, function, unknown