Hystrix support (#42)

* Hystrix support

* Updates to observer pattern

* Minor tweak

* Removes weak adds comments

* Stats

* docs

* [skip ci] updates readme

* documentation update [ci skip]

* documentation update [ci skip]

* documentation update [ci skip]

Author: Aaron Liberatore

README.md

@@ -11,6 +11,7 @@ The Circuit Breaker design pattern is used to increase application stability, im
 * [Installation](#installation)
 * [Usage](#usage)
 * [API](#api)
+* [License](#license)
 
 ## Swift version
 The latest version of CircuitBreaker works with the `4.0.3` and newer version of the Swift binaries. You can download this version of the Swift binaries by following this [link](https://swift.org/download/#releases).
@@ -28,7 +29,7 @@ To leverage the CircuitBreaker package in your Swift application, you should spe
 
      dependencies: [
          // Swift 4
-         .package(url: "https://github.com/IBM-Swift/CircuitBreaker.git", .upToNextMajor(from: "4.0.0")),
+         .package(url: "https://github.com/IBM-Swift/CircuitBreaker.git", .upToNextMajor(from: "5.0.0")),
          ...
 
      ])
@@ -189,6 +190,20 @@ CircuitBreaker(timeout: Int = 1000, resetTimeout: Int = 60000, maxFailures: Int
  * `command` Contextual function to circuit break, which allows user defined failures (the context provides an indirect reference to the corresponding circuit breaker instance).
 
 ### Stats
+
+#### Tracked Stats:
+ * Total Requests
+ * Concurrent Requests
+ * Rejected Requests
+ * Successful Responses
+ * Average Execution Response Time
+ * Average Total Response Time
+ * Failed Responses
+ * Total Timeouts
+ * Total Latency
+ * Total Execution Latency
+ * Hystrix Compliant Snapshot
+
 ```swift
 ...
 // Create CircuitBreaker
@@ -199,18 +214,37 @@ breaker.run(commandArgs: (a: 10, b: 20), fallbackArgs: "Something went wrong.")
 
 // Log Stats snapshot
 breaker.snapshot()
+
+// Hystrix compliant snapshot
+let snapshot = breaker.snapshot
 ...
 ```
 
-#### Tracked Stats:
- * Total Requests
- * Concurrent Requests
- * Rejected Requests
- * Successful Responses
- * Average Response Time
- * Failed Responses
- * Total Timeouts
- * Total Latency
+#### Observing stats
+The CircuitBreaker library provides an interface for observing new CircuitBreaker instances in order to register and track stat changes. In the initialization of a CircuitBreaker instance, the linked monitors are notified of its instantiation allowing them to begin tracking the instance's stats. The CircuitBreaker instance exposes a Hystrix compliant stat snapshot to the monitor which can then be processed accordingly.
+
+```swift
+
+/// Initialize stat monitors
+let monitor1 = SwiftMetrics()
+let monitor2 = ...
+...
+let monitorN = ...
+
+/// Register monitors
+CircuitBreaker.addMonitor(monitor1)
+CircuitBreaker.addMonitor(monitor2)
+...
+CircuitBreaker.addMonitor(monitorN)
+
+// Create instances of CircuitBreaker
+let circuit1 = CircuitBreaker()
+let circuit2 = CircuitBreaker()
+...
+let circuitN = CircuitBreaker()
+```
+
+As mentioned above, the initializer takes care of notifying each one of the monitors of the new CircuitBreaker instance.
 
 ## License
 This Swift package is licensed under Apache 2.0. Full license text is available in [LICENSE](LICENSE).

Sources/CircuitBreaker/BreakerError.swift

@@ -1,5 +1,5 @@
 /**
-* Copyright IBM Corporation 2017
+* Copyright IBM Corporation 2017, 2018
 *
 * Licensed under the Apache License, Version 2.0 (the "License");
 * you may not use this file except in compliance with the License.

Sources/CircuitBreaker/Bulkhead.swift

@@ -1,5 +1,5 @@
 /**
- * Copyright IBM Corporation 2017
+ * Copyright IBM Corporation 2017, 2018
  *
  * Licensed under the Apache License, Version 2.0 (the "License");
  * you may not use this file except in compliance with the License.

Sources/CircuitBreaker/CircuitBreaker.swift

@@ -1,5 +1,5 @@
 /**
-* Copyright IBM Corporation 2017
+* Copyright IBM Corporation 2017, 2018
 *
 * Licensed under the Apache License, Version 2.0 (the "License");
 * you may not use this file except in compliance with the License.
@@ -31,6 +31,12 @@ public class CircuitBreaker<A, B> {
 
   // MARK: Public Fields
 
+  /// Name of Circuit Breaker Instance
+  public private(set) var name: String
+
+  // Name of Circuit Breaker Group
+  public private(set) var group: String?
+
   /// Execution timeout for command contect (Default: 1000 ms)
   public let timeout: Int
 
@@ -75,6 +81,8 @@ public class CircuitBreaker<A, B> {
   /// Initializes CircuitBreaker instance with asyncronous context command (Advanced usage)
   ///
   /// - Parameters:
+  ///   - name: name of the circuit instance
+  ///   - group: optional group description
   ///   - timeout: Execution timeout for command contect (Default: 1000 ms)
   ///   - resetTimeout: Timeout to reset circuit (Default: 6000 ms)
   ///   - maxFailures: Maximum number of failures allowed before opening circuit (Default: 5)
@@ -86,13 +94,17 @@ public class CircuitBreaker<A, B> {
   ///   - fallback: Function user specifies to signal timeout or fastFail completion.
   ///     Required format: (BreakerError, (fallbackArg1, fallbackArg2,...)) -> Void
   ///
-  public init(timeout: Int = 1000,
+  public init(name: String,
+              group: String? = nil,
+              timeout: Int = 1000,
               resetTimeout: Int = 60000,
               maxFailures: Int = 5,
               rollingWindow: Int = 10000,
               bulkhead: Int = 0,
               command: @escaping AnyContextFunction<A>,
               fallback: @escaping AnyFallback<B>) {
+    self.name = name
+    self.group = group
     self.timeout = timeout
     self.resetTimeout = resetTimeout
     self.maxFailures = maxFailures
@@ -101,6 +113,10 @@ public class CircuitBreaker<A, B> {
     self.command = command
     self.failures = FailureQueue(size: maxFailures)
     self.bulkhead = (bulkhead > 0) ? Bulkhead.init(limit: bulkhead) : nil
+
+    // Link to Observers
+
+    MonitorCollection.sharedInstance.values.forEach { $0.register(breakerRef: self) }
   }
 
   // MARK: Class Methods
@@ -122,41 +138,37 @@ public class CircuitBreaker<A, B> {
 
       if let bulkhead = self.bulkhead {
           bulkhead.enqueue {
-              self.callFunction(commandArgs: commandArgs, fallbackArgs: fallbackArgs)
+              self.callFunction(startTime: startTime, commandArgs: commandArgs, fallbackArgs: fallbackArgs)
           }
       } else {
-          callFunction(commandArgs: commandArgs, fallbackArgs: fallbackArgs)
+          callFunction(startTime: startTime, commandArgs: commandArgs, fallbackArgs: fallbackArgs)
       }
 
-      self.breakerStats.trackLatency(latency: Int(Date().timeIntervalSince(startTime)))
-
     case .closed:
       let startTime = Date()
 
       if let bulkhead = self.bulkhead {
           bulkhead.enqueue {
-              self.callFunction(commandArgs: commandArgs, fallbackArgs: fallbackArgs)
+              self.callFunction(startTime: startTime, commandArgs: commandArgs, fallbackArgs: fallbackArgs)
           }
       } else {
-          callFunction(commandArgs: commandArgs, fallbackArgs: fallbackArgs)
+          callFunction(startTime: startTime, commandArgs: commandArgs, fallbackArgs: fallbackArgs)
       }
-
-      self.breakerStats.trackLatency(latency: Int(Date().timeIntervalSince(startTime)))
     }
   }
 
   /// Method to print current stats
-  public func snapshot() {
+  public func logSnapshot() {
     breakerStats.snapshot()
   }
 
   /// Method to notifcy circuit of a completion with a failure
-  func notifyFailure(error: BreakerError, fallbackArgs: B) {
+  internal func notifyFailure(error: BreakerError, fallbackArgs: B) {
     handleFailure(error: error, fallbackArgs: fallbackArgs)
   }
 
   /// Method to notifcy circuit of a successful completion
-  func notifySuccess() {
+  internal func notifySuccess() {
     handleSuccess()
   }
 
@@ -180,9 +192,9 @@ public class CircuitBreaker<A, B> {
   }
 
   /// Wrapper for calling and handling CircuitBreaker command
-  private func callFunction(commandArgs: A, fallbackArgs: B) {
+  private func callFunction(startTime: Date, commandArgs: A, fallbackArgs: B) {
 
-    let invocation = Invocation(breaker: self, commandArgs: commandArgs, fallbackArgs: fallbackArgs)
+    let invocation = Invocation(startTime: startTime, breaker: self, commandArgs: commandArgs, fallbackArgs: fallbackArgs)
 
     setTimeout { [weak invocation, weak self] in
       if invocation?.nofityTimedOut() == true {
@@ -251,6 +263,7 @@ public class CircuitBreaker<A, B> {
   private func handleSuccess() {
     semaphoreCircuit.wait()
     Log.verbose("Handling success...")
+
     if state == .halfopen {
       close()
     }
@@ -300,3 +313,16 @@ public class CircuitBreaker<A, B> {
     resetTimer?.resume()
   }
 }
+
+extension CircuitBreaker: StatsProvider {
+
+  /// Method to create link a StatsMonitor Instance
+  public static func addMonitor(monitor: StatsMonitor) {
+    MonitorCollection.sharedInstance.values.append(monitor)
+  }
+
+  /// Property to compute snapshot
+  public var snapshot: Snapshot {
+    return Snapshot(name: name, group: group, stats: self.breakerStats, state: breakerState)
+  }
+}

Sources/CircuitBreaker/Collection.swift

@@ -1,5 +1,5 @@
 /**
- * Copyright IBM Corporation 2017
+ * Copyright IBM Corporation 2017, 2018
  *
  * Licensed under the Apache License, Version 2.0 (the "License");
  * you may not use this file except in compliance with the License.

Sources/CircuitBreaker/Invocation.swift

@@ -1,5 +1,5 @@
 /**
- * Copyright IBM Corporation 2017
+ * Copyright IBM Corporation 2017, 2018
  *
  * Licensed under the Apache License, Version 2.0 (the "License");
  * you may not use this file except in compliance with the License.
@@ -32,6 +32,10 @@ public class Invocation<A, B> {
   /// Completion state of invocation
   private(set) var completed: Bool = false
 
+  private let startTime: Date
+
+  private let startExecutionTime = Date()
+
   // Semaphore to avoid race conditions in the state of the invocation
   private let semaphoreCompleted = DispatchSemaphore(value: 1)
 
@@ -42,10 +46,11 @@ public class Invocation<A, B> {
   ///   - breaker CircuitBreaker Instance
   ///   - commandArgs Arguments for command context
   ///
-  public init(breaker: CircuitBreaker<A, B>, commandArgs: A, fallbackArgs: B) {
+  public init(startTime: Date, breaker: CircuitBreaker<A, B>, commandArgs: A, fallbackArgs: B) {
     self.breaker = breaker
     self.commandArgs = commandArgs
     self.fallbackArgs = fallbackArgs
+    self.startTime = startTime
   }
 
   /// Marks invocation as having timed out if and only if the execution
@@ -56,6 +61,9 @@ public class Invocation<A, B> {
     if !self.completed {
       setTimedOut()
       semaphoreCompleted.signal()
+      // Revisit Execution versus Total Latency
+      // Track latency for timeout in same way as brakes (https://github.com/awolden/brakes/)
+      breaker?.breakerStats.trackTotalLatency(latency: Int(Date().timeIntervalSince(startTime)))
       return true
     }
     semaphoreCompleted.signal()
@@ -83,6 +91,8 @@ public class Invocation<A, B> {
       self.setCompleted()
       semaphoreCompleted.signal()
       breaker?.notifySuccess()
+      breaker?.breakerStats.trackTotalLatency(latency: Int(Date().timeIntervalSince(startTime)))
+      breaker?.breakerStats.trackExecutionLatency(latency: Int(Date().timeIntervalSince(startTime)))
       return
     }
     semaphoreCompleted.signal()
@@ -99,6 +109,8 @@ public class Invocation<A, B> {
       self.setCompleted()
       semaphoreCompleted.signal()
       breaker?.notifyFailure(error: error, fallbackArgs: fallbackArgs)
+      breaker?.breakerStats.trackTotalLatency(latency: Int(Date().timeIntervalSince(startTime)))
+      breaker?.breakerStats.trackExecutionLatency(latency: Int(Date().timeIntervalSince(startTime)))
       return
     }
     semaphoreCompleted.signal()

Sources/CircuitBreaker/Monitor.swift

@@ -0,0 +1,41 @@
+/**
+ * Copyright IBM Corporation 2017, 2018
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ **/
+
+import Foundation
+
+/// Protocol identifying a stats monitor
+public protocol StatsMonitor {
+
+  /// References to monitored CircuitBreaker instances
+  var refs: [StatsProvider] { get set }
+
+  /// Method to register a stats provider
+  ///
+  /// - Parameters:
+  ///   - breakerRef: The StatsProvider to monitor
+  func register(breakerRef: StatsProvider)
+
+}
+
+/// Protocol identifying a stats provider
+public protocol StatsProvider: class {
+
+  /// Registers a monitor for a breaker reference
+  static func addMonitor(monitor: StatsMonitor)
+
+  /// Histrix compliant instance
+  var snapshot: Snapshot { get }
+}

Sources/CircuitBreaker/MonitorCollection.swift

@@ -0,0 +1,25 @@
+/**
+ * Copyright IBM Corporation 2017, 2018
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ **/
+
+internal struct MonitorCollection {
+
+  internal var values: [StatsMonitor] = []
+
+  internal static var sharedInstance = MonitorCollection()
+
+  private init() {}
+
+}