Backport: Add batch_read_size config to Winlogbeat (elastic#2642)

This configuration option allows users to control the number of event log records that are read, processed, and published in its event loop.

* Update changelog.

Backport of elastic#2641

Author: andrewkroh

CHANGELOG.asciidoc

@@ -47,6 +47,8 @@ https://github.com/elastic/beats/compare/v1.3.1...1.3[Check the HEAD diff]
 
 *Winlogbeat*
 
+- Add `event_logs.batch_read_size` configuration option. {pull}2642[2642]
+
 ==== Deprecated
 
 *Affecting all Beats*

winlogbeat/beat/winlogbeat.go

@@ -135,8 +135,9 @@ func (eb *Winlogbeat) Run(b *beat.Beat) error {
 		debugf("Initializing EventLog[%s]", eventLogConfig.Name)
 
 		eventLog, err := eventlog.New(eventlog.Config{
-			Name: eventLogConfig.Name,
-			API:  eventLogConfig.API,
+			Name:          eventLogConfig.Name,
+			API:           eventLogConfig.API,
+			BatchReadSize: eventLogConfig.BatchReadSize,
 		})
 		if err != nil {
 			return fmt.Errorf("Failed to create new event log for %s. %v",

winlogbeat/config/config.go

@@ -105,9 +105,10 @@ func (mc MetricsConfig) Validate() error {
 // EventLogConfig holds the configuration data that specifies which event logs
 // to monitor.
 type EventLogConfig struct {
-	Name        string
-	IgnoreOlder string `yaml:"ignore_older"`
-	API         string
+	Name          string
+	IgnoreOlder   string `yaml:"ignore_older"`
+	BatchReadSize int    `yaml:"batch_read_size"`
+	API           string
 }
 
 // Validate validates the EventLogConfig data and returns an error describing

winlogbeat/docs/reference/configuration/winlogbeat-options.asciidoc

@@ -1,3 +1,6 @@
+:vista_and_newer: This option is only available on operating systems +
+  supporting the Windows Event Log API (Microsoft Windows Vista and newer).
+
 [[configuration-winlogbeat-options]]
 === Winlogbeat
 
@@ -54,6 +57,17 @@ winlogbeat:
     - name: Application
 --------------------------------------------------------------------------------
 
+===== event_logs.batch_read_size
+
+The maximum number of event log records to read from the Windows API in a single
+batch. The default batch size is 100. *{vista_and_newer}*
+
+Winlogbeat starts a goroutine (a lightweight thread) to read from each
+individual event log. The goroutine reads a batch of event log records using the
+Windows API, applies any processors to the events, publishes them to the
+configured outputs, and waits for an acknowledgement from the outputs before
+reading additional event log records.
+
 [[configuration-winlogbeat-options-event_logs-name]]
 ===== event_logs.name
 

winlogbeat/eventlog/factory.go

@@ -10,6 +10,7 @@ import (
 type Config struct {
 	Name          string // Name of the event log or channel.
 	RemoteAddress string // Remote computer to connect to. Optional.
+	BatchReadSize int    // The number of events to read in one batch.
 
 	API string // Name of the API to use. Optional.
 }

winlogbeat/eventlog/wineventlog.go

@@ -138,10 +138,14 @@ func (l *winEventLog) Close() error {
 // newWinEventLog creates and returns a new EventLog for reading event logs
 // using the Windows Event Log.
 func newWinEventLog(c Config) (EventLog, error) {
+	if c.BatchReadSize <= 0 {
+		c.BatchReadSize = defaultMaxNumRead
+	}
+
 	return &winEventLog{
 		channelName:  c.Name,
 		remoteServer: c.RemoteAddress,
-		maxRead:      defaultMaxNumRead,
+		maxRead:      c.BatchReadSize,
 		renderBuf:    make([]byte, renderBufferSize),
 		logPrefix:    fmt.Sprintf("WinEventLog[%s]", c.Name),
 	}, nil

winlogbeat/eventlog/wineventlog_test.go

@@ -0,0 +1,54 @@
+// +build windows
+
+package eventlog
+
+import (
+	"testing"
+
+	"github.com/stretchr/testify/assert"
+)
+
+func TestWinEventLogBatchReadSize(t *testing.T) {
+	configureLogp()
+	log, err := initLog(providerName, sourceName, eventCreateMsgFile)
+	if err != nil {
+		t.Fatal(err)
+	}
+	defer func() {
+		err := uninstallLog(providerName, sourceName, log)
+		if err != nil {
+			t.Fatal(err)
+		}
+	}()
+
+	// Publish test messages:
+	for k, m := range messages {
+		err = log.Report(m.eventType, k, []string{m.message})
+		if err != nil {
+			t.Fatal(err)
+		}
+	}
+
+	batchReadSize := 2
+	eventlog, err := newWinEventLog(Config{Name: providerName, BatchReadSize: 2})
+	if err != nil {
+		t.Fatal(err)
+	}
+	err = eventlog.Open(0)
+	if err != nil {
+		t.Fatal(err)
+	}
+	defer func() {
+		err := eventlog.Close()
+		if err != nil {
+			t.Fatal(err)
+		}
+	}()
+
+	records, err := eventlog.Read()
+	if err != nil {
+		t.Fatal(err)
+	}
+
+	assert.Len(t, records, batchReadSize)
+}