Package com.ultikits.ultitools.manager

Class UltiPanelLogTransmitter

java.lang.Object
com.ultikits.ultitools.manager.UltiPanelLogTransmitter
@Internal public class UltiPanelLogTransmitter extends Object
UltiPanel log transmitter. Implements the log-transmission functionality per the API documentation specification.
Version:
1.0.0
Author:
UltiKits

  • Field Details

    • MAX_QUEUE_SIZE

      private static final int MAX_QUEUE_SIZE
      See Also:

    • MIN_INTERVAL_MS

      static final int MIN_INTERVAL_MS
      Shared lower bound for the batch-send interval (#432/WR-01). Before this fix, the boot-time path (LogStreamManager.loadBatchConfiguration()) clamped to 1000ms via its own inline Math.max(1000, interval), while this class's own setIntervalMs(int) -- reachable live, over the panel's WebSocket config action -- only rejected a non-positive value, leaving every value from 1ms up accepted. A panel operator (or a compromised/buggy panel session) could drive the batch scheduler to a roughly 1000-sends/second cadence indefinitely. Both entry points now share this one constant instead of the live path being silently more permissive than the boot path.
      See Also:

    • DISCARD_REPORT_INTERVAL_MS

      static final long DISCARD_REPORT_INTERVAL_MS
      Shortest time between two reports of records discarded on queue overflow (#486).
      See Also:

    • webSocketClient

      private final UltiPanelWebSocketClient webSocketClient

    • serverId

      private final String serverId

    • agentSink

      private final UltiPanelLogTransmitter.AgentLineSink agentSink
      Where the v2 connection takes each redacted line (plan 22-22), or null for the v1 transmitter. When set, sendLog(java.lang.String, java.lang.String, java.lang.String, java.lang.Throwable) and replayLog(java.lang.String, java.lang.String, java.lang.String, java.lang.Throwable, long) hand every entry buildLogData(java.lang.String, java.lang.String, java.lang.String, java.lang.Throwable, long) built to this sink and nothing else: no v1 queue, batch or frame is ever built.

    • logTransmissionEnabled

      private final AtomicBoolean logTransmissionEnabled

    • batchModeLock

      private final Object batchModeLock
      Guards the batch-vs-immediate DISPATCH DECISION in sendLog(java.lang.String, java.lang.String, java.lang.String, java.lang.Throwable) against the enabled-to-disabled TRANSITION in setBatchEnabled(boolean) (Gate-2 finding, round 6). Before this lock, setBatchEnabled(false) wrote this.batchEnabled = false before flushLogs() drained the existing queue, so a concurrent sendLog() call could read the new false value and send a NEWER record immediately while flushLogs() was still draining OLDER queued records -- the newer record then arrived at the panel before the older ones, breaking FIFO order. Holding this lock across both sendLog's dispatch decision and the WHOLE disable transition means every sendLog call observes the transition as atomic: either entirely before it (queued, and included in the flush that is about to run) or entirely after it (flag already false, sent immediately) -- never during.

    • externalDrainMode

      private final AtomicBoolean externalDrainMode

    • externalSizeThresholdCallback

      private volatile Runnable externalSizeThresholdCallback
      Invoked from addToBatch(JsonObject) whenever the queue reaches batchSize WHILE externalDrainMode is active (Gate-2 finding, round 6). Without this, reaching the size threshold under external drain mode had no effect at all -- sendBatch()'s own early-return for external drain mode silently swallowed the threshold crossing that would otherwise have triggered an immediate send, so a documented size threshold never actually shortened delivery latency while monitoring was active, and a sustained burst could fill MAX_QUEUE_SIZE and start discarding old entries despite repeatedly crossing the threshold. null by default (no external owner wired up); LogStreamManager.initialize(com.ultikits.ultitools.websocket.UltiPanelWebSocketClient) sets it alongside setExternalDrainMode(boolean).

    • externalDrainCoordinationLock

      private volatile Object externalDrainCoordinationLock
      External coordination lock for flushLogs() (Gate-2 finding, round 9). null by default (no external owner wired up -- flushLogs() runs unsynchronized, its original behaviour). LogStreamManager.initialize(com.ultikits.ultitools.websocket.UltiPanelWebSocketClient) wires this to ServerMonitorManager's OWN logDrainLock (the same object ServerMonitorManager.drainAndSendLogsOnly(com.ultikits.ultitools.manager.UltiPanelLogTransmitter) and its inline sendBatchUpdate drain already synchronize on) alongside setExternalSizeThresholdCallback(java.lang.Runnable). Without this, a live batchConfig.enabled: false update's flush ran entirely outside that lock, so it could still send NEWER records ahead of OLDER ones the monitor had already polled but not yet sent, even after round 7's monitor-side-only serialization.

    • batchEnabled

      private volatile boolean batchEnabled

    • batchSize

      private volatile int batchSize

    • intervalMs

      private volatile int intervalMs

    • logQueue

      private final ConcurrentLinkedQueue<com.google.gson.JsonObject> logQueue

    • batchScheduler

      private final ScheduledExecutorService batchScheduler

    • heldBatch

      private final AtomicReference<com.google.gson.JsonArray> heldBatch
      The one batch whose send failed, kept to be sent before anything newer (#486). Before 6.3.0 a batch was polled off logQueue and then sent, and nothing put it back when the send failed, so a socket that closed around the send lost the batch without a trace. While a batch is held, neither this class's own sender nor an external drain (drainQueue(int)) takes a newer record, so the panel still receives records in order. At most one batch is held, so a permanently failing send retries once per attempt and never spins, and the held records count against nothing else (the queue stays bounded by MAX_QUEUE_SIZE).

      A failed send is one that throws or after which the client no longer reports itself connected (the closed-socket path of UltiPanelWebSocketClient#sendMessage returns without sending). Whether the panel processed a frame that was sent cannot be known at this layer, so delivery stays best effort: a batch whose connection drops just after it was written may be sent again after reconnecting.

    • discardedRecords

      private final AtomicLong discardedRecords
      Records discarded on queue overflow since the last report (#486).

    • lastDiscardReportMs

      private final AtomicLong lastDiscardReportMs
      When the last overflow report was logged; Long.MIN_VALUE before the first.

    • batchSenderTask

      private volatile ScheduledFuture<?> batchSenderTask
      The currently-scheduled batch-send task, or null while batching is disabled.

      Tracked so setIntervalMs(int) and setBatchEnabled(boolean) can cancel and resubmit it -- before #432, the interval was baked into the one scheduleWithFixedDelay call the constructor made, and the setter only mutated the field without ever touching the already-running task.

    • replayQueue

      private final Deque<com.google.gson.JsonObject> replayQueue
      The start-up records the early capture replayed into this transmitter (#487), oldest first, not yet delivered. Guarded by batchModeLock.

      Kept apart from logQueue because the replay has its own fixed chunking, independent of the live batch settings (as of 6.3.0): log_batch messages of at most REPLAY_CHUNK_MAX_BYTES, the first at once and then one every REPLAY_SPACING_MS. With live batching off, every replayed record used to be its own message -- about 350 at once on connect on a real server -- and the panel's per-client quota (50 messages in 10 seconds) rejected them. It is not bounded here: the early capture already bounds what it replays (2,000 records, about 512 KiB).

    • REPLAY_CHUNK_MAX_BYTES

      static final int REPLAY_CHUNK_MAX_BYTES
      The most bytes one replay message may take, serialized as the client sends it.

      Measured against the UltiPanel Worker (websocket-server.ts): its quota counts messages, 50 per 10 seconds per client, and a rejected message counts too; its log_batch handler accepts an array of any length and sets no size limit of its own; the Cloudflare platform caps one received WebSocket message at 1 MiB. 64 KiB stays a sixteenth of that cap, also for the copy the Worker broadcasts to each subscribed panel, and holds a few hundred typical start-up lines, so a full start-up buffer drains in about ten messages.

      See Also:

    • REPLAY_SPACING_MS

      static final long REPLAY_SPACING_MS
      The time between two replay messages: at most 10 of the panel's 50 messages per 10 seconds, leaving the rest for live traffic.
      See Also:

    • REPLAY_FRAME_RESERVE_BYTES

      private static final int REPLAY_FRAME_RESERVE_BYTES
      Room kept in each replay message for its envelope: type, server ID, timestamp, brackets.
      See Also:

    • TRUNCATED_MARK

      private static final String TRUNCATED_MARK
      Appended to an entry shortened to fit one replay message.
      See Also:

    • WIRE_GSON

      private static final com.google.gson.Gson WIRE_GSON
      Serializes as UltiPanelWebSocketClient does, to measure what goes on the wire.

    • replaySenderTask

      private volatile ScheduledFuture<?> replaySenderTask
      The scheduled replay-send task, or null when nothing is left to replay.

  • Constructor Details

  • Method Details

    • v1ServerId

      private static String v1ServerId(String serverId)
      The v1 constructor's server ID, defaulted as in 6.3.0; also writes that constructor's one INFO line per run, here because nothing may precede its this(...) call.

    • sendLog

      public void sendLog(String level, String message, String source, Throwable throwable)
      Sends a log message to the backend.
      Parameters:
      level - the log level (info, warning, error, debug)
      message - the log message
      source - the log source (e.g.: "server", "plugin:name")
      throwable - the exception object (optional)

    • buildLogData

      private com.google.gson.JsonObject buildLogData(String level, String message, String source, Throwable throwable, long timestamp)
      Builds one log entry as the panel receives it.

      This is the only place a cloud log entry is built, for the live stream (sendLog(java.lang.String, java.lang.String, java.lang.String, java.lang.Throwable)) and for the start-up replay (replayLog(java.lang.String, java.lang.String, java.lang.String, java.lang.Throwable, long)) alike, so the message and the stack trace are passed through SensitiveCommandRegistry.redactText(java.lang.String) here and no path can bypass it. The replay builds its entries when the stream opens, so a command a module registers while it enables also covers lines captured before it enabled. Only this copy is redacted: the LogRecord, the local console and the log files keep the original.

      Parameters:
      level - the log level, info when blank
      message - the message
      source - the log source, server when blank
      throwable - the exception, or null
      timestamp - the entry's time, in milliseconds since the epoch
      Returns:
      the entry

    • offerToAgent

      private void offerToAgent(String level, String message, String source, Throwable throwable, long timestamp)
      Hands one line to the v2 sink: the message and stack trace buildLogData(java.lang.String, java.lang.String, java.lang.String, java.lang.Throwable, long) redacted, the stack trace appended to the message because log.lines has no stack-trace field. Never throws: a log line must not break the logging call that produced it.

    • julLevel

      private static Level julLevel(String level)
      The JUL level of a panel level name, as SystemLogHandler#mapLogLevel produced it.

    • redact

      private static String redact(String text)
      Redacts registered commands' arguments from text bound for the panel. Without a framework instance nothing can have been registered, so the text is returned unchanged.

    • replayLog

      public void replayLog(String level, String message, String source, Throwable throwable, long timestamp)
      Queues one start-up record the early capture replays (#487) for delivery in batches.

      Queuing is an in-memory hand-off and does not depend on the connection: the capture has already given the record up, so a socket that closes while the capture is still handing its records over (#677) must not lose the rest. Nothing is sent here; startReplay()'s sender sends only while the client reports itself connected, and records still queued when the panel reconnects on a new client are handed to the next transmitter (takePending()). The transmission switch still applies -- nothing is queued while transmission is off, or when there is no client -- and the queue holds at most what EarlyLogCapture kept: at most 2,000 records and about 512 KiB, and nothing once it has waited five minutes for the stream.

      Parameters:
      level - the log level (info, warning, error, debug)
      message - the log message
      source - the log source (e.g.: "server", "plugin:name")
      throwable - the exception object (optional)
      timestamp - when the record was logged -- kept, so the panel can place a replayed line among the live ones that may arrive while the replay is still being sent

    • wireBytes

      private static int wireBytes(com.google.gson.JsonElement element)

    • fitReplayEntry

      private static com.google.gson.JsonObject fitReplayEntry(com.google.gson.JsonObject entry)
      Shortens an entry that would not fit one replay message on its own -- the stack trace first, then the message -- so it is sent shortened rather than not at all.

    • startReplay

      public void startReplay()
      Starts delivering the queued replay: the first message now, then one every REPLAY_SPACING_MS until none is left, each holding as many records, oldest first, as fit in REPLAY_CHUNK_MAX_BYTES. The live batch.size and batch.interval do not apply. A message that does not go out is kept, at the front, for the next attempt.

      Live records are not held back while the replay is sent: with live batching off they still go out one by one at once, so a live line can reach the panel before the last replayed messages. Each replayed entry keeps the time its record was logged.

    • sendReplayBatch

      void sendReplayBatch()
      Sends the next replay message -- as many queued records, oldest first, as fit in REPLAY_CHUNK_MAX_BYTES -- if the panel is connected; stops the replay task once nothing is left. Package-private for tests.

    • stopReplaySender

      private void stopReplaySender()

    • getReplayQueueSize

      public int getReplayQueueSize()
      How many replayed records are still waiting to be sent.
      Returns:
      the number of replayed records not yet delivered

    • info

      public void info(String message, String source)
      Convenience methods: send a log at a specific level.

    • warning

      public void warning(String message, String source)

    • error

      public void error(String message, String source, Throwable throwable)

    • debug

      public void debug(String message, String source)

    • sendLogImmediately

      private void sendLogImmediately(com.google.gson.JsonObject logData)
      Sends a single log entry immediately.

    • addToBatch

      private void addToBatch(com.google.gson.JsonObject logData)
      Adds a log entry to the batch queue.

    • startBatchSender

      private void startBatchSender()
      (Re)starts the batch-send task at the current intervalMs. Cancels whatever task was previously scheduled first, so this is safe to call to both start fresh and reschedule (#432).

    • scheduledTick

      private void scheduledTick()
      One run of the batch-send task: send what is due, then report overflow discards. The report is made here, outside batchModeLock, never from inside addToBatch(com.google.gson.JsonObject): logging from there would re-enter this pipeline while the lock is held.

    • reportDiscardedRecords

      void reportDiscardedRecords(long now)
      Logs one WARNING with the number of records discarded on queue overflow since the last report, at most once per DISCARD_REPORT_INTERVAL_MS (#486). Nothing is logged when nothing was discarded. The count goes to the server log only; the panel protocol is unchanged.
      Parameters:
      now - the current time in milliseconds

    • deliver

      private boolean deliver(com.google.gson.JsonObject frame)
      Sends one frame and reports whether it went out: it did not if sendMessage threw, or if the client no longer reports itself connected afterwards -- the closed-socket path of sendMessage returns without sending (#486).

    • pollBatch

      private com.google.gson.JsonArray pollBatch()
      Pulls up to batchSize records off the queue, oldest first.

    • sendHeldBatch

      private boolean sendHeldBatch()
      Sends the batch held from a failed send, if there is one (#486).
      Returns:
      true when nothing is held any more (none was, or it went out), false when the held batch could not be sent and is still held

    • deliverBatch

      private boolean deliverBatch(com.google.gson.JsonArray logs)

    • stopBatchSender

      private void stopBatchSender()
      Cancels the currently-scheduled batch-send task, if any, and clears the reference.

    • setIntervalMs

      public void setIntervalMs(int intervalMs)
      Sets the batch-send interval, in milliseconds, and -- unlike the field it replaces as a plain Lombok setter -- reschedules the already-running batch-send task to actually use it (#432). A no-op, deliberately not touching the scheduler, when the requested value equals the current one, so a live-config-update round-trip that resends the same value does not churn the scheduler. Has no effect on the scheduler while batching is disabled (there is no running task to reschedule); the new value still takes effect the next time batching is enabled.
      Parameters:
      intervalMs - the new interval; must be at least MIN_INTERVAL_MS
      Throws:
      IllegalArgumentException - if intervalMs is below MIN_INTERVAL_MS (including zero or negative) -- the previous interval is left in effect

    • setBatchSize

      public void setBatchSize(int batchSize)
      Sets the batch send-threshold size, rejecting anything below 1 (Gate-2 finding). A value of zero or negative would make sendBatch()'s own for (int i = 0; i < batchSize ...) loop consume nothing on every scheduled run, while addToBatch(JsonObject)'s logQueue.size() >= batchSize check is simultaneously always true (any non-negative queue size satisfies >= 0 or >= a negative number) -- so every single enqueued record would trigger an immediate sendBatch() call that dequeues nothing, silently stalling delivery while burning CPU on every log line, rather than the panel request being rejected outright. LogStreamManager#loadBatchConfiguration()'s own boot-time path already clamped to Math.max(1, batchSize); the live panel path had no lower bound at all before this fix, unlike setIntervalMs(int)'s pre-existing floor.
      Parameters:
      batchSize - the new batch size; must be at least 1
      Throws:
      IllegalArgumentException - if batchSize is below 1 -- the previous value is left in effect

    • setBatchEnabled

      public void setBatchEnabled(boolean batchEnabled)
      Enables or disables batched delivery, and -- unlike the field it replaces as a plain Lombok setter -- actually starts or stops the scheduled batch-send task to match (#432): disabling cancels it outright rather than leaving it running with nothing useful to flush; re-enabling starts a fresh one at the current intervalMs. A no-op when the requested value equals the current one.
      Parameters:
      batchEnabled - whether batched delivery should be active

    • sendBatch

      private void sendBatch()
      Sends the batched logs. When externalDrainMode is true, only drops entries that exceed the queue cap and does not send.

      Gate-2 finding (round 10): this method is ALSO reached directly, on the scheduler's own thread, by the periodically-firing batchSenderTask. Before this fix, that call was completely unsynchronized -- a live batchConfig.enabled: false update could hold batchModeLock across its own flushLogs() call while the scheduled task fired concurrently on the scheduler's thread and called this method too, both polling the SAME logQueue and both calling webSocketClient.sendMessage(...) independently -- interleaving queue polls between two frames with no ordering guarantee between the two sendMessage calls. Synchronizing the whole body on batchModeLock serializes the scheduled tick against setBatchEnabled(false)'s own flush, against flushLogs() however else it is reached (it now also acquires this same lock, see its own javadoc), and against sendLog's own dispatch decision (reentrant when this method is reached via addToBatch's size-threshold trigger, since that call chain already holds the lock). No new deadlock risk: this method never acquires any OTHER lock, so it can only ever be the innermost link in any lock chain that reaches it.

    • drainQueue

      public com.google.gson.JsonArray drainQueue(int maxItems)
      Drains log entries from the queue, returning them as a JsonArray. For external callers to use (e.g. ServerMonitorManager's batch_update).
      Parameters:
      maxItems - the maximum number of entries to take
      Returns:
      a JsonArray of log entries

    • holdUndelivered

      public void holdUndelivered(com.google.gson.JsonArray logs)
      Gives back a batch an external drain took with drainQueue(int) but could not send, so the next drainQueue(int) -- or this class's own sender -- sends it before any newer record (#486). ServerMonitorManager calls this when its batch_update frame carrying these records failed.
      Parameters:
      logs - the undelivered records, oldest first; ignored when null or empty
      Since:
      6.3.0

    • takePending

      public com.google.gson.JsonArray takePending()
      Removes and returns every record this transmitter has not sent -- replayed start-up records not yet delivered first, then the held batch, then the queue, oldest first -- so LogStreamManager can hand them to the transmitter that replaces this one when the panel reconnects on a new client (#486). Call it after shutdown(), whose flush sends what the old connection still can.
      Returns:
      the unsent records, oldest first; empty when there are none
      Since:
      6.3.0

    • adoptPending

      public void adoptPending(com.google.gson.JsonArray pending)
      Queues records a replaced transmitter could not send (takePending()) ahead of anything this one logs afterwards, so they reach the panel first and in order (#486). The queue's own bound applies: beyond MAX_QUEUE_SIZE the oldest are discarded and counted.
      Parameters:
      pending - the records, oldest first; ignored when null
      Since:
      6.3.0

    • setExternalDrainMode

      public void setExternalDrainMode(boolean enabled)
      Sets external drain mode. When enabled, sendBatch() no longer sends logs automatically; they are obtained externally via drainQueue() instead.
      Parameters:
      enabled - whether to enable external drain mode

    • isExternalDrainMode

      public boolean isExternalDrainMode()
      Checks whether external drain mode is active.

    • setExternalSizeThresholdCallback

      public void setExternalSizeThresholdCallback(Runnable callback)
      Sets the callback invoked when the queue reaches batchSize while external drain mode is active (Gate-2 finding, round 6). Pass null to clear it.
      Parameters:
      callback - a no-argument, non-blocking callback; called on whichever thread sendLog(String, String, String, Throwable) happened to run on

    • setExternalDrainCoordinationLock

      public void setExternalDrainCoordinationLock(Object lock)
      Sets the lock flushLogs() coordinates disable-time flushes against (Gate-2 finding, round 9). Pass null to clear it (flushLogs runs unsynchronized again).
      Parameters:
      lock - any object usable as a monitor; the SAME instance must be used by whatever external code also drains this transmitter's queue (see this field's own javadoc)

    • holdsExternalDrainLock

      private boolean holdsExternalDrainLock()
      Whether the current thread holds externalDrainCoordinationLock (ServerMonitorManager's logDrainLock) -- and so must not take batchModeLock (#584).

      The lock order. Every path that holds both locks takes batchModeLock first: flushLogs() explicitly, and sendLog(java.lang.String, java.lang.String, java.lang.String, java.lang.Throwable) through addToBatch(com.google.gson.JsonObject)'s size-threshold callback (ServerMonitorManager#drainLogsNow). A thread holding the drain lock that then asked for batchModeLock would invert that order and deadlock against either of them. The drain thread itself never calls into this class's locked methods -- drainQueue(int), holdUndelivered(com.google.gson.JsonArray) and the getters it uses take no lock -- so the only way it could reach batchModeLock is a line logged on that thread while it holds the drain lock, re-entering through SystemLogHandler. On Paper that includes anything printed to System.out/System.err: SysoutCatcher re-logs it through the calling plugin's logger.

      Why such a line is dropped from the stream. The framework's own lines on that thread are about the send in progress and are logged through PanelConnectionLog, which the stream never carries anyway; any other line still reaches the console, which is written by a different handler. Queuing or sending it without the lock instead would bypass the dispatch decision batchModeLock serializes against setBatchEnabled(boolean)'s flush.

      Returns:
      true when a drain lock is wired and the current thread holds it

    • getStackTrace

      private String getStackTrace(Throwable throwable)
      Gets the exception stack trace as a string.

    • determineLoggerName

      private String determineLoggerName(String source)
      Determines the logger name.

    • getDefaultServerId

      private static String getDefaultServerId()
      Gets the default server ID.

    • setLogTransmissionEnabled

      public void setLogTransmissionEnabled(boolean enabled)
      Enables/disables log transmission.

    • isLogTransmissionEnabled

      public boolean isLogTransmissionEnabled()
      Checks whether log transmission is enabled.

    • getQueueSize

      public int getQueueSize()
      Gets the current queue size.

    • flushLogs

      public void flushLogs()
      Immediately sends every log currently in the queue. Temporarily disables external drain mode to make sure the logs actually get sent.

      Gate-2 finding (round 10): now ALSO acquires batchModeLock as the OUTERMOST lock, before the external coordination lock below -- sendBatch() (which this method reaches via doFlushLogs()) is itself now synchronized on batchModeLock (see its own javadoc), so without this method also acquiring it first, a caller that had NOT already taken batchModeLock (e.g. shutdown(), or SystemLogHandler.flush()/close()) would acquire the coordination lock FIRST and only then try for batchModeLock inside sendBatch() -- the reverse of the order every other path in this class already establishes (batchModeLock before logDrainLock, see setExternalDrainCoordinationLock(Object)'s own javadoc), and a reverse-order acquisition on two different threads is exactly how a lock-ordering deadlock happens. Taking batchModeLock first here keeps every acquisition path in this class consistent with that one order, so this addition introduces no new deadlock risk -- verified by enumerating every lock-acquiring path in this class and ServerMonitorManager (see this plan's gate record).

    • doFlushLogs

      private void doFlushLogs()

    • heldSize

      private int heldSize()

    • shutdown

      public void shutdown()
      Shuts down the log transmitter.