[Opossum] Add Stats and fix symbols.

Adds the supporting classes and interfaces for a breaker: Status, Stats, Bucket, Window, and HystrixStats. Fixes a bug with the `timeout` option where `false` was not provided as a valid value. Fixes the attributes of the main `CircuitBreaker` to not be symbols, but instead their correct types. This is a bit confusing when looking at the source of the lib, but while symbols are used extensively, they are not exported or made available via the class. They are only used internally and the classes attributes are exposed via getters. Fixes the global export of `promisify` and `stats` from the module as well as being tacked onto the default exported function. Beefs up the tests by copying examples used in the lib docs.
2026-10-06 15:57:13 +00:00 · 2018-12-31 07:38:33 -07:00
parent efb66b22be
commit 3a7ead14f2
2 changed files with 268 additions and 71 deletions
@@ -3,6 +3,7 @@
 // Definitions by: Quinn Langille <https://github.com/quinnlangille>
 //                 Willy Zhang <https://github.com/merufm>
 //                 Lance Ball <https://github.com/lance>
+//                 Matt R. Wilson <https://github.com/mastermatt>
 // Definitions: https://github.com/DefinitelyTyped/DefinitelyTyped
 // TypeScript Version: 2.8

@@ -13,63 +14,190 @@ import { EventEmitter } from "events";
 export type Action = (...args: any[]) => any;

 export class CircuitBreaker extends EventEmitter {
+    constructor(action: Action, options: CircuitBreakerOptions);
+
+    readonly name: string;
+    readonly group: string;
+    readonly enabled: boolean;
+    readonly pendingClose: boolean;
+    readonly closed: boolean;
+    readonly opened: boolean;
+    readonly halfOpen: boolean;
+    readonly status: Status;
+    readonly stats: Stats;
+    readonly hystrixStats: HystrixStats;
+    readonly warmUp: boolean;
+    readonly volumeThreshold: number;
+
    clearCache(): void;
+    open(): void;
    close(): void;
    disable(): void;
    enable(): void;
-    fallback(func?: (...args: any[]) => any): this;
+    fallback(func: Action | CircuitBreaker): this;
    fire(...args: any[]): Promise<any>;
-    healthCheck(func: (...args: any[]) => Promise<any>, interval?: number): Promise<any>;
-    on(event: string | symbol, listener: (...args: any[]) => void): this;
-    open(): void;
-    promisify(action: Action): Promise<Action>;
-    stats(): stream.Transform;
-
-    static readonly name: symbol;
-    static readonly group: symbol;
-    static readonly pendingClose: symbol;
-    static readonly closed: symbol;
-    static readonly opened: symbol;
-    static readonly halfOpen: symbol;
-    static readonly status: symbol;
-    static readonly hystrixStats: symbol;
-    static readonly enabled: symbol;
-    static readonly warmUp: symbol;
-    static readonly volumeThreshold: symbol;
+    healthCheck(
+        func: (...args: any[]) => Promise<any>,
+        interval?: number
+    ): void;
 }

 export enum Event {
-    cacheHit = 'cacheHit',
-    cacheMiss = 'cacheMiss',
-    close = 'close',
-    failure = 'failure',
-    fallback = 'fallback',
-    fire = 'fire',
-    halfOpen = 'halfOpen',
-    healthCheckFailed = 'health-check-failed',
-    open = 'open',
-    reject = 'reject',
-    semaphoreLocked = 'semaphore-locked',
-    success = 'success',
-    timeout = 'timeout'
+    cacheHit = "cacheHit",
+    cacheMiss = "cacheMiss",
+    close = "close",
+    failure = "failure",
+    fallback = "fallback",
+    fire = "fire",
+    halfOpen = "halfOpen",
+    healthCheckFailed = "health-check-failed",
+    open = "open",
+    reject = "reject",
+    semaphoreLocked = "semaphore-locked",
+    success = "success",
+    timeout = "timeout"
 }

 export interface CircuitBreakerOptions {
-    timeout?: number;
+    /**
+     * The time in milliseconds that action should be allowed to execute before timing out.
+     * Timeout can be disabled by setting this to `false`.
+     */
+    timeout?: number | false;
+
+    /**
+     * The number of times the circuit can fail before opening.
+     * @deprecated see options.errorThresholdPercentage
+     */
    maxFailures?: number;
+
+    /**
+     * The time in milliseconds to wait before setting the breaker to `halfOpen` state, and trying the action again.
+     */
    resetTimeout?: number;
+
+    /**
+     * Sets the duration of the statistical rolling window, in milliseconds.
+     * This is how long Opossum keeps metrics for the circuit breaker to use and for publishing.
+     * @default 10000
+     */
    rollingCountTimeout?: number;
+
+    /**
+     * Sets the number of buckets the rolling statistical window is divided into.
+     * So, if options.rollingCountTimeout is 10,000, and options.rollingCountBuckets is 10, then the
+     * statistical window will be 1,000 1 second snapshots in the statistical window.
+     * @default 10
+     */
    rollingCountBuckets?: number;
+
+    /**
+     * The circuit name to use when reporting stats.
+     * Defaults to the name of the action function then falls back to a UUID
+     */
    name?: string;
+
+    /**
+     * A grouping key for reporting.
+     * Defaults to the computed value of options.name
+     */
+    group?: string;
+
+    /**
+     * This property indicates whether execution latencies should be tracked and calculated as percentiles.
+     * If they are disabled, all summary statistics (mean, percentiles) are returned as -1.
+     * @default true
+     */
    rollingPercentilesEnabled?: boolean;
+
+    /**
+     * The number of concurrent requests allowed.
+     * If the number currently executing function calls is equal to options.capacity, further calls
+     * to `fire()` are rejected until at least one of the current requests completes.
+     * @default MAX_SAFE_INTEGER
+     */
    capacity?: number;
+
+    /**
+     * The error percentage at which to open the circuit and start short-circuiting requests to fallback.
+     */
    errorThresholdPercentage?: number;
+
+    /**
+     * Whether this circuit is enabled upon construction.
+     * @default true
+     */
    enabled?: boolean;
+
+    /**
+     * Determines whether to allow failures without opening the circuit during a brief warmup period
+     * This can help in situations where no matter what your `errorThresholdPercentage` is, if the
+     * first execution times out or fails, the circuit immediately opens.
+     * @default false
+     */
    allowWarmUp?: boolean;
+
+    /**
+     * The minimum number of requests within the rolling statistical window that must exist before
+     * the circuit breaker can open. This is similar to `allowWarmUp` in that no matter how many
+     * failures there are, if the number of requests within the statistical window does not exceed
+     * this threshold, the circuit will remain closed.
+     * @default 0
+     */
    volumeThreshold?: number;
+
+    /**
+     * If set to true, the value from the first call to `fire` will be cached an subsequent calls
+     * will not execute the `action` function, but return the cached value instead.
+     * @default false
+     */
+    cache?: boolean;
 }

-export default function circuitBreaker(
-    action: Action,
-    options?: CircuitBreakerOptions
-): CircuitBreaker;
+export interface Status extends EventEmitter {
+    stats: Stats;
+    window: Window;
+
+    increment(property: string, latencyRunTime?: number): void;
+    open(): void;
+    close(): void;
+}
+
+export interface Bucket {
+    failures: number;
+    fallbacks: number;
+    successes: number;
+    rejects: number;
+    fires: number;
+    timeouts: number;
+    cacheHits: number;
+    cacheMisses: number;
+    semaphoreRejections: number;
+    percentiles: { [percentile: number]: number };
+    latencyTimes: number[];
+}
+
+export type Window = Bucket[];
+
+export interface Stats extends Bucket {
+    latencyMean: number;
+}
+
+export class HystrixStats {
+    constructor(circuit: CircuitBreaker);
+
+    getHystrixStream(): stream.Transform;
+}
+
+export function promisify(action: Action): (...args: any[]) => Promise<any>;
+export const stats: stream.Transform;
+
+interface index {
+    (action: Action, options: CircuitBreakerOptions): CircuitBreaker;
+
+    promisify: (action: Action) => (...args: any[]) => Promise<any>;
+    stats: stream.Transform;
+}
+
+export const circuitBreaker: index;
+export default circuitBreaker;
@@ -1,48 +1,117 @@
-import { Transform } from "stream";
-import circuitBreaker, { CircuitBreaker, CircuitBreakerOptions } from "opossum";
+import * as fs from "fs";
+import circuitBreaker, {
+    CircuitBreaker,
+    CircuitBreakerOptions,
+    promisify,
+    stats,
+    Stats,
+    Window
+} from "opossum";

-const _blank = () => {};
-const async_blank = () => new Promise(resolve => null);
+let readFile = promisify(fs.readFile);
+stats.removeAllListeners();

-const testNoOptions: CircuitBreaker = circuitBreaker(_blank);
+let breaker: CircuitBreaker;
+const callbackNoArgs = () => console.log("foo");

-const options: CircuitBreakerOptions = {
-    timeout: 1,
-    maxFailures: 1,
-    resetTimeout: 1,
-    rollingCountTimeout: 1,
-    rollingCountBuckets: 1,
-    name: "testing",
+breaker = circuitBreaker(() => true, {
+    timeout: false,
+    maxFailures: 50,
+    resetTimeout: 10,
+    rollingCountTimeout: 500,
+    rollingCountBuckets: 20,
+    name: "test",
+    group: "group",
    rollingPercentilesEnabled: true,
    capacity: 1,
    errorThresholdPercentage: 1,
    enabled: true,
    allowWarmUp: true,
-    volumeThreshold: 1
+    volumeThreshold: 1,
+    cache: true
+});
+
+breaker.name; // $ExpectType string
+breaker.group; // $ExpectType string
+breaker.enabled; // $ExpectType boolean
+breaker.pendingClose; // $ExpectType boolean
+breaker.closed; // $ExpectType boolean
+breaker.opened; // $ExpectType boolean
+breaker.halfOpen; // $ExpectType boolean
+breaker.warmUp; // $ExpectType boolean
+breaker.volumeThreshold; // $ExpectType number
+breaker.status.stats.latencyMean; // $ExpectType number
+breaker.stats.latencyTimes; // $ExpectType number[]
+breaker.hystrixStats.getHystrixStream().removeAllListeners();
+
+breaker.clearCache(); // $ExpectType void
+breaker.open(); // $ExpectType void
+breaker.close(); // $ExpectType void
+breaker.disable(); // $ExpectType void
+breaker.enable(); // $ExpectType void
+
+// The following are examples are from the libs README and official documentation
+// https://nodeshift.github.io/opossum/index.html.
+
+function asyncFunctionThatCouldFail(x: any, y: any) {
+    return new Promise((resolve, reject) => {
+        // Do something, maybe on the network or a disk
+        resolve([x, y]);
+    });
+}
+
+const options: CircuitBreakerOptions = {
+    timeout: 3000, // If our function takes longer than 3 seconds, trigger a failure
+    errorThresholdPercentage: 50, // When 50% of requests fail, trip the circuit
+    resetTimeout: 30000 // After 30 seconds, try again.
 };
+breaker = circuitBreaker(asyncFunctionThatCouldFail, options);

-const testWithOptions: CircuitBreaker = circuitBreaker(_blank, options);
-const shouldBeAPromise: Promise<any> = testWithOptions.promisify(_blank);
-const shouldBeATransformStream: Transform = testWithOptions.stats();
+breaker
+    .fire("foo")
+    .then(console.log)
+    .catch(console.error);

-const healthCheck: Promise<any> = testWithOptions.healthCheck(async_blank, 8);
-const shouldBeACircuitBreaker: CircuitBreaker = testWithOptions.fallback();
-const shouldBeAPromiseGeneric: Promise<any> = testWithOptions.fire();
+breaker = circuitBreaker(asyncFunctionThatCouldFail, options);
+// if asyncFunctionThatCouldFail starts to fail, firing the breaker
+// will trigger our fallback function
+breaker.fallback(() => "Sorry, out of service right now");
+breaker.on("fallback", result => console.log(result));

-const shouldBeVoid = (): void => testWithOptions.enable();
-const shouldBeVoid2 = (): void => testWithOptions.disable();
-const shouldBeVoid3 = (): void => testWithOptions.close();
-const shouldBeVoid4 = (): void => testWithOptions.open();
-const shouldBeVoid5 = (): void => testWithOptions.clearCache();
+breaker = circuitBreaker(callbackNoArgs, options);

-const shouldBeSymbol: symbol = CircuitBreaker.name;
-const shouldBeSymbol2: symbol = CircuitBreaker.group;
-const shouldBeSymbol3: symbol = CircuitBreaker.pendingClose;
-const shouldBeSymbol4: symbol = CircuitBreaker.closed;
-const shouldBeSymbol5: symbol = CircuitBreaker.opened;
-const shouldBeSymbol6: symbol = CircuitBreaker.halfOpen;
-const shouldBeSymbol7: symbol = CircuitBreaker.hystrixStats;
-const shouldBeSymbol8: symbol = CircuitBreaker.status;
-const shouldBeSymbol9: symbol = CircuitBreaker.enabled;
-const shouldBeSymbol10: symbol = CircuitBreaker.warmUp;
-const shouldBeSymbol11: symbol = CircuitBreaker.volumeThreshold;
+breaker.fallback(callbackNoArgs);
+
+breaker.on("success", result => console.log(result));
+breaker.on("timeout", callbackNoArgs);
+breaker.on("reject", callbackNoArgs);
+breaker.on("open", callbackNoArgs);
+breaker.on("halfOpen", callbackNoArgs);
+breaker.on("close", callbackNoArgs);
+breaker.on("fallback", data => console.log(data));
+
+readFile = circuitBreaker.promisify(fs.readFile);
+breaker = circuitBreaker(readFile, options);
+
+breaker
+    .fire("./package.json", "utf-8")
+    .then(console.log)
+    .catch(console.error);
+
+breaker = circuitBreaker(fs.readFile, {});
+
+breaker.hystrixStats.getHystrixStream().pipe(process.stdout);
+
+// Creates a 1 second window consisting of ten time slices,
+// each 100ms long.
+const circuit = circuitBreaker(fs.readFile, {
+    rollingCountBuckets: 10,
+    rollingCountTimeout: 1000
+});
+
+// get the cumulative statistics for the last second
+const theStats: Stats = breaker.status.stats;
+
+// get the array of 10, 1 second time slices for the last second
+const window: Window = breaker.status.window;
+window[0].fires; // $ExpectType number