Skip to content
Last9 named a Gartner Cool Vendor in AI for SRE Observability for 2025! Read more →
Last9
Talk to an Expert

Falcon

Monitor Falcon WSGI applications with OpenTelemetry instrumentation for high-performance API observability

Instrument your Falcon WSGI application with OpenTelemetry to send comprehensive telemetry data to Last9. This integration provides automatic instrumentation for HTTP requests, middleware, and resources, giving you complete visibility into your high-performance Falcon API’s behavior.

Prerequisites

  • Python 3.7 or higher
  • Falcon 3.0 or higher
  • Last9 account with OTLP endpoint configured

Installation

Install the required OpenTelemetry packages for Falcon instrumentation:

The easiest way to get started is using automatic instrumentation:

# Install core OpenTelemetry packages
pip install opentelemetry-api opentelemetry-sdk opentelemetry-exporter-otlp opentelemetry-distro


# Bootstrap to detect and install instrumentation packages
opentelemetry-bootstrap -a requirements
opentelemetry-bootstrap -a install

Configuration

  1. Set Environment Variables

    Configure the required environment variables for Last9 OTLP integration:

    export OTEL_SERVICE_NAME="your-falcon-service"
    export OTEL_EXPORTER_OTLP_ENDPOINT="$last9_otlp_endpoint"
    export OTEL_EXPORTER_OTLP_HEADERS="Authorization=$last9_otlp_auth_header"
    export OTEL_TRACES_EXPORTER="otlp"
    export OTEL_TRACES_SAMPLER="always_on"
    export OTEL_RESOURCE_ATTRIBUTES="service.name=your-falcon-service,service.version=1.0.0,deployment.environment=production"
    export OTEL_LOG_LEVEL="error"

  2. Run with Automatic Instrumentation

    The simplest way to instrument your Falcon application:

    # For development with gunicorn
    opentelemetry-instrument gunicorn --bind 0.0.0.0:8000 app:app
    

    # For development server
    opentelemetry-instrument python app.py
    

    # For production with uWSGI
    opentelemetry-instrument uwsgi --http :8000 --wsgi-file app.py --callable app

Manual Instrumentation

For more control over the instrumentation process:

Basic Setup

# app.py
import falcon
from opentelemetry import trace
from opentelemetry.exporter.otlp.proto.http.trace_exporter import OTLPSpanExporter
from opentelemetry.sdk.trace import TracerProvider
from opentelemetry.sdk.trace.export import BatchSpanProcessor
from opentelemetry.sdk.resources import Resource
from opentelemetry.instrumentation.falcon import FalconInstrumentor
from opentelemetry.instrumentation.requests import RequestsInstrumentor
import os


# Configure resource information
resource = Resource.create({
    "service.name": os.getenv("OTEL_SERVICE_NAME", "falcon-app"),
    "service.version": "1.0.0",
    "deployment.environment": os.getenv("DEPLOYMENT_ENV", "development"),
})


# Set up the tracer provider
trace.set_tracer_provider(TracerProvider(resource=resource))


# Configure OTLP exporter
otlp_exporter = OTLPSpanExporter(
    endpoint=os.getenv("OTEL_EXPORTER_OTLP_ENDPOINT"),
    headers={"Authorization": os.getenv("OTEL_EXPORTER_OTLP_HEADERS", "").replace("Authorization=", "")},
)


# Add span processor
span_processor = BatchSpanProcessor(otlp_exporter)
trace.get_tracer_provider().add_span_processor(span_processor)


# Create Falcon app
app = falcon.App()


# Instrument Falcon and other libraries
FalconInstrumentor().instrument_app(app)
RequestsInstrumentor().instrument()


# Get tracer for custom spans
tracer = trace.get_tracer(__name__)


# Sample resources
class HealthResource:
    def on_get(self, req, resp):
        resp.media = {"status": "healthy", "service": "falcon-api"}


class UsersResource:
    def on_get(self, req, resp):
        with tracer.start_as_current_span("get_users_business_logic") as span:
            span.set_attributes({
                "operation.type": "read",
                "resource.name": "users"
            })


            # Simulate business logic
            users = [
                {"id": 1, "name": "John Doe", "email": "john@example.com"},
                {"id": 2, "name": "Jane Smith", "email": "jane@example.com"},
            ]


            span.set_attribute("users.count", len(users))
            resp.media = {"users": users}


    def on_post(self, req, resp):
        with tracer.start_as_current_span("create_user_business_logic") as span:
            span.set_attributes({
                "operation.type": "create",
                "resource.name": "user"
            })


            user_data = req.media
            span.set_attribute("user.name", user_data.get("name", ""))
            span.set_attribute("user.email", user_data.get("email", ""))


            # Simulate user creation
            new_user = {
                "id": 123,
                "name": user_data["name"],
                "email": user_data["email"]
            }


            span.set_attribute("user.id", new_user["id"])
            resp.status = falcon.HTTP_201
            resp.media = new_user


class UserResource:
    def on_get(self, req, resp, user_id):
        with tracer.start_as_current_span("get_user_by_id") as span:
            span.set_attributes({
                "operation.type": "read",
                "resource.name": "user",
                "user.id": user_id
            })


            # Simulate user lookup
            if user_id == "404":
                span.set_attribute("user.found", False)
                raise falcon.HTTPNotFound(description="User not found")


            user = {
                "id": user_id,
                "name": f"User {user_id}",
                "email": f"user{user_id}@example.com"
            }


            span.set_attribute("user.found", True)
            resp.media = user


    def on_put(self, req, resp, user_id):
        with tracer.start_as_current_span("update_user") as span:
            span.set_attributes({
                "operation.type": "update",
                "resource.name": "user",
                "user.id": user_id
            })


            user_data = req.media
            span.set_attribute("user.name", user_data.get("name", ""))
            span.set_attribute("user.email", user_data.get("email", ""))


            updated_user = {
                "id": user_id,
                "name": user_data.get("name", f"User {user_id}"),
                "email": user_data.get("email", f"user{user_id}@example.com")
            }


            resp.media = updated_user


# Add routes
app.add_route('/health', HealthResource())
app.add_route('/users', UsersResource())
app.add_route('/users/{user_id}', UserResource())


if __name__ == '__main__':
    from wsgiref import simple_server
    httpd = simple_server.make_server('127.0.0.1', 8000, app)
    print("Serving on http://127.0.0.1:8000")
    httpd.serve_forever()

Custom Middleware with Tracing

import time
from opentelemetry import trace
from opentelemetry.trace import Status, StatusCode


class TimingMiddleware:
    def __init__(self):
        self.tracer = trace.get_tracer(__name__)


    def process_request(self, req, resp):
        req.context.start_time = time.time()


    def process_response(self, req, resp, resource, req_succeeded):
        if hasattr(req.context, 'start_time'):
            duration = time.time() - req.context.start_time


            # Get current span and add timing information
            span = trace.get_current_span()
            if span.is_recording():
                span.set_attribute("http.request_duration_ms", round(duration * 1000, 2))
                span.set_attribute("http.request_succeeded", req_succeeded)


class LoggingMiddleware:
    def __init__(self):
        self.tracer = trace.get_tracer(__name__)


    def process_request(self, req, resp):
        span = trace.get_current_span()
        if span.is_recording():
            span.set_attributes({
                "http.user_agent": req.user_agent or "",
                "http.remote_addr": req.remote_addr or "",
                "http.content_length": req.content_length or 0,
                "http.query_string": req.query_string or ""
            })


    def process_response(self, req, resp, resource, req_succeeded):
        span = trace.get_current_span()
        if span.is_recording() and not req_succeeded:
            span.set_status(Status(StatusCode.ERROR, "Request failed"))


# Add middleware to app
app.add_middleware(TimingMiddleware())
app.add_middleware(LoggingMiddleware())

Database Integration

Example with SQLAlchemy:

import sqlalchemy as sa
from sqlalchemy.orm import sessionmaker
from opentelemetry.instrumentation.sqlalchemy import SQLAlchemyInstrumentor


# Create database engine
DATABASE_URL = "postgresql://user:password@localhost/dbname"
engine = sa.create_engine(DATABASE_URL)
Session = sessionmaker(bind=engine)


# Instrument SQLAlchemy
SQLAlchemyInstrumentor().instrument(
    engine=engine,
    service="falcon-db",
)


class DatabaseUsersResource:
    def on_get(self, req, resp):
        with tracer.start_as_current_span("fetch_users_from_database") as span:
            session = Session()
            try:
                # This database query will be automatically traced
                users = session.execute(
                    sa.text("SELECT id, name, email FROM users ORDER BY id")
                ).fetchall()


                span.set_attribute("db.rows_returned", len(users))


                resp.media = {
                    "users": [
                        {"id": user.id, "name": user.name, "email": user.email}
                        for user in users
                    ]
                }
            finally:
                session.close()


    def on_post(self, req, resp):
        with tracer.start_as_current_span("create_user_in_database") as span:
            user_data = req.media
            span.set_attribute("user.name", user_data.get("name", ""))
            span.set_attribute("user.email", user_data.get("email", ""))


            session = Session()
            try:
                # Insert new user (automatically traced)
                result = session.execute(
                    sa.text(
                        "INSERT INTO users (name, email) VALUES (:name, :email) RETURNING id"
                    ),
                    {"name": user_data["name"], "email": user_data["email"]}
                )
                user_id = result.scalar()
                session.commit()


                span.set_attribute("user.id", user_id)


                resp.status = falcon.HTTP_201
                resp.media = {
                    "id": user_id,
                    "name": user_data["name"],
                    "email": user_data["email"]
                }
            except Exception as e:
                session.rollback()
                span.record_exception(e)
                span.set_status(Status(StatusCode.ERROR, str(e)))
                raise falcon.HTTPInternalServerError(description="Database error")
            finally:
                session.close()


# Add database routes
app.add_route('/db/users', DatabaseUsersResource())

Production Deployment

Gunicorn Configuration

# gunicorn_config.py
import multiprocessing


# Server socket
bind = "0.0.0.0:8000"
backlog = 2048


# Worker processes
workers = multiprocessing.cpu_count() * 2 + 1
worker_class = "sync"
worker_connections = 1000
max_requests = 1000
max_requests_jitter = 50
preload_app = True
timeout = 30
keepalive = 60


# Logging
loglevel = "info"
accesslog = "-"
errorlog = "-"
access_log_format = '%(h)s %(l)s %(u)s %(t)s "%(r)s" %(s)s %(b)s "%(f)s" "%(a)s" %(D)s'


# OpenTelemetry configuration
def post_fork(server, worker):
    server.log.info(f"Worker {worker.pid} spawned")


def worker_exit(server, worker):
    server.log.info(f"Worker {worker.pid} exited")

Start with:

opentelemetry-instrument gunicorn -c gunicorn_config.py app:app

Docker Configuration

# Dockerfile
FROM python:3.11-slim


WORKDIR /app


COPY requirements.txt .
RUN pip install --no-cache-dir -r requirements.txt


COPY . .


# Set OpenTelemetry environment variables
ENV OTEL_SERVICE_NAME=falcon-docker-app
ENV OTEL_EXPORTER_OTLP_ENDPOINT=$last9_otlp_endpoint
ENV OTEL_EXPORTER_OTLP_HEADERS="Authorization=$last9_otlp_auth_header"
ENV OTEL_RESOURCE_ATTRIBUTES="deployment.environment=docker,service.version=1.0.0"


EXPOSE 8000


# Run with OpenTelemetry instrumentation
CMD ["opentelemetry-instrument", "gunicorn", "-c", "gunicorn_config.py", "app:app"]

uWSGI Configuration

# uwsgi.ini
[uwsgi]
module = app:app
master = true
processes = 4
socket = 0.0.0.0:8000
protocol = http
die-on-term = true
vacuum = true
max-requests = 1000


# OpenTelemetry environment
env = OTEL_SERVICE_NAME=falcon-uwsgi-app
env = OTEL_EXPORTER_OTLP_ENDPOINT=$last9_otlp_endpoint
env = OTEL_EXPORTER_OTLP_HEADERS=Authorization=$last9_otlp_auth_header
env = OTEL_RESOURCE_ATTRIBUTES=deployment.environment=production

Start with:

opentelemetry-instrument uwsgi --ini uwsgi.ini

Error Handling and Custom Hooks

import logging
from opentelemetry.trace import Status, StatusCode


class ErrorHandlingMiddleware:
    def __init__(self):
        self.tracer = trace.get_tracer(__name__)
        self.logger = logging.getLogger(__name__)


    def process_request(self, req, resp):
        # Add request ID for tracing
        req.context.request_id = req.headers.get('X-Request-ID', 'unknown')


    def process_response(self, req, resp, resource, req_succeeded):
        span = trace.get_current_span()


        if span.is_recording():
            span.set_attribute("request.id", req.context.request_id)


            if not req_succeeded:
                span.set_status(Status(StatusCode.ERROR, "Request processing failed"))


# Custom error handlers
def handle_404(ex, req, resp, params):
    span = trace.get_current_span()
    if span.is_recording():
        span.set_attribute("http.status_code", 404)
        span.set_attribute("error.type", "not_found")


def handle_500(ex, req, resp, params):
    span = trace.get_current_span()
    if span.is_recording():
        span.record_exception(ex)
        span.set_status(Status(StatusCode.ERROR, str(ex)))


# Add error handlers
app.add_error_handler(falcon.HTTPNotFound, handle_404)
app.add_error_handler(Exception, handle_500)

Troubleshooting

Common Issues

  1. No traces appearing:

    • Verify environment variables are correctly set
    • Check Last9 endpoint connectivity
    • Enable debug logging: export OTEL_LOG_LEVEL=debug

  2. Missing request spans:

    • Ensure FalconInstrumentor is properly applied
    • Check that automatic instrumentation is running

  3. Performance impact:

    • Use sampling: export OTEL_TRACES_SAMPLER_ARG=0.1
    • Monitor memory usage with instrumentation enabled
    • Consider using BatchSpanProcessor for production

Debug Mode

Enable detailed logging:

import logging
logging.getLogger("opentelemetry").setLevel(logging.DEBUG)

Or via environment:

export OTEL_LOG_LEVEL=debug

Monitoring Capabilities

This integration automatically captures:

  • HTTP Requests: All incoming API requests to resources
  • Resource Methods: Individual method execution (on_get, on_post, etc.)
  • Middleware Execution: Processing time and middleware chain
  • Database Operations: SQL queries and transactions
  • External HTTP Calls: Outbound requests via requests library
  • Exception Tracking: Detailed error information and stack traces
  • Custom Business Logic: Through manual instrumentation

Best Practices

  1. Resource Naming: Use descriptive resource class names
  2. Custom Attributes: Add meaningful business context to spans
  3. Error Handling: Implement proper error recording in spans
  4. Middleware Order: Place tracing middleware early in the chain
  5. Database Connections: Create connections after instrumentation
  6. Sampling: Configure appropriate sampling rates for production

Your Falcon application will now provide comprehensive telemetry data to Last9, enabling detailed performance monitoring of your high-performance WSGI APIs.