Observability Basics

Seven examples to get you started with observability for AWS SDK calls - five basic and two intermediate.

Each example shows the same instrumentation pattern in both Python (boto3) and TypeScript (AWS SDK for JavaScript v3).

Prerequisites

Python: pip install boto3 (boto3 1.43.x, Python 3.10+). TypeScript: npm install @aws-sdk/client-s3 @aws-sdk/client-dynamodb (Node.js 18+).
A logger already configured - the standard library logging module in Python, or any structured logger in TypeScript (these examples use console as a stand-in).
Credentials and a default region configured, same as any other SDK call.

Basic Examples

1. Log a Request Id and Latency

The smallest useful instrumentation: time the call, then log its request id.

# --- Python (boto3) ---
import time
import logging
import boto3
 
logger = logging.getLogger("app")
s3 = boto3.client("s3")
 
start = time.monotonic()
response = s3.list_buckets()
duration_ms = (time.monotonic() - start) * 1000
 
logger.info(
    "s3.list_buckets ok request_id=%s duration_ms=%.1f",
    response["ResponseMetadata"]["RequestId"],
    duration_ms,
)

// --- TypeScript (AWS SDK v3) ---
import { S3Client, ListBucketsCommand } from "@aws-sdk/client-s3";
 
const s3 = new S3Client({});
 
const start = Date.now();
const response = await s3.send(new ListBucketsCommand({}));
const durationMs = Date.now() - start;
 
console.log(
  `s3.ListBuckets ok requestId=${response.$metadata.requestId} durationMs=${durationMs}`,
);

The request id uniquely identifies this call on AWS's side, useful for support cases.
time.monotonic() and Date.now() are safe for measuring elapsed time; avoid wall-clock time for durations.
This costs a few lines and no dependencies, and already answers "was this call slow."
Nothing about the bucket contents is logged - only metadata about the call.

Related: Observability for AWS SDK Applications - why this matters.

2. Log the Operation Name

Add the operation name so one log line is self-describing without context from surrounding code.

# --- Python (boto3) ---
import time
import logging
import boto3
 
logger = logging.getLogger("app")
ddb = boto3.client("dynamodb")
 
def timed_call(operation_name, fn):
    start = time.monotonic()
    response = fn()
    duration_ms = (time.monotonic() - start) * 1000
    logger.info(
        "%s ok request_id=%s duration_ms=%.1f",
        operation_name,
        response["ResponseMetadata"]["RequestId"],
        duration_ms,
    )
    return response
 
item = timed_call(
    "dynamodb.GetItem",
    lambda: ddb.get_item(TableName="Orders", Key={"id": {"S": "order-42"}}),
)

// --- TypeScript (AWS SDK v3) ---
import { DynamoDBClient, GetItemCommand } from "@aws-sdk/client-dynamodb";
 
const ddb = new DynamoDBClient({});
 
async function timedCall<T extends { $metadata: { requestId?: string } }>(
  operationName: string,
  fn: () => Promise<T>,
): Promise<T> {
  const start = Date.now();
  const response = await fn();
  const durationMs = Date.now() - start;
  console.log(
    `${operationName} ok requestId=${response.$metadata.requestId} durationMs=${durationMs}`,
  );
  return response;
}
 
const item = await timedCall("dynamodb.GetItem", () =>
  ddb.send(new GetItemCommand({ TableName: "Orders", Key: { id: { S: "order-42" } } })),
);

Wrapping the call in a small helper avoids repeating the timing boilerplate at every call site.
The operation name (dynamodb.GetItem) should match the AWS API name for easy cross-referencing with docs and CloudTrail.
This pattern scales to any client and any operation without modification.
Keep the helper thin; it should never inspect or log the item contents.

3. Log Failures With the Error Code

A failed call should log the AWS error code, not just "it failed."

# --- Python (boto3) ---
import logging
import boto3
from botocore.exceptions import ClientError
 
logger = logging.getLogger("app")
s3 = boto3.client("s3")
 
try:
    s3.get_object(Bucket="my-bucket", Key="missing.txt")
except ClientError as err:
    logger.warning(
        "s3.get_object failed code=%s request_id=%s",
        err.response["Error"]["Code"],
        err.response["ResponseMetadata"]["RequestId"],
    )

// --- TypeScript (AWS SDK v3) ---
import { S3Client, GetObjectCommand } from "@aws-sdk/client-s3";
 
const s3 = new S3Client({});
 
try {
  await s3.send(new GetObjectCommand({ Bucket: "my-bucket", Key: "missing.txt" }));
} catch (err) {
  const e = err as { name: string; $metadata?: { requestId?: string } };
  console.warn(`s3.GetObject failed code=${e.name} requestId=${e.$metadata?.requestId}`);
}

err.response["Error"]["Code"] (boto3) and err.name (SDK v3) give a stable, matchable error code.
Always include the request id on failures - it is the single field AWS support asks for first.
Log at warning or error level so failures stand out from normal info traffic.
Never log the full exception's request context if it includes input parameters that may carry sensitive data.

Related: Structured Logging Around SDK Calls - a fuller logging pattern.

4. Add a Correlation Id

Pass a correlation id through your own request so all its downstream SDK calls share one identifier.

# --- Python (boto3) ---
import logging
import uuid
import boto3
 
logger = logging.getLogger("app")
s3 = boto3.client("s3")
 
def handle_upload(correlation_id: str | None = None):
    correlation_id = correlation_id or str(uuid.uuid4())
    response = s3.put_object(Bucket="my-bucket", Key="notes/hello.txt", Body=b"hi")
    logger.info(
        "s3.put_object ok correlation_id=%s request_id=%s",
        correlation_id,
        response["ResponseMetadata"]["RequestId"],
    )

// --- TypeScript (AWS SDK v3) ---
import { randomUUID } from "crypto";
import { S3Client, PutObjectCommand } from "@aws-sdk/client-s3";
 
const s3 = new S3Client({});
 
async function handleUpload(correlationId = randomUUID()) {
  const response = await s3.send(
    new PutObjectCommand({ Bucket: "my-bucket", Key: "notes/hello.txt", Body: "hi" }),
  );
  console.log(`s3.PutObject ok correlationId=${correlationId} requestId=${response.$metadata.requestId}`);
}

A correlation id (often from an incoming HTTP request header) lets you group every SDK call made while serving one user request.
The AWS request id identifies the call to AWS; the correlation id identifies the call to you. Log both.
This is the manual precursor to what a tracer does automatically with trace and span ids.
Generate the correlation id once per incoming request, not per SDK call.

5. Log Retry Attempts

Both SDKs retry transient failures automatically; logging when they do makes hidden retries visible.

# --- Python (boto3) ---
import logging
import boto3
 
logger = logging.getLogger("app")
ddb = boto3.client("dynamodb")
 
def log_retry(**kwargs):
    logger.info("dynamodb retry attempt=%s", kwargs.get("attempt"))
 
ddb.meta.events.register("needs-retry.dynamodb.PutItem", log_retry)
ddb.put_item(TableName="Visits", Item={"id": {"S": "user-7"}})

// --- TypeScript (AWS SDK v3) ---
import { DynamoDBClient, PutItemCommand } from "@aws-sdk/client-dynamodb";
 
const ddb = new DynamoDBClient({});
 
ddb.middlewareStack.add(
  (next) => async (args) => {
    try {
      return await next(args);
    } catch (err) {
      console.log("dynamodb retry observed", err);
      throw err;
    }
  },
  { step: "finalizeRequest", tags: ["RETRY_LOGGING"] },
);
 
await ddb.send(new PutItemCommand({ TableName: "Visits", Item: { id: { S: "user-7" } } }));

boto3 exposes a needs-retry lifecycle event per operation you can hook into.
SDK v3's retry behavior is easiest to observe via its built-in retry strategy metadata or middleware, since retries happen inside the client's send pipeline.
Seeing retry counts in logs often explains latency that would otherwise look unexplained.
Do not treat every retry as an error - occasional throttling retries are normal and expected.

Intermediate Examples

6. Build a Reusable Logging Wrapper

Consolidate timing, request id, and error logging into one wrapper used across all clients.

# --- Python (boto3) ---
import time
import logging
from botocore.exceptions import ClientError
 
logger = logging.getLogger("app")
 
def observed_call(operation_name, fn):
    start = time.monotonic()
    try:
        response = fn()
        duration_ms = (time.monotonic() - start) * 1000
        logger.info(
            "%s ok request_id=%s duration_ms=%.1f",
            operation_name,
            response["ResponseMetadata"]["RequestId"],
            duration_ms,
        )
        return response
    except ClientError as err:
        duration_ms = (time.monotonic() - start) * 1000
        logger.warning(
            "%s failed code=%s duration_ms=%.1f",
            operation_name,
            err.response["Error"]["Code"],
            duration_ms,
        )
        raise

// --- TypeScript (AWS SDK v3) ---
async function observedCall<T extends { $metadata: { requestId?: string } }>(
  operationName: string,
  fn: () => Promise<T>,
): Promise<T> {
  const start = Date.now();
  try {
    const response = await fn();
    console.log(`${operationName} ok requestId=${response.$metadata.requestId} durationMs=${Date.now() - start}`);
    return response;
  } catch (err) {
    const e = err as { name: string };
    console.warn(`${operationName} failed code=${e.name} durationMs=${Date.now() - start}`);
    throw err;
  }
}

One wrapper used across every client call keeps logging consistent without repeating the try/timing code.
This is the natural place to later add a tracer span, since it already brackets each call.
Keep the wrapper generic - it should not know about any specific service's fields.
Reuse it for both reads and writes, across every client in the codebase.

Related: AWS X-Ray Tracing for SDK-Calling Services - turning this wrapper into a trace subsegment.

7. Log Aggregate Call Counts

Track how many calls per operation your service makes, not just individual latencies.

# --- Python (boto3) ---
import logging
from collections import Counter
 
logger = logging.getLogger("app")
call_counts: Counter = Counter()
 
def record_call(operation_name: str):
    call_counts[operation_name] += 1
    if sum(call_counts.values()) % 100 == 0:
        logger.info("sdk call counts snapshot=%s", dict(call_counts))

// --- TypeScript (AWS SDK v3) ---
const callCounts = new Map<string, number>();
let total = 0;
 
function recordCall(operationName: string) {
  callCounts.set(operationName, (callCounts.get(operationName) ?? 0) + 1);
  total += 1;
  if (total % 100 === 0) {
    console.log("sdk call counts snapshot", Object.fromEntries(callCounts));
  }
}

Per-operation call counts catch a runaway loop long before it shows up as a bill or a throttling error.
In production, prefer emitting this as a real metric (CloudWatch custom metric or an OTel counter) rather than a periodic log line.
This complements per-call logging - one shows individual latency, the other shows volume trends.
Watch for a single operation dominating the counts; that is often a missing pagination or batching opportunity.

Stack versions: This page was written for boto3 1.43.x (Python 3.10+) and the AWS SDK for JavaScript v3 (Node.js 18+).

Each example shows the same instrumentation pattern in both Python (boto3) and TypeScript (AWS SDK for JavaScript v3).

Prerequisites

Python: pip install boto3 (boto3 1.43.x, Python 3.10+). TypeScript: npm install @aws-sdk/client-s3 @aws-sdk/client-dynamodb (Node.js 18+).
A logger already configured - the standard library logging module in Python, or any structured logger in TypeScript (these examples use console as a stand-in).
Credentials and a default region configured, same as any other SDK call.

Basic Examples

1. Log a Request Id and Latency

The smallest useful instrumentation: time the call, then log its request id.

# --- Python (boto3) ---
import time
import logging
import boto3
 
logger = logging.getLogger("app")
s3 = boto3.client("s3")
 
start = time.monotonic()
response = s3.list_buckets()
duration_ms = (time.monotonic() - start) * 1000
 
logger.info(
    "s3.list_buckets ok request_id=%s duration_ms=%.1f",
    response["ResponseMetadata"]["RequestId"],
    duration_ms,
)

// --- TypeScript (AWS SDK v3) ---
import { S3Client, ListBucketsCommand } from "@aws-sdk/client-s3";
 
const s3 = new S3Client({});
 
const start = Date.now();
const response = await s3.send(new ListBucketsCommand({}));
const durationMs = Date.now() - start;
 
console.log(
  `s3.ListBuckets ok requestId=${response.$metadata.requestId} durationMs=${durationMs}`,
);

The request id uniquely identifies this call on AWS's side, useful for support cases.
time.monotonic() and Date.now() are safe for measuring elapsed time; avoid wall-clock time for durations.
This costs a few lines and no dependencies, and already answers "was this call slow."
Nothing about the bucket contents is logged - only metadata about the call.

Related: Observability for AWS SDK Applications - why this matters.

2. Log the Operation Name

Add the operation name so one log line is self-describing without context from surrounding code.

# --- Python (boto3) ---
import time
import logging
import boto3
 
logger = logging.getLogger("app")
ddb = boto3.client("dynamodb")
 
def timed_call(operation_name, fn):
    start = time.monotonic()
    response = fn()
    duration_ms = (time.monotonic() - start) * 1000
    logger.info(
        "%s ok request_id=%s duration_ms=%.1f",
        operation_name,
        response["ResponseMetadata"]["RequestId"],
        duration_ms,
    )
    return response
 
item = timed_call(
    "dynamodb.GetItem",
    lambda: ddb.get_item(TableName="Orders", Key={"id": {"S": "order-42"}}),
)

// --- TypeScript (AWS SDK v3) ---
import { DynamoDBClient, GetItemCommand } from "@aws-sdk/client-dynamodb";
 
const ddb = new DynamoDBClient({});
 
async function timedCall<T extends { $metadata: { requestId?: string } }>(
  operationName: string,
  fn: () => Promise<T>,
): Promise<T> {
  const start = Date.now();
  const response = await fn();
  const durationMs = Date.now() - start;
  console.log(
    `${operationName} ok requestId=${response.$metadata.requestId} durationMs=${durationMs}`,
  );
  return response;
}
 
const item = await timedCall("dynamodb.GetItem", () =>
  ddb.send(new GetItemCommand({ TableName: "Orders", Key: { id: { S: "order-42" } } })),
);

Wrapping the call in a small helper avoids repeating the timing boilerplate at every call site.
The operation name (dynamodb.GetItem) should match the AWS API name for easy cross-referencing with docs and CloudTrail.
This pattern scales to any client and any operation without modification.
Keep the helper thin; it should never inspect or log the item contents.

3. Log Failures With the Error Code

A failed call should log the AWS error code, not just "it failed."

# --- Python (boto3) ---
import logging
import boto3
from botocore.exceptions import ClientError
 
logger = logging.getLogger("app")
s3 = boto3.client("s3")
 
try:
    s3.get_object(Bucket="my-bucket", Key="missing.txt")
except ClientError as err:
    logger.warning(
        "s3.get_object failed code=%s request_id=%s",
        err.response["Error"]["Code"],
        err.response["ResponseMetadata"]["RequestId"],
    )

// --- TypeScript (AWS SDK v3) ---
import { S3Client, GetObjectCommand } from "@aws-sdk/client-s3";
 
const s3 = new S3Client({});
 
try {
  await s3.send(new GetObjectCommand({ Bucket: "my-bucket", Key: "missing.txt" }));
} catch (err) {
  const e = err as { name: string; $metadata?: { requestId?: string } };
  console.warn(`s3.GetObject failed code=${e.name} requestId=${e.$metadata?.requestId}`);
}

err.response["Error"]["Code"] (boto3) and err.name (SDK v3) give a stable, matchable error code.
Always include the request id on failures - it is the single field AWS support asks for first.
Log at warning or error level so failures stand out from normal info traffic.
Never log the full exception's request context if it includes input parameters that may carry sensitive data.

Related: Structured Logging Around SDK Calls - a fuller logging pattern.

4. Add a Correlation Id

Pass a correlation id through your own request so all its downstream SDK calls share one identifier.

# --- Python (boto3) ---
import logging
import uuid
import boto3
 
logger = logging.getLogger("app")
s3 = boto3.client("s3")
 
def handle_upload(correlation_id: str | None = None):
    correlation_id = correlation_id or str(uuid.uuid4())
    response = s3.put_object(Bucket="my-bucket", Key="notes/hello.txt", Body=b"hi")
    logger.info(
        "s3.put_object ok correlation_id=%s request_id=%s",
        correlation_id,
        response["ResponseMetadata"]["RequestId"],
    )

// --- TypeScript (AWS SDK v3) ---
import { randomUUID } from "crypto";
import { S3Client, PutObjectCommand } from "@aws-sdk/client-s3";
 
const s3 = new S3Client({});
 
async function handleUpload(correlationId = randomUUID()) {
  const response = await s3.send(
    new PutObjectCommand({ Bucket: "my-bucket", Key: "notes/hello.txt", Body: "hi" }),
  );
  console.log(`s3.PutObject ok correlationId=${correlationId} requestId=${response.$metadata.requestId}`);
}

A correlation id (often from an incoming HTTP request header) lets you group every SDK call made while serving one user request.
The AWS request id identifies the call to AWS; the correlation id identifies the call to you. Log both.
This is the manual precursor to what a tracer does automatically with trace and span ids.
Generate the correlation id once per incoming request, not per SDK call.

5. Log Retry Attempts

Both SDKs retry transient failures automatically; logging when they do makes hidden retries visible.

# --- Python (boto3) ---
import logging
import boto3
 
logger = logging.getLogger("app")
ddb = boto3.client("dynamodb")
 
def log_retry(**kwargs):
    logger.info("dynamodb retry attempt=%s", kwargs.get("attempt"))
 
ddb.meta.events.register("needs-retry.dynamodb.PutItem", log_retry)
ddb.put_item(TableName="Visits", Item={"id": {"S": "user-7"}})

// --- TypeScript (AWS SDK v3) ---
import { DynamoDBClient, PutItemCommand } from "@aws-sdk/client-dynamodb";
 
const ddb = new DynamoDBClient({});
 
ddb.middlewareStack.add(
  (next) => async (args) => {
    try {
      return await next(args);
    } catch (err) {
      console.log("dynamodb retry observed", err);
      throw err;
    }
  },
  { step: "finalizeRequest", tags: ["RETRY_LOGGING"] },
);
 
await ddb.send(new PutItemCommand({ TableName: "Visits", Item: { id: { S: "user-7" } } }));

boto3 exposes a needs-retry lifecycle event per operation you can hook into.
SDK v3's retry behavior is easiest to observe via its built-in retry strategy metadata or middleware, since retries happen inside the client's send pipeline.
Seeing retry counts in logs often explains latency that would otherwise look unexplained.
Do not treat every retry as an error - occasional throttling retries are normal and expected.

Intermediate Examples

6. Build a Reusable Logging Wrapper

Consolidate timing, request id, and error logging into one wrapper used across all clients.

# --- Python (boto3) ---
import time
import logging
from botocore.exceptions import ClientError
 
logger = logging.getLogger("app")
 
def observed_call(operation_name, fn):
    start = time.monotonic()
    try:
        response = fn()
        duration_ms = (time.monotonic() - start) * 1000
        logger.info(
            "%s ok request_id=%s duration_ms=%.1f",
            operation_name,
            response["ResponseMetadata"]["RequestId"],
            duration_ms,
        )
        return response
    except ClientError as err:
        duration_ms = (time.monotonic() - start) * 1000
        logger.warning(
            "%s failed code=%s duration_ms=%.1f",
            operation_name,
            err.response["Error"]["Code"],
            duration_ms,
        )
        raise

// --- TypeScript (AWS SDK v3) ---
async function observedCall<T extends { $metadata: { requestId?: string } }>(
  operationName: string,
  fn: () => Promise<T>,
): Promise<T> {
  const start = Date.now();
  try {
    const response = await fn();
    console.log(`${operationName} ok requestId=${response.$metadata.requestId} durationMs=${Date.now() - start}`);
    return response;
  } catch (err) {
    const e = err as { name: string };
    console.warn(`${operationName} failed code=${e.name} durationMs=${Date.now() - start}`);
    throw err;
  }
}

One wrapper used across every client call keeps logging consistent without repeating the try/timing code.
This is the natural place to later add a tracer span, since it already brackets each call.
Keep the wrapper generic - it should not know about any specific service's fields.
Reuse it for both reads and writes, across every client in the codebase.

Related: AWS X-Ray Tracing for SDK-Calling Services - turning this wrapper into a trace subsegment.

7. Log Aggregate Call Counts

Track how many calls per operation your service makes, not just individual latencies.

# --- Python (boto3) ---
import logging
from collections import Counter
 
logger = logging.getLogger("app")
call_counts: Counter = Counter()
 
def record_call(operation_name: str):
    call_counts[operation_name] += 1
    if sum(call_counts.values()) % 100 == 0:
        logger.info("sdk call counts snapshot=%s", dict(call_counts))

// --- TypeScript (AWS SDK v3) ---
const callCounts = new Map<string, number>();
let total = 0;
 
function recordCall(operationName: string) {
  callCounts.set(operationName, (callCounts.get(operationName) ?? 0) + 1);
  total += 1;
  if (total % 100 === 0) {
    console.log("sdk call counts snapshot", Object.fromEntries(callCounts));
  }
}

Per-operation call counts catch a runaway loop long before it shows up as a bill or a throttling error.
In production, prefer emitting this as a real metric (CloudWatch custom metric or an OTel counter) rather than a periodic log line.
This complements per-call logging - one shows individual latency, the other shows volume trends.
Watch for a single operation dominating the counts; that is often a missing pagination or batching opportunity.

Stack versions: This page was written for boto3 1.43.x (Python 3.10+) and the AWS SDK for JavaScript v3 (Node.js 18+).

Observability Basics

Buscar en la documentación

Observability Basics