Package com.ultikits.ultitools.manager

Class AgentLogBatcher

java.lang.Object

com.ultikits.ultitools.manager.AgentLogBatcher

final class AgentLogBatcher
extends Object

Buffers console log lines and releases them as protocol 2 log.lines frames, within the contract's per-line, per-frame and rate limits.

Redaction. This class never redacts. Its only producer must be the redacting builders of this phase (UltiPanelLogTransmitter#buildLogData, the error-report collector's serialisation and the player-event builders): they hand over text in which a sensitive command's arguments are already replaced by *** (D-20). Two producers with two redaction paths would let one of them miss, so the batcher takes already-redacted text and states that contract here instead of repeating the work.

Limits. Every limit comes from CloudChannel.limits() on each flush, because a reconnect negotiates a fresh welcome. offer(long, java.util.logging.Level, java.lang.String) has no channel, so it applies the limits of the last flush and, before the first one, the contract's own defaults (which are the values an unnegotiated connection uses too). Nothing about a local drop is ever reported to the gateway: the contract has no such message (S-56), so the gateway tells the browser about gaps itself. Only a local FINE line summarises what was dropped, at most once a minute.

Threading. offer(long, java.util.logging.Level, java.lang.String) may be called from any thread, flush(com.ultikits.ultitools.websocket.CloudChannel) from the connection's scheduler and from the caller that saw offer(long, java.util.logging.Level, java.lang.String) report a full batch, and reset() from the session's close path; every method is safe to call concurrently and nothing blocks on I/O beyond the channel's own send.

Package-private: wired by the session plan (22-22), not part of the public API.

Nested Class Summary

Nested Classes

Modifier and Type	Class	Description
private static final class	AgentLogBatcher.Line	A buffered line, kept unformatted so limits are applied when the flush reads them.

Field Summary

Fields

Modifier and Type	Field	Description
private final ArrayDeque<AgentLogBatcher.Line>	buffered
private final LongSupplier	clock
private static final long	DROP_REPORT_INTERVAL_MS	Local diagnostics only: how often the drop summary may be written.
private long	dropped
private long	dropsSinceReport
private static final int	FRAME_OVERHEAD_BYTES	UTF-8 bytes the channel's envelope adds around payload.lines: v, a 26-character id, the type, a 16-digit timestamp and the {"lines":[]} wrapper.
private long	lastDropReportAt
private long	lastRefillAt
private static final String	LEVEL_DEBUG
private static final String	LEVEL_ERROR	The protocol level names.
private static final String	LEVEL_INFO
private static final String	LEVEL_WARN
private CloudLimits	limits
private static final Logger	LOGGER	Runs the network call and the socket's state machine; never re-enters this batcher.
private double	tokens
(package private) static final String	TRUNCATION_SUFFIX	Appended to a message that had to be cut, so a reader knows the line is incomplete.

Constructor Summary

Constructors

Constructor	Description
AgentLogBatcher(LongSupplier clock)	Creates a batcher.

Method Summary

Modifier and Type	Method	Description
private void	countDrop()
(package private) long	droppedByRateLimit()	Reports how many lines were dropped locally, by the rate bucket or by the bounded buffer.
(package private) void	flush(CloudChannel channel)	Sends the buffered lines as one log.lines frame, if the channel can take it.
private static int	frameOverheadBytes()	Measures the envelope the channel adds around the payload, so a frame is never built larger than maxFrameBytes before the channel sees it.
private static String	levelName(Level level)	Maps a JUL level onto the contract's four names by thresholds, so custom levels map the same way the built-in ones do: >= 1000 ERROR (Level.SEVERE), >= 900 WARN (Level.WARNING), >= 700 INFO (Level.INFO and Level.CONFIG), anything lower DEBUG (Level.FINE and below).
private static com.google.gson.JsonObject	lineObject(long ts, String level, String message)	Builds one log.lines line object: exactly ts, level and message, which is all the schema allows.
(package private) boolean	offer(long ts, Level level, String redactedMessage)	Offers one console line.
private static String	prefix(String text, int codePoints)	Returns the first codePoints characters of text, counted in code points so a surrogate pair is either fully in or fully out.
private void	refill(long now)	Adds the tokens earned since the last call, capped at the burst the current limits allow.
private void	reportDrops(long now)	Writes the local drop summary, at most once per interval and only when something was dropped since the last one.
(package private) void	reset()	Forgets every buffered line, refills the rate bucket and zeroes the local drop counters.
private void	trimToCapacity()	Drops the oldest buffered lines until the buffer fits one burst.
private static String	truncate(long ts, String levelName, String message, int lineMaxBytes)	Cuts a message so its serialised line fits lineMaxBytes, appending TRUNCATION_SUFFIX.
private static int	utf8Bytes(String text)

Methods inherited from class java.lang.Object

clone, equals, finalize, getClass, hashCode, notify, notifyAll, toString, wait, wait, wait

Field Details

TRUNCATION_SUFFIX

static final String TRUNCATION_SUFFIX

Appended to a message that had to be cut, so a reader knows the line is incomplete.

See Also:

• Constant Field Values

LEVEL_ERROR

private static final String LEVEL_ERROR

The protocol level names.

See Also:

• Constant Field Values

LEVEL_WARN

private static final String LEVEL_WARN

See Also:

• Constant Field Values

LEVEL_INFO

private static final String LEVEL_INFO

See Also:

• Constant Field Values

LEVEL_DEBUG

private static final String LEVEL_DEBUG

See Also:

• Constant Field Values

LOGGER

private static final Logger LOGGER

Runs the network call and the socket's state machine; never re-enters this batcher.

DROP_REPORT_INTERVAL_MS

private static final long DROP_REPORT_INTERVAL_MS

Local diagnostics only: how often the drop summary may be written.

See Also:

• Constant Field Values

FRAME_OVERHEAD_BYTES

private static final int FRAME_OVERHEAD_BYTES

UTF-8 bytes the channel's envelope adds around payload.lines: v, a 26-character id, the type, a 16-digit timestamp and the {"lines":[]} wrapper. The real id is 26 characters and a real timestamp 13 digits, so this is an upper bound.

clock

private final LongSupplier clock

buffered

private final ArrayDeque<AgentLogBatcher.Line> buffered

limits

private CloudLimits limits

tokens

private double tokens

lastRefillAt

private long lastRefillAt

dropped

private long dropped

dropsSinceReport

private long dropsSinceReport

lastDropReportAt

private long lastDropReportAt

Constructor Details

AgentLogBatcher

AgentLogBatcher(LongSupplier clock)

Creates a batcher.

Parameters:

clock - the clock whose milliseconds drive the rate bucket; the same unit as the ts values passed to offer(long, java.util.logging.Level, java.lang.String), so a test can fake it

Method Details

offer

boolean offer(long ts,
 Level level,
 String redactedMessage)

Offers one console line. Never throws.

A line over the token bucket's limit is dropped and counted; so is the oldest buffered line when the buffer would exceed one burst (a closed channel that keeps receiving lines must not grow the heap).

The full-batch signal. The contract asks for a flush every second and whenever a batch is full, so the caller must flush as soon as this method reports one: a gateway that admits more than log.batchMaxLines lines a second would otherwise fill the burst-sized buffer within one tick and lose lines the token bucket had accepted. The signal describes the buffer, not this line, so it also stays true for a line the token bucket dropped while a full batch was already waiting.

Parameters:

ts - the line's timestamp in milliseconds

level - the JUL level; null counts as Level.INFO

redactedMessage - the text with sensitive command arguments already replaced by ***; null counts as an empty message

Returns:

true while at least log.batchMaxLines lines are buffered, so the caller can flush now instead of waiting for the next tick; a flush that sends (or drops) them clears it

flush

void flush(CloudChannel channel)

Sends the buffered lines as one log.lines frame, if the channel can take it.

Limits are read from channel.limits() on every flush, because a reconnect's new welcome may negotiate different ones. Nothing is sent while the channel is closed; the lines stay buffered for the next open channel, and a frame the socket refuses is put back for the next attempt.

Parameters:

channel - the connection to send on; null is ignored

reset

void reset()

Forgets every buffered line, refills the rate bucket and zeroes the local drop counters. Called when the session closes: the next session starts clean.

droppedByRateLimit

long droppedByRateLimit()

Reports how many lines were dropped locally, by the rate bucket or by the bounded buffer. Nothing of this reaches the gateway.

Returns:

the count since the last reset()

refill

private void refill(long now)

Adds the tokens earned since the last call, capped at the burst the current limits allow.

Parameters:

now - the current clock value in milliseconds

trimToCapacity

private void trimToCapacity()

Drops the oldest buffered lines until the buffer fits one burst.

countDrop

private void countDrop()

reportDrops

private void reportDrops(long now)

Writes the local drop summary, at most once per interval and only when something was dropped since the last one. Called from flush(com.ultikits.ultitools.websocket.CloudChannel), the once-a-second tick, so a burst is summarised by its total rather than by its first line. The count is the cumulative number dropped since the last reset(), the same figure droppedByRateLimit() reports. FINE on the batcher's own logger: it is a diagnostic for the operator and must not depend on the connection it describes.

Parameters:

now - the current clock value in milliseconds

levelName

private static String levelName(Level level)

Maps a JUL level onto the contract's four names by thresholds, so custom levels map the same way the built-in ones do: >= 1000 ERROR (Level.SEVERE), >= 900 WARN (Level.WARNING), >= 700 INFO (Level.INFO and Level.CONFIG), anything lower DEBUG (Level.FINE and below).

Parameters:

level - the JUL level, never null

Returns:

the contract level name

truncate

private static String truncate(long ts,
 String levelName,
 String message,
 int lineMaxBytes)

Cuts a message so its serialised line fits lineMaxBytes, appending TRUNCATION_SUFFIX. The cut lands on a code-point boundary, so a surrogate pair is never split into an unpaired half.

Parameters:

ts - the line's timestamp, whose digits are part of the measured size

levelName - the contract level name

message - the full redacted message

lineMaxBytes - the largest UTF-8 size the serialised line object may have

Returns:

the message unchanged when it fits, otherwise a cut message ending in the suffix

prefix

private static String prefix(String text,
 int codePoints)

Returns the first codePoints characters of text, counted in code points so a surrogate pair is either fully in or fully out.

Parameters:

text - the text

codePoints - how many code points to keep

Returns:

the prefix

lineObject

private static com.google.gson.JsonObject lineObject(long ts,
 String level,
 String message)

Builds one log.lines line object: exactly ts, level and message, which is all the schema allows.

Parameters:

ts - the line's timestamp

level - the contract level name

message - the redacted text

Returns:

the line object

frameOverheadBytes

private static int frameOverheadBytes()

Measures the envelope the channel adds around the payload, so a frame is never built larger than maxFrameBytes before the channel sees it.

Returns:

the upper-bound UTF-8 size of the frame without its lines

utf8Bytes

private static int utf8Bytes(String text)