Welcome to Part 5 of the tutorial series: “Build a Production-Ready User Management Microservice with Flask and SQLAlchemy: A Step-by-Step Guide”. In this part, we’ll add logging and monitoring to the application to track requests, errors, and performance. By the end of this tutorial, your microservice will be production-ready with proper logging and monitoring in place.

What you learn in part 5

• How to add logging to track requests, errors, and important events.
• How to configure log levels and log file rotation.
• How to set up Prometheus and Grafana to monitor application performance.
• How to handle logging for Flask applications using the @jwt_required decorator.
• How to test logging and monitoring.

Prerequisites

Before we begin, ensure you’ve completed Part 1, Part 2, Part 3, and Part 4 of the series. You should have:

• A Flask application with SQLAlchemy configured.
• A User model with password hashing.
• CRUD operations for the User model.
• JWT authentication for securing endpoints.
• Error handling and input validation.

Logging helps track requests, errors, and important events in your application. Let’s configure logging in the Flask app.

Update app/__init__.py

Add logging configuration to the create_app function:

import logging
from logging.handlers import RotatingFileHandler
import os

def create_app():
    app = Flask(__name__)
    app.config.from_object(Config)

    db.init_app(app)
    jwt.init_app(app)

    # Configure logging
    if not app.debug:
        # Create logs directory if it doesn't exist
        if not os.path.exists('logs'):
            os.mkdir('logs')

        # Set up a rotating file handler
        file_handler = RotatingFileHandler(
            'logs/microservice.log', maxBytes=10240, backupCount=10
        )
        file_handler.setFormatter(logging.Formatter(
            '%(asctime)s %(levelname)s: %(message)s [in %(pathname)s:%(lineno)d]'
        ))
        file_handler.setLevel(logging.INFO)
        app.logger.addHandler(file_handler)

        # Set the log level for the application
        app.logger.setLevel(logging.INFO)
        app.logger.info('Microservice startup')

    with app.app_context():
        from . import routes
        app.register_blueprint(routes.bp)

        db.create_all()

    return app

The code configures logging for the Flask application when it’s not in debug mode.

1. Creates a logs directory
   The application checks if the logs directory exists. If it doesn’t, it creates the directory to store log files.
2. Sets up a rotating file handler
   The code initializes a RotatingFileHandler to write logs to logs/microservice.log. It limits the file size to 10,240 bytes (10 KB) and keeps up to 10 backup log files to prevent excessive disk usage.
3. Formats log messages
   The formatter structures log messages to include the timestamp, log level, message, and the file/line number where the log was generated.
4. Adds the handler to the Flask logger
   The application attaches the file handler to Flask’s built-in logger, ensuring logs are stored in microservice.log.
5. Sets the logging level
   The logger records only INFO level and higher messages, filtering out lower-priority debug logs.
6. Logs the application startup
   When the application starts, it logs a message: "Microservice startup".

This setup helps monitor errors and application activity while keeping logs manageable through rotation. 🚀

Step 2: Add Logging to Routes

Add logging to the routes to track requests and errors.

Update app/routes.py

Add logging to the /api/register and /api/login endpoints:

from flask import Blueprint, request, jsonify, current_app
from flask_jwt_extended import create_access_token, jwt_required, get_jwt_identity
from .schemas import user_schema, ValidationError
from .models import User
from . import db

bp = Blueprint('api', __name__, url_prefix='/api')

@bp.route('/test', methods=['GET'])
def test():
    return jsonify({"message": "Welcome to the user management microservice"})

@bp.route('/register', methods=['POST'])
def register():
    try:
        # Validate and deserialize the input data
        data = user_schema.load(request.get_json())
    except ValidationError as err:
        current_app.logger.error(f"Validation: {err.messages}")
        return jsonify({"error": "Validation Error", "message": err.messages}), 400
    
    # Check if username or email already exists
    if User.query.filter_by(username=data['username']).first():
        current_app.logger.warning(f"Username already exists: {data['username']}")
        return jsonify({"error": "Username already exists"}), 400
    
    if User.query.filter_by(email=data['email']).first():
        current_app.logger.warning(f"Email already exists: {data['email']}")
        return jsonify({"error": "Email already exists"}), 400
    
    # Create a new user
    new_user = User(username=data['username'], email=data['email'])
    new_user.set_password(data['password'])
    db.session.add(new_user)
    db.session.commit()

    return jsonify({
        "id": new_user.id,
        "username": new_user.username,
        "email": new_user.email
    }), 201


@bp.route('/login', methods=['POST'])
def login():
    try:
        # Validate and deserialize the input data
        data = user_schema.load(request.get_json)
    except ValidationError as err:
        current_app.logger.error(f"Validation error: {err.messages}")
        return jsonify({"error": "Validation Error", "message": err.messages}), 400

    username = data['username']
    password = data['password']

    # Find the user
    user = User.query.filter_by(username=username).first()
    if user and user.check_password(password):
        access_token = create_access_token(identity=str(user.id))
        current_app.logger.info(f"User logged in: {user.username}")
        return jsonify(access_token=access_token), 200
    
    current_app.logger.warning(f"Invalid login attempt for username: {username}")
    return jsonify({"error": "Invalid credentials"}), 401


@bp.route('/users', methods=['GET'])
@jwt_required()
def get_all_users():
    users = User.query.all()
    return jsonify([{
        "id": user.id,
        "username": user.username,
        "email": user.email
    } for user in users])

@bp.route('/users/<int:id>', methods=['GET'])
@jwt_required()
def get_user(id):
    user = User.query.get_or_404(id)
    return jsonify({
        "id": user.id,
        "username": user.username,
        "email": user.email
    })

@bp.route('/users/<int:id>', methods=['PUT'])
@jwt_required()
def update_user(id):
    user = User.query.get_or_404(id)
    data = request.get_json()

    if 'username' in data:
        user.username = data['username']
    if 'email' in data:
        user.email = data['email']
    if 'password' in data:
        user.set_password(data['password'])

    db.session.commit()
    return jsonify({
        "id": user.id,
        "username": user.username,
        "email": user.email
    })

@bp.route('/users/<int:id>', methods=['DELETE'])
@jwt_required()
def delete_user(id):
    user = User.query.get_or_404(id)
    db.session.delete(user)
    db.session.commit()
    return jsonify({"message": "User deleted successfully"})

The code uses current_app to log errors, warnings, and important events in the Flask application.

Why Use current_app?

Flask’s current_app provides access to the Flask application context within a blueprint. Since blueprints don’t inherently have direct access to the app instance, we use current_app to interact with the application’s logger.

How We Use current_app for Logging?

1. Logging Validation Errors

   current_app.logger.error(f"Validation error: {err.messages}")
   

   When user input fails validation, the application logs an error message with details about what went wrong.

2. Logging Duplicate Username or Email Attempts

   current_app.logger.warning(f"Username already exists: {data['username']}")
   current_app.logger.warning(f"Email already exists: {data['email']}")
   

   If a user tries to register with an existing username or email, the system logs a warning, indicating a potential duplicate entry.

3. Logging Successful User Registration

   current_app.logger.info(f"New user registered: {new_user.username}")
   

   When a new user registers successfully, the app logs an info-level message with the username.

4. Logging Successful Logins

   current_app.logger.info(f"User logged in: {user.username}")
   

   When a user logs in, the system logs an info-level message to track successful authentication attempts.

5. Logging Invalid Login Attempts

   current_app.logger.warning(f"Invalid login attempt for username: {username}")
   

   If login fails due to incorrect credentials, the system logs a warning to track potential unauthorized access attempts.

Why This Matters?

• Improved Debugging: Logs help developers track errors and diagnose issues.
• Security Monitoring: Unauthorized login attempts and repeated failed validations can indicate suspicious activity.
• User Activity Tracking: Logging successful registrations and logins helps monitor application usage.

By using current_app.logger, the blueprint remains independent while still leveraging the centralized logging system of the main Flask application. 🚀

Handling @jwt_required Decorator

When using the @jwt_required decorator, you can log JWT-related events such as token validation failures or unauthorized access attempts. Here’s an example:

@bp.route('/protected', methods=['GET'])
@jwt_required()
def protected():
    current_user_id = get_jwt_identity()
    user = User.query.get(current_user_id)
    if not user:
        current_app.logger.warning(f"Unauthorized access attempt by user ID: {current_user_id}")
        return jsonify({"error": "User not found"}), 404

    current_app.logger.info(f"Authorized access by user: {user.username}")
    return jsonify({"message": f"Hello, {user.username}!"}), 200

The code defines a protected route that requires authentication using JWT.

1. Require Authentication
   @jwt_required() ensures only users with a valid JWT token can access the route.
2. Retrieve and Verify User
   • get_jwt_identity() extracts the user ID from the token.
   • The app queries the database for the user. If not found, it logs a warning and returns a 404 error.
3. Log Access and Respond
   • If the user exists, the app logs an info-level message.
   • It returns a 200 response with a personalized greeting.

This approach secures access, prevents unauthorized requests, and logs user activity for debugging. 🚀

Step 3: Set up Prometheus and Grafana

Prometheus and Grafana are powerful tools for monitoring and visualizing application metrics. Here’s how to set them up in a way that’s easy to follow and free of errors.

Step 3.1: Install Docker (if not already installed)

If you don’t have Docker installed, follow these steps to set it up:

On Linux

1. Install Docker:
   Run the following commands in your terminal:

   sudo apt-get update
   sudo apt-get install docker.io

2. Start Docker:
   Start the Docker service and enable it to run on boot:

   sudo systemctl start docker
   sudo systemctl enable docker

3. Verify Installation:
   Check if Docker is installed correctly:

   docker --version

On macOS

1. Download Docker Desktop:
   • Go to the Docker Desktop for Mac page.
   • Download and install Docker Desktop.
2. Start Docker:
   • Open Docker Desktop from your Applications folder.
   • Verify installation by running:

     docker --version

On Windows

1. Download Docker Desktop:
   • Go to the Docker Desktop for Windows page.
   • Download and install Docker Desktop.
2. Start Docker:
   • Open Docker Desktop from your Start menu.
   • Verify installation by running in Command Prompt:

     docker --version

Step 3.2: Create the docker-compose.yml File

1. Where to place the file:
   • Create a file named docker-compose.yml in the root directory of your project (the same folder where your Flask app’s main files are located).
2. Add the following content to docker-compose.yml:

   services:
     prometheus:
       image: prom/prometheus
       container_name: prometheus
       ports:
         - "9090:9090"
       volumes:
         - ./prometheus.yml:/etc/prometheus/prometheus.yml  # Mount the Prometheus config file
       command:
         - '--config.file=/etc/prometheus/prometheus.yml'
   
     grafana:
       image: grafana/grafana
       container_name: grafana
       ports:
         - "3000:3000"
       volumes:
         - grafana-storage:/var/lib/grafana
       environment:
         - GF_SECURITY_ADMIN_PASSWORD=admin  # Set the default admin password
   
   volumes:
     grafana-storage:  # Define a volume for Grafana data persistence

   This Docker Compose file sets up Prometheus and Grafana for monitoring.

   1. Prometheus Service
      • Uses the prom/prometheus image.
      • Exposes port 9090.
      • Mounts prometheus.yml as its configuration file.
   2. Grafana Service
      • Uses the grafana/grafana image.
      • Exposes port 3000.
      • Stores data in a persistent volume.
      • Sets the admin password to "admin".
   3. Persistent Storage
      • Defines grafana-storage to retain Grafana data.

   This setup enables real-time monitoring with Prometheus and Grafana. 🚀

Step 3.3: Create the prometheus.yml file

1. Where to Place the File:
   • Create a file named prometheus.yml in the root directory of your project (the same folder as docker-compose.yml).
2. Add the following content to prometheus.yml:

   global:
     scrape_interval: 15s  # How often to scrape metrics
   
   scrape_configs:
     - job_name: 'flask_app'
       static_configs:
         - targets: ['host.docker.internal:5000']  # Replace with your Flask app's host and port

   This Prometheus configuration controls how it collects metrics.

   1. Global Settings
      • Prometheus scrapes metrics every 15 seconds (scrape_interval: 15s).
   2. Scrape configuration for Flask app
      • Prometheus targets the Flask application running on host.docker.internal:5000.
      • The job_name: 'flask_app' identifies the Flask app in Prometheus.

   This setup ensures Prometheus regularly collects metrics from the Flask app. 🚀

Step 3.4: Start Prometheus and Grafana

1. Navigate to the Project Directory:
   Open a terminal and navigate to the root directory of your project (where docker-compose.yml and prometheus.yml are located).
2. Start the Services:
   Run the following command to start Prometheus and Grafana:

   docker-compose up -d

3. Verify the Services:
   • Prometheus: Open http://localhost:9090 in your browser.
   • Grafana: Open http://localhost:3000 in your browser.
     • Log in with the default credentials:
       • Username: admin
       • Password: admin

Step 3.5: Expose Flask metrics

To expose metrics from your Flask application, use the prometheus-flask-exporter library.

1. Install the Library:Run the following command to install the library:

   pip install prometheus-flask-exporter

2. Update app/__init__.py:Initialize the Prometheus metrics exporter in your Flask app:

   import os
   import logging
   from logging.handlers import RotatingFileHandler
   from prometheus_flask_exporter import PrometheusMetrics
   from flask import Flask, jsonify, request
   from flask_sqlalchemy import SQLAlchemy
   from flask_jwt_extended import JWTManager
   from config import Config
   
   db = SQLAlchemy()
   jwt = JWTManager()
   
   def create_app():
       app = Flask(__name__)
       app.config.from_object(Config)
   
       db.init_app(app)
       jwt.init_app(app)
   
       # Initialize Prometheus metrics
       metrics = PrometheusMetrics(app)  # Enable default metrics
   
       # Configure logging
       if not app.debug:
           # Create logs directory if it doesn't exist
           if not os.path.exists('logs'):
               os.mkdir('logs')
   
           # Set up a rotating file handler
           file_handler = RotatingFileHandler(
               'logs/microservice.log', maxBytes=10240, backupCount=10
           )
           file_handler.setFormatter(logging.Formatter(
               '%(asctime)s %(levelname)s: %(message)s [in %(pathname)s:%(lineno)d]'
           ))
           file_handler.setLevel(logging.INFO)
           app.logger.addHandler(file_handler)
   
           # Set the log level for the application
           app.logger.setLevel(logging.INFO)
           app.logger.info('Microservice startup')
   
       @app.errorhandler(404)
       def not_found(error):
           return  jsonify({"error": "Not Found", "message": "The requested resource was not found"}), 404
       
       @app.errorhandler(500)
       def internal_error(error):
           return jsonify({"error": "Internal Server Error", "message": "Something went wrong on the server"}), 500
   
       with app.app_context():
           from . import routes
           app.register_blueprint(routes.bp)
   
           db.create_all()
   
       return app
   
   

   This code enables Prometheus metrics in the Flask application.

   • PrometheusMetrics(app) automatically collects default metrics like request counts, response times, and errors.
   • Prometheus scrapes these metrics to monitor the app’s performance and health.

   This setup helps track API usage and detect issues efficiently. 🚀

3. Access Metrics:
   • Start your Flask application.
   • Metrics will be available at http://localhost:5000/metrics.

Step 3.6: Visualize metrics in Grafana

1. Log in to Grafana:
   • Open http://localhost:3000 in your browser.
   • Log in with the username admin and password admin.
2. Add Prometheus as a Data Source:
   • Go to Configuration > Data Sources.
   • Click Add data source.
   • Select Prometheus.
   • Set the URL to http://prometheus:9090 (if using Docker) or http://localhost:9090 (if running locally).
   • Click Save & Test.
3. Create a Dashboard:
   • Go to Create > Dashboard.
   • Add a new panel.
   • Use Prometheus queries to visualize metrics. For example:
     • Request Rate: rate(http_request_duration_seconds_count[1m])
     • Error Rate: rate(http_request_duration_seconds_count{status=~"5.."}[1m])
     • Response Time: histogram_quantile(0.95, sum(rate(http_request_duration_seconds_bucket[1m])) by (le)
4. Save the Dashboard:
   • Give your dashboard a name and save it.

Your microservice is now ready to be monitored with Prometheus and Grafana!

Helpful Links

Here are some helpful links to assist developers who may encounter issues with configuring Grafana and Prometheus:

Prometheus documentation

• Prometheus Getting Started
• Prometheus Configuration
• Prometheus Querying

Grafana documentation

• Grafana Getting Started
• Grafana Data Sources
• Grafana Dashboards

Troubleshooting

• Prometheus Troubleshooting
• Grafana Troubleshooting

Full code for part 5

You can find the complete code for this tutorial in the GitHub repository.

What’s Next?

In Part 6, we’ll dive into writing unit tests for your Flask application. Unit testing is a critical step in ensuring your microservice is reliable, maintainable, and free of bugs. Stay tuned!