tcp input and kafka output processors

.travis.yml

@@ -13,6 +13,9 @@ install:
     - go get -v golang.org/x/tools/cmd/cover
     - go get -v github.com/stretchr/testify/assert
     - go get -v github.com/smartystreets/goconvey
+    - go get -v github.com/segmentio/kafka-go
+    - go get -v github.com/golang/snappy
+    - go get -v github.com/pierrec/lz4
 script:
     - ./go.test.sh
 after_success:

cmd/bitfan/commands/plugins.go

@@ -29,6 +29,7 @@ import (
 	inputstdout "github.com/vjeantet/bitfan/processors/input-stdout"
 	sysloginput "github.com/vjeantet/bitfan/processors/input-syslog"
 	tail "github.com/vjeantet/bitfan/processors/input-tail"
+	tcpinput "github.com/vjeantet/bitfan/processors/input-tcp"
 	twitter "github.com/vjeantet/bitfan/processors/input-twitter"
 	udpinput "github.com/vjeantet/bitfan/processors/input-udp"
 	unixinput "github.com/vjeantet/bitfan/processors/input-unix"
@@ -39,6 +40,7 @@ import (
 	fileoutput "github.com/vjeantet/bitfan/processors/output-file"
 	glusterfsoutput "github.com/vjeantet/bitfan/processors/output-glusterfs"
 	httpoutput "github.com/vjeantet/bitfan/processors/output-http"
+	kafkaoutput "github.com/vjeantet/bitfan/processors/output-kafka"
 	tcpoutput "github.com/vjeantet/bitfan/processors/output-tcp"
 	mongodb "github.com/vjeantet/bitfan/processors/output-mongodb"
 	null "github.com/vjeantet/bitfan/processors/output-null"
@@ -73,6 +75,7 @@ func init() {
 	initPlugin("input", "beats", beatsinput.New)
 	initPlugin("input", "rabbitmq", rabbitmqinput.New)
 	initPlugin("input", "udp", udpinput.New)
+	initPlugin("input", "tcp", tcpinput.New)
 	initPlugin("input", "syslog", sysloginput.New)
 	initPlugin("input", "unix", unixinput.New)
 	initPlugin("input", "readfile", file.New)
@@ -128,6 +131,7 @@ func init() {
 	initPlugin("output", "rabbitmq", rabbitmqoutput.New)
 	initPlugin("output", "email", email.New)
 	initPlugin("output", "http", httpoutput.New)
+	initPlugin("output", "kafka", kafkaoutput.New)
 	initPlugin("output", "tcp", tcpoutput.New)
 	initPlugin("output", "sql", sqlprocessor.New)
 	initPlugin("output", "template", templateprocessor.New)

docs/data/processors/kafkaoutput.json

@@ -0,0 +1,133 @@
+{
+  "Behavior": "",
+  "Doc": "",
+  "DocShort": "",
+  "ImportPath": "github.com/vjeantet/bitfan/processors/output-kafka",
+  "Name": "kafkaoutput",
+  "Options": {
+    "Doc": "",
+    "Options": [
+      {
+        "Alias": ",squash",
+        "DefaultValue": null,
+        "Doc": "",
+        "ExampleLS": "",
+        "Name": "processors.CommonOptions",
+        "PossibleValues": null,
+        "Required": false,
+        "Type": "processors.CommonOptions"
+      },
+      {
+        "Alias": "bootstrap_servers",
+        "DefaultValue": null,
+        "Doc": "Bootstrap Servers ( \"host:port\" )",
+        "ExampleLS": "",
+        "Name": "BootstrapServers",
+        "PossibleValues": null,
+        "Required": false,
+        "Type": "string"
+      },
+      {
+        "Alias": "brokers",
+        "DefaultValue": null,
+        "Doc": "Broker list",
+        "ExampleLS": "",
+        "Name": "Brokers",
+        "PossibleValues": null,
+        "Required": false,
+        "Type": "array"
+      },
+      {
+        "Alias": "topic_id",
+        "DefaultValue": null,
+        "Doc": "Kafka topic",
+        "ExampleLS": "",
+        "Name": "TopicID",
+        "PossibleValues": null,
+        "Required": true,
+        "Type": "string"
+      },
+      {
+        "Alias": "client_id",
+        "DefaultValue": null,
+        "Doc": "Kafka client id",
+        "ExampleLS": "",
+        "Name": "ClientID",
+        "PossibleValues": null,
+        "Required": false,
+        "Type": "string"
+      },
+      {
+        "Alias": "balancer",
+        "DefaultValue": null,
+        "Doc": "Balancer ( roundrobin, hash or leastbytes )",
+        "ExampleLS": "",
+        "Name": "Balancer",
+        "PossibleValues": null,
+        "Required": false,
+        "Type": "string"
+      },
+      {
+        "Alias": "max_attempts",
+        "DefaultValue": null,
+        "Doc": "Max Attempts",
+        "ExampleLS": "",
+        "Name": "MaxAttempts",
+        "PossibleValues": null,
+        "Required": false,
+        "Type": "int"
+      },
+      {
+        "Alias": "queue_size",
+        "DefaultValue": null,
+        "Doc": "Queue Size",
+        "ExampleLS": "",
+        "Name": "QueueSize",
+        "PossibleValues": null,
+        "Required": false,
+        "Type": "int"
+      },
+      {
+        "Alias": "batch_size",
+        "DefaultValue": null,
+        "Doc": "Batch Size",
+        "ExampleLS": "",
+        "Name": "BatchSize",
+        "PossibleValues": null,
+        "Required": false,
+        "Type": "int"
+      },
+      {
+        "Alias": "keepalive",
+        "DefaultValue": null,
+        "Doc": "Keep Alive ( in seconds )",
+        "ExampleLS": "",
+        "Name": "KeepAlive",
+        "PossibleValues": null,
+        "Required": false,
+        "Type": "int"
+      },
+      {
+        "Alias": "io_timeout",
+        "DefaultValue": null,
+        "Doc": "IO Timeout ( in seconds )",
+        "ExampleLS": "",
+        "Name": "IOTimeout",
+        "PossibleValues": null,
+        "Required": false,
+        "Type": "int"
+      },
+      {
+        "Alias": "acks",
+        "DefaultValue": null,
+        "Doc": "Required Acks ( number of replicas that must acknowledge write. -1 for all replicas )",
+        "ExampleLS": "",
+        "Name": "RequiredAcks",
+        "PossibleValues": null,
+        "Required": false,
+        "Type": "int"
+      }
+    ]
+  },
+  "Ports": []
+}

docs/data/processors/tcpinput.json

@@ -0,0 +1,43 @@
+{
+  "Behavior": "",
+  "Doc": "",
+  "DocShort": "",
+  "ImportPath": "github.com/vjeantet/bitfan/processors/input-tcp",
+  "Name": "tcpinput",
+  "Options": {
+    "Doc": "",
+    "Options": [
+      {
+        "Alias": ",squash",
+        "DefaultValue": null,
+        "Doc": "",
+        "ExampleLS": "",
+        "Name": "processors.CommonOptions",
+        "PossibleValues": null,
+        "Required": false,
+        "Type": "processors.CommonOptions"
+      },
+      {
+        "Alias": "port",
+        "DefaultValue": null,
+        "Doc": "TCP port number to listen on",
+        "ExampleLS": "",
+        "Name": "Port",
+        "PossibleValues": null,
+        "Required": false,
+        "Type": "int"
+      },
+      {
+        "Alias": "read_buffer_size",
+        "DefaultValue": null,
+        "Doc": "Message buffer size",
+        "ExampleLS": "",
+        "Name": "ReadBufferSize",
+        "PossibleValues": null,
+        "Required": false,
+        "Type": "int"
+      }
+    ]
+  },
+  "Ports": []
+}

processors/input-tcp/README.md

@@ -0,0 +1,36 @@
+# TCPINPUT
+
+
+## Synopsys
+
+
+|     SETTING      | TYPE | REQUIRED | DEFAULT VALUE |
+|------------------|------|----------|---------------|
+| port             | int  | false    |             0 |
+| read_buffer_size | int  | false    |             0 |
+
+
+## Details
+
+### port
+* Value type is int
+* Default value is `0`
+
+TCP port number to listen on
+
+### read_buffer_size
+* Value type is int
+* Default value is `0`
+
+Message buffer size
+
+
+
+## Configuration blueprint
+
+```
+tcp {
+	port => 123
+	read_buffer_size => 123
+}
+```

processors/input-tcp/tcp_test.go

@@ -0,0 +1,24 @@
+package tcpinput
+
+import (
+	"github.com/stretchr/testify/assert"
+	"github.com/vjeantet/bitfan/processors/doc"
+	"github.com/vjeantet/bitfan/processors/testutils"
+	"testing"
+)
+
+func TestNew(t *testing.T) {
+	p := New()
+	_, ok := p.(*processor)
+	assert.Equal(t, ok, true, "New() should return a processor")
+}
+func TestDoc(t *testing.T) {
+	assert.IsType(t, &doc.Processor{}, New().(*processor).Doc())
+}
+func TestConfigure(t *testing.T) {
+	conf := map[string]interface{}{}
+	ctx := testutils.NewProcessorContext()
+	p := New()
+	err := p.Configure(ctx, conf)
+	assert.Nil(t, err, "Configure() processor without error")
+}