code: def test_collections_get(client, optimize_graph, s, a, b):
      da = pytest.importorskip("dask.array")
  
      async def f(dask_worker):
          await dask_worker.set_resources(**{"A": 1})
  
      client.run(f, workers=[a["address"]])
  
      with dask.annotate(resources={"A": 1}):
          x = da.random.random(100, chunks=(10,)) + 1
  
      x.compute(optimize_graph=optimize_graph)
  
      def g(dask_worker):
          return len(dask_worker.log)
  
      logs = client.run(g)
      assert logs[a["address"]]
      assert not logs[b["address"]]
  
file_path: distributed/distributed/tests/test_resources.py
imports: from __future__ import annotations
  import asyncio
  import pytest
  import dask
  from dask import delayed
  from distributed import Lock, Worker
  from distributed.client import wait
  from distributed.utils_test import NO_AMM, gen_cluster, inc, lock_inc, slowadd, slowinc
  from distributed.worker_state_machine import ComputeTaskEvent, Execute, ExecuteFailureEvent, ExecuteSuccessEvent, FreeKeysEvent, LongRunningMsg, RescheduleEvent, TaskFinishedMsg
label_desc: Use image node_id 5 for calling a global function with example usage: test_collections_get(client, optimize_graph, s, a, b) and returns: len