|
| 1 | +/** |
| 2 | + * @fileoverview The traces queue, cunsume traces and sends in batches |
| 3 | + */ |
| 4 | +const EventEmitter = require('events'); |
| 5 | +const axios = require('axios'); |
| 6 | +const https = require('https'); |
| 7 | +const http = require('http'); |
| 8 | +const utils = require('../src/utils.js'); |
| 9 | +const config = require('./config.js'); |
| 10 | + |
| 11 | + |
| 12 | +/** |
| 13 | + * Session for the post requests to the collector |
| 14 | + */ |
| 15 | +const session = axios.create({ |
| 16 | + headers: { Authorization: `Bearer ${config.getConfig().token}` }, |
| 17 | + timeout: config.getConfig().sendTimeout, |
| 18 | + httpAgent: new http.Agent({ keepAlive: true }), |
| 19 | + httpsAgent: new https.Agent({ keepAlive: true }), |
| 20 | +}); |
| 21 | + |
| 22 | +/** |
| 23 | + * Post given batch to epsagon's infrastructure. |
| 24 | + * @param {*} batchObject The batch data to send. |
| 25 | + * @returns {Promise} a promise that is resolved after the batch is posted. |
| 26 | + */ |
| 27 | +async function postBatch(batchObject) { |
| 28 | + utils.debugLog(`[QUEUE] Posting batch to ${config.getConfig().traceCollectorURL}...`); |
| 29 | + const cancelTokenSource = axios.CancelToken.source(); |
| 30 | + const handle = setTimeout(() => { |
| 31 | + cancelTokenSource.cancel('Timeout sending batch!'); |
| 32 | + }, config.getConfig().sendTimeout); |
| 33 | + |
| 34 | + return session.post( |
| 35 | + config.getConfig().traceCollectorURL, |
| 36 | + batchObject, |
| 37 | + { |
| 38 | + cancelToken: cancelTokenSource.token, |
| 39 | + } |
| 40 | + ).then((res) => { |
| 41 | + clearTimeout(handle); |
| 42 | + utils.debugLog('[QUEUE] Batch posted!'); |
| 43 | + return res; |
| 44 | + }).catch((err) => { |
| 45 | + clearTimeout(handle); |
| 46 | + if (err.config && err.config.data) { |
| 47 | + utils.debugLog(`[QUEUE] Error sending trace. Batch size: ${err.config.data.length}`); |
| 48 | + } else { |
| 49 | + utils.debugLog(`[QUEUE] Error sending trace. Error: ${err}`); |
| 50 | + } |
| 51 | + utils.debugLog(`[QUEUE] ${err ? err.stack : err}`); |
| 52 | + return err; |
| 53 | + }); |
| 54 | +} |
| 55 | + |
| 56 | +/** |
| 57 | + * The trace queue class |
| 58 | + * @param {function} batchSender function to send batch traces |
| 59 | + */ |
| 60 | +class TraceQueue extends EventEmitter.EventEmitter { |
| 61 | + /** |
| 62 | + * EventEmitter class |
| 63 | + */ |
| 64 | + constructor() { |
| 65 | + super(); |
| 66 | + this.batchSender = postBatch; |
| 67 | + this.initQueue(); |
| 68 | + } |
| 69 | + |
| 70 | + /** |
| 71 | + * Update the queue config |
| 72 | + */ |
| 73 | + updateConfig() { |
| 74 | + this.maxBatchSizeBytes = config.getConfig().maxBatchSizeBytes; |
| 75 | + this.batchSize = config.getConfig().batchSize; |
| 76 | + this.maxQueueSizeBytes = config.getConfig().maxQueueSizeBytes; |
| 77 | + } |
| 78 | + |
| 79 | + /** |
| 80 | + * Init queue event listners |
| 81 | + */ |
| 82 | + initQueue() { |
| 83 | + this.updateConfig(); |
| 84 | + this.removeAllListeners(); |
| 85 | + this.flush(); |
| 86 | + this.on('traceQueued', function traceQueued() { |
| 87 | + if (this.byteSizeLimitReached()) { |
| 88 | + utils.debugLog(`[QUEUE] Queue Byte size reached ${this.currentByteSize} Bytes, releasing batch...`); |
| 89 | + this.emit('releaseRequest', Math.max(this.currentSize - 1, 1)); |
| 90 | + } else if (this.batchSizeReached()) { |
| 91 | + utils.debugLog(`[QUEUE] Queue size reached ${this.currentSize}, releasing batch... `); |
| 92 | + this.emit('releaseRequest'); |
| 93 | + } |
| 94 | + return this; |
| 95 | + }); |
| 96 | + |
| 97 | + this.on('releaseRequest', function releaseRequest(count = this.batchSize) { |
| 98 | + try { |
| 99 | + const batch = this.queue.splice(0, count); |
| 100 | + utils.debugLog('[QUEUE] Releasing batch...'); |
| 101 | + this.subtractFromCurrentByteSize(batch); |
| 102 | + this.emit('batchReleased', batch); |
| 103 | + } catch (err) { |
| 104 | + utils.debugLog('[QUEUE] Failed releasing batch!'); |
| 105 | + utils.debugLog(`[QUEUE] ${err}`); |
| 106 | + } |
| 107 | + return this; |
| 108 | + }); |
| 109 | + |
| 110 | + this.on('batchReleased', async function batchReleased(batch) { |
| 111 | + utils.debugLog('[QUEUE] Sending batch...'); |
| 112 | + const batchJSON = batch.map(trace => trace.traceJSON); |
| 113 | + this.batchSender(batchJSON); |
| 114 | + }); |
| 115 | + process.on('exit', function releaseAndClearQueue() { |
| 116 | + this.emit('releaseRequest'); |
| 117 | + this.removeAllListeners(); |
| 118 | + }); |
| 119 | + } |
| 120 | + |
| 121 | + /** |
| 122 | + * Push trace to queue, emit event, and check if queue max queue length reached, |
| 123 | + * if it does, send batch. |
| 124 | + * @param {object} traceJson Trace JSON |
| 125 | + * @returns {TraceQueue} This trace queue |
| 126 | + */ |
| 127 | + push(traceJson) { |
| 128 | + try { |
| 129 | + if (this.currentByteSize >= this.maxQueueSizeBytes) { |
| 130 | + utils.debugLog(`[QUEUE] Discardig trace, queue size reached max size of ${this.currentByteSize} Bytes`); |
| 131 | + return this; |
| 132 | + } |
| 133 | + const timestamp = Date.now(); |
| 134 | + const json = traceJson; |
| 135 | + const string = JSON.stringify(json); |
| 136 | + const byteLength = string.length; |
| 137 | + // eslint-disable-next-line object-curly-newline |
| 138 | + const trace = { json, string, byteLength, timestamp }; |
| 139 | + this.queue.push(trace); |
| 140 | + this.addToCurrentByteSize([trace]); |
| 141 | + utils.debugLog(`[QUEUE] Trace size ${byteLength} Bytes pushed to queue`); |
| 142 | + utils.debugLog(`[QUEUE] Queue size: ${this.currentSize} traces, total size of ${this.currentByteSize} Bytes`); |
| 143 | + this.emit('traceQueued', trace); |
| 144 | + } catch (err) { |
| 145 | + utils.debugLog(`[QUEUE] Failed pushing trace to queue: ${err}`); |
| 146 | + } |
| 147 | + return this; |
| 148 | + } |
| 149 | + |
| 150 | + /** |
| 151 | + * add given trace byte size to total byte size |
| 152 | + * @param {Array} traces Trace object array |
| 153 | + */ |
| 154 | + addToCurrentByteSize(traces) { |
| 155 | + traces.forEach((trace) => { |
| 156 | + this.currentByteSize += trace.byteLength; |
| 157 | + }); |
| 158 | + } |
| 159 | + |
| 160 | + /** |
| 161 | + * subtract given trace byte size to total byte size |
| 162 | + * @param {Array} traces Trace object array |
| 163 | + */ |
| 164 | + subtractFromCurrentByteSize(traces) { |
| 165 | + traces.forEach((trace) => { |
| 166 | + this.currentByteSize -= trace.byteLength; |
| 167 | + this.currentByteSize = Math.max(this.currentByteSize, 0); |
| 168 | + }); |
| 169 | + } |
| 170 | + |
| 171 | + /** |
| 172 | + * Queue size getter |
| 173 | + * @returns {Number} Queue length |
| 174 | + */ |
| 175 | + get currentSize() { |
| 176 | + return this.queue.length; |
| 177 | + } |
| 178 | + |
| 179 | + |
| 180 | + /** |
| 181 | + * Checks if queue size reached batch size |
| 182 | + * @returns {Boolean} Indicator for if current queue size is larger than batch size definition |
| 183 | + */ |
| 184 | + batchSizeReached() { |
| 185 | + return this.currentSize >= this.batchSize; |
| 186 | + } |
| 187 | + |
| 188 | + /** |
| 189 | + * Checks if queue byte size reached its limit |
| 190 | + * @returns {Boolean} Indicator for if current queue byte size is larger than byte size definition |
| 191 | + */ |
| 192 | + byteSizeLimitReached() { |
| 193 | + return this.currentByteSize >= this.maxBatchSizeBytes; |
| 194 | + } |
| 195 | + |
| 196 | + /** |
| 197 | + * Flush queue |
| 198 | + */ |
| 199 | + flush() { |
| 200 | + this.queue = []; |
| 201 | + this.currentByteSize = 0; |
| 202 | + } |
| 203 | +} |
| 204 | + |
| 205 | +const traceQueue = new TraceQueue(); |
| 206 | + |
| 207 | +module.exports.getInstance = () => traceQueue; |
0 commit comments