Welcome back! In this lesson, we will focus on the critical aspects of monitoring, logging, and observability in distributed systems. These practices help you track system performance, monitor workflow failures, and gain visibility into the health and behavior of your applications. For large-scale systems like ERP/CRM platforms, having a comprehensive monitoring and logging infrastructure is essential for debugging, troubleshooting, and optimizing performance.
Table of Contents
-
Introduction to Observability in Distributed Systems
-
Setting Up Logging in Flask Applications
-
Distributed Logging with Elastic Stack (ELK)
-
Monitoring Workflows and Tasks with Prometheus and Grafana
-
Adding Observability to the DSL
-
Practical Example: Monitoring an Order Fulfillment System
-
Conclusion and Next Steps
1. Introduction to Observability in Distributed Systems
What is Observability?
Observability is the practice of understanding the internal state of a system by collecting and analyzing outputs such as logs, metrics, and traces. In distributed systems, observability is critical because services often span multiple nodes and systems, making it harder to pinpoint issues without visibility into each part of the system.
Key Components of Observability:
-
Logging: Capturing detailed logs of system events, errors, and informational messages.
-
Metrics: Measuring system performance, including resource usage, task completion times, error rates, etc.
-
Tracing: Tracking requests as they traverse through multiple services to provide insights into how different components interact.
-
Alerting: Setting up notifications for critical system events or failures (e.g., task failure rates, service downtime).
Why Observability Matters for Workflows:
For systems that implement complex workflows, such as order processing or approval chains, observability allows you to:
-
Monitor Workflow Progress: Track the flow of tasks through different states.
-
Detect Failures Early: Identify and alert on failed tasks or services before they impact users.
-
Optimize Performance: Analyze task execution times and resource usage to improve system performance.
2. Setting Up Logging in Flask Applications
Basic Logging in Flask
Flask supports logging out of the box, and you can easily configure it to log messages at different levels: DEBUG, INFO, WARNING, ERROR, and CRITICAL.
Step 1: Configuring Flask Logging
Here’s how to set up logging in a Flask application:
import logging
from flask import Flask
app = Flask(name)
Configure basic logging
logging.basicConfig(filename=‘app.log’, level=logging.INFO)
@app.route(’/’)
def home():
app.logger.info(‘Home page accessed’)
return “Welcome to the home page!”
@app.route(‘/order/int:order_id/fulfill’, methods=
)
def fulfill_order(order_id):
try:
app.logger.info(f’Fulfilling order {order_id}’)
Workflow processing logic here
return “Order fulfilled”
except Exception as e:
app.logger.error(f’Error fulfilling order {order_id}: {str(e)}’)
return “Failed to fulfill order”, 500
Explanation:
-
logging.basicConfig(): Configures the logging output, log level, and log file location.
-
app.logger: Flask’s logger is used to log messages at different levels (INFO, ERROR, etc.).
Step 2: Capturing Workflow Logs
For each transition or state change in a workflow, you should log relevant information, such as the current state, the task being performed, and any errors that occur.
Example: Logging in the Workflow Code:
@app.route(‘/order/int:order_id/fulfill’, methods=
)
def fulfill_order(order_id):
workflow = OrderProcessingWorkflow()
try:
workflow.fulfill(‘FulfillmentSystem’)
app.logger.info(f’Order {order_id} transitioned to Fulfilled’)
Run distributed tasks
notify_warehouse.delay(order_id)
notify_customer.delay(order_id)
app.logger.info(f’Concurrent tasks for order {order_id} started’)
Update order status
order = Order.query.get(order_id)
order.state = ‘Completed’
db.session.commit()
app.logger.info(f’Order {order_id} marked as Completed’)
return jsonify({‘status’: ‘success’, ‘new_state’: ‘Completed’}), 200
except Exception as e:
app.logger.error(f’Error fulfilling order {order_id}: {str(e)}’)
return jsonify({‘status’: ‘error’, ‘message’: str(e)}), 500
3. Distributed Logging with Elastic Stack (ELK)
What is the ELK Stack?
The ELK Stack (Elasticsearch, Logstash, Kibana) is a powerful toolset for managing logs in distributed systems:
-
Elasticsearch: A distributed search engine that stores logs.
-
Logstash: A log pipeline tool that ingests logs and sends them to Elasticsearch.
-
Kibana: A visualization tool for exploring logs stored in Elasticsearch.
Step 1: Setting Up the ELK Stack
- Install Elasticsearch, Logstash, and Kibana. You can install them using Docker:
docker run -d —name elasticsearch -p 9200:9200 -e “discovery.type=single-node” elasticsearch:7.10.0
docker run -d —name logstash -p 5044:5044 logstash:7.10.0
docker run -d —name kibana -p 5601:5601 kibana:7.10.0
- Configure Logstash to receive logs from your Flask application. Create a configuration file (logstash.conf):
input {
file {
path => “/path/to/app.log”
start_position => “beginning”
}
}
output {
elasticsearch {
hosts =>
index => “flask-logs-%{+YYYY.MM.dd}”
}
}
- Start Logstash with the configuration file:
docker exec -it logstash logstash -f /path/to/logstash.conf
Step 2: Sending Logs to Logstash
You can configure your Flask app to send logs to Logstash by writing to a file that Logstash monitors.
import logging
logging.basicConfig(filename=‘app.log’, level=logging.INFO)
Now, any logs generated by the Flask app will be picked up by Logstash and stored in Elasticsearch.
Step 3: Visualizing Logs with Kibana
Open Kibana at http://localhost:5601, create an index pattern (flask-logs-*), and explore the logs generated by your Flask application.
4. Monitoring Workflows and Tasks with Prometheus and Grafana
What is Prometheus?
Prometheus is an open-source system monitoring and alerting toolkit. It collects metrics from different services and stores them in a time-series database. Prometheus is often paired with Grafana for visualization.
Step 1: Setting Up Prometheus
Install Prometheus using Docker:
docker run -d —name prometheus -p 9090:9090 prom/prometheus
Configure Prometheus to scrape metrics from your Flask application by adding a prometheus.yml configuration file:
scrape_configs:
- job_name: ‘flask-app’
static_configs:
- targets:
Step 2: Exposing Metrics in Flask
Install Prometheus-Flask-Exporter to expose metrics from your Flask app:
pip install prometheus-flask-exporter
Add the metrics exporter to your Flask app:
from prometheus_flask_exporter import PrometheusMetrics
app = Flask(name)
metrics = PrometheusMetrics(app)
@app.route(’/’)
def home():
return “Welcome to the home page!”
@app.route(‘/order/int:order_id/fulfill’, methods=
)
def fulfill_order(order_id):
workflow = OrderProcessingWorkflow()
workflow.fulfill(‘FulfillmentSystem’)
return “Order fulfilled”
Step 3: Monitoring Workflow Metrics
Prometheus will automatically scrape metrics such as request counts, latencies, and error rates from your Flask app. You can also define custom metrics to monitor specific workflow steps:
from prometheus_client import Counter
order_fulfilled = Counter(‘order_fulfilled’, ‘Number of orders fulfilled’)
@app.route(‘/order/int:order_id/fulfill’, methods=
)
def fulfill_order(order_id):
workflow = OrderProcessingWorkflow()
workflow.fulfill(‘FulfillmentSystem’)
order_fulfilled.inc() # Increment the counter each time an order is fulfilled
return “Order fulfilled”
Step 4: Visualizing Metrics with Grafana
- Install Grafana using Docker:
docker run -d —name grafana -p 3000:3000 grafana/grafana
- Configure Prometheus as a data source in Grafana, and create dashboards to visualize metrics such as:
-
Number of orders fulfilled.
-
Average task completion times.
-
Error rates for distributed tasks.
5. Adding Observability to the DSL
Extending the DSL to Define Metrics and Alerts
We can extend the DSL to allow non-technical users to define metrics and alerts for workflows. For example:
workflow OrderProcessing {
state
Pending {
transition approve -> Approved by Manager;
transition reject -> Rejected by Manager;
}
state Approved {
transition fulfill -> Fulfilled by FulfillmentSystem;
metric ordersFulfilled;
alert onFailure -> NotifyAdmin if errorCount > 5;
}
}
Explanation:
-
Metric Definition: The metric keyword defines a custom metric (ordersFulfilled) to track how many orders have been fulfilled.
-
Alert Definition: The alert keyword defines an alert that triggers when the error count exceeds a threshold (errorCount > 5).
Step 1: Generating Monitoring Code for Metrics and Alerts
The DSL can generate code to expose metrics and configure alerts. For example, metrics can be generated using Prometheus, and alerts can be configured in Prometheus Alertmanager.
Example: Generating Metric and Alert Code:
from prometheus_client import Counter, Gauge
orders_fulfilled = Counter(‘orders_fulfilled’, ‘Number of orders fulfilled’)
error_count = Gauge(‘error_count’, ‘Number of errors encountered’)
@app.route(‘/order/int:order_id/fulfill’, methods=
)
def fulfill_order(order_id):
try:
workflow = OrderProcessingWorkflow()
workflow.fulfill(‘FulfillmentSystem’)
orders_fulfilled.inc()
except Exception as e:
error_count.inc()
if error_count._value.get() > 5:
notify_admin(f”Error threshold exceeded for order {order_id}”)
raise e
This code automatically tracks workflow metrics and sends alerts when errors exceed a predefined threshold.
6. Practical Example: Monitoring an Order Fulfillment System
Let’s bring together everything we’ve covered---logging, metrics, and alerts---to implement a monitored and observable order fulfillment system.
DSL for Observable Order Workflow:
workflow OrderProcessing {
state Pending {
transition approve -> Approved by Manager;
transition reject -> Rejected by Manager;
}
state Approved {
transition fulfill -> Fulfilled by FulfillmentSystem;
metric ordersFulfilled;
alert onFailure -> NotifyAdmin if errorCount > 5;
}
state Shipped {
// End of workflow
}
}
Generated Code with Metrics and Alerts:
from prometheus_client import Counter, Gauge
orders_fulfilled = Counter(‘orders_fulfilled’, ‘Number of orders fulfilled’)
error_count = Gauge(‘error_count’, ‘Number of errors encountered’)
@app.route(‘/order/int:order_id/fulfill’, methods=
)
def fulfill_order(order_id):
try:
workflow = OrderProcessingWorkflow()
workflow.fulfill(‘FulfillmentSystem’)
orders_fulfilled.inc()
app.logger.info(f’Order {order_id} fulfilled’)
except Exception as e:
error_count.inc()
app.logger.error(f’Error fulfilling order {order_id}: {str(e)}’)
if error_count._value.get() > 5:
notify_admin(f”Error threshold exceeded for order {order_id}”)
return jsonify({‘status’: ‘error’, ‘message’: str(e)}), 500
This system logs workflow events, tracks metrics for fulfilled orders, and triggers alerts when error thresholds are exceeded.
7. Conclusion and Next Steps
In this lesson, we’ve:
-
Set up logging in Flask and integrated it with the ELK stack for distributed log management.
-
Implemented monitoring using Prometheus and Grafana for tracking metrics and alerts.
-
Extended the DSL to define metrics and alerts for workflows.
-
Built a practical example of an observable order fulfillment system.
This approach gives you visibility into the health and performance of your system, making it easier to detect issues, optimize workflows, and ensure that tasks are executed smoothly.
Practical Exercise
Task: Extend the DSL to support tracing across multiple services. Implement distributed tracing using a tool like Jaeger to track requests across different microservices involved in a workflow.
Hints:
-
Use OpenTelemetry to instrument your services for tracing.
-
Visualize traces using Jaeger to follow the flow of requests between services.
Looking Ahead
In the next lesson, we’ll explore security best practices for distributed systems, focusing on securing workflows, managing access control, and protecting sensitive data.
Feel free to reach out if you have any questions. See you in the next lesson!