Refactor httpd to organized app using FastAPI (#255)

* Remove all locks

* Init - added dynamic body resources - runs, functions, artifacts, schedules

* Added logs

* Added projects

* Added pipelines

* Added submit

* Added files

* little fixes

* Added workflows

* Added tags

* Added healthz

* order

* quotes

* lint

* Nicer docs

* fix sqldb rundb mass

* Added auth verifier dep

* fixed sqldb tests

* globals to singletons

* projects fixes

* only sql test with in memory

* Project tests working

* Added test artifacts

* added test schedules

* tag test working

* remove httpd

* imports

* remove httpd requirements

* Add app dockerfile and requirements

* remove httpd dockerfile

* export the right port

* oops

* Create db table once

* remove httpd tests

* fix app tests

* remove db file after tests

* Fix projects cache + add test

* enable makefile's run-app

* remove added abstract methods from interface cause httpdb does not fulfill

* Fix test httpdb server mode

* httpdb fixes

* fix dockerfile

* requirements change

* ok

* remove httpd from places (replace with app)

* Remove redundant command from api deployment yamls

* must return a list

* forgot returns

* fix artifacts

* app -> api (and also mlrun-api -> api)

* add api package to setup.py

* sub module

* to relative imports

* Revert "to relative imports"

This reverts commit 955ec9c.

* adding packages

* oops

* bugfix

* logging - remove later

* Fixing test_dbs

* Fixed filedb proxy as well

* fix test notebooks

* fix test_kfp

* fixing api tests

* generalize db stuff

* no SessionLocal

* oops

* POST -> GET

* don't use uvicorn-gunicorn-fastapi-docker as base docker file

* oops

* final

* chmod

* docker file fixes

* change mlrun-ui to ui

* retry official image but with max workers 1

* Revert "retry official image but with max workers 1"

This reverts commit 753e9a2.

* Fix submit pipeline endpoint bugs

* extend healthz response

* move requirements to api

* re-enable python -m mlrun db

* remove un-needed requirements from docs also

* back to old image tags

* leftovers

* stam

* upgrade tests python version (FastAPI dependencies with yield require Python 3.7 or above)

* Fix API dockerfile CMD

* python 3.7 instead

Makefile

@@ -39,15 +39,15 @@ test: clean
 	    -rf \
 	    tests
 
-.PHONY: run-httpd
-run-httpd:
+.PHONY: run-api
+run-api:
 	python -m mlrun db
 
-.PHONY: docker-db
-docker-httpd:
+.PHONY: docker-api
+docker-api:
 	docker build \
 	    -f dockerfiles/mlrun-api/Dockerfile \
-	    -t mlrun/mlrun-httpd .
+	    -t mlrun/mlrun-api .
 
 .PHONY: circleci
 circleci:

dockerfiles/api-requirements.txt

@@ -0,0 +1,4 @@
+croniter==0.3.31
+fastapi==0.54.1
+uvicorn==0.11.3
+pydantic==1.5.1

dockerfiles/mlrun-api/Dockerfile

@@ -11,7 +11,7 @@
 # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 # See the License for the specific language governing permissions and
 # limitations under the License.
-ARG PYTHON_VER=3.6
+ARG PYTHON_VER=3.7
 
 FROM python:${PYTHON_VER}-slim
 
@@ -20,7 +20,10 @@ LABEL org="iguazio.com"
 
 ENV PYTHON_VER=${PYTHON_VER}
 
-RUN apt-get update && apt-get install -y gcc git-core
+RUN apt-get update && apt-get install -y \
+  gcc \
+  git-core \
+  vim
 
 RUN python -m pip install --upgrade --no-cache pip
 
@@ -31,15 +34,12 @@ COPY ./dockerfiles/*requirements.txt ./
 COPY ./requirements.txt ./
 
 RUN python -m pip install \
-    -r httpd-requirements.txt \
+    -r api-requirements.txt \
     -r requirements.txt
 
-COPY . .
-
 ENV MLRUN_httpdb__dirpath=/mlrun/db
 ENV MLRUN_httpdb__port=8080
+COPY . .
 VOLUME /mlrun/db
-CMD gunicorn \
-    --bind=0.0.0.0:$MLRUN_httpdb__port \
-    --worker-class gevent \
-    mlrun.db.httpd:app
+
+CMD ["python",  "-m",  "mlrun",  "db"]

dockerfiles/test/Dockerfile

@@ -11,7 +11,7 @@
 # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 # See the License for the specific language governing permissions and
 # limitations under the License.
-ARG PYTHON_VER=3.6
+ARG PYTHON_VER=3.7
 
 FROM python:${PYTHON_VER}-slim
 
@@ -46,7 +46,7 @@ COPY ./requirements.txt ./
 
 RUN pip install \
     -r dev-requirements.txt \
-    -r httpd-requirements.txt \
+    -r api-requirements.txt \
     -r requirements.txt
 
 COPY . .

mlrun/__main__.py

@@ -450,7 +450,7 @@ def db(port, dirpath):
     if dirpath is not None:
         env['MLRUN_httpdb__dirpath'] = dirpath
 
-    cmd = [executable, '-m', 'mlrun.db.httpd']
+    cmd = [executable, '-m', 'mlrun.api.main']
     child = Popen(cmd, env=env)
     returncode = child.wait()
     if returncode != 0:

mlrun/api/api/api.py

@@ -0,0 +1,19 @@
+from fastapi import APIRouter, Depends
+
+from mlrun.api.api import deps
+from mlrun.api.api.endpoints import artifacts, files, functions, healthz, logs, pipelines, projects, runs, schedules, \
+    submit, tags, workflows
+
+api_router = APIRouter()
+api_router.include_router(artifacts.router, tags=["artifacts"], dependencies=[Depends(deps.AuthVerifier)])
+api_router.include_router(files.router, tags=["files"], dependencies=[Depends(deps.AuthVerifier)])
+api_router.include_router(functions.router, tags=["functions"], dependencies=[Depends(deps.AuthVerifier)])
+api_router.include_router(healthz.router, tags=["healthz"])
+api_router.include_router(logs.router, tags=["logs"], dependencies=[Depends(deps.AuthVerifier)])
+api_router.include_router(pipelines.router, tags=["pipelines"], dependencies=[Depends(deps.AuthVerifier)])
+api_router.include_router(projects.router, tags=["projects"], dependencies=[Depends(deps.AuthVerifier)])
+api_router.include_router(runs.router, tags=["runs"], dependencies=[Depends(deps.AuthVerifier)])
+api_router.include_router(schedules.router, tags=["schedules"], dependencies=[Depends(deps.AuthVerifier)])
+api_router.include_router(submit.router, tags=["submit"], dependencies=[Depends(deps.AuthVerifier)])
+api_router.include_router(tags.router, tags=["tags"], dependencies=[Depends(deps.AuthVerifier)])
+api_router.include_router(workflows.router, tags=["workflows"], dependencies=[Depends(deps.AuthVerifier)])

mlrun/api/api/deps.py

@@ -0,0 +1,65 @@
+from base64 import b64decode
+from http import HTTPStatus
+from typing import Generator
+from sqlalchemy.orm import Session
+
+from fastapi import Request
+
+from mlrun.api.api.utils import log_and_raise
+from mlrun.api.db.session import create_session, close_session
+from mlrun.config import config
+
+
+def get_db_session() -> Generator[Session, None, None]:
+    try:
+        db_session = create_session()
+        yield db_session
+    finally:
+        close_session(db_session)
+
+
+class AuthVerifier:
+    _basic_prefix = 'Basic '
+    _bearer_prefix = 'Bearer '
+
+    def __init__(self, request: Request):
+        self.username = None
+        self.password = None
+        self.token = None
+
+        cfg = config.httpdb
+
+        header = request.headers.get('Authorization', '')
+        if self._basic_auth_required(cfg):
+            if not header.startswith(self._basic_prefix):
+                log_and_raise(HTTPStatus.UNAUTHORIZED, reason="missing basic auth")
+            user, password = self._parse_basic_auth(header)
+            if user != cfg.user or password != cfg.password:
+                log_and_raise(HTTPStatus.UNAUTHORIZED, reason="bad basic auth")
+            self.username = user
+            self.password = password
+        elif self._bearer_auth_required(cfg):
+            if not header.startswith(self._bearer_prefix):
+                log_and_raise(HTTPStatus.UNAUTHORIZED, reason="missing bearer auth")
+            token = header[len(self._bearer_prefix):]
+            if token != cfg.token:
+                log_and_raise(HTTPStatus.UNAUTHORIZED, reason="bad basic auth")
+            self.token = token
+
+    @staticmethod
+    def _basic_auth_required(cfg):
+        return cfg.user or cfg.password
+
+    @staticmethod
+    def _bearer_auth_required(cfg):
+        return cfg.token
+
+    @staticmethod
+    def _parse_basic_auth(header):
+        """
+        parse_basic_auth('Basic YnVnczpidW5ueQ==')
+        ['bugs', 'bunny']
+        """
+        b64value = header[len(AuthVerifier._basic_prefix):]
+        value = b64decode(b64value).decode()
+        return value.split(':', 1)

mlrun/api/api/endpoints/artifacts.py

@@ -0,0 +1,99 @@
+import asyncio
+from http import HTTPStatus
+from typing import List
+
+from fastapi import APIRouter, Depends, Request, Query
+from sqlalchemy.orm import Session
+
+from mlrun.api.api import deps
+from mlrun.api.api.utils import log_and_raise
+from mlrun.api.singletons import get_db
+from mlrun.config import config
+from mlrun.utils import logger
+
+router = APIRouter()
+
+
+# curl -d@/path/to/artifcat http://localhost:8080/artifact/p1/7&key=k
+@router.post("/artifact/{project}/{uid}/{key:path}")
+def store_artifact(
+        request: Request,
+        project: str,
+        uid: str,
+        key: str,
+        tag: str = "",
+        iter: int = 0,
+        db_session: Session = Depends(deps.get_db_session)):
+    data = None
+    try:
+        data = asyncio.run(request.json())
+    except ValueError:
+        log_and_raise(HTTPStatus.BAD_REQUEST, reason="bad JSON body")
+
+    logger.debug(data)
+    get_db().store_artifact(db_session, key, data, uid, iter=iter, tag=tag, project=project)
+    return {}
+
+
+# curl http://localhost:8080/artifact/p1/tags
+@router.get("/projects/{project}/artifact-tags")
+def list_artifact_tags(
+        project: str,
+        db_session: Session = Depends(deps.get_db_session)):
+    tags = get_db().list_artifact_tags(db_session, project)
+    return {
+        "project": project,
+        "tags": tags,
+    }
+
+
+# curl http://localhost:8080/projects/my-proj/artifact/key?tag=latest
+@router.get("/projects/{project}/artifact/{key:path}")
+def read_artifact(
+        project: str,
+        key: str,
+        tag: str = "latest",
+        iter: int = 0,
+        db_session: Session = Depends(deps.get_db_session)):
+    data = get_db().read_artifact(db_session, key, tag=tag, iter=iter, project=project)
+    return {
+        "data": data,
+    }
+
+
+# curl -X DELETE http://localhost:8080/artifact/p1&key=k&tag=t
+@router.delete("/artifact/{project}/{uid}")
+def del_artifact(
+        project: str,
+        uid: str,
+        key: str,
+        tag: str = "",
+        db_session: Session = Depends(deps.get_db_session)):
+    get_db().del_artifact(db_session, key, tag, project)
+    return {}
+
+
+# curl http://localhost:8080/artifacts?project=p1?label=l1
+@router.get("/artifacts")
+def list_artifacts(
+        project: str = config.default_project,
+        name: str = None,
+        tag: str = None,
+        labels: List[str] = Query([]),
+        db_session: Session = Depends(deps.get_db_session)):
+    artifacts = get_db().list_artifacts(db_session, name, project, tag, labels)
+    return {
+        "artifacts": artifacts,
+    }
+
+
+# curl -X DELETE http://localhost:8080/artifacts?project=p1?label=l1
+@router.delete("/artifacts")
+def del_artifacts(
+        project: str = "",
+        name: str = "",
+        tag: str = "",
+        labels: List[str] = Query([]),
+        db_session: Session = Depends(deps.get_db_session)):
+    get_db().del_artifacts(db_session, name, project, tag, labels)
+    return {}

mlrun/api/api/endpoints/files.py

@@ -0,0 +1,77 @@
+import mimetypes
+from http import HTTPStatus
+
+from fastapi import APIRouter, Query, Request, Response
+
+from mlrun.api.api.utils import log_and_raise, get_obj_path, get_secrets
+from mlrun.datastore import get_object_stat, StoreManager
+
+router = APIRouter()
+
+
+# curl http://localhost:8080/api/files?schema=s3&path=mybucket/a.txt
+@router.get("/files")
+def get_files(
+        request: Request,
+        schema: str = "",
+        objpath: str = Query("", alias="path"),
+        user: str = "",
+        size: int = 0,
+        offset: int = 0):
+    _, filename = objpath.split(objpath)
+
+    objpath = get_obj_path(schema, objpath, user=user)
+    if not objpath:
+        log_and_raise(HTTPStatus.NOT_FOUND, path=objpath, err="illegal path prefix or schema")
+
+    secrets = get_secrets(request)
+    body = None
+    try:
+        stores = StoreManager(secrets)
+        obj = stores.object(url=objpath)
+        if objpath.endswith("/"):
+            listdir = obj.listdir()
+            return {
+                "listdir": listdir,
+            }
+
+        body = obj.get(size, offset)
+    except FileNotFoundError as e:
+        log_and_raise(HTTPStatus.NOT_FOUND, path=objpath, err=str(e))
+    if body is None:
+        log_and_raise(HTTPStatus.NOT_FOUND, path=objpath)
+
+    ctype, _ = mimetypes.guess_type(objpath)
+    if not ctype:
+        ctype = "application/octet-stream"
+    return Response(content=body, media_type=ctype, headers={"x-suggested-filename": filename})
+
+
+# curl http://localhost:8080/api/filestat?schema=s3&path=mybucket/a.txt
+@router.get("/filestat")
+def get_filestat(
+        request: Request,
+        schema: str = "",
+        path: str = "",
+        user: str = ""):
+    _, filename = path.split(path)
+
+    path = get_obj_path(schema, path, user=user)
+    if not path:
+        log_and_raise(HTTPStatus.NOT_FOUND, path=path, err="illegal path prefix or schema")
+    secrets = get_secrets(request)
+    stat = None
+    try:
+        stat = get_object_stat(path, secrets)
+    except FileNotFoundError as e:
+        log_and_raise(HTTPStatus.NOT_FOUND, path=path, err=str(e))
+
+    ctype, _ = mimetypes.guess_type(path)
+    if not ctype:
+        ctype = "application/octet-stream"
+
+    return {
+        "size": stat.size,
+        "modified": stat.modified,
+        "mimetype": ctype,
+    }