Тесты, линтеры, Prometheus, Grafana, логирование
Непрерывная интеграция и доставка, мониторинг производительности и логирование в production.
.github/workflows/ci.yml:
name: CI
on:
push:
branches: [main]
pull_request:
branches: [main]
jobs:
test:
runs-on: ubuntu-latest
services:
postgres:
image: postgres:15
env:
POSTGRES_PASSWORD: test
options: >-
--health-cmd pg_isready
--health-interval 10s
--health-timeout 5s
--health-retries 5
ports:
- 5432:5432
steps:
- uses: actions/checkout@v3
- name: Set up Python
uses: actions/setup-python@v4
with:
python-version: '3.11'
- name: Install dependencies
run: |
pip install poetry
poetry install
- name: Lint
run: |
poetry run flake8
poetry run black --check
poetry run mypy .
- name: Test
run: |
poetry run pytest --cov=app --cov-report=xml
env:
DATABASE_URL: postgresql://postgres:test@localhost:5432/postgres
- name: Upload coverage
uses: codecov/codecov-action@v3
build:
needs: test
runs-on: ubuntu-latest
if: github.ref == 'refs/heads/main'
steps:
- uses: actions/checkout@v3
- name: Build Docker image
run: docker build -t myapp:${{ github.sha }} .
- name: Push to registry
run: |
docker login -u ${{ secrets.DOCKER_USER }} -p ${{ secrets.DOCKER_PASS }}
docker push myapp:${{ github.sha }}.gitlab-ci.yml:
stages:
- lint
- test
- build
- deploy
lint:
stage: lint
image: python:3.11
script:
- pip install poetry
- poetry install
- poetry run flake8
- poetry run black --check
test:
stage: test
image: python:3.11
services:
- postgres:15
variables:
POSTGRES_DB: test
POSTGRES_USER: postgres
POSTGRES_PASSWORD: ""
script:
- poetry install
- poetry run pytest --cov=app
build:
stage: build
image: docker:latest
services:
- docker:dind
script:
- docker build -t registry.gitlab.com/project/app:$CI_COMMIT_SHA .
- docker push registry.gitlab.com/project/app:$CI_COMMIT_SHA
only:
- main
deploy:
stage: deploy
image: alpine:latest
script:
- apk add openssh-client
- ssh -o StrictHostKeyChecking=no user@server "cd /opt/app && docker-compose pull && docker-compose up -d"
only:
- mainpip install prometheus-fastapi-instrumentatorfrom fastapi import FastAPI
from prometheus_fastapi_instrumentator import Instrumentator
app = FastAPI()
@app.on_event("startup")
async def startup():
Instrumentator().instrument(app).expose(app)
# Метрики доступны на /metricshttp_requests_total — всего запросовhttp_request_duration_seconds — время ответаhttp_requests_in_progress — запросов в работеglobal:
scrape_interval: 15s
scrape_configs:
- job_name: 'fastapi'
static_configs:
- targets: ['api:8000']
metrics_path: '/metrics'version: '3.8'
services:
prometheus:
image: prom/prometheus
volumes:
- ./prometheus.yml:/etc/prometheus/prometheus.yml
- prometheus_data:/prometheus
command:
- '--config.file=/etc/prometheus/prometheus.yml'
- '--storage.tsdb.path=/prometheus'
ports:
- "9090:9090"
grafana:
image: grafana/grafana
volumes:
- grafana_data:/var/lib/grafana
environment:
- GF_SECURITY_ADMIN_PASSWORD=admin
- GF_USERS_ALLOW_SIGN_UP=false
ports:
- "3000:3000"
depends_on:
- prometheus
volumes:
prometheus_data:
grafana_data:import logging
import json
from pythonjsonlogger import jsonlogger
logger = logging.getLogger(__name__)
def setup_logging():
logHandler = logging.StreamHandler()
formatter = jsonlogger.JsonFormatter(
fmt="%(asctime)s %(levelname)s %(name)s %(message)s"
)
logHandler.setFormatter(formatter)
logger.addHandler(logHandler)
logger.setLevel(logging.INFO)
@app.on_event("startup")
async def startup():
setup_logging()
logger.info("Application started")from fastapi import Request
import time
@app.middleware("http")
async def log_requests(request: Request, call_next):
start_time = time.time()
logger.info(
f"Request: {request.method} {request.url.path}",
extra={
"method": request.method,
"path": request.url.path,
"client_ip": request.client.host
}
)
response = await call_next(request)
duration = time.time() - start_time
logger.info(
f"Response: {response.status_code} in {duration:.3f}s",
extra={
"status_code": response.status_code,
"duration_ms": round(duration * 1000, 2)
}
)
return responseglobal:
smtp_smarthost: 'smtp.example.com:587'
smtp_from: 'alertmanager@example.com'
route:
group_by: ['alertname']
receiver: 'email'
receivers:
- name: 'email'
email_configs:
- to: 'team@example.com'
alert_rules:
- alert: HighErrorRate
expr: rate(http_requests_total{status=~"5.."}[5m]) > 0.1
for: 5m
annotations:
summary: "High error rate detected"pip install opentelemetry-api opentelemetry-sdk opentelemetry-exporter-jaegerfrom opentelemetry import trace
from opentelemetry.exporter.jaeger.thrift import JaegerExporter
from opentelemetry.sdk.trace import TracerProvider
from opentelemetry.sdk.trace.export import BatchSpanProcessor
trace.set_tracer_provider(TracerProvider())
jaeger_exporter = JaegerExporter(
agent_host_name="localhost",
agent_port=6831,
)
trace.get_tracer_provider().add_span_processor(
BatchSpanProcessor(jaeger_exporter)
)
tracer = trace.get_tracer(__name__)
@app.get("/users/{user_id}")
def get_user(user_id: int):
with tracer.start_as_current_span("get_user"):
user = db.query(User).get(user_id)
return userВопросы ещё не добавлены
Вопросы для этой подтемы ещё не добавлены.