Initial commit

.gitignore

@@ -0,0 +1,8 @@
+/.idea
+/build
+.DS_Store
+/saltbeat
+/saltbeat.test
+*.pyc
+*.iml
+vendor

.travis.yml

@@ -0,0 +1,43 @@
+sudo: required
+dist: trusty
+services:
+  - docker
+
+language: go
+
+go:
+  - 1.6
+
+os:
+  - linux
+  - osx
+
+env:
+  matrix:
+    - TARGETS="check"
+    - TARGETS="-C saltbeat testsuite"
+
+  global:
+    # Cross-compile for amd64 only to speed up testing.
+    - GOX_FLAGS="-arch amd64"
+
+addons:
+  apt:
+    packages:
+      - python-virtualenv
+
+before_install:
+  # Redo the travis setup but with the elastic/libbeat path. This is needed so the package path is correct
+  - mkdir -p $HOME/gopath/src/github.com/elastic/beats/
+  - rsync -az ${TRAVIS_BUILD_DIR}/ $HOME/gopath/src/github.com/elastic/beats/
+  - export TRAVIS_BUILD_DIR=$HOME/gopath/src/github.com/elastic/beats/
+  - cd $HOME/gopath/src/github.com/elastic/beats/
+
+install:
+  - true
+
+script:
+  - make $TARGETS
+
+after_success:
+  # Copy full.cov to coverage.txt because codecov.io requires this file

LICENSE

@@ -0,0 +1,13 @@
+Copyright (c) 2016 Martin Hoefling
+
+Licensed under the Apache License, Version 2.0 (the "License");
+you may not use this file except in compliance with the License.
+You may obtain a copy of the License at
+
+    http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and
+limitations under the License.

Makefile

@@ -0,0 +1,42 @@
+BEATNAME=saltbeat
+BEAT_DIR=github.com/martinhoefling
+SYSTEM_TESTS=false
+TEST_ENVIRONMENT=false
+ES_BEATS=./vendor/github.com/elastic/beats
+GOPACKAGES=$(shell glide novendor)
+PREFIX?=.
+
+# Path to the libbeat Makefile
+-include $(ES_BEATS)/libbeat/scripts/Makefile
+
+.PHONY: init
+init:
+	glide update --no-recursive
+	make update
+	git init
+
+.PHONY: commit
+commit:
+	git add README.md CONTRIBUTING.md
+	git commit -m "Initial commit"
+	git add LICENSE
+	git commit -m "Add the LICENSE"
+	git add .gitignore
+	git commit -m "Add git settings"
+	git add .
+	git reset -- .travis.yml
+	git commit -m "Add saltbeat"
+	git add .travis.yml
+	git commit -m "Add Travis CI"
+
+.PHONY: update-deps
+update-deps:
+	glide update --no-recursive --strip-vcs
+
+# This is called by the beats packer before building starts
+.PHONY: before-build
+before-build:
+
+# Create binary packages for the beat
+pack: create-packer
+	cd dev-tools/packer; make deps images saltbeat

README.md

@@ -0,0 +1,131 @@
+# Saltbeat
+
+Welcome to Saltbeat.
+
+Ensure that this folder is at the following location:
+`${GOPATH}/github.com/martinhoefling`
+
+## Getting Started with Saltbeat
+
+### Requirements
+
+* [Golang](https://golang.org/dl/) 1.6
+* [Glide](https://github.com/Masterminds/glide) >= 0.10.0
+
+### Init Project
+To get running with Saltbeat, run the following command:
+
+```
+make init
+```
+
+To commit the first version before you modify it, run:
+
+```
+make commit
+```
+
+It will create a clean git history for each major step. Note that you can always rewrite the history if you wish before pushing your changes.
+
+To push Saltbeat in the git repository, run the following commands:
+
+```
+git remote set-url origin https://github.com/martinhoefling/saltbeat
+git push origin master
+```
+
+For further development, check out the [beat developer guide](https://www.elastic.co/guide/en/beats/libbeat/current/new-beat.html).
+
+### Build
+
+To build the binary for Saltbeat run the command below. This will generate a binary
+in the same directory with the name saltbeat.
+
+```
+make
+```
+
+
+### Run
+
+To run Saltbeat with debugging output enabled, run:
+
+```
+./saltbeat -c saltbeat.yml -e -d "*"
+```
+
+
+### Test
+
+To test Saltbeat, run the following command:
+
+```
+make testsuite
+```
+
+alternatively:
+```
+make unit-tests
+make system-tests
+make integration-tests
+make coverage-report
+```
+
+The test coverage is reported in the folder `./build/coverage/`
+
+
+### Package
+
+To be able to package Saltbeat the requirements are as follows:
+
+ * [Docker Environment](https://docs.docker.com/engine/installation/) >= 1.10
+ * $GOPATH/bin must be part of $PATH: `export PATH=${PATH}:${GOPATH}/bin`
+
+To cross-compile and package Saltbeat for all supported platforms, run the following commands:
+
+```
+cd dev-tools/packer
+make deps
+make images
+make
+```
+
+### Update
+
+Each beat has a template for the mapping in elasticsearch and a documentation for the fields
+which is automatically generated based on `etc/fields.yml`.
+To generate etc/saltbeat.template.json and etc/saltbeat.asciidoc
+
+```
+make update
+```
+
+
+### Cleanup
+
+To clean  Saltbeat source code, run the following commands:
+
+```
+make fmt
+make simplify
+```
+
+To clean up the build directory and generated artifacts, run:
+
+```
+make clean
+```
+
+
+### Clone
+
+To clone Saltbeat from the git repository, run the following commands:
+
+```
+mkdir -p ${GOPATH}/github.com/martinhoefling
+cd ${GOPATH}/github.com/martinhoefling
+git clone https://github.com/martinhoefling/saltbeat
+```
+
+
+For further development, check out the [beat developer guide](https://www.elastic.co/guide/en/beats/libbeat/current/new-beat.html).

beater/saltbeat.go

@@ -0,0 +1,153 @@
+package beater
+
+import (
+	"fmt"
+	"time"
+
+	"github.com/elastic/beats/libbeat/beat"
+	"github.com/elastic/beats/libbeat/common"
+	"github.com/elastic/beats/libbeat/logp"
+	"github.com/elastic/beats/libbeat/publisher"
+
+	"bytes"
+	"net"
+	"reflect"
+
+	"github.com/martinhoefling/saltbeat/config"
+	"github.com/ugorji/go/codec"
+)
+
+type Saltbeat struct {
+	beatConfig       *config.Config
+	done             chan struct{}
+	messages         chan map[string]interface{}
+	socketConnection *net.UnixConn
+	client           publisher.Client
+}
+
+// Creates beater
+func New() *Saltbeat {
+	logp.Debug("beater", "Creating new beater")
+	return &Saltbeat{
+		done:     make(chan struct{}),
+		messages: make(chan map[string]interface{}),
+	}
+}
+
+/// *** Beater interface methods ***///
+
+func (bt *Saltbeat) Config(b *beat.Beat) error {
+	logp.Debug("beater", "Configuring beater")
+
+	// Load beater beatConfig
+	err := b.RawConfig.Unpack(&bt.beatConfig)
+	if err != nil {
+		return fmt.Errorf("Error reading config file: %v", err)
+	}
+
+	return nil
+}
+
+func (bt *Saltbeat) Setup(b *beat.Beat) error {
+	logp.Debug("beater", "Setting up beater")
+	// Setting default period if not set
+	if bt.beatConfig.Saltbeat.MasterEventPub == "" {
+		bt.beatConfig.Saltbeat.MasterEventPub = "/var/run/salt/master/master_event_pub.ipc"
+	}
+	bt.client = b.Publisher.Connect()
+
+	var err error
+	logp.Info("Opening socket %s", bt.beatConfig.Saltbeat.MasterEventPub)
+	bt.socketConnection, err = net.DialUnix("unix", nil, &net.UnixAddr{bt.beatConfig.Saltbeat.MasterEventPub, "unix"})
+	if err != nil {
+		return err
+	}
+	err = bt.socketConnection.CloseWrite()
+	if err != nil {
+		return err
+	}
+
+	go func() {
+		var err error
+		var handle codec.MsgpackHandle
+		handle.MapType = reflect.TypeOf(map[string]interface{}(nil))
+		for {
+			logp.Debug("message", "Waiting for message")
+			message_decoder := codec.NewDecoder(bt.socketConnection, &handle)
+			var message map[string]interface{}
+			err = message_decoder.Decode(&message)
+			if err != nil {
+				logp.WTF(err.Error())
+			}
+			logp.Debug("message", "Message read")
+			bt.messages <- message
+		}
+	}()
+
+	return nil
+}
+
+func parseMessage(handle codec.MsgpackHandle, message map[string]interface{}) (string, map[string]interface{}) {
+	body := message["body"].([]byte)
+	newline := byte(10)
+	splitted := bytes.SplitN(body, []byte{newline, newline}, 2)
+
+	tag := string(splitted[0])
+	logp.Debug("message", "Message tag is %s", tag)
+
+	payload_bytes := splitted[1]
+	payload_decoder := codec.NewDecoderBytes(payload_bytes, &handle)
+
+	var payload map[string]interface{}
+	err := payload_decoder.Decode(&payload)
+	if err != nil {
+		logp.WTF(err.Error())
+	}
+
+	logp.Debug("message", "Decoded payload is %s", payload)
+	return tag, payload
+}
+
+func (bt *Saltbeat) Run(b *beat.Beat) error {
+	logp.Info("saltbeat is running! Hit CTRL-C to stop it.")
+
+	var err error
+	var handle codec.MsgpackHandle
+	handle.MapType = reflect.TypeOf(map[string]interface{}(nil))
+	handle.RawToString = true
+
+	for {
+		select {
+		case <-bt.done:
+			return nil
+		case message := <-bt.messages:
+			tag, payload := parseMessage(handle, message)
+			logp.Debug("publish", "Publishing event")
+
+			event := common.MapStr{
+				"@timestamp": common.Time(time.Now()),
+				"type":       b.Name,
+				"tag":        tag,
+				"payload":    payload,
+			}
+
+			ok := bt.client.PublishEvent(event)
+			if !ok {
+				logp.Debug("publish", "Cannot publish event")
+				logp.WTF(err.Error())
+			}
+			logp.Debug("publish", "Published")
+		}
+	}
+}
+
+func (bt *Saltbeat) Cleanup(b *beat.Beat) error {
+	logp.Info("Closing socket %s", bt.beatConfig.Saltbeat.MasterEventPub)
+	bt.socketConnection.Close()
+	return nil
+}
+
+func (bt *Saltbeat) Stop() {
+	close(bt.done)
+	close(bt.messages)
+}