Skip to content

Commit fba2bb6

Browse files
johnsimonsjohnsimons
committed
Document the ingestion pipeline and its telemetry
docs/ingestion-pipeline.md covers the shape of the pipeline, how back pressure reaches the broker, what a shutdown and a hard cancellation each do to messages in flight, the three settings per instance with their ranges and what raising each one is actually for, and a table of which storage takes concurrent batches with the reason in each case. It also states the ordering rule that concurrency does not excuse, which is that the end state has to be settled by the times the events happened rather than by which batch commits last. docs/telemetry.md gains the error instance's instruments and loses the claim that PrintMetrics is how you observe error ingestion. The two instances now publish the same shape, so the sections are laid out to be read side by side and the PromQL examples say which prefix to swap. Added a short section on what the shapes mean when tuning, since the settings and the metrics that tell you how to set them now exist together. The batch duration's `full` result no longer means the transport's concurrency, it means the configured batch size, and error-ingestion-design.md no longer says a batch is what one receive cycle handed over. Three things the docs got wrong. Batch size cannot exceed the transport's concurrency, because a receive does not return until its batch has been written, so a timeout does not let a batch accumulate across receive cycles and the setting is only useful downwards. Parallel writers mean custom enrichers are called concurrently, which the settings section now says. The consecutive batch failure gauge counts whole batches failing for any reason, including announcing and forwarding, and is not the condition the custom check and the health endpoints report.
1 parent b11ad24 commit fba2bb6

3 files changed

Lines changed: 176 additions & 9 deletions

File tree

‎docs/error-ingestion-design.md‎

Lines changed: 5 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -8,9 +8,11 @@ for the relational persisters (PostgreSQL and SQL Server) and explains the desig
88
are not obvious from the code, in particular why the write path is hand-written SQL rather than
99
ordinary change-tracked entity saves.
1010

11-
The unit of ingestion is a **batch**: the transport hands the ingester up to `MaximumConcurrency`
12-
messages at a time, and the whole batch is written in a single database transaction. The relevant
13-
types are `EFIngestionUnitOfWork` (accumulation), `FailedMessageBatchWriter` (the write), and the
11+
The unit of ingestion is a **batch**, and the whole batch is written in a single database
12+
transaction. How batches are assembled, how many are written at once, and what can be tuned about
13+
that is covered in [ingestion-pipeline.md](ingestion-pipeline.md); by default a batch holds up to
14+
`MaximumConcurrency` messages and several are written concurrently. The relevant types here are
15+
`EFIngestionUnitOfWork` (accumulation), `FailedMessageBatchWriter` (the write), and the
1416
per-provider `IFailedMessageIngestionSqlDialect` implementations (the statements that differ by
1517
provider). Retry claim insertion is a separate persistence capability behind
1618
`IRetryBatchSqlDialect`; it is used by `RetryBatchStore`, not by the ingestion unit of work.

‎docs/ingestion-pipeline.md‎

Lines changed: 129 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,129 @@
1+
# Ingestion pipeline
2+
3+
Both instances take messages off a queue, batch them, and write each batch to storage. That
4+
machinery is one class, `IngestionPipeline` in `ServiceControl.Infrastructure`, used by
5+
`ErrorIngestion` and `AuditIngestion`. This document covers how it behaves, what can be tuned, and
6+
why the answer to "can batches be written in parallel" belongs to the storage rather than to the
7+
instance.
8+
9+
For what the error instance then does with a batch, see [error-ingestion-design.md](error-ingestion-design.md).
10+
For the metrics it publishes, see [telemetry.md](telemetry.md).
11+
12+
## Shape
13+
14+
```
15+
transport receivers ──► message channel ──► batch assembler ──► batch channel ──► writers
16+
(MaxConcurrency) (bounded) (single reader) (bounded) (1..n)
17+
```
18+
19+
A receiver calls `Enqueue` and then waits on the message's `TaskCompletionSource`. That is what
20+
makes the receive commit only after the message has been written: nothing is acknowledged to the
21+
broker until the batch it landed in is in storage. A failed batch faults the completion sources of
22+
its own messages and nothing else, so the transport redelivers exactly those.
23+
24+
Both channels are bounded. When storage slows down, the batch channel fills, then the message
25+
channel fills, and then the receivers block in `Enqueue`. Back pressure reaches the broker instead
26+
of a queue growing in memory.
27+
28+
The assembler is the only reader of the message channel, which is what keeps several writers fed
29+
without any of them competing for messages.
30+
31+
### Shutdown
32+
33+
`StopAsync` stops receiving first, under the shutdown token rather than a cancelled one, so
34+
messages already in flight finish and their receives commit. It then completes the pipeline and
35+
waits for it to drain, and only then tears the transport infrastructure down, because batches
36+
still forward through its dispatcher.
37+
38+
A hard cancellation does not silently drop what is in flight. The assembler abandons the batch it
39+
was building and whatever is left in the message channel, any batch no writer picked up is
40+
abandoned, and a writer fails the batch it was holding. Every one of those receives is answered,
41+
so they are redelivered rather than left waiting for a shutdown that has already happened.
42+
43+
## Settings
44+
45+
Named per instance, and read from that instance's settings root
46+
(`ServiceControl/...` and `ServiceControl.Audit/...`).
47+
48+
| Setting | Default | Range | What it does |
49+
| --- | --- | --- | --- |
50+
| `ErrorIngestionBatchSize` / `AuditIngestionBatchSize` | the transport's `MaximumConcurrencyLevel` | 1 to 1000 | The most messages one write handles |
51+
| `ErrorIngestionMaxParallelWriters` / `AuditIngestionMaxParallelWriters` | the storage decides, see below | 1 to 16 | How many batches are written at once |
52+
| `ErrorIngestionBatchTimeout` / `AuditIngestionBatchTimeout` | `00:00:00` | 0 to 5 seconds | How long a batch that is not yet full waits for more messages |
53+
54+
As environment variables:
55+
56+
```bash
57+
SERVICECONTROL_ErrorIngestionBatchSize=50
58+
SERVICECONTROL_ErrorIngestionMaxParallelWriters=8
59+
SERVICECONTROL_ErrorIngestionBatchTimeout=00:00:00.100
60+
61+
SERVICECONTROL_AUDIT_AuditIngestionBatchSize=50
62+
SERVICECONTROL_AUDIT_AuditIngestionMaxParallelWriters=8
63+
SERVICECONTROL_AUDIT_AuditIngestionBatchTimeout=00:00:00.100
64+
```
65+
66+
### Batch size
67+
68+
The default, the transport's concurrency, is also the ceiling. A receive does not return until the
69+
batch carrying its message has been written, so the transport holds every one of its concurrency
70+
slots open and no more than that many messages can ever be waiting. Setting a batch size above the
71+
transport's concurrency is therefore unreachable, with or without a batch timeout: the batch never
72+
fills, and a timeout only delays what has already arrived. Larger batches come from raising
73+
`MaximumConcurrencyLevel`, and this setting is what holds writes below it when a storage is happier
74+
with smaller ones.
75+
76+
### Batch timeout
77+
78+
Zero means a partial batch is written rather than waited on, which is what the ingestion did before
79+
the setting existed. A non-zero value trades latency for fewer, larger writes: at volume it costs
80+
nothing, because a full batch never waits, and at a trickle it delays each message by up to the
81+
timeout. Start at 100ms if a storage is clearly happier with larger batches. It cannot make a batch
82+
larger than the transport's concurrency, only fuller.
83+
84+
### Parallel writers
85+
86+
Only raise this for a storage whose writes are safe to interleave. Batches commit in whatever order
87+
they finish, so this is not a free throughput knob, and the pipeline will not let you turn it on
88+
where it is unsafe.
89+
90+
More than one writer also means more than one batch is being enriched and announced at a time, so a
91+
custom `IEnrichImportedErrorMessages` or `IEnrichImportedAuditMessages` has to be thread safe. A
92+
single writer used to serialise them.
93+
94+
## Which storages take concurrent batches
95+
96+
Each ingestion unit of work factory answers `SupportsConcurrentBatches`. Where it says no, the
97+
pipeline uses one writer whatever is configured, and logs a warning if that overrules a setting
98+
someone actually asked for rather than a default.
99+
100+
| Storage | Concurrent batches | Why |
101+
| --- | --- | --- |
102+
| Error, SQL Server and PostgreSQL | yes | The batch writer was built for it: upserts guarded by the attempt times so the newer attempt wins whichever transaction commits last, inserts that tolerate a competing writer's identical row, and a consistent lock order. Running several `--error-ingestion-only` hosts against one database already depends on all of it. |
103+
| Error, RavenDB | no | Failed messages are merged by patch scripts that read and rewrite one document, and nothing orders two patches of the same document against each other. `--error-ingestion-only` refuses to start on RavenDB for the same reason. |
104+
| Audit, RavenDB | yes | Audit documents are independent. Nothing merges two of them, and every batch gets its own bulk insert operation. |
105+
| Audit, in memory | no | Test storage only. |
106+
107+
A storage that says yes gets four writers by default.
108+
109+
### The ordering that concurrency does not excuse
110+
111+
Concurrent writers mean two batches touching the same message can commit in either order, which is
112+
the same condition several ingestion hosts already create. Storage is responsible for making the
113+
end state the same either way, and for failed messages that means comparing the times the events
114+
happened rather than trusting arrival order:
115+
116+
- a retry acknowledgement resolves a message only if no attempt newer than the retry has been
117+
stored, because such an attempt means the message failed again afterwards
118+
- an attempt moves the status only if it is strictly newer than the newest attempt already stored,
119+
so a redelivery of the attempt already there cannot undo a resolve or an archive
120+
121+
## Where things live
122+
123+
| | |
124+
| --- | --- |
125+
| `ServiceControl.Infrastructure/Ingestion/IngestionPipeline.cs` | The channels, the assembler and the writers |
126+
| `ServiceControl.Infrastructure/Ingestion/IngestionSettingsReader.cs` | Reading, validating and resolving the settings above |
127+
| `ServiceControl.Infrastructure/Ingestion/Metrics/` | The metric scopes both instances report through |
128+
| `ServiceControl/Operations/ErrorIngestion.cs` | Transport, watchdog and fault policy for the error queue |
129+
| `ServiceControl.Audit/Auditing/AuditIngestion.cs` | The same for the audit queue |

‎docs/telemetry.md‎

Lines changed: 42 additions & 6 deletions
Original file line numberDiff line numberDiff line change
@@ -2,18 +2,37 @@
22

33
Instances can be configured to emit telemetry to aid in performance testing or troubleshooting performance-related issues.
44

5+
Both the error and the audit instance report their ingestion the same way. Set `OtlpEndpointUrl`
6+
in that instance's settings root to a valid [OTLP endpoint url](https://opentelemetry.io/docs/specs/otel/protocol/exporter/#configuration-options).
7+
Only GRPC endpoints are supported at this stage.
8+
9+
The instruments differ only in their prefix and in the categories a message can fall into, so the
10+
same dashboard works for both with the prefix swapped. What the batches being measured actually are
11+
is covered in [ingestion-pipeline.md](ingestion-pipeline.md).
12+
513
## Error
614

7-
Setting `ServiceControl/PrintMetrics` to `true` will print metrics to the logs at `INFO` level.
15+
Meter `Particular.ServiceControl`, configured with `ServiceControl/OtlpEndpointUrl`.
816

9-
## Audit
17+
- `sc.error.ingestion.batch_duration_seconds` - Message batch processing duration in seconds
18+
- `result` - Whether the batch was written at its configured size: `full`, `partial` or `failed`
19+
- `sc.error.ingestion.storage_duration_seconds` - The storage write of a batch, without the announcing and forwarding around it
20+
- `sc.error.ingestion.message_duration_seconds` - Error message processing duration in seconds
21+
- `message.category` - What the message is: `failed-message` or `retry-confirmation`
22+
- `result` - The outcome: `success`, `failed` or `skipped` (if the message was filtered out and skipped)
23+
- `sc.error.ingestion.failures_total` - Failure counter
24+
- `message.category` - What the message is: `failed-message` or `retry-confirmation`
25+
- `result` - How the failure was resolved: `retry` or `stored-poison`
26+
- `sc.error.ingestion.consecutive_batch_failures_total` - Consecutive batch failures
1027

11-
Set `ServiceControl.Audit/OtlpEndpointUrl` to a valid [OTLP endpoint url](https://opentelemetry.io/docs/specs/otel/protocol/exporter/#configuration-options). Only GRPC endpoints are supported at this stage.
28+
`ServiceControl/PrintMetrics` predates this and no longer has anything to print.
1229

13-
The following ingestion metrics with their corresponding dimensions are available:
30+
## Audit
31+
32+
Meter `Particular.ServiceControl.Audit`, configured with `ServiceControl.Audit/OtlpEndpointUrl`.
1433

1534
- `sc.audit.ingestion.batch_duration_seconds` - Message batch processing duration in seconds
16-
- `result` - Indicates if the full batch size was used (batch size == max concurrency of the transport): `full`, `partial` or `failed`
35+
- `result` - Whether the batch was written at its configured size: `full`, `partial` or `failed`
1736
- `sc.audit.ingestion.message_duration_seconds` - Audit message processing duration in seconds
1837
- `message.category` - Indicates the category of the message ingested: `audit-message`, `saga-update` or `control-message`
1938
- `result` - Indicates the outcome of the operation: `success`, `failed` or `skipped` (if the message was filtered out and skipped)
@@ -22,12 +41,29 @@ The following ingestion metrics with their corresponding dimensions are availabl
2241
- `result` - Indicates how the failure was resolved: `retry` or `stored-poison`
2342
- `sc.audit.ingestion.consecutive_batch_failures_total` - Consecutive batch failures
2443

25-
Example queries in PromQL:
44+
## Reading the ingestion metrics
45+
46+
Example queries in PromQL, shown for audit. Swap `sc_audit` for `sc_error` for the error instance.
2647

2748
- Ingestion rate: `sum (rate(sc_audit_ingestion_message_duration_seconds_count[5m])) by (exported_job)`
2849
- Failure rate: `sum(rate(sc_audit_ingestion_failures_total[5m])) by (exported_job,result)`
2950
- Message duration: `histogram_quantile(0.9,sum(rate(sc_audit_ingestion_message_duration_seconds_bucket[5m])) by (le,exported_job))`
3051

52+
What the shapes mean when tuning:
53+
54+
- `result="full"` dominating the batch duration histogram means batches are being filled, so the
55+
limit is the write and not the arrival of messages. More writers, or a larger batch size, is
56+
where to look.
57+
- `result="partial"` dominating at high throughput means the pipeline is writing before batches
58+
fill. A batch timeout is what makes them accumulate.
59+
- Message duration far above batch duration means messages are queueing behind the writers rather
60+
than being slow to write.
61+
- `consecutive_batch_failures_total` counts whole batches that failed in a row, whatever ended them:
62+
the storage write, but equally announcing or forwarding. It is not a tuning signal, and it is not
63+
the same thing the ingestion custom check and the health endpoints report, which is the critical
64+
failure the watchdog acts on. A gauge climbing while health stays clear means batches are failing
65+
and being retried.
66+
3167
Example Grafana dashboard - https://github.com/andreasohlund/Docker/blob/main/otel-monitoring/grafana-platform-template.json
3268

3369
## Monitoring

0 commit comments

Comments
 (0)