Add mode to generate event per line of command output

_meta/beat.full.yml

@@ -20,3 +20,6 @@ execbeat:
       # the type defines the document type these entries should be stored
       # in. Default: execbeat
       document_type: execbeat
+
+      # Send message for each line of output
+      line_mode: false

_meta/beat.yml

@@ -20,3 +20,6 @@ execbeat:
       # the type defines the document type these entries should be stored
       # in. Default: execbeat
       #document_type:
+
+      # Send message for each line of output
+      #line_mode: false

_meta/fields.yml

@@ -23,3 +23,30 @@
           type: keyword
           description: >
             Exit code of the command executed by Execbeat.
+    - name: line
+      type: group
+      fields:
+        - name: command
+          type: keyword
+          description: >
+            The command executed by Execbeat.
+
+        - name: line
+          type: keyword
+          description: >
+            Single Line from output produced by the command executed by Execbeat.
+
+        - name: line_number
+          type: long
+          description: >
+            Line number of the line from output produced by the command executed by Execbeat.
+
+        - name: source
+          type: keyword
+          description: >
+            Source of the line. Either stdout or stderr.
+
+        - name: exitCode
+          type: keyword
+          description: >
+            Exit code of the command executed by Execbeat.

_meta/kibana/index-pattern/execbeat.json

@@ -1,5 +1,5 @@
 {
-  "fields": "[{\"count\": 0, \"analyzed\": false, \"aggregatable\": true, \"name\": \"beat.name\", \"searchable\": true, \"indexed\": true, \"doc_values\": true, \"type\": \"string\", \"scripted\": false}, {\"count\": 0, \"analyzed\": false, \"aggregatable\": true, \"name\": \"beat.hostname\", \"searchable\": true, \"indexed\": true, \"doc_values\": true, \"type\": \"string\", \"scripted\": false}, {\"count\": 0, \"analyzed\": false, \"aggregatable\": true, \"name\": \"beat.version\", \"searchable\": true, \"indexed\": true, \"doc_values\": true, \"type\": \"string\", \"scripted\": false}, {\"count\": 0, \"analyzed\": false, \"aggregatable\": true, \"name\": \"@timestamp\", \"searchable\": true, \"indexed\": true, \"doc_values\": true, \"type\": \"date\", \"scripted\": false}, {\"count\": 0, \"analyzed\": false, \"aggregatable\": true, \"name\": \"tags\", \"searchable\": true, \"indexed\": true, \"doc_values\": true, \"type\": \"string\", \"scripted\": false}, {\"count\": 0, \"analyzed\": false, \"aggregatable\": true, \"name\": \"fields\", \"searchable\": true, \"indexed\": true, \"doc_values\": true, \"scripted\": false}, {\"count\": 0, \"analyzed\": false, \"aggregatable\": true, \"name\": \"meta.cloud.provider\", \"searchable\": true, \"indexed\": true, \"doc_values\": true, \"type\": \"string\", \"scripted\": false}, {\"count\": 0, \"analyzed\": false, \"aggregatable\": true, \"name\": \"meta.cloud.instance_id\", \"searchable\": true, \"indexed\": true, \"doc_values\": true, \"type\": \"string\", \"scripted\": false}, {\"count\": 0, \"analyzed\": false, \"aggregatable\": true, \"name\": \"meta.cloud.machine_type\", \"searchable\": true, \"indexed\": true, \"doc_values\": true, \"type\": \"string\", \"scripted\": false}, {\"count\": 0, \"analyzed\": false, \"aggregatable\": true, \"name\": \"meta.cloud.availability_zone\", \"searchable\": true, \"indexed\": true, \"doc_values\": true, \"type\": \"string\", \"scripted\": false}, {\"count\": 0, \"analyzed\": false, \"aggregatable\": true, \"name\": \"meta.cloud.project_id\", \"searchable\": true, \"indexed\": true, \"doc_values\": true, \"type\": \"string\", \"scripted\": false}, {\"count\": 0, \"analyzed\": false, \"aggregatable\": true, \"name\": \"meta.cloud.region\", \"searchable\": true, \"indexed\": true, \"doc_values\": true, \"type\": \"string\", \"scripted\": false}, {\"count\": 0, \"analyzed\": false, \"aggregatable\": true, \"name\": \"exec.command\", \"searchable\": true, \"indexed\": true, \"doc_values\": true, \"type\": \"string\", \"scripted\": false}, {\"count\": 0, \"analyzed\": false, \"aggregatable\": true, \"name\": \"exec.stdout\", \"searchable\": true, \"indexed\": true, \"doc_values\": true, \"type\": \"string\", \"scripted\": false}, {\"count\": 0, \"analyzed\": false, \"aggregatable\": true, \"name\": \"exec.stderr\", \"searchable\": true, \"indexed\": true, \"doc_values\": true, \"type\": \"string\", \"scripted\": false}, {\"count\": 0, \"analyzed\": false, \"aggregatable\": true, \"name\": \"exec.exitCode\", \"searchable\": true, \"indexed\": true, \"doc_values\": true, \"type\": \"string\", \"scripted\": false}]", 
+  "fields": "[{\"count\": 0, \"analyzed\": false, \"aggregatable\": true, \"name\": \"beat.name\", \"searchable\": true, \"indexed\": true, \"doc_values\": true, \"type\": \"string\", \"scripted\": false}, {\"count\": 0, \"analyzed\": false, \"aggregatable\": true, \"name\": \"beat.hostname\", \"searchable\": true, \"indexed\": true, \"doc_values\": true, \"type\": \"string\", \"scripted\": false}, {\"count\": 0, \"analyzed\": false, \"aggregatable\": true, \"name\": \"beat.version\", \"searchable\": true, \"indexed\": true, \"doc_values\": true, \"type\": \"string\", \"scripted\": false}, {\"count\": 0, \"analyzed\": false, \"aggregatable\": true, \"name\": \"@timestamp\", \"searchable\": true, \"indexed\": true, \"doc_values\": true, \"type\": \"date\", \"scripted\": false}, {\"count\": 0, \"analyzed\": false, \"aggregatable\": true, \"name\": \"tags\", \"searchable\": true, \"indexed\": true, \"doc_values\": true, \"type\": \"string\", \"scripted\": false}, {\"count\": 0, \"analyzed\": false, \"aggregatable\": true, \"name\": \"fields\", \"searchable\": true, \"indexed\": true, \"doc_values\": true, \"scripted\": false}, {\"count\": 0, \"analyzed\": false, \"aggregatable\": true, \"name\": \"meta.cloud.provider\", \"searchable\": true, \"indexed\": true, \"doc_values\": true, \"type\": \"string\", \"scripted\": false}, {\"count\": 0, \"analyzed\": false, \"aggregatable\": true, \"name\": \"meta.cloud.instance_id\", \"searchable\": true, \"indexed\": true, \"doc_values\": true, \"type\": \"string\", \"scripted\": false}, {\"count\": 0, \"analyzed\": false, \"aggregatable\": true, \"name\": \"meta.cloud.machine_type\", \"searchable\": true, \"indexed\": true, \"doc_values\": true, \"type\": \"string\", \"scripted\": false}, {\"count\": 0, \"analyzed\": false, \"aggregatable\": true, \"name\": \"meta.cloud.availability_zone\", \"searchable\": true, \"indexed\": true, \"doc_values\": true, \"type\": \"string\", \"scripted\": false}, {\"count\": 0, \"analyzed\": false, \"aggregatable\": true, \"name\": \"meta.cloud.project_id\", \"searchable\": true, \"indexed\": true, \"doc_values\": true, \"type\": \"string\", \"scripted\": false}, {\"count\": 0, \"analyzed\": false, \"aggregatable\": true, \"name\": \"meta.cloud.region\", \"searchable\": true, \"indexed\": true, \"doc_values\": true, \"type\": \"string\", \"scripted\": false}, {\"count\": 0, \"analyzed\": false, \"aggregatable\": true, \"name\": \"exec.command\", \"searchable\": true, \"indexed\": true, \"doc_values\": true, \"type\": \"string\", \"scripted\": false}, {\"count\": 0, \"analyzed\": false, \"aggregatable\": true, \"name\": \"exec.stdout\", \"searchable\": true, \"indexed\": true, \"doc_values\": true, \"type\": \"string\", \"scripted\": false}, {\"count\": 0, \"analyzed\": false, \"aggregatable\": true, \"name\": \"exec.stderr\", \"searchable\": true, \"indexed\": true, \"doc_values\": true, \"type\": \"string\", \"scripted\": false}, {\"count\": 0, \"analyzed\": false, \"aggregatable\": true, \"name\": \"exec.exitCode\", \"searchable\": true, \"indexed\": true, \"doc_values\": true, \"type\": \"string\", \"scripted\": false}, {\"count\": 0, \"analyzed\": false, \"aggregatable\": true, \"name\": \"line.command\", \"searchable\": true, \"indexed\": true, \"doc_values\": true, \"type\": \"string\", \"scripted\": false}, {\"count\": 0, \"analyzed\": false, \"aggregatable\": true, \"name\": \"line.line\", \"searchable\": true, \"indexed\": true, \"doc_values\": true, \"type\": \"string\", \"scripted\": false}, {\"count\": 0, \"analyzed\": false, \"aggregatable\": true, \"name\": \"line.line_number\", \"searchable\": true, \"indexed\": true, \"doc_values\": true, \"type\": \"number\", \"scripted\": false}, {\"count\": 0, \"analyzed\": false, \"aggregatable\": true, \"name\": \"line.source\", \"searchable\": true, \"indexed\": true, \"doc_values\": true, \"type\": \"string\", \"scripted\": false}, {\"count\": 0, \"analyzed\": false, \"aggregatable\": true, \"name\": \"line.exitCode\", \"searchable\": true, \"indexed\": true, \"doc_values\": true, \"type\": \"string\", \"scripted\": false}]", 
   "fieldFormatMap": "{\"@timestamp\": {\"id\": \"date\"}}", 
   "timeFieldName": "@timestamp", 
   "title": "execbeat-*"

beater/execevent.go

@@ -9,7 +9,8 @@ type ExecEvent struct {
 	ReadTime     time.Time
 	DocumentType string
 	Fields       map[string]string
-	Exec         Exec
+	Exec         *Exec
+	Line         *Line
 }
 
 type Exec struct {
@@ -19,11 +20,24 @@ type Exec struct {
 	ExitCode int    `json:"exitCode"`
 }
 
+type Line struct {
+	Command    string `json:"command,omitempty"`
+	Source     string `json:"source"`
+	LineNumber int    `json:"line_number"`
+	Line       string `json:"line"`
+	ExitCode   int    `json:"exitCode"`
+}
+
 func (h *ExecEvent) ToMapStr() common.MapStr {
 	event := common.MapStr{
 		"@timestamp": common.Time(h.ReadTime),
 		"type":       h.DocumentType,
-		"exec":       h.Exec,
+	}
+	if h.Exec != nil {
+		event["exec"] = h.Exec
+	}
+	if h.Line != nil {
+		event["line"] = h.Line
 	}
 
 	if h.Fields != nil {

beater/execevent_test.go

@@ -7,7 +7,7 @@ import (
 	"time"
 )
 
-func TestHttpEventToMapStr(t *testing.T) {
+func TestExecEventToMapStr(t *testing.T) {
 	now := time.Now()
 	fields := make(map[string]string)
 	fields["field1"] = "value1"
@@ -20,17 +20,52 @@ func TestHttpEventToMapStr(t *testing.T) {
 	event.Fields = fields
 	event.DocumentType = "test"
 	event.ReadTime = now
-	event.Exec = command
+	event.Exec = &command
 	mapStr := event.ToMapStr()
 	_, fieldsExist := mapStr["fields"]
 	assert.True(t, fieldsExist)
 	_, execExist := mapStr["exec"]
 	assert.True(t, execExist)
 	assert.Equal(t, "test", mapStr["type"])
 	assert.Equal(t, common.Time(now), mapStr["@timestamp"])
+
+	_, lineExist := mapStr["line"]
+	assert.False(t, lineExist)
+}
+
+func TestExecEventWithLineToMapStr(t *testing.T) {
+	now := time.Now()
+	fields := make(map[string]string)
+	fields["field1"] = "value1"
+	fields["field2"] = "value2"
+	command := Line{}
+	command.Command = "foo"
+	command.Line = "test"
+	command.Source = "test"
+	command.LineNumber = 1
+
+	event := ExecEvent{}
+	event.Fields = fields
+	event.DocumentType = "test"
+	event.ReadTime = now
+	event.Line = &command
+
+	mapStr := event.ToMapStr()
+
+	_, fieldsExist := mapStr["fields"]
+	assert.True(t, fieldsExist)
+
+	_, execExist := mapStr["exec"]
+	assert.False(t, execExist)
+
+	_, lineExist := mapStr["line"]
+	assert.True(t, lineExist)
+
+	assert.Equal(t, "test", mapStr["type"])
+	assert.Equal(t, common.Time(now), mapStr["@timestamp"])
 }
 
-func TestHttpEventToMapStrWIthEmptyFields(t *testing.T) {
+func TestExecEventToMapStrWIthEmptyFields(t *testing.T) {
 	event := ExecEvent{}
 	mapStr := event.ToMapStr()
 	_, fieldsExist := mapStr["fields"]

beater/executor.go

@@ -45,10 +45,38 @@ func (e *Executor) Run() {
 	}
 
 	cron := cron.New()
-	cron.AddFunc(e.schedule, func() { e.runOneTime() })
+	cron.AddFunc(e.schedule, func() {
+		e.runOneTime()
+	})
 	cron.Start()
 }
 
+func (e *Executor) sendLines(buf bytes.Buffer, source string, cmdName string, exitCode int, now time.Time) {
+	n := 0
+	for _, s := range strings.Split(buf.String(), "\n") {
+		if len(s) > 0 {
+			lineEvent := Line{
+				Command:    cmdName,
+				Source:     source,
+				LineNumber: n,
+				Line:       s,
+				ExitCode:   exitCode,
+			}
+
+			event := ExecEvent{
+				ReadTime:     now,
+				DocumentType: e.documentType,
+				Fields:       e.config.Fields,
+				Line:         &lineEvent,
+			}
+
+			e.execbeat.client.PublishEvent(event.ToMapStr())
+
+			n += 1
+		}
+	}
+}
+
 func (e *Executor) runOneTime() error {
 	var cmd *exec.Cmd
 	var cmdArgs []string
@@ -92,21 +120,26 @@ func (e *Executor) runOneTime() error {
 		}
 	}
 
-	commandEvent := Exec{
-		Command:  cmdName,
-		StdOut:   stdout.String(),
-		StdErr:   stderr.String(),
-		ExitCode: exitCode,
-	}
+	if e.config.LineMode {
+		e.sendLines(stdout, "stdout", cmdName, exitCode, now)
+		e.sendLines(stderr, "stderr", cmdName, exitCode, now)
+	} else {
+		commandEvent := Exec{
+			Command:  cmdName,
+			StdOut:   stdout.String(),
+			StdErr:   stderr.String(),
+			ExitCode: exitCode,
+		}
 
-	event := ExecEvent{
-		ReadTime:     now,
-		DocumentType: e.documentType,
-		Fields:       e.config.Fields,
-		Exec:         commandEvent,
-	}
+		event := ExecEvent{
+			ReadTime:     now,
+			DocumentType: e.documentType,
+			Fields:       e.config.Fields,
+			Exec:         &commandEvent,
+		}
 
-	e.execbeat.client.PublishEvent(event.ToMapStr())
+		e.execbeat.client.PublishEvent(event.ToMapStr())
+	}
 
 	return nil
 }

config/config.go

@@ -14,6 +14,7 @@ type ExecConfig struct {
 	Schedule     string
 	Command      string
 	Args         string
+	LineMode     bool              `config:"line_mode"`
 	DocumentType string            `config:"document_type"`
 	Fields       map[string]string `config:"fields"`
 }

docs/fields.asciidoc

@@ -160,3 +160,44 @@ type: keyword
 Exit code of the command executed by Execbeat.
 
 
+
+[float]
+=== line.command
+
+type: keyword
+
+The command executed by Execbeat.
+
+
+[float]
+=== line.line
+
+type: keyword
+
+Single Line from output produced by the command executed by Execbeat.
+
+
+[float]
+=== line.line_number
+
+type: long
+
+Line number of the line from output produced by the command executed by Execbeat.
+
+
+[float]
+=== line.source
+
+type: keyword
+
+Source of the line. Either stdout or stderr.
+
+
+[float]
+=== line.exitCode
+
+type: keyword
+
+Exit code of the command executed by Execbeat.
+
+

execbeat.full.yml

@@ -21,6 +21,9 @@ execbeat:
       # in. Default: execbeat
       document_type: execbeat
 
+      # Send message for each line of output
+      line_mode: false
+
 #================================ General ======================================
 
 # The name of the shipper that publishes the network data. It can be used to group

execbeat.template-es2x.json

@@ -68,6 +68,33 @@
             }
           }
         },
+        "line": {
+          "properties": {
+            "command": {
+              "ignore_above": 1024,
+              "index": "not_analyzed",
+              "type": "string"
+            },
+            "exitCode": {
+              "ignore_above": 1024,
+              "index": "not_analyzed",
+              "type": "string"
+            },
+            "line": {
+              "ignore_above": 1024,
+              "index": "not_analyzed",
+              "type": "string"
+            },
+            "line_number": {
+              "type": "long"
+            },
+            "source": {
+              "ignore_above": 1024,
+              "index": "not_analyzed",
+              "type": "string"
+            }
+          }
+        },
         "meta": {
           "properties": {
             "cloud": {

execbeat.template.json

@@ -58,6 +58,29 @@
             }
           }
         },
+        "line": {
+          "properties": {
+            "command": {
+              "ignore_above": 1024,
+              "type": "keyword"
+            },
+            "exitCode": {
+              "ignore_above": 1024,
+              "type": "keyword"
+            },
+            "line": {
+              "ignore_above": 1024,
+              "type": "keyword"
+            },
+            "line_number": {
+              "type": "long"
+            },
+            "source": {
+              "ignore_above": 1024,
+              "type": "keyword"
+            }
+          }
+        },
         "meta": {
           "properties": {
             "cloud": {

execbeat.yml

@@ -21,6 +21,9 @@ execbeat:
       # in. Default: execbeat
       #document_type:
 
+      # Send message for each line of output
+      #line_mode: false
+
 #================================ General =====================================
 
 # The name of the shipper that publishes the network data. It can be used to group