<commit_before>/*
 * This Source Code Form is subject to the terms of the Mozilla Public
 * License, v. 2.0. If a copy of the MPL was not distributed with this
 * file, You can obtain one at http://mozilla.org/MPL/2.0/.
 */

/*
 * Copyright (c) 2015, Joyent, Inc.
 */

/**
 * A Transform stream from changelog entries to user objects
 */

// expect:

var assert = require('assert-plus');
var Transform = require('stream').Transform;
var util = require('util');
var mod_uuid = require('node-uuid');
var mod_vasync = require('vasync');

function UfdsUserStream(opts) {
    assert.object(opts, 'opts');
    assert.object(opts.log, 'opts.log');
    assert.object(opts.ufds, 'opts.ufds');

    Transform.call(this, { objectMode: true });

    var self = this;

    self.log = opts.log.child({ component: 'UfdsUserStream'}, true);
    self.ufdsClient = opts.ufds;
    self.dead = false;

    // each obj is a changelog with a parsed 'changes' property.
    function ufdsWorker(obj, cb) {
        var err;
        var uuid;
        var account;

        if (self.dead) {
            return cb();
        }

        if (!(obj && obj.changes && obj.changes.uuid) ||
            !obj.changes.uuid.length) {
            err = new Error(util.format,
                'Missing uuid for changenumber %d', obj.changenumber);
            self.log.error({ err: err, changelog: obj },
                'Can\'t determine UUID for changelog');
            return cb(err);
        } else if (obj.changes.uuid.length > 1) {
            err = new Error(util.format,
                'Changelog %d has extra uuid', obj.changenumber);
            self.log.error({ err: err, changelog: obj },
                'Can\'t determine UUID for changelog %d', obj.changenumber);
            return cb(err);
        } else {
            uuid = obj.changes.uuid[0];
        }

        if ((obj && obj.changes && obj.changes.account) &&
            obj.changes.account.length !== 1) {
            account = obj.changes.account[0];
        }

        self.ufdsClient.getUser(uuid, account, function (ufdsErr, user) {
            if (ufdsErr && ufdsErr.name === 'NoSuchObjectError') {
                return cb();
            } else if (ufdsErr && ufdsErr.statusCode >= 500 &&
                ufdsErr.statusCode < 600) {
                var timeout = 10000;
                self.log.warn(ufdsErr, 'UFDS server error fetching user %s.' +
                    ' Retrying in %sms', uuid, timeout);
                setTimeout(function () {
                    self.ufdsQueue.push(wrapTask(obj, cb));
                    self.log.debug({ npending: self.ufdsQueue.npending,
                        queued: self.ufdsQueue.queued.length },
                        'push: ufdsQueue has %d tasks running, %d queued',
                        self.ufdsQueue.npending, self.ufdsQueue.queued.length);
                }, timeout);
                return cb();<commit_msg>NAPI-249 add logging to determine what sort of changelogs are causing these errors.<commit_after>/*
 * This Source Code Form is subject to the terms of the Mozilla Public
 * License, v. 2.0. If a copy of the MPL was not distributed with this
 * file, You can obtain one at http://mozilla.org/MPL/2.0/.
 */

/*
 * Copyright (c) 2015, Joyent, Inc.
 */

/**
 * A Transform stream from changelog entries to user objects
 */

// expect:

var assert = require('assert-plus');
var Transform = require('stream').Transform;
var util = require('util');
var mod_uuid = require('node-uuid');
var mod_vasync = require('vasync');

function UfdsUserStream(opts) {
    assert.object(opts, 'opts');
    assert.object(opts.log, 'opts.log');
    assert.object(opts.ufds, 'opts.ufds');

    Transform.call(this, { objectMode: true });

    var self = this;

    self.log = opts.log.child({ component: 'UfdsUserStream'}, true);
    self.ufdsClient = opts.ufds;
    self.dead = false;

    // each obj is a changelog with a parsed 'changes' property.
    function ufdsWorker(obj, cb) {
        var err;
        var uuid;
        var account;

        if (self.dead) {
            return cb();
        }

        if (!(obj && obj.changes && obj.changes.uuid) ||
            !obj.changes.uuid.length) {
            err = new Error(util.format,
                'Missing uuid for changenumber %d', obj.changenumber);
            self.log.error({ err: err, changelog: obj },
                'Can\'t determine UUID for changelog');
            return cb(err);
        } else if (obj.changes.uuid.length > 1) {
            err = new Error(util.format,
                'Changelog %d has extra uuid', obj.changenumber);
            self.log.error({ err: err, changelog: obj },
                'Can\'t determine UUID for changelog %d', obj.changenumber);
            return cb(err);
        } else {
            self.log.debug({
                changelog: obj
            }, 'found changelog to process');
            uuid = obj.changes.uuid[0];
        }

        if ((obj && obj.changes && obj.changes.account) &&
            obj.changes.account.length !== 1) {
            account = obj.changes.account[0];
        }

        self.ufdsClient.getUser(uuid, account, function (ufdsErr, user) {
            if (ufdsErr && ufdsErr.name === 'NoSuchObjectError') {
                return cb();
            } else if (ufdsErr && ufdsErr.statusCode >= 500 &&
                ufdsErr.statusCode < 600) {
                var timeout = 10000;
                self.log.warn(ufdsErr, 'UFDS server error fetching user %s.' +
                    ' Retrying in %sms', uuid, timeout);
                setTimeout(function () {
                    self.ufdsQueue.push(wrapTask(obj, cb));
                    self.log.debug({ npending: self.ufdsQueue.npending,
                        queued: self.ufdsQueue.queued.length },
                        'push: ufdsQueue has %d tasks running, %d queued',
                        self.ufdsQueue.npending, self.ufdsQueue.queued.length);
                }, timeout);
                return cb();
{'fraction_non_alphanumeric': 0.12883333333333333, 'fraction_numerical': 0.008166666666666666, 'mean_word_length': 1.8837097549255166, 'pattern_counts': {'":': 0, '<': 5, '<?xml version=': 0, '>': 7, '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': '4085037', 'n_tokens_mistral': 1831, 'n_tokens_neox': 1745, 'n_words': 476}