Logstash output: shrinkWindow is unexpectedly called twice?

spacewander · August 31, 2016, 9:57am

Hey guys:

I have just read the code in:

elastic/beats/blob/master/libbeat/outputs/logstash/async.go

package logstash

import (
	"net"
	"time"

	"github.com/elastic/beats/libbeat/beat"
	"github.com/elastic/beats/libbeat/common/atomic"
	"github.com/elastic/beats/libbeat/logp"
	"github.com/elastic/beats/libbeat/outputs"
	"github.com/elastic/beats/libbeat/outputs/transport"
	"github.com/elastic/beats/libbeat/publisher"
	"github.com/elastic/go-lumber/client/v2"
)

type asyncClient struct {
	*transport.Client
	observer outputs.Observer
	client   *v2.AsyncClient
	win      *window

This file has been truncated. show original

, and find something strange here:

github.com

elastic/beats/blob/74c1d512694e02d1d7134d6f6ee42dce9ca02d9c/libbeat/outputs/logstash/async.go#L111


	ref := &msgRef{
		count:     1,
		batch:     data,
		batchSize: len(data),
		win:       &c.win,
		cb:        cb,
		err:       nil,
	}


	for len(data) > 0 {
		n, err := c.publishWindowed(ref, data)


		debug("%v events out of %v events sent to logstash. Continue sending",
			n, len(data))


		data = data[n:]
		if err != nil {
			c.win.shrinkWindow()
			_ = c.Close()


			logp.Err("Failed to publish events caused by: %v", err)

Once c.publishWindowed returns an err, the send window will be shrinked by this line:

github.com

elastic/beats/blob/74c1d512694e02d1d7134d6f6ee42dce9ca02d9c/libbeat/outputs/logstash/async.go#L118


	}


	for len(data) > 0 {
		n, err := c.publishWindowed(ref, data)


		debug("%v events out of %v events sent to logstash. Continue sending",
			n, len(data))


		data = data[n:]
		if err != nil {
			c.win.shrinkWindow()
			_ = c.Close()


			logp.Err("Failed to publish events caused by: %v", err)
			eventsNotAcked.Add(int64(len(data)))
			return err
		}
	}
	ref.dec()


	return nil

However, there will be an asynchronous call in:

github.com

elastic/beats/blob/74c1d512694e02d1d7134d6f6ee42dce9ca02d9c/libbeat/outputs/logstash/async.go#L177


	}
}


func (r *msgRef) done(n uint32) {
	ackedEvents.Add(int64(n))
	r.batch = r.batch[n:]
	r.win.tryGrowWindow(r.batchSize)
	r.dec()
}


func (r *msgRef) fail(n uint32, err error) {
	ackedEvents.Add(int64(n))
	r.err = err
	r.batch = r.batch[n:]
	r.win.shrinkWindow()
	r.dec()
}


func (r *msgRef) dec() {
	i := atomic.AddInt32(&r.count, -1)
	if i > 0 {

, which also shrinks the window.

Hence, if sending data to logstash failed, the window will be shrinked twice, one is synchronous and the other is asynchronous.

Are there any mistakes in my derivation? Is the behaviour expected or just a bug?

steffens · August 31, 2016, 11:04am

Have you got some logs confirming you observation? run beats in foreground with -e -d 'logstash' and kill running logstash.

The window is supposed to shrink with every batch(window) that has failed. Originally, a batch not being send by publishWindowed, will not be handled by async handler. That is (*msgReg).fail() was not be executed. But this changed at some point. I think you're right, shrinkWindow is executed twice.

Good news is, with beats input being rewritten recently, I'm planning to remove the windowing support. The windowing was used to figure an ok-batch size so to not having to reconnect every so often or continuously fail with logstash being overloaded. The rewrite sports a keep-alive signal, making the dynamic windowing in output plugin superfluous.

spacewander · September 1, 2016, 2:27am

Thank you for your reply.
The way to reproduce the problem:

Build the filebeat from master branch.

Modify filebeat.yml with

output.logstash:
  # The Logstash hosts
  hosts: ["localhost:5044"]
  loadbalance: true
  pipelining: 1
  bulk_max_size: 2

Then run logstash, and also filebeat
Kill the logstash
Get log like this:

2016/09/01 02:13:47.070020 async.go:138: DBG  Try to publish 2 events to logstash with window size 10
2016/09/01 02:13:47.657564 async.go:114: DBG  2 events out of 2 events sent to logstash. Continue sending
2016/09/01 02:13:47.657596 async.go:138: DBG  Try to publish 2 events to logstash with window size 2
2016/09/01 02:13:47.660661 async.go:114: DBG  0 events out of 2 events sent to logstash. Continue sending
2016/09/01 02:13:47.660685 async.go:68: DBG  close connection

It seems that window size shrinks twice, from 10 to 5, and then from 5 to 2, between two publishing.

steffens · September 1, 2016, 11:31am

Nice. You're totally right!!! Great you found this!

Also the eventsNotAcked counter might be wrong, as batch is reported to be failed twice. Once during error handling and once by callback. Shrinking window + updating counter should only happen in (*msgRef).dec(). Also (*msgRef).fail() is not reporting the failed send. Maybe the log message should be moved too.

You want to create a PR fixing this bug (I see you signed CA already)? Given you did dig through async IO-code finding it, I'd rather see your name next to the fix then mine. Honor to whom honor is due

system · September 21, 2016, 9:57am

This topic was automatically closed after 21 days. New replies are no longer allowed.