Fluentd: Fluentd accept protobuf messages

[marcosflobo]

Check CONTRIBUTING guideline first and here is the list to help us investigate the problem.

Is your feature request related to a problem? Please describe.

Nowadays, Fluentd does not support receive protobuf (Protocol Buffers) messages via HTTP. There are many use cases where Fluentd could be used in systems that "talk" each other via protobuf messages since this format is really lightweight.

I'm not talking about a protobuf parse, I mean just accept protobuf messages. Today, if you send to Fluentd a protobuf message via HTTP is says:

400 Bad Request
'json' or 'msgpack' parameter is required

Describe the solution you'd like

I'd like that, in a configuration similar to:

<source>
# Here I would receive the protobuf message
@type http
port 9880
bind 0.0.0.0
body_size_limit 32m
keepalive_timeout 10s
</source>

<match hello>
# And here I'd send the protobuf message to another endpoint
@type http

endpoint "http://localhost:9999"
open_timeout 2

<buffer>
flush_interval 1s
</buffer>
</match>

The source would accept messages in protobuf format too.

Describe alternatives you've considered

Since Fluentd does not accept protobuf messages, an alternative would use another product similar to Fluentd for doing the same.

Additional context

As nice to have, this feature request could be extended to FluentBit project too

[repeatedly]

http input plugin supports parser plugin. So if protobuf parser exists, you can use it for payload parsing.

https://docs.fluentd.org/input/http#handle-other-formats-using-parser-plugins

[marcosflobo]

Thank you for your answer.

I do not find the protobuf parser on that list.

In my case I do not want to parse the protobuf, since another service would receive the bytes and it will do the parse. I just want to send the bytes through the Fluentd.

[cosmo0920]

AFAIK, protobuf parser plugin does not exist.

@marcosflobo What version of your using protobuf? protobuf verison should exist 2 and 3.

I'm currently investigating how to create protobuf parser plugin.

In my case I do not want to parse the protobuf, since another service would receive the bytes and it will do the parse. I just want to send the bytes through the Fluentd.

Like logstash's one, Fluentd protobuf parser will require to handle protocol buffer definitions.

[marcosflobo]

@cosmo0920 in my use case, it's both version 2 and 3, we have different protobuf message versions for different use cases

[cosmo0920]

Thanks for the reply.
I'd released protbuf parser plugin which supports protobuf 2 and 3 for Fluentd v1.
https://github.com/cosmo0920/fluent-plugin-parser-protobuf
Note that this plugin requires to compile protobuf definitions into Ruby class format with Protocol Buffers compilers and load compiled Ruby class files.

[marcosflobo]

@cosmo0920 this is a great job, congratulations.

I have a couple of questions about this parser.

• A parse plugin will be executed after the input plugin. If this is true, how can you configure fluentd to accept protobuf messages in the input plugin, so after your parser will take over?
• Does your parser have any limitation in terms of Content-Type? There are a couple of Content-Type values allowed for protobuf, such as application/x-protobuf and application/protobuf
• After the protobuf is parsed by your parser, in which format the data is transformed? JSON?

[cosmo0920]

• A parse plugin will be executed after the input plugin. If this is true, how can you configure fluentd to accept protobuf messages in the input plugin, so after your parser will take over?

Please use it within <parse> directive.
https://docs.fluentd.org/input/http#less-than-parse-greater-than-directive

• Does your parser have any limitation in terms of Content-Type? There are a couple of Content-Type values allowed for protobuf, such as application/x-protobuf and application/protobuf

Parser plugin itself doesn't handle MIME type. This should be handled in in_http.
If you use <parse> directive, in_http refers <parse> directive settings:
https://github.com/fluent/fluentd/blob/master/lib/fluent/plugin/in_http.rb#L88

• After the protobuf is parsed by your parser, in which format the data is transformed? JSON?

Protobuf parses protbuf binary/string and then returns EventTime/Integer and Hash object pair.

[cosmo0920]

I've confirmed that protobuf parser plugin can be working with the following settings and protobuf class definition:

ProtoBuf definition

# cat test/data/protobuf3/simple.proto                                                                                                                                                  

syntax = "proto3";
import "google/protobuf/timestamp.proto";

message SearchRequest {
  string query = 1;
  int32 page_number = 2;
  int32 result_per_page = 3;
  enum Corpus {
    UNIVERSAL = 0;
    WEB = 1;
    IMAGES = 2;
    LOCAL = 3;
    NEWS = 4;
    PRODUCTS = 5;
    VIDEO = 6;
  }
  Corpus corpus = 4;
  google.protobuf.Timestamp timestamp = 5;
}

Compiled Ruby class from ProtoBuf definition

# cat test/data/protobuf3/simple_pb.rb                                                                                                                                                  
# Generated by the protocol buffer compiler.  DO NOT EDIT!
# source: simple.proto

require 'google/protobuf'

require 'google/protobuf/timestamp_pb'
Google::Protobuf::DescriptorPool.generated_pool.build do
  add_file("simple.proto", :syntax => :proto3) do
    add_message "SearchRequest" do
      optional :query, :string, 1
      optional :page_number, :int32, 2
      optional :result_per_page, :int32, 3
      optional :corpus, :enum, 4, "SearchRequest.Corpus"
      optional :timestamp, :message, 5, "google.protobuf.Timestamp"
    end
    add_enum "SearchRequest.Corpus" do
      value :UNIVERSAL, 0
      value :WEB, 1
      value :IMAGES, 2
      value :LOCAL, 3
      value :NEWS, 4
      value :PRODUCTS, 5
      value :VIDEO, 6
    end
  end
end

SearchRequest = ::Google::Protobuf::DescriptorPool.generated_pool.lookup("SearchRequest").msgclass
SearchRequest::Corpus = ::Google::Protobuf::DescriptorPool.generated_pool.lookup("SearchRequest.Corpus").enummodule

Fluentd Configuration

# cat fluent.conf
<source>
  @type http
  port 8080
  <parse>
    @type protobuf
    class_name SearchRequest
    class_file "#{File.expand_path(File.join('test', 'data', 'protobuf3', 'simple_pb.rb'))}"
  </parse>
</source>
<match protobuf>
  @type stdout
</match>

Sending protobuf Example script

# cat test_http.rb
require 'google/protobuf'
require "net/http"

require_relative "test/data/protobuf3/simple_pb"

def encoded_simple_binary
  request = SearchRequest.new(query: "q=Fluentd",
                              page_number: 404,
                              result_per_page: 10,
                              corpus: :WEB,
                              timestamp: Time.now)
  SearchRequest.encode(request)
end

uri = URI.parse("http://localhost:8080/protobuf")
params = encoded_simple_binary
req = Net::HTTP.new(uri.host, uri.port)
req.post(uri.path, params.to_s)

Result

$ bundle exec ruby test_http.rb

Then, Fluentd gets:

% bundle exec fluentd -c fluent.conf -p lib/fluent/plugin
<snip>
2020-06-01 16:45:43 +0900 [info]: #0 fluentd worker is now running worker=0
2020-06-01 16:45:46.377515000 +0900 protobuf: {"query":"q=Fluentd","page_number":404,"result_per_page":10,"corpus":"WEB","timestamp":{"seconds":1590997546,"nanos":371940000}}

---

Note that protobuf parser can parse only one Protobuf class per <parse> directive. If you want to use multiple Protobuf classes parsing, you should define multiple http endpoints with <parse> directives.

Hope this helps.

[repeatedly]

Thanks @cosmo0920 ! Awesome work.
The issue is closed.

[marcosflobo]

@cosmo0920 thank you very much for your work on this, I'll give it a try to your parser and I'll give you feedback

[marcosflobo]

@cosmo0920 which dependencies do we have to include in fluentd to be able to use your parser with a .proto version 2?

[cosmo0920]

With configuration parameter:
https://github.com/fluent-plugins-nursery/fluent-plugin-parser-protobuf#for-protobuf-2