Forwarding

Data Structures
struct	LogForwardHandlers
struct	LogForwardConfig
	Optional forwarder settings; zero in any field takes the default. More...
struct	LogForwardStats
	What a forwarder has done so far. More...

Macros
#define	LOG_FORWARD_SPOOL_DEFAULT   (4 * 1024 * 1024)
	Bytes of spooled frames a forwarder holds by default before it starts dropping the oldest.
#define	LOG_FORWARD_SEGMENT_DEFAULT   (256 * 1024)
#define	LOG_FORWARD_MAXHOPS_DEFAULT   4
	Instances a record may pass through by default before a forwarder refuses to pass it on again.

Typedefs
typedef struct LogForwarder	LogForwarder
	Opaque handle to a registered forwarder.
typedef struct LogForwardHandlers	LogForwardHandlers
typedef struct LogForwardConfig	LogForwardConfig
	Optional forwarder settings; zero in any field takes the default.
typedef struct LogForwardStats	LogForwardStats
	What a forwarder has done so far.

Functions
LogForwarder *	logforwardRegister (int maxlevel, strref chanfilter, const LogForwardHandlers *handlers, void *ctx, const LogForwardConfig *config)
LogDest *	logForwardDest (LogForwarder *fwd)
void	logforwardUnregister (LogForwarder *fwd)
void	logForwardResume (LogForwarder *fwd)
void	logForwardDisconnected (LogForwarder *fwd)
void	logForwardConnected (LogForwarder *fwd)
bool	logForwardTake (LogForwarder *fwd, Buffer *out)
bool	logForwardRecv (LogForwarder *fwd, const uint8 *buf, size_t len)
bool	logForwardApplySub (LogForwarder *fwd, const LogSubSpec *spec)
bool	logForwardCatalog (LogForwarder *fwd, Buffer *out)
void	logForwardStats (LogForwarder *fwd, LogForwardStats *out)

Detailed Description

A destination that encodes the records it receives and hands the bytes to your transport.

cx owns none of the transport. It does not listen, connect, read, write or authenticate. You supply a send callback; cx supplies the codec, a bounded spool for when send says no, loop prevention, and the destination plumbing. Your connection stays entirely yours, so log traffic can share a stream you already have, sit inside your own authentication, or travel over something that is not a socket.

static bool mySend(void *ctx, const uint8 *buf, size_t len)
{
    return myTransportWrite(ctx, buf, len);   // false means "not now"
}
static const LogForwardHandlers kHandlers = { .send = mySend };
 
LogForwarder *fwd = logforwardRegister(LOG_Info, _SL("app/**"), &kHandlers, conn, NULL);
...
logForwardResume(fwd);   // when the transport can take more

Backpressure

send returns false for "not now". Everything from that record on is spooled until logForwardResume(), which drains the spool through send again. The spool is bounded: past its limit the oldest records are dropped and a gap record takes their place, so what the receiver ends up with is the newest traffic plus an honest statement of what is missing.

The spool is memory only, and it is a second copy. Local destinations are the system of record; forwarding is layered on top of them.

Loops

A forwarder's send runs transport code, transport code logs, and those records would come back to the forwarder. Left alone that loop sustains itself with no application activity and does not recover, so cx closes it in two places:

• cx's own transport is never forwarded. A forwarder does not receive anything logged to cx/net or beneath it, however its filter is written. This is not configurable. It costs the central copy of cx's network diagnostics, which still reach every local destination.
• Anything logged inside a log-owned callback stays local. That covers send itself, and withLogLocal() extends it to your own code.

Neither reaches an application transport of your own that logs about its sends from another thread after the send returned. If you have one, do not subscribe a forwarder to its channel.

Records refused for a loop reason are counted; see logForwardStats().

Subscription

A forwarder ships nothing until a receiver asks. There is no locally configured "forward everything to host X": the level and channel filter given at registration say what this process is willing to send, and a subscription arriving from the far end says what it actually wants within that. Until one does, the forwarder is silent and its call sites cost what they cost with nobody listening.

A subscription is applied either from the bytes a receiver sent – logForwardRecv() – or directly, for an application with a control plane of its own – logForwardApplySub(). The two produce identical routing.

Applying one re-binds the destination, so channel filters, per-channel levels and the call-site gate all recompute: a subsystem nobody has subscribed to costs nothing, and one that is subscribed to at Debug starts producing records that were compiled in but dormant. That is fleet-wide verbosity control per subsystem, at runtime, paid for only where somebody is listening.

If a boot window (logBootWindowBegin()) is still open when the first subscription arrives, what it retained is sent ahead of any live record, so a receiver that connects during startup still gets the startup traffic that preceded it. Give the window no deadline if the wait might be longer than the default one.

Macro Definition Documentation

LOG_FORWARD_SEGMENT_DEFAULT

#define LOG_FORWARD_SEGMENT_DEFAULT   (256 * 1024)

Bytes per spool segment by default

The spool drops whole segments, and each one repeats the declarations it needs, so a small value turns a long outage into mostly declarations while a large one makes each drop coarse.

Definition at line 116 of file logforward.h.

Typedef Documentation

LogForwardHandlers

typedef struct LogForwardHandlers LogForwardHandlers

What a forwarder needs from your transport

The table is borrowed, not copied, so it must outlive the forwarder. A static const one is the usual shape.

Function Documentation

logForwardApplySub()

bool logForwardApplySub	(	LogForwarder *	fwd,
		const LogSubSpec *	spec
	)

Apply a subscription directly

For an application whose control plane is its own. Identical in effect to the same subscription arriving through logForwardRecv().

Parameters

fwd	Forwarder to configure
spec	What to send; NULL unsubscribes, returning the forwarder to silence

Returns

false if the forwarder's destination is no longer registered

LogSubSpec spec = { .maxlevel = LOG_Info };
saInit(&spec.patterns, string, 1);
saPush(&spec.patterns, string, _S"app/db/**");
logForwardApplySub(fwd, &spec);
saDestroy(&spec.patterns);

logForwardCatalog()

bool logForwardCatalog	(	LogForwarder *	fwd,
		Buffer *	out
	)

Encode this process's channel inventory

What an operator browses to find out what this binary is capable of logging, before deciding what to subscribe to. Send the bytes back over the same connection.

Channels appear as they are interned, which for most is the first time something logs to them. Call sites are not included.

Parameters

fwd	Forwarder to describe
out	Receives the frames, replacing anything already in the buffer

Returns

false if the catalog could not be encoded

Buffer frames = 0;
logForwardCatalog(fwd, &frames);
myTransportWrite(conn, frames->data, frames->len);
bufDestroy(&frames);

logForwardConnected()

void logForwardConnected

(

LogForwarder *

fwd

)

Tell a forwarder it has a connection again

Replays whatever is spooled through send, then goes live. If send refuses during the replay the rest stays spooled until logForwardResume().

Parameters

fwd	Forwarder that has reconnected

logForwardDest()

LogDest * logForwardDest

(

LogForwarder *

fwd

)

A forwarder's underlying destination

For the destination-level calls a forwarder has no wrapper of its own – extra filter rules with logDestAddFilter(), a different drain group with logDestSetGroup(). Do not unregister it directly; use logforwardUnregister().

Parameters

fwd	Forwarder to inspect

Returns

Its destination handle

logDestAddFilter(logForwardDest(fwd), _SL("app/debug/**"), true);

logForwardDisconnected()

void logForwardDisconnected

(

LogForwarder *

fwd

)

Tell a forwarder its connection is gone

Records are spooled from this point instead of being sent, and the segment in progress is closed so that what is spooled stays decodable on its own.

Parameters

fwd	Forwarder that lost its connection

logForwardRecv()

bool logForwardRecv	(	LogForwarder *	fwd,
		const uint8 *	buf,
		size_t	len
	)

Feed a forwarder bytes its receiver sent

Control frames are applied; anything else is ignored, so this is safe to call with whatever arrives on the connection. A malformed stream fails the call, after which the forwarder accepts no more of it – close the connection.

Call this from one thread at a time: it holds a decoder for the connection it is reading, and a half-delivered frame belongs to whoever is feeding it. Everything else on a forwarder may be called from any thread.

Parameters

fwd	Forwarder to feed
buf	Bytes received
len	Number of bytes

Returns

false if the receiver's stream is malformed

if (!logForwardRecv(fwd, buf, n))
    netsocketClose(sock);

logforwardRegister()

LogForwarder * logforwardRegister	(	int	maxlevel,
		strref	chanfilter,
		const LogForwardHandlers *	handlers,
		void *	ctx,
		const LogForwardConfig *	config
	)

Register a forwarder

The forwarder is an ordinary log destination that happens to encode what it receives. It lands in the remote drain group, so a transport that stalls cannot hold up the local file writes you would need in order to find out why.

A forwarder starts connected but unsubscribed: it sends nothing until logForwardRecv() or logForwardApplySub() says what a receiver wants. Call logForwardDisconnected() as well if there is no transport yet.

Parameters

maxlevel	Most verbose level this forwarder may ever send; a subscription asking for more than this is clamped to it
chanfilter	Channels this forwarder may ever send, as a path pattern; NULL means every unrestricted channel. A subscription narrows this and can never widen it.
handlers	Transport callbacks; borrowed, must outlive the forwarder
ctx	Passed back to the callbacks
config	Optional settings; NULL takes every default

Returns

Forwarder handle, or NULL on failure

LogForwarder *fwd = logforwardRegister(LOG_Info, _SL("app/**"), &kHandlers, conn, NULL);

logForwardResume()

void logForwardResume

(

LogForwarder *

fwd

)

Tell a forwarder the transport can take more

Drains the spool through send until it is empty or send refuses again. Callable from any thread.

Parameters

fwd	Forwarder to resume

logForwardStats()

void logForwardStats	(	LogForwarder *	fwd,
		LogForwardStats *	out
	)

Read a forwarder's counters

Parameters

fwd	Forwarder to inspect
out	Receives the counters

logForwardTake()

bool logForwardTake	(	LogForwarder *	fwd,
		Buffer *	out
	)

Take spooled frames instead of being handed them

For a transport that would rather pull. Hands back the oldest run of complete frames and removes it from the spool. Wrap whatever you then do with the bytes in withLogLocal(), so that a transport which logs about its own sends cannot feed itself.

Parameters

fwd	Forwarder to take from
out	Receives the frames, replacing anything already in the buffer

Returns

false if nothing is spooled

Buffer frames = 0;
while (logForwardTake(fwd, &frames)) {
    withLogLocal() { myTransportWrite(conn, frames->data, frames->len); }
}
bufDestroy(&frames);

logforwardUnregister()

void logforwardUnregister

(

LogForwarder *

fwd

)

Unregister a forwarder

Anything still spooled is discarded. The close handler runs once the log system has finished with the destination.

Parameters

fwd	Forwarder to unregister; invalid afterwards