benmur / riemann-scala-client Goto Github PK

I've run into an issue with roundtripping a Riemann event with a metric of type Long. The scenario is roughly as follows:

1. I create and send an Event with a metric of type Long.
2. I query for that event and receive a metric of type Float.
3. :(

I've debugged the issue and it appears to be caused by a difference in the way riemann-scala-client transcodes metrics versus https://github.com/aphyr/riemann-java-client. Note in particular that when writing a metric value of type Long, the Riemann Java client sets both the int64 and float metric values on the underlying protobuf. Why? I don't know. But it does. Therefore, when the Riemann server answers a query for an event with a long metric, it sets the float value as well.

When riemann-scala-client decodes the event protobuf, it looks for a float value before a long: https://github.com/benmur/riemann-scala-client/blob/master/src/main/scala/net/benmur/riemann/client/Serializers.scala#L53. And therein lies the confusion.

I looked carefully but I actually couldn't find a method in the Riemann Java client code which decodes an event protobuf's metric into a single canonical value. I guess it's up to the user of the library to decide how to decode metric values. However, I did find code within the Riemann Ruby client codebase which decodes an event protobuf's metric: https://github.com/aphyr/riemann-ruby-client/blob/master/lib/riemann/event.rb#L192. It looks for a double value first, then a long, then a float. In that scheme, a long metric encoded by the java client would indeed be decoded as a long (even though the float metric value exists as well in the underlying protobuf).

I've thought about how to rectify this. The easiest fix (by far) is to simply change the metric decoding to work like the java client. However, this would introduce an incompatibility with previous versions. (Perhaps a warning and a bump in version to 0.4.0 would suffice.)

A more ambitious fix would be to support both decoding schemes, at least as a bridge to cross-compatibility: a "legacy" mode preserving the previous decoding behavior and a "standard" or "java" mode emulating the java client behavior. This is something I've sketched out, but it's pretty ambitious and touches a lot of code.

What do you think?

I just restarted my Riemann server and saw my client app enter a failure state wherein every attempt to send data to Riemann failed, logging this:

2016-01-19 04:34:51,895 ERROR n.b.r.c.ReliableIO$TcpConnectionActor akka://foo/user/riemann-tcp-client-1/io - could not send or receive data
java.io.EOFException: null
        at java.io.DataInputStream.readInt(DataInputStream.java:392) ~[na:1.7.0_65]
        at net.benmur.riemann.client.ReliableIO$TcpConnectionActor$$anonfun$receive$2.applyOrElse(ReliableIO.scala:90) ~[foo-249.jar:0.0]
        at akka.actor.Actor$class.aroundReceive(Actor.scala:467) ~[foo-249.jar:0.0]
        at net.benmur.riemann.client.ReliableIO$TcpConnectionActor.aroundReceive(ReliableIO.scala:79) ~[foo-249.jar:0.0]
        at akka.actor.ActorCell.receiveMessage(ActorCell.scala:516) [foo-249.jar:0.0]
        at akka.actor.ActorCell.invoke(ActorCell.scala:487) [foo-249.jar:0.0]
        at akka.dispatch.Mailbox.processMailbox(Mailbox.scala:238) [foo-249.jar:0.0]
        at akka.dispatch.Mailbox.run(Mailbox.scala:220) [foo-249.jar:0.0]
        at akka.dispatch.ForkJoinExecutorConfigurator$AkkaForkJoinTask.exec(AbstractDispatcher.scala:397) [foo-249.jar:0.0]
        at scala.concurrent.forkjoin.ForkJoinTask.doExec(ForkJoinTask.java:260) [foo-249.jar:0.0]
        at scala.concurrent.forkjoin.ForkJoinPool$WorkQueue.runTask(ForkJoinPool.java:1339) [foo-249.jar:0.0]
        at scala.concurrent.forkjoin.ForkJoinPool.runWorker(ForkJoinPool.java:1979) [foo-249.jar:0.0]
        at scala.concurrent.forkjoin.ForkJoinWorkerThread.run(ForkJoinWorkerThread.java:107) [foo-249.jar:0.0]
2016-01-19 04:34:51,896 INFO  akka.actor.DeadLetterActorRef akka://foo/deadLetters - Message [com.aphyr.riemann.Proto$Msg] from Actor[akka://foo/user/riemann-tcp-client-1/io#916728015] to Actor[akka://foo/deadLetters] was not delivered. [270] dead letters encountered. This logging can be turned off or adjusted with configuration settings 'akka.log-dead-letters' and 'akka.log-dead-letters-during-shutdown'.

Restarting the client app fixed the problem, leading me to assume that there's some state in the Riemann client library that isn't getting reset after a connection fails.

riemann-scala-client version 0.3.4

that would be great. Thanks!

Will there be any updates on this repository?

Implement connection closing.

To allow fire-and-forget of small events, while making sending big events and querying possible.

Retry failed Writes after reconnecting (with a counter and backoff time, to be gentle)

Hi,

are you still actively pursuing this? Or did you let the project silently die? If so, what were the reasons? Did you loose interest or just hit a problem with it?

Jan

Currently there is no way to send event attribute values (key-value pairs) to Riemann.
Would be nice to have this (The java Proto.Event#Builder already contains setAttributes methods ).
I can create a PR to add support for this if you like.

Currently if the connection to Riemann fails more than twice within a one second window the associated IO actor is stopped and no further retries are attempted.

The following class, taken from the "Minimum viable use case" does not compile. Errors are below.
Using 0.3.3-SNAPSHOT.

package com.example.myproject

import net.benmur.riemann.client._
//import net.benmur.riemann.client.EventSenderDSL._
import RiemannClient._

import akka.actor.ActorSystem
import akka.util.Timeout

import java.net.InetSocketAddress
import java.util.concurrent.TimeUnit._

import com.twitter.inject.Logging

/**
 * Riemann proxy client.
 */
object Metrics extends Logging
{
  private implicit val system = ActorSystem()
  private implicit def timeout = Timeout(5, SECONDS)

  val metrics = riemannConnectAs[Unreliable] to new InetSocketAddress("localhost", 5555)
  service("service name") | state("warning") |>> metrics
}

compile
[info] Compiling 9 Scala sources to /.../target/scala-2.11/classes...
[error] /.../Metrics.scala:23: value |>> is not a member of net.benmur.riemann.client.EventPart
[error] service("service name") | state("warning") |>> metrics
[error] ^
[error] one error found
error Compilation failed
[error] Total time: 2 s, completed Jul 22, 2015 10:56:48 AM