Services API Reference

Overview

This document describes the built-in services provided by Bunker OS that can be invoked from Open World applications through Turmux RPC over vsock communication (documented in Reference RPC Implementation: Turmux).

These pre-installed services provide core functionality for:

• Logging: Store application and system logs in secure storage

• Data Storage: Persistent key-value storage (encrypted at rest)

• Telemetry: Time-series metrics and performance data

• Monitoring: Security events, system state changes, anomalies

Services are accessed using Turmux RPC, which handles message routing, serialization, and delivery. See Reference RPC Implementation: Turmux for the complete Turmux RPC protocol specification and language binding details.

Note

API Stability: These service APIs are still under development. The method signatures, parameter schemas, and return formats may change in future releases. Applications should implement appropriate error handling and version negotiation when calling these services.

Note

Availability: The complete list of services might not be available for all supported target platform. Please refer to platform specific documentation.

Logger Service (logger)

Stores application and system logs with timestamps, severity levels, and structured metadata.

Purpose

• Central logging for Open World applications

• Secure storage (protected from Open World corruption)

• Searchable log database

• Retention policies (automatic archival)

Methods

Note

Implementation: The service methods described below are implemented using Turmux RPC and protobuf messages. The parameter descriptions and return types shown here represent the logical interface. Actual implementation requires defining corresponding .proto message definitions (see Reference RPC Implementation: Turmux for protocol details). Each service’s .proto file is included in the reference image build.

store(log_entry)

Store a log entry.

Parameters:

{
    "level": str,           # "DEBUG", "INFO", "WARNING", "ERROR", "CRITICAL"
    "message": str,         # Log message (max 1000 chars)
    "timestamp": int,       # Unix timestamp (seconds)
    "source": str,          # Application name (max 50 chars)
    "module": str,          # Optional: module/component
    "request_id": str,      # Optional: correlation ID for tracing
    "tags": [str],          # Optional: arbitrary tags for filtering
    "metadata": dict        # Optional: additional context
}

Return:

{
    "log_id": str,          # Unique log ID (UUID)
    "status": "ok",
    "timestamp": int        # Server-assigned timestamp
}

Example (Python):

import turmux_python
import time
from logger_pb2 import LogRequest, LogResponse

# Connect via vsock to Bunker OS services
client = turmux_python.TurmuxClient.connect_vsock(cid=3, port=1024)

# Create and serialize request
request = LogRequest(
    level="ERROR",
    message="Database connection timeout after 30 seconds",
    timestamp=int(time.time()),
    source="db_monitor",
    module="connection_handler",
    request_id="req-12345",
    tags=["database", "network", "timeout"],
    metadata={
        "host": "db.internal",
        "port": 5432,
        "retry_count": 3
    }
)

# Make RPC call
response_bytes = client.call("logger.store", request.SerializeToString())

# Deserialize response
response = LogResponse()
response.ParseFromString(response_bytes)

print(f"Logged with ID: {response.log_id}")

Error Responses:

INVALID_LEVEL: level must be one of DEBUG, INFO, WARNING, ERROR, CRITICAL
MESSAGE_TOO_LONG: message exceeds 1000 characters
STORAGE_FULL: secure storage capacity exceeded

list(filters)

Query stored logs with optional filtering.

Parameters:

{
    "level": str,           # Optional: filter by level
    "source": str,          # Optional: filter by source
    "tags": [str],          # Optional: filter by tags (AND logic)
    "since": int,           # Optional: start timestamp
    "until": int,           # Optional: end timestamp
    "limit": int,           # Optional: max results (default 100, max 1000)
    "offset": int           # Optional: pagination offset (default 0)
}

Return:

{
    "logs": [
        {
            "log_id": str,
            "level": str,
            "message": str,
            "timestamp": int,
            "source": str,
            "module": str,
            "request_id": str,
            "tags": [str],
            "metadata": dict
        },
        ...
    ],
    "total": int,           # Total matching logs (across all pages)
    "offset": int,          # Current offset
    "limit": int            # Current limit
}

Example:

import turmux_python
import time
from logger_pb2 import ListLogsRequest, ListLogsResponse

# Get all ERROR logs from the last hour
one_hour_ago = int(time.time()) - 3600

request = ListLogsRequest(
    level="ERROR",
    since=one_hour_ago,
    limit=50
)

client = turmux_python.TurmuxClient.connect_vsock(cid=3, port=1024)
response_bytes = client.call("logger.list", request.SerializeToString())

response = ListLogsResponse()
response.ParseFromString(response_bytes)

for log in response.logs:
    print(f"[{log.source}] {log.message}")

delete(filters)

Delete logs matching criteria (e.g., older than N days).

Parameters:

{
    "until": int,           # Delete logs before this timestamp
    "level": str,           # Optional: only delete this level
    "source": str,          # Optional: only delete from this source
    "dry_run": bool         # Optional: preview without deleting (default false)
}

Return:

{
    "deleted": int,         # Number of deleted logs
    "remaining": int        # Total logs still stored
}

Storage Service (storage)

Key-value storage with encryption at rest in Bunker OS. Used for:

• Application configuration and state

• Secure credentials (API keys, tokens)

• Binary blobs (certificates, keys)

• Small databases

Purpose

• Persistent, encrypted storage

• Protected from Open World access/modification

• Atomic operations

• TTL (Time-To-Live) support for temporary data

Methods

write(key, value, options)

Store a key-value pair.

Parameters:

{
    "key": str,             # Key name (max 256 chars, alphanumeric + underscore)
    "value": str/bytes,     # Value (string or base64-encoded binary)
    "ttl": int,             # Optional: time-to-live in seconds (0=permanent)
    "tags": [str],          # Optional: for organization
    "metadata": dict        # Optional: additional context
}

Return:

{
    "key": str,
    "status": "ok",
    "stored_bytes": int,
    "created": int,         # Unix timestamp
    "expires": int          # Unix timestamp (if TTL set)
}

Example:

import turmux_python
from storage_pb2 import WriteRequest, WriteResponse

# Store an API key
request = WriteRequest(
    key="api_keys/stripe",
    value="sk_live_51234567890abcdef",
    ttl=86400,  # 24 hours
    tags=["payment", "stripe"],
    metadata={"env": "production", "rotation_date": 1710704400}
)

client = turmux_python.TurmuxClient.connect_vsock(cid=3, port=1024)
response_bytes = client.call("storage.write", request.SerializeToString())

response = WriteResponse()
response.ParseFromString(response_bytes)
print(f"Stored: {response.key}")

read(key)

Retrieve a stored value.

Parameters:

{
    "key": str              # Key name
}

Return:

{
    "key": str,
    "value": str/bytes,     # Base64-encoded if binary
    "created": int,
    "expires": int,         # null if no TTL
    "metadata": dict,
    "tags": [str]
}

Example:

import turmux_python
from storage_pb2 import ReadRequest, ReadResponse

request = ReadRequest(key="api_keys/stripe")

client = turmux_python.TurmuxClient.connect_vsock(cid=3, port=1024)
response_bytes = client.call("storage.read", request.SerializeToString())

response = ReadResponse()
response.ParseFromString(response_bytes)
api_key = response.value

delete(key)

Delete a stored value.

Parameters:

{
    "key": str
}

Return:

{
    "status": "ok",
    "deleted": bool
}

list(prefix)

List all keys matching a prefix.

Parameters:

{
    "prefix": str,          # Key prefix (e.g., "api_keys/")
    "limit": int            # Optional: max results (default 100)
}

Return:

{
    "keys": [
        {
            "key": str,
            "created": int,
            "expires": int,
            "size_bytes": int,
            "tags": [str]
        },
        ...
    ],
    "total": int
}

Telemetry Service (telemetry)

Time-series metrics and performance data. Used for:

• System metrics (CPU, memory, temperature)

• Application metrics (requests, latency, errors)

• Business metrics (transactions, user activity)

• Historical analysis and dashboards

Purpose

• Efficient time-series storage

• Aggregation and summarization

• Long-term retention with compression

• Queryable by time range and tags

Methods

put(metric_name, value, timestamp, tags)

Store a metric data point.

Parameters:

{
    "metric_name": str,     # Metric identifier (e.g., "cpu_usage")
    "value": float,         # Numeric value
    "timestamp": int,       # Unix timestamp (seconds)
    "tags": dict,           # Optional: dimensions (e.g., {"host": "server1", "core": "0"})
    "metadata": dict        # Optional: additional context
}

Return:

{
    "status": "ok",
    "metric_id": str,
    "stored_at": int
}

Example:

import turmux_python
import time
import psutil
from telemetry_pb2 import PutRequest, PutResponse

# Store CPU usage metric
request = PutRequest(
    metric_name="system.cpu.usage",
    value=psutil.cpu_percent(interval=1),
    timestamp=int(time.time()),
    tags={"host": "app-server-01", "instance": "prod"}
)

client = turmux_python.TurmuxClient.connect_vsock(cid=3, port=1024)
response_bytes = client.call("telemetry.put", request.SerializeToString())

response = PutResponse()
response.ParseFromString(response_bytes)

query(metric_name, filters)

Query time-series data with aggregation.

Parameters:

{
    "metric_name": str,     # Which metric to query
    "since": int,           # Start timestamp
    "until": int,           # End timestamp
    "tags_filter": dict,    # Optional: filter by tag values
    "aggregate": str,       # Optional: "avg", "min", "max", "sum", "count"
    "bucket_size": int      # Optional: aggregation interval in seconds
}

Return:

{
    "metric_name": str,
    "datapoints": [
        {
            "timestamp": int,
            "value": float,
            "tags": dict,
            "count": int        # number of points in bucket (if aggregated)
        },
        ...
    ],
    "total_points": int,
    "query_time_ms": int
}

Example:

import turmux_python
import time
from telemetry_pb2 import QueryRequest, QueryResponse

# Get average CPU usage over last 24 hours (5-minute buckets)
one_day_ago = int(time.time()) - 86400

request = QueryRequest(
    metric_name="system.cpu.usage",
    since=one_day_ago,
    until=int(time.time()),
    aggregate="avg",
    bucket_size=300  # 5 minutes
)

client = turmux_python.TurmuxClient.connect_vsock(cid=3, port=1024)
response_bytes = client.call("telemetry.query", request.SerializeToString())

response = QueryResponse()
response.ParseFromString(response_bytes)

for dp in response.datapoints:
    print(f"{dp.timestamp}: {dp.value:.2f}%")

Monitoring Service (monitoring)

System and security events. Used for:

• Security events (unauthorized access attempts)

• State changes (service started/stopped)

• Anomalies (unusual behavior detected)

• Alerts and notifications

Purpose

• Record significant events with context

• Detect patterns and anomalies

• Support incident investigation

• Compliance audit trail

Methods

event(event_type, severity, description, context)

Record a monitoring event.

Parameters:

{
    "event_type": str,      # "SECURITY", "STATE", "PERFORMANCE", "ANOMALY"
    "severity": str,        # "LOW", "MEDIUM", "HIGH", "CRITICAL"
    "description": str,     # Human-readable description
    "timestamp": int,       # Unix timestamp
    "source": str,          # Component generating event
    "context": dict,        # Event-specific details
    "tags": [str]           # Optional: categorization
}

Return:

{
    "event_id": str,        # Unique event ID
    "status": "ok",
    "timestamp": int
}

Event Examples:

Security Event:

import turmux_python
import time
from monitoring_pb2 import EventRequest, EventResponse

request = EventRequest(
    event_type="SECURITY",
    severity="HIGH",
    description="Failed authentication attempt",
    timestamp=int(time.time()),
    source="auth_handler",
    context={
        "username": "admin",
        "ip_address": "192.168.1.100",
        "attempts": 5,
        "action": "login"
    },
    tags=["authentication", "login_failure"]
)

client = turmux_python.TurmuxClient.connect_vsock(cid=3, port=1024)
response_bytes = client.call("monitoring.event", request.SerializeToString())

response = EventResponse()
response.ParseFromString(response_bytes)

State Event:

import turmux_python
import time
from monitoring_pb2 import EventRequest, EventResponse

request = EventRequest(
    event_type="STATE",
    severity="MEDIUM",
    description="Service restarted unexpectedly",
    timestamp=int(time.time()),
    source="systemd",
    context={
        "service": "database",
        "exit_code": 139,
        "uptime_seconds": 3600,
        "restart_count": 3
    },
    tags=["service", "restart"]
)

client = turmux_python.TurmuxClient.connect_vsock(cid=3, port=1024)
response_bytes = client.call("monitoring.event", request.SerializeToString())

response = EventResponse()
response.ParseFromString(response_bytes)

Anomaly Event:

import turmux_python
import time
from monitoring_pb2 import EventRequest, EventResponse

request = EventRequest(
    event_type="ANOMALY",
    severity="CRITICAL",
    description="Unusual memory usage detected",
    timestamp=int(time.time()),
    source="anomaly_detector",
    context={
        "metric": "memory_usage",
        "current_value": 92.5,
        "baseline": 45.0,
        "deviation_percent": 105,
        "threshold": 85.0
    },
    tags=["memory", "performance"]
)

client = turmux_python.TurmuxClient.connect_vsock(cid=3, port=1024)
response_bytes = client.call("monitoring.event", request.SerializeToString())

response = EventResponse()
response.ParseFromString(response_bytes)

query(filters)

Query recorded events.

Parameters:

{
    "event_type": str,      # Optional: filter by type
    "severity": str,        # Optional: minimum severity (HIGH includes HIGH and CRITICAL)
    "since": int,           # Optional: start timestamp
    "until": int,           # Optional: end timestamp
    "source": str,          # Optional: filter by source
    "tags": [str],          # Optional: filter by tags (AND logic)
    "limit": int,           # Optional: max results (default 100)
    "offset": int           # Optional: pagination offset
}

Return:

{
    "events": [
        {
            "event_id": str,
            "event_type": str,
            "severity": str,
            "description": str,
            "timestamp": int,
            "source": str,
            "context": dict,
            "tags": [str]
        },
        ...
    ],
    "total": int
}

Example:

import turmux_python
import time
from monitoring_pb2 import QueryRequest, QueryResponse

# Get all CRITICAL events from last 24 hours
one_day_ago = int(time.time()) - 86400

request = QueryRequest(
    severity="CRITICAL",
    since=one_day_ago,
    limit=50
)

client = turmux_python.TurmuxClient.connect_vsock(cid=3, port=1024)
response_bytes = client.call("monitoring.query", request.SerializeToString())

response = QueryResponse()
response.ParseFromString(response_bytes)

for event in response.events:
    print(f"[{event.source}] {event.description}")

Service Status

All services support a status() method (no parameters):

import turmux_python
from logger_pb2 import StatusRequest, StatusResponse

request = StatusRequest()

client = turmux_python.TurmuxClient.connect_vsock(cid=3, port=1024)
response_bytes = client.call("logger.status", request.SerializeToString())

response = StatusResponse()
response.ParseFromString(response_bytes)

print(f"Service: {response.service}")
print(f"Status: {response.status}")
print(f"Uptime: {response.uptime_seconds}s")

Error Handling

All services follow consistent error handling through Turmux RPC. When a service method encounters an error, the error is communicated back through the Response envelope:

message Response {
  int32 msg_id = 1;            // Correlates with request
  bool success = 2;            // Success flag (false on error)
  string error = 3;            // Error message (populated if success=false)
  bytes payload = 4;           // Response data (if success=true)
}

When a service method fails, the response will have success=false and an error code in the error field. Common error codes across all services:

INVALID_PARAMS      - Parameter validation failed
SERVICE_ERROR       - Internal service error
STORAGE_FULL        - Storage capacity exceeded
TIMEOUT             - Operation timed out
PERMISSION_DENIED   - Not authorized to perform operation
NOT_FOUND           - Requested resource not found

Rate Limiting

Services implement rate limits to prevent abuse and resource exhaustion. The following are reference values that can be configured through the meta-layer (see Meta Layers Reference):

Logger service:    1000 calls/second (per connection)
Storage service:   500 calls/second
Telemetry service: 10000 points/second
Monitoring service: 500 events/second

When a rate limit is exceeded, configurable policies can be applied by Bunker OS, including:

• Throttling: Delay responses to slow down the caller

• Rejection: Return error responses to requests exceeding the limit

• Hypervisor Actions: Request the hypervisor to take corrective action (e.g., reboot Open World VM if deemed necessary)

The specific policy applied depends on the severity of the violation and the meta-layer configuration.

Complete Service Matrix

Services and Methods

Service	Method	Purpose
logger	store()	Store application log entry
logger	list()	Query logs with filters
logger	delete()	Purge old logs
storage	write()	Store key-value pair
storage	read()	Retrieve value by key
storage	delete()	Delete stored value
storage	list()	Enumerate keys by prefix
telemetry	put()	Store metric data point
telemetry	query()	Query time-series data
monitoring	event()	Record monitoring event
monitoring	query()	Query events with filters

Complete Example: Multi-Service Integration

import turmux_python
import time
import os
from logger_pb2 import LogRequest, LogResponse
from storage_pb2 import WriteRequest, WriteResponse, ReadRequest, ReadResponse
from telemetry_pb2 import PutRequest, PutResponse
from monitoring_pb2 import EventRequest, EventResponse

class ApplicationMonitor:
    def __init__(self):
        # Connect via vsock to Bunker OS services
        self.client = turmux_python.TurmuxClient.connect_vsock(cid=3, port=1024)

    def log_startup(self):
        """Log application startup"""
        request = LogRequest(
            level="INFO",
            message="Application started",
            timestamp=int(time.time()),
            source="app_monitor",
            tags=["startup"]
        )
        response_bytes = self.client.call("logger.store", request.SerializeToString())
        response = LogResponse()
        response.ParseFromString(response_bytes)

    def store_config(self, config_dict):
        """Store application configuration"""
        for key, value in config_dict.items():
            request = WriteRequest(
                key=f"app_config/{key}",
                value=str(value),
                tags=["configuration"]
            )
            response_bytes = self.client.call("storage.write", request.SerializeToString())
            response = WriteResponse()
            response.ParseFromString(response_bytes)

    def record_metrics(self, cpu, memory, disk):
        """Record system metrics"""
        timestamp = int(time.time())

        for metric_name, value in [
            ("system.cpu", cpu),
            ("system.memory", memory),
            ("system.disk", disk)
        ]:
            request = PutRequest(
                metric_name=metric_name,
                value=value,
                timestamp=timestamp,
                tags={"host": os.hostname()}
            )
            response_bytes = self.client.call("telemetry.put", request.SerializeToString())

    def record_error(self, error_msg, context):
        """Log error and record monitoring event"""
        # Log it
        log_request = LogRequest(
            level="ERROR",
            message=error_msg,
            timestamp=int(time.time()),
            source="app_monitor",
            metadata=context
        )
        self.client.call("logger.store", log_request.SerializeToString())

        # Also record as monitoring event
        event_request = EventRequest(
            event_type="PERFORMANCE",
            severity="HIGH",
            description=error_msg,
            timestamp=int(time.time()),
            source="app_monitor",
            context=context
        )
        self.client.call("monitoring.event", event_request.SerializeToString())

# Usage
monitor = ApplicationMonitor()
monitor.log_startup()
monitor.store_config({"db_host": "localhost", "timeout": 30})
monitor.record_metrics(cpu=45.2, memory=62.5, disk=72.1)
monitor.record_error("Database connection failed", {"retry_count": 3})

Next Steps

• See Custom Service Development to create your own services

• See Fluent Bit Log Forwarding for integration with Fluentbit

• See Falco Runtime Security Monitoring for integration with Falco