all files / src/ breaker.js

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
1×
1×
 
1×
1×
 
 
 
 
 
 
 
 
 
1×
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
const { commandFactory } = require('hystrixjs');
const request = require('request');
 
const promisify = require('util.promisify');
const callbackify = require('util').callbackify || function (promiser) {
    return function (...args) {
        const callback = args.pop();
        promiser.apply(this, args).then(
            result => callback(null, result),
            error => callback(error)
        );
    };
};
 
module.exports = (redis, options) => {
    const service = promisify((command_obj, callback) => {
        command_obj.callback = callback;
    });
 
    const fallback = promisify((error, [command_obj], callback) => {
        command_obj.command = command_obj.callback = null;
        callback(error);
    });
 
    const breaker = commandFactory.getOrCreate(options.name)
        .circuitBreakerErrorThresholdPercentage(options.errorThreshold)
        .timeout(options.timeout)
        .run(service)
        .circuitBreakerRequestVolumeThreshold(options.volumeThreshold)
        .requestVolumeRejectionThreshold(options.rejectionThreshold)
        .circuitBreakerSleepWindowInMilliseconds(options.sleepWindow)
        .statisticalWindowLength(options.windowLength)
        .statisticalWindowNumberOfBuckets(options.windowBuckets)
        .fallbackTo(fallback)
        .build();
 
    const { createClient } = redis;
 
    redis.createClient = function () {
        const client = createClient.apply(this, arguments);
 
        const  { internal_send_command } = client;
 
        client.internal_send_command = function (command_obj) {
            const { command, callback = function () {} } = command_obj;
 
            if (!command) return true;
 
            callbackify(breaker.execute).call(breaker, command_obj, callback);
 
            return internal_send_command.call(this, command_obj);
        };
 
        return client;
    };
 
    if (options.influxdbUrl) {
        request.get('http://169.254.169.254/latest/meta-data/instance-id', (error, response) => {
            const instance = response.body;
 
            const { hystrixSSEStream } = require('hystrixjs');
 
            hystrixSSEStream.toObservable(options.statsWindow).subscribe(json => {
                const data = JSON.parse(json);
 
                Object.keys(data.latencyExecute).forEach(key => {
                    data[`latencyExecute_${key}`] = data.latencyExecute[key];
                });
 
                Object.keys(data.latencyTotal).forEach(key => {
                    data[`latencyTotal_${key}`] = data.latencyTotal[key];
                });
 
                const tags = [
                    'name',
                    'group'
                ];
 
                const fields = [
                    'isCircuitBreakerOpen',
                    'errorPercentage',
                    'errorCount',
                    'requestCount',
                    'rollingCountFailure',
                    'rollingCountTimeout',
                    'rollingCountSuccess',
                    'rollingCountShortCircuited',
                    'rollingCountBadRequests',
                    'rollingCountCollapsedRequests',
                    'rollingCountExceptionsThrown',
                    'rollingCountFallbackFailure',
                    'rollingCountFallbackRejection',
                    'rollingCountFallbackSuccess',
                    'rollingCountResponsesFromCache',
                    'rollingCountSemaphoreRejected',
                    'rollingCountThreadPoolRejected',
                    'currentConcurrentExecutionCount',
                    'latencyExecute_mean',
                    'latencyExecute_0',
                    'latencyExecute_25',
                    'latencyExecute_50',
                    'latencyExecute_75',
                    'latencyExecute_90',
                    'latencyExecute_95',
                    'latencyExecute_99',
                    'latencyExecute_99.5',
                    'latencyExecute_100',
                    'latencyTotal_mean',
                    'latencyTotal_0',
                    'latencyTotal_25',
                    'latencyTotal_50',
                    'latencyTotal_75',
                    'latencyTotal_90',
                    'latencyTotal_95',
                    'latencyTotal_99',
                    'latencyTotal_99.5',
                    'latencyTotal_100'
                ];
 
                const extract = (keys, obj) => {
                    return keys.map(key => {
                        const value = obj[key].toString().replace(/\W/g, '_');
                        return [key, value].join('=');
                    });
                };
 
                const parts = [
                    ['hystrix', `instance=${instance}`, ...extract(tags, data)],
                    [...extract(fields, data)],
                    [data.currentTime]
                ];
 
                request.post({
                    url: `http://${options.influxdbUrl}:8086/write?db=telegraf&precision=ms`,
                    body: parts.map(part => part.join(',')).join(' ')
                });
            });
        });
    }
 
    return redis;
};