Route Activity Log API Events to Logging Service with Fluent Bit

Overview

This tutorial demonstrates how to stream IONOS CLOUD Activity Log data into the IONOS CLOUD Logging Service (powered by Loki) using Fluent Bit. You will build an automated pipeline that uses a Python script to fetch activity events, forwards them as structured JSON, and enables real-time querying in Grafana.

Target audience

This tutorial benefits cloud administrators, DevOps engineers, and security operators who manage IONOS CLOUD contracts and want a centralized visibility into account activity. Readers benefit from basic familiarity with:

• Linux administration (systemd, shell commands, file editing)

• Python scripting

• Log aggregation concepts and Grafana queries

• The IONOS CLOUD Console and API authentication

What you will learn

• How to create an IONOS CLOUD Logging Pipeline for generic TCP log sources

• How to install and configure Fluent Bit on a Linux virtual machine

• How to fetch Activity Log events from the IONOS CLOUD API with a Python script

• How to parse and forward JSON events to Loki through Fluent Bit

• How to query and filter activity events in Grafana using LogQL

• How to troubleshoot common ingestion problems

Before you begin

Ensure you have:

• An active IONOS CLOUD account with permission to view contracts and create Logging Pipelines

• An IONOS CLOUD API token with read access to the Activity Log API. Generate one from the Token Manager.

• A Linux virtual machine (Ubuntu 20.04 or later recommended) with internet access and sudo privileges

• Access to a Grafana instance connected to your IONOS CLOUD Logging Service. See the Logging Service documentation for setup details.

Cost considerations

This tutorial creates billable resources, including an IONOS CLOUD Logging Pipeline and a Linux VM that hosts Fluent Bit. Charges incur based on ingested log volume and VM runtime.

Delete the Logging Pipeline and the VM after you finish to avoid ongoing charges. For current rates, see IONOS CLOUD price list (EUR).

Procedure

Architecture

The diagram below shows the data flow from the Activity Log API to Grafana:

The pipeline consists of three components:

• A Python fetcher script queries the IONOS CLOUD Activity Log API for every contract and prints each event as raw JSON.

• Fluent Bit runs the script periodically through the exec input plugin, parses the JSON, and forwards events to both stdout (for debugging) and the IONOS CLOUD Logging Service (for centralized analysis).

• Grafana queries and filters the stored logs using LogQL.

Step 1: Create an IONOS CLOUD Logging Pipeline

Follow the Create a Logging Service pipeline guide and, when you configure the log source, use these values:

• Source type: Generic

• Protocol: HTTP

• Tag: a unique name such as activityLog

Important:

• Save the authentication key immediately. The Logging Service displays the authentication key only once on pipeline creation.

• Store it in a secure location, such as a password manager. Record the HTTPS endpoint shown in the pipeline details to reference it in the Fluent Bit configuration.

Step 2: Install Fluent Bit on the VM

1

Run the official Fluent Bit installer.

The installer detects the Linux distribution and installs the latest stable release.

curl https://raw.githubusercontent.com/fluent/fluent-bit/master/install.sh | sh

2

Add the IONOS CLOUD API token to the service configuration.

Create a systemd drop-in override so the token survives package upgrades:

sudo systemctl edit fluent-bit

Add the following content in the editor that opens, replacing the placeholder with your token:

[Service]
Environment=IONOS_API_TOKEN=<your_IONOS_API_TOKEN>

Save and close. systemd writes the override to /etc/systemd/system/fluent-bit.service.d/override.conf and reloads the unit automatically.

3

Start and enable the service.

sudo systemctl start fluent-bit
sudo systemctl enable fluent-bit
sudo systemctl status fluent-bit

Verify that the service is active (running) before continuing.

Step 3: Install Python prerequisites

Install python3 and the requests library:

sudo apt update
sudo apt install -y python3 python3-pip
pip3 install requests

Step 4: Create the Python fetcher script

Create the script directory, the persistent state directory, and the fetcher file:

sudo mkdir -p /opt/fluent-bit/scripts/ /var/lib/fluent-bit/
sudo vim /opt/fluent-bit/scripts/fetch_activity_logs.py

Paste the following code into the file:

#!/usr/bin/env python3
import os
import sys
import json
import time
import requests
from datetime import datetime, timezone, timedelta
from typing import List, Dict, Any

# === Configuration ===
API_BASE = "https://api.ionos.com/activitylog/v1"
TOKEN = os.getenv("IONOS_API_TOKEN")
STATE_FILE = "/var/lib/fluent-bit/activity_log_state.json"
FETCH_WINDOW = 300  # seconds back if no state file found
LIMIT = 1000
REQUEST_TIMEOUT = 30


# === Helpers ===
def load_state() -> Dict[str, Any]:
    if os.path.exists(STATE_FILE):
        try:
            with open(STATE_FILE, "r") as f:
                return json.load(f)
        except Exception:
            return {}
    return {}


def save_state(state: Dict[str, Any]) -> None:
    with open(STATE_FILE, "w") as f:
        json.dump(state, f)


def zulu(dt: datetime) -> str:
    """Format datetime as YYYY-MM-DDTHH:MM:SSZ (no microseconds)."""
    return dt.astimezone(timezone.utc).strftime("%Y-%m-%dT%H:%M:%SZ")


def list_contracts(session: requests.Session) -> List[int]:
    url = f"{API_BASE}/contracts"
    r = session.get(
        url,
        headers={"Authorization": f"Bearer {TOKEN}"},
        timeout=REQUEST_TIMEOUT,
    )
    r.raise_for_status()
    data = r.json()
    if isinstance(data, list):
        return [c.get("id") for c in data if isinstance(c, dict) and c.get("id") is not None]
    if isinstance(data, dict) and "items" in data and isinstance(data["items"], list):
        return [c.get("id") for c in data["items"] if isinstance(c, dict) and c.get("id") is not None]
    return []


def parse_entries_from_response(resp_json: Any) -> List[Dict[str, Any]]:
    entries: List[Dict[str, Any]] = []
    if isinstance(resp_json, dict):
        if "items" in resp_json and isinstance(resp_json["items"], list):
            for it in resp_json["items"]:
                if isinstance(it, dict):
                    entries.append(it)
        elif "hits" in resp_json and isinstance(resp_json["hits"], dict) and "hits" in resp_json["hits"]:
            for h in resp_json["hits"]["hits"]:
                src = h.get("_source") or h.get("source") or h
                if isinstance(src, dict):
                    entries.append(src)
    elif isinstance(resp_json, list):
        for it in resp_json:
            if isinstance(it, dict):
                entries.append(it)
    return entries


def fetch_logs_for_contract(
    session: requests.Session,
    contract_id: int,
    start_time: datetime,
    end_time: datetime,
    limit: int = LIMIT,
) -> List[Dict[str, Any]]:
    offset = 0
    all_entries: List[Dict[str, Any]] = []
    while True:
        params = {
            "dateStart": zulu(start_time),
            "dateEnd": zulu(end_time),
            "limit": limit,
            "offset": offset,
        }
        url = f"{API_BASE}/contracts/{contract_id}"
        r = session.get(
            url,
            headers={"Authorization": f"Bearer {TOKEN}", "Accept": "application/json"},
            params=params,
            timeout=REQUEST_TIMEOUT,
        )
        r.raise_for_status()
        resp_json = r.json()
        entries = parse_entries_from_response(resp_json)
        if not entries:
            break
        all_entries.extend(entries)
        if len(entries) < limit:
            break
        offset += limit
        time.sleep(0.1)
    return all_entries


# === Main ===
def main():
    if not TOKEN:
        print("Missing IONOS_API_TOKEN environment variable", file=sys.stderr)
        sys.exit(1)
    session = requests.Session()
    state = load_state()
    now = datetime.now(timezone.utc)
    last_run = state.get("last_run")
    if last_run:
        try:
            since = datetime.fromisoformat(last_run)
            if since.tzinfo is None:
                since = since.replace(tzinfo=timezone.utc)
        except Exception:
            since = now - timedelta(seconds=FETCH_WINDOW)
    else:
        since = now - timedelta(seconds=FETCH_WINDOW)
    try:
        contracts = list_contracts(session)
    except Exception as e:
        print(f"Error listing contracts: {e}", file=sys.stderr)
        sys.exit(1)
    for contract in contracts:
        try:
            entries = fetch_logs_for_contract(session, contract, since, now, limit=LIMIT)
            for entry in entries:
                # Print raw JSON entry
                print(json.dumps(entry, ensure_ascii=False))
        except Exception as e:
            print(f"Error fetching contract {contract}: {e}", file=sys.stderr)
    state["last_run"] = now.isoformat()
    save_state(state)


if __name__ == "__main__":
    main()

Make the script executable:

sudo chmod +x /opt/fluent-bit/scripts/fetch_activity_logs.py

(Optional) Test the script manually:

export IONOS_API_TOKEN=<your_API_token>
python3 /opt/fluent-bit/scripts/fetch_activity_logs.py | head -n 3

The script prints up to three JSON event lines if your contracts contain recent activity.

Step 5: Configure the Fluent Bit JSON parser

Open the parser configuration file:

sudo vim /etc/fluent-bit/parsers.conf

Verify that the json parser exists. If not, add the following block:

[PARSER]
    Name        json
    Format      json
    Time_Key    time
    Time_Format %d/%b/%Y:%H:%M:%S %z

Step 6: Configure the Fluent Bit input and outputs

Edit the main configuration file:

sudo vim /etc/fluent-bit/fluent-bit.conf

Append the following sections at the bottom of the file:

[SERVICE]
    Flush        1
    Daemon       Off
    Log_Level    info
    Parsers_File parsers.conf

[INPUT]
    Name         exec
    Command      python3 /opt/fluent-bit/scripts/fetch_activity_logs.py
    Interval_Sec 300
    Tag          <TAG_NAME>
    Parser       json

[OUTPUT]
    Name            http
    Match           <TAG_NAME>
    Host            <HTTPS_ENDPOINT>
    Port            443
    URI             /<TAG_NAME>
    Format          json
    Header          APIKEY <PIPELINE_AUTH_KEY>
    tls             on

[OUTPUT]
    Name   stdout
    Match  <TAG_NAME>

Replace the placeholders:

• <TAG_NAME>: The tag you defined when creating the Logging Pipeline (for example, activityLog). It becomes both the routing match and the URI path.

• <HTTPS_ENDPOINT>: The HTTPS endpoint host returned by the pipeline (no scheme, no path).

• <PIPELINE_AUTH_KEY>: The authentication key from pipeline creation. It is sent in the APIKEY header on every request.

Remove the stdout output for production

The second [OUTPUT] block writes events to stdout for debugging. Remove it once you confirm that events reach Loki.

Restart Fluent Bit to apply the changes:

sudo systemctl restart fluent-bit

Step 7: Test end-to-end ingestion

1

Run the script manually.

Confirm the script prints JSON lines:

python3 /opt/fluent-bit/scripts/fetch_activity_logs.py | head -n 3

2

Inspect Fluent Bit output.

Follow the service logs:

sudo journalctl -u fluent-bit -f

JSON entries appear through the stdout output. Wait up to 300 seconds (the Interval_Sec value) or lower the interval temporarily for faster feedback.

3

Query Loki in Grafana.

In Grafana, open Explore, select the Loki data source, and run:

{tag_name="<your_TAG_NAME>"}

For example:

{tag_name="activityLog"}

Activity events from your contracts appear in the results panel.

Final result

You have built a complete pipeline that pulls IONOS CLOUD Activity Log events into the Logging Service. Fluent Bit runs the Python fetcher every five minutes, parses each JSON event, and forwards it to Loki through a secure TCP endpoint. Grafana queries return live activity data across all your contracts.

Grafana queries

Use these LogQL examples in Grafana Explore to filter activity events.

Filter by username:

{tag_name="activityLog"} |= "\"username\":\"<[email protected]>\""

Filter by principal.sourceService:

{tag_name="activityLog"} |= "\"sourceService\":\"DCD\""
{tag_name="activityLog"} |= "\"sourceService\":\"Provisioning\""
{tag_name="activityLog"} |= "\"sourceService\":\"PUBLIC_REST\""

Filter by event type:

{tag_name="activityLog"} |= "\"type\":\"Provision\""
{tag_name="activityLog"} |= "\"type\":\"Error\""

Troubleshooting checklist

Symptom	Resolution
Token missing	Verify the systemd drop-in sets IONOS_API_TOKEN. Run sudo systemctl edit fluent-bit, then sudo systemctl restart fluent-bit.
Parser errors	Confirm /etc/fluent-bit/parsers.conf exists and Parsers_File parsers.conf appears in the [SERVICE] section.
Events visible in stdout but not in Loki	Re-check the http output: Host, Port, URI, and the APIKEY header value must match the pipeline.
exec input does not run	Ensure the script is executable (chmod +x) and starts with #!/usr/bin/env python3.
Missed events after a restart	The script stores the last run timestamp in /var/lib/fluent-bit/activity_log_state.json. Delete the file to force a fresh fetch: sudo rm /var/lib/fluent-bit/activity_log_state.json.

Conclusion

You created an automated pipeline that streams IONOS CLOUD Activity Log data into the Logging Service and visualizes it in Grafana. The setup gives your team a single place to audit user actions, detect errors, and investigate provisioning events across all contracts.

Next steps:

• Build Grafana dashboards that group events by user, service, or contract.

• Configure Grafana alerts on critical event types such as Error.

• Review the Logging Service documentation for retention and security options.

• Explore the Activity Log API reference to extend the fetcher with additional metadata.

• For one-off exports without a streaming pipeline, see Download activity log entries.