Protobuf issue

[bernd]

Hi,

i'm trying to send an event to riemann the following way but failing pretty bad:

logstash kafka output (with riemann codec) -> kafka topic -> kafka-riemann plugin (https://github.com/pyr/riemann-kafka) -> riemann

logstash codec code snippet:

    public
    def encode(event)
        riemann_event = Riemann::Event.new
        [...]

        riemann_message = Riemann::Message.new :events => [riemann_event]
        @on_event.call(event, riemann_message.encode_with_length)
    end # def encode                                                      

I'm getting the following exc on riemann side:

ERROR [2015-06-10 06:24:57,209] clojure-agent-send-off-pool-3 - riemann.plugin.kafka - could not decode protobuf msg
com.google.protobuf.InvalidProtocolBufferException: Protocol message contained an invalid tag (zero).
        at com.google.protobuf.InvalidProtocolBufferException.invalidTag(InvalidProtocolBufferException.java:89)
        at com.google.protobuf.CodedInputStream.readTag(CodedInputStream.java:108)
        at com.aphyr.riemann.Proto$Msg.<init>(Proto.java:3883)
        at com.aphyr.riemann.Proto$Msg.<init>(Proto.java:3847)
        at com.aphyr.riemann.Proto$Msg$1.parsePartialFrom(Proto.java:3970)
        at com.aphyr.riemann.Proto$Msg$1.parsePartialFrom(Proto.java:3965)
        at com.google.protobuf.AbstractParser.parsePartialFrom(AbstractParser.java:141)
        at com.google.protobuf.AbstractParser.parseFrom(AbstractParser.java:176)
        at com.google.protobuf.AbstractParser.parseFrom(AbstractParser.java:188)
        at com.google.protobuf.AbstractParser.parseFrom(AbstractParser.java:193)
        at com.google.protobuf.AbstractParser.parseFrom(AbstractParser.java:49)
        at com.aphyr.riemann.Proto$Msg.parseFrom(Proto.java:4227)
        at sun.reflect.NativeMethodAccessorImpl.invoke0(Native Method)
        at sun.reflect.NativeMethodAccessorImpl.invoke(NativeMethodAccessorImpl.java:57)
        at sun.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:43)
        at java.lang.reflect.Method.invoke(Method.java:606)
        at clojure.lang.Reflector.invokeMatchingMethod(Reflector.java:93)
        at clojure.lang.Reflector.invokeStaticMethod(Reflector.java:207)
        at riemann.plugin.kafka$safe_decode.invoke(kafka.clj:20)
        at riemann.plugin.kafka$start_kafka_thread$fn__450.invoke(kafka.clj:43)
        at clojure.core$binding_conveyor_fn$fn__4145.invoke(core.clj:1910)
        at clojure.lang.AFn.call(AFn.java:18)
        at java.util.concurrent.FutureTask.run(FutureTask.java:262)
        at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1145)
        at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:615)
        at java.lang.Thread.run(Thread.java:745)

Any idea/hint what could be the issue or how i could debug further?

Regards
Bernd

[N-H Project]

+1

I have a similar issue.
I'm sending events from CollectD to Kafka and everything works ok.
Then, I'm connecting Kafka to Riemann, but I keep getting this msg:
"riemann.plugin.kafka - could not decode protobuf msg . . .  

Protocol message tag had invalid wire type."
Once a while the error is :

"Protocol message end-group tag did not match expected tag."
or

"While parsing a protocol message, the input ended unexpectedly in the middle of a field.  This could mean either than the input has been truncated or that an embedded message misreported its own length."

I'd be glad for any help :)

Nitz

[Tahir Rauf]

+2

[krakatoa1987]

I have a similar problem here. Logstash input is serializing events as json so the riemann-kafka plugin fails with the same exception. I found this fork: https://github.com/weichuliu/riemann-kafka, which lets you specify a custom msg decoder, but still trying to figure out how to process json :D  

[Aphyr]

On 12/15/2015 10:47 AM, krakatoa1987 wrote:
> I have a similar problem here. Logstash input is serializing events as json so
> the riemann-kafka plugin fails with the same exception. I found this fork:
> https://github.com/weichuliu/riemann-kafka, which lets you specify a custom msg
> decoder, but still trying to figure out how to process json :D

Ah! Try cheshire, which comes packaged with Riemann by default. :)

https://github.com/dakrone/cheshire

--Kyle