20250213 ‐ set up python monitoring - cywongg/2025 GitHub Wiki

To set up a heartbeat for your Python application and monitor its status based on logs, follow these steps with minimal changes:

1. Add Heartbeat Logging to the Python App

Modify your application to log a heartbeat at regular intervals using a background thread.

import logging
import threading
import time

logger = logging.getLogger(__name__)

def heartbeat():
    """Logs heartbeat every 30 seconds."""
    while True:
        logger.info("HEARTBEAT")
        time.sleep(30)  # Adjust interval as needed

# Start the heartbeat thread on application startup
heartbeat_thread = threading.Thread(target=heartbeat, daemon=True)
heartbeat_thread.start()

Key Points:

• Minimal code change: Adds a daemon thread that logs "HEARTBEAT" every 30 seconds.
• Ensure your application already logs errors/warnings (e.g., using logger.error("...")).

---

2. Set Up Filebeat on the Machine

Install and configure Filebeat to ship logs to Elasticsearch.

Install Filebeat

# On Debian/Ubuntu
curl -L -O https://artifacts.elastic.co/downloads/beats/filebeat/filebeat-8.7.1-amd64.deb
sudo dpkg -i filebeat-8.7.1-amd64.deb

# Or use your system's package manager

Configure Filebeat

Edit /etc/filebeat/filebeat.yml:

filebeat.inputs:
- type: filestream
  enabled: true
  paths:
    - /path/to/your/python_app.log  # Path to your application log

output.elasticsearch:
  hosts: ["your-elasticsearch-host:9200"]
  username: "elastic"  # If authentication is enabled
  password: "password"

Start Filebeat

sudo systemctl enable filebeat
sudo systemctl start filebeat

---

3. Create Elasticsearch/Kibana Alerts

Use Kibana to set up rules that check for missing heartbeats or errors.

a. Create a Data View in Kibana

• Go to Stack Management > Data Views > Create Data View.
• Name: python_app_logs, Index pattern: filebeat-*.

b. Set Up Heartbeat Alert

• Go to Observability > Alerts > Manage Rules > Create Rule.
• Rule Type: "Elasticsearch query".
• Query:

  {
    "query": {
      "bool": {
        "must_not": {
          "exists": { "field": "message" }
        },
        "filter": [
          { "term": { "message": "HEARTBEAT" } },
          { "range": { "@timestamp": { "gte": "now-35s" } } }
        ]
      }
    }
  }
  

• Check every: 30 seconds.
• Alert if: Number of documents is below 1 (no heartbeat in 35s).

c. Set Up Error/Warning Alert

• Create another rule with this query:

  {
    "query": {
      "bool": {
        "should": [
          { "match": { "log.level": "ERROR" } },
          { "match": { "log.level": "WARN" } }
        ]
      }
    }
  }
  

• Alert if: Number of documents is above 0.

d. Trigger Actions (Optional)

• Send notifications (Email, Slack) or write to a status index using Kibana Actions.

---

4. Verify the Setup

1. Check Logs in Kibana:
   • Use Discover to see HEARTBEAT entries and errors.
2. Test Alerts:
   • Stop the Python app to trigger the heartbeat alert.
   • Inject a test error log to trigger the error alert.

---

Minimal Machine Changes

• Only requires:
  1. Installing/configuring Filebeat.
  2. Ensuring the Python app writes logs to a file.
• No need for exposed ports or HTTP endpoints on the app.

Alternative: Custom Script with Cron

If you prefer not to use Filebeat:

#!/bin/bash
LOG_FILE="/path/to/app.log"
STATUS=$(awk -v d1="$(date -d '35 seconds ago' '+%Y-%m-%dT%H:%M:%S')" '
  /HEARTBEAT/ && $0 > d1 { heartbeat=1 }
  /ERROR|WARN/ { error=1 }
  END { if (!heartbeat || error) print "down"; else print "up" }' "$LOG_FILE")

curl -X POST "http://elasticsearch:9200/app_status/_doc" -H 'Content-Type: application/json' -d '{
  "@timestamp": "'$(date -u +"%Y-%m-%dT%H:%M:%SZ")'",
  "status": "'"$STATUS"'"
}'

Schedule with Cron:

# Run every minute
* * * * * /path/to/script.sh

---

This setup ensures your app's status is monitored via logs and reported to Elasticsearch with minimal changes.