Takes a number threshold and a time period in seconds duration. If the condition “the event metric is > to the threshold” is valid for all events received during at least the period duration, valid events received after the duration period will be passed on until an invalid event arrives. :metric should not be nil (it will produce exceptions).

(above-dt {:threshold 100 :duration 10}
  (debug))

In this example, if the events :metric field are greater than 100 for more than 10 seconds, events are passed downstream.

Execute children into the specific async queue. The async queue should be defined in the I/O configuration file.

(async-queue! :my-queue
  (info))

Takes a number threshold and a time period in seconds duration. If the condition the event metric is < to the threshold is valid for all events received during at least the period duration, valid events received after the duration period will be passed on until an invalid event arrives. :metric should not be nil (it will produce exceptions).

(below-dt {:threshold 100 :duration 10}
(debug))

In this example, if the events :metric field are lower than 100 for more than 10 seconds, events are passed downstream.

Takes two numbers, low and high, and a time period in seconds, duration. If the condition the event metric is > low and < high is valid for all events received during at least the period duration, valid events received after the duration period will be passed on until an invalid event arrives. :metric should not be nil (it will produce exceptions).

(between-dt {:low 50 :high 100 :duration 10}
  (debug))

In this example, if the events :metric field are between 50 ans 100 for more than 10 seconds, events are passed downstream.

Get the min event from the last dt seconds.

(bottom {:duration 10}
  (info))

You can pass a :delay key to the configuration in order to tolerate late events. In that case, events from previous windows will be flushed after this delay:

(bottom {:duration 10 :delay 5}
  (info))

Split stream by field Every time an event arrives with a new value of field, this action invokes its child forms to return a new, distinct set of streams for that particular value.

(by {:fields [:host :service]}
  (fixed-time-window {:duration 60}))

This example generates a moving window for each host/service combination.

You can also pass the :gc-interval and :fork-ttl keys to the action. This will enable garbage collections of children actions, executed every :gc-interval (in seconds) and which will remove actions which didn’t receive events since :fork-ttl seconds

(by {:fields [:host :service [:a :nested-key]
     :gc-interval 3600
     :fork-ttl 1800}
  (fixed-time-window {:duration 60}))

Passes on events only if the field passed as parameter differs from the previous one. The init parameter is the default value for the stream.

(changed {:field :state :init "ok"})

For example, this action will let event pass if the :state field vary, the initial value being ok.

This stream is useful to get only events making a transition.

It also supported nested fields:

(changed {:field [:nested :field] :init "ok"})

Returns a list of the latest non-expired events (by fields) every dt seconds (at best).

(coalesce {:duration 10 :fields [:host :service]}
  (debug)

In this example, the latest event for each host/service combination will be kept and forwarded downstream. The debug action will then receive this list of events. Expired events will be removed from the list.

coalesce supports nested fields:

(coalesce {:duration 10 :fields [:host [:nested :field]]}
  (debug)

Receives a list of events, returns the bottom N events with the lowest metrics.

(fixed-time-window {:duration 60}
  (coll-bottom 5
    (info)))

Count the number of events. Should receive a list of events from the previous stream. The most recent event is used as a base to create the new event, and its :metric field is set to the number of events received as input.

(fixed-time-window {:duration 60}
  (coll-count
    (debug)))

Receives a list of events which should represent an always-increasing counter and returns the latest event with its :metric field set to the value of the increase between the oldest and the latest event.

If it receives for example:

[{:time 1 :metric 10} {:time 9 :metric 20} {:time 20 :metric 30}}

It will produces (30-10 = 20):

{:time 20 :metric 20}

Events produced with a negative metric (which can happen if the counter is resetted) are not send downstream.

Returns the event with the biggest metric. Should receive a list of events from the previous stream.

(fixed-event-window {:size 10}
  (coll-max
    (debug)))

Get the event the biggest metric on windows of 10 events.

Check top and smax as well.

Computes the events mean (on metric). Should receive a list of events from the previous stream. The most recent event is used as a base to create the new event

(fixed-event-window {:size 10}
  (coll-mean
    (debug)))

Computes the mean on windows of 10 events

Returns the event with the smallest metric. Should receive a list of events from the previous stream.

(fixed-event-window {:size 10}
  (coll-min
    (debug)))

Get the event the smallest metric on windows of 10 events.

Check bottom and smin as well.

Receives a list of events and selects one event from that period for each point. If point is 0, takes the lowest metric event. If point is 1, takes the highest metric event. 0.5 is the median event, and so forth. Forwards each of these events to children. The event has the point appended the :quantile key. Useful for extracting histograms and percentiles.

(fixed-event-window {:size 10}
  (coll-percentiles [0.5 0.75 0.98 0.99]))

Divide the first event :metrìc field by all subsequent events :metric. Return a new event containing the new :metric.

Should receive a list of events from the previous stream.

Computes the rate on a list of events. Should receive a list of events from the previous stream. The latest event is used as a base to build the new event.

(fixed-event-window {:size 3}
  (coll-rate
    (debug)))

If this example receives the events:

{:metric 1 :time 1} {:metric 2 :time 2} {:metric 1 :time 3}

The stream will return {:metric 2 :time 3}

Indeed, (1+2+1)/2 = 3 (we divide by 2 because we have 2 seconds between the min and max events time).

Sort events based on the field passed as parameter Should receive a list of events from the previous stream.

(fixed-event-window {:size 10}
  (coll-sort :time
    (debug)))

Sum all the events :metric fields Should receive a list of events from the previous stream.

(fixed-event-window {:size 10}
  (coll-sum
    (debug)))

Sum all :metric fields for windows of 10 events

Receives a list of events, returns the top N events with the highest metrics.

(fixed-time-window {:duration 60}
  (coll-top 5
    (info)))

Like where but should receive a list of events.

(fixed-time-window {:duration 60}
  (coll-where [:and [:= :host "foo"]
                    [:> :metric 10]))

A stream which detects if a condition (f event) is true during dt seconds. Takes conditions (like in the where action) and a time period dt in seconds. If the condition is valid for all events received during at least the period dt, valid events received after the dt period will be passed on until an invalid event arrives. Skips events that are too old or that do not have a timestamp.

Keep all events in state critical.

(critical
  (error))

In this example, all events with :state “critical” will be logged.

Takes a time period in seconds duration. If all events received during at least the period duration have :state critical, new critical events received after the duration period will be passed on until an invalid event arrives.

(critical-dt {:duration 10}
  (debug))

In this example, if the events :state are “critical” for more than 10 seconds, events are passed downstream.

Executes a custom action. Custom actions are defined in the Mirabelle configuration file. The actomn can then be called (by name) using this custom action.

(custom :my-custom-action ["parameters"]
  (info))

Differentiate metrics with respect to time. Emits an event for each event received, but with metric equal to the difference between the current event and the previous one, divided by the difference in their times. Skips events without metrics.

(ddt
  (info))

If ddt receives {:metric 1 :time 1} and {:metric 10 :time 4}, it will produce {:metric (/ 9 3) :time 4}.

Like ddt but do not forward events with negative metrics. This can be used for counters which may be reseted to zero for example.

Print the event in the logs using the debug level

(increment
  (debug))

Decrement the event :metric field.

(decrement
  (index [:host]))

Set a default value for an event

(default :state "ok"
  (info))

In this example, all events where :state is not set will be updated with :state to “ok”.

Dissociates an entry from a nested associative structure returning a new nested structure. keys is a sequence of keys. Any empty maps that result will not be present in the new structure.

Print the event in the logs using the error level

(increment
  (debug))

Exponential weighted moving average. Constant space and time overhead. Passes on each event received, but with metric adjusted to the moving average. Does not take the time between events into account. R is the ratio between successive events: r=1 means always return the most recent metric; r=1/2 means the current event counts for half, the previous event for 1/4, the previous event for 1/8, and so on.

Build a new event from an Exception and from the event which caused it.

Takes two actions. If an exception is thrown in the first action, an event representing this exception is emitted in in the second action.

(exception-stream
  (bad-action)
  (error))

Here, if bad-action throws, an event will be built (using the exception->event function) and sent to the error action (which will log it).

Keep expired events.

(expired
  (increment))

In this example, all expired events will be forwarded to the incrementstream.

Keep expired events.

Takes a key as parameter and send downstream the content of this key.

(extract :base-event
  (info))

If extract receives in this example {:time 1 :base-event {:time 1 :service "foo" :host "bar"}, info will receive the content of :base-time.

Returns a fixed-sized window of events.

(fixed-event-window {:size 5}
  (debug))

This example will return a vector events partitioned 5 by 5.

A fixed window over the event stream in time. Emits vectors of events, such that each vector has events from a distinct n-second interval. Windows do not overlap; each event appears at most once in the output stream.

(fixed-time-window {:duration 60}
  (coll-max
    (info)))

You can pass a :delay key to the configuration in order to tolerate late events. In that case, previous windows will be flushed after this delay:

(fixed-time-window {:duration 60 :delay 30}
  (coll-max
    (info)))

Convert a field or multiple fields from base64 to string. Fields values should be string.

(sdo
  ;; you can pass one field
  (from-base64 :host)
  ;; or a list of fields
  (from-base64 [:host :service]))

Include an configuration file by path into the configuration. The file will be read using the aero (https://github.com/juxt/aero/) library. The config variable supports these optional options:

• :profile: the aero profile to use. By default, Mirabelle will read (and convert to a Clojure keyword) the PROFILE environment variable during compilation. You can override this value by setting :profile.
• :variables: variables to pass to the configuration file. You can use the #mirabelle/var reader in order to define variables in your EDN file.

This allows you to use the same configuration snippet (eventually templated) from multiple streams (or multiple parts of the same stream)

(includes"/etc/mirabelle/includes/my-actions.clj {:profile :dev
                                                   :variables {:foo "bar"})

Increment the event :metric field.

(increment
  (index [:host]))

Insert events into the index. Events are indexed using the keys passed as parameter.

(index [:host :service])

This example will index events by host and services.

Print the event in the logs using the info level

(increment
  (info))

Discard all events in test mode. Else, forward to children. You can use this stream to avoid side effects in test mode.

Takes a field or a list of fields, and converts the values associated to these fields from json to edn.

(with :my-field "{"foo": "bar"}
  (json-fields [:my-field]))

In this example, we associate to :my-field a json string and then we call json-fields on it. :my-field will now contain an edn map built from the json data, with keywords as keys.

Keep only the specified keys for events.

(keep-keys [:host :metric :time :environment :description]
  (info))

Also works with nested keys:

(keep-keys [:host :metric :time [:a :nested-key]]
  (info))

Takes an event or a list of events. Returns an event (or a list of events depending of the input) with all events tagged “mirabelle/discard” filtered. Returns nil if all events are filtered.

Generic logger

Get the mean of event metrics from the last dt seconds.

(mean {:duration 10}
  (info))

You can pass a :delay key to the configuration in order to tolerate late events. In that case, events from previous windows will be flushed after this delay:

(mean {:duration 10 :delay 5}
  (info))

A sliding window of the last few events. Every time an event arrives, calls children with a vector of the last n events, from oldest to newest. Ignores event times. Example:

(moving-event-window {:size 5}
  (coll-mean (info))

A sliding window of all events with times within the last n seconds. Uses the maximum event time as the present-time horizon. Every time a new event arrives within the window, emits a vector of events in the window to children.

Events without times accrue in the current window.

Keep non-expired events.

(not-expired
  (increment))

In this example, all non-expired events will be forwarded to the `increment`stream.

Keep non-expired events.

Push events to an external system.

Outputs are configured into the main Mirabelle configuration file. If you create a new output named :influxdb for example, you can use output! to push all events into this I/O:

(output! :influxdb)

Outputs are automatically discarded in test mode.

Takes two numbers, low and high, and a time period in seconds, duration. If the condition the event metric is < low or > high is valid for all events received during at least the period duration, valid events received after the duration period will be passed on until an invalid event arrives. :metric should not be nil (it will produce exceptions).

(outside-dt {:low 50 :high 100 :duration 10}
  (debug))

In this example, if the events :metric field are outside the 50-100 range for more than 10 seconds, events are passed downstream.

Passes on events only when their metric is greater than x.

(over 10
  (info))

Takes a list of conditions. Like coalesce, project will return the most recent events matching the conditions.

(project [[:= :service "enqueues"]
          [:= :service "dequeues"]]
  (coll-quotient
    (with :service "enqueues per dequeue"
      (info))))

We divide here the latest event for the “enqueues” :service by the latest event from the “dequeues” one.

Publish events in the given channel.

(publish! :my-channel)

Users can then subscribe to channels using the websocket engine.

Everytime this action receives an event, it will expires events from the index (every dt seconds) and reinject them into a stream (default to the current stream if not specified).

(reaper 5)

(reaper 5 :custom-stream)

Reinject an event into the streaming system. By default, events are reinject into the real time engine. You can reinject events to a specific stream by passing the destination stream as parameter.

(reinject)

This example reinjects events into the real stream engine.

(reinject :foo)

This example reinjects events into the stream named :foo.

Rename events keys.

(rename-keys {:host :service
              :environment :env}

In this example, the :host key will be renamed :service and the :environment key is renamed :env. Existing values will be overrided.

Multiplies the event :metric field by the factor passed as parameter.

(scale 1000
  (info

This example will multiply the :metric field for all events by 1000.

Remove a key (or a list of keys) from the events.

(sdissoc :host (info))

(sdissoc [:environment :host] (info))

(sdissoc [:environment [:nested :key] (info))

Send events to children. useful when you want to send the same events to multiple downstream actions.

(sdo
  (increment)
  (decrement))

Here, events arriving in sdo will be forwarded to both increment and decrement.

Streaming flatten. Calls children with each event in events. Events should be a sequence.

This stream can be used to “flat” a sequence of events (emitted by a time window stream for example).

(fixed-event-window {:size 5}
  (sflatten
    (info)))

Takes the content of multiple event keys, and use them to build a string value and assign it to a given key.

(sformat "%s-foo-%s" :format-test [:host :service])

If the event {:host "machine" :service "bar"} is passed to this action the event will become {:host "machine" :service "bar" :format-test "machine-foo-bar"}.

More information about availables formatters in the Clojure documentation: https://clojuredocs.org/clojure.core/format

Send downstream the event with the biggest :metric every time it receives an event

(smax
  (info))

If the events {:time 1 :metric 10} {:time 2 :metric 3} {:time 3 :metric 11} are injected, info will receive:

{:time 1 :metric 10} {:time 1 :metric 10} {:time 3 :metric 11}

Send downstream the event with the lowest :metric every time it receives an event

(smin
  (info))

If the events {:time 1 :metric 10} {:time 2 :metric 3} {:time 3 :metric 11} are injected, info will receive:

{:time 1 :metric 10} {:time 2 :metric 3} {:time 3 :metric 11}

Split by conditions.

(split
  [:> :metric 10] (debug)
  [:> :metric 5] (info)
  (error)

In this example, all events with :metric > 10 will go into the debug stream, all events with :metric > 5 in the info stream, and other events will to the default stream which is “error”.

The default stream is optional, if not set all events not matching a condition will be discarded.

Streaming sort. Takes a configuration containing a :duration and a :field key. The action will buffer events during :duration seconds and then send the events downstream one by one, sorted by :field.

(ssort {:duration 10 :field :time}
  (info))

This example will sort events based on the :time field.

For example, if it get as input:

{:time 1} {:time 10} {:time 4} {:time 9} {:time 13} {:time 31}

Info will receive these events:

{:time 1} {:time 4} {:time 9} {:time 10} {:time 13}

You can add a :delay key to the action configuration in order to tolerate late events:

(ssort {:duration 10 :field :time :delay 10}
  (info))

In this example, events from previous windows will be sent with a delay of 10 seconds.

ssort supports sorting on a nested field (example :field [:nested :field])

Takes a duration (dt) in second and a field name as parameter. Returns events where the value of the field specified as second argument is equal to the value of the field for the last event, for at least dt seconds. Events can be buffered for dt seconds before being forwarded in order to see if they are stable or not.

Events should arrive in order (old events will be dropped).

You can use this stream to remove flapping states for example.

(stable 10 :state
  (info))

In this example, events will be forwarded of the value of the :state key is the same for at least 10 seconds.

Support nested fields:

(stable 10 [:nested :field]
  (info))
```.

Creates a new stream. This action takes a map where the :name key, which will be the name of the stream, is mandatory.

Entrypoint for all streams.

(streams
  (stream {:name :fobar}
    (info))
  (stream {:name :foo}
    (info)))

Sum the events field from the last dt seconds.

(sum {:duration 10}
  (info))

You can pass a :delay key to the configuration in order to tolerate late events. In that case, events from previous windows will be flushed after this delay:

(sum {:duration 10 :delay 5}
  (info))

Adds a new tag, or set of tags, to events which flow through.

(tag "foo"
  (info))

This example adds the tag “foo” to events.

(tag ["foo" "bar"] (info))

This example adds the tag “foo” and “bar” to events.

Passes on events where all tags are present.

(tagged-all "foo"
  (info))

This example keeps only events tagged “foo”.

(tagged-all ["foo" "bar"] (info))

This example keeps only events tagged “foo” and “bar”.

Save events into the tap. Noop outside tests.

(where [:= :service "foo"]
  (tap :foo)

In test mode, all events with :service “foo” will be saved in a tap named :foo

Bufferize all received events in the state (an atom) passed as parameter

Let N event pass at most every duration seconds. Can be used for example to avoid sending to limit the number of alerts sent to an external system.

(throttle {:count 3 :duration 10}
  (error))

In this example, throttle will let 3 events pass at most every 10 seconds. Other events, or events with no time, are filtered.

Convert a field or multiple fields to base64. Fields values should be string.

(sdo
  ;; you can pass one field
  (to-base64 :host)
  ;; or a list of fields
  (to-base64 [:host :service]))

Get the max event from the last dt seconds.

(top {:duration 10}
  (info))

You can pass a :delay key to the configuration in order to tolerate late events. In that case, events from previous windows will be flushed after this delay:

(top {:duration 10 :delay 5}
  (info))

Passes on events only when their metric is under than x.

(under 10
  (info))

Removes a tag, or set of tags, from events which flow through.

(untag "foo" index)

This example removes the tag “foo” from events.

(untag ["foo" "bar"] index)

This example removes the tags “foo” and “bar” from events

Keep all events in state warning.

(warning
  (warning))

In this example, all events with :state “warning” will be logged.

Filter events based on conditions. Each condition is a vector composed of the function to apply on the field, the field to extract from the event, and the event itself. Multiple conditions can be added by using :or or :and.

(where [:= :metric 4])

Here, we keep only events where the :metric field is equal to 4.

(where [:and [:= :host "foo"]
             [:> :metric 10])

Here, we keep only events with :host = foo and with :metric > 10

Set an event field to the given value.

(with :state "critical"
  (debug))

This example set the field :state to “critical” for events.

A map can also be provided:

(with {:service "foo" :state "critical"}
  (debug))

This example set the the field :service to “foo” and the field :state to “critical” for events.