Balena API Interaction
The Balena API interaction code is responsible for retrieving metrics from the Balena cloud. It handles authentication, API requests, and data parsing to provide a Prometheus-compatible output.
Authentication
The Balena API requires authentication using a Balena API token. The token is stored in the BALENA_TOKEN environment variable.
BALENA_TOKEN = os.getenv("BALENA_TOKEN", False)
Retrieving Fleets
The get_balena_fleets function retrieves all the fleets that the user has access to. It uses the Balena API endpoint /v6/application with a filter to select only directly accessible fleets.
def get_balena_fleets(self):
"""
Get all the fleets that the user has access to
and return all corresponding fleet IDs
"""
fleets = []
headers = {
"Authorization": f"Bearer {BALENA_TOKEN}",
"Content-Type": "application/json",
}
response = requests.get(
"https://api.balena-cloud.com/v6/application?$filter=is_directly_accessible_by__user/any(dau:1 eq 1)",
headers=headers,
)
if not response.ok:
print("Error: {}".format(response.text))
sys.exit(1)
for fleet in response.json()["d"]:
fleets.append((fleet["id"]))
return fleets
Retrieving Fleet Metrics
The get_fleet_metrics function retrieves the number of online devices in a given fleet. It uses the Balena API endpoint /v6/application/{fleet_id} with an expansion to include the count of online devices in the fleet.
def get_fleet_metrics(self, fleet_id):
headers = {
"Authorization": f"Bearer {BALENA_TOKEN}",
"Content-Type": "application/json",
}
response = requests.get(
f"https://api.balena-cloud.com/v6/application({fleet_id})?$expand=owns__device/$count($filter=is_online eq true)",
headers=headers,
)
if not response.ok:
print("Error: {}".format(response.text))
sys.exit(1)
device_online_count = response.json()["d"][0]["owns__device"]
fleet_name = response.json()["d"][0]["app_name"]
return fleet_name, device_online_count
Data Collection
The collect function gathers metrics for each fleet and constructs a Prometheus metric family. It iterates through the list of fleets, retrieves metrics for each fleet using get_fleet_metrics, and adds the metrics to the GaugeMetricFamily object.
def collect(self):
gauge = GaugeMetricFamily(
"balena_devices_online", "Devices by status", labels=["fleet_name"]
)
for fleet_id in self.get_balena_fleets():
fleet_name, device_online_count = self.get_fleet_metrics(str(fleet_id))
gauge.add_metric([fleet_name], float(device_online_count))
return [gauge]
Data Exporter
The main function of the code starts an HTTP server on port 8000 and registers the BalenaCollector with the Prometheus registry. The while True loop sleeps for the specified CRAWL_INTERVAL before collecting metrics again. The CRAWL_INTERVAL is set to 60 seconds by default, but can be overridden with the CRAWL_INTERVAL environment variable.
def main():
if not BALENA_TOKEN:
print("Please set the BALENA_TOKEN environment variable")
sys.exit(1)
start_http_server(8000)
REGISTRY.register(BalenaCollector())
while True:
time.sleep(int(CRAWL_INTERVAL))
Testing
Unit tests are available in the tests/test_exporter.py file. The tests cover the functionality of the get_balena_fleets, get_fleet_metrics, and collect functions.
def setUp(self):
self.collector = BalenaCollector()
class TestBalenaCollector(unittest.TestCase):
def setUp(self):
self.collector = BalenaCollector()
@mock.patch("main.BalenaCollector.get_fleet_metrics")
def test_collect(self, mock_metrics):
with mock.patch.object(self.collector, "get_balena_fleets") as mock_fleets:
mock_fleets.return_value = ["fleet1", "fleet2"]
mock_metrics.side_effect = [("test_fleet", 3), ("test_fleet", 3)]
result = list(self.collector.collect()[0].samples)
expected = [('balena_devices_online', {'fleet_name': 'test_fleet'}, 3),
('balena_devices_online', {'fleet_name': 'test_fleet'}, 3)]
result = [(s.name, s.labels, s.value) for s in result]
self.assertEqual(result, expected)
def test_get_balena_fleets(self):
with mock.patch("main.requests.get") as mock_get:
mock_get.return_value.ok = True
mock_get.return_value.json.return_value = {"d": [{"id": "fleet1"}, {"id": "fleet2"}]}
result = list(self.collector.get_balena_fleets())
expected = ["fleet1", "fleet2"]
self.assertEqual(result, expected)
def test_get_fleet_metrics(self):
with mock.patch("main.requests.get") as mock_get:
mock_get.return_value.ok = True
mock_get.return_value.json.return_value = {"d": [{"owns__device": 3, "app_name": "test_fleet"}]}
result = self.collector.get_fleet_metrics("fleet1")
expected = ("test_fleet", 3)
self.assertEqual(result, expected)
Top-Level Directory Explanations
tests/ - This directory contains all the unit and integration tests for the project. It includes the __init__.py file which makes it a package, and specific test files like test_exporter.py.
