Flowmetal

A shining mercurial metal laden with sensors and almost infinitely reconfigurable.

The stuff of which robots and servitors are made.

Flowmetal is a substrate for automation. It attempts to provide a programming environment wherein programs are durable, evented and asynchronous aimed at what would traditionally be described as scripting or coordination.

Let's unpack these terms.

Durable - programs and their state are not dynamic and RAM located as with traditional models of processes. Instead programs and their state are always persisted to storage. This allows programs to sleep for a long time or even move seamlessly between machines.

Evented - durability is implemented in an event sourced style. Each program retails - or at least has the opportunity to retain - both a log of any external events and of its own execution. This allows for programs to easily rebuild their state, simplifies the durable programming model, and greatly simplifies debugging as intermediary states are retained and inspectable.

This also allows for external systems such as REST callback APIs, databases and such to easily integrate with Flowmetal programs as event sources. It also allows bidirectional communication between Flowmetal programs and other more traditional programming environments. Anything that can communicate with Flowmetal can provide function implementations, or call Flowmetal programs!

Asynchronous - thanks to Flowmetal's evented execution model, waiting for slow external events either synchronously or asynchronously is second nature! Flowmetal is especially good at waiting for very, very slow external operations. Stuff like webhooks and batch processes.

Scripting - the tradeoff Flowmetal makes for the evented model is that it's slow. While Flowmetal foreign functions could be fast, Flowmetal's interpreter isn't designed for speed. It's designed for eventing and ensuring durability. This makes Flowmetal suitable for interacting with and coordinating other systems, but it's not gonna win any benchmark games.

Wait what?

Okay. In simpler words, Flowmetal is an interpreted lisp which can use a datastore of your choice for durability. Other systems can attach to Flowmetal's datastore and send events to and receive them from Flowmetal. For instance Flowmetal contains a reference implementation of a HTTP callback connector and of a HTTP request connector.

A possible Flowmetal setup looks something like this -

                      +----------------------------+
                    +---------------------------+  |
                  +--------------------------+  |--+
                  | External HTTP service(s) |--+
                  +--------------------------+
                     ^                  ^
                     |                  |
                     v                  v
   +-----------------------+     +------------------------+
   | HTTP server connector |     | HTTP request connector |
   +-----------------------+     +------------------------+
                     ^                  ^
                     |                  |
                     v                  v
                    +--------------------+
                    | Shared event store |
                    +--------------------+
                             ^
                             |
                             v
                +--------------------------+
                | Flowmetal interpreter(s) |
                +--------------------------+

In this setup, the Flowmetal interpreters are able to interact with an external HTTP service; sending and receiving webhooks with Flowmetal programs waiting for those external events to arrive.

For instance this program would use the external connector stubs to build up interaction(s) with an external system.

(defpackage flowmetal.time
  (defenum time-unit
    "Calendar-independent durations."
    +milliseconds+
    +seconds+
    +hours+
    +days+)

  (defrecord duration [num : int?, scale : time-unit?]
     "A type for representing scalar durations.")

  (defn as-milliseconds [d : duration?]
                        : duration?
     "Normalize a duration to a number of milliseconds."
     (match d
        [(duration x +days+)         (duration (* x 24) +hours+)]
        [(duration x +hours+)        (duration (* x 60) +minutes+)]
        [(duration x +minutes+)      (duration (* x 60) +seconds+)]
        [(duration x +seconds+)      (duration (* x 1000) +milliseconds+)]
        [(duration x +milliseconds+) d]))

  ;; A type of one value used to represent an error
  (defenum timeout
    +timeout+)
    
  (defendpoint with-timeout!
     [d : duration?
      f : (fn? [] : a)]
     : a

    )

)

(defpackage com.twitter.wilson
  (require
    ;; The log lets you record status information into a program's trace
    [flowmetal.log
     :refer [log!]]
    ;; The time system lets you put bounds on the implicit awaiting Flowmetal does
    [flowmetal.time
     :refer [with-timeout!, timeout?, make-duration, duration?, +seconds+, +hours+, sleep!]]
    ;; JSON. Simple enough
    [flowmetal.json
     :refer [loads, dumps, json?]]
    ;; Extensions! Provided by other systems.
    ;;
    ;; This one allows for an external service to receive HTTP callbacks on Flowmetal's behalf.
    [http.callback
     :refer [make-callback!, get-callback!, callback?]]
     ;; This one allows for an external service to make HTTP requests on Flowmetal's behalf.
    [http.request
     :refer [post!, error?, dns-error?, connection-error?, response-error?]])

  (defenum stage
    +reboot+
    +bios-update+
    +reinstall+)

  ;; FIXME: how to do table optimization?
  (defn fib [x]
    (match x
       [0 1]
       [1 1]
       [_ (+ (fib (- x 1) (- x 2)))]))

  (defn retry-http [f
                      : (fn? [] a?)
                    backoff-fn
                      : (fn? [int?] duration?)
                      :default (fn [x : int?]
                                   : duration?
                                  (make-duration (fib x) +seconds+))
                    backoff-count
                      : int?
                      :default 0]
                    : a
    """The implementation of HTTP with retrying."""
    (let [response (f)]
      (if (not (error? response))
         response
         ;; FIXME: how does auth denied get represented?
         (if (or (dns-error? response)
                 (connection-error? response)
                 (response-error? response))
             (do (sleep (backoff-fn backoff-count))
                 (retry-http* f backoff-fn (+ backoff-count 1)))))))

  (defn job [hostname
               : str?
             stages
               : (list? stage?)
             job-timeout
               : duration?
               :default (duration 3 :hours)]
             : (union? [timeout? json?])
    """Run a wilson job, wait for the callback and process the result.

    By default the job is only waited for three hours.

    """
    (let [callback : callback? (make-callback!)
          job (retry-http
                (fn []
                  (post "http://wilson.local.twitter.com"
                   :data
                   (dumps
                     {:host hostname
                      :stages [stages]
                      :callbacks [{:type :http, :url callback}]}))))]

      (let [result (with-timeout! (duration 3 :hours)
                     (fn []
                        (get-callback callback)))]
         (if-not (timeout? result)
           (loads result)
           result))))
           
    )

Comparisons to Apache Airflow are at least in this setup pretty apt, although Flowmetal's durable execution model makes it much more suitable for providing reliable workflows and its DSL is more approachable.

License

Mirrored from https://git.arrdem.com/arrdem/flowmetal

Published under the MIT license. See LICENSE.md