> For the complete documentation index, see [llms.txt](https://docs.ionos.com/cloud/llms.txt). Markdown versions of documentation pages are available by appending `.md` to page URLs; this page is available as [Markdown](https://docs.ionos.com/cloud/tutorials/observability/logging-service/route-activity-logs-to-logging-service.md).

# Route Activity Log API Events to Logging Service with Fluent Bit

## Overview

This tutorial demonstrates how to stream [<mark style="color:blue;">IONOS CLOUD Activity Log</mark>](https://docs.ionos.com/cloud/observability/activity-logs) data into the <code class="expression">space.vars.ionos\_cloud</code> 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 <code class="expression">space.vars.ionos\_cloud</code> 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 <code class="expression">space.vars.ionos\_cloud</code> Console and API authentication

## What you will learn

* How to create an <code class="expression">space.vars.ionos\_cloud</code> 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 <code class="expression">space.vars.ionos\_cloud</code> 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 <code class="expression">space.vars.ionos\_cloud</code> account with permission to view contracts and create Logging Pipelines
* An <code class="expression">space.vars.ionos\_cloud</code> API token with read access to the Activity Log API. Generate one from the [<mark style="color:blue;">Token Manager</mark>](https://docs.ionos.com/cloud/set-up-ionos-cloud/management/identity-access-management/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 <code class="expression">space.vars.ionos\_cloud</code> Logging Service. See the [<mark style="color:blue;">Logging Service documentation</mark>](https://docs.ionos.com/cloud/observability/logging-service) for setup details.

## Cost considerations

This tutorial creates billable resources, including an <code class="expression">space.vars.ionos\_cloud</code> 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 [<mark style="color:blue;">IONOS CLOUD price list (EUR)</mark>](https://docs.ionos.com/cloud/support/general-information/price-list/ionos-cloud-eur-en).

## Procedure

### Architecture

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

```mermaid
flowchart LR
    API["IONOS Activity<br/>Log API"]
    PY["Python fetcher<br/>script"]
    FB["Fluent Bit<br/>exec input + JSON parser"]
    DBG["stdout<br/>(debug only)"]
    LOKI["IONOS Logging Service<br/>(Loki)"]
    GRAF["Grafana<br/>Explore + LogQL"]

    API -->|HTTPS bearer token| PY
    PY -->|JSON event lines| FB
    FB --> DBG
    FB -->|HTTPS POST| LOKI
    LOKI --> GRAF
```

The pipeline consists of three components:

* A **Python fetcher script** queries the <code class="expression">space.vars.ionos\_cloud</code> 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 <code class="expression">space.vars.ionos\_cloud</code> Logging Service (for centralized analysis).
* **Grafana** queries and filters the stored logs using `LogQL`.

### Step 1: Create an IONOS CLOUD Logging Pipeline

Follow the [<mark style="color:blue;">Create a Logging Service pipeline</mark>](https://docs.ionos.com/cloud/observability/logging-service/dcd-how-tos/create-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`

{% hint style="warning" %}
**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.
  {% endhint %}

### Step 2: Install Fluent Bit on the VM

{% stepper %}
{% step %}
**Run the official Fluent Bit installer.**

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

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

{% endstep %}

{% step %}
**Add the IONOS CLOUD API token to the service configuration.**

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

```bash
sudo systemctl edit fluent-bit
```

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

```ini
[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.
{% endstep %}

{% step %}
**Start and enable the service.**

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

Verify that the service is `active (running)` before continuing.
{% endstep %}
{% endstepper %}

### Step 3: Install Python prerequisites

Install `python3` and the `requests` library:

```bash
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:

```bash
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:

```python
#!/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:

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

(Optional) Test the script manually:

```bash
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:

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

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

```ini
[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:

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

Append the following sections at the bottom of the file:

```ini
[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.

{% hint style="info" %}
**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.
{% endhint %}

Restart Fluent Bit to apply the changes:

```bash
sudo systemctl restart fluent-bit
```

### Step 7: Test end-to-end ingestion

{% stepper %}
{% step %}
**Run the script manually.**

Confirm the script prints JSON lines:

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

{% endstep %}

{% step %}
**Inspect Fluent Bit output.**

Follow the service logs:

```bash
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.
{% endstep %}

{% step %}
**Query Loki in Grafana.**

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

```logql
{tag_name="<your_TAG_NAME>"}
```

For example:

```logql
{tag_name="activityLog"}
```

Activity events from your contracts appear in the results panel.
{% endstep %}
{% endstepper %}

### Final result

You have built a complete pipeline that pulls <code class="expression">space.vars.ionos\_cloud</code> 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:**

```logql
{tag_name="activityLog"} |= "\"username\":\"<user@example.com>\""
```

**Filter by `principal.sourceService`:**

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

**Filter by event type:**

```logql
{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 <code class="expression">space.vars.ionos\_cloud</code> 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 [<mark style="color:blue;">Logging Service documentation</mark>](https://docs.ionos.com/cloud/observability/logging-service) for retention and security options.
* Explore the [<mark style="color:blue;">Activity Log API reference</mark>](https://api.ionos.com/docs/activitylog/v1/) to extend the fetcher with additional metadata.
* For one-off exports without a streaming pipeline, see [<mark style="color:blue;">Download activity log entries</mark>](https://docs.ionos.com/cloud/observability/activity-logs/how-tos/download-activity-log-entries).


---

# Agent Instructions
This documentation is published with GitBook. GitBook is the documentation platform designed so that both humans and AI agents can read, navigate, and reason over technical content effectively. Learn more at gitbook.com.

## Querying This Documentation
If you need additional information that is not directly available in this page, you can query the documentation dynamically by asking a question.

Perform an HTTP GET request on the current page URL with the `ask` query parameter:

```
GET https://docs.ionos.com/cloud/tutorials/observability/logging-service/route-activity-logs-to-logging-service.md?ask=<question>
```

The question should be specific, self-contained, and written in natural language.
The response will contain a direct answer to the question and relevant excerpts and sources from the documentation.

Use this mechanism when the answer is not explicitly present in the current page, you need clarification or additional context, or you want to retrieve related documentation sections.