Log Pipeline Optimization — Tuning Log Processing for Performance

// Optimized log processor with buffering
class OptimizedLogPipeline {
  constructor(options = {}) {
    this.buffer = [];
    this.batchSize = options.batchSize || 500;
    this.flushInterval = options.flushInterval || 1000;
    this.compressionLevel = options.compressionLevel || 6;
    this.processing = false;

    this.startFlushInterval();
  }

  async process(entry) {
    this.buffer.push(entry);

    if (this.buffer.length >= this.batchSize) {
      await this.flush();
    }
  }

  async flush() {
    if (this.buffer.length === 0 || this.processing) return;
    this.processing = true;

    const batch = this.buffer.splice(0, this.batchSize);

    try {
      // Compress batch
      const compressed = await this.compressBatch(batch);

      // Send to multiple destinations in parallel
      await Promise.all([
        this.sendToElasticsearch(compressed),
        this.sendToS3(batch),
        this.sendToMetrics(batch)
      ]);
    } catch (err) {
      console.error('Pipeline flush failed, buffering for retry:', err.message);
      this.buffer.unshift(...batch);
    } finally {
      this.processing = false;
    }
  }

  async compressBatch(batch) {
    const data = batch.map(e => JSON.stringify(e)).join('\n');
    return new Promise((resolve, reject) => {
      zlib.gzip(data, { level: this.compressionLevel }, (err, result) => {
        if (err) reject(err);
        else resolve(result);
      });
    });
  }

  async sendToElasticsearch(compressed) {
    // Bulk API with compressed body
    return axios.post('https://elastic:9200/_bulk', compressed, {
      headers: { 'Content-Encoding': 'gzip', 'Content-Type': 'application/x-ndjson' },
      timeout: 10000
    });
  }

  async sendToS3(batch) {
    const key = `logs/${new Date().toISOString().slice(0, 10)}/${uuidv4()}.json.gz`;
    const compressed = await this.compressBatch(batch);
    return s3.putObject({
      Bucket: 'scanapp-logs',
      Key: key,
      Body: compressed,
      ContentEncoding: 'gzip'
    });
  }

  startFlushInterval() {
    setInterval(() => this.flush(), this.flushInterval);
  }

  // Backpressure handling
  async checkBackpressure() {
    const pressure = this.buffer.length / this.batchSize;
    if (pressure > 10) {
      // Drop debug logs under backpressure
      this.buffer = this.buffer.filter(e => e.level !== 'DEBUG');
    }
    return pressure;
  }
}