lab_4 до интеграции grafana

.gitignore

@@ -1,4 +1,5 @@
 ### Python
+__pycache__/
 *.pyc
 
 ### Jupyter
@@ -9,3 +10,5 @@
 # virtual environments
 .venv/
 .venv*/
+# .env files
+*.env

README.md

@@ -8,9 +8,15 @@
 ](https://www.kaggle.com/datasets/vijayaadithyanvg/car-price-predictionused-cars/data) —
 продажа подержанных автомобилей на рынке в Индии.
 
-## Установка
+## Сервис предсказания цен
 
-### Общий порядок
+См. `services/ml_service.md`.
+
+## Исследовательская часть проекта
+
+### Установка
+
+#### Общий порядок
 
 **Внимание**: Здесь описан только общий порядок установки. Определённые части проекта могут требовать установки по отдельным инструкциям.
 
@@ -47,13 +53,13 @@
 
 5. **При необходимости** скачайте данные. Каноническое расположение для данных проекта: `data/`.
 
-### Зависимости
+#### Зависимости
 
-#### Общие зависимости
+##### Общие зависимости
 
 Зависимости — пакеты Python — записаны в файле `requirements/requirements.txt` (см. **Пакеты Python**).
 
-#### Пакеты Python
+##### Пакеты Python
 
 Установка/обновление пакетов Python в активное окружение из файла `requirements/requirements.txt`:
 
@@ -61,10 +67,10 @@
 pip install -U -r requirements/requirements.txt
 ```
 
-## Разведочный анализ данных (EDA)
+### Разведочный анализ данных (EDA)
 
 См. `eda/README.md`.
 
-## Исследование и настройка предсказательной модели
+### Исследование и настройка предсказательной модели
 
 См. `research/README.md`.

run_mlflow_server.sh

@@ -2,7 +2,7 @@
 
 set -eu
 
-. _mlflow_config_common.sh
+. ./_mlflow_config_common.sh
 
 DEFAULT_ARTIFACT_ROOT="./mlflow/mlartifacts/"
 

services/compose.env.template

@@ -0,0 +1,2 @@
+GF_SECURITY_ADMIN_USER=admin
+GF_SECURITY_ADMIN_PASSWORD=admin

services/compose.yaml

@@ -0,0 +1,48 @@
+name: mpei-iis-system
+
+services:
+
+  prices-predictor:
+    image: ml_service:2
+    ports:
+      - "8010:8000"
+    volumes:
+      - './models:/models'
+
+  load-tester:
+    image: load_tester:1
+    environment:
+      API_BASE_URL: "http://prices-predictor:8000/api"
+    deploy:
+      replicas: 2
+
+  prometheus:
+    image: prom/prometheus:v3.7.3
+    ports:
+      - "9090:9090"
+    command:
+      - "--config.file=/etc/prometheus/prometheus.yaml"
+    volumes:
+      - "./prometheus/prometheus.yaml:/etc/prometheus/prometheus.yaml:ro"
+
+  grafana:
+    image: grafana/grafana:12.4.0-20012734117
+    ports:
+      - "3000:3000"
+    #environment:
+    #  GF_SECURITY_ADMIN_USER: "$__file{/run/secrets/grafana-admin-user}"
+    #  GF_SECURITY_ADMIN_PASSWORD: "$__file{/run/secrets/grafana-admin-password}"
+    #secrets:
+    #  - grafana-admin-user
+    #  - grafana-admin-password
+    environment:
+      GF_SECURITY_ADMIN_USER: "${GF_SECURITY_ADMIN_USER:-admin}"
+      GF_SECURITY_ADMIN_PASSWORD: "${GF_SECURITY_ADMIN_PASSWORD:-admin}"
+
+#secrets:
+#
+#  grafana-admin-user:
+#    environment: GF_SECURITY_ADMIN_USER
+#
+#  grafana-admin-password:
+#    environment: GF_SECURITY_ADMIN_PASSWORD

services/load_tester/.dockerignore

@@ -0,0 +1,6 @@
+### Python
+__pycache__/
+*.pyc
+
+### Project
+*.unused

services/load_tester/Dockerfile

@@ -0,0 +1,13 @@
+FROM python:3.11-slim
+
+WORKDIR /load_tester
+
+COPY ./requirements.txt .
+RUN pip install --no-cache-dir -r requirements.txt
+
+COPY . .
+
+CMD ["python", "-m", "tester"]
+
+# docker build -t load_tester:1 services/load_tester/
+# docker run -e "API_BASE_URL=http://prices-predictor:8000/api" load_tester:1

services/load_tester/requirements.txt

@@ -0,0 +1 @@
+requests >=2.32.5,<3

services/load_tester/tester.py

@@ -0,0 +1,280 @@
+from argparse import ArgumentParser
+from collections.abc import Callable, MutableMapping
+from dataclasses import dataclass, asdict
+from enum import Enum
+import logging
+from os import getenv
+from random import randint, uniform, expovariate, choice
+from signal import SIGINT, SIGTERM, signal
+import sys
+from time import sleep
+from types import FrameType
+from typing import Any, cast
+
+from requests import RequestException, Response, Session
+
+
+def fixup_payload_enum_value(mapping: MutableMapping[str, Any], key: str) -> None:
+    mapping[key] = mapping[key].value
+
+
+ENDPOINT_URL: str = '/predict'
+
+
+class FuelType(Enum):
+    PETROL = 'petrol'
+    DIESEL = 'diesel'
+    CNG = 'cng'
+
+
+class SellingType(Enum):
+    DEALER  = 'dealer'
+    INDIVIDUAL  = 'individual'
+
+
+class TransmissionType(Enum):
+    MANUAL  = 'manual'
+    AUTOMATIC   = 'automatic'
+
+
+@dataclass
+class PricePredictionFeatures:
+    selling_price: float
+    driven_kms: float
+    age: float
+    fuel_type: FuelType
+    selling_type: SellingType
+    transmission_type: TransmissionType
+
+
+MAX_RETRIES_DEFAULT = 3
+
+
+def exp_delay_from_attempt_number(attempt_i: int) -> float:
+    return 0.2 * (2 ** attempt_i)
+
+
+def post_item(
+    session: Session, url: str, item_id: int, features: PricePredictionFeatures,
+    *, max_retries: int = MAX_RETRIES_DEFAULT,
+) -> Response:
+    if max_retries < 0:
+        raise ValueError('max_retries must be >= 0')
+    payload = asdict(features)
+    for k in ('fuel_type', 'selling_type', 'transmission_type'):
+        fixup_payload_enum_value(payload, k)
+    excs = []
+    for attempt_i in range(max_retries + 1):
+        try:
+            response = session.post(url, params={'item_id': item_id}, json=payload, timeout=10)
+        except RequestException as err:
+            excs.append(err)
+            sleep(exp_delay_from_attempt_number(attempt_i))
+        else:
+            return response
+    assert len(excs) > 0
+    # XXX: ...
+    raise IOError(
+        f'Failed to post an item in {max_retries + 1} attempts;'
+         ' see the latest exception in __cause__'
+    ) from excs[-1]
+
+
+def generate_request_data() -> tuple[int, PricePredictionFeatures]:
+    item_id = randint(1, 100)
+    features = PricePredictionFeatures(
+        selling_price=round(uniform(2.0, 16.0), 2),
+        driven_kms=round(uniform(0.0, 100000.0), 0),
+        age=round(uniform(0.0, 10.0), 1),
+        fuel_type=choice(list(FuelType)),
+        selling_type=choice(list(SellingType)),
+        transmission_type=choice(list(TransmissionType)),
+    )
+    return (item_id, features)
+
+
+INTERVAL_MEAN_DEFAULT = 4.0
+INTERVAL_BOUNDS_DEFAULT: tuple[float | None, float | None] = (0.5, 10.0)
+
+
+class Requester:
+
+    def __init__(
+        self,
+        base_url: str,
+        interval_mean: float = INTERVAL_MEAN_DEFAULT,
+        interval_bounds: tuple[float | None, float | None] = INTERVAL_BOUNDS_DEFAULT,
+        *, max_retries: int = MAX_RETRIES_DEFAULT,
+    ):
+        self.base_url = base_url
+        self.interval_mean = interval_mean
+        self.interval_bounds = interval_bounds
+        self.max_retries = max_retries
+        self._session = Session()
+        self._stop_requested: bool = False
+
+    @property
+    def endpoint(self) -> str:
+        endpoint_url = ENDPOINT_URL
+        if (len(endpoint_url) > 0) and (not endpoint_url.startswith('/')):
+            endpoint_url = '/' + endpoint_url
+        return (self.base_url + endpoint_url)
+
+    @property
+    def session(self) -> Session:
+        return self._session
+
+    @property
+    def stop_requested(self) -> bool:
+        return self._stop_requested
+
+    def stop(self) -> None:
+        self._stop_requested = True
+
+    def _decide_delay(self) -> float:
+        interval_bounds = self.interval_bounds
+        val = expovariate(1. / self.interval_mean)
+        if interval_bounds[0] is not None:
+            val = max(val, interval_bounds[0])
+        if interval_bounds[1] is not None:
+            val = min(val, interval_bounds[1])
+        return val
+
+    def run(self) -> None:
+        while not self._stop_requested:
+            item_id, features = generate_request_data()
+            try:
+                response = post_item(
+                    self._session, self.endpoint, item_id, features, max_retries=self.max_retries,
+                )
+            except IOError as err:
+                logging.warning('%s: %s', str(err), str(err.__cause__))
+                raise err
+            else:
+                logging.debug('Success: %s %s', response.status_code, response.reason)
+            sleep(self._decide_delay())
+
+
+def _build_termination_handler(requester: Requester) -> Callable[[int, FrameType | None], None]:
+    def termination_handler(sig: int, frame: FrameType | None) -> None:
+        _ = sig
+        _ = frame
+        requester.stop()
+    return termination_handler
+
+
+def _configure_logging(level: int, quiet: bool) -> None:
+    if quiet:
+        level = logging.CRITICAL + 1
+    logging.basicConfig(
+        level=level, format='%(asctime)s %(levelname)s %(message)s', stream=sys.stderr,
+    )
+
+
+def _setup_signal_handlers(requester: Requester) -> None:
+    termination_handler = _build_termination_handler(requester)
+    for sig in (SIGINT, SIGTERM):
+        signal(sig, termination_handler)
+
+
+def _validate_cli_interval_bound(string: str) -> float | None:
+    string = string.lower()
+    if string in ('', 'null', 'none'):
+        return None
+    return float(string)
+
+
+def _validate_cli_interval_bounds(string: str) -> tuple[float | None, float | None]:
+    string = string.lower()
+    if string in ('', 'null', 'none'):
+        return (None, None)
+    min_string, max_string = string.split(',', 1)
+    return cast(
+        tuple[float | None, float | None],
+        tuple(map(_validate_cli_interval_bound, (min_string, max_string)))
+    )
+
+
+def _validate_cli_max_retries(string: str) -> int:
+    val = int(string)
+    if val < 0:
+        raise ValueError(f'Max retries should be >=0, given {val}')
+    return val
+
+
+def _validate_cli_logging_level(string: str) -> int:
+    return {
+        'debug':    logging.DEBUG,
+        'info': logging.INFO,
+        'warning':  logging.WARNING,
+        'error':    logging.ERROR,
+        'critical': logging.CRITICAL,
+    }[string]
+
+
+def parse_args(argv):
+    parser = ArgumentParser(
+        description=(
+            'Регулярная отправка POST-запросов на эндпоинт предсказания цены.'
+            ' Остановка по SIGINT / SIGTERM.'
+        ),
+        allow_abbrev=False,
+        exit_on_error=True,
+    )
+    parser.add_argument('base_url', type=str, nargs='?')
+    parser.add_argument('--interval-mean', type=float, dest='interval_mean')
+    parser.add_argument(
+        '--interval-bounds', type=_validate_cli_interval_bounds, dest='interval_bounds',
+    )
+    parser.add_argument(
+        '--max-retries',
+        type=_validate_cli_max_retries,
+        default=MAX_RETRIES_DEFAULT,
+        dest='max_retries',
+    )
+    parser.add_argument('-q', '--quiet', action='store_true', dest='quiet')
+    parser.add_argument(
+        '--log-level',
+        default=logging.WARNING,
+        type=_validate_cli_logging_level,
+        dest='logging_level',
+    )
+    args = parser.parse_args(argv[1:])
+    if args.base_url is None:
+        args.base_url = getenv('API_BASE_URL')
+        if args.base_url is None:
+            raise RuntimeError('No API base URL specified')
+    if (args.interval_mean is not None) and (args.interval_mean <= 0):
+        raise ValueError(f'Interval mean should be > 0, given {args.interval_mean}')
+    if (
+        (args.interval_bounds is not None)
+        and all((b is not None) for b in args.interval_bounds)
+        and (args.interval_bounds[0] > args.interval_bounds[1])
+    ):
+        raise ValueError(f'Interval bounds should be b_1 <= b_2, given {args.interval_bounds!r}')
+    if args.interval_mean is not None:
+        if args.interval_bounds is None:
+            args.interval_bounds = ((args.interval_mean / 5), (args.interval_mean * 5))
+    else:
+        args.interval_mean = INTERVAL_MEAN_DEFAULT
+        args.interval_bounds = INTERVAL_BOUNDS_DEFAULT
+    return args
+
+
+def main(argv):
+    args = parse_args(argv)
+    _configure_logging(args.logging_level, args.quiet)
+    logging.debug('Creating a Requester with base URL: %s', args.base_url)
+    requester = Requester(
+        args.base_url,
+        interval_mean=args.interval_mean,
+        interval_bounds=args.interval_bounds,
+        max_retries=args.max_retries,
+    )
+    _setup_signal_handlers(requester)
+    requester.run()
+    return 0
+
+
+if __name__ == '__main__':
+    sys.exit(int(main(sys.argv) or 0))

services/load_tester/tester_wrapper.py.unused

@@ -0,0 +1,37 @@
+from os import (
+    getenv,
+    #kill,
+)
+#from signal import SIGINT, SIGTERM
+from subprocess import (
+    run,
+    #Popen,
+)
+import sys
+
+
+#_child_proc: Popen | None = None
+
+
+#def _forward_signal(sig, frame):
+#    _ = frame
+#    if _child_proc is None:
+#        return
+#    if _child_proc.pid in (SIGINT, SIGTERM):
+#        kill(_child_proc.pid, sig)
+#    else:
+#        raise RuntimeError(f'Attempted to forward an unexpected signal: {sig}')
+
+
+def main(argv):
+    argv = list(argv)   # copy
+    base_url = getenv('API_BASE_URL')
+    if base_url is None:
+        raise RuntimeError('API_BASE_URL is not specified')
+    argv.append(base_url)   # HACK: ...
+    result = run(['python', '-m', 'tester', *argv[1:]], check=False)
+    return result.returncode
+
+
+if __name__ == '__main__':
+    sys.exit(int(main(sys.argv) or 0))

services/ml_service/.dockerignore

@@ -0,0 +1,3 @@
+### Python
+__pycache__/
+*.pyc

services/ml_service/Dockerfile

@@ -0,0 +1,19 @@
+FROM python:3.11-slim
+
+WORKDIR /service
+
+COPY ./requirements.txt .
+RUN pip install --no-cache-dir -r requirements.txt
+
+COPY . .
+
+VOLUME /models
+
+EXPOSE 8000/tcp
+
+ENV MODELS_PATH=/models
+
+CMD ["uvicorn", "app:app", "--host", "0.0.0.0", "--port", "8000"]
+
+# docker build -t ml_service:2 services/ml_service/
+# docker run -v "$(pwd)/services/models:/models" -p 8000:8000 ml_service:2

services/ml_service/README.md

@@ -0,0 +1,71 @@
+# Сервис предсказания цен
+
+Веб-сервис предсказания цен на подержанные автомобили; только stateless API. Об используемой предсказательной модели см. `research/README.md`.
+
+## API
+
+**Базовый URL**: `/api`. Все указанные далее URL записаны **относительно базового URL**, если не указано иное.
+
+* Полная интерактивная документация (Swagger UI): `/docs`.
+
+* Предсказать цену подержанного автомобиля: `/predict` (POST).
+
+  Пример запроса:
+
+  * requst query: `item_id=16` (параметр `item_id` необходим!);
+
+  * request body:
+
+    ```json
+    {
+        "selling_price": 5.59,
+        "driven_kms": 27000.0,
+        "age": 5.0,
+        "fuel_type": "petrol",
+        "selling_type": "dealer",
+        "transmission_type": "manual"
+    }
+    ```
+
+  * response body:
+
+    ```json
+    {
+        "item_id": 16,
+        "price": 3.743508852258851
+    }
+    ```
+
+* Тестовый эндпоинт: `/` (GET).
+
+  Возвращает простой демонстрационный объект JSON.
+
+  Может использоваться для проверки состояния сервиса.
+
+## Развёртывание
+
+### Файл модели
+
+Файл используемой предсказательной модели `model.pkl` можно извлечь из MLFlow скриптом `services/models/fetch_model_as_pickle_from_mlflow.py`. Файл модели можно разместить в директории проекта, а именно в `services/models/`.
+
+Например, извлечь модель по имени (`<model-name>`) и версии (`<model-version>`) (например, `UsedCardPricePredictionFinal/1`) (команда запускается из корневой директории проекта &mdash; от этого зависит путь к создаваемому файлу):
+
+    python services/models/fetch_model_as_pickle_from_mlflow.py --model "models:/<model-name>/<model-version>" services/models/model.pkl
+
+Можно указать адрес tracking сервера MLFlow, например: `--tracking-uri "http://localhost:5000"`.
+
+Информация о других опциях доступна:
+
+    python services/models/fetch_model_as_pickle_from_mlflow.py --help
+
+### Образ Docker
+
+Сборка образа (замените `<version>` на номер версии) (команда запускается из корневой директории проекта &mdash; от этого зависит путь к директории):
+
+    docker build -t ml_service:<version> services/ml_service/
+
+Запуск образа (замените `<version>` на номер версии образа, `<models-dir>` на **абсолютный** путь к директории, где размещён файл предсказательной модели `model.pkl`, `<port>` на порт для запуска веб-сервиса (например, `8000`)):
+
+    docker run -v "<models-dir>:/models" -p <port>:8000 ml_service:<version>
+
+Модель может быть размещена в директории проекта; тогда, например, при запуске команды из корна проекта: `$(pwd)/services/models` (здесь `$(pwd)` используется потому, что необходим абсолютный путь).

services/ml_service/app/__init__.py

@@ -0,0 +1,2 @@
+from ._meta import PACKAGE_PATH
+from .main import app

services/ml_service/app/_meta.py

@@ -0,0 +1,4 @@
+from pathlib import Path
+
+
+PACKAGE_PATH = Path(__file__).parent

services/ml_service/app/main.py

@@ -0,0 +1,72 @@
+from os import getenv
+from pathlib import Path
+
+from fastapi import FastAPI
+from prometheus_fastapi_instrumentator import Instrumentator
+from pydantic import BaseModel, Field
+
+from ._meta import PACKAGE_PATH
+from .predictor import (
+    FuelType, SellingType, TransmissionType, PricePredictionFeatures, PricePredictor,
+)
+
+
+MODELS_PATH = getenv('MODELS_PATH', None)
+if MODELS_PATH is not None:
+    MODELS_PATH = Path(MODELS_PATH)
+else:
+    SERVICES_PATH = PACKAGE_PATH.parents[1]
+    assert SERVICES_PATH.name == 'services'
+    MODELS_PATH = SERVICES_PATH / 'models'
+
+MODEL_PATH = MODELS_PATH / 'model.pkl'
+
+
+predictor = PricePredictor(MODEL_PATH)
+
+
+API_BASE_PATH = '/api'
+
+
+app = FastAPI(
+    title='Сервис ML',
+    version='0.1.0',
+    root_path=API_BASE_PATH,
+    #redoc_url=None,
+)
+
+
+_ = (
+    Instrumentator(excluded_handlers=['/metrics'])
+    .instrument(app)
+    .expose(app, endpoint='/metrics')
+)
+
+
+@app.get('/', summary='Тестовый эндпоинт')
+async def root():
+    return {'Hello': 'World'}
+
+
+class PricePredictionRequest(BaseModel):
+
+    selling_price: float = Field(..., gt=0)
+    driven_kms: float = Field(..., ge=0)
+    age: float = Field(..., ge=0)
+    fuel_type: FuelType
+    selling_type: SellingType
+    transmission_type: TransmissionType
+
+
+@app.post('/predict', summary='Предсказать цену подержанного автомобиля')
+def predict_price(item_id: int, req: PricePredictionRequest):
+    features = PricePredictionFeatures(
+        selling_price=req.selling_price,
+        driven_kms=req.driven_kms,
+        age=req.age,
+        fuel_type=req.fuel_type,
+        selling_type=req.selling_type,
+        transmission_type=req.transmission_type,
+    )
+    pred = predictor.predict(features)
+    return {'item_id': item_id, 'price': pred}

services/ml_service/app/predictor.py

@@ -0,0 +1,111 @@
+from dataclasses import dataclass
+from enum import Enum
+from itertools import chain
+from pandas import DataFrame
+from pickle import load
+from prometheus_client import Counter, Histogram
+
+
+def open_model_file(file, *, buffering=-1, opener=None, **kwargs_extra):
+    open_kwargs_extra = {}
+    if 'closefd' in kwargs_extra:
+        open_kwargs_extra['closefd'] = kwargs_extra.pop('closefd')
+    if len(kwargs_extra) > 0:
+        raise TypeError(
+            'Unexpected keyword arguments given: {}'
+            .format(', '.join(map(repr, kwargs_extra.keys())))
+        )
+    return open(file, 'rb', buffering=buffering, opener=opener)
+
+
+def load_model_from_file(file):
+    return load(file)
+
+
+def load_model_from_path(path, *, buffering=-1, opener=None, **kwargs_extra):
+    open_kwargs_extra = {}
+    for k in ('closefd',):
+        if k in kwargs_extra:
+            open_kwargs_extra[k] = kwargs_extra.pop(k)
+    if len(kwargs_extra) > 0:
+        raise TypeError(
+            'Unexpected keyword arguments given: {}'.format(', '.join(kwargs_extra.keys()))
+        )
+    with open_model_file(
+        path, buffering=buffering, opener=opener, **open_kwargs_extra,
+    ) as model_file:
+        return load_model_from_file(model_file)
+
+
+class FuelType(Enum):
+    PETROL = 'petrol'
+    DIESEL = 'diesel'
+    CNG = 'cng'
+
+
+class SellingType(Enum):
+    DEALER  = 'dealer'
+    INDIVIDUAL  = 'individual'
+
+
+class TransmissionType(Enum):
+    MANUAL  = 'manual'
+    AUTOMATIC   = 'automatic'
+
+
+@dataclass
+class PricePredictionFeatures:
+    selling_price: float
+    driven_kms: float
+    age: float
+    fuel_type: FuelType
+    selling_type: SellingType
+    transmission_type: TransmissionType
+
+
+metric_prediction_latency = Histogram(
+    'model_prediction_seconds', 'Время вычислений в модели',
+    buckets=(
+        list(chain.from_iterable((v * (10 ** p) for v in (1, 2, 5)) for p in range(-4, (1 + 1))))
+        + [float('+inf')]
+    ),
+)
+
+metric_prediction_errors = Counter(
+    'model_prediction_errors_total', 'Ошибки вычислений в модели по типу', ('error_type',),
+)
+
+metric_prediction_value = Histogram(
+    'model_prediction_value', 'Предсказанное значение цены',
+    buckets=(
+        list(chain.from_iterable((v * (10 ** p) for v in (1, 2, 5)) for p in range(-1, (2 + 1))))
+        + [float('+inf')]
+    ),
+)
+
+
+class PricePredictor:
+
+    def __init__(self, model_path):
+        self._model = load_model_from_path(model_path)
+
+    def predict(self, features):
+        # WARN: порядок столбцов вроде имеет значение
+        features_df = DataFrame([{
+            'selling_price':    features.selling_price,
+            'driven_kms':   features.driven_kms,
+            'fuel_type':    features.fuel_type.value,
+            'selling_type': features.selling_type.value,
+            'transmission': features.transmission_type.value,
+            'age':  features.age,
+        }])
+        try:
+            with metric_prediction_latency.time():
+                predictions = self._model.predict(features_df)
+        except Exception as err:
+            metric_prediction_errors.labels(error_type=type(err).__name__).inc()
+            raise
+        assert len(predictions) == 1
+        value = float(predictions[0])
+        metric_prediction_value.observe(value)
+        return value

services/ml_service/requirements.txt

@@ -0,0 +1,7 @@
+fastapi ~=0.120.4
+mlxtend ~=0.23.4
+pandas >=2.3.1,<3
+prometheus_client ~=0.23.1
+prometheus_fastapi_instrumentator >=7.0.2,<8
+scikit-learn >=1.7.2,<2
+uvicorn ~=0.38.0

services/models/.gitignore

@@ -0,0 +1 @@
+*.pkl

services/models/fetch_model_as_pickle_from_mlflow.py

@@ -0,0 +1,83 @@
+from argparse import ArgumentParser
+from pathlib import Path
+from pickle import dump
+from sys import exit as sys_exit, argv as sys_argv
+
+from mlflow import set_tracking_uri, set_registry_uri
+from mlflow.sklearn import load_model
+
+
+MLFLOW_TRACKING_URI_DEFAULT = 'http://localhost:5000'
+
+
+def open_file_for_model(file, *, buffering=-1, opener=None, **kwargs_extra):
+    open_kwargs_extra = {}
+    if 'closefd' in kwargs_extra:
+        open_kwargs_extra['closefd'] = kwargs_extra.pop('closefd')
+    if len(kwargs_extra) > 0:
+        raise TypeError(
+            'Unexpected keyword arguments given: {}'
+            .format(', '.join(map(repr, kwargs_extra.keys())))
+        )
+    return open(file, 'wb', buffering=buffering, opener=opener)
+
+
+def dump_model_to_file(model, file):
+    return dump(model, file)
+
+
+def dump_model_to_path(model, path, *, buffering=-1, opener=None, **kwargs_extra):
+    open_kwargs_extra = {}
+    for k in ('closefd',):
+        if k in kwargs_extra:
+            open_kwargs_extra[k] = kwargs_extra.pop(k)
+    if len(kwargs_extra) > 0:
+        raise TypeError(
+            'Unexpected keyword arguments given: {}'
+            .format(', '.join(map(repr, kwargs_extra.keys())))
+        )
+    with open_file_for_model(
+        path, buffering=buffering, opener=opener, **open_kwargs_extra,
+    ) as model_file:
+        return dump_model_to_file(model, model_file)
+
+
+def parse_args(argv):
+    parser = ArgumentParser(
+        description=(
+            'Скачать модель с tracking server MLFlow и сохранить в локальный файл pickle'
+        ),
+        allow_abbrev=False,
+        exit_on_error=True,
+    )
+    model_ref_parser = parser.add_mutually_exclusive_group(required=True)
+    model_ref_parser.add_argument('-m', '--model', type=str, dest='model_uri')
+    model_ref_parser.add_argument('--run', type=str, dest='run_id')
+    parser.add_argument(
+        '--tracking-uri', default=MLFLOW_TRACKING_URI_DEFAULT, type=str, dest='tracking_uri',
+    )
+    parser.add_argument('--registry-uri', type=str, dest='registry_uri')
+    parser.add_argument('out_path', default=Path('.'), type=Path)
+    args = parser.parse_args(argv)
+    return args
+
+
+def main(argv):
+    args = parse_args(argv)
+    set_tracking_uri(args.tracking_uri)
+    if args.registry_uri is not None:
+        set_registry_uri(args.registry_uri)
+    if args.model_uri is not None:
+        model_uri = args.model_uri
+    elif args.run_id is not None:
+        model_uri = f'runs:/{args.run_id}/model'
+    else:
+        assert False
+        return 1
+    model = load_model(model_uri)
+    dump_model_to_path(model, args.out_path)
+    return 0
+
+
+if __name__ == '__main__':
+    sys_exit(int(main(sys_argv) or 0))

services/prometheus/prometheus.yaml

@@ -0,0 +1,15 @@
+global:
+  scrape_interval: 15s
+  scrape_timeout: 5s
+
+scrape_configs:
+
+  - job_name: "prices_predictor"
+    static_configs:
+      - targets:
+        - "prices-predictor:8000"
+    scheme: http
+    metrics_path: "/metrics"
+    #relabel_configs:
+    #  - source_labels: ["__address__"]
+    #    target_labels: "instance"