NAME
Punk::Observe::Ingest - the OTLP receiver
SYNOPSIS
use Punk::Observe::Ingest;
my $app = Punk::Observe::Ingest->new(
max_body => 16 * 1024 * 1024,
auth => sub { my ($env) = @_; resolve_tenant($env) },
on_batch => sub {
my ($tenant, $signal, $body, $encoding) = @_;
append_to_wal($tenant, $signal, $body);
},
)->to_app;A PSGI application serving /v1/traces, /v1/metrics and /v1/logs over application/x-protobuf and application/json, with optional gzip or deflate request bodies.
PSGI rather than a set of Punk routes, so that the engine mounts into any application: Punk's mount, a bare plackup, or Punk::Plugin::Observe. The three transports differ only in how bytes become records; everything after that - the limits, the WAL append, the response - is one code path, and sharing it is what keeps them from drifting into three behaviours.
to_app hands back the receiver itself rather than a closure around it, so a request reaches the decoder without a Perl frame in front of it.
METHODS
new
Punk::Observe::Ingest->new(%options)max_body-
The largest request body accepted, in bytes. Default 16MB. Enforced from
CONTENT_LENGTHbefore the body is read, and again against the decompressed size. max_ratio-
How far a compressed body may expand, as a multiple of what arrived. Defaults to
$Punk::Observe::Ingest::MAX_RATIO, which is 20. max_records-
The per-batch record cap. Records over it are rejected and reported through
partial_success; see "STATUS CODES". Unset means uncapped. auth-
A coderef given the PSGI environment, returning a tenant identifier or undef. See "AUTHENTICATION".
on_batch-
A coderef called as
($tenant, $signal, $body, $encoding, $out)once per accepted batch, where$bodyis the decoded request body and$outcarriesrecords,rejected,dropped_bad_traceandclamped_durations. A false return or a die is a503. grpc-
Dies. See "GRPC".
to_app
The PSGI application: a coderef taking $env and returning a PSGI response.
call
$ingest->call($env)The same thing as a method, for a caller that would rather mount the object.
STATUS CODES
These are a data-retention decision rather than a formality. An OTLP client retries 429, 502, 503 and 504 and drops everything else, so getting them backwards means either losing telemetry or being retried forever.
200 decoded and stored
200 decoded, some records rejected - carries partial_success
400 malformed body
401 no ingest credential, where one is required
404 not an OTLP path
405 not a POST
413 body over max_body, or a compressed body that cannot be inflated
within it
415 unsupported content type
503 the store is unavailable - retryable, with Retry-AfterA partial rejection is a 200 with a partial_success body, never a 4xx. That is the channel that says "I kept 9,600 of these and rejected 400", and Punk::OpenTelemetry's exporter reads and reports it. A 4xx would make the client re-send the whole batch indefinitely at exactly the moment the server is under pressure.
A batch that decodes to zero records is a 200 with no append. An empty frame would carry a nonsense timestamp span, and a reader pruning on it skips exactly the wrong frames.
GRPC
Not available, deliberately.
A gRPC call returns HTTP 200 even when it fails; the outcome lives entirely in the grpc-status and grpc-message HTTP/2 trailers. PSGI has no trailer channel, and Hyperman's HTTP/2 path has none either. An endpoint serving gRPC without them would answer 200 with no status, which an exporter reads as complete success while every batch is discarded.
Passing grpc => 1 therefore dies at construction, with the reason, so the problem surfaces where somebody can act on it rather than as silent data loss. Use the OTLP/HTTP endpoints.
COMPRESSION
gzip and deflate request bodies are accepted. The size limit applies to the decompressed body as well as the compressed one, bounded by max_ratio (default 20). Without that, a forty-kilobyte body expanding to gigabytes is a denial of service that needs no exploit.
The two encodings are not the same stream and are not treated as one: gzip is gzip framing and deflate is the zlib wrapper it names. A body labelled as one and framed as the other is refused rather than guessed at, as is a body that stops part way through - both are 413. Passing either on would turn a mislabelled or truncated encoding into a malformed OTLP payload somewhere further down, with the real cause thrown away.
Where the build found no zlib, the core decompressors do the work instead, under the same ceiling.
AUTHENTICATION
auth is a coderef given the PSGI environment, returning a tenant identifier or undef. Undef is a 401, and so is a resolver that dies - a credential that cannot be checked has not been checked.
It runs before anything is read or decoded, so an unauthenticated caller cannot make the server spend on either.
With no auth, every batch is accepted and attributed to the default tenant. That is the right default for a self-hosted install on a private network, and it is the seam a multi-tenant deployment replaces - the engine does not change, the resolver does.
The credential is read once, here, and no later code takes a tenant identifier from anything a client can influence.
FUNCTIONS
The receiver itself is the object above. These are the decoding steps it is built from, reachable on their own.
count
my $out = Punk::Observe::Ingest::count($bytes, $signal, 'protobuf');Decodes an OTLP protobuf batch and reports what is in it without building the records. This is what the receiving path does: decode, count, append, reply, with no Perl value per record anywhere.
{ ok => 1, records => 600, dropped_bad_trace => 0, clamped_durations => 0 }$signal is traces, metrics or logs. The encoding must be protobuf; anything else is fatal, as is an unknown signal.
To get the records themselves, use "decode" in Punk::Observe::Decode.
decode_json
my $out = Punk::Observe::Ingest::decode_json($doc, $signal);Turns an already-parsed OTLP/JSON document into records. $doc is what File::Raw::JSON::file_json_decode returned.
{ ok => 1, records => [ ... ], dropped_bad_trace => 0, clamped_durations => 0 }The records have the shape in "decode" in Punk::Observe::Decode, and must match what the protobuf path produces for the same batch. Four rules make that true, and each is a way telemetry silently goes missing when it is broken:
- 1. Keys are
lowerCamelCase. Asnake_casekey parses as a message with every field absent, which is proto3's own semantics: accepted, stored, and empty. - 2. An identifier may be hex or base64, and both must land on the same bytes. A trace whose ids are spelled two ways splits in half and presents as data simply being missing.
- 3. A 64-bit value is a string. As a JSON number a nanosecond timestamp loses its last digits.
- 4. An enum may be its name or its number.
decode_append
my $r = Punk::Observe::Ingest::decode_append(
$wal_path, $body, $signal, $encoding,
$fsync_policy, $fsync_interval_ns, $want_records);Decodes a batch and appends it to the write-ahead log in one pass. This is the ingest path.
$body is the wire bytes when $encoding is protobuf, and the document File::Raw::JSON::file_json_decode returned when it is json.
{
ok => 1,
appended => 1,
opened => 1,
n => 480, # records decoded
frames => 1,
bytes => 42880,
fsyncs => 0,
records => [ ... ], # only when want_records is true
dropped_bad_trace => 0,
clamped_durations => 0,
}The decoder's record array and its arena go straight into the log's writev. Nothing is copied, and no SV is built for a record unless $want_records asks for one - which is what an on_records observer costs and why it is off by default.
Read appended, not only ok. ok says the batch decoded; appended says the bytes reached the log, and errno carries the system error when it did not. A batch that decoded and was not stored must be refused to its exporter, or the exporter drops the only other copy.
An empty batch is a success that writes nothing and does not create the file.
partial_success_pb
my $body = Punk::Observe::Ingest::partial_success_pb($rejected, $message);The OTLP partial_success response body, in protobuf. $message is truncated to 40 characters.
This is the channel that says "I kept 9,600 of these and rejected 400", and Punk::OpenTelemetry's exporter reads it. The JSON equivalent is built by the receiver directly.
SEE ALSO
Punk::Observe, Punk::Observe::Decode, Punk::Observe::WAL, Punk::Observe::Query, Punk::OpenTelemetry