Add index support in test mode

[mcorbin]

Hi,

I want to implement the possibility to test/query the index in test mode. Actually, streams like maintenance-mode are not testable.

My last attempt to do it was #821, and i have tried to do something better but i have trouble doing it.

To test the index we need riemann.config/core defined, because the index stream update the index in this core. Actually, only riemann.config/next-core is defined (because in test mode riemann.config/apply! is not called).
Furthermore, we cannot import riemann.config in riemann.test because of cyclic dependency.

I tried that:

In test.clj, i defined a test-core variable instead of streams

(def ^:dynamic *test-core*
  nil)

Now, in bin.clj, i added:

(reset! riemann.config/core @riemann.config/next-core)
;; bind *test-core* instead of *streams*
(binding [test/*test-core* @config/core]
   ...
)

Now, i have riemann.config/core and riemann.test/*test-core* defined, and i can use *test-core* in test.clj (for example in inject! with (inject! (:streams *test-core*) events)).
I can also clear the index between each tests, and can query the index in tests.

I also tried to add expiration support in test mode. The reaper uses (Thread/sleep interval), so it won't work.

My first tought was to define a (periodically-expire interval) function in test.clj. Once called (in a test for example), it will schedule using riemann.time/every! a sort of reaper, expiring events.
Unfortunately, inject! call (time.controlled/reset-time!) so my scheduled task is cancelled.

And this is where i am, and i'm a bit stuck, and will really appreciate help/guidance. I think Riemann really need this feature.

[mcorbin]

So, i maybe have a solution, i hope it will be good enough. First, i created a *test-core* variable like i explained before.
Like i said, inject! call (time.controlled/reset-time!) so i cannot schedule a "reaper" outside of it.

So i defined a new variable in test.clj (yeah more state...) :

(def ^:dynamic *test-config*
  {})

This variable will contains the configuration for each test (example soon).

Now, i added in inject!, after reset-time! these lines:

(when-let [periodically-expire-opts (:periodically-expire *test-config*)]
         (periodically-expire-test periodically-expire-opts streams))

If in my *test-config* variable the :periodically-expire key is defined, i call periodically-expire-test. periodically-expire-test schedule a task (with every!) that expire events and send them in streams.

I wrote a macro with-test-config to easily set *test-config*. I can now use the index in test mode, and test expired events:

(streams
 (where (not (expired? event))
   (index))
 (where (expired? event)
  (tap :expired)))

(tests
 (deftest test-expired
   (with-test-config {:periodically-expire {:keep-keys [:host :service :tags]}}
     (let [result (inject! [{:host "foo" :tags ["t1"] :time 0 :ttl 10}
                            {:host "bar" :time 15 :ttl 10}
                            {:host "baz" :time 30 :ttl 10}
                            {:host "foobar" :time 40 :ttl 10}
                            ])]
       (is (=(:expired result)
              [{:host "foo", :tags ["t1"], :state "expired", :time 20}
               {:host "bar", :state "expired", :time 30}]))))))

I can also test the maintenance-mode? stream:

(defn maintenance-mode?
  "Is Riemann currently in maintenance mode?"
  []
  ; Take an expression representing a query for maintenance mode
  (->> '(and (= :host nil)
             (= :service "maintenance-mode"))
       ; Search the current Riemann core's index for any matching events
       (riemann.index/search (:index @core))
       ; Take the first match
       first
       ; Find its state
       :state
       ; Is it the string "active"?
       (= "active")))

(streams
 (where (host "maintenance")
   ;; index a maintenance event
   (with {:host nil
          :service "maintenance-mode"}
     (index)))
 (where (and (not (host "maintenance"))
             (not (maintenance-mode?)))
   ;; index all events excepts maintenance events
   (tap :maintenance-mode)))

(tests
 (deftest maintenance-test
   (let [result (inject! [{:host "foo" :service "a" :time 0}
                          {:host "bar" :service "a" :time 1}
                          ;; enable maintenance
                          {:host "maintenance" :state "active" :time 2}
                          ;; these events will not be present in the tap
                          {:host "bar" :service "b" :time 3}
                          {:host "bar" :service "c" :time 4}
                          ;; disable maintenance
                          {:host "maintenance" :state "not active" :time 5}
                          {:host "bar" :service "d" :time 6}])]
     (is (= (:maintenance-mode result)
            [{:host "foo" :service "a" :time 0}
             {:host "bar" :service "a" :time 1}
             {:host "bar" :service "d" :time 6}])))))

Also, reinject will work in test mode.

What do you think about it ?

[mcorbin]

A new PR with a new approach ;)

I first did the same thing than in my previous PR: using a core in test mode and not just streams.

The Riemann reaper (used to expire events) is actually a service (a thread-service to be exact). It uses Thread/sleep to periodically expire events.

The problem was during tests. In test mode, side effects (like firing a batch/fixed-time-window) are handled by riemann.time.controlled, but the reaper was not a task running on the Riemann scheduler.

So, i rewrote the reaper to be a task. I created a new namespace (riemann.reaper) because i had issues with cyclic dependencies.

(periodically-expire) will now start a reaper in the Riemann scheduler, expiring events.

But i had a new problem. The reaper is now a task, but in test mode, the inject! function removes all tasks from the Riemann scheduler (to prevents side effects between tests). I had to start the reaper in the inject function, with the user configuration.

That's why i created a test-config variable in riemann.test, and did this in periodically-expire:

(defn periodically-expire
  "Sets up a reaper for this core. See riemann.reaper/reaper."
  ([]
   (periodically-expire 10))
  ([& args]
   (if test/*testing*
     (swap! test/test-config (fn [state] (assoc state :periodically-expire args)))
     (reaper/reaper args core))))

In test mode, i update test-config with the periodically-expire params.

Now, a fresh reaper is started at each inject, with the correct configuration, and events expired when the time advance. No macro like in the previous PR, it just works.

It could breaks existing user tests, because users should now take into account events expiration. They can also use (io) before indexing events to prevents it.

I hope this is my final attempt to implement index testing in test mode :D

[mcorbin]

Done ! ;)