func (lj *logstash) PublishEvent(
	signaler op.Signaler,
	opts outputs.Options,
	data outputs.Data,
) error {
	return lj.mode.PublishEvent(signaler, opts, data)
}
TODO: update Outputer interface to support multiple events for batch-like
processing (e.g. for filebeat). Batch like processing might reduce
send/receive overhead per event for other implementors too.
package logstash
import (
	"expvar"
	"time"
	"github.com/elastic/go-lumber/log"
	"github.com/elastic/beats/libbeat/common"
	"github.com/elastic/beats/libbeat/common/op"
	"github.com/elastic/beats/libbeat/logp"
	"github.com/elastic/beats/libbeat/outputs"
	"github.com/elastic/beats/libbeat/outputs/mode"
	"github.com/elastic/beats/libbeat/outputs/mode/modeutil"
	"github.com/elastic/beats/libbeat/outputs/transport"
)
var debug = logp.MakeDebug("logstash")
var (
	ackedEvents            = expvar.NewInt("libbeat.logstash.published_and_acked_events")
	eventsNotAcked         = expvar.NewInt("libbeat.logstash.published_but_not_acked_events")
	publishEventsCallCount = expvar.NewInt("libbeat.logstash.call_count.PublishEvents")
	statReadBytes   = expvar.NewInt("libbeat.logstash.publish.read_bytes")
	statWriteBytes  = expvar.NewInt("libbeat.logstash.publish.write_bytes")
	statReadErrors  = expvar.NewInt("libbeat.logstash.publish.read_errors")
	statWriteErrors = expvar.NewInt("libbeat.logstash.publish.write_errors")
)
const (
	defaultWaitRetry = 1 * time.Second
	defaultMaxWaitRetry = 60 * time.Second
)
func init() {
	log.Logger = logstashLogger{}
	outputs.RegisterOutputPlugin("logstash", new)
}
func new(beatName string, cfg *common.Config, _ int) (outputs.Outputer, error) {
	if !cfg.HasField("index") {
		cfg.SetString("index", -1, beatName)
	}
	output := &logstash{}
	if err := output.init(cfg); err != nil {
		return nil, err
	}
	return output, nil
}
type logstash struct {
	mode  mode.ConnectionMode
	index string
}
func (lj *logstash) init(cfg *common.Config) error {
	config := defaultConfig
	if err := cfg.Unpack(&config); err != nil {
		return err
	}
	tls, err := outputs.LoadTLSConfig(config.TLS)
	if err != nil {
		return err
	}
	transp := &transport.Config{
		Timeout: config.Timeout,
		Proxy:   &config.Proxy,
		TLS:     tls,
		Stats: &transport.IOStats{
			Read:        statReadBytes,
			Write:       statWriteBytes,
			ReadErrors:  statReadErrors,
			WriteErrors: statWriteErrors,
		},
	}
	logp.Info("Max Retries set to: %v", config.MaxRetries)
	m, err := initConnectionMode(cfg, &config, transp)
	if err != nil {
		return err
	}
	lj.mode = m
	lj.index = config.Index
	return nil
}
func initConnectionMode(
	cfg *common.Config,
	config *logstashConfig,
	transp *transport.Config,
) (mode.ConnectionMode, error) {
	sendRetries := config.MaxRetries
	maxAttempts := sendRetries + 1
	if sendRetries < 0 {
		maxAttempts = 0
	}
	settings := modeutil.Settings{
		Failover:     !config.LoadBalance,
		MaxAttempts:  maxAttempts,
		Timeout:      config.Timeout,
		WaitRetry:    defaultWaitRetry,
		MaxWaitRetry: defaultMaxWaitRetry,
	}
	if config.Pipelining == 0 {
		clients, err := modeutil.MakeClients(cfg, makeClientFactory(config, transp))
		if err != nil {
			return nil, err
		}
		return modeutil.NewConnectionMode(clients, settings)
	}
	clients, err := modeutil.MakeAsyncClients(cfg, makeAsyncClientFactory(config, transp))
	if err != nil {
		return nil, err
	}
	return modeutil.NewAsyncConnectionMode(clients, settings)
}
func makeClientFactory(
	cfg *logstashConfig,
	tcfg *transport.Config,
) modeutil.ClientFactory {
	compressLvl := cfg.CompressionLevel
	maxBulkSz := cfg.BulkMaxSize
	to := cfg.Timeout
	return func(host string) (mode.ProtocolClient, error) {
		t, err := transport.NewClient(tcfg, "tcp", host, cfg.Port)
		if err != nil {
			return nil, err
		}
		return newLumberjackClient(t, compressLvl, maxBulkSz, to, cfg.Index)
	}
}
func makeAsyncClientFactory(
	cfg *logstashConfig,
	tcfg *transport.Config,
) modeutil.AsyncClientFactory {
	compressLvl := cfg.CompressionLevel
	maxBulkSz := cfg.BulkMaxSize
	queueSize := cfg.Pipelining - 1
	to := cfg.Timeout
	return func(host string) (mode.AsyncProtocolClient, error) {
		t, err := transport.NewClient(tcfg, "tcp", host, cfg.Port)
		if err != nil {
			return nil, err
		}
		return newAsyncLumberjackClient(t, queueSize, compressLvl, maxBulkSz, to, cfg.Index)
	}
}
func (lj *logstash) Close() error {
	return lj.mode.Close()
}