Iggy¶

The imbi.common.iggy module publishes analytics rows to Apache Iggy. The Iggy ClickHouse sink reads the streams and writes the rows to ClickHouse. No Imbi service inserts rows into ClickHouse directly. See ADR 0019.

The module needs the databases extra, which installs apache-iggy. That package can need a Rust toolchain to build. See Installation.

Streams and topics¶

TOPICS maps each stream name to its topics. The stream name is also the name of the ClickHouse table that the stream goes to. A topic is the feature that makes the rows, for example github or maintenance.

TOPICS is the only list of streams and topics. imbi-api makes the sink configuration from it and serves that configuration to the Iggy connectors runtime over HTTP. The runtime reads the configuration one time, at startup. When you add a topic, restart the connectors runtime.

For the full list of streams and the tables they go to, see ClickHouse Client.

Lifecycle¶

initialize() connects to the server and creates every stream and topic in TOPICS that does not exist. The sink stops at startup if a topic in its configuration does not exist, so a service must create the topics before the sink starts. initialize() returns False when it cannot connect, and raises PublishError when it connects but cannot create a topic.

The client is a singleton per process. Services call initialize() from a lifespan hook and aclose() on shutdown:

import contextlib

from imbi.common import iggy


@contextlib.asynccontextmanager
async def iggy_hook():
    if await iggy.initialize() is False:
        raise RuntimeError('Iggy initialization failed')
    async with contextlib.aclosing(iggy):
        yield

When an error shows that the connection is lost, the client discards itself. The next call connects again. A service therefore recovers after an Iggy restart without a process restart.

Publish rows¶

publish() sends one message per Pydantic model. All models in one call must be of the same type. The message body is the model dumped by alias, so the keys must be the column names of the table.

from imbi.common import iggy

await iggy.publish('commits', 'github', [commit_record])

Use columns to send only some keys, in a given order. Use headers to set Iggy user headers on every message.

publish_rows() sends rows that are already dictionaries, for example a row read back from ClickHouse and sent again with a higher version.

Both functions raise ValueError for an empty list, and PublishError when the send fails. A publish creates the topic first if it does not exist.

Metrics¶

The client counts messages on two Prometheus counters, labelled by stream and topic:

topic_status() returns a TopicStatus for each topic: the message count, the size, and the members of the sink consumer group. A group with more members than members_owning has a member that reads nothing. That is how a stuck sink shows. stored_bytes() returns the total size of all the streams.

Configuration¶

The client reads settings.Iggy from the environment only. A config.toml has no [iggy] section.

The user name and password must not contain @ or :. The SDK cannot parse these characters.

API Reference¶

TOPICS: dict[str, tuple[str, ...]] = {
    'agent_task_events': ('agent_tasks',),
    'agent_usage': ('agent_tasks',),
    'commit_drift': ('drift',),
    'commits': ('github', 'maintenance'),
    'document_read_events': ('documents',),
    'document_read_sessions': ('documents',),
    'document_versions': ('documents',),
    'email_audit': ('email',),
    'events': (
        'comments',
        'documents',
        'gateway',
        'lifecycle',
        'projects',
    ),
    'maintenance_log': ('maintenance',),
    'operations_log': (
        'api',
        'configuration',
        'deployments',
        'maintenance',
    ),
    'pull_request_review_requests': ('github',),
    'pull_request_reviews': ('github',),
    'pull_requests': ('github',),
    'release_component_batches': ('sbom',),
    'release_components': ('sbom',),
    'scheduler_runs': ('scheduler',),
    'score_history': ('scoring',),
    'secret_audit_events': ('secrets',),
    'tags': ('github',),
    'workflow_run_events': ('workflows',),
    'workflow_runs': ('workflows',),
}

initialize() -> bool

68
69
70

async def initialize() -> bool:
    """Create a new client and test the connection."""
    return await client.Iggy.get_instance().initialize()

aclose() -> None

73
74
75

async def aclose() -> None:
    """Discard the Iggy client."""
    await client.Iggy.get_instance().aclose()

publish(
    stream: str,
    topic: str,
    models: list[BaseModel],
    *,
    columns: list[str] | None = None,
    headers: dict[str, str] | None = None,
) -> None

 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121

async def publish(
    stream: str,
    topic: str,
    models: list[pydantic.BaseModel],
    *,
    columns: list[str] | None = None,
    headers: dict[str, str] | None = None,
) -> None:
    """Publish one message per model to a stream's topic.

    Args:
        stream: The name of the stream to publish to
        topic: The name of the topic within that stream
        models: List of Pydantic models to publish (all the same type)
        columns: Restrict each payload to these keys, in this order
        headers: Iggy ``user_headers`` set on every message

    Raises:
        ValueError: If models is empty or the models are mixed types
        PublishError: If the send fails
    """
    await client.Iggy.get_instance().publish(
        stream, topic, models, columns=columns, headers=headers
    )

publish_rows(
    stream: str,
    topic: str,
    rows: list[dict[str, Any]],
    *,
    headers: dict[str, str] | None = None,
) -> None

124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150

async def publish_rows(
    stream: str,
    topic: str,
    rows: list[dict[str, typing.Any]],
    *,
    headers: dict[str, str] | None = None,
) -> None:
    """Publish one message per already-rendered row to a stream's topic.

    For producers whose row is a column-to-value mapping rather than a
    model, such as a row read back from ClickHouse and re-published with
    a bumped version. Each row is sent as-is, so its keys must be the
    table's column names.

    Args:
        stream: The name of the stream to publish to
        topic: The name of the topic within that stream
        rows: The rows to publish, one message each
        headers: Iggy ``user_headers`` set on every message

    Raises:
        ValueError: If rows is empty
        PublishError: If the send fails
    """
    await client.Iggy.get_instance().publish_rows(
        stream, topic, rows, headers=headers
    )

ensure_topic(stream: str, topic: str) -> None

78
79
80

async def ensure_topic(stream: str, topic: str) -> None:
    """Create the stream and topic when they do not exist yet."""
    await client.Iggy.get_instance().ensure_topic(stream, topic)

ping() -> None

83
84
85

async def ping() -> None:
    """Round-trip a ping to the server to prove the connection works."""
    await client.Iggy.get_instance().ping()

topic_status() -> list[TopicStatus]

93
94
95

async def topic_status() -> list[TopicStatus]:
    """The state of every topic in `TOPICS` and its sink group."""
    return await client.Iggy.get_instance().topic_status()

stored_bytes() -> int

88
89
90

async def stored_bytes() -> int:
    """Total bytes Iggy holds across every stream in `TOPICS`."""
    return await client.Iggy.get_instance().stored_bytes()