Repository: mlrun/mlrun

Issue:
Move HTTP routes under prefix
Move the HTTP DB routes under a prefix (such as `/api/v1`). We'd like to be able to add UI on top of it as well

Discussion:

Patch:
diff --git a/mlrun/__init__.py b/mlrun/__init__.py
index 16c968e89..efde0b6fd 100644
--- a/mlrun/__init__.py
+++ b/mlrun/__init__.py
@@ -20,5 +20,5 @@ from .model import RunTemplate, NewRun, NewTask, RunObject
 from .kfpops import mlrun_op
 from .config import config as mlconf
 from .runtimes import new_model_server
-from .platforms import mount_v3io
+from .platforms import mount_v3io, v3io_cred
 from .datastore import get_object
diff --git a/mlrun/__main__.py b/mlrun/__main__.py
index 236f5f9de..8b35aa7ac 100644
--- a/mlrun/__main__.py
+++ b/mlrun/__main__.py
@@ -181,11 +181,12 @@ def build(dest, command, source, base_image, secret_name,
 @click.option('--dashboard', '-d', default='', help='nuclio dashboard url')
 @click.option('--project', '-p', default='', help='container registry secret name')
 @click.option('--model', '-m', multiple=True, help='input artifact')
-@click.option('--kind', '-k', help='runtime sub kind')
+@click.option('--kind', '-k', default=None, help='runtime sub kind')
 @click.option('--tag', default='', help='version tag')
+@click.option('--env', '-e', multiple=True, help='environment variables')
 @click.option('--verbose', is_flag=True, help='verbose log')
-def deploy(spec, source, dashboard, project, model, tag, kind, verbose):
-    """Deploy model"""
+def deploy(spec, source, dashboard, project, model, tag, kind, env, verbose):
+    """Deploy model or function"""
     if spec:
         runtime = py_eval(spec)
     else:
@@ -198,13 +199,16 @@ def deploy(spec, source, dashboard, project, model, tag, kind, verbose):
     f.spec.source = source
     if kind:
         f.spec.function_kind = kind
+    if env:
+        for k, v in list2dict(env).items():
+            f.set_env(k, v)
     f.verbose = verbose
     if model:
         models = list2dict(model)
         for k, v in models.items():
             f.add_model(k, v)
 
-    addr = f.deploy(dashboard=dashboard, project=project, tag=tag)
+    addr = f.deploy(dashboard=dashboard, project=project, tag=tag, kind=kind)
     print('function deployed, address={}'.format(addr))
     with open('/tmp/output', 'w') as fp:
         fp.write(addr)
@@ -272,7 +276,7 @@ def get(kind, name, selector, namespace, uid, project, tag, db, extra_args):
         print(tabulate(df, headers='keys'))
 
     else:
-        print('currently only get pods [name] is supported')
+        print('currently only get pods | runs | artifacts [name] are supported')
 
 
 @main.command()
diff --git a/mlrun/db/base.py b/mlrun/db/base.py
index be3ac8791..5ba674793 100644
--- a/mlrun/db/base.py
+++ b/mlrun/db/base.py
@@ -32,7 +32,7 @@ class RunDBInterface(ABC):
         pass
 
     @abstractmethod
-    def get_log(self, uid, project=''):
+    def get_log(self, uid, project='', offset=0, size=0):
         pass
 
     @abstractmethod
diff --git a/mlrun/db/filedb.py b/mlrun/db/filedb.py
index bdcd9ed58..7394cb981 100644
--- a/mlrun/db/filedb.py
+++ b/mlrun/db/filedb.py
@@ -28,6 +28,7 @@ from ..utils import logger
 run_logs = 'runs'
 artifacts_dir = 'artifacts'
 functions_dir = 'functions'
+default_project = 'default'
 _missing = object()
 
 
@@ -50,13 +51,15 @@ class FileRunDB(RunDBInterface):
         # TODO: handle append
         self._datastore.put(filepath, body)
 
-    def get_log(self, uid, project='', offset=0):
+    def get_log(self, uid, project='', offset=0, size=0):
         filepath = self._filepath(run_logs, project, uid, '') + '.log'
         if pathlib.Path(filepath).is_file():
             with open(filepath, 'rb') as fp:
                 if offset:
                     fp.seek(offset)
-                return fp.read()
+                if not size:
+                    size = 2**18
+                return fp.read(size)
         return None
 
     def store_run(self, struct, uid, project='', commit=False):
@@ -230,11 +233,9 @@ class FileRunDB(RunDBInterface):
             tag = ''
         if tag:
             key = '/' + key
-        if project:
-            return path.join(self.dirpath, '{}/{}/{}{}'.format(
-                table, project, tag, key))
-        else:
-            return path.join(self.dirpath, '{}/{}{}'.format(table, tag, key))
+        project = project or default_project
+        return path.join(self.dirpath, '{}/{}/{}{}'.format(
+            table, project, tag, key))
 
     _encodings = {
         '.yaml': ('to_yaml', dict_to_yaml),
diff --git a/mlrun/db/httpd.py b/mlrun/db/httpd.py
index 3970628b7..113cf8032 100644
--- a/mlrun/db/httpd.py
+++ b/mlrun/db/httpd.py
@@ -65,13 +65,12 @@ def bearer_auth_required(cfg):
 
 @app.before_request
 def check_auth():
-    if request.path == '/healthz':
+    if request.path == '/api/healthz':
         return
 
     cfg = config.httpdb
 
     header = request.headers.get('Authorization', '')
-    print(header)
     try:
         if basic_auth_required(cfg):
             if not header.startswith(basic_prefix):
@@ -104,9 +103,9 @@ def catch_err(fn):
 
 
 # curl -d@/path/to/job.json http://localhost:8080/submit
-@app.route('/submit', methods=['POST'])
-@app.route('/submit/', methods=['POST'])
-@app.route('/submit/<path:func>', methods=['POST'])
+@app.route('/api/submit', methods=['POST'])
+@app.route('/api/submit/', methods=['POST'])
+@app.route('/api/submit/<path:func>', methods=['POST'])
 @catch_err
 def submit_job(func=''):
     try:
@@ -114,30 +113,36 @@ def submit_job(func=''):
     except ValueError:
         return json_error(HTTPStatus.BAD_REQUEST, reason='bad JSON body')
 
-    print("FUNC: ", func)
+    logger.info('submit_job: func %s', func)
     url = data.get('functionUrl')
     function = data.get('function')
     task = data.get('task')
     if not (function or url) or not task:
-        return json_error(HTTPStatus.BAD_REQUEST,
-                          reason='bad JSON, need to include function/url and task objects')
+        return json_error(
+            HTTPStatus.BAD_REQUEST,
+            reason='bad JSON, need to include function/url and task objects',
+        )
 
-    # TODO: block exec for function['kind'] in ['', 'local]  (must be a remote/container runtime)
+    # TODO: block exec for function['kind'] in ['', 'local]  (must be a
+    # remote/container runtime)
 
     try:
         if url:
             resp = new_function(command=url).run(task)
         else:
             resp = new_function(runtime=function).run(task)
-        print(resp.to_yaml())
+        logger.info('resp: %s', resp.to_yaml())
     except RunError as err:
-        return json_error(HTTPStatus.BAD_REQUEST, reason='runtime error: {}'.format(err))
+        return json_error(
+            HTTPStatus.BAD_REQUEST,
+            reason='runtime error: {}'.format(err),
+        )
 
     return jsonify(ok=True, data=resp.to_dict())
 
 
 # curl -d@/path/to/log http://localhost:8080/log/prj/7?append=true
-@app.route('/log/<project>/<uid>', methods=['POST'])
+@app.route('/api/log/<project>/<uid>', methods=['POST'])
 @catch_err
 def store_log(project, uid):
     append = strtobool(request.args.get('append', 'no'))
@@ -147,7 +152,7 @@ def store_log(project, uid):
 
 
 # curl http://localhost:8080/log/prj/7
-@app.route('/log/<project>/<uid>', methods=['GET'])
+@app.route('/api/log/<project>/<uid>', methods=['GET'])
 def get_log(project, uid):
     data = _file_db.get_log(uid, project)
     if data is None:
@@ -156,7 +161,7 @@ def get_log(project, uid):
     return data
 
 # curl -d @/path/to/run.json http://localhost:8080/run/p1/3?commit=yes
-@app.route('/run/<project>/<uid>', methods=['POST'])
+@app.route('/api/run/<project>/<uid>', methods=['POST'])
 @catch_err
 def store_run(project, uid):
     commit = strtobool(request.args.get('commit', 'no'))
@@ -171,7 +176,7 @@ def store_run(project, uid):
 
 
 # curl -X PATCH -d @/path/to/run.json http://localhost:8080/run/p1/3?commit=yes
-@app.route('/run/<project>/<uid>', methods=['PATCH'])
+@app.route('/api/run/<project>/<uid>', methods=['PATCH'])
 @catch_err
 def update_run(project, uid):
     try:
@@ -185,7 +190,7 @@ def update_run(project, uid):
 
 
 # curl http://localhost:8080/run/p1/3
-@app.route('/run/<project>/<uid>', methods=['GET'])
+@app.route('/api/run/<project>/<uid>', methods=['GET'])
 @catch_err
 def read_run(project, uid):
     data = _file_db.read_run(uid, project)
@@ -193,7 +198,7 @@ def read_run(project, uid):
 
 
 # curl -X DELETE http://localhost:8080/run/p1/3
-@app.route('/run/<project>/<uid>', methods=['DELETE'])
+@app.route('/api/run/<project>/<uid>', methods=['DELETE'])
 @catch_err
 def del_run(project, uid):
     _file_db.del_run(uid, project)
@@ -201,7 +206,7 @@ def del_run(project, uid):
 
 
 # curl http://localhost:8080/runs?project=p1&name=x&label=l1&label=l2&sort=no
-@app.route('/runs', methods=['GET'])
+@app.route('/api/runs', methods=['GET'])
 @catch_err
 def list_runs():
     name = request.args.get('name', '')
@@ -224,7 +229,7 @@ def list_runs():
     return jsonify(ok=True, runs=runs)
 
 # curl -X DELETE http://localhost:8080/runs?project=p1&name=x&days_ago=3
-@app.route('/runs', methods=['DELETE'])
+@app.route('/api/runs', methods=['DELETE'])
 @catch_err
 def del_runs():
     name = request.args.get('name', '')
@@ -238,7 +243,7 @@ def del_runs():
 
 
 # curl -d@/path/to/artifcat http://localhost:8080/artifact/p1/7&key=k
-@app.route('/artifact/<project>/<uid>/<path:key>', methods=['POST'])
+@app.route('/api/artifact/<project>/<uid>/<path:key>', methods=['POST'])
 @catch_err
 def store_artifact(project, uid, key):
     try:
@@ -252,14 +257,14 @@ def store_artifact(project, uid, key):
 
 
 # curl http://localhost:8080/artifact/p1/tag/key
-@app.route('/artifact/<project>/<tag>/<path:key>', methods=['GET'])
+@app.route('/api/artifact/<project>/<tag>/<path:key>', methods=['GET'])
 @catch_err
 def read_artifact(project, tag, key):
     data = _file_db.read_artifact(key, tag, project)
     return data
 
 # curl -X DELETE http://localhost:8080/artifact/p1&key=k&tag=t
-@app.route('/artifact/<project>/<uid>', methods=['DELETE'])
+@app.route('/api/artifact/<project>/<uid>', methods=['DELETE'])
 @catch_err
 def del_artifact(project, uid):
     key = request.args.get('key')
@@ -271,7 +276,7 @@ def del_artifact(project, uid):
     return jsonify(ok=True)
 
 # curl http://localhost:8080/artifacts?project=p1?label=l1
-@app.route('/artifacts', methods=['GET'])
+@app.route('/api/artifacts', methods=['GET'])
 @catch_err
 def list_artifacts():
     name = request.args.get('name', '')
@@ -283,7 +288,7 @@ def list_artifacts():
     return jsonify(ok=True, artifacts=artifacts)
 
 # curl -X DELETE http://localhost:8080/artifacts?project=p1?label=l1
-@app.route('/artifacts', methods=['DELETE'])
+@app.route('/api/artifacts', methods=['DELETE'])
 @catch_err
 def del_artifacts():
     name = request.args.get('name', '')
@@ -295,7 +300,7 @@ def del_artifacts():
     return jsonify(ok=True)
 
 # curl -d@/path/to/func.json http://localhost:8080/func/prj/7?tag=0.3.2
-@app.route('/func/<project>/<name>', methods=['POST'])
+@app.route('/api/func/<project>/<name>', methods=['POST'])
 @catch_err
 def store_function(project, name):
     try:
@@ -310,7 +315,7 @@ def store_function(project, name):
 
 
 # curl http://localhost:8080/log/prj/7?tag=0.2.3
-@app.route('/func/<project>/<name>', methods=['GET'])
+@app.route('/api/func/<project>/<name>', methods=['GET'])
 @catch_err
 def get_function(project, name):
     tag = request.args.get('tag', '')
@@ -319,7 +324,7 @@ def get_function(project, name):
 
 
 # curl http://localhost:8080/funcs?project=p1&name=x&label=l1&label=l2
-@app.route('/funcs', methods=['GET'])
+@app.route('/api/funcs', methods=['GET'])
 @catch_err
 def list_functions():
     name = request.args.get('name', '')
@@ -335,7 +340,7 @@ def list_functions():
     )
 
 
-@app.route('/healthz', methods=['GET'])
+@app.route('/api/healthz', methods=['GET'])
 def health():
     return 'OK\n'
 
diff --git a/mlrun/db/httpdb.py b/mlrun/db/httpdb.py
index 8bc7b349f..0363c8b8e 100644
--- a/mlrun/db/httpdb.py
+++ b/mlrun/db/httpdb.py
@@ -48,7 +48,7 @@ class HTTPRunDB(RunDBInterface):
         return f'{cls}({self.base_url!r})'
 
     def _api_call(self, method, path, error=None, params=None, body=None):
-        url = f'{self.base_url}/{path}'
+        url = f'{self.base_url}/api/{path}'
         kw = {
             key: value
             for key, value in (('params', params), ('data', body))
@@ -85,7 +85,7 @@ class HTTPRunDB(RunDBInterface):
         error = f'store log {project}/{uid}'
         self._api_call('POST', path, error, params, body)
 
-    def get_log(self, uid, project=''):
+    def get_log(self, uid, project='', offset=0, size=0):
         path = self._path_of('log', project, uid)
         error = f'get log {project}/{uid}'
         resp = self._api_call('GET', path, error)
diff --git a/mlrun/kfpops.py b/mlrun/kfpops.py
index e7fe6a01a..a623d32cb 100644
--- a/mlrun/kfpops.py
+++ b/mlrun/kfpops.py
@@ -35,7 +35,7 @@ def write_kfpmeta(struct):
         json.dump(metrics, f)
 
     output_artifacts, out_dict = get_kfp_outputs(
-        struct['status'].get(run_keys.artifacts, []))
+        struct['status'].get(run_keys.artifacts, []), struct['metadata'].get('labels', {}))
 
     for key in struct['spec'].get(run_keys.outputs, []):
         val = 'None'
@@ -71,7 +71,7 @@ def write_kfpmeta(struct):
         json.dump(metadata, f)
 
 
-def get_kfp_outputs(artifacts):
+def get_kfp_outputs(artifacts, labels):
     outputs = []
     out_dict = {}
     for output in artifacts:
@@ -83,6 +83,11 @@ def get_kfp_outputs(artifacts):
         if target.startswith('v3io:///'):
             target = target.replace('v3io:///', 'http://v3io-webapi:8081/')
 
+        user = labels.get('v3io_user', '') or environ.get('V3IO_USERNAME', '')
+        if target.startswith('/User/'):
+            user = user or 'admin'
+            target = 'http://v3io-webapi:8081/users/' + user + target[5:]
+
         viewer = output.get('viewer', '')
         if viewer in ['web-app', 'chart']:
             meta = {'type': 'web-app',
@@ -201,20 +206,25 @@ def mlrun_op(name: str = '', project: str = '', function=None,
 
     if runobj:
         handler = handler or runobj.spec.handler_name
-        params = params or runobj.spec.parameters or {}
-        hyperparams = hyperparams or runobj.spec.hyperparams or {}
+        params = params or runobj.spec.parameters
+        hyperparams = hyperparams or runobj.spec.hyperparams
         param_file = param_file or runobj.spec.param_file
         selector = selector or runobj.spec.selector
-        inputs = inputs or runobj.spec.inputs or {}
+        inputs = inputs or runobj.spec.inputs
         outputs = outputs or runobj.spec.outputs
         in_path = in_path or runobj.spec.input_path
         out_path = out_path or runobj.spec.output_path
-        secrets = secrets or runobj.spec.secret_sources or []
+        secrets = secrets or runobj.spec.secret_sources
         project = project or runobj.metadata.project
 
     if hyperparams or param_file:
         outputs.append('iteration_results')
 
+    params = params or {}
+    hyperparams = hyperparams or {}
+    inputs = inputs or {}
+    secrets = secrets or []
+
     if name:
         cmd += ['--name', name]
     for s in secrets:
diff --git a/mlrun/run.py b/mlrun/run.py
index b37260171..7e11a84c7 100644
--- a/mlrun/run.py
+++ b/mlrun/run.py
@@ -17,8 +17,8 @@ from base64 import b64decode
 from copy import deepcopy
 from os import environ, path, makedirs
 from tempfile import mktemp
-
 import yaml
+from nuclio import build_file
 
 from .execution import MLClientCtx
 from .model import RunObject
@@ -272,29 +272,35 @@ def parse_command(runtime, url):
         update_in(runtime, 'spec.args', arg_list[1:])
 
 
-def code_to_function(name='', filename='', handler='', runtime=None,
-                     image=None):
+def code_to_function(name='', filename='', handler='', runtime='',
+                     image=None, embed_code = True):
     """convert code or notebook to function object with embedded code
     code stored in the function spec and can be refreshed using .with_code()
-    eliminate the need to build container images everytime we edit the code
+    eliminate the need to build container images every time we edit the code
 
-    :param name:      function name
-    :param filename:  blank for current notebook, or path to .py/.ipynb file
-    :param handler:   name of function handler (if not main)
-    :param runtime:   optional, runtime type local, job, dask, mpijob, ..
-    :param image:     optional, container image
+    :param name:       function name
+    :param filename:   blank for current notebook, or path to .py/.ipynb file
+    :param handler:    name of function handler (if not main)
+    :param runtime:    optional, runtime type local, job, dask, mpijob, ..
+    :param image:      optional, container image
+    :param embed_code: embed the source code into the function spec
 
     :return:
            function object
     """
-    if runtime == 'nuclio':
+    if runtime.startswith('nuclio'):
         r = RemoteRuntime()
+        kind = runtime[runtime.rfind(':')+1:] if ':' in runtime else None
+        if embed_code:
+            bname, spec, code = build_file(filename, handler=handler or 'handler', kind=kind)
+            name = name or bname
+            r.spec.base_spec = spec
+        else:
+            r.spec.source = filename
+            r.spec.function_handler = handler
         r.metadata.name = name
-        r.spec.source = filename
-        r.spec.function_handler = handler
         return r
 
-    from nuclio import build_file
     bname, spec, code = build_file(filename, handler=handler)
 
     if runtime is None or runtime in ['', 'local']:
diff --git a/mlrun/runtimes/base.py b/mlrun/runtimes/base.py
index fe3a34492..0761fd5ca 100644
--- a/mlrun/runtimes/base.py
+++ b/mlrun/runtimes/base.py
@@ -193,6 +193,8 @@ class BaseRuntime(ModelObj):
         if self.spec.rundb:
             self._db_conn = get_run_db(self.spec.rundb).connect(self._secrets)
 
+        if 'V3IO_USERNAME' in environ:
+            meta.labels['v3io_user'] = environ.get('V3IO_USERNAME')
         meta.labels['kind'] = self.kind
         meta.labels['owner'] = meta.labels.get('owner', getpass.getuser())
         add_code_metadata(meta.labels)
diff --git a/mlrun/runtimes/function.py b/mlrun/runtimes/function.py
index c0446b362..a10324972 100644
--- a/mlrun/runtimes/function.py
+++ b/mlrun/runtimes/function.py
@@ -135,8 +135,8 @@ class RemoteRuntime(BaseRuntime):
     def add_model(self, key, model):
         if model.startswith('v3io://'):
             model = '/User/' + '/'.join(model.split('/')[5:])
-        if '://' not in model:
-            model = 'file://' + model
+        #if '://' not in model:
+        #    model = 'file://' + model
         if not model.endswith('/'):
             model = model[:model.rfind('/')]
         self.set_env('SERVING_MODEL_{}'.format(key), model)


Test patch:
diff --git a/tests/test_httpdb.py b/tests/test_httpdb.py
index cd3b17264..d819aa7c2 100644
--- a/tests/test_httpdb.py
+++ b/tests/test_httpdb.py
@@ -51,7 +51,7 @@ def start_server(dirpath, log_file, env_config):
     proc = Popen(cmd, env=env, stdout=log_file, stderr=log_file, cwd=root)
     url = f'http://localhost:{port}'
 
-    health_url = f'{url}/healthz'
+    health_url = f'{url}/api/healthz'
     timeout = 30
     if not wait_for_server(health_url, timeout):
         raise RuntimeError('server did not start after {timeout}sec')
diff --git a/tests/test_xdocker.py b/tests/test_xdocker.py
index 1cc492e88..28e00d174 100644
--- a/tests/test_xdocker.py
+++ b/tests/test_xdocker.py
@@ -53,7 +53,7 @@ def build_docker():
         out = run(cmd, stdout=PIPE, check=True)
         cid = out.stdout.decode('utf-8').strip()
 
-        url = f'http://localhost:{port}/healthz'
+        url = f'http://localhost:{port}/api/healthz'
         timeout = 30
         assert wait_for_server(url, timeout), \
             f'server failed to start after {timeout} seconds, url={url}'