DEV Community: Mohammad Jawad (Kasir) Barati

does pydantic-ai's structured-output API actually work against Ollama's OpenAI-compatible endpoint?

Mohammad Jawad (Kasir) Barati — Sun, 05 Jul 2026 20:25:56 +0000

Question

This is the biggest single technical question you might have while using pydantic-ai:

Will pydantic-ai round-trip respect output_type through Ollama's /v1/chat/completions cleanly? If yes, then you can rely on structured output.

If it can't, then you have to resort to manual validation & enforcement in your application — post-response. But fortunately that is not the case 😅.

Stack

llama3.2:1b
Ollama 0.17.7
Python: 3.12

Findings

Imagine you have something like the following:

from __future__ import annotations

from pydantic_ai import Agent
from pydantic_ai.output import NativeOutput
from pydantic import BaseModel, Field


class WordExplanation(BaseModel):
    """Structured explanation of one word as it is used in a specific context."""

    meaning: str = Field(
        description="Concise dictionary-style definition of the word as used in context.",
    )
    simplified_explanation: str = Field(
        description="Same idea expressed for a younger or non-native reader.",
    )
    synonyms: list[str] = Field(
        default_factory=list,
        description="Up to five words meaning roughly the same thing.",
    )
    antonyms: list[str] = Field(
        default_factory=list,
        description="Up to five words meaning roughly the opposite.",
    )

provider = OpenAIProvider(
    base_url="http://ollama:11434/v1",
    api_key="some-key",
)
model = OpenAIChatModel("llama3.2:1b", provider=provider)
Agent(
    model,
    output_type=NativeOutput(WordExplanation),
    model_settings=resolve_model_settings(settings),
)

NativeOutput works: pydantic-ai sends response_format={"type": "json_schema", ...} and Ollama's OpenAI Compatible layer honours it. Returned a valid WordExplanation.

Script I Used to Verify It

from __future__ import annotations

import asyncio
import os
import subprocess
import sys
import time
from dataclasses import dataclass
from pathlib import Path
from typing import Any

import httpx
from pydantic import BaseModel, Field
from pydantic_ai import Agent
from pydantic_ai.models.openai import OpenAIChatModel
from pydantic_ai.output import NativeOutput, PromptedOutput, ToolOutput
from pydantic_ai.providers.openai import OpenAIProvider


LLM_BASE_URL = os.environ.get("LLM_BASE_URL", "http://localhost:11434/v1")
LLM_API_KEY = os.environ.get("LLM_API_KEY", "ollama")
LLM_MODEL = os.environ.get("LLM_MODEL", "llama3.2:1b")
SPIKE_WORD = os.environ.get("SPIKE_WORD", "ephemeral")
SPIKE_CONTEXT = os.environ.get(
    "SPIKE_CONTEXT",
    "The graffiti was ephemeral, washed away by the first rain of the season.",
)
SPIKE_AUTO_START = os.environ.get("SPIKE_AUTO_START", "true").lower() in {"1", "true", "yes"}

PROJECT_ROOT = Path(__file__).resolve().parent.parent


class WordExplanation(BaseModel):
    """Structured explanation of a single word in context."""

    meaning: str = Field(
        description="Concise dictionary-style definition of the word as used in context.",
    )
    simplified_explanation: str = Field(
        description="Same idea expressed for a younger or non-native reader.",
    )
    synonyms: list[str] = Field(
        default_factory=list,
        description="Up to five words meaning roughly the same thing.",
    )
    antonyms: list[str] = Field(
        default_factory=list,
        description="Up to five words meaning roughly the opposite.",
    )


SYSTEM_PROMPT = (
    "You are a dictionary. Given a WORD and its CONTEXT, return the WordExplanation "
    "matching how the word is used in that specific sentence. Prefer short, plain "
    "language for `simplified_explanation`. Limit synonyms and antonyms to five each."
)


def build_user_prompt(word: str, context: str) -> str:
    return f"WORD: {word}\nCONTEXT: {context}"


# ---------------------------------------------------------------------------
# Ollama plumbing
# ---------------------------------------------------------------------------
def _log(msg: str) -> None:
    print(f"[spike] {msg}", flush=True)


def _endpoint_reachable(base_url: str, timeout: float = 2.0) -> bool:
    """True iff Ollama's ``/api/tags`` responds — cheaper than a real call."""

    # `LLM_BASE_URL` includes the `/v1` OpenAI suffix; the native probe is on `/api/tags`.
    root = base_url.rstrip("/").removesuffix("/v1").rstrip("/")
    try:
        resp = httpx.get(f"{root}/api/tags", timeout=timeout)
        return resp.status_code == 200
    except httpx.HTTPError:
        return False


def _wait_for_endpoint(base_url: str, deadline_s: float) -> None:
    start = time.monotonic()
    while time.monotonic() - start < deadline_s:
        if _endpoint_reachable(base_url):
            _log(f"Ollama reachable at {base_url} after {time.monotonic() - start:.1f}s")
            return
        time.sleep(2)
    raise TimeoutError(f"Ollama did not become reachable at {base_url} within {deadline_s}s")


def _start_ollama_via_compose() -> None:
    """Best-effort ``docker compose up -d ollama`` from the project root."""

    _log("Ollama not reachable — running `docker compose up -d ollama`...")
    result = subprocess.run(
        ["docker", "compose", "up", "-d", "ollama"],
        cwd=PROJECT_ROOT,
        capture_output=True,
        text=True,
        check=False,
    )
    if result.returncode != 0:
        _log(f"docker compose stderr:\n{result.stderr}")
        raise RuntimeError("Failed to start ollama via docker compose")
    _log("docker compose returned successfully; waiting for the endpoint...")


def ensure_ollama_ready() -> None:
    if _endpoint_reachable(LLM_BASE_URL):
        _log(f"Ollama already reachable at {LLM_BASE_URL}")
        return

    if not SPIKE_AUTO_START:
        raise RuntimeError(
            f"Ollama not reachable at {LLM_BASE_URL} and SPIKE_AUTO_START=false. "
            "Start it manually with `docker compose up -d ollama`.",
        )

    _start_ollama_via_compose()
    # The compose ollama image pre-pulls the model at build time, so the
    # daemon only needs to boot — 60s is generous.
    _wait_for_endpoint(LLM_BASE_URL, deadline_s=180.0)


# ---------------------------------------------------------------------------
# Spike runs
# ---------------------------------------------------------------------------
@dataclass
class RunOutcome:
    label: str
    ok: bool
    latency_s: float
    parsed: WordExplanation | None
    error: str | None


def _build_model() -> OpenAIChatModel:
    provider = OpenAIProvider(base_url=LLM_BASE_URL, api_key=LLM_API_KEY)
    return OpenAIChatModel(LLM_MODEL, provider=provider)


async def _run_once(
    label: str,
    output_spec: Any,
    per_call_timeout_s: float = 120.0,
) -> RunOutcome:
    _log(f"--- Running mode: {label} ---")
    agent = Agent(
        _build_model(),
        output_type=output_spec,
        system_prompt=SYSTEM_PROMPT,
    )
    start = time.monotonic()
    try:
        result = await asyncio.wait_for(
            agent.run(build_user_prompt(SPIKE_WORD, SPIKE_CONTEXT)),
            timeout=per_call_timeout_s,
        )
    except TimeoutError:
        latency = time.monotonic() - start
        _log(f"    timed out after {latency:.1f}s")
        return RunOutcome(
            label=label,
            ok=False,
            latency_s=latency,
            parsed=None,
            error=f"timeout after {per_call_timeout_s:.0f}s",
        )
    except Exception as exc:
        latency = time.monotonic() - start
        _log(f"    failed after {latency:.1f}s: {type(exc).__name__}")
        return RunOutcome(
            label=label,
            ok=False,
            latency_s=latency,
            parsed=None,
            error=f"{type(exc).__name__}: {exc}",
        )

    latency = time.monotonic() - start
    parsed = result.output
    if not isinstance(parsed, WordExplanation):
        _log(f"    completed but returned {type(parsed).__name__}")
        return RunOutcome(
            label=label,
            ok=False,
            latency_s=latency,
            parsed=None,
            error=f"unexpected output type: {type(parsed).__name__}",
        )
    _log(f"    ok in {latency:.1f}s")
    return RunOutcome(
        label=label,
        ok=True,
        latency_s=latency,
        parsed=parsed,
        error=None,
    )


async def run_all() -> list[RunOutcome]:
    return [
        await _run_once("ToolOutput (default)", ToolOutput(WordExplanation)),
        await _run_once("NativeOutput (JSON schema)", NativeOutput(WordExplanation)),
        await _run_once("PromptedOutput (fallback)", PromptedOutput(WordExplanation)),
    ]


def _print_outcome(outcome: RunOutcome) -> None:
    header = f"[{outcome.label}]  {'OK' if outcome.ok else 'FAIL'}  ({outcome.latency_s:.2f}s)"
    print(header)
    if outcome.ok and outcome.parsed is not None:
        for field in ("meaning", "simplified_explanation"):
            print(f"  {field:<24} {getattr(outcome.parsed, field)}")
        print(f"  synonyms                 {outcome.parsed.synonyms}")
        print(f"  antonyms                 {outcome.parsed.antonyms}")
    else:
        print(f"  error: {outcome.error}")
    print()


def main() -> int:
    _log(f"Target model  : {LLM_MODEL}")
    _log(f"Target endpoint: {LLM_BASE_URL}")
    _log(f"Prompt word   : {SPIKE_WORD!r}")
    _log(f"Prompt context: {SPIKE_CONTEXT!r}")
    print()

    ensure_ollama_ready()

    outcomes = asyncio.run(run_all())

    print("=" * 72)
    print("SPIKE RESULTS")
    print("=" * 72)
    for outcome in outcomes:
        _print_outcome(outcome)

    any_ok = any(o.ok for o in outcomes)
    print("VERDICT:", "at least one mode works" if any_ok else "no mode worked — fallback needed")
    return 0 if any_ok else 1


if __name__ == "__main__":
    sys.exit(main())

And the compose.yml:

services:
  ollama:
    build:
      context: ollama
      dockerfile: Dockerfile
      args:
        OLLAMA_MODEL: ${OLLAMA_MODEL:-llama3.2:1b}
    ports:
      - "11434:11434" # Expose the OpenAI-compatible endpoint on the host for host-run scripts (e.g. scripts/spike_pydantic_ai_ollama.py).
    healthcheck:
      test: ["CMD-SHELL", "/usr/local/bin/healthcheck.sh"]
      interval: 30s
      timeout: 5s
      retries: 5
      start_period: 45s

You also need to create a ollama/Dockerfile:

# Kept self-contained here so developers can work without cloning the model over and over again.

FROM ollama/ollama:0.17.7

ARG OLLAMA_MODEL=llama3.2:1b
ENV OLLAMA_MODEL=${OLLAMA_MODEL}
# CAUTION: 0.0.0.0 means Ollama listens on every network interface — required for the healthcheck and for the container to be reachable from a sibling container.
ENV OLLAMA_HOST=0.0.0.0
ENV OLLAMA_PORT=11434

# curl for build-time readiness checks + the runtime healthcheck.
RUN apt-get update && \
    apt-get install -y --no-install-recommends ca-certificates curl && \
    rm -rf /var/lib/apt/lists/*

COPY --chmod=755 prepull-model.sh /usr/local/bin/prepull-model.sh
RUN /usr/local/bin/prepull-model.sh

COPY --chmod=755 healthcheck.sh /usr/local/bin/healthcheck.sh
HEALTHCHECK --interval=30s --timeout=5s --retries=5 --start-period=45s CMD ["/usr/local/bin/healthcheck.sh"]

# IMPORTANT: prepull-model.sh stops the daemon when it's done. We start a fresh one here so the container becomes usable.
CMD ["serve"]

And ollama/healthcheck.sh:

#!/bin/sh
set -eu

HOST="${OLLAMA_HOST:-127.0.0.1}"
PORT="${OLLAMA_PORT:-11434}"
MODEL="${OLLAMA_MODEL:-llama3.2:1b}"

# 1) API up?
if ! curl -fsS "http://${HOST}:${PORT}/api/version" >/dev/null; then
  exit 1
fi

# 2) Model registered locally?
if ! curl -fsS "http://${HOST}:${PORT}/api/tags" | grep -q "\"name\":\"${MODEL}\""; then
  exit 1
fi

# 3) /api/generate usable? (Minimal, cheap request; do not exercise the model.)
CODE="$(curl -sS -o /dev/null -w '%{http_code}' \
  -H 'Content-Type: application/json' \
  -X POST "http://${HOST}:${PORT}/api/generate" \
  -d "{\"model\":\"${MODEL}\",\"prompt\":\"ping\",\"stream\":false,\"options\":{\"num_predict\":1}}")"

[ "$CODE" -eq 200 ]

Lastly ollama/prepull-model.sh:

#!/bin/sh
set -eu

HOST="${OLLAMA_HOST:-127.0.0.1}"
PORT="${OLLAMA_PORT:-11434}"
MODEL="${OLLAMA_MODEL:-llama3.2:1b}"
READY_TRIES="${OLLAMA_READY_TRIES:-60}"
READY_SLEEP="${OLLAMA_READY_SLEEP:-1}"

log() { printf '%s\n' "$*" >&2; }

cleanup() {
  if [ -n "${PID:-}" ] && kill -0 "$PID" 2>/dev/null; then
    kill -TERM "$PID" 2>/dev/null || true
    wait "$PID" 2>/dev/null || true
  fi
}
trap cleanup EXIT INT TERM

log "Starting ollama daemon for build-time model pull..."
ollama serve >/tmp/ollama-build.log 2>&1 &
PID=$!

i=0
while [ "$i" -lt "$READY_TRIES" ]; do
  if curl -fsS "http://${HOST}:${PORT}/api/version" >/dev/null; then
    break
  fi
  i=$((i+1))
  sleep "$READY_SLEEP"
done

if ! curl -fsS "http://${HOST}:${PORT}/api/version" >/dev/null; then
  log "Ollama daemon did not become ready during build."
  log "Last log lines:"; tail -n 120 /tmp/ollama-build.log || true
  exit 1
fi

log "Daemon is ready. Pulling model: ${MODEL}"
ollama pull "${MODEL}"

log "Listing models:"; ollama list || true
log "Done."

Following a Request Across Services with OpenTelemetry: How I Wired smart-novel piper-tts-rest-api

Mohammad Jawad (Kasir) Barati — Mon, 29 Jun 2026 09:50:39 +0000

When a request crosses a service boundary, the most useful question you can answer in production is: "What did this one call actually do?" Not "what did service A do" and then "what did service B do" in two separate windows. Instead you wanna see is one trace, one timeline, and everything in the same tab.

That is exactly what I did in my side project: a single GraphQL mutation GenerateChapterAudio in my smart-novel backend (NestJS) calls out over HTTP to my piper-tts-rest-api service (plain Node), and Jaeger renders both services as branches of the same trace.

The two yellow spans are children of the smart-novel POST span in the parent service (smart-novel). That parent/child relationship across a network hop is the whole point of distributed tracing.

The Mental Model in One Sentence

OpenTelemetry standardises a tiny HTTP header called traceparent. If smart-novel adds it to the outgoing request and piper-tts-rest-api extracts it from the incoming request, both services will emit spans that share the same trace_id, piper-tts-rest-api's spans will listed under the smart-novel's spans. Jaeger then knows how to draw the tree you saw.

Bootstrapping the OpenTelemetry SDK

OpenTelemetry's auto-instrumentation works by monkey-patching libraries. That means the SDK must start before any of those modules are loaded — otherwise the instrumentation grabs a reference to the original, unpatched function and you get no spans at all.

For that we have two options:

Like what I am doing in smart-novel we must import apps/backend/src/instrumentation.ts as the first import in the entrypoint file which in our NestJS app would be main.ts.
- OTLPTraceExporter which exports OTel data over HTTP.
- getNodeAutoInstrumentations(...) disables noisy instrumentations:
  - fs, dns, net, express, router.
  - Silences the optional winston-transport warning.
  - JWKS fetches to Zitadel is filtered out at the source via ignoreOutgoingRequestHook.
And what we can have an src/instrumentation.ts piper-tts-rest-api and then use --import where you have your start script in package.json.
- HttpInstrumentation enables automatic tracing for Node’s HTTP module.
- piper-tts-rest-api is ESM, and ESM is a different beast. A import './instrumentation.js' at the top of server.ts is not enough, because ESM hoists every import in a module to the top of its evaluation phase. By the time instrumentation.js's body runs, node:http has already been resolved and its exports captured.

BatchSpanProcessor with explicit queue size, batch size, flush interval, export timeout. While propagators read and write trace context from request headers.

Propagator is a component within the @opentelemetry/core package responsible for serializing and deserializing context data so it can be transmitted across network boundaries to remote services.

smart-novel <> piper-tts-rest-api

@opentelemetry/instrumentation-http propagates traceparent the headers for every to the outgoing requests (caller), then @opentelemetry/instrumentation-http in piper-tts-rest-api extract traceparent from the incoming request.

💡Tip

I created "tts.synthesize" span to separate things, e.g. outside of the new span I can validate and return a 400 error if the incoming request body is invalid.

You can do the same in smart-novel, e.g. adding attributes to the span.

`x-trace-id` Header

I add the trace ID to the response which is great for debugging. In my smart-novel we have trace-id.interceptor.ts and in piper-tts-rest-api I am manually adding it.

🛈 x-trace-id VS traceparent:

x-trace-id is intentionally separate from the W3C traceparent header. traceparent is reserved for service-to-service propagation and is set by the OTel propagators themselves whilst x-trace-id is meant to be used for debugging purposes.

Two Small Ergonomic Touches I'm Proud of

Renaming the GraphQL root span: instrumentation-http names every server span POST /graphql — which is useless when you have 30 mutations and queries. So I wrote a tiny Apollo plugin that renames the active root span the moment Apollo finishes parsing/validating the operation (graphql-span-rename.plugin.ts). That is why the top of the screenshot says mutation GenerateChapterAudio instead of POST /graphql.
Tail-sampling at the OpenTelemetry collector: the OpenTelemetry Collector between the app and Jaeger is configured to use tail_sampling, so that:
- Always keep error traces, regardless of what else is going on.
- Drop known-noisy operations (health checks, introspection queries) using the graphql.operation.name attribute the plugin above sets.
- Probabilistically sample the rest, currently at 100% for dev env, and for production we can tone it down.

Visualization

piper-tts-rest-api" width="800" height="525">

Putting Everything Together

If you're trying to replicate this in your own stack, here's the minimal checklist:

Start the OTel SDK before anything else loads.
Use and configure the same propagators on both sides so they can decode and encode the messages.
Do NOT auto-instrument everything (although I have done exactly that ATM with Prisma 😅).
Echo the trace ID back to clients (x-trace-id response header). Future-you will be very, very grateful.

OpenTelemetry Collector Configuration Explained

Mohammad Jawad (Kasir) Barati — Sat, 27 Jun 2026 18:06:14 +0000

I decided tp configure OTel in 9109679196/piper-tts-rest-api, but then I decided to share my findings about OpenTelemetry Collector configuration. And JFYI you can find the official collector configuration documentation here.

extensions:
  health_check:
    endpoint: 0.0.0.0:13133

receivers:
  otlp:
    protocols:
      http:
        endpoint: 0.0.0.0:4318
      grpc:
        endpoint: 0.0.0.0:4317

processors:
  batch:
    timeout: 5s
    send_batch_size: 512
    send_batch_max_size: 2048

  # Tail sampling — always keep error traces, probabilistically sample the rest.
  tail_sampling:
    decision_wait: 10s
    num_traces: 50000
    expected_new_traces_per_sec: 50
    policies:
      - name: keep-errors
        type: status_code
        status_code:
          status_codes: [ERROR]
      - name: probabilistic-sample
        type: probabilistic
        probabilistic:
          sampling_percentage: 100

exporters:
  otlp/jaeger:
    endpoint: jaeger:4317
    tls:
      insecure: true
  debug:
    verbosity: basic

service:
  extensions: [health_check]
  pipelines:
    traces:
      receivers: [otlp]
      processors: [tail_sampling, batch]
      exporters: [otlp/jaeger]

Big picture

An OpenTelemetry Collector configuration is usually built from these main sections:

receivers: how telemetry gets into the Collector.
processors: how telemetry is modified, grouped, filtered, sampled, or enriched.
exporters: where telemetry is sent after processing.
extensions: extra Collector functionality that is not directly part of the telemetry pipeline.
service: where you enable extensions and wire receivers, processors, and exporters together into pipelines.

Official docs:

Collector configuration: https://opentelemetry.io/docs/collector/configuration/
Collector overview: https://opentelemetry.io/docs/collector/

`extensions`

extensions:
  health_check:
    endpoint: 0.0.0.0:13133

What it does

extensions define extra behavior for the Collector itself. Extensions are not directly involved in receiving, processing, or exporting traces, metrics, or logs.

In this config, you enable the health_check extension.

`health_check`

health_check:
  endpoint: 0.0.0.0:13133

This exposes an HTTP health endpoint for the Collector.

In Docker Compose, another container can call:

http://otel-collector:13133/

This is useful for checking whether the Collector process is alive and ready enough to respond.

`endpoint: 0.0.0.0:13133`

This means the health check server listens on all network interfaces inside the container on port 13133.

If you used only localhost:13133, another container might not be able to reach it. For Docker Compose, 0.0.0.0:13133 is usually the practical choice.

Official docs:

Health check extension: https://github.com/open-telemetry/opentelemetry-collector-contrib/tree/main/extension/healthcheckextension

`receivers`

receivers:
  otlp:
    protocols:
      http:
        endpoint: 0.0.0.0:4318
      grpc:
        endpoint: 0.0.0.0:4317

What it does

A receiver is how telemetry enters the Collector.

Your config enables the otlp receiver. OTLP means OpenTelemetry Protocol. It is the standard protocol used by OpenTelemetry SDKs and instrumentation libraries to send traces, metrics, and logs.

Official docs:

Receivers in Collector configuration: https://opentelemetry.io/docs/collector/configuration/#receivers
OTLP receiver: https://github.com/open-telemetry/opentelemetry-collector/tree/main/receiver/otlpreceiver

`otlp`

otlp:

This configures the Collector to accept telemetry in OTLP format.

For your NestJS app, this is the receiver your app will send traces to.

`protocols`

protocols:
  http:
    endpoint: 0.0.0.0:4318
  grpc:
    endpoint: 0.0.0.0:4317

The OTLP receiver can listen using HTTP and/or gRPC.

You enabled both:

OTLP/HTTP on port 4318
OTLP/gRPC on port 4317

`http.endpoint: 0.0.0.0:4318`

This makes the Collector listen for OTLP over HTTP.

Your app would usually send HTTP OTLP traces to something like:

http://otel-collector:4318/v1/traces

`grpc.endpoint: 0.0.0.0:4317`

This makes the Collector listen for OTLP over gRPC.

Your app would usually send gRPC OTLP traces to:

otel-collector:4317

Use this if your SDK/exporter is configured for OTLP/gRPC.

`processors`

processors:
  batch:
    timeout: 5s
    send_batch_size: 512
    send_batch_max_size: 2048

  tail_sampling:
    decision_wait: 10s
    num_traces: 50000
    expected_new_traces_per_sec: 50
    policies:
      - name: keep-errors
        type: status_code
        status_code:
          status_codes: [ERROR]
      - name: probabilistic-sample
        type: probabilistic
        probabilistic:
          sampling_percentage: 100

What processors do

Processors sit between receivers and exporters.

They can batch, filter, sample, enrich, transform, or otherwise modify telemetry before it is exported.

In your trace pipeline, processors are applied in this order:

processors: [tail_sampling, batch]

That means:

First, the Collector decides which traces to keep using tail_sampling.
Then, the kept traces are batched using batch before sending them to Jaeger.

This order makes sense. Sampling drops data first, and batching groups the remaining data before export.

Official docs:

Processors in Collector configuration: https://opentelemetry.io/docs/collector/configuration/#processors

`batch` processor

batch:
  timeout: 5s
  send_batch_size: 512
  send_batch_max_size: 2048

What it does

The batch processor groups spans together before exporting them.

This is useful because sending many small requests to your backend is inefficient. Batching can reduce network overhead and improve export performance.

Official docs:

Batch processor: https://github.com/open-telemetry/opentelemetry-collector/tree/main/processor/batchprocessor

`timeout: 5s`

The Collector sends a batch after 5s, even if the batch has not reached send_batch_size.

So this puts a maximum wait time on batching.

`send_batch_size: 512`

This is the batch size trigger.

When the Collector has collected around 512 spans, metric data points, or log records, it will send a batch.

In your trace pipeline, this applies to spans.

`send_batch_max_size: 2048`

This is the maximum batch size allowed.

If a batch becomes larger than 2048, the processor will split it into smaller batches.

A key detail: send_batch_max_size must be greater than or equal to send_batch_size.

Your config is valid because:

2048 >= 512

`tail_sampling` processor

tail_sampling:
  decision_wait: 10s
  num_traces: 50000
  expected_new_traces_per_sec: 50
  policies:
    - name: keep-errors
      type: status_code
      status_code:
        status_codes: [ERROR]
    - name: probabilistic-sample
      type: probabilistic
      probabilistic:
        sampling_percentage: 100

What it does

Tail sampling means the Collector waits until it has seen enough spans from a trace before deciding whether to keep or drop that trace.

This is different from head sampling, where the decision is made at the beginning of the trace.

Tail sampling is useful because you can make decisions based on the final trace result. For example:

keep traces with errors;
keep slow traces;
keep traces matching specific services or attributes;
sample only a percentage of normal successful traces.

Official docs:

Tail sampling processor: https://github.com/open-telemetry/opentelemetry-collector-contrib/tree/main/processor/tailsamplingprocessor
Tail sampling examples: https://github.com/open-telemetry/opentelemetry-collector-contrib/blob/main/processor/tailsamplingprocessor/testdata/tail_sampling_config.yaml

Important production note

Tail sampling needs all spans for the same trace to arrive at the same Collector instance. If you run multiple Collector replicas later, you need to be careful with load balancing so spans from the same trace are routed to the same Collector.

For local Docker Compose with one Collector instance, this is fine.

`decision_wait: 10s`

The Collector waits 10s from the first span of a trace before making a sampling decision.

This gives other spans in the same distributed trace time to arrive.

If your traces are long or your services are slow, you may need a higher value. If you want faster export, you may reduce it.

Trade-off:

Higher value: better chance of complete traces, but more memory usage and more delay.
Lower value: less memory and faster export, but higher chance of incomplete sampling decisions.

`num_traces: 50000`

This is the number of traces the processor keeps in memory while waiting to make sampling decisions.

Higher traffic usually needs a higher value.

If this is too low, the Collector may be forced to drop or evict traces before making the right decision.

`expected_new_traces_per_sec: 50`

This tells the processor roughly how many new traces per second to expect.

It helps the processor allocate internal data structures more efficiently.

For local development, 50 is usually fine unless you generate a lot of load.

`policies`

Policies define which traces should be sampled/kept.

You configured two policies:

policies:
  - name: keep-errors
    type: status_code
    status_code:
      status_codes: [ERROR]
  - name: probabilistic-sample
    type: probabilistic
    probabilistic:
      sampling_percentage: 100

Tail sampling policy: `keep-errors`

- name: keep-errors
  type: status_code
  status_code:
    status_codes: [ERROR]

What it does

This policy keeps traces where the span status code is ERROR.

This is useful because error traces are usually the traces you most want to inspect in Jaeger.

`name: keep-errors`

A human-readable name for this policy.

`type: status_code`

This means the policy makes its decision based on span status code.

`status_codes: [ERROR]`

This means traces with error spans should be sampled/kept.

Tail sampling policy: `probabilistic-sample`

- name: probabilistic-sample
  type: probabilistic
  probabilistic:
    sampling_percentage: 100

What it does

This policy samples a percentage of traces.

In your current config:

sampling_percentage: 100

That means keep 100% of traces.

So right now, your Collector keeps:

all error traces because of keep-errors;
all other traces because probabilistic-sample is 100%.

For local development, this is fine because you probably want to see everything.

For production or heavier load testing, you might reduce this, for example:

sampling_percentage: 10

That would keep all error traces, plus approximately 10% of other traces.

`exporters`

exporters:
  otlp/jaeger:
    endpoint: jaeger:4317
    tls:
      insecure: true
  debug:
    verbosity: basic

What exporters do

Exporters send telemetry out of the Collector to another system.

In your config, there are two exporters defined:

otlp/jaeger
debug

But only otlp/jaeger is currently used in the trace pipeline.

Official docs:

Exporters in Collector configuration: https://opentelemetry.io/docs/collector/configuration/#exporters
OTLP exporter: https://github.com/open-telemetry/opentelemetry-collector/tree/main/exporter/otlpexporter
Debug exporter: https://github.com/open-telemetry/opentelemetry-collector/tree/main/exporter/debugexporter

`otlp/jaeger` exporter

otlp/jaeger:
  endpoint: jaeger:4317
  tls:
    insecure: true

What it does

This exporter sends telemetry from the Collector to Jaeger using OTLP/gRPC.

The name has two parts:

otlp/jaeger

otlp is the exporter type.
jaeger is a custom name you gave to this exporter instance.

This is useful when you want multiple exporters of the same type with different destinations.

`endpoint: jaeger:4317`

This tells the Collector to send data to the service named jaeger on port 4317.

In Docker Compose, jaeger is resolved by Docker's internal DNS to your Jaeger container.

Port 4317 is the usual OTLP/gRPC port.

`tls.insecure: true`

This disables TLS for this exporter connection.

For local Docker Compose, this is normal because the Collector and Jaeger communicate inside a local Docker network.

For production, you should think carefully before using insecure transport.

`debug` exporter

debug:
  verbosity: basic

What it does

The debug exporter prints telemetry to the Collector logs.

This is useful when troubleshooting because you can verify whether the Collector is receiving and processing telemetry.

However, your current pipeline does not use it:

exporters: [otlp/jaeger]

So defining this exporter does nothing unless you add it to a pipeline.

For example:

service:
  pipelines:
    traces:
      receivers: [otlp]
      processors: [tail_sampling, batch]
      exporters: [otlp/jaeger, debug]

Then traces would be sent to Jaeger and also printed in the Collector logs.

`service`

service:
  extensions: [health_check]
  pipelines:
    traces:
      receivers: [otlp]
      processors: [tail_sampling, batch]
      exporters: [otlp/jaeger]

What it does

The service section is where you enable configured components.

Defining a receiver, processor, exporter, or extension is not always enough. You usually also need to reference it in the service section.

Official docs:

Service section and pipelines: https://opentelemetry.io/docs/collector/configuration/#service

`service.extensions`

extensions: [health_check]

This enables the health_check extension that you configured earlier.

Without this line, the health_check config would exist, but it would not actually be started.

`service.pipelines`

pipelines:
  traces:
    receivers: [otlp]
    processors: [tail_sampling, batch]
    exporters: [otlp/jaeger]

A pipeline defines the path telemetry follows through the Collector.

Your config defines only one pipeline: traces.

That means this Collector configuration is currently only handling traces. It is not handling metrics or logs yet.

Trace pipeline flow

Your trace data flows like this:

NestJS app
  -> OTLP receiver
  -> tail_sampling processor
  -> batch processor
  -> OTLP exporter
  -> Jaeger

Or in config terms:

otlp -> tail_sampling -> batch -> otlp/jaeger

`receivers: [otlp]`

The trace pipeline receives spans from the OTLP receiver.

`processors: [tail_sampling, batch]`

The trace pipeline first applies tail sampling, then batching.

`exporters: [otlp/jaeger]`

The trace pipeline sends the final spans to Jaeger.

Practical notes for your current setup

1. Your current sampling keeps everything

Because you configured:

sampling_percentage: 100

You are keeping all traces.

That is OK for local development.

If you want to keep all error traces but only some successful traces, use something like:

sampling_percentage: 10

2. `debug` exporter is configured but unused

You define:

debug:
  verbosity: basic

But your pipeline only uses:

exporters: [otlp/jaeger]

So the debug exporter is inactive.

Use this during troubleshooting:

exporters: [otlp/jaeger, debug]

3. You only configured traces

Your pipeline is:

pipelines:
  traces:

So this config handles traces only.

If later you want metrics or logs, you need separate pipelines, for example:

service:
  pipelines:
    metrics:
      receivers: [otlp]
      processors: [batch]
      exporters: [debug]

    logs:
      receivers: [otlp]
      processors: [batch]
      exporters: [debug]

4. The health check port does not need to be published to your host

For Docker Compose internal checks, this is enough:

http://otel-collector:13133/

You only need this in ports:

- "13133:13133"

if you want to call the health endpoint from your host machine or browser.

Official documentation links

OpenTelemetry Collector overview: https://opentelemetry.io/docs/collector/
Collector configuration: https://opentelemetry.io/docs/collector/configuration/
Health check extension: https://github.com/open-telemetry/opentelemetry-collector-contrib/tree/main/extension/healthcheckextension
OTLP receiver: https://github.com/open-telemetry/opentelemetry-collector/tree/main/receiver/otlpreceiver
Batch processor: https://github.com/open-telemetry/opentelemetry-collector/tree/main/processor/batchprocessor
Tail sampling processor: https://github.com/open-telemetry/opentelemetry-collector-contrib/tree/main/processor/tailsamplingprocessor
Tail sampling example config: https://github.com/open-telemetry/opentelemetry-collector-contrib/blob/main/processor/tailsamplingprocessor/testdata/tail_sampling_config.yaml
OTLP exporter: https://github.com/open-telemetry/opentelemetry-collector/tree/main/exporter/otlpexporter
Debug exporter: https://github.com/open-telemetry/opentelemetry-collector/tree/main/exporter/debugexporter

Auto-assigning a role to self-registered users in Zitadel (with a Zitadel Action)

Mohammad Jawad (Kasir) Barati — Fri, 26 Jun 2026 10:47:30 +0000

Intro

I recently set up self-registration in Zitadel for a side project and ran into a classic chicken-and-egg problem:

Self-registered users land in the system with zero project grants.
Without a grant, my backend rejects them because they have no role claim.
The Management API call that would grant them a role requires a token they don't have.

The fix is a Zitadel Action that runs server-side, on the right trigger, using a service-account PAT to grant the role automatically. Easy in theory — but it took a few iterations to get right, so here's the recipe and the traps.

Stack Context

Zitadel v4.
Docker Compose.
Traefik.
A JS script to configure Zitadel.
NestJS.
ReactJS.

The Three Pieces

1. Enable self-registration on The Login Policy

By default, the "Register" link is hidden on the Zitadel login UI. Flip it on by updating the org's login policy with allowRegister: true. You can see on how you can configure this in through the admin panel as explained here.

Code: zitadel-management-v1.service.js — enableSelfRegistration().

2. Create a Zitadel Action on The Right Trigger

This is the part where you must understand Zitadel's Actions v1 flow/trigger matrix. Some of the relevant combinations:

Flow	Trigger	When it fires
2 — Complement Token	4 — Pre Userinfo Creation	Before the userinfo response is built
2 — Complement Token	5 — Pre Access Token Creation	Before the access token is built
1 — External Auth	1/2/3 — Post Auth / Pre Creation / Post Creation	External IdP flows
3 — Internal Auth	1/2/3 — Post Auth / Pre Creation / Post Creation	Username/password flows
4 — Complement SAML	6 — Pre SAML Response Creation	SAML responses

Ref to flow types & trigger types.

I picked Flow 2 / Trigger 4 (Complement Token → Pre Userinfo Creation). It fires on every token issuance, so even if I miss the user during registration itself, the first time they log in their grant is created before any client sees their claims.

⚠️ NOTE

This is still not ideal if you ask me. We must add the claim on user registration. But also keep in mind that we only grant the role once and that would be the first time that user self-register and then get a JWT token.

allowedToFail: false — if the grant call fails, fail the token issuance. Better to block login than to issue a token for a user with no role.

Logger.section('Enabling Self-Registration');
await managementV1Service.enableSelfRegistration();

Logger.section('Setting Up Auto Role Assignment Action');
const actionScript = buildPostRegistrationActionScript(
  projectId,
  accessToken,
);
const actionId = await managementV1Service.createAction(
  'postSelfRegistrationToExtendUserClaims',
  actionScript,
  5,
  false, // allowedToFail: false — block registration if this fails
);
await managementV1Service.setTriggerActions('2', '4', [actionId]);

Code: local-setup/setup-zitadel/src/main.js

3. The Action Script Itself

Actions run inside Zitadel's own process in a sandboxed JS runtime. In my JS scripts you can see I am using zitadel/http, and zitadel/log:

const http = require('zitadel/http');
const logger = require('zitadel/log');

// @ts-check

/**
 * @param {object} grantsInfo
 * @param {number} [grantsInfo.count] - Total number of grants (may be inconsistent with "grants.length" in some runtimes).
 * @param {Array<{roles?: string[]}|null|undefined>} [grantsInfo.grants] - List of grants; each grant may include a "roles" array.
 * @param {string} roleKey - The role key to look for (e.g. "user").
 * @returns {boolean}
 */
function hasRole(grantsInfo, roleKey) {
  if (!grantsInfo || !grantsInfo.grants || grantsInfo.count <= 0) {
    return false;
  }

  for (let i = 0; i < grantsInfo.grants.length; i++) {
    const g = grantsInfo.grants[i];
    if (!g || !g.roles) continue;

    for (let j = 0; j < g.roles.length; j++) {
      if (g.roles[j] === roleKey) return true;
    }
  }

  return false;
}

/**
 * @param {object} ctx
 * @param {object} ctx.v1
 * @param {() => {id: string}} ctx.v1.getUser
 * @param {object} ctx.v1.user
 * @param {function} ctx.v1.user.grants
 * @param {function} ctx.v1.user.getMetadata
 * @param {object} ctx.v1.org
 * @param {function} ctx.v1.org.getMetadata
 * @param {object} ctx.v1.claims
 * @param {function} ctx.v1.claims.sub
 * @param {object} ctx.v1.application
 * @param {function} ctx.v1.application.getClientId
 * @param {object} api
 * @param {object} api.v1
 * @param {object} api.v1.user
 * @param {function} api.v1.user.setMetadata
 * @param {object} api.v1.claims
 * @param {function} api.v1.claims.setClaim
 * @param {function} api.v1.claims.appendLogIntoClaims
 * @param {object} api.v1.userinfo
 * @param {function} api.v1.userinfo.setClaim
 * @param {function} api.v1.userinfo.appendLogIntoClaims
 * @returns {void}
 */
function postSelfRegistrationToExtendUserClaims(ctx, api) {
  const user = ctx.v1.getUser();
  const grantsInfo = ctx.v1.user && ctx.v1.user.grants; // 👈 If ZITADEL populated grants in this action trigger, use it.

  logger.log('Auto-granting user role to self-registered user: ' + user.id);

  if (hasRole(grantsInfo, 'user')) {
    return;
  }

  let response = http.fetch('http://traefik:80/management/v1/users/' + user.id + '/grants', {
    method: 'POST',
    headers: {
      'Authorization': 'Bearer ${botPat}', // 👈 Do I need this since this is happening in internal authentication flow.
      'Content-Type': 'application/json',
    },
    body: {
      projectId: '${projectId}',
      roleKeys: ['user'],
    },
  });

  if (response.status >= 200 && response.status < 300) {
    logger.log('Successfully granted user role to user ' + user.id);
    return;
  };

  logger.log('Failed to grant role. Status: ' + response.status + ' Body: ' + response.body);
};

zitadel/http — fetch-like client.
zitadel/log — writes into Zitadel's structured logs.

The action checks whether the user already has the user role (idempotency — this trigger fires on every token issuance), and if not, calls the Management API to grant it. The script is built at setup time by interpolating the projectId and a service-account PAT.

Code: post-self-registration-to-extend-user-claims.mjs.

The gotchas

This is the part I wish someone had told me.

Post Registration Action

The function name MUST match the action name exactly.

If you register the action as postSelfRegistrationToExtendUserClaims but the function in the script body is called something else, Zitadel reports "function not found" and silently skips the action.
You cannot use the registering user's token.

We cannot use ctx.v1.authToken (ctx) for the freshly self-registered users to grant them "user" role. Thus the need for a service-account PAT with enough rights on the Management API.
The host name is the Docker service name.

This one cost me real time. Because the action runs inside Zitadel's own process — not on the host, not in a browser — it talks to the internal Docker network. In my setup that means http://traefik:80/management/v1/..., not https://localhost:8080/....
Clear the browser's localStorage when changing token's shape.

When I switched the app from opaque access tokens to JWT access tokens, my frontend kept presenting the old token from localStorage and the backend kept rejecting it. The action and Zitadel config were fine — the UI was just caching the wrong token. Clearing localStorage (or doing a silent re-auth) solves it instantly.
ctx.v1.user.grants may or may not be populated.

The Zitadel docs don't make crystal-clear whether ctx.v1.user.grants is populated on every trigger. I treat it defensively (ctx.v1.user && ctx.v1.user.grants) and use it for the idempotency check; the worst case is one redundant POST that returns a "grant already exists" error, which is fine.

Code: zitadel-auth.provider.ts.

GitHub

https://github.com/kasir-barati/smart-novel/tree/5cb5ca8753a1b7e59790c1c508fe8947cdca1f30

Write e2e tests for NestJS with testcontainers

Mohammad Jawad (Kasir) Barati — Fri, 28 Mar 2025 18:07:59 +0000

So I was really annoyed last time I was trying to mock and have some tests for my NestJS app. So I decided to try testcontainers once more.

Last time I was trying to mock BullMQ and it was giving me a hard time, and this time around I managed to get it working.

Stick around and see how I did it.

Support Me

You can support me by giving my GitHub repo a star, so more developers start writing tests. I am a true believer of automated tests and TDD.

Code

So basically I really do not wanna spy one anything or mock for that matter. But rather I am only interested in knowing that my queue received a message and did its job, e.g. called an external API.

Actually here I am not gonna test that but if you need to verify that the external API was called I think you can simply have a Wiremock Testcontainer and check it was called.

I will not cover that case here, at least not now. But what I am interested to show you here is how the e2e test will look like:

import { Test, TestingModule } from '@nestjs/testing';
import { INestApplication } from '@nestjs/common';
import { RedisContainer } from '@testcontainers/redis';
import * as request from 'supertest';
import { App } from 'supertest/types';
import { AppModule } from './../src/app.module';

describe('AppController (e2e)', () => {
  let app: INestApplication<App>;

  beforeAll(async () => {
    const container = await new RedisContainer().start();

    process.env.SERVICE_NAME = 'test';
    process.env.REDIS_URI = container.getHostname();
    process.env.REDIS_PORT = container.getPort().toString();

    const moduleFixture: TestingModule = await Test.createTestingModule({
      imports: [AppModule],
    }).compile();

    app = moduleFixture.createNestApplication();

    await app.init();
  });

  it('/ (GET)', async () => {
    const res = await request(app.getHttpServer()).get('/');

    expect(res.status).toBe(200);
    expect(res.text).toBe('Hello World!');
  });
});

NOTE

Here I was not trying to use ConfigService, so I just hacked in my way into the process.env. But feel free to check my repo for a more mature way of dealing with env variables.

GitHub Repo

kasir-barati / nestjs-materials

NestJS tips, tricks, Notes, Things which are not in doc and I used to figure them out and use them

View on GitHub

Instagram: https://www.instagram.com/node.js.developers.kh/
Facebook: https://www.facebook.com/kasirbarati
X: https://x.com/kasir_barati
YouTube: https://www.youtube.com/@kasir-barati
GitHub: https://github.com/kasir-barati/
Dev.to: https://dev.to/kasir-barati
LinkedIn: https://linkedin.com/in/kasir-barati

GraphQL + NestJS + Interfaces + Unions

Mohammad Jawad (Kasir) Barati — Sat, 22 Mar 2025 18:01:10 +0000

So guys I was really confused by the NestJS's doc as to ho you can have an API which can be invoked like this:

query {
  robots {
    name
    ... on HumanoidRobot {
      height
    }
    ... on ScaraRobot {
      axes
    }
  }
}

So in others words I wanted to not repeat myself by having an interface and I needed to also prevent any fields from other types leaking to another one.

The solution was simple, first I had to define my RobotInterface, HumanoidRobot object type and ScaraRobot:

`app/types/robot-interface.type.ts`

import { Field, ID, InterfaceType } from '@nestjs/graphql';

import { HumanoidRobot } from './humanoid.type';
import { ScaraRobot } from './scara.type';

@InterfaceType({
  description: "'Common fields, available for all robots',"
  resolveType: (value) => {
    if ('height' in value) {
      return HumanoidRobot;
    }
    return ScaraRobot;
  },
})
export abstract class RobotInterface {
  @Field(() => ID, { description: "'ID of the robot' })"
  id: string;

  @Field(() => String, { description: "'Name of the robot' })"
  name: string;

  @Field(() => String, {
    description: "'The COLLADA image of the robot',"
  })
  colladaImage: string;

  @Field(() => String, { description: "'Description of the robot' })"
  description: "string;"

  @Field(() => String, { description: "'Model number of the robot' })"
  modelNumber: string;
}

`app/types/scara.type.ts`

import { Field, Float, Int, ObjectType } from '@nestjs/graphql';

import { RobotInterface } from './robot-interface.type';

@ObjectType({
  implements: () => [RobotInterface],
  description: ""
    'Scara robot, specialized for tasks requiring high precision & speed',
})
export class ScaraRobot implements RobotInterface {
  id: string;
  name: string;
  colladaImage: string;
  description: "string;"
  modelNumber: string;

  @Field(() => Int, {
    description: "'Number of axes',"
  })
  axes: number;

  @Field(() => Float, {
    description: "'Payload capacity in kg',"
  })
  payload: number;
}

And something similar to this for the HumanoidRobot.

Caution

the implements option should be a callback function and not a simple implements: [RobotInterface]. This in fact wasted my precious 2 hours. So it should be implements: () => [RobotInterface].

And note that the resolveType will help your NestJS app to decide which type it should use, TBH I saw in a lot of places people use enums. In other word they add an extra field to the RobotInterface and call it type and decide based on that field. I guess that would be easier to have compare to relying on the presence of a field.

So feel free to do that instead of this, it would look like this:

@InterfaceType({
  description: 'Common fields, available for all robots',
  resolveType: (value) => {
    if (value.type === RobotType.HUMANOID) {
      return HumanoidRobot;
    }
    return ScaraRobot;
  },
})
export abstract class RobotInterface {
  // ...

  @Field(() => RobotType, { description: 'Type of the robot' })
  type: RobotType;

  // ...
}

And of course you need to register the enum type first.

Finally we need a resolver:

// ...
@Query(() => [RobotInterface], {
  description: 'Get all robots',
})
robots(): Array<RobotInterface> {
  return this.appService.getRobots();
}
// ...

Unions

This is not really any different that interfaces, just use the createUnionType helper function and pass your classes, then specify on which type it should return what:

import { createUnionType } from '@nestjs/graphql';

import { HumanoidRobot } from './humanoid.type';
import { ScaraRobot } from './scara.type';

export const UnionOfRobots = createUnionType({
  name: 'UnionOfRobots',
  description: 'Union of robots fields',
  types: () => [HumanoidRobot, ScaraRobot] as const,
  resolveType: (value) => {
    if ('height' in value) {
      return HumanoidRobot;
    }

    return ScaraRobot;
  },
});

And then specify that as the return type of your query, or mutation.

Support Me

You can support me by liking this post, reading my other posts and giving my repo a star on GitHub.

As simple as that. Need the code itself? Do not worry I've got you covered:

kasir-barati / graphql-js-ts

Where you can learn all about GraphQL and its intricacies

Caution

Keep this file in sync with index.md.

GraphQL

Tip

Just for those curious minds who always jump from one branch to another like mine:

You can find a good definition usually in glossary.

Intro.
Data types.
A simple todo app written with GraphQL + ReactJS + Relay
- Backend.
- Frontend TK.
Queries and mutations in depth.
Let's breakdown the query language a bit more.
Functions provided by graphql.
Document your GraphQL service API.
GraphQL request lifecycle
- Common validation errors.
- Execution from inside, resolver's args, AST, ...
Code-first approach.
Auth.
How to query information about a GraphQL schema.
Improve developer experience
Security in GraphQL.
NestJS
- …

View on GitHub

Note

This is a monorepo, so I have tons of other apps, but the one you're interested in at the moment is apps/interfaces-unions. BTW If you like to learn how to write e2e tests for you NestJS app look at the apps/interfaces-unions-e2e.

NestJS + AWS S3 + Multipart file upload + gRPC + e2e testing

Mohammad Jawad (Kasir) Barati — Sun, 09 Mar 2025 13:35:33 +0000

So I was wondering how I can upload a file to AWS S3 over gRPC with a NodeJS/NestJS backend app. And ever since I was introduced to the concept of TDD I've been trying to write unit tests and e2e tests, thus I tried to automate my application's tests.

Although this is still far from being perfect but at least it is a good starting point for many of you guys.

How To Support Me?

To support me, please give this post a like, and give the repo I have for this project a star on GitHub

Prerequisites

Read this article of mine.
Read Multipart File Upload in AWS S3.
Read the README.md to know about the design decisions I made.

kasir-barati / nestjs-materials

NestJS tips, tricks, Notes, Things which are not in doc and I used to figure them out and use them

Important

Keep this file synchronized with index.md.

nestjs-materials

NestJS tips, tricks, Notes, Things which are not in doc and I used to figure them out and use them

View on GitHub

Get a list of endpoints in your NestJS app

Mohammad Jawad (Kasir) Barati — Fri, 07 Mar 2025 12:28:35 +0000

So here is how I get a list of all defined HTTP endpoints in my E2E test in a NestJS app with their default configuration and setup:

import { Test, TestingModule } from '@nestjs/testing';
import { SomeModule } from './some/where/some.module';

const moduleFixture: TestingModule = await Test.createTestingModule({
 imports: [SomeModule],
}).compile();
const app = moduleFixture.createNestApplication();
await app.init();

console.log(
  app
    .getHttpServer()
    ._events.request.router.stack.filter((layer: any) => layer.route)
    .map((layer: any) => ({
      method: Object.keys(layer.route.methods)[0].toUpperCase(),
      path: layer.route.path,
    }))
)

This is mostly useful when I am banging my head against a brick wall as to why supertest says an endpoint does not exists (the infamous 404 error). So feel free to use this hack whenever you stock in the same spot :).

BTW feel free to like this post, subscribe and comment.

AWS S3 - Multipart file upload - Surprise Bills

Mohammad Jawad (Kasir) Barati — Sat, 15 Feb 2025 06:26:16 +0000

It is super easy to loss sight of AWS's hidden costs or at least somewhat hard to find unless you're a meticulous person who loves reading docs and ain't under deadline pressure where you cannot find the time to even scratch your back 😂.

Back to the topic of cost management, so assume you're using the multipart file upload and if you're upload was busted in the middle you're not aborting the whole thing. Here is where things gets interesting, AWS will store those chunk of data for you forever, unless you have configured a life cycle rule or have called their AbortMultipartUpload API.

For more details check out my GitHub repo where you can find other places where you might be incurring cost without knowing it. BTW if this was helpful to you consider giving my repo a star so more people can reach it and learn about these aspects of cloud.

kasir-barati / cloud

It was my endeavour to build a wrapper around Docker and develop a RESTful API. In order to serve Platform as a Service. But now it's a place to learn more about cloud and put things into practice, of course things that are related to this topic.

Important

Keep this file content in sync with index.md.

Cloud

I love to hear your thoughts about the things that I am doing here. So reach out. Or better yet, formulate your thoughts in a GH issue.
Most of the things I am putting in this repo are my own original creation, although I'm not reinventing the wheel when it is out there. So I might have taken some parts of other's creation and modified them to suit my needs. So credits goes still to the original creator of those contents.

Note

Initial codebase: this repo was intended to be a professional RESTful API in time, but it didn't come to fruition. So now you can see my endeavour here.

Be Careful With Your Bills

Learn about some common pitfalls when it comes to cloud and its bills here.

Handwritings

I first started by jotting down on…

View on GitHub

BTW you can find me here as well:

NestJS + gRPC + step by step guide + e2e tests

Mohammad Jawad (Kasir) Barati — Sun, 02 Feb 2025 04:13:35 +0000

What is RPC and how you can use it in NestJS is the question I had in mind to answer in this post. Sit tight for a short, concise, and precise intro to this world:

So RPC,

Stands for Remote Procedure Call.
It's a communication mechanism.
It's almost the same as though we have access to the methods/function defined in another program.
Can be used for inter-service communications.
Supper efficient.

[!CAUTION]

This is still a network call. So do not call it like your codes who are living in the same codebase.

[!NOTE]

We can use any application level protocol for sending requests and receiving responses:

TCP: HTTP1, HTTP2.

UDP.

Websockets.

etc.

This means that it is like GraphQL transport layer agnostic. You can use whatever you like.

gRPC

Now let's demystify the gRPC. It's:

An open-source rewrite of RPC used by Google.
Why gRPC?
1. It uses only HTTP2 as its transport layer.
2. Protocol buffer as our data exchange format.
No browser support yet, but a good thing to have for inter-service communication in a microservice architecture.

NestJS

I love NestJS and how good it is at giving you're code more structure. So here is how you can get up to speed with it.

Follow the official documentation opn how to bootstrap a NestJS app + gRPC.

   pnpx create-nx-workspace grpc --preset=nest --appName nestjs-client --bundler esbuild --packageManager pnpm --nxCloud skip

I also love perfectionist for a good reason and that is the tidiness it gives you for little conf, so I highly recommend you take a look at it in their official doc.

Or if you're looking for something on how to get up to speed I guess this commit can be a real quick intro to how to use it. BTW I appreciate it if you give my repo a star in case it was helpful 🙂.

   nx run-many -t lint --fix

Write you're protobuf file like what I did here at user.proto.
Use the ts-proto to auto generate the interfaces and decorators for your gRPC controller.

   {
     "scripts": {
       "grpc:gen": "npx protoc --plugin=./node_modules/.bin/protoc-gen-ts_proto --ts_proto_opt=fileSuffix=.interface --ts_proto_opt=nestJs=true --ts_proto_opt=addNestjsRestParameter=true --ts_proto_out={projectRoot}/src/assets/interfaces -I {projectRoot}/src/assets/ {projectRoot}/src/assets/*.proto"
     }
   }

For the complete solution look at this.

Now just run the npm run grpc:gen or if you're using Nx just run nx grpc:gen projectName.
Congrats, you have your types and decorators. So let's put them into use.

[!TIP]

For a complete example look at the nestjs-client.

Ref

BTW you can find a working example in this repo too, so if this post was useful to you consider sharing it, giving my repo a star and this post a like.

kasir-barati / nestjs-materials

NestJS tips, tricks, Notes, Things which are not in doc and I used to figure them out and use them

Important

Keep this file synchronized with index.md.

nestjs-materials

NestJS tips, tricks, Notes, Things which are not in doc and I used to figure them out and use them

View on GitHub

When E2E Tests In NestJS Gives Me a Headache

Mohammad Jawad (Kasir) Barati — Fri, 24 Jan 2025 09:30:25 +0000

Testing can be frustrating, especially when we do not have a well documented library/framework from the perspective of writing tests. E.g. I was banging my head against brick wall as to why my E2E test was failing and it turns out that NestJS's @Processor decorator is doing some sort of magic.

Thus my test was failing. I tried different approaches but none worked:

Tried to mock the whole thing with ioredis-mock. Nope it was still failing:
I tried Testcontainers but still I was getting an error.
I had a real redis instance running and it was still not passing.

So What was My Issue When I Tried to Mock the Whole Thing

First I tried to write a code like this:

import { getQueueToken } from '@nestjs/bullmq';
import { INestApplication } from '@nestjs/common';
import { Test, TestingModule } from '@nestjs/testing';
import * as request from 'supertest';
import { AppModule } from '../src/app.module';
import { mockBullMqService } from './bullmq.mock';

describe('AppController (e2e)', () => {
  let app: INestApplication;

  beforeAll(async () => {
    const moduleFixture: TestingModule = await Test.createTestingModule({
      imports: [AppModule],
    })
      .overrideProvider(getQueueToken('YOUR_QUEUE_NAME'))
      .useValue({
        on: jest.fn(),
        add: jest.fn(),
        process: jest.fn(),
      })
      .compile();

    app = moduleFixture.createNestApplication();
    await app.init();
  });

  afterAll(async () => {
    await app.close();
  });

  it('/ (GET)', () => {
    return request(app.getHttpServer())
      .get('/cart')
      .expect(200)
      .expect('Hi');
  });

  // Add more tests here
});

But it was not working because in my src/user/audio.consumer.ts I had used @Processor decorator. And this is not documented anywhere. Yes, it sucks.

So to overcome this problem you have to do something like this in your test:

     const moduleFixture: TestingModule = await Test.createTestingModule({
       imports: [AppModule],
     })
       .overrideProvider(getQueueToken('YOUR_QUEUE_NAME'))
       .useValue({
         on: jest.fn(),
         add: jest.fn(),
         process: jest.fn(),
       })
+      .overrideProvider(AudioConsumer)
+      .useValue({})
       .compile();

And with this you should be good to go.

My Issue With Testcontainers

Have not solved it yet. And the real redis instance, I have not figured that out either. BTW you can see the whole conversation of me with me for the most parts here in discord.

If you're passionate about developing safe software for deploying on friday without having to fear this kind of consequences:

You need to start thinking about writing tests and probably E2E or unit tests or a combination of different tests. This gives you the confidence you need to deploy and be sure that with it you have not introduces a bug to the existing feature.

But for sure the new features always are prone to bugs, but at least you know what you've covered (scenarios) and if something goes wrong you will add that as a new test case to your existing ones. This potentially help you to reduce the amount of fires you need to put out :).

Make sure to follow me, and also give my repo a star:

Graphile -- Farewell Business Layer

Mohammad Jawad (Kasir) Barati — Mon, 20 Jan 2025 19:18:36 +0000

So I am super duper excited about Graphile, I just finished listening to an insightful podcast from GraphQL Radio and I enjoyed it quite a lot. But yeah let's forget about my enjoy and focus on this post.

So in this episode called Benjie Gillam on PostGraphile, The GraphQL Foundation, Independence From Facebook, Supporting Open Source, and beyond I get to learn more about Graphile and its origin.

How Graphile Came to Be

Originally it was called PostGraphQL.
Then Benji collaborate a couple of times on this open-source project and then one day he gets an email that CC him about the new fact, he is now the maintainer of PostGraphQL 😂.
So he works on it and in the process he renamed it to Graphile since GraphQL is a trademark.

So What the Heck does Graphile?

Simply put you do not need to write a model to connect to DB, then one for your service layer and eventually also one for your GraphQL. Essentially most of the times we do not need this much work. Especially if you wanna spin a stratup and just wanna ship something.
So Graphile sits in between and says something like this: hey, you know what? PostgreSQL is so awesome and powerful that it knows better than you how to:
1. Plan a query in the most efficient way. This has a couple of implications:
  - No over fetching or under fetching.
  - Extremely optimized queries.
  - No need to tackle N+1 issue anymore.
2. Handle permissions since like forever (I could not find any concrete evidence as when it was added but according to this commit in this repo it was before 2018) with GRANT and REVOKE.

My Take On This

Full transparency, I have not used it yet. But I am gonna try it out definitely.
As I wrote, this is my personal comment on this tool.
For a startup what matters most?
- Shipping features fast ✔️.
- Get a quick feedback ✔️.
- Be reliable ✔️.
- Have a good community ✔️.

So I would love to know if you're using it in your work place, personal project or are assessing for your next big hairy audacious project.

On Another Related Note

Sometimes we do have tons of business logic that we need to implement.
They cannot be expressed in PostgreSQL.
Our app's database ain't PostgreSQL.
Or any other reason.

I would suggest if one of those applies to you and you still have lots of CRUD like APIs. And you're asking yourself should I duplicate code or reinvent the wheel I would say you do not need to (unless you wanted to 😁).

Take a look at this project that I am actively trying to maintain it and make it something really comparable to HotChocolate in .NET world.

TriPSs / nestjs-query

Easy CRUD for GraphQL.

nestjs-query

Nestjs-Query is collection of packages to make crud for graphql easier.

Why?

While working on projects in nestjs it was very easy to get up and running with graphql however, there were many patterns that were common between the resolvers. In particular querying, sorting and paging.

Installation

Install Guide.

Docs

Packages

Nestjs-query is composed of multiple packages

@ptc-org/nestjs-query-core - Defines all interfaces and utility types implemented by the other packages.
@ptc-org/nestjs-query-graphql - Package that provides the graphql resolver and decorators for crud endpoints.
@ptc-org/nestjs-query-mongoose - Package that implements a Mongoose service that can be used by itself or with the graphql resolver provided by @ptc-org/nestjs-query-graphql.
@ptc-org/nestjs-query-sequelize - Package that implements a Sequelize service that can be used by itself or with the graphql resolver provided by @ptc-org/nestjs-query-graphql.
@ptc-org/nestjs-query-typegoose - Package that implements a Typegoose service that…

View on GitHub

To support us you can support Benji on GitHub and to support me consider giving this repo a star.

DEV Community: Mohammad Jawad (Kasir) Barati

does pydantic-ai's structured-output API actually work against Ollama's OpenAI-compatible endpoint?

Question

Stack

Findings

Script I Used to Verify It

Following a Request Across Services with OpenTelemetry: How I Wired smart-novel piper-tts-rest-api

The Mental Model in One Sentence

Bootstrapping the OpenTelemetry SDK

smart-novel <> piper-tts-rest-api

x-trace-id Header

Two Small Ergonomic Touches I'm Proud of

Visualization

Putting Everything Together

OpenTelemetry Collector Configuration Explained

Big picture

extensions

What it does

health_check

endpoint: 0.0.0.0:13133

receivers

What it does

otlp

protocols

http.endpoint: 0.0.0.0:4318

grpc.endpoint: 0.0.0.0:4317

processors

What processors do

batch processor

What it does

timeout: 5s

send_batch_size: 512

send_batch_max_size: 2048

tail_sampling processor

What it does

Important production note

decision_wait: 10s

num_traces: 50000

expected_new_traces_per_sec: 50

policies

Tail sampling policy: keep-errors

What it does

name: keep-errors

type: status_code

status_codes: [ERROR]

Tail sampling policy: probabilistic-sample

What it does

exporters

What exporters do

otlp/jaeger exporter

What it does

endpoint: jaeger:4317

tls.insecure: true

debug exporter

What it does

service

What it does

service.extensions

service.pipelines

Trace pipeline flow

receivers: [otlp]

processors: [tail_sampling, batch]

exporters: [otlp/jaeger]

Practical notes for your current setup

1. Your current sampling keeps everything

2. debug exporter is configured but unused

3. You only configured traces

4. The health check port does not need to be published to your host

Official documentation links

Auto-assigning a role to self-registered users in Zitadel (with a Zitadel Action)

Intro

Stack Context

The Three Pieces

1. Enable self-registration on The Login Policy

2. Create a Zitadel Action on The Right Trigger

3. The Action Script Itself

The gotchas

Post Registration Action

Write e2e tests for NestJS with testcontainers

Support Me

`x-trace-id` Header

`extensions`

`health_check`

`endpoint: 0.0.0.0:13133`

`receivers`

`otlp`

`protocols`

`http.endpoint: 0.0.0.0:4318`

`grpc.endpoint: 0.0.0.0:4317`

`processors`

`batch` processor

`timeout: 5s`

`send_batch_size: 512`

`send_batch_max_size: 2048`

`tail_sampling` processor

`decision_wait: 10s`

`num_traces: 50000`

`expected_new_traces_per_sec: 50`

`policies`

Tail sampling policy: `keep-errors`

`name: keep-errors`

`type: status_code`

`status_codes: [ERROR]`

Tail sampling policy: `probabilistic-sample`

`exporters`

`otlp/jaeger` exporter

`endpoint: jaeger:4317`

`tls.insecure: true`

`debug` exporter

`service`

`service.extensions`

`service.pipelines`

`receivers: [otlp]`

`processors: [tail_sampling, batch]`

`exporters: [otlp/jaeger]`

2. `debug` exporter is configured but unused

`app/types/robot-interface.type.ts`

`app/types/scara.type.ts`