By default, Dask already copies the exception and traceback wherever they
occur and reraises that exception locally.  If your task failed with a
``ZeroDivisionError`` remotely, then you'll get a ``ZeroDivisionError`` in your
interactive session.  Similarly you'll see a full traceback of where this error
occurred, which, just like in normal Python, can help you to identify the
troublesome spot in your code.

However, you cannot use the ``pdb`` module or ``%debug`` IPython magics with
these tracebacks to look at the value of variables during failure.  You can
only inspect things visually.  Additionally, the top of the traceback may be
filled with functions that are Dask-specific and not relevant to your
problem, so you can safely ignore these.

Both the single-machine and distributed schedulers do this.


Use the Single-Threaded Scheduler

>>> x.compute(scheduler='single-threaded')

If a remote task fails, we can collect the function and all inputs, bring them
to the local thread, and then rerun the function in hopes of triggering the
same exception locally where normal debugging tools can be used.

With the single-machine schedulers, use the ``rerun_exceptions_locally=True``
keyword:

.. code-block:: python

   >>> x.compute(rerun_exceptions_locally=True)

On the distributed scheduler use the ``recreate_error_locally`` method on
anything that contains ``Futures``:

.. code-block:: python

   >>> x.compute()
   ZeroDivisionError(...)

   >>> %pdb
   >>> future = client.compute(x)
   >>> client.recreate_error_locally(future)


Remove Failed Futures Manually

First, the distributed scheduler has a number of `diagnostic tools
<https://distributed.dask.org/en/latest/diagnosing-performance.html>`_ showing dozens of
recorded metrics like CPU, memory, network, and disk use, a history of previous
tasks, allocation of tasks to workers, worker memory pressure, work stealing,
open file handle limits, etc.  *Many* problems can be correctly diagnosed by
inspecting these pages.  By default, these are available at
``http://scheduler:8787/`` where ``scheduler`` should be replaced by the address of the
scheduler. See `diagnosing performance docs
<https://distributed.dask.org/en/latest/diagnosing-performance.html>`_ for more information.

Logs
~~~~

The scheduler, workers, and client all emits logs using `Python's standard
logging module <https://docs.python.org/3/library/logging.html>`_.  By default,
these emit to standard error.  When Dask is launched by a cluster job scheduler
(SGE/SLURM/YARN/Mesos/Marathon/Kubernetes/whatever), that system will track
these logs and will have an interface to help you access them.  If you are
launching Dask on your own, they will probably dump to the screen unless you
`redirect stderr to a file
<https://en.wikipedia.org/wiki/Redirection_(computing)#Redirecting_to_and_from_the_standard_file_handles>`_
.

You can control the logging verbosity in the :doc:`../configuration`, for example,
the ``~/.config/dask/*.yaml`` files.
Defaults currently look like the following:

.. code-block:: yaml

   logging:
     distributed: info
     distributed.client: warning
     bokeh: error

Logging for specific components like ``distributed.client``,  ``distributed.scheduler``,
``distributed.nanny``,  ``distributed.worker``, etc. can each be independently configured.
So, for example, you could add a line like ``distributed.worker: debug`` to get
*very* verbose output from the workers.

Furthermore, you can explicitly assign handlers to loggers. The following example
assigns both file ("output.log") and console output to the scheduler and workers.
See the `python logging`_ documentation for information on the meaning of
specific terms here.

.. code-block:: yaml

    logging:
      version: 1
      handlers:
        file:
          class: logging.handlers.RotatingFileHandler
          filename: output.log
          level: INFO
        console:
          class: logging.StreamHandler
          level: INFO
      loggers:
        distributed.worker:
          level: INFO
          handlers:
            - file
            - console
        distributed.scheduler:
          level: INFO
          handlers:
            - file
            - console

.. _python logging: https://docs.python.org/3/library/logging.html


LocalCluster
------------

If you are using the distributed scheduler from a single machine, you may be
setting up workers manually using the command line interface or you may be
using `LocalCluster <https://distributed.dask.org/en/latest/api.html#cluster>`_
which is what runs when you just call ``Client()``:

.. code-block:: python

   >>> from dask.distributed import Client, LocalCluster
   >>> client = Client()  # This is actually the following two commands

   >>> cluster = LocalCluster()
   >>> client = Client(cluster.scheduler.address)

LocalCluster is useful because the scheduler and workers are in the same
process with you, so you can easily inspect their `state
<https://distributed.dask.org/en/latest/scheduling-state.html>`_ while
they run (they are running in a separate thread):

.. code-block:: python

   >>> cluster.scheduler.processing
   {'worker-one:59858': {'inc-123', 'add-443'},
    'worker-two:48248': {'inc-456'}}

You can also do this for the workers *if* you run them without nanny processes:

.. code-block:: python

   >>> cluster = LocalCluster(nanny=False)
   >>> client = Client(cluster)

This can be very helpful if you want to use the Dask distributed API and still
want to investigate what is going on directly within the workers.  Information
is not distilled for you like it is in the web diagnostics, but you have full
low-level access.