<commit_before>'use strict'

var afterAll = require('after-all')
var stackman = require('stackman')({ filter: '/node_modules/opbeat/' })
var Transaction = require('./transaction')
var parsers = require('../parsers')
var request = require('../request')

var MAX_QUEUE_SIZE = Infinity
var MAX_SEND_DELAY = 60000

var Instrumentation = module.exports = function (client) {
  if (!(this instanceof Instrumentation)) return new Instrumentation(client)
  this._client = client
  this._queue = []
}

Instrumentation.prototype.add = function (transaction) {
  if (this._client.active && this._client._ff_instrument) {
    this._client.logger.trace('[%s] pushing transaction to queue', transaction._uuid)
    this._queue.push(transaction)
  } else {
    this._client.logger.trace('[%s] ignoring transaction', transaction._uuid)
  }
  if (this._queue.length >= MAX_QUEUE_SIZE) this._send()
  else if (this._queue.length && !this._timeout) this._queueSend()
}

Instrumentation.prototype.startTransaction = function (name, type, result) {
  return new Transaction(this._client, name, type, result)
}

Instrumentation.prototype._send = function () {
  this._client.logger.trace('flushing transaction queue')
  clearTimeout(this._timeout)
  this._timeout = null
  this._flush(function (result) {
    request.transactions(this._client, result)
  }.bind(this))
}

Instrumentation.prototype._queueSend = function () {
  this._client.logger.trace('setting timer to flush transaction queue')
  this._timeout = setTimeout(this._send.bind(this), MAX_SEND_DELAY)
  this._timeout.unref()
}

Instrumentation.prototype._flush = function (cb) {
  var transactions = groupTransactions(this._queue)
  var traces = [].concat.apply([], this._queue.map(function (trans) {
    return trans.traces
  }))
  var groups = groupTraces(traces)
  var raw = rawTransactions(this._queue)
  this._queue = []

  addStackTracesToTraceGroups(groups, traces, function () {
    cb({
      transactions: transactions,
      traces: {
        groups: groups,
        raw: raw
      }
    })
  })
}

function groupTransactions (transactions) {
  var groups = groupByMinute(transactions, transactionGroupingKey)

  return Object.keys(groups).map(function (key) {
    var trans = groups[key][0]
    var durations = groups[key].map(function (trans) {
      return trans.duration()
    })
    return {
      transaction: trans.name,
      result: trans.result,
      kind: trans.type,
      timestamp: groupingTs(trans._start).toISOString(),
      durations: durations
    }
  })
}

function rawTransactions (transactions) {
  return transactions.map(function (trans) {
    return [trans.duration()].concat(rawTraces(trans.traces))
  })
}

function rawTraces (traces) {
  return traces.map(function (trace) {
    return [trace._groupIndex, trace.startTime(), trace.duration()]
  })
}

function addStackTracesToTraceGroups (groups, traces, cb) {
  var processed = []
  var next = afterAll(cb)
  traces.forEach(function (trace) {
    var index = trace._groupIndex
    if (~processed.indexOf(index)) return
    processed.push(index)
    var done = next()
    traceFrames(trace, function (frames) {
      groups[index].extra._frames = frames
      done()
    })
  })
}

function groupTraces (traces) {
  var groups = groupByMinute(traces, traceGroupingKey)

  return Object.keys(groups).map(function (key, index) {
    var trace = groups[key][0]
    groups[key].forEach(function (trace) {
      trace._groupIndex = index
    })
    return {
      transaction: trace.transaction.name,
      signature: trace.signature,
      kind: trace.type,
      timestamp: groupingTs(trace._start).toISOString(),
      parents: trace.ancestors(),
      extra: trace.extra || {}
    }
  })
}

function traceFrames (trace, cb) {
  if (trace._stackObj.frames) return process.nextTick(cb.bind(null, trace._stackObj.frames))
  stackman(trace._stackObj.err, function (stack) {
    var frames = stack.frames.map(parsers.parseCallsite)
    trace._stackObj.frames = frames
    cb(frames)
  })
}

function groupByMinute (arr, grouper) {
  var groups = {}

  arr.forEach(function (obj) {
    var key = grouper(obj)
    if (key in groups) groups[key].push(obj)
    else groups[key] = [obj]
  })

  return groups
}

function groupingTs (ts) {
  return new Date(ts.getFullYear(), ts.getMonth(), ts.getDate(), ts.getHours(), ts.getMinutes())
}

function transactionGroupingKey (trans) {
  return groupingTs(trans._start).getTime() + '|' + trans.name + '|' + trans.result + '|' + trans.type
}

function traceGroupingKey (trace) {
  var ancestors = trace.ancestors().map(function (trace) { return trace.signature }).join('|')
  return groupingTs(trace._start).getTime() + '|' + trace.transaction.name + '|' + ancestors + '|' + trace.signature
}<commit_msg>Group traces based on transaction start time<commit_after>'use strict'

var afterAll = require('after-all')
var stackman = require('stackman')({ filter: '/node_modules/opbeat/' })
var Transaction = require('./transaction')
var parsers = require('../parsers')
var request = require('../request')

var MAX_QUEUE_SIZE = Infinity
var MAX_SEND_DELAY = 60000

var Instrumentation = module.exports = function (client) {
  if (!(this instanceof Instrumentation)) return new Instrumentation(client)
  this._client = client
  this._queue = []
}

Instrumentation.prototype.add = function (transaction) {
  if (this._client.active && this._client._ff_instrument) {
    this._client.logger.trace('[%s] pushing transaction to queue', transaction._uuid)
    this._queue.push(transaction)
  } else {
    this._client.logger.trace('[%s] ignoring transaction', transaction._uuid)
  }
  if (this._queue.length >= MAX_QUEUE_SIZE) this._send()
  else if (this._queue.length && !this._timeout) this._queueSend()
}

Instrumentation.prototype.startTransaction = function (name, type, result) {
  return new Transaction(this._client, name, type, result)
}

Instrumentation.prototype._send = function () {
  this._client.logger.trace('flushing transaction queue')
  clearTimeout(this._timeout)
  this._timeout = null
  this._flush(function (result) {
    request.transactions(this._client, result)
  }.bind(this))
}

Instrumentation.prototype._queueSend = function () {
  this._client.logger.trace('setting timer to flush transaction queue')
  this._timeout = setTimeout(this._send.bind(this), MAX_SEND_DELAY)
  this._timeout.unref()
}

Instrumentation.prototype._flush = function (cb) {
  var transactions = groupTransactions(this._queue)
  var traces = [].concat.apply([], this._queue.map(function (trans) {
    return trans.traces
  }))
  var groups = groupTraces(traces)
  var raw = rawTransactions(this._queue)
  this._queue = []

  addStackTracesToTraceGroups(groups, traces, function () {
    cb({
      transactions: transactions,
      traces: {
        groups: groups,
        raw: raw
      }
    })
  })
}

function groupTransactions (transactions) {
  var groups = groupByMinute(transactions, transactionGroupingKey)

  return Object.keys(groups).map(function (key) {
    var trans = groups[key][0]
    var durations = groups[key].map(function (trans) {
      return trans.duration()
    })
    return {
      transaction: trans.name,
      result: trans.result,
      kind: trans.type,
      timestamp: groupingTs(trans._start).toISOString(),
      durations: durations
    }
  })
}

function rawTransactions (transactions) {
  return transactions.map(function (trans) {
    return [trans.duration()].concat(rawTraces(trans.traces))
  })
}

function rawTraces (traces) {
  return traces.map(function (trace) {
    return [trace._groupIndex, trace.startTime(), trace.duration()]
  })
}

function addStackTracesToTraceGroups (groups, traces, cb) {
  var processed = []
  var next = afterAll(cb)
  traces.forEach(function (trace) {
    var index = trace._groupIndex
    if (~processed.indexOf(index)) return
    processed.push(index)
    var done = next()
    traceFrames(trace, function (frames) {
      groups[index].extra._frames = frames
      done()
    })
  })
}

function groupTraces (traces) {
  var groups = groupByMinute(traces, traceGroupingKey)

  return Object.keys(groups).map(function (key, index) {
    var trace = groups[key][0]
    groups[key].forEach(function (trace) {
      trace._groupIndex = index
    })
    return {
      transaction: trace.transaction.name,
      signature: trace.signature,
      kind: trace.type,
      timestamp: groupingTs(trace._start).toISOString(),
      parents: trace.ancestors(),
      extra: trace.extra || {}
    }
  })
}

function traceFrames (trace, cb) {
  if (trace._stackObj.frames) return process.nextTick(cb.bind(null, trace._stackObj.frames))
  stackman(trace._stackObj.err, function (stack) {
    var frames = stack.frames.map(parsers.parseCallsite)
    trace._stackObj.frames = frames
    cb(frames)
  })
}

function groupByMinute (arr, grouper) {
  var groups = {}

  arr.forEach(function (obj) {
    var key = grouper(obj)
    if (key in groups) groups[key].push(obj)
    else groups[key] = [obj]
  })

  return groups
}

function groupingTs (ts) {
  return new Date(ts.getFullYear(), ts.getMonth(), ts.getDate(), ts.getHours(), ts.getMinutes())
}

function transactionGroupingKey (trans) {
  return groupingTs(trans._start).getTime() + '|' + trans.name + '|' + trans.result + '|' + trans.type
}

function traceGroupingKey (trace) {
  var ancestors = trace.ancestors().map(function (trace) { return trace.signature }).join('|')
  return groupingTs(trace.transaction._start).getTime() + '|' + trace.transaction.name + '|' + ancestors + '|' + trace.signature
}
{'fraction_non_alphanumeric': 0.13689046674323901, 'fraction_numerical': 0.0014618356479064426, 'mean_word_length': 3.947314049586777, 'pattern_counts': {'":': 0, '<': 3, '<?xml version=': 0, '>': 5, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '1049371', 'n_tokens_mistral': 2997, 'n_tokens_neox': 2947, 'n_words': 632}