redducklabs-platform-observability-sdk 1.0.2

Python SDK for Platform Observability log ingestion and analytics

Platform Observability Python SDK

A Python SDK for ingesting logs and interacting with the Platform Observability service by Red Duck Labs.

Installation

From PyPI (after publishing)

pip install redducklabs-platform-observability-sdk

From Source (Development)

cd sdks/python
pip install -e .

Quick Start

Basic Usage

from platform_observability import ObservabilityClient, LogEntry, LogLevel

# Initialize client
client = ObservabilityClient(
    api_key="YOUR_API_KEY",
    base_url="https://observability.redducklabs.com/api/v1"
)

# Send a single log
client.ingest_log(
    message="Application started",
    level=LogLevel.INFO,
    source="my-application",
    labels={"app": "my-app", "environment": "production"},
    fields={"version": "1.0.0", "user_id": "user123"}
)

Using the ObservabilityLogger (Recommended)

The ObservabilityLogger provides a higher-level interface with automatic batching and Python logging integration:

from platform_observability import ObservabilityLogger

# Initialize logger with labels for Grafana filtering
logger = ObservabilityLogger(
    api_key="YOUR_API_KEY",
    name="my-logger",
    base_url="https://observability.redducklabs.com/api/v1",
    source="my-application",
    labels={
        "app": "my-app",
        "environment": "production",
        "version": "1.0.0"
    },
    batch_size=100,      # Batch up to 100 logs
    flush_interval=5     # Auto-flush every 5 seconds
)

# Log messages with different levels
logger.info("User logged in", user_id="user123", ip="192.168.1.1")
logger.warning("High memory usage", memory_percent=85.5)
logger.error("Failed to connect to database", error_code="DB_CONN_FAILED")

# Ensure all logs are sent before exit
logger.flush()
logger.close()

Configuration

Environment Variables

You can configure the SDK using environment variables:

# Required
export RDL_API_KEY="your_api_key_here"

# Optional
export RDL_LOG_ENDPOINT="https://observability.redducklabs.com/api/v1"
export RDL_BATCH_SIZE="100"
export RDL_FLUSH_INTERVAL="5.0"

Then use them in your code:

import os
from platform_observability import ObservabilityLogger

logger = ObservabilityLogger(
    api_key=os.getenv('RDL_API_KEY'),
    base_url=os.getenv('RDL_LOG_ENDPOINT', 'https://observability.redducklabs.com/api/v1'),
    source="my-app",
    labels={"app": "my-app", "environment": os.getenv('ENVIRONMENT', 'dev')}
)

Important Concepts

Labels vs Fields

Labels (low-cardinality, indexed):

• Used for filtering and querying in Grafana
• Should have limited unique values
• Examples: app, environment, version, region

Fields (high-cardinality, not indexed):

• Additional context for debugging
• Can have many unique values
• Examples: user_id, request_id, duration_ms

Log Format

Logs are sent in this format:

{
    "message": "User action completed",
    "level": "info",
    "source": "my-app",
    "labels": {
        "app": "my-app",
        "environment": "production"
    },
    "fields": {
        "user_id": "user123",
        "action": "purchase",
        "amount": 99.99
    },
    "timestamp": "2024-01-25T10:30:00Z"
}

Querying Logs in Grafana

Once logs are ingested, query them in Grafana using LogQL:

# All logs from your app
{app="my-app"}

# Filter by environment
{app="my-app", environment="production"}

# Search for errors
{app="my-app"} |= "error"

# Parse JSON and filter
{app="my-app"} | json | user_id="user123"

Advanced Usage

Custom Batching

client = ObservabilityClient(
    api_key="YOUR_API_KEY",
    base_url="https://observability.redducklabs.com/api/v1",
    batch_size=200,        # Larger batches
    batch_timeout=10.0,    # Wait longer before sending
    enable_batching=True   # Enable automatic batching
)

# Add logs to batch queue
for i in range(100):
    client.add_to_batch(
        message=f"Log entry {i}",
        level=LogLevel.INFO,
        source="batch-test"
    )

# Manually flush when needed
client.flush_batch()

Error Handling

from platform_observability import ObservabilityClient
from platform_observability.exceptions import (
    AuthenticationError,
    RateLimitError,
    ValidationError
)

client = ObservabilityClient(
    api_key="YOUR_API_KEY",
    skip_connection_check=True  # Skip initial health check
)

try:
    client.ingest_log("Test message", level="info")
except AuthenticationError:
    print("Invalid API key")
except RateLimitError:
    print("Rate limit exceeded, retry later")
except ValidationError as e:
    print(f"Invalid log format: {e}")

Integration with Python Logging

import logging
from platform_observability import ObservabilityHandler

# Create custom handler
handler = ObservabilityHandler(
    api_key="YOUR_API_KEY",
    base_url="https://observability.redducklabs.com/api/v1",
    source="my-app",
    labels={"app": "my-app", "environment": "dev"}
)

# Add to Python logger
logger = logging.getLogger(__name__)
logger.addHandler(handler)
logger.setLevel(logging.INFO)

# Now regular Python logging will be sent to Platform Observability
logger.info("This goes to Platform Observability")
logger.error("This error too", extra={"user_id": "user123"})

API Key Management

# List API keys
keys = client.list_api_keys()
for key in keys:
    print(f"{key.name}: {key.masked_key}")

# Create new API key
new_key = client.create_api_key(
    name="ci-pipeline",
    description="API key for CI/CD pipeline"
)
print(f"New key created: {new_key.key}")

# Rotate API key
rotated = client.rotate_api_key(key_id="key_123")
print(f"New key value: {rotated['key']}")

# Delete API key
client.delete_api_key(key_id="key_123")

Usage Analytics

from datetime import datetime, timedelta

# Get usage metrics
end_time = datetime.now()
start_time = end_time - timedelta(days=7)

metrics = client.get_usage_analytics(
    start_time=start_time,
    end_time=end_time
)

print(f"Total logs ingested: {metrics.total_logs_ingested}")
print(f"Data volume: {metrics.total_data_volume_gb:.2f} GB")
print(f"Average logs per request: {metrics.average_logs_per_request:.1f}")

CLI Tool

After installation, use the obs-cli command-line tool:

# Set up configuration
obs-cli config set api_key YOUR_API_KEY
obs-cli config set endpoint https://observability.redducklabs.com/api/v1

# Send a log
obs-cli log "Application deployed" --level info --label app=my-app

# List API keys
obs-cli keys list

# Get usage statistics
obs-cli usage --days 7

Development

Requirements

• Python >= 3.8
• pip

Setup Development Environment

# Clone repository
git clone https://github.com/redducklabs/platform-observability.git
cd platform-observability/sdks/python

# Install in development mode with dependencies
pip install -e .[dev]

Running Tests

# Run all tests
pytest

# Run with coverage
pytest --cov=platform_observability

# Run specific test file
pytest tests/test_client.py

Building Package

# Install build tools
pip install build

# Build wheel and source distribution
python -m build

# Output will be in dist/
ls dist/

Code Quality

# Format code
black platform_observability tests

# Lint code
flake8 platform_observability tests

# Type checking
mypy platform_observability

Troubleshooting

Logs Not Appearing in Grafana?

1. Verify API Key: Ensure your API key is valid and active
2. Check Endpoint: Confirm you're using the correct endpoint URL
3. Validate Labels: Ensure labels are properly set (required for Grafana queries)
4. Flush Logs: Call logger.flush() to ensure batched logs are sent
5. Check Errors: Enable debug logging to see any errors:

   import logging
   logging.basicConfig(level=logging.DEBUG)
   

Connection Errors?

If you're getting connection errors during initialization:

client = ObservabilityClient(
    api_key="YOUR_API_KEY",
    base_url="https://observability.redducklabs.com/api/v1",
    skip_connection_check=True  # Skip initial health check
)

Rate Limiting?

The SDK automatically retries on rate limit errors. You can also check rate limit status:

# After making requests
rate_limit_info = client.get_rate_limit_info()
print(f"Remaining requests: {rate_limit_info.get('remaining')}")
print(f"Reset time: {rate_limit_info.get('reset')}")

Examples

See the examples/ directory for complete working examples:

• examples/sample-application/sdk_only_example.py - Basic SDK usage
• examples/sample-application/sdk_continuous_example.py - Continuous log generation
• examples/sample-application/app.py - FastAPI integration example

Support

For issues, questions, or contributions:

• GitHub Issues: https://github.com/redducklabs/platform-observability/issues
• Documentation: https://docs.redducklabs.com/observability

License

MIT License - see LICENSE file for details