-
Notifications
You must be signed in to change notification settings - Fork 0
/
server.py
124 lines (105 loc) · 4.23 KB
/
server.py
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
from flask import Flask, request, jsonify, Response
from functools import wraps
import os
import logging
from dotenv import load_dotenv
from prometheus_client import CollectorRegistry, Gauge, generate_latest, CONTENT_TYPE_LATEST
# Load environment variables from .env file
load_dotenv()
app = Flask(__name__)
# Setup logging
logging.basicConfig(level=logging.DEBUG) # Set to DEBUG for detailed output
logger = logging.getLogger(__name__)
logger.setLevel(logging.DEBUG)
# Retrieve the credentials from the environment variables
USERNAME = os.getenv("FLASK_USERNAME")
PASSWORD = os.getenv("FLASK_PASSWORD")
FLASK_HOST = os.getenv("FLASK_HOST")
FLASK_PORT = os.getenv("FLASK_PORT")
# Create Prometheus registry
registry = CollectorRegistry()
# Define your metrics (Gauges, Counters, etc.)
location_lat = Gauge('device_latitude', 'Latitude of the device', registry=registry)
location_lon = Gauge('device_longitude', 'Longitude of the device', registry=registry)
battery_level = Gauge('device_battery_level', 'Battery level of the device', registry=registry)
velocity = Gauge('device_velocity', 'Velocity of the device', registry=registry)
altitude = Gauge('device_altitude', 'Altitude of the device', registry=registry)
accuracy = Gauge('device_accuracy', 'Accuracy of the device', registry=registry)
prometheus_obj_and_parameters = [(
location_lat, 'lat'),
(location_lon, 'lon'),
(battery_level, 'batt'),
(velocity, 'vel'),
(altitude, 'alt'),
(accuracy, 'acc')
]
def check_auth(username, password):
"""Check if a username/password combination is valid."""
return username == USERNAME and password == PASSWORD
def authenticate():
"""Sends a 401 response that enables basic auth"""
return Response(
'Could not verify your access level for that URL.\n'
'You have to login with proper credentials', 401,
{'WWW-Authenticate': 'Basic realm="Login Required"'})
def requires_auth(f):
@wraps(f)
def decorated(*args, **kwargs):
auth = request.authorization
if not auth or not check_auth(auth.username, auth.password):
return authenticate()
return f(*args, **kwargs)
return decorated
@app.route('/health', methods=['GET'])
def health_check():
return jsonify({"status": "success", "message": "Server is healthy"}), 200
@app.route('/metrics', methods=['GET'])
@requires_auth
def metrics():
logger.debug("Metrics endpoint reached")
# Expose the metrics for Prometheus scraping
return Response(generate_latest(registry), mimetype=CONTENT_TYPE_LATEST)
def process_json(data):
def process_value(value):
try:
return float(value)
except ValueError:
return 0
for prometheus_obj, key in prometheus_obj_and_parameters:
try:
value = process_value(data[key])
prometheus_obj.set(value)
logger.debug(f"Set {prometheus_obj} to {value}")
except:
logger.error(f"Error setting {prometheus_obj} to {value}")
return True
@app.route('/api', methods=['POST'])
@requires_auth
def receive_json():
if request.is_json:
data = request.get_json()
# Check if the data is a list (i.e., multiple JSONs sent at once)
if isinstance(data, list):
logger.debug(f"Received a batch of {len(data)} JSON objects.")
successes, failures = 0, 0
for item in data:
if process_json(item):
successes += 1
else:
failures += 1
return jsonify({
"status": "success",
"message": f"Processed {successes} JSON objects, {failures} failed."
}), 200
# Handle single JSON object
elif isinstance(data, dict):
logger.debug("Received a single JSON object.")
if process_json(data):
return jsonify({"status": "success", "message": "Processed JSON"}), 200
else:
return jsonify({"status": "failure", "message": "Error processing JSON"}), 400
else:
logger.warning("Request body is not JSON.")
return jsonify({"status": "failure", "message": "Request body must be JSON"}), 400
if __name__ == '__main__':
app.run(host=FLASK_HOST, port=FLASK_PORT)