Metadata-Version: 2.5
Name: aaa-sentinel-health
Version: 0.1.1
Summary: health+json endpoints and flow trace propagation for AAA Sentinel monitoring
Project-URL: Homepage, https://app.aaa-sentinel.com/developer/docs#health
Project-URL: Issues, https://app.aaa-sentinel.com/contact
License-Expression: MIT
License-File: LICENSE
Keywords: aws-lambda,azure-functions,cloudevents,health+json,health-check,monitoring,pubsub,synthetic-monitoring,traceparent,w3c-trace-context
Classifier: Operating System :: OS Independent
Classifier: Programming Language :: Python :: 3
Classifier: Programming Language :: Python :: 3.9
Classifier: Programming Language :: Python :: 3.10
Classifier: Programming Language :: Python :: 3.11
Classifier: Programming Language :: Python :: 3.12
Classifier: Programming Language :: Python :: 3.13
Classifier: Programming Language :: Python :: 3.14
Classifier: Topic :: System :: Monitoring
Classifier: Typing :: Typed
Requires-Python: >=3.9
Description-Content-Type: text/markdown

# aaa-sentinel-health

Two small helpers for services watched by [AAA Sentinel](https://aaa-sentinel.com) monitoring, with no dependencies:

- **Health endpoints** in `application/health+json`, the format the *health contract* monitor reads per component (also understood by load balancers and other monitors).
- **Flow trace propagation**: a *flow* monitor sends a test transaction through your chain (a queue, a function, an event bus, an HTTP endpoint) and checks it reaches every step on time. Your code passes its trace on to the next step and can skip real work for it.

Python 3.9 or later. Typed. It reads the same marks and writes the same responses as the Node package `@aaa-sentinel/health`; timeouts are in seconds.

```sh
pip install aaa-sentinel-health
```

## Health endpoint

```python
import os
import time
import urllib.request

from aaa_sentinel_health import run_health_checks


def database():
    db.execute("select 1")  # nothing or True: pass


def queue():
    return "pass" if queue_depth() < 1000 else "warn"  # or a status


def payments():
    started = time.monotonic()
    urllib.request.urlopen("https://payments.internal/ping", timeout=2)
    return {"observedValue": round((time.monotonic() - started) * 1000), "observedUnit": "ms"}  # or a result


# AWS Lambda behind API Gateway or a function URL: the result is the HTTP response.
def handler(event, context):
    return run_health_checks({"database": database, "queue": queue, "payments": payments}, version=os.environ.get("APP_VERSION"), timeout=2)
```

- A check that raises or takes longer than `timeout` (2 s by default, 60 s at most) is a `fail` for that component; the endpoint still answers. The checks run at the same time, each in its own daemon thread. One still running after its timeout is left behind, and until it ends the same check answers `fail` at once instead of starting another thread: give your own clients a timeout too.
- The overall status is the worst check. `fail` answers **503**, `pass` and `warn` answer **200**. `minimum_status="warn"` keeps it at `warn` or worse (during maintenance, for example).
- Nothing, `True` or whatever your client hands back (a cursor, a response) is a `pass`, and none of it is published; `False` is a `fail`. A status other than `pass`, `warn` or `fail` counts as `fail` (`ok`/`up` and `error`/`down` are read as their synonyms). Result fields can be written in camelCase or snake_case (`observed_value`); only the fields of the format leave.
- Error messages stay out of the response unless you pass `expose_errors=True`: they often contain hosts or connection strings, and health endpoints are usually public.

With asyncio, `run_health_checks_async` takes `async def` checks (cancelled on timeout) and plain ones (run in a thread):

```python
# FastAPI
@app.get("/health")
async def health():
    r = await run_health_checks_async({"database": ping_db, "cache": ping_cache})
    return Response(r["body"], r["statusCode"], r["headers"])
```

Other runtimes:

```python
# Flask
r = run_health_checks(checks)
return Response(r["body"], r["statusCode"], r["headers"])

# Azure Functions
return func.HttpResponse(r["body"], status_code=r["statusCode"], headers=r["headers"])
```

Already know the results? `health_response({"db": {"status": "pass"}})` builds the same response without running anything, and `health_body(...)` gives just the document.

## Flow trace

The test transaction carries two marks: `x-aaa-trace` / `aaaTraceId` (a 32-hex trace id) and a W3C `traceparent`. `propagate_trace` finds them in whatever started your code: an SQS, SNS, EventBridge, DynamoDB Streams or Kinesis event or record (also SNS or EventBridge delivered through SQS), a Lambda destination, a direct invocation, an API Gateway or ALB event; an Azure Functions Service Bus message, Event Hubs event, Event Grid event or HTTP request, or an `azure-servicebus` or `azure-eventhub` message; a boto3 SQS `Message`; a Google Pub/Sub message, push request or Eventarc event; a Kafka message (kafka-python, confluent-kafka, or an MSK or self-managed Kafka event in Lambda); a CloudEvent (structured, a batch, or binary through HTTP, Pub/Sub, Kafka or AMQP, with the mark as the `aaatraceid` extension); a Flask, Django, Starlette or aiohttp request, its headers or a WSGI environ; or a JSON string. It never raises.

Of a web request only the headers, and a body the framework has already read, are looked at: reading the body itself would consume the stream your view still needs. Objects from other libraries are not opened, so a property never runs code it should not (an ORM relation, a lazy body). In an Event Hubs batch, a test event is found by its body: every event of a batch carries the properties of the whole batch.

Only the AAA Sentinel mark makes a flow test: a `traceparent` from your own OpenTelemetry tracing is real traffic.

```python
import json

from aaa_sentinel_health import propagate_trace


def handler(event, context):
    for record in event["Records"]:
        trace = propagate_trace(record)  # None for real traffic
        if trace:
            # Hand it on to the next step, in whatever channel that step uses:
            sqs.send_message(QueueUrl=queue_url, MessageBody="{}", MessageAttributes=trace.message_attributes)
            requests.post(next_url, headers=trace.headers, json=trace.fields, timeout=5)
            events.put_events(Entries=[{"Source": source, "DetailType": detail_type, "Detail": json.dumps({**detail, **trace.fields})}])
            trace.log("order stored")  # {"message": "order stored <trace_id>", ...}
            continue  # skip charges, emails and real writes for the test transaction
        process_order(record)
```

- `trace.trace_id`: the id the monitor looks for. A DynamoDB or S3 step finds the item or object whose key contains it (`ORDER#{trace_id}`, `invoices/{trace_id}.pdf`): write the test transaction under that key.
- `trace.log(message)`: one JSON line at INFO level, on standard output, whose `message` ends with the trace id. A "line in the logs" step with the text `order stored {traceId}` matches `trace.log("order stored")`. Pass `logger=` to write it through your own `logging.Logger`.
- `trace.traceparent`: same trace id with a new span id, so tools such as X-Ray, Application Insights, Cloud Trace or any OpenTelemetry backend show the synthetic transaction.
- `trace.headers`, `trace.message_attributes`, `trace.attributes`, `trace.fields`, `trace.extensions`: the marks ready for an HTTP call, an SQS/SNS message (boto3), Service Bus `application_properties` or Pub/Sub attributes, a CloudEvent you emit, or a JSON body, EventBridge detail, Event Grid `data`, DynamoDB item or Cosmos DB / Firestore document.
- `is_flow_test(event)`: just the yes/no.

### Azure and Google Cloud

```python
# Azure Functions (Python v2), Service Bus trigger.
@app.service_bus_queue_trigger(arg_name="message", queue_name="orders", connection="SERVICEBUS")
def consume(message: func.ServiceBusMessage):
    trace = propagate_trace(message)
    if trace:
        sender.send_messages(ServiceBusMessage(json.dumps({**next_message, **trace.fields}), application_properties=trace.attributes))
        # A Cosmos DB step reads the document by id, with the id as partition key unless the step names another.
        container.create_item({"id": trace.trace_id, **trace.fields})
        return
    process_order(message)


# A 2nd gen function (Functions Framework) or Cloud Run behind Eventarc; for a Pub/Sub push request, pass `request.get_json()`.
@functions_framework.cloud_event
def consume(cloud_event):
    trace = propagate_trace(cloud_event)
    if trace:
        publisher.publish(next_topic, json.dumps({**next_message, **trace.fields}).encode(), **trace.attributes)
        firestore.collection("orders").document(trace.trace_id).set(trace.fields)  # a Firestore step reads orders/{traceId}
        return
    process_order(cloud_event)
```

When your code emits a CloudEvent, add `trace.extensions` to its attributes (`CloudEvent({**attributes, **trace.extensions}, data)`): the CloudEvents SDK sends `aaatraceid` and `traceparent` with the prefix of each binding (`ce-`, `ce_`, `cloudEvents_`), and the next step reads them. The names are exported as `TRACE_HEADER`, `TRACEPARENT_HEADER` and `CE_TRACE_EXTENSION`.

### Callback steps

A *callback* step is marked by your own code calling the flow's callback URL (shown on the flow page, it carries its own secret):

```python
import os

from aaa_sentinel_health import mark_step

mark_step(os.environ.get("AAA_CALLBACK_URL"), trace, "invoice")
```

`step` is the step's name as written in the flow, or its position among all the flow's steps (1 is the first). It does nothing for real traffic (`trace` is `None`) or without a URL, never raises and gives up after 3 s (`timeout=`): marking a step must not break or slow the code it watches. It returns `True` when the step was recorded. In asyncio code, run it with `await asyncio.to_thread(mark_step, url, trace, "invoice")`.

## License

MIT
