feat!: add maxBufferSize option + update batching API

Author: Julien-R44

README.md

@@ -20,9 +20,6 @@ import type { LokiOptions } from 'pino-loki'
 const transport = pino.transport<LokiOptions>({
   target: "pino-loki",
   options: {
-    batching: true,
-    interval: 5,
-
     host: 'https://my-loki-instance:3100',
     basicAuth: {
       username: "username",
@@ -102,11 +99,44 @@ If false, errors when sending logs to Loki will be displayed in the console. Def
 
 #### `batching`
 
-Should logs be sent in batch mode. Defaults to `true`.
+Batching configuration. When enabled, logs are accumulated in a buffer and sent to Loki at regular intervals, reducing the number of HTTP requests. Batching is **enabled by default**.
+
+```ts
+// Batching enabled with default options (interval: 5s, maxBufferSize: 10000)
+pinoLoki({ host: '...' })
+
+// Batching with custom options
+pinoLoki({
+  host: '...',
+  batching: {
+    interval: 2,        // Send logs every 2 seconds
+    maxBufferSize: 5000 // Keep max 5000 logs in buffer
+  }
+})
+
+// Batching disabled - logs sent immediately
+pinoLoki({ host: '...', batching: false })
+```
+
+##### `batching.interval`
+
+The interval at which batched logs are sent, in seconds. Defaults to `5`.
 
-#### `interval`
+##### `batching.maxBufferSize`
 
-The interval at which batched logs are sent in seconds. Defaults to `5`.
+Maximum number of logs to keep in the buffer. When the buffer is full, **oldest logs are dropped** (FIFO) to make room for new ones. Defaults to `10000`.
+
+This prevents memory issues (OOM) if Loki becomes unavailable - without this limit, the buffer would grow indefinitely. Set to `0` for unlimited buffer (probably not really recommended).
+
+```ts
+pinoLoki({
+  host: '...',
+  batching: {
+    interval: 10,
+    maxBufferSize: 50000
+  }
+})
+```
 
 #### `replaceTimestamp` 
 
@@ -195,23 +225,23 @@ node foo | pino-loki --hostname=http://hostname:3100
 ```
 $ pino-loki -h
 Options:
-  -V, --version                  output the version number
-  -u, --user <user>              Loki username
-  -p, --password <password>      Loki password
-  --hostname <hostname>          URL for Loki
-  --endpoint <endpoint>          Path to the Loki push API
-  --headers <headers>            Headers to be sent to Loki (Example: "X-Scope-OrgID=your-id,another-header=another-value")
-  -b, --batch                    Should logs be sent in batch mode
-  -i, --interval <interval>      The interval at which batched logs are sent in seconds
-  -t, --timeout <timeout>        Timeout for request to Loki
-  -s, --silenceErrors            If false, errors will be displayed in the console
-  -r, --replaceTimestamp         Replace pino logs timestamps with Date.now()
-  -l, --labels <label>           Additional labels to be added to all Loki logs
-  -a, --convertArrays            If true, arrays will be converted to objects
-  -pl, --propsLabels <labels>    Fields in log line to convert to Loki labels (comma separated values)
-  --structuredMetaKey <key>      Key in the pino log object that contains structured metadata
-  --no-stdout                    Disable output to stdout
-  -h, --help                     display help for command
+  -v, --version                          Print version number and exit
+  -u, --user <user>                      Loki username
+  -p, --password <password>              Loki password
+  --hostname <hostname>                  URL for Loki (default: http://localhost:3100)
+  --endpoint <endpoint>                  Path to the Loki push API (default: /loki/api/v1/push)
+  --headers <headers>                    Headers to be sent to Loki (Example: "X-Scope-OrgID=your-id,another=value")
+  -b, --batching                         Should logs be sent in batch mode (default: true)
+  -i, --batching-interval <interval>     The interval at which batched logs are sent in seconds (default: 5)
+  --batching-max-buffer-size <size>      Maximum number of logs to buffer (default: 10000, 0 for unlimited)
+  -t, --timeout <timeout>                Timeout for request to Loki in ms (default: 2000)
+  -s, --silenceErrors                    If set, errors will not be displayed in the console
+  -r, --replaceTimestamp                 Replace pino logs timestamps with Date.now()
+  -l, --labels <label>                   Additional labels to be added to all Loki logs (JSON)
+  --convertArrays                        If set, arrays will be converted to objects
+  --propsLabels <labels>                 Fields in log line to convert to Loki labels (comma separated)
+  --structuredMetaKey <key>              Key in the pino log object that contains structured metadata
+  -h, --help                             Print this help message and exit
 ```
 
 ## Examples
@@ -270,7 +300,13 @@ And you should be good to go! You can check our [full example](./examples/adonis
 Out-of-order Loki errors can occur due to the asynchronous nature of Pino. The fix to this is to allow for out-of-order logs in the Loki configuration. The reason why Loki doesn't have this enabled by default is because Promtail accounts for ordering constraints, however the same issue can also happen with promtail in high-load or when working with distributed networks.
 
 ## Dropped logs
-If any network issues occur, the logs can be dropped. The recommendation is therefore to implement a failover solution, this will vary greatly from system to system.
+
+Logs can be dropped in two scenarios:
+
+1. **Network issues**: If Loki is unreachable, logs in the current batch will be lost.
+2. **Buffer overflow**: When batching is enabled and the buffer reaches `maxBufferSize` (default: 10,000), the oldest logs are dropped to make room for new ones. This prevents memory exhaustion if Loki becomes unavailable for an extended period.
+
+For critical applications, consider implementing a failover solution or adjusting `maxBufferSize` based on your memory constraints and acceptable data loss.
 
 ## Node v18+ Required
 As the pino-loki library uses the native Node fetch, any consumer must be using a version of Node greater than v18.0.0.

examples/batching.ts

@@ -13,8 +13,7 @@ const transport = pino.transport<LokiOptions>({
   target: '../dist/index.mjs',
 
   options: {
-    batching: true,
-    interval: 2,
+    batching: { interval: 2 },
 
     // These labels will be added to every log
     labels: { application: 'MY-APP' },

src/cli/args.ts

@@ -34,18 +34,23 @@ export const options = {
     default: '/loki/api/v1/push',
     help: 'Path to the Loki push API',
   },
-  'batch': {
+  'batching': {
     type: 'boolean',
     default: true,
     short: 'b',
     help: 'Should logs be sent in batch mode',
   },
-  'interval': {
+  'batching-interval': {
     type: 'string',
     default: '5',
     short: 'i',
     help: 'The interval at which batched logs are sent in seconds',
   },
+  'batching-max-buffer-size': {
+    type: 'string',
+    default: '10000',
+    help: 'Maximum number of logs to buffer (0 for unlimited)',
+  },
   'timeout': { type: 'string', default: '2000', short: 't', help: 'Timeout for request to Loki' },
   'silenceErrors': {
     type: 'boolean',

src/cli/index.ts

@@ -50,8 +50,14 @@ export const createPinoLokiConfigFromArgs = () => {
     endpoint: values.endpoint,
     timeout: values.timeout ? Number(values.timeout) : undefined,
     silenceErrors: values.silenceErrors,
-    batching: values.batch,
-    interval: values.interval ? Number(values.interval) : undefined,
+    batching: values.batching === false
+      ? false
+      : {
+          interval: values['batching-interval'] ? Number(values['batching-interval']) : undefined,
+          maxBufferSize: values['batching-max-buffer-size']
+            ? Number(values['batching-max-buffer-size'])
+            : undefined,
+        },
     replaceTimestamp: values.replaceTimestamp,
     labels: values.labels ? JSON.parse(values.labels) : undefined,
     propsToLabels: propsLabels,

src/index.ts

@@ -3,7 +3,23 @@ import abstractTransportBuild from 'pino-abstract-transport'
 import debug from './debug.ts'
 import { LogPusher } from './log_pusher.ts'
 import { LokiLogLevel } from './constants.ts'
-import type { PinoLog, LokiOptions } from './types.ts'
+import type { PinoLog, LokiOptions, BatchingOptions } from './types.ts'
+
+interface ResolvedBatching {
+  enabled: boolean
+  interval: number
+  maxBufferSize: number
+}
+
+function resolveBatching(batching: LokiOptions['batching']): ResolvedBatching {
+  if (batching === false) return { enabled: false, interval: 5, maxBufferSize: 10_000 }
+
+  return {
+    enabled: true,
+    interval: batching?.interval ?? 5,
+    maxBufferSize: batching?.maxBufferSize ?? 10_000,
+  }
+}
 
 /**
  * Resolves the options for the Pino Loki transport
@@ -14,8 +30,7 @@ function resolveOptions(options: LokiOptions) {
     endpoint: options.endpoint ?? 'loki/api/v1/push',
     timeout: options.timeout ?? 30_000,
     silenceErrors: options.silenceErrors ?? false,
-    batching: options.batching ?? true,
-    interval: options.interval ?? 5,
+    batching: resolveBatching(options.batching),
     replaceTimestamp: options.replaceTimestamp ?? false,
     propsToLabels: options.propsToLabels ?? [],
     convertArrays: options.convertArrays ?? false,
@@ -36,21 +51,23 @@ function pinoLoki(userOptions: LokiOptions) {
 
   return abstractTransportBuild(
     async (source) => {
-      if (options.batching) {
+      if (options.batching.enabled) {
         batchInterval = setInterval(async () => {
           debug(`Batch interval reached, sending ${pinoLogBuffer.length} logs to Loki`)
 
-          if (pinoLogBuffer.length === 0) {
-            return
-          }
+          if (pinoLogBuffer.length === 0) return
 
           logPusher.push(pinoLogBuffer)
           pinoLogBuffer = []
-        }, options.interval! * 1000)
+        }, options.batching.interval * 1000)
       }
 
       for await (const obj of source) {
-        if (options.batching) {
+        if (options.batching.enabled) {
+          if (options.batching.maxBufferSize > 0 && pinoLogBuffer.length >= options.batching.maxBufferSize) {
+            const dropped = pinoLogBuffer.shift()
+            debug(`[PinoLoki] Buffer full, dropping oldest log: ${JSON.stringify(dropped)}`)
+          }
           pinoLogBuffer.push(obj)
           continue
         }
@@ -64,7 +81,7 @@ function pinoLoki(userOptions: LokiOptions) {
        * and clear the interval
        */
       async close() {
-        if (options.batching) {
+        if (options.batching.enabled) {
           clearInterval(batchInterval!)
           await logPusher.push(pinoLogBuffer)
         }
@@ -74,4 +91,4 @@ function pinoLoki(userOptions: LokiOptions) {
 }
 
 export default pinoLoki
-export { LokiLogLevel, pinoLoki, type LokiOptions, type PinoLog }
+export { LokiLogLevel, pinoLoki, type LokiOptions, type PinoLog, type BatchingOptions }

src/types.ts

@@ -28,6 +28,26 @@ export interface PinoLog {
   [key: string]: any
 }
 
+/**
+ * Batching configuration options
+ */
+export interface BatchingOptions {
+  /**
+   * The interval at which batched logs are sent in seconds
+   *
+   * @default 5
+   */
+  interval?: number
+
+  /**
+   * Maximum number of logs to buffer before dropping the oldest ones.
+   * Set to 0 for unlimited buffer (not recommended).
+   *
+   * @default 10_000
+   */
+  maxBufferSize?: number
+}
+
 /**
  * Options for the Pino-Loki transport
  */
@@ -59,18 +79,11 @@ export interface LokiOptions {
   silenceErrors?: boolean
 
   /**
-   * Should logs be sent in batch mode
-   *
-   * @default true
-   */
-  batching?: boolean
-
-  /**
-   * The interval at which batched logs are sent in seconds
+   * Batching configuration. Set to `false` to disable batching entirely.
    *
-   * @default 5
+   * @default { enabled: true, interval: 5, maxBufferSize: 10_000 }
    */
-  interval?: number
+  batching?: false | BatchingOptions
 
   /**
    * Replace pino logs timestamps with Date.now()

tests/integration/loki.spec.ts

@@ -80,8 +80,7 @@ test.group('Loki integration', () => {
       { level: 'info' },
       pinoLoki({
         ...credentials,
-        batching: true,
-        interval: 1,
+        batching: { interval: 1 },
         labels: { application },
       }),
     )
@@ -107,8 +106,7 @@ test.group('Loki integration', () => {
       target: '../../dist/index.mjs',
       options: {
         ...credentials,
-        batching: true,
-        interval: 10,
+        batching: { interval: 10 },
         labels: { application },
       },
     })