out_http: Add option to reuse connections #4330

Garfield96 · 2023-10-22T12:53:44Z

Which issue(s) this PR fixes:
N/A

What this PR does / why we need it:
The HTTP output plugin recreates a connection for every HTTP request. In combination with TLS and high network latencies, this can result in very poor performance. This PR adds an option to reuse connections.

The implementation caches one connection per flush thread. The cache is part of the HTTP plugin instance and is implemented as an array. When a thread creates its first request, it gets assigned a slot (id) in this array and the id is stored in a thread local variable. Since this slot is exclusively used by a single thread, no synchronization is required. If the endpoint changes at runtime, the old connection will be closed and replaced by a new connection to the new endpoint. During shutdown, all open connections are finished in the close method.

This implementation works very well for a static endpoint. If the endpoint changes frequently, there will be little benefit. However, it can be assumed that most users use a static endpoint.

Benchmark
I tested the throughput with two Fluentd instances on a single machine. One instance acting as sender and the other as receiver. Even though the connection creation was very fast, because both instances were located on the same machine and TLS was not enabled, activating connection reuse doubled the throughput.

Sender configuration:

<source>
  @type sample
  sample {"hello":"world"}
  tag sample
  rate 5000
</source>

<match **>
  @type http

  endpoint http://localhost:9000
  open_timeout 2
  reuse_connections true
  <format>
    @type json
  </format>
  <buffer>
    flush_interval 10s
    chunk_limit_records 1
    flush_thread_count 32
  </buffer>
</match>

Receiver configuration:

<system>
  workers 1
</system>

<source>
  @type http
  port 9000
  bind 0.0.0.0
  body_size_limit 32m
  keepalive_timeout 10s
</source>

<match **>
  @type flowcounter_simple
  unit second
</match>

Docs Changes:
fluent/fluentd-docs-gitbook#501

Release Note:
out_http: Add option to reuse connections

Garfield96 · 2023-10-30T16:33:25Z

Hi @daipom,
are you fine with this change? If no bigger changes are required, I'll add some tests.

Garfield96 · 2023-11-17T16:33:19Z

Hi @daipom @ashie ,
I added a test case which shows that the feature works for the general case as well as for interrupted connections. Do you have an estimate when this feature can be released?

daipom · 2023-11-18T01:46:52Z

Thanks for this enhancement.
Sorry for my late response.
I was busy releasing v1.16.3 😢, but it was finally released on 11/14!

Now, I'm working on releasing fluent-package v5.0.2, which contains Fluentd v1.16.3.
It will be released at the end of this month.

Since this adds the new option, it will be released on v1.17.0 (fluent-package v5.1.0).
That date is not yet fixed, I think, but probably be early next year.
I'm sorry for not seeing the details yet, but I will see this for v1.17.0 this month or next month.

lib/fluent/plugin/out_http.rb

Garfield96 · 2024-01-13T21:59:28Z

@ashie @daipom Can you please review this PR as well as #4331? Thanks

daipom · 2024-01-15T08:34:53Z

Sorry for my late response.
I'm seeing this.

It does not affect the existing specifications.
It looks basically good.

However, there are a few points that concern me. I will comment on them later.

daipom

Sorry for my late response and the overall change request.

It looks good, but I believe we can use more simple logic.
Would you please consider if it is possible to use this kind of simple logic?

It is becoming difficult to maintain Fluentd due to the complexity of the code.
I want to keep the code as simple as possible.

Here are the main points.

Avoid nil initialization of the cache array.
Avoid managing the array index with mutex.
Share the logic make_request by using the &block argument.

If you have any concerns or ideas, please let me know.

daipom · 2024-01-23T11:13:12Z

lib/fluent/plugin/out_http.rb

+      @cache = []
+      @cache_id_mutex = Mutex.new
+      @cache_entry = Struct.new(:uri, :conn)


I believe that we can manage these caches more simply.

Suggested change

@cache = []

@cache_id_mutex = Mutex.new

@cache_entry = Struct.new(:uri, :conn)

@cache_by_thread = {}

Why do you define the struct in the constructor?
It would be normal to define it directly under the class.

class HTTPOutput < Output Fluent::Plugin.register_output('http', self) class RetryableResponse < StandardError; end CacheEntry = Struct.new(:uri, :conn) ...

Why do you define the struct in the constructor?
It would be normal to define it directly under the class.

I fixed this in d57402e

daipom · 2024-01-23T11:14:31Z

lib/fluent/plugin/out_http.rb

+      # Close all open connections
+      @cache.each {|entry| entry.conn.finish if entry.conn&.started? }


Suggested change

# Close all open connections

@cache.each {|entry| entry.conn.finish if entry.conn&.started? }

@cache_by_thread.each_value do |cache|

cache.conn.finish if cache.conn.started?

end

daipom · 2024-01-23T11:15:35Z

lib/fluent/plugin/out_http.rb

+      @cache = Array.new(actual_flush_thread_count, @cache_entry.new("", nil)) if @reuse_connections
+      @cache_id = 0
+


Suggested change

@cache = Array.new(actual_flush_thread_count, @cache_entry.new("", nil)) if @reuse_connections

@cache_id = 0

We should not use the struct with nil or empty values if possible.
It makes code management difficult.

lib/fluent/plugin/out_http.rb

daipom · 2024-01-23T11:20:33Z

lib/fluent/plugin/out_http.rb

+    def make_request_cached(uri, req)
+      id = Thread.current.thread_variable_get(plugin_id)
+      if id.nil?
+        @cache_id_mutex.synchronize {
+          id = @cache_id
+          @cache_id += 1
+        }
+        Thread.current.thread_variable_set(plugin_id, id)
+      end
+      uri_str = uri.to_s
+      if @cache[id].uri != uri_str
+        @cache[id].conn.finish if @cache[id].conn&.started?
+        http =  if @proxy_uri
+                  Net::HTTP.start(uri.host, uri.port, @proxy_uri.host, @proxy_uri.port, @proxy_uri.user, @proxy_uri.password, @http_opt)
+                else
+                  Net::HTTP.start(uri.host, uri.port, @http_opt)
+                end
+        @cache[id] = @cache_entry.new(uri_str, http)
+      end
+      @cache[id].conn.request(req)
+    end


Sorry for the overall change, but I believe we can use more simple logic like this.

Suggested change

def make_request_cached(uri, req)

id = Thread.current.thread_variable_get(plugin_id)

if id.nil?

@cache_id_mutex.synchronize {

id = @cache_id

@cache_id += 1

}

Thread.current.thread_variable_set(plugin_id, id)

end

uri_str = uri.to_s

if @cache[id].uri != uri_str

@cache[id].conn.finish if @cache[id].conn&.started?

http = if @proxy_uri

Net::HTTP.start(uri.host, uri.port, @proxy_uri.host, @proxy_uri.port, @proxy_uri.user, @proxy_uri.password, @http_opt)

else

Net::HTTP.start(uri.host, uri.port, @http_opt)

end

@cache[id] = @cache_entry.new(uri_str, http)

end

@cache[id].conn.request(req)

end

def make_request_cached(uri, req)

thread_name = Thread.current.name

@cache_by_thread[thread_name] ||= CacheEntry.new(uri.to_s, make_request(uri, req))

cache = @cache_by_thread[thread_name]

unless cache.uri == uri.to_s

cache.conn.finish if cache.conn.started?

@cache_by_thread[thread_name] = CacheEntry.new(uri.to_s, make_request(uri, req))

cache = @cache_by_thread[thread_name]

end

cache.conn.request(req)

end

Note: we can use the thread name flush_thread_{i} for the key.

fluentd/lib/fluent/plugin/output.rb

Lines 506 to 511 in 0893c39

@buffer_config.flush_thread_count.times do |i|

thread_title = "flush_thread_#{i}".to_sym

thread_state = FlushThreadState.new(nil, nil, Mutex.new, ConditionVariable.new)

thread = thread_create(thread_title) do

flush_thread_run(thread_state)

end

Garfield96 · 2024-02-04T11:27:24Z

Hi @daipom,

Thanks for the review. I agree that using a hash makes the code simpler. Unfortunately, hashes are not thread-safe in Ruby and inserting the CacheEntries could result in data-races. Therefore, we would again need a mutex. The array solution had the advantage, that the mutex was only used during the first request of a thread to retrieve the index. Afterwards, the code was lock-free.

daipom · 2024-02-05T01:43:24Z

@Garfield96
Thanks for considering!

Unfortunately, hashes are not thread-safe in Ruby and inserting the CacheEntries could result in data-races.

I think it's basically thread-safe, although some of the features are not thread-safe (e.g. default_proc (https://bugs.ruby-lang.org/issues/19237)).

||= in my suggestion seems to be not thread-safe, but Thread.current.name is a unique key to the thread.

So, I don't expect my suggestion to cause data races.
Have you actually checked the data race?

Signed-off-by: Christian Norbert Menges <[email protected]>

Signed-off-by: Daijiro Fukuda <[email protected]> Co-authored-by: Daijiro Fukuda <[email protected]>

daipom · 2024-04-30T03:23:49Z

Rebased to the latest master, and add my DCO.

Signed-off-by: Daijiro Fukuda <[email protected]>

daipom · 2024-04-30T03:48:49Z

I have added only the following additional fixes.

Move the cache struct definition outside of the constructor.
Make some parameter names more specific.

I have not added commits about the logic simplification that I'm suggesting because we don't have a consensus in the discussion about data race concerns (#4330 (comment)).

daipom

Looks good to me, although there are still some discussions about code simplification.

ashie · 2024-04-30T04:27:48Z

lib/fluent/plugin/out_http.rb

@@ -302,16 +318,41 @@ def create_request(chunk, uri)
      req
    end

+    def make_request_cached(uri, req)
+      id = Thread.current.thread_variable_get(plugin_id)


It's not appropriate to use plugin_id as a key, it should be used with a suffix like "#{plugin_id}_connection_cache_id".

In addition, we prefer to use Thread#[] instead of Thread#thread_variable_get if there is no particular reason.

Thanks. I agree.
Done in 9ce635c.

and following: * Use `Thread#[]` style * because it is the common for Fluentd code * Rename `connection_cache_id` to `connection_cache_next_id` * for clarity. Signed-off-by: Daijiro Fukuda <[email protected]>

ashie

LGTM

ashie · 2024-04-30T05:40:21Z

Thanks!

daipom · 2024-04-30T05:55:30Z

Thanks so much @Garfield96 !
Sorry for taking a long time. 😢

New feature of Fluentd v1.17.0. Related: fluent/fluentd#4330 Signed-off-by: Daijiro Fukuda <[email protected]>

daipom added the enhancement Feature request or improve operations label Oct 30, 2023

daipom added this to the v1.17.0 milestone Oct 30, 2023

Garfield96 force-pushed the out-http-reuse-connections branch from b00f90e to 7ba811d Compare November 6, 2023 12:17

ashie reviewed Nov 21, 2023

View reviewed changes

lib/fluent/plugin/out_http.rb Outdated Show resolved Hide resolved

Garfield96 requested a review from ashie December 17, 2023 21:27

daipom reviewed Jan 23, 2024

View reviewed changes

Garfield96 added 3 commits April 30, 2024 12:17

out_http: Add option to reuse connections

5844e96

Signed-off-by: Christian Norbert Menges <[email protected]>

out_http: Add test for connection recreation

7a5662a

Signed-off-by: Christian Norbert Menges <[email protected]>

out_http: Use snake case

fa5b183

Signed-off-by: Christian Norbert Menges <[email protected]>

daipom force-pushed the out-http-reuse-connections branch from 7bb7553 to fdf8914 Compare April 30, 2024 03:22

Apply suggestions from code review

af18531

Signed-off-by: Daijiro Fukuda <[email protected]> Co-authored-by: Daijiro Fukuda <[email protected]>

daipom force-pushed the out-http-reuse-connections branch from fdf8914 to af18531 Compare April 30, 2024 03:23

daipom added 2 commits April 30, 2024 12:31

Move struct definition outside of constructor

d57402e

Signed-off-by: Daijiro Fukuda <[email protected]>

Make some parameter names more specific

6763f3d

Signed-off-by: Daijiro Fukuda <[email protected]>

daipom approved these changes Apr 30, 2024

View reviewed changes

ashie reviewed Apr 30, 2024

View reviewed changes

Use more specific thread key name for cache-id

9ce635c

and following: * Use `Thread#[]` style * because it is the common for Fluentd code * Rename `connection_cache_id` to `connection_cache_next_id` * for clarity. Signed-off-by: Daijiro Fukuda <[email protected]>

ashie approved these changes Apr 30, 2024

View reviewed changes

ashie merged commit 8804a80 into fluent:master Apr 30, 2024
15 of 16 checks passed

daipom added a commit to fluent/fluentd-docs-gitbook that referenced this pull request May 15, 2024

out_http: Add new option: reuse_connections

a6b89c4

New feature of Fluentd v1.17.0. Related: fluent/fluentd#4330 Signed-off-by: Daijiro Fukuda <[email protected]>

daipom mentioned this pull request May 15, 2024

out_http: Add new option: reuse_connections fluent/fluentd-docs-gitbook#501

Merged

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

out_http: Add option to reuse connections #4330

out_http: Add option to reuse connections #4330

Garfield96 commented Oct 22, 2023 •

edited by daipom

Loading

Garfield96 commented Oct 30, 2023

Garfield96 commented Nov 17, 2023

daipom commented Nov 18, 2023

Garfield96 commented Jan 13, 2024

daipom commented Jan 15, 2024

daipom left a comment

daipom Jan 23, 2024

daipom Apr 30, 2024

daipom Jan 23, 2024

daipom Jan 23, 2024

daipom Jan 23, 2024

Garfield96 commented Feb 4, 2024

daipom commented Feb 5, 2024 •

edited

Loading

daipom commented Apr 30, 2024

daipom commented Apr 30, 2024

daipom left a comment

ashie Apr 30, 2024

daipom Apr 30, 2024

ashie left a comment

ashie commented Apr 30, 2024

daipom commented Apr 30, 2024

		# Close all open connections
		@cache.each {\|entry\| entry.conn.finish if entry.conn&.started? }

		@cache = Array.new(actual_flush_thread_count, @cache_entry.new("", nil)) if @reuse_connections
		@cache_id = 0

	@buffer_config.flush_thread_count.times do \|i\|
	thread_title = "flush_thread_#{i}".to_sym
	thread_state = FlushThreadState.new(nil, nil, Mutex.new, ConditionVariable.new)
	thread = thread_create(thread_title) do
	flush_thread_run(thread_state)
	end

out_http: Add option to reuse connections #4330

out_http: Add option to reuse connections #4330

Conversation

Garfield96 commented Oct 22, 2023 • edited by daipom Loading

Garfield96 commented Oct 30, 2023

Garfield96 commented Nov 17, 2023

daipom commented Nov 18, 2023

Garfield96 commented Jan 13, 2024

daipom commented Jan 15, 2024

daipom left a comment

Choose a reason for hiding this comment

daipom Jan 23, 2024

Choose a reason for hiding this comment

daipom Apr 30, 2024

Choose a reason for hiding this comment

daipom Jan 23, 2024

Choose a reason for hiding this comment

daipom Jan 23, 2024

Choose a reason for hiding this comment

daipom Jan 23, 2024

Choose a reason for hiding this comment

Garfield96 commented Feb 4, 2024

daipom commented Feb 5, 2024 • edited Loading

daipom commented Apr 30, 2024

daipom commented Apr 30, 2024

daipom left a comment

Choose a reason for hiding this comment

ashie Apr 30, 2024

Choose a reason for hiding this comment

daipom Apr 30, 2024

Choose a reason for hiding this comment

ashie left a comment

Choose a reason for hiding this comment

ashie commented Apr 30, 2024

daipom commented Apr 30, 2024

Garfield96 commented Oct 22, 2023 •

edited by daipom

Loading

daipom commented Feb 5, 2024 •

edited

Loading