Observability

Every call in an iii system passes through the engine. Because every worker communicates through the engine, it can trace work as it moves from one worker to the next, giving you observability across the whole system without each worker instrumenting itself. The automatically supplied iii-observability worker turns that information into OpenTelemetry traces, metrics, and logs. Do not declare it in either config.yaml or worker-compose.yaml.

This page is a quick tour. For the complete configuration and the full list of query functions, see the iii-observability worker docs.

Logging

Logger from the observability SDK sends each log line of your worker to the engine, so the line carries the trace and span of the call that logged it. Each level takes a message and optional structured data. The examples on this page use a link store: links::process appends one link to a store file, and retries the write.

Node / TypeScript
Python
Rust
Go

import { appendFile } from "node:fs/promises";
import { registerWorker } from "iii-sdk";
import { Logger } from "@iii-dev/helpers/observability";

const url = process.env.III_URL;
if (!url) throw new Error("III_URL must be set");
const worker = registerWorker(url);

// without OpenTelemetry, Logger falls back to console.*
const logger = new Logger();

type Link = { slug: string; url: string };

worker.registerFunction("links::process", async (link: Link) => {
  logger.debug("cache lookup", { key: `link:${link.slug}` });

  for (let attempt = 1; attempt <= 3; attempt += 1) {
    try {
      await appendFile("links.jsonl", `${JSON.stringify(link)}\n`);
      logger.info("processing link", { slug: link.slug, attempt });
      return { slug: link.slug };
    } catch (err) {
      if (attempt === 3) {
        logger.error("failed to persist", { slug: link.slug, err: String(err) });
        throw err;
      }
      logger.warn("retrying write", { attempt });
    }
  }
});

import json
import os

from iii import InitOptions, register_worker
from iii_helpers.observability import Logger

worker = register_worker(os.environ["III_URL"], InitOptions(worker_name="links"))
# without OpenTelemetry, Logger falls back to logging, logger iii.logger
logger = Logger()


def process_link(link: dict) -> dict:
    logger.debug("cache lookup", {"key": f"link:{link['slug']}"})

    for attempt in range(1, 4):
        try:
            with open("links.jsonl", "a", encoding="utf-8") as store:
                store.write(json.dumps(link) + "\n")
            logger.info("processing link", {"slug": link["slug"], "attempt": attempt})
            return {"slug": link["slug"]}
        except OSError as err:
            if attempt == 3:
                logger.error("failed to persist", {"slug": link["slug"], "err": str(err)})
                raise
            logger.warn("retrying write", {"attempt": attempt})
    return {"slug": link["slug"]}


worker.register_function("links::process", process_link)

use iii_helpers::observability::Logger;
use iii_sdk::{Error, InitOptions, RegisterFunction, register_worker};
use schemars::JsonSchema; // schemars 0.8, the version iii-sdk uses
use serde::{Deserialize, Serialize};
use serde_json::json;
use std::io::Write;

#[derive(Debug, Serialize, Deserialize, JsonSchema)]
struct Link {
    slug: String,
    url: String,
}

#[derive(Debug, Serialize, Deserialize, JsonSchema)]
struct Processed {
    slug: String,
}

let url = std::env::var("III_URL").expect("III_URL must be set");
let worker = register_worker(&url, InitOptions::default());

worker.register_function(
    "links::process",
    RegisterFunction::new(|link: Link| {
        // without OpenTelemetry, Logger falls back to the `tracing` crate
        let logger = Logger::new();
        let slug = link.slug.clone();
        logger.debug("cache lookup", Some(json!({ "key": format!("link:{slug}") })));

        for attempt in 1..=3 {
            let appended = std::fs::OpenOptions::new()
                .create(true)
                .append(true)
                .open("links.jsonl")
                .and_then(|mut store| writeln!(store, "{}", json!(&link)));

            match appended {
                Ok(()) => {
                    logger.info(
                        "processing link",
                        Some(json!({ "slug": slug, "attempt": attempt })),
                    );
                    return Ok(Processed { slug });
                }
                Err(err) if attempt == 3 => {
                    logger.error(
                        "failed to persist",
                        Some(json!({ "slug": slug, "err": err.to_string() })),
                    );
                    return Err(Error::from(err.to_string()));
                }
                Err(_) => logger.warn("retrying write", Some(json!({ "attempt": attempt }))),
            }
        }
        Ok(Processed { slug })
    }),
);

package main

import (
	"context"
	"encoding/json"
	"log"
	"os"

	iii "github.com/iii-hq/iii/sdk/packages/go/iii"
	"github.com/iii-hq/iii/sdk/packages/go/iii/observability"
)

type Link struct {
	Slug string `json:"slug"`
	URL  string `json:"url"`
}

// without OpenTelemetry, Logger falls back to the standard log package
var logger observability.Logger

func processLink(ctx context.Context, data json.RawMessage) (any, error) {
	var link Link
	if err := json.Unmarshal(data, &link); err != nil {
		return nil, err
	}

	logger.Debug(ctx, "cache lookup", map[string]any{"key": "link:" + link.Slug})

	for attempt := 1; attempt <= 3; attempt++ {
		line, err := json.Marshal(link)
		if err == nil {
			var store *os.File
			store, err = os.OpenFile("links.jsonl", os.O_APPEND|os.O_CREATE|os.O_WRONLY, 0o644)
			if err == nil {
				_, err = store.Write(append(line, '\n'))
				_ = store.Close()
			}
		}
		if err == nil {
			logger.Info(ctx, "processing link", map[string]any{"slug": link.Slug, "attempt": attempt})
			return map[string]any{"slug": link.Slug}, nil
		}
		if attempt == 3 {
			logger.Error(ctx, "failed to persist", map[string]any{"slug": link.Slug, "err": err.Error()})
			return nil, err
		}
		logger.Warn(ctx, "retrying write", map[string]any{"attempt": attempt})
	}
	return map[string]any{"slug": link.Slug}, nil
}

func main() {
	client := iii.RegisterWorker(os.Getenv("III_URL"))

	if _, err := client.RegisterFunction("links::process", processLink); err != nil {
		log.Fatal(err)
	}

	if err := client.Connect(context.Background()); err != nil {
		log.Fatal(err)
	}
	select {}
}

Where the lines go

The engine prints each line in its own output, labeled with the worker’s OpenTelemetry service name (set it with OTEL_SERVICE_NAME), unless the observability worker sets logs_console_output: false. It also stores each line: iii call engine::logs::list returns it, and the console shows it next to the trace it was logged in. Under iii compose --up, the engine’s output goes to engine.log in the compose state directory, .iii/compose/<namespace>/ by default (see Compose logs). Before the worker’s OpenTelemetry export starts, or when it is off (OTEL_ENABLED=false), Logger falls back to the language’s own logger, and the line stays in the worker’s process. You can also emit one log line and find it in the stored logs, without a worker:

# emit a log line
iii call engine::log::info --json '{"message":"processing link"}'

# find it in the stored logs
iii call engine::logs::list | grep -C 10 "processing link"

Logger does not write to your worker’s own output. To print a line in the worker’s terminal, use the print function of the language.

Traces

A trace captures the spans of a call as it moves across workers, the headline view for debugging a request end to end. Traces come from worker function invocations: a worker records a span each time it handles a call. Built-in engine::* functions are not traced by default (set III_OTEL_TRACE_BUILTINS=true to include them), so call a worker function to produce a trace. The Logging examples register links::process, a worker function:

# call the worker function to produce a trace
iii call links::process slug=iii-intro url=https://iii.dev

# count the recorded traces
iii call engine::traces::list | jq '.total'

# print the span tree of the most recent trace
iii call engine::traces::tree trace_id=$(iii call engine::traces::list | jq -r '.traces[-1].trace_id')

Traces export on a short delay. If engine::traces::list reports 0, wait a few seconds and call links::process again.

To run a function when a log record or a span is recorded, observe observability::on-log or observability::on-span: see Observers.

Health and stored data

Health

# overall health status
iii call engine::health::check | jq -r .status

Clear stored data

# clear stored logs
iii call engine::logs::clear

# clear stored traces
iii call engine::traces::clear

Logs populate quickly. engine::logs::clear followed immediately by engine::logs::list can still return new logs generated in between the two calls.

Getting Started

Using iii

Understanding iii

Creating Workers

Upgrading

Observability

Logging

Where the lines go

Traces

Health and stored data

Health

Clear stored data

​Logging

​Where the lines go

​Traces

​Health and stored data

​Health

​Clear stored data

Logging

Where the lines go

Traces

Health and stored data

Health

Clear stored data