Repository: dcs4cop/xcube (python)

Issue:
Enable distributed cube computing
**Is your feature request related to a problem? Please describe.**

`xcube serve` supports datasets that are computed on-the-fly, e.g. applying temporal aggregation. This requires high data throughput and CPU resources, therefore requests based on computed datasets often respond too slow and then time out. Such computations may be much faster if computation is distributed on a dedicated cluster.

`xcube gen` may be parallelized so that individual input files (usually spatial time slices) are transformed in a distributed way and then results are combined chronologically into the desired cube.

The same may apply to other xcube commands that perform heavy computing on chunks of data.

**Describe the solution you'd like**

Add option to xcube CLI commands that configure how the command is executed in a distributed manner. The option is TBD.

**Describe alternatives you've considered**

None, besides using larger machines.

**Additional context**

As we use xarray and xarray uses dask, the solution should be based on [Dask Distributed](http://distributed.dask.org). 


Pull request description:
Dask Distributed
This PR closes #58 and incorporates the following changes:

* Added global `xcube --scheduler <scheduler>` option for Dask distributed computing (#58)
* Added global `xcube --traceback` option, removed local `xcube gen --traceback` option
* Making use of common CLI options via new @cli_option_traceback and @cli_option_scheduler decorators in new `xcube.util.cliutil` module.

Regarding the latter point, we should follow this pattern to reuse common CLI arguments and options such as `--input`, `--output`, `--format`, etc! This pattern **reduces code duplication** and makes the CLI **consistent** across sub-commands. --> update Dev Guide.

Patch:
diff --git a/CHANGES.md b/CHANGES.md
index 69ba1cd2..b3b64655 100644
--- a/CHANGES.md
+++ b/CHANGES.md
@@ -4,6 +4,8 @@
 
 * Restructured and clarified code base (#27)
 * Moved to Python 3.7 (#25)
+* Added global `xcube --scheduler <scheduler>` option for Dask distributed computing (#58)
+* Added global `xcube --traceback` option, removed local `xcube gen --traceback` option
 * Completed version 1 of an xcube developer guide.
 * Added `xcube serve` command (#43) 
 * `xcube serve`: Time-series web API now also returns "uncertainty" (#48)
diff --git a/README.md b/README.md
index 81755acd..3519d7d3 100644
--- a/README.md
+++ b/README.md
@@ -83,12 +83,26 @@ To start a demo using docker use the following commands
     $ xcube --help
     Usage: xcube [OPTIONS] COMMAND [ARGS]...
     
-      Xcube Toolkit
+    Xcube Toolkit
     
     Options:
-      --version  Show the version and exit.
-      --help     Show this message and exit.
-    
+      --version                Show the version and exit.
+      --traceback              Enable tracing back errors by dumping the Python
+                               call stack. Pass as very first option to also trace
+                               back error during command-line validation.
+
+      --scheduler <scheduler>  Enable distributed computing using the Dask
+                               scheduler identified by <scheduler>. <scheduler>
+                               can have the form <address>?<keyword>=<value>,...
+                               where <address> is <host> or <host>:<port> and
+                               specifies the scheduler's address in your network.
+                               For more information on distributed computing using
+                               Dask, refer to http://distributed.dask.org/. Pairs
+                               of <keyword>=<value> are passed to the Dask client.
+                               Refer to http://distributed.dask.org/en/latest/api.
+                               html#distributed.Client
+      --help                   Show this message and exit.
+
     Commands:
       chunk     (Re-)chunk dataset.
       dump      Dump contents of a dataset.
@@ -96,8 +110,9 @@ To start a demo using docker use the following commands
       gen       Generate data cube.
       grid      Find spatial data cube resolutions and adjust bounding boxes.
       level     Generate multi-resolution levels.
-      server    Serve data cubes via web service.
+      serve     Serve data cubes via web service.
       vars2dim  Convert cube variables into new dimension.
+    
 
 
 ## `xcube chunk`
diff --git a/xcube/cli/cli.py b/xcube/cli/cli.py
index 6052b04d..bcf6a795 100644
--- a/xcube/cli/cli.py
+++ b/xcube/cli/cli.py
@@ -1,30 +1,17 @@
-from typing import Any, Dict
+import sys
 
 import click
 
 from xcube.cli.gen import gen
-from xcube.cli.serve import serve
 from xcube.cli.grid import grid
+from xcube.cli.serve import serve
+from xcube.util.cliutil import parse_cli_kwargs, new_cli_ctx_obj, handle_cli_exception, cli_option_traceback, \
+    cli_option_scheduler
 from xcube.version import version
 
 DEFAULT_TILE_SIZE = 512
 
 
-def _parse_kwargs(value: str, metavar: str = None) -> Dict[str, Any]:
-    if value:
-        try:
-            return eval(f"dict({value})", {}, {})
-        except Exception:
-            if metavar:
-                message = f"Invalid value for {metavar}: {value!r}"
-            else:
-                message = f"Invalid value: {value!r}"
-            raise click.ClickException(message)
-    else:
-        return dict()
-
-
-# noinspection PyShadowingBuiltins
 @click.command(name="extract")
 @click.argument('cube', metavar='<cube>')
 @click.argument('coords', metavar='<coords>')
@@ -85,7 +72,7 @@ def chunk(input, output, format=None, params=None, chunks=None):
     """
     chunk_sizes = None
     if chunks:
-        chunk_sizes = _parse_kwargs(chunks, metavar="<chunks>")
+        chunk_sizes = parse_cli_kwargs(chunks, metavar="<chunks>")
         for k, v in chunk_sizes.items():
             if not isinstance(v, int) or v <= 0:
                 raise click.ClickException("Invalid value for <chunks>, "
@@ -93,7 +80,7 @@ def chunk(input, output, format=None, params=None, chunks=None):
 
     write_kwargs = dict()
     if params:
-        write_kwargs = _parse_kwargs(params, metavar="<params>")
+        write_kwargs = parse_cli_kwargs(params, metavar="<params>")
 
     from xcube.util.dsio import guess_dataset_format
     format_name = format if format else guess_dataset_format(output)
@@ -111,6 +98,7 @@ def chunk(input, output, format=None, params=None, chunks=None):
         write_dataset(chunked_dataset, output_path=output, format_name=format_name, **write_kwargs)
 
 
+# noinspection PyShadowingBuiltins
 @click.command(name="level")
 @click.argument('input', metavar='<input>')
 @click.option('--output', '-o', metavar='<output>',
@@ -198,15 +186,15 @@ def dump(dataset, variable, encoding):
         print(text)
 
 
-# noinspection PyShadowingBuiltins
+# noinspection PyShadowingBuiltins,PyUnusedLocal
 @click.command(name="vars2dim")
 @click.argument('cube', metavar='<cube>')
-@click.option('--dim_name', '-d', metavar='<dim-name>',
-              default='var',
-              help='Name of the new dimension into variables. Defaults to "var".')
 @click.option('--var_name', '-v', metavar='<var-name>',
               default='data',
               help='Name of the new variable that includes all variables. Defaults to "data".')
+@click.option('--dim_name', '-d', metavar='<dim-name>',
+              default='var',
+              help='Name of the new dimension into variables. Defaults to "var".')
 @click.option('--output', '-o', metavar='<output>',
               help="Output file.")
 @click.option('--format', '-f', metavar='<format>', type=click.Choice(['zarr', 'netcdf']),
@@ -235,10 +223,12 @@ def vars2dim(cube, var_name, dim_name, output=None, format=None):
         write_dataset(converted_dataset, output_path=output, format_name=format_name)
 
 
-# noinspection PyShadowingBuiltins
+# noinspection PyShadowingBuiltins,PyUnusedLocal
 @click.group()
 @click.version_option(version)
-def cli():
+@cli_option_traceback
+@cli_option_scheduler
+def cli(traceback=False, scheduler=None):
     """
     Xcube Toolkit
     """
@@ -255,7 +245,13 @@ cli.add_command(serve)
 
 
 def main(args=None):
-    cli.main(args=args)
+    # noinspection PyBroadException
+    ctx_obj = new_cli_ctx_obj()
+    try:
+        exit_code = cli.main(args=args, obj=ctx_obj, standalone_mode=False)
+    except Exception as e:
+        exit_code = handle_cli_exception(e, traceback_mode=ctx_obj.get("traceback", False))
+    sys.exit(exit_code)
 
 
 if __name__ == '__main__':
diff --git a/xcube/cli/gen.py b/xcube/cli/gen.py
index c918df94..2dcdfbd4 100644
--- a/xcube/cli/gen.py
+++ b/xcube/cli/gen.py
@@ -20,18 +20,15 @@
 # SOFTWARE.
 
 import click
-import sys
-import traceback
 
-from xcube.util.dsio import query_dataset_io
 from xcube.api.gen.config import get_config_dict
 from xcube.api.gen.defaults import DEFAULT_OUTPUT_DIR, DEFAULT_OUTPUT_NAME, \
     DEFAULT_OUTPUT_RESAMPLING, DEFAULT_OUTPUT_FORMAT
-from xcube.api.gen.iproc import InputProcessor
 from xcube.api.gen.gen import gen_cube
+from xcube.api.gen.iproc import InputProcessor
+from xcube.util.dsio import query_dataset_io
 from xcube.util.objreg import get_obj_registry
 from xcube.util.reproject import NAME_TO_GDAL_RESAMPLE_ALG
-from xcube.version import version
 
 input_processor_names = [input_processor.name
                          for input_processor in get_obj_registry().get_all(type=InputProcessor)]
@@ -40,7 +37,6 @@ resampling_algs = NAME_TO_GDAL_RESAMPLE_ALG.keys()
 
 
 @click.command(name='gen', context_settings={"ignore_unknown_options": True})
-@click.version_option(version)
 @click.argument('input_files', metavar='INPUT_FILES', nargs=-1)
 @click.option('--proc', '-p', metavar='INPUT_PROCESSOR', type=click.Choice(input_processor_names),
               help=f'Input processor type name. '
@@ -72,8 +68,6 @@ resampling_algs = NAME_TO_GDAL_RESAMPLE_ALG.keys()
               help='Fallback spatial resampling algorithm to be used for all variables. '
               f'Defaults to {DEFAULT_OUTPUT_RESAMPLING!r}. '
               f'The choices for the resampling algorithm are: {resampling_algs}')
-@click.option('--traceback', default=False, is_flag=True,
-              help='On error, print Python traceback.')
 @click.option('--append', '-a', default=False, is_flag=True,
               help='Append successive outputs.')
 @click.option('--dry_run', default=False, is_flag=True,
@@ -93,7 +87,6 @@ def gen(input_files: str,
         region: str,
         variables: str,
         resampling: str,
-        traceback: bool,
         append: bool,
         dry_run: bool,
         info: bool,
@@ -113,7 +106,6 @@ def gen(input_files: str,
     output_region = region
     output_variables = variables
     output_resampling = resampling
-    traceback_mode = traceback
     append_mode = append
     dry_run = dry_run
     info_mode = info
@@ -123,30 +115,17 @@ def gen(input_files: str,
         print(_format_info())
         return 0
 
-    try:
-        config = get_config_dict(locals())
-    except ValueError as e:
-        return _handle_error(e, traceback_mode)
+    config = get_config_dict(locals())
 
-    try:
-        gen_cube(append_mode=append_mode,
-                 dry_run=dry_run,
-                 monitor=print,
-                 sort_mode=sort_mode,
-                 **config)
-    except Exception as e:
-        return _handle_error(e, traceback_mode)
+    gen_cube(append_mode=append_mode,
+             dry_run=dry_run,
+             monitor=print,
+             sort_mode=sort_mode,
+             **config)
 
     return 0
 
 
-def _handle_error(e, traceback_mode):
-    print(f'error: {e}', file=sys.stderr)
-    if traceback_mode:
-        traceback.print_exc(file=sys.stderr)
-    return 2
-
-
 def _format_info():
     # noinspection PyUnresolvedReferences
     input_processors = get_obj_registry().get_all(type=InputProcessor)
diff --git a/xcube/util/cliutil.py b/xcube/util/cliutil.py
new file mode 100644
index 00000000..b83c5974
--- /dev/null
+++ b/xcube/util/cliutil.py
@@ -0,0 +1,110 @@
+from typing import Dict, Any, Optional
+
+import click
+
+
+def new_cli_ctx_obj():
+    return {
+        "traceback": False,
+        "scheduler": None,
+    }
+
+
+def cli_option_traceback(func):
+    """Decorator for adding a pre-defined, reusable CLI option `--traceback`."""
+
+    # noinspection PyUnusedLocal
+    def _callback(ctx: click.Context, param: click.Option, value: bool):
+        ctx_obj = ctx.ensure_object(dict)
+        if ctx_obj is not None:
+            ctx_obj["traceback"] = value
+        return value
+
+    return click.option(
+        '--traceback',
+        is_flag=True,
+        help="Enable tracing back errors by dumping the Python call stack. "
+             "Pass as very first option to also trace back error during command-line validation.",
+        callback=_callback)(func)
+
+
+def cli_option_scheduler(func):
+    """Decorator for adding a pre-defined, reusable CLI option `--scheduler`."""
+
+    # noinspection PyUnusedLocal
+    def _callback(ctx: click.Context, param: click.Option, value: Optional[str]):
+        if not value:
+            return
+
+        address_and_kwargs = value.split("?", 2)
+        if len(address_and_kwargs) == 2:
+            address, kwargs_string = address_and_kwargs
+            kwargs = parse_cli_kwargs(kwargs_string, metavar="<scheduler>")
+        else:
+            address, = address_and_kwargs
+            kwargs = dict()
+
+        try:
+            # The Dask Client registers itself as the default Dask scheduler, and so runs dask.array used by xarray
+            import distributed
+            scheduler_client = distributed.Client(address, **kwargs)
+            ctx_obj = ctx.ensure_object(dict)
+            if ctx_obj is not None:
+                ctx_obj["scheduler"] = scheduler_client
+            return scheduler_client
+        except ValueError as e:
+            raise click.BadParameter(f'Failed to create Dask scheduler client: {e}') from e
+
+    return click.option(
+        '--scheduler',
+        metavar='<scheduler>',
+        help="Enable distributed computing using the Dask scheduler identified by <scheduler>. "
+             "<scheduler> can have the form <address>?<keyword>=<value>,... where <address> "
+             "is <host> or <host>:<port> and specifies the scheduler's address in your network. "
+             "For more information on distributed computing "
+             "using Dask, refer to http://distributed.dask.org/. "
+             "Pairs of <keyword>=<value> are passed to the Dask client. "
+             "Refer to http://distributed.dask.org/en/latest/api.html#distributed.Client",
+        callback=_callback)(func)
+
+
+def parse_cli_kwargs(value: str, metavar: str = None) -> Dict[str, Any]:
+    """
+    Parse a string value of the form [<kw>=<arg>{,<kw>=<arg>}] into a dictionary.
+    <kw> must be a valid Python identifier, <arg> must be a Python literal.
+
+    :param value: A string value.
+    :param metavar: Name of a meta-variable used in CLI.
+    :return: a dictionary of keyword-arguments
+    """
+    if value:
+        try:
+            return eval(f"dict({value})", {}, {})
+        except Exception:
+            if metavar:
+                message = f"Invalid value for {metavar}: {value!r}"
+            else:
+                message = f"Invalid value: {value!r}"
+            raise click.ClickException(message)
+    else:
+        return dict()
+
+
+def handle_cli_exception(e: BaseException, exit_code: int = None, traceback_mode: bool = False) -> int:
+    import sys
+    if isinstance(e, click.Abort):
+        print(f'Aborted!')
+        exit_code = exit_code or 1
+    elif isinstance(e, click.ClickException):
+        e.show(file=sys.stderr)
+        exit_code = exit_code or e.exit_code
+    elif isinstance(e, OSError):
+        print(f'OS error: {e}', file=sys.stderr)
+        exit_code = exit_code or 2
+    else:
+        print(f'Internal error: {e}', file=sys.stderr)
+        exit_code = exit_code or 3
+    if traceback_mode:
+        import traceback
+        traceback.print_exc(file=sys.stderr)
+    return exit_code


Test patch:
diff --git a/test/cli/test_cli.py b/test/cli/test_cli.py
index a5ed343b..8ad5c624 100644
--- a/test/cli/test_cli.py
+++ b/test/cli/test_cli.py
@@ -6,8 +6,9 @@ from typing import List
 import click
 import click.testing
 import xarray as xr
+
 from xcube.api.new import new_cube
-from xcube.cli.cli import cli, _parse_kwargs
+from xcube.cli.cli import cli
 from xcube.util.dsio import rimraf
 
 TEST_NC_FILE = "test.nc"
@@ -258,22 +259,3 @@ class LevelTest(CliTest):
         self.assertEqual("Error: <num-levels-max> must be a positive integer\n",
                          result.output)
         self.assertEqual(1, result.exit_code)
-
-
-class ParseTest(unittest.TestCase):
-
-    def test_parse_kwargs(self):
-        self.assertEqual(dict(),
-                         _parse_kwargs("", metavar="<chunks>"))
-        self.assertEqual(dict(time=1, lat=256, lon=512),
-                         _parse_kwargs("time=1, lat=256, lon=512", metavar="<chunks>"))
-
-        with self.assertRaises(click.ClickException) as cm:
-            _parse_kwargs("45 * 'A'", metavar="<chunks>")
-        self.assertEqual("Invalid value for <chunks>: \"45 * 'A'\"",
-                         f"{cm.exception}")
-
-        with self.assertRaises(click.ClickException) as cm:
-            _parse_kwargs("9==2")
-        self.assertEqual("Invalid value: '9==2'",
-                         f"{cm.exception}")
diff --git a/test/util/test_cliutil.py b/test/util/test_cliutil.py
new file mode 100644
index 00000000..e3a11194
--- /dev/null
+++ b/test/util/test_cliutil.py
@@ -0,0 +1,34 @@
+import unittest
+
+import click
+
+from xcube.util.cliutil import parse_cli_kwargs, handle_cli_exception
+
+
+class ClickUtilTest(unittest.TestCase):
+
+    def test_parse_cli_kwargs(self):
+        self.assertEqual(dict(),
+                         parse_cli_kwargs("", metavar="<chunks>"))
+        self.assertEqual(dict(time=1, lat=256, lon=512),
+                         parse_cli_kwargs("time=1, lat=256, lon=512", metavar="<chunks>"))
+
+        with self.assertRaises(click.ClickException) as cm:
+            parse_cli_kwargs("45 * 'A'", metavar="<chunks>")
+        self.assertEqual("Invalid value for <chunks>: \"45 * 'A'\"",
+                         f"{cm.exception}")
+
+        with self.assertRaises(click.ClickException) as cm:
+            parse_cli_kwargs("9==2")
+        self.assertEqual("Invalid value: '9==2'",
+                         f"{cm.exception}")
+
+    def test_handle_cli_exception(self):
+        self.assertEqual(1, handle_cli_exception(click.Abort(), traceback_mode=True))
+        self.assertEqual(1, handle_cli_exception(click.ClickException("bad handle"), traceback_mode=True))
+        self.assertEqual(2, handle_cli_exception(OSError("bad handle"), traceback_mode=True))
+        self.assertEqual(3, handle_cli_exception(ValueError("bad handle"), traceback_mode=True))
+        self.assertEqual(20, handle_cli_exception(click.Abort(), exit_code=20))
+        self.assertEqual(20, handle_cli_exception(click.ClickException("bad handle"), exit_code=20))
+        self.assertEqual(20, handle_cli_exception(OSError("bad handle"), exit_code=20))
+        self.assertEqual(20, handle_cli_exception(ValueError("bad handle"), exit_code=20))