Reinject is causing a stack overflow

[Matt Spaulding]

I'm using the following config and wondering why my call to reinject events is causing a stack overflow.  The idea is that I want to pre-process all events and tag them so they don't get processed again.  Maybe there's a better way to do this.

(def email (mailer {:from "no-r...@example.com"}))

(def send-email

        (email "mye...@example.com"))

(def service-rules [{:service #"^cpu/percent-(.*)$" :rewrite "cpu $1"}

                    {:service #"^memory/percent-(.*)$" :rewrite "memory $1"}])

(defn service-rewrite-rules

        [rules]

        (let [matcher (fn [s1 s2] (if (string? s1) (= s1 s2) (re-find s1 s2)))]

                (fn [{:keys [service] :as e}]

                        (or (first

                                (for [{:keys [rewrite] :as rule} rules

                                      :when (matcher (:service rule) service)]

                                        (assoc e :service (if (string? (:service rule)) rewrite

                                                                (str/replace service (:service rule) rewrite)))))

                                e))))

(def hour-rollup

        (partial rollup 5 3600))

(def service-rewrite

        (service-rewrite-rules service-rules))

(let [index (index)]

        (streams

                (where (not (tagged "processed"))

                        service-rewrite (tag "processed") #(info %) reinject))

        (streams

                (tagged "processed"

                        (by [:host :service]

                                        (hour-rollup

                                                (where (service "cpu/percent-user") send-email)))))

        (streams

                (tagged "processed"

                        (by [:host :service]

                                        (hour-rollup

                                                (where (service "cpu user") send-email))))))

Stack overflow error:

WARN [2016-07-11 13:32:26,589] defaultEventExecutorGroup-2-1 - riemann.config - riemann.config$reinject@e3104e8 threw

java.lang.StackOverflowError: null

        at sun.misc.URLClassPath$JarLoader.getResource(URLClassPath.java:1034)

        at sun.misc.URLClassPath$JarLoader.getResource(URLClassPath.java:1013)

        at sun.misc.URLClassPath.getResource(URLClassPath.java:212)

        at java.net.URLClassLoader$1.run(URLClassLoader.java:365)

        at java.net.URLClassLoader$1.run(URLClassLoader.java:362)

        at java.security.AccessController.doPrivileged(Native Method)

        at java.net.URLClassLoader.findClass(URLClassLoader.java:361)

        at java.lang.ClassLoader.loadClass(ClassLoader.java:424)

        at sun.misc.Launcher$AppClassLoader.loadClass(Launcher.java:331)

        at java.lang.ClassLoader.loadClass(ClassLoader.java:357)

        at java.lang.Class.forName0(Native Method)

        at java.lang.Class.forName(Class.java:264)

        at ch.qos.logback.classic.spi.PackagingDataCalculator.bestEffortLoadClass(PackagingDataCalculator.java:239)

        at ch.qos.logback.classic.spi.PackagingDataCalculator.computeBySTEP(PackagingDataCalculator.java:138)

        at ch.qos.logback.classic.spi.PackagingDataCalculator.populateUncommonFrames(PackagingDataCalculator.java:113)

        at ch.qos.logback.classic.spi.PackagingDataCalculator.populateFrames(PackagingDataCalculator.java:105)

        at ch.qos.logback.classic.spi.PackagingDataCalculator.calculate(PackagingDataCalculator.java:57)

        at ch.qos.logback.classic.spi.ThrowableProxy.calculatePackagingData(ThrowableProxy.java:147)

        at ch.qos.logback.classic.spi.LoggingEvent.<init>(LoggingEvent.java:124)

        at ch.qos.logback.classic.Logger.buildLoggingEventAndAppend(Logger.java:440)

        at ch.qos.logback.classic.Logger.filterAndLog_0_Or3Plus(Logger.java:396)

        at ch.qos.logback.classic.Logger.warn(Logger.java:713)

        at clojure.tools.logging$eval3$fn__8.invoke(NO_SOURCE_FILE:0)

        at clojure.tools.logging.impl$fn__4800$G__4782__4811.invoke(impl.clj:16)

        at clojure.tools.logging$log_STAR_.invokeStatic(logging.clj:64)

        at clojure.tools.logging$log_STAR_.invoke(logging.clj:43)

        at riemann.config$eval57$stream__61$fn__66.invoke(riemann.config:39)

        at riemann.config$eval57$stream__61.invoke(riemann.config:39)

        at riemann.core$stream_BANG_$fn__8608.invoke(core.clj:19)

        at riemann.core$stream_BANG_.invokeStatic(core.clj:18)

        at riemann.core$stream_BANG_.invoke(core.clj:14)

        at riemann.config$reinject.invokeStatic(config.clj:230)

        at riemann.config$reinject.invoke(config.clj:225)

        at riemann.config$eval57$stream__61$fn__66.invoke(riemann.config:39)

        at riemann.config$eval57$stream__61.invoke(riemann.config:39)

        at riemann.core$stream_BANG_$fn__8608.invoke(core.clj:19)

        at riemann.core$stream_BANG_.invokeStatic(core.clj:18)

        at riemann.core$stream_BANG_.invoke(core.clj:14)

        at riemann.config$reinject.invokeStatic(config.clj:230)

        at riemann.config$reinject.invoke(config.clj:225)

        at riemann.config$eval57$stream__61$fn__66.invoke(riemann.config:39)

...........

[Aphyr]

On 07/11/2016 12:39 PM, Matt Spaulding wrote:
> I'm using the following config and wondering why my call to reinject events is
> causing a stack overflow. The idea is that I want to pre-process all events and
> tag them so they don't get processed again. Maybe there's a better way to do this.

(streams
(where (not (tagged "processed"))
service-rewrite (tag "processed") #(info %) reinject))

This is an infinite loop. You're sending unprocessed events to service-rewrite
and discarding them, tagging untagged events as processed and discarding
*those*, logging them, then sending the untagged events back to the top level.
http://riemann.io/howto.html#understanding-the-riemann-stream-model and
http://riemann.io/howto.html#reinject-events may be helpful.

--Kyle

[Matt Spaulding]

On Monday, July 11, 2016 at 10:50:04 AM UTC-7, aphyr wrote:

This is an infinite loop. You're sending unprocessed events to service-rewrite
and discarding them, tagging untagged events as processed and discarding
*those*, logging them, then sending the untagged events back to the top level.
http://riemann.io/howto.html#understanding-the-riemann-stream-model and
http://riemann.io/howto.html#reinject-events may be helpful.

Thanks for the explanation.  I've changed the code to this and it seems to be doing what I want now.

(streams
    (where (not (tagged "processed"))

        (tag "processed"
            (smap (comp #(info %) service-rewrite) reinject))))

[Aphyr]

On 07/11/2016 01:21 PM, Matt Spaulding wrote:
> (streams
> (where (not (tagged "processed"))
> (tag "processed"
> (smap (comp #(info %) service-rewrite) reinject))))

Since (info x) returns nil, this reinjects nothing. I think you want (smap
service-rewrite reinject).

--Kyle

[Matt Spaulding]

On Monday, July 11, 2016 at 11:23:53 AM UTC-7, aphyr wrote:

Since (info x) returns nil, this reinjects nothing. I think you want (smap
service-rewrite reinject).

Quite right.  I assumed since I was getting logs that everything was working.  Thanks for the help.