11import { Heap } from 'heap-js' ;
22import { SdkComponent } from '@temporalio/common' ;
33import { native } from '@temporalio/core-bridge' ;
4- import { DefaultLogger , LogEntry , Logger , LogTimestamp } from './logger' ;
4+ import { DefaultLogger , FlushableLogger , LogEntry , Logger , LogTimestamp } from './logger' ;
55
66/**
77 * A log collector that accepts log entries either through the TS `Logger` interface (e.g. used by
@@ -25,10 +25,60 @@ export class NativeLogCollector {
2525
2626 protected buffer = new Heap < LogEntry > ( ( a , b ) => Number ( a . timestampNanos - b . timestampNanos ) ) ;
2727
28+ /**
29+ * A timer that periodically flushes the buffer to the downstream logger.
30+ */
31+ protected flushIntervalTimer : NodeJS . Timeout ;
32+
33+ /**
34+ * The minimum time an entry should be buffered before getting flushed.
35+ *
36+ * Increasing this value allows the buffer to do a better job of correctly reordering messages
37+ * emitted from different sources (notably from Workflow executions through Sinks, and from Core)
38+ * based on their absolute timestamps, but also increases latency of logs.
39+ *
40+ * The minimum buffer time requirement only applies as long as the buffer is not full. Once the
41+ * buffer reaches its maximum size, older messages are unconditionally flushed, to prevent
42+ * unbounded growth of the buffer.
43+ *
44+ * TODO(JWH): Is 100ms a reasonable compromise? That might seem a little high on latency, but to
45+ * be useful, that value needs to exceed the time it typically takes to process
46+ * Workflow Activations, let's say above the expected P90, but that's highly variable
47+ * across our user base, and we don't really have field data anyway.
48+ * We can revisit depending on user feedback.
49+ */
50+ protected readonly minBufferTimeMs = 100 ;
51+
52+ /**
53+ * Interval between flush passes checking for expired messages.
54+ *
55+ * This really is redundant, since Core itself is expected to flush its buffer every 10 ms, and
56+ * we're checking for expired messages when it does. However, Core will only flush if it has
57+ * accumulated at least one message; when Core's log level is set to WARN or higher, it may be
58+ * many seconds, and even minutes, between Core's log messages, resulting in very rare flush
59+ * from that end, which cause considerable delay on flushing log messages from other sources.
60+ */
61+ protected readonly flushPassIntervalMs = 100 ;
62+
63+ /**
64+ * The maximum number of log messages to buffer before flushing.
65+ *
66+ * When the buffer reaches this limit, older messages are unconditionally flushed (i.e. without
67+ * regard to the minimum buffer time requirement), to prevent unbounded growth of the buffer.
68+ */
69+ protected readonly maxBufferSize = 2000 ;
70+
2871 constructor ( downstream : Logger ) {
29- this . logger = new DefaultLogger ( 'TRACE' , ( entry ) => this . buffer . add ( entry ) ) ;
72+ this . logger = new DefaultLogger ( 'TRACE' , this . appendOne . bind ( this ) ) ;
73+ ( this . logger as FlushableLogger ) . flush = this . flush . bind ( this ) ;
74+ ( this . logger as FlushableLogger ) . close = this . close . bind ( this ) ;
75+
3076 this . downstream = downstream ;
3177 this . receive = this . receive . bind ( this ) ;
78+
79+ // Flush the buffer every so often.
80+ // Unref'ed so that it doesn't prevent the process from exiting.
81+ this . flushIntervalTimer = setInterval ( this . flushExpired . bind ( this ) , this . flushPassIntervalMs ) . unref ( ) ;
3282 }
3383
3484 /**
@@ -44,14 +94,20 @@ export class NativeLogCollector {
4494 this . buffer . add ( log ) ;
4595 }
4696 }
47- this . flush ( ) ;
97+ this . flushUnconditionally ( ) ;
98+ this . flushExpired ( ) ;
4899 } catch ( _e ) {
49100 // We're not allowed to throw from here, and conversion errors have already been handled in
50101 // convertFromNativeLogEntry(), so an error at this point almost certainly indicates a problem
51102 // with the downstream logger. Just swallow it, there's really nothing else we can do.
52103 }
53104 }
54105
106+ private appendOne ( entry : LogEntry ) : void {
107+ this . buffer . add ( entry ) ;
108+ this . flushUnconditionally ( ) ;
109+ }
110+
55111 private convertFromNativeLogEntry ( entry : native . JsonString < native . LogEntry > ) : LogEntry | undefined {
56112 try {
57113 const log = JSON . parse ( entry ) as native . LogEntry ;
@@ -78,15 +134,49 @@ export class NativeLogCollector {
78134 }
79135
80136 /**
81- * Flush all buffered logs into the logger supplied to the constructor/
137+ * Flush messages that have exceeded their required minimal buffering time.
82138 */
83- flush ( ) : void {
84- for ( const entry of this . buffer ) {
139+ private flushExpired ( ) : void {
140+ const threadholdTimeNanos = BigInt ( Date . now ( ) - this . minBufferTimeMs ) * 1_000_000n ;
141+ for ( ; ; ) {
142+ const entry = this . buffer . peek ( ) ;
143+ if ( ! entry || entry . timestampNanos > threadholdTimeNanos ) break ;
144+ this . buffer . pop ( ) ;
145+
85146 this . downstream . log ( entry . level , entry . message , {
86147 [ LogTimestamp ] : entry . timestampNanos ,
87148 ...entry . meta ,
88149 } ) ;
89150 }
90- this . buffer . clear ( ) ;
151+ }
152+
153+ /**
154+ * Flush messages without regard to the time threshold, up to a given number of messages.
155+ *
156+ * If no limit is provided, flushes messages in excess to the maximum buffer size.
157+ */
158+ private flushUnconditionally ( maxFlushCount ?: number ) : void {
159+ if ( maxFlushCount === undefined ) {
160+ maxFlushCount = this . buffer . size ( ) - this . maxBufferSize ;
161+ }
162+
163+ while ( maxFlushCount -- > 0 ) {
164+ const entry = this . buffer . pop ( ) ;
165+ if ( ! entry ) break ;
166+
167+ this . downstream . log ( entry . level , entry . message , {
168+ [ LogTimestamp ] : entry . timestampNanos ,
169+ ...entry . meta ,
170+ } ) ;
171+ }
172+ }
173+
174+ public flush ( ) : void {
175+ this . flushUnconditionally ( Number . MAX_SAFE_INTEGER ) ;
176+ }
177+
178+ public close ( ) : void {
179+ this . flush ( ) ;
180+ clearInterval ( this . flushIntervalTimer ) ;
91181 }
92182}
0 commit comments