Prometheus Remote Writer

https://github.com/xykong/prometheus-remote-writer
Admin
Prometheus Remote Writer is a Python library designed to simplify writing time-series data to...
Tokens:11,448
Snippets:74
Trust Score:6.7
License:Apache-2.0
Update:7 months ago
Show doc for...
Tokens:
Raw
### Install uv and Project Dependencies

Source: https://github.com/xykong/prometheus-remote-writer/blob/master/AGENTS.md

Installs the uv dependency manager and then installs the project's development dependencies using uv. It also shows how to activate a local virtual environment and install dependencies within it.

```bash
# Install uv (if not already installed)
curl -LsSf https://astral.sh/uv/install.sh | sh

# Install dependencies
uv pip install -e ".[dev]"

# OR use the local venv
source .venv/bin/activate
uv pip install -e ".[dev]"
```

--------------------------------

### Quick Start: Send data to Prometheus Remote Write endpoint

Source: https://github.com/xykong/prometheus-remote-writer/blob/master/README.md

A basic example demonstrating how to initialize the RemoteWriter and send a sample data payload. It includes setting the target URL and authentication headers. The data format is a list of dictionaries, each representing a time series with metrics, values, and timestamps.

```python
from prometheus_remote_writer import RemoteWriter

# Create a RemoteWriter instance
writer = RemoteWriter(
    url='https://your-prometheus-server/api/v1/write',
    headers={'Authorization': 'Bearer YOUR_ACCESS_TOKEN'}
)

# Prepare the data to send
data = [
    {
        'metric': {'__name__': 'cpu_usage', 'host': 'server1'},
        'values': [23.5, 24.1, 22.8],
        'timestamps': [1609459200000, 1609459260000, 1609459320000]
    }
]

# Send the data
writer.send(data)
```

--------------------------------

### Context Manager Implementation (Python)

Source: https://github.com/xykong/prometheus-remote-writer/blob/master/AGENTS.md

Demonstrates implementing context manager support for a class using `__enter__` and `__exit__` methods. This allows for proper resource management, such as setup and cleanup operations.

```python
class MyResource:
    def __enter__(self):
        # Setup logic
        return self

    def __exit__(self, exc_type, exc, tb):
        # Cleanup logic
        self.close()

    def close(self):
        # Actual closing implementation
```

--------------------------------

### Python Testing with pytest and unittest.mock

Source: https://github.com/xykong/prometheus-remote-writer/blob/master/AGENTS.md

Shows the standard imports for testing in Python projects using pytest and the `unittest.mock` library. This setup is used for creating tests and mocking dependencies.

```python
# Use pytest, unittest.mock for mocking
from unittest.mock import patch, MagicMock
import pytest
```

--------------------------------

### Initialize RemoteWriter with Default and Full Configuration (Python)

Source: https://context7.com/xykong/prometheus-remote-writer/llms.txt

Demonstrates how to initialize the RemoteWriter class with default settings and provides a comprehensive example with various configuration options. These options include URL, headers, authentication, proxies, retry policies, batching parameters, and timestamp handling.

```python
from prometheus_remote_writer import RemoteWriter

# Basic initialization with defaults
writer = RemoteWriter(url="https://prometheus.example.com/api/v1/write")

# Full configuration example
writer = RemoteWriter(
    url="https://prometheus.example.com/api/v1/write",
    headers={"X-Scope-OrgID": "tenant-1"},
    timeout=30.0,  # Request timeout in seconds (or tuple for connect/read)
    auth={"bearer_token": "your-secret-token"},  # Or {"username": "user", "password": "pass"}
    proxies={"http": "http://proxy:8080", "https": "https://proxy:8080"},
    retries=5,  # Number of retry attempts
    backoff_factor=0.5,  # Exponential backoff multiplier
    status_forcelist=(429, 500, 502, 503, 504),  # HTTP codes to retry
    pool_connections=10,  # Connection pool size
    pool_maxsize=50,  # Max pool size
    max_series_per_request=2000,  # Batch size limit
    max_bytes_per_request=4_000_000,  # Max compressed payload size
    sort_labels=True,  # Sort labels alphabetically
    auto_convert_seconds_to_ms=True,  # Auto-convert second timestamps to ms
    strict_timestamps=False,  # Raise error on second timestamps
    verify=True,  # SSL verification (True, False, or path to CA bundle)
    user_agent="my-app/1.0",
)

print(writer)
# Output: RemoteWriter(url='https://prometheus.example.com/api/v1/write', headers={...}, timeout=30.0)
```

--------------------------------

### Python Type Annotations

Source: https://github.com/xykong/prometheus-remote-writer/blob/master/AGENTS.md

Demonstrates mandatory type annotations for all Python functions and methods. It shows correct usage with the `typing` module and contrasts it with incorrect, unannotated examples.

```python
# ✓ CORRECT
def send(self, metrics: Sequence[MetricItem]) -> SendResult:
    """Send metrics to Prometheus."""
    pass

def _normalize_timestamps(
    self,
    timestamps: Sequence[Union[int, float]],
    metric_index: int,
) -> List[int]:
    """Normalize timestamps to milliseconds."""
    pass

# ✗ WRONG - Missing type hints
def send(self, metrics):
    pass
```

--------------------------------

### Install Prometheus Remote Writer using pip

Source: https://github.com/xykong/prometheus-remote-writer/blob/master/README.md

This command installs the prometheus-remote-writer Python package using pip. Ensure Python 3.8 or higher is installed. This is the primary method for adding the library to your project.

```bash
pip install prometheus-remote-writer
```

--------------------------------

### Configure RemoteWriter to use an HTTP proxy

Source: https://github.com/xykong/prometheus-remote-writer/blob/master/README.md

This example illustrates how to configure the RemoteWriter to route its requests through an HTTP or HTTPS proxy. The `proxies` parameter takes a dictionary specifying the proxy URLs for different protocols.

```python
from prometheus_remote_writer import RemoteWriter

writer = RemoteWriter(
    url='https://your-prometheus-server/api/v1/write',
    proxies={'http': 'http://proxy.example.com:8080', 'https': 'https://proxy.example.com:8080'}
)
```

--------------------------------

### Configure RemoteWriter with a custom CA certificate

Source: https://github.com/xykong/prometheus-remote-writer/blob/master/README.md

This example demonstrates configuring the RemoteWriter to use a custom CA certificate bundle for SSL verification. This is an alternative to disabling verification when dealing with custom certificate authorities. Provide the path to your CA bundle file using the `verify` parameter.

```python
from prometheus_remote_writer import RemoteWriter

writer = RemoteWriter(
    url='https://mimir.example.com/api/v1/push',
    headers={'X-Scope-OrgID': 'test-id'},
    verify='/path/to/ca-bundle.crt'  # Custom CA certificate
)
```

--------------------------------

### Integrate with VictoriaMetrics using Remote Writer (Python)

Source: https://context7.com/xykong/prometheus-remote-writer/llms.txt

Shows how to configure and use `RemoteWriter` to send metrics to a VictoriaMetrics instance. It includes setting the endpoint, batching configurations, and converting time units. The example also demonstrates querying data back from VictoriaMetrics using its API.

```python
from prometheus_remote_writer import RemoteWriter
import time
import requests

# VictoriaMetrics remote write endpoint
writer = RemoteWriter(
    url="http://localhost:8428/api/v1/write",
    max_series_per_request=5000,  # VictoriaMetrics handles large batches well
    auto_convert_seconds_to_ms=True,
)

# Send metrics
current_time_ms = int(time.time()) * 1000
metrics = [
    {
        "metric": {
            "__name__": "node_cpu_seconds_total",
            "cpu": "0",
            "mode": "user",
            "instance": "server-01:9100",
            "job": "node_exporter",
        },
        "values": [12345.67, 12346.89, 12348.12],
        "timestamps": [current_time_ms - 60000, current_time_ms - 30000, current_time_ms],
    },
    {
        "metric": {
            "__name__": "node_memory_MemAvailable_bytes",
            "instance": "server-01:9100",
            "job": "node_exporter",
        },
        "values": [8589934592, 8489934592, 8389934592],
        "timestamps": [current_time_ms - 60000, current_time_ms - 30000, current_time_ms],
    },
]

with writer:
    result = writer.send(metrics)
    print(f"Sent {result.series_sent} series with {result.samples_sent} samples to VictoriaMetrics")

# Query the data via VictoriaMetrics API
query_url = "http://localhost:8428/api/v1/export?match=node_cpu_seconds_total"
response = requests.get(query_url)
print(f"Query result: {response.text}")

```

--------------------------------

### Run Prometheus Remote Writer unit tests with pytest

Source: https://github.com/xykong/prometheus-remote-writer/blob/master/README.md

This command executes the unit tests for the prometheus-remote-writer project using the pytest framework. Ensure pytest is installed in your environment to run this command.

```bash
pytest
```

--------------------------------

### Configure RemoteWriter with Bearer or Basic Authentication

Source: https://github.com/xykong/prometheus-remote-writer/blob/master/README.md

Examples showing how to configure the RemoteWriter for different authentication methods. The `auth` parameter accepts a dictionary for either Bearer token authentication or Basic HTTP authentication, providing flexibility for secure connections.

```python
# Bearer token authentication
writer = RemoteWriter(
    url='https://your-prometheus-server/api/v1/write',
    auth={'bearer_token': 'YOUR_ACCESS_TOKEN'}
)

# Basic authentication
writer = RemoteWriter(
    url='https://your-prometheus-server/api/v1/write',
    auth={'username': 'user', 'password': 'pass'}
)
```

--------------------------------

### Send Metrics using RemoteWriter.send() (Python)

Source: https://context7.com/xykong/prometheus-remote-writer/llms.txt

Shows how to use the `send()` method to convert, compress, and send metrics to a configured endpoint. It includes defining metrics in the `MetricItem` structure and processing the `SendResult` object returned by the method. An example of sending an empty list of metrics is also provided.

```python
from prometheus_remote_writer import RemoteWriter, MetricItem
from typing import List

writer = RemoteWriter(
    url="https://mimir.example.com/api/v1/push",
    headers={"X-Scope-OrgID": "production"},
    auth={"bearer_token": "secret-token"},
    max_series_per_request=1000,
)

# Define metrics following MetricItem structure
metrics: List[MetricItem] = [
    {
        "metric": {
            "__name__": "http_requests_total",
            "method": "GET",
            "status": "200",
            "endpoint": "/api/users",
        },
        "values": [150.0, 175.0, 200.0, 225.0],
        "timestamps": [1717200000000, 1717200060000, 1717200120000, 1717200180000],  # Milliseconds
    },
    {
        "metric": {
            "__name__": "http_request_duration_seconds",
            "method": "POST",
            "endpoint": "/api/orders",
        },
        "values": [0.125, 0.089, 0.156],
        "timestamps": [1717200000000, 1717200060000, 1717200120000],
    },
    {
        "metric": {
            "__name__": "cpu_usage_percent",
            "host": "server-01",
            "region": "us-east-1",
        },
        "values": [45.2, 52.8, 48.1, 55.3, 61.0],
        "timestamps": [1717200000000, 1717200060000, 1717200120000, 1717200180000, 1717200240000],
    },
]

# Send metrics and get result
result = writer.send(metrics)

print(f"Requests sent: {result.requests_sent}")
print(f"Series sent: {result.series_sent}")
print(f"Samples sent: {result.samples_sent}")
print(f"Last response status: {result.last_response.status_code if result.last_response else 'N/A'}")

# Output:
# Requests sent: 1
# Series sent: 3
# Samples sent: 12
# Last response status: 200

# Empty metrics return immediately without making requests
empty_result = writer.send([])
print(f"Empty send - requests: {empty_result.requests_sent}")  # Output: 0

```

--------------------------------

### Send Valid and Invalid Metrics with Error Handling (Python)

Source: https://context7.com/xykong/prometheus-remote-writer/llms.txt

Demonstrates sending valid metrics and handling potential `RuntimeError` for HTTP/network issues and `ValueError` for invalid metric data. It shows examples of metrics missing keys or having length mismatches in values and timestamps.

```python
from prometheus_remote_writer import RemoteWriter
import logging

logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)

# Assuming 'writer' is an initialized RemoteWriter instance
# For demonstration, let's mock it:
class MockWriter:
    def send(self, metrics):
        if not metrics:
            raise ValueError("No metrics provided")
        if "__name__" not in metrics[0]["metric"]:
            raise ValueError("Metric item at index 0 missing required keys: 'metric', 'values', 'timestamps'")
        if len(metrics[0]["values"]) != len(metrics[0]["timestamps"]):
            raise ValueError("Metric item at index 0 has length mismatch: values=3 vs timestamps=2")
        class MockResult:
            samples_sent = len(metrics[0]["values"])
            requests_sent = 1
        return MockResult()

writer = MockWriter()

# Valid metrics
metrics = [
    {
        "metric": {"__name__": "app_requests", "service": "api"},
        "values": [100, 150, 200],
        "timestamps": [1717200000000, 1717200060000, 1717200120000],
    }
]

try:
    result = writer.send(metrics)
    logger.info(f"Successfully sent {result.samples_sent} samples in {result.requests_sent} requests")
except RuntimeError as e:
    # HTTP errors (4xx, 5xx) or network failures
    logger.error(f"Failed to send metrics: {e}")
    # Example: RuntimeError: HTTP error posting to https://...: status=401, body=Unauthorized
except ValueError as e:
    # Invalid input data
    logger.error(f"Invalid metric data: {e}")

# Invalid metrics examples that raise ValueError
invalid_metrics_missing_keys = [
    {"metric": {"__name__": "test"}, "values": [1.0]}  # Missing 'timestamps'
]

invalid_metrics_length_mismatch = [
    {
        "metric": {"__name__": "test"},
        "values": [1.0, 2.0, 3.0],
        "timestamps": [1717200000000, 1717200060000],  # Length mismatch
    }
]

try:
    writer.send(invalid_metrics_missing_keys)
except ValueError as e:
    print(f"Error: {e}")
    # Error: Metric item at index 0 missing required keys: 'metric', 'values', 'timestamps'

try:
    writer.send(invalid_metrics_length_mismatch)
except ValueError as e:
    print(f"Error: {e}")
    # Error: Metric item at index 0 has length mismatch: values=3 vs timestamps=2

```

--------------------------------

### Python Error Handling Patterns

Source: https://github.com/xykong/prometheus-remote-writer/blob/master/AGENTS.md

Provides examples of robust error handling in Python, including raising `RuntimeError` for network/HTTP errors with context and `ValueError` for invalid input. It emphasizes preserving exception chains using `raise ... from e`.

```python
# Raise RuntimeError for network/HTTP errors with context
try:
    resp = self._session.post(url, ...)
    resp.raise_for_status()
except requests.HTTPError as e:
    raise RuntimeError(
        f"HTTP error posting to {self.url}: status={status}, body={body_snippet}"
    ) from e
except requests.RequestException as e:
    raise RuntimeError(
        f"Network error posting to {self.url}: {e.__class__.__name__}: {e}"
    ) from e

# Raise ValueError for invalid input with details
if len(values) != len(timestamps):
    raise ValueError(
        f"Metric item at index {idx} has length mismatch: "
        f"values={len(values)} vs timestamps={len(timestamps)}"
    )
```

--------------------------------

### Configure RemoteWriter to ignore self-signed certificates

Source: https://github.com/xykong/prometheus-remote-writer/blob/master/README.md

This example shows how to configure the RemoteWriter to disable SSL certificate verification, which is useful when connecting to Prometheus servers with self-signed certificates. The `verify=False` parameter overrides default SSL checks.

```python
from prometheus_remote_writer import RemoteWriter

writer = RemoteWriter(
    url='https://mimir.example.com/api/v1/push',
    headers={'X-Scope-OrgID': 'test-id'},
    verify=False  # Disable SSL verification
)
```

--------------------------------

### Generate Protobuf Files

Source: https://github.com/xykong/prometheus-remote-writer/blob/master/AGENTS.md

Command to generate Protocol Buffer (protobuf) files, typically used for serialization and communication. This is often a prerequisite for running or testing certain parts of the project.

```bash
# Generate protobuf files
make proto
```

--------------------------------

### Build and Check Project Artifacts

Source: https://github.com/xykong/prometheus-remote-writer/blob/master/AGENTS.md

Commands for building the Python package and running a comprehensive check that includes cleaning, linting, testing, and tox. This ensures the project is in a releasable state.

```bash
# Build package
python -m build

# Run all checks (clean, lint, test, tox)
make check
```

--------------------------------

### Lint Project Code with flake8

Source: https://github.com/xykong/prometheus-remote-writer/blob/master/AGENTS.md

Shows how to run flake8 for linting the project code, with options for critical errors only or a full check including warnings and complexity limits. It also mentions using the Makefile for linting.

```bash
# Run flake8 (critical errors only)
flake8 . --count --select=E9,F63,F7,F82 --show-source --statistics

# Run full flake8 (with warnings)
flake8 . --count --exit-zero --max-complexity=10 --max-line-length=127 --statistics

# OR use Makefile
make lint
```

--------------------------------

### Run Project Tests with pytest

Source: https://github.com/xykong/prometheus-remote-writer/blob/master/AGENTS.md

Demonstrates various ways to run tests for the prometheus-remote-writer project using pytest. This includes running all tests, specific files or functions, with verbose output, and with code coverage.

```bash
# Run all tests
make test
# OR
pytest

# Run specific test file
pytest tests/test_remote_writer_mocks.py

# Run single test function
pytest tests/test_remote_writer_mocks.py::TestRemoteWriter::test_initialization_defaults

# Run with verbose output
pytest -v

# Run with coverage
pytest --cov=prometheus_remote_writer
```

--------------------------------

### Test RemoteWriter Initialization and Sending Messages (Python)

Source: https://github.com/xykong/prometheus-remote-writer/blob/master/AGENTS.md

Demonstrates testing the RemoteWriter class, including default initialization and successful message sending using mocks. It utilizes `unittest.mock.patch` for isolating dependencies.

```python
from unittest.mock import patch, MagicMock
from requests import Response

# Assuming RemoteWriter is defined elsewhere
# from prometheus_remote_writer import RemoteWriter

class TestRemoteWriter:
    def test_initialization_defaults(self):
        writer = RemoteWriter(url="http://example.com")
        assert writer.url == "http://example.com"
        
    @patch.object(requests.Session, "post")
    def test_send_message_success(self, mock_post):
        mock_resp = MagicMock(spec=Response)
        mock_resp.status_code = 200
        mock_post.return_value = mock_resp
        # Test logic...

# Skip unimplemented tests
@pytest.mark.skip(reason="This test is not yet implemented")
def test_future_feature():
    pass
```

--------------------------------

### SSL/TLS Configuration for RemoteWriter (Python)

Source: https://context7.com/xykong/prometheus-remote-writer/llms.txt

Demonstrates how to configure SSL/TLS certificate verification for secure connections with the RemoteWriter. It covers enabling default verification, disabling it, using custom CA bundles, and combining SSL settings with authentication. Requires the 'prometheus_remote_writer' library.

```python
from prometheus_remote_writer import RemoteWriter

# Default: SSL verification enabled
secure_writer = RemoteWriter(
    url="https://prometheus.example.com/api/v1/write",
    verify=True,  # Default behavior
)

# Disable SSL verification (for development/self-signed certs)
insecure_writer = RemoteWriter(
    url="https://internal-prometheus.local/api/v1/write",
    verify=False,  # Disables certificate verification
)

# Custom CA certificate bundle
custom_ca_writer = RemoteWriter(
    url="https://prometheus.internal.corp/api/v1/write",
    verify="/etc/ssl/certs/internal-ca-bundle.crt",  # Path to CA bundle
)

# Combined with authentication
production_writer = RemoteWriter(
    url="https://secure-metrics.example.com/api/v1/write",
    auth={"bearer_token": "eyJhbGciOiJIUzI1NiIsInR5cCI6IkpXVCJ9..."},
    verify="/etc/ssl/certs/company-ca.pem",
    timeout=(5.0, 30.0),  # (connect_timeout, read_timeout)
)
```

--------------------------------

### Integrate with Grafana Mimir using Remote Writer (Python)

Source: https://context7.com/xykong/prometheus-remote-writer/llms.txt

Demonstrates sending metrics to Grafana Mimir, highlighting configuration for multi-tenancy with `X-Scope-OrgID` headers and bearer token authentication. It also shows setting request size limits and sorting labels for Mimir compatibility.

```python
from prometheus_remote_writer import RemoteWriter
import time

# Grafana Mimir with multi-tenancy
writer = RemoteWriter(
    url="https://mimir.example.com/api/v1/push",
    headers={
        "X-Scope-OrgID": "tenant-production",  # Required for multi-tenant Mimir
    },
    auth={"bearer_token": "mimir-api-token"},
    max_series_per_request=1000,
    max_bytes_per_request=1_000_000,  # 1MB limit
    sort_labels=True,  # Mimir benefits from sorted labels
    timeout=(10.0, 60.0),
)

# Generate metrics for multiple services
services = ["api-gateway", "user-service", "order-service", "payment-service"]
current_time = int(time.time()) * 1000

metrics = []
for i, service in enumerate(services):
    metrics.append({
        "metric": {
            "__name__": "service_requests_total",
            "service": service,
            "environment": "production",
            "region": "us-east-1",
        },
        "values": [1000 + i * 100, 1050 + i * 100, 1100 + i * 100],
        "timestamps": [current_time - 120000, current_time - 60000, current_time],
    })
    metrics.append({
        "metric": {
            "__name__": "service_latency_seconds",
            "service": service,
            "quantile": "0.99",
        },
        "values": [0.5 + i * 0.1, 0.55 + i * 0.1, 0.52 + i * 0.1],
        "timestamps": [current_time - 120000, current_time - 60000, current_time],
    })

with writer:
    result = writer.send(metrics)
    print(f"Pushed {result.series_sent} series to Mimir tenant 'tenant-production'")
    print(f"Total samples: {result.samples_sent}, Requests made: {result.requests_sent}")

```

--------------------------------

### Clean Build Artifacts

Source: https://github.com/xykong/prometheus-remote-writer/blob/master/AGENTS.md

Command to remove generated build artifacts and temporary files from the project directory. This is useful for ensuring a clean build environment.

```bash
# Clean build artifacts
make clean
```

--------------------------------

### TypedDict for API Contracts (Python)

Source: https://github.com/xykong/prometheus-remote-writer/blob/master/AGENTS.md

Shows how to use `TypedDict` to define the structure of dictionaries used for API contracts. This enhances type checking and code readability for data exchange.

```python
from typing import Dict, List, Union

class MetricItem(TypedDict):
    metric: Dict[str, str]
    values: List[Union[int, float]]
```

--------------------------------

### Organize Python Imports

Source: https://github.com/xykong/prometheus-remote-writer/blob/master/AGENTS.md

Illustrates the recommended way to organize import statements in Python code. Imports are grouped into standard library, third-party, and local imports, with specific formatting for multi-line imports.

```python
# 1. Standard library imports (grouped by category)
import logging
from dataclasses import dataclass
from typing import (
    Dict,
    List,
    Optional,
    Sequence,
    Tuple,
)

# 2. Third-party imports
import requests
import snappy
from requests import Response, Session

# 3. Local imports (with # noqa if needed for generated code)
from prometheus_remote_writer.proto.remote_pb2 import WriteRequest  # noqa
```

--------------------------------

### Error Handling with RemoteWriter (Python)

Source: https://context7.com/xykong/prometheus-remote-writer/llms.txt

Shows how to implement error handling for the `send()` method of the RemoteWriter, which can raise `RuntimeError` for network issues or `ValueError` for invalid input. It includes basic logging configuration and setting retry parameters. Requires the 'prometheus_remote_writer' library and 'logging'.

```python
from prometheus_remote_writer import RemoteWriter
import logging

logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)

writer = RemoteWriter(
    url="https://prometheus.example.com/api/v1/write",
    retries=3,
    backoff_factor=0.5,
    logger=logger,
)

```

--------------------------------

### Automatic Timestamp Conversion with RemoteWriter (Python)

Source: https://context7.com/xykong/prometheus-remote-writer/llms.txt

Illustrates the automatic conversion of timestamps from seconds to milliseconds by the RemoteWriter, controlled by `auto_convert_seconds_to_ms` and `strict_timestamps`. It shows how to configure logging to observe conversion warnings and how strict mode handles invalid timestamps. Requires the 'prometheus_remote_writer' library and 'logging'.

```python
from prometheus_remote_writer import RemoteWriter
import logging
import time

# Configure logging to see conversion warnings
logging.basicConfig(level=logging.WARNING)
logger = logging.getLogger("prometheus_remote_writer")

writer = RemoteWriter(
    url="https://prometheus.example.com/api/v1/write",
    auto_convert_seconds_to_ms=True,  # Default: auto-convert seconds to ms
    strict_timestamps=False,  # Default: don't raise on second timestamps
    logger=logger,
)

# Timestamps in seconds will be auto-converted with a warning
metrics_with_seconds = [
    {
        "metric": {"__name__": "sensor_temperature", "location": "datacenter-1"},
        "values": [23.5, 24.1, 22.8],
        "timestamps": [1609459200, 1609459260, 1609459320],  # Unix seconds
    }
]

# Sends successfully after conversion (logs warning)
result = writer.send(metrics_with_seconds)
# Warning: Metric[0]: timestamps appear to be in seconds; auto-converted to milliseconds.

# Timestamps already in milliseconds pass through unchanged
metrics_with_ms = [
    {
        "metric": {"__name__": "sensor_temperature", "location": "datacenter-2"},
        "values": [25.0, 26.2],
        "timestamps": [1609459200000, 1609459260000],  # Already milliseconds
    }
]
result = writer.send(metrics_with_ms)  # No warning

# Strict mode raises ValueError on second timestamps
strict_writer = RemoteWriter(
    url="https://prometheus.example.com/api/v1/write",
    auto_convert_seconds_to_ms=False,
    strict_timestamps=True,
)

try:
    strict_writer.send(metrics_with_seconds)
except ValueError as e:
    print(f"Error: {e}")
    # Error: Metric[0]: timestamps appear to be in seconds (< 10000000000); strict_timestamps=True, refusing to send.
```

--------------------------------

### Dataclass for Data Containers (Python)

Source: https://github.com/xykong/prometheus-remote-writer/blob/master/AGENTS.md

Illustrates the use of Python's `dataclass` decorator for creating simple data container objects. Dataclasses reduce boilerplate code for attribute initialization and representation.

```python
from dataclasses import dataclass

@dataclass
class SendResult:
    requests_sent: int
    series_sent: int
```

--------------------------------

### Context Manager and Resource Cleanup for RemoteWriter (Python)

Source: https://context7.com/xykong/prometheus-remote-writer/llms.txt

Illustrates the use of the `RemoteWriter` class as a context manager for automatic resource cleanup. It emphasizes the importance of using the `close()` method or the `with` statement to properly release HTTP session resources after use.

```python
from prometheus_remote_writer import RemoteWriter
import time

# Example usage with context manager (preferred for automatic cleanup)
with RemoteWriter(url="https://prometheus.example.com/api/v1/write") as writer:
    # Send metrics here
    # writer.send([...])
    print("Writer initialized within context.")
    # Resources will be automatically closed when exiting the 'with' block

print("Writer context exited, resources cleaned up.")

# Example usage with explicit close()
writer_explicit = RemoteWriter(url="https://prometheus.example.com/api/v1/write")
try:
    # Send metrics here
    # writer_explicit.send([...])
    print("Writer initialized explicitly.")
finally:
    writer_explicit.close()  # Ensure resources are closed
    print("Writer closed explicitly.")

```

--------------------------------

### Send Metrics using RemoteWriter (Python)

Source: https://context7.com/xykong/prometheus-remote-writer/llms.txt

Demonstrates how to send metrics to a Prometheus remote write endpoint using the RemoteWriter class. It shows both context manager usage for automatic resource management and manual cleanup with try-finally blocks. Requires the 'prometheus_remote_writer' library.

```python
from prometheus_remote_writer import RemoteWriter
import time

# Using context manager (recommended)
with RemoteWriter(
    url="https://prometheus.example.com/api/v1/write",
    auth={"username": "metrics-writer", "password": "secure-password"},
) as writer:
    metrics = [
        {
            "metric": {"__name__": "process_memory_bytes", "process": "worker"},
            "values": [1024000, 1048576, 1073152],
            "timestamps": [
                int(time.time() - 120) * 1000,
                int(time.time() - 60) * 1000,
                int(time.time()) * 1000,
            ],
        }
    ]
    result = writer.send(metrics)
    print(f"Sent {result.samples_sent} samples")
# Session automatically closed here

# Manual cleanup (alternative approach)
writer = RemoteWriter(url="https://prometheus.example.com/api/v1/write")
try:
    result = writer.send(metrics)
finally:
    writer.close()  # Explicitly close the session
```

--------------------------------

### RemoteWriter Class Initialization

Source: https://context7.com/xykong/prometheus-remote-writer/llms.txt

The RemoteWriter class is the primary interface for sending metrics. It allows for extensive configuration including endpoint URL, custom headers, authentication methods, SSL verification, proxy settings, retry policies, and batching parameters.

```APIDOC
## RemoteWriter Class Initialization

### Description
The `RemoteWriter` class is the main entry point for sending metrics. It supports extensive configuration options including custom headers, authentication, SSL verification, proxies, retry policies, and batching parameters.

### Method
Initialization

### Endpoint
N/A (Class Initialization)

### Parameters
#### Path Parameters
None

#### Query Parameters
None

#### Request Body
None

### Request Example
```python
from prometheus_remote_writer import RemoteWriter

# Basic initialization with defaults
writer = RemoteWriter(url="https://prometheus.example.com/api/v1/write")

# Full configuration example
writer = RemoteWriter(
    url="https://prometheus.example.com/api/v1/write",
    headers={"X-Scope-OrgID": "tenant-1"},
    timeout=30.0,  # Request timeout in seconds (or tuple for connect/read)
    auth={"bearer_token": "your-secret-token"},  # Or {"username": "user", "password": "pass"}
    proxies={"http": "http://proxy:8080", "https": "https://proxy:8080"},
    retries=5,  # Number of retry attempts
    backoff_factor=0.5,  # Exponential backoff multiplier
    status_forcelist=(429, 500, 502, 503, 504),  # HTTP codes to retry
    pool_connections=10,  # Connection pool size
    pool_maxsize=50,  # Max pool size
    max_series_per_request=2000,  # Batch size limit
    max_bytes_per_request=4_000_000,  # Max compressed payload size
    sort_labels=True,  # Sort labels alphabetically
    auto_convert_seconds_to_ms=True,  # Auto-convert second timestamps to ms
    strict_timestamps=False,  # Raise error on second timestamps
    verify=True,  # SSL verification (True, False, or path to CA bundle)
    user_agent="my-app/1.0",
)

print(writer)
# Output: RemoteWriter(url='https://prometheus.example.com/api/v1/write', headers={...}, timeout=30.0)
```

### Response
#### Success Response (Initialization)
Returns an instance of the `RemoteWriter` class.

#### Response Example
```json
{
  "instance": "<RemoteWriter object at 0x...>"
}
```
```

--------------------------------

### RemoteWriter.send() Method

Source: https://context7.com/xykong/prometheus-remote-writer/llms.txt

The `send()` method is responsible for converting metrics into the protobuf format, applying Snappy compression, and transmitting them to the configured remote write endpoint. It handles automatic batching for large payloads and returns a `SendResult` object containing statistics about the operation.

```APIDOC
## RemoteWriter.send() Method

### Description
The `send()` method converts metrics to protobuf format, compresses with Snappy, and sends to the configured endpoint. It automatically batches large payloads and returns a `SendResult` with statistics about the operation.

### Method
POST

### Endpoint
N/A (Method of RemoteWriter instance)

### Parameters
#### Path Parameters
None

#### Query Parameters
None

#### Request Body
- **metrics** (List[MetricItem]) - Required - A list of metric data points to send. Each `MetricItem` should follow the structure:
  - **metric** (Dict) - Required - A dictionary containing the metric name and its labels.
    - **`__name__`** (str) - Required - The name of the metric.
    - **(label_name)** (str) - Optional - Additional labels for the metric.
  - **values** (List[float]) - Required - A list of metric values.
  - **timestamps** (List[int]) - Required - A list of timestamps corresponding to the values, in milliseconds.

### Request Example
```python
from prometheus_remote_writer import RemoteWriter, MetricItem
from typing import List

writer = RemoteWriter(
    url="https://mimir.example.com/api/v1/push",
    headers={"X-Scope-OrgID": "production"},
    auth={"bearer_token": "secret-token"},
    max_series_per_request=1000,
)

# Define metrics following MetricItem structure
metrics: List[MetricItem] = [
    {
        "metric": {
            "__name__": "http_requests_total",
            "method": "GET",
            "status": "200",
            "endpoint": "/api/users",
        },
        "values": [150.0, 175.0, 200.0, 225.0],
        "timestamps": [1717200000000, 1717200060000, 1717200120000, 1717200180000],  # Milliseconds
    },
    {
        "metric": {
            "__name__": "http_request_duration_seconds",
            "method": "POST",
            "endpoint": "/api/orders",
        },
        "values": [0.125, 0.089, 0.156],
        "timestamps": [1717200000000, 1717200060000, 1717200120000],
    },
    {
        "metric": {
            "__name__": "cpu_usage_percent",
            "host": "server-01",
            "region": "us-east-1",
        },
        "values": [45.2, 52.8, 48.1, 55.3, 61.0],
        "timestamps": [1717200000000, 1717200060000, 1717200120000, 1717200180000, 1717200240000],
    },
]

# Send metrics and get result
result = writer.send(metrics)

print(f"Requests sent: {result.requests_sent}")
print(f"Series sent: {result.series_sent}")
print(f"Samples sent: {result.samples_sent}")
print(f"Last response status: {result.last_response.status_code if result.last_response else 'N/A'}")

# Output:
# Requests sent: 1
# Series sent: 3
# Samples sent: 12
# Last response status: 200

# Empty metrics return immediately without making requests
empty_result = writer.send([])
print(f"Empty send - requests: {empty_result.requests_sent}")  # Output: 0
```

### Response
#### Success Response (200 OK)
- **requests_sent** (int) - The number of HTTP requests made.
- **series_sent** (int) - The total number of unique series sent across all requests.
- **samples_sent** (int) - The total number of samples sent across all requests.
- **last_response** (Optional[requests.Response]) - The `requests.Response` object from the last successful HTTP request, or None if no requests were made.

#### Response Example
```json
{
  "requests_sent": 1,
  "series_sent": 3,
  "samples_sent": 12,
  "last_response": {
    "status_code": 200,
    "reason": "OK",
    "headers": {...}
  }
}
```
```

--------------------------------

### Context Manager and Resource Cleanup

Source: https://context7.com/xykong/prometheus-remote-writer/llms.txt

The RemoteWriter class implements the context manager protocol, allowing for automatic cleanup of resources, such as HTTP session connections, when exiting a `with` block. Explicitly calling the `close()` method also ensures proper resource release.

```APIDOC
## Context Manager and Resource Cleanup

### Description
`RemoteWriter` supports the context manager protocol for automatic resource cleanup. Use the `close()` method or context manager to properly release HTTP session resources.

### Method
Context Manager (`__enter__`, `__exit__`) and `close()` method

### Endpoint
N/A

### Parameters
None

### Request Example
```python
from prometheus_remote_writer import RemoteWriter
import time

# Using the context manager
with RemoteWriter(url="https://prometheus.example.com/api/v1/write") as writer:
    # Use the writer instance here
    writer.send([...])
    time.sleep(1)
# HTTP session is automatically closed upon exiting the 'with' block

# Alternatively, using the close() method
writer = RemoteWriter(url="https://prometheus.example.com/api/v1/write")
try:
    writer.send([...])
finally:
    writer.close() # Explicitly close the session
```

### Response
#### Success Response (Cleanup)
Resources are cleaned up without returning specific data.

#### Response Example
N/A
```

--------------------------------

### Python Naming Conventions

Source: https://github.com/xykong/prometheus-remote-writer/blob/master/AGENTS.md

Specifies Python naming conventions for classes (PascalCase), constants (UPPER_SNAKE_CASE), functions/methods (snake_case), private methods (_snake_case), and variables (snake_case).

```python
# Classes: PascalCase
class RemoteWriter:
    pass

class SendResult:
    pass

# Constants: UPPER_SNAKE_CASE
DEFAULT_HEADERS = {...}
MS_THRESHOLD = 10_000_000_000

# Functions/methods: snake_case
def send(self, metrics):
    pass

def _normalize_timestamps(self):  # Private: prefix with _
    pass

# Variables: snake_case
total_samples = 0
is_seconds = True
```

--------------------------------

### Python Formatting Standards

Source: https://github.com/xykong/prometheus-remote-writer/blob/master/AGENTS.md

Outlines Python code formatting conventions, including a maximum line length of 127 characters, 4-space indentation, preference for single quotes for strings, and the use of triple double-quotes for docstrings.

```python
# Line length: Max 127 characters (flake8 configured)
# Indentation: 4 spaces (standard Python)
# String quotes: Single quotes 'string' preferred (based on codebase pattern)
# Docstrings: Use triple double-quotes """docstring"""
```
Tokens:
Raw