Kafka and nats executer

.goreleaser.yml

@@ -34,6 +34,16 @@ builds:
     id: dkron-executor-rabbitmq
     binary: dkron-executor-rabbitmq
 
+  - <<: *xbuild
+    main: ./builtin/bins/dkron-executor-nats/
+    id: dkron-executor-nats
+    binary: dkron-executor-nats
+
+  - <<: *xbuild
+    main: ./builtin/bins/dkron-executor-kafka/
+    id: dkron-executor-kafka
+    binary: dkron-executor-kafka
+
   - <<: *xbuild
     main: ./builtin/bins/dkron-processor-files/
     id: dkron-processor-files

builtin/bins/dkron-executor-kafka/kafka.go

@@ -0,0 +1,92 @@
+package main
+
+import (
+	"errors"
+	"log"
+
+	"github.com/Shopify/sarama"
+	"github.com/armon/circbuf"
+
+	dkplugin "github.com/distribworks/dkron/v3/plugin"
+	dktypes "github.com/distribworks/dkron/v3/plugin/types"
+)
+
+const (
+	// maxBufSize limits how much data we collect from a handler.
+	// This is to prevent Serf's memory from growing to an enormous
+	// amount due to a faulty handler.
+	maxBufSize = 500000
+)
+
+// Kafka process kafka request
+type Kafka struct {
+}
+
+// Execute Process method of the plugin
+// "executor": "kafka",
+// "executor_config": {
+//     "brokerAddress": "192.168.59.103:9092", // kafka broker url
+//     "message": "",
+//     "topic": "publishTopic"
+// }
+func (s *Kafka) Execute(args *dktypes.ExecuteRequest, cb dkplugin.StatusHelper) (*dktypes.ExecuteResponse, error) {
+
+	out, err := s.ExecuteImpl(args)
+	resp := &dktypes.ExecuteResponse{Output: out}
+	if err != nil {
+		resp.Error = err.Error()
+	}
+	return resp, nil
+}
+
+// ExecuteImpl produce message on Kafka broker
+func (s *Kafka) ExecuteImpl(args *dktypes.ExecuteRequest) ([]byte, error) {
+
+	output, _ := circbuf.NewBuffer(maxBufSize)
+
+	var debug bool
+	if args.Config["debug"] != "" {
+		debug = true
+		log.Printf("config  %#v\n\n", args.Config)
+	}
+
+	if args.Config["brokerAddress"] == "" {
+
+		return output.Bytes(), errors.New("brokerAddress is empty")
+	}
+
+	if args.Config["topic"] == "" {
+		return output.Bytes(), errors.New("topic is empty")
+	}
+	config := sarama.NewConfig()
+	config.Producer.RequiredAcks = sarama.WaitForAll
+	config.Producer.Retry.Max = 5
+	config.Producer.Return.Successes = true
+	config.Producer.Return.Errors = true
+
+	brokers := []string{args.Config["brokerAddress"]}
+	producer, err := sarama.NewSyncProducer(brokers, config)
+	if err != nil {
+		// Should not reach here
+
+		if debug {
+			log.Printf("sarama  %#v\n\n", config)
+		}
+		return output.Bytes(), err
+	}
+	defer producer.Close()
+
+	msg := &sarama.ProducerMessage{
+		Topic: args.Config["topic"],
+		Value: sarama.StringEncoder(args.Config["message"]),
+	}
+
+	_, _, err = producer.SendMessage(msg)
+
+	if err != nil {
+		return output.Bytes(), err
+	}
+
+	output.Write([]byte("Result: successfully produced the message on Kafka broker\n"))
+	return output.Bytes(), nil
+}

builtin/bins/dkron-executor-kafka/kafka_test.go

@@ -0,0 +1,27 @@
+package main
+
+import (
+	"fmt"
+	"testing"
+
+	dktypes "github.com/distribworks/dkron/v3/plugin/types"
+)
+
+func TestProduceExecute(t *testing.T) {
+	pa := &dktypes.ExecuteRequest{
+		JobName: "testJob",
+		Config: map[string]string{
+			"topic":         "test",
+			"brokerAddress": "testaddress",
+			"message":       "{\"hello\":11}",
+			"debug":         "true",
+		},
+	}
+	kafka := &Kafka{}
+	output, err := kafka.Execute(pa, nil)
+	fmt.Println(string(output.Output))
+	fmt.Println(err)
+	if err != nil {
+		t.Fatal(err)
+	}
+}

builtin/bins/dkron-executor-kafka/main.go

@@ -0,0 +1,18 @@
+package main
+
+import (
+	dkplugin "github.com/distribworks/dkron/v3/plugin"
+	"github.com/hashicorp/go-plugin"
+)
+
+func main() {
+	plugin.Serve(&plugin.ServeConfig{
+		HandshakeConfig: dkplugin.Handshake,
+		Plugins: map[string]plugin.Plugin{
+			"executor": &dkplugin.ExecutorPlugin{Executor: &Kafka{}},
+		},
+
+		// A non-nil value here enables gRPC serving for this plugin...
+		GRPCServer: plugin.DefaultGRPCServer,
+	})
+}

builtin/bins/dkron-executor-nats/main.go

@@ -0,0 +1,18 @@
+package main
+
+import (
+	dkplugin "github.com/distribworks/dkron/v3/plugin"
+	"github.com/hashicorp/go-plugin"
+)
+
+func main() {
+	plugin.Serve(&plugin.ServeConfig{
+		HandshakeConfig: dkplugin.Handshake,
+		Plugins: map[string]plugin.Plugin{
+			"executor": &dkplugin.ExecutorPlugin{Executor: &Nats{}},
+		},
+
+		// A non-nil value here enables gRPC serving for this plugin...
+		GRPCServer: plugin.DefaultGRPCServer,
+	})
+}

builtin/bins/dkron-executor-nats/nats.go

@@ -0,0 +1,78 @@
+package main
+
+import (
+	"errors"
+	"log"
+
+	"github.com/armon/circbuf"
+	"github.com/nats-io/nats.go"
+
+	dkplugin "github.com/distribworks/dkron/v3/plugin"
+	dktypes "github.com/distribworks/dkron/v3/plugin/types"
+)
+
+const (
+	// maxBufSize limits how much data we collect from a handler.
+	// This is to prevent Serf's memory from growing to an enormous
+	// amount due to a faulty handler.
+	maxBufSize = 500000
+)
+
+// Nats process http request
+type Nats struct {
+}
+
+// Execute Process method of the plugin
+// "executor": "nats",
+// "executor_config": {
+//     "url": "tls://nats.demo.io:4443", // nats server url
+//     "message": "",
+//     "subject": "Subject",
+//     "userName":"test@hbh.dfg",
+//     "password":"dfdffs"
+// }
+func (s *Nats) Execute(args *dktypes.ExecuteRequest, cb dkplugin.StatusHelper) (*dktypes.ExecuteResponse, error) {
+
+	out, err := s.ExecuteImpl(args)
+	resp := &dktypes.ExecuteResponse{Output: out}
+	if err != nil {
+		resp.Error = err.Error()
+	}
+	return resp, nil
+}
+
+// ExecuteImpl do http request
+func (s *Nats) ExecuteImpl(args *dktypes.ExecuteRequest) ([]byte, error) {
+
+	output, _ := circbuf.NewBuffer(maxBufSize)
+
+	var debug bool
+	if args.Config["debug"] != "" {
+		debug = true
+		log.Printf("config  %#v\n\n", args.Config)
+	}
+
+	if args.Config["url"] == "" {
+
+		return output.Bytes(), errors.New("url is empty")
+	}
+
+	if args.Config["subject"] == "" {
+		return output.Bytes(), errors.New("subject is empty")
+	}
+	nc, err := nats.Connect(args.Config["url"], nats.UserInfo(args.Config["userName"], args.Config["password"]))
+
+	if err != nil {
+		return output.Bytes(), errors.New("error connecting to NATS")
+	}
+
+	nc.Publish(args.Config["subject"], []byte(args.Config["message"]))
+
+	output.Write([]byte("Result: Message successfully sent\n"))
+
+	if debug {
+		log.Printf("request  %#v\n\n", nc)
+	}
+
+	return output.Bytes(), nil
+}

builtin/bins/dkron-executor-nats/nats_test.go

@@ -0,0 +1,27 @@
+package main
+
+import (
+	"fmt"
+	"testing"
+
+	dktypes "github.com/distribworks/dkron/v3/plugin/types"
+)
+
+func TestPublishExecute(t *testing.T) {
+	pa := &dktypes.ExecuteRequest{
+		JobName: "testJob",
+		Config: map[string]string{
+			"subject": "opcuaReadRequest",
+			"url":     "localhost:4222",
+			"message": "{\"hello\":11}",
+			"debug":   "true",
+		},
+	}
+	nats := &Nats{}
+	output, err := nats.Execute(pa, nil)
+	fmt.Println(string(output.Output))
+	fmt.Println(err)
+	if err != nil {
+		t.Fatal(err)
+	}
+}

go.mod

@@ -2,6 +2,7 @@ module github.com/distribworks/dkron/v3
 
 require (
 	github.com/DataDog/datadog-go v4.0.0+incompatible // indirect
+	github.com/Shopify/sarama v1.19.0
 	github.com/armon/circbuf v0.0.0-20190214190532-5111143e8da2
 	github.com/armon/go-metrics v0.3.6
 	github.com/aws/aws-sdk-go v1.36.31 // indirect
@@ -31,6 +32,7 @@ require (
 	github.com/kardianos/osext v0.0.0-20190222173326-2bc1f35cddc0
 	github.com/mattn/go-shellwords v1.0.11
 	github.com/mitchellh/go-testing-interface v1.14.1 // indirect
+	github.com/nats-io/nats.go v1.9.1
 	github.com/philhofer/fwd v1.0.0 // indirect
 	github.com/prometheus/client_golang v1.9.0
 	github.com/robfig/cron/v3 v3.0.1

go.sum

@@ -176,8 +176,11 @@ github.com/dnaeon/go-vcr v1.0.1 h1:r8L/HqC0Hje5AXMu1ooW8oyQyOFv4GxqpL0nRP7SLLY=
 github.com/dnaeon/go-vcr v1.0.1/go.mod h1:aBB1+wY4s93YsC3HHjMBMrwTj2R9FHDzUr9KyGc8n1E=
 github.com/docker/spdystream v0.0.0-20160310174837-449fdfce4d96/go.mod h1:Qh8CwZgvJUkLughtfhJv5dyTYa91l1fOUCrgjqmcifM=
 github.com/dustin/go-humanize v0.0.0-20171111073723-bb3d318650d4/go.mod h1:HtrtbFcZ19U5GC7JDqmcUSB87Iq5E25KnS6fMYU6eOk=
+github.com/eapache/go-resiliency v1.1.0 h1:1NtRmCAqadE2FN4ZcN6g90TP3uk8cg9rn9eNK2197aU=
 github.com/eapache/go-resiliency v1.1.0/go.mod h1:kFI+JgMyC7bLPUVY133qvEBtVayf5mFgVsvEsIPBvNs=
+github.com/eapache/go-xerial-snappy v0.0.0-20180814174437-776d5712da21 h1:YEetp8/yCZMuEPMUDHG0CW/brkkEp8mzqk2+ODEitlw=
 github.com/eapache/go-xerial-snappy v0.0.0-20180814174437-776d5712da21/go.mod h1:+020luEh2TKB4/GOp8oxxtq0Daoen/Cii55CzbTV6DU=
+github.com/eapache/queue v1.1.0 h1:YOEu7KNc61ntiQlcEeUIoDTJ2o8mQznoNvUhiigpIqc=
 github.com/eapache/queue v1.1.0/go.mod h1:6eCeP0CKFpHLu8blIFXhExK/dRa7WDZfr6jVFPTqq+I=
 github.com/edsrzf/mmap-go v1.0.0/go.mod h1:YO35OhQPt3KJa3ryjFM5Bs14WD66h8eGKpfaBNrHW5M=
 github.com/elazarl/goproxy v0.0.0-20180725130230-947c36da3153/go.mod h1:/Zj4wYkgs4iZTTu3o/KG3Itv/qCCa8VVMlb3i9OVuzc=
@@ -288,6 +291,7 @@ github.com/golang/protobuf v1.4.2 h1:+Z5KGCizgyZCbGh1KZqA0fcLLkwbsjIzS4aV2v7wJX0
 github.com/golang/protobuf v1.4.2/go.mod h1:oDoupMAO8OvCJWAcko0GGGIgR6R6ocIYbsSw735rRwI=
 github.com/golang/protobuf v1.4.3 h1:JjCZWpVbqXDqFVmTfYWEVTMIYrL/NPdPSCHPJ0T/raM=
 github.com/golang/protobuf v1.4.3/go.mod h1:oDoupMAO8OvCJWAcko0GGGIgR6R6ocIYbsSw735rRwI=
+github.com/golang/snappy v0.0.0-20180518054509-2e65f85255db h1:woRePGFeVFfLKN/pOkfl+p/TAqKOfFu+7KPlMVpok/w=
 github.com/golang/snappy v0.0.0-20180518054509-2e65f85255db/go.mod h1:/XxbfmMg8lxefKM7IXC3fBNl/7bRcc72aCRzEWrmP2Q=
 github.com/google/btree v0.0.0-20180813153112-4030bb1f1f0c/go.mod h1:lNA+9X1NB3Zf8V7Ke586lFgjr2dZNuvo3lPJSGZ5JPQ=
 github.com/google/btree v1.0.0 h1:0udJVsspx3VBr5FwtLhQQtuAsVc79tTq0ocGIPAU6qo=
@@ -560,11 +564,15 @@ github.com/mwitkow/go-conntrack v0.0.0-20161129095857-cc309e4a2223/go.mod h1:qRW
 github.com/mwitkow/go-conntrack v0.0.0-20190716064945-2f068394615f/go.mod h1:qRWi+5nqEBWmkhHvq77mSJWrCKwh8bxhgT7d/eI7P4U=
 github.com/mxk/go-flowrate v0.0.0-20140419014527-cca7078d478f/go.mod h1:ZdcZmHo+o7JKHSa8/e818NopupXU1YMK5fe1lsApnBw=
 github.com/nats-io/jwt v0.3.0/go.mod h1:fRYCDE99xlTsqUzISS1Bi75UBJ6ljOJQOAAu5VglpSg=
+github.com/nats-io/jwt v0.3.2 h1:+RB5hMpXUUA2dfxuhBTEkMOrYmM+gKIZYS1KjSostMI=
 github.com/nats-io/jwt v0.3.2/go.mod h1:/euKqTS1ZD+zzjYrY7pseZrTtWQSjujC7xjPc8wL6eU=
 github.com/nats-io/nats-server/v2 v2.1.2/go.mod h1:Afk+wRZqkMQs/p45uXdrVLuab3gwv3Z8C4HTBu8GD/k=
+github.com/nats-io/nats.go v1.9.1 h1:ik3HbLhZ0YABLto7iX80pZLPw/6dx3T+++MZJwLnMrQ=
 github.com/nats-io/nats.go v1.9.1/go.mod h1:ZjDU1L/7fJ09jvUSRVBR2e7+RnLiiIQyqyzEE/Zbp4w=
 github.com/nats-io/nkeys v0.1.0/go.mod h1:xpnFELMwJABBLVhffcfd1MZx6VsNRFpEugbxziKVo7w=
+github.com/nats-io/nkeys v0.1.3 h1:6JrEfig+HzTH85yxzhSVbjHRJv9cn0p6n3IngIcM5/k=
 github.com/nats-io/nkeys v0.1.3/go.mod h1:xpnFELMwJABBLVhffcfd1MZx6VsNRFpEugbxziKVo7w=
+github.com/nats-io/nuid v1.0.1 h1:5iA8DT8V7q8WK2EScv2padNa/rTESc1KdnPw4TC2paw=
 github.com/nats-io/nuid v1.0.1/go.mod h1:19wcPz3Ph3q0Jbyiqsd0kePYG7A95tJPxeL+1OSON2c=
 github.com/nicolai86/scaleway-sdk v1.10.2-0.20180628010248-798f60e20bb2 h1:BQ1HW7hr4IVovMwWg0E0PYcyW8CzqDcVmaew9cujU4s=
 github.com/nicolai86/scaleway-sdk v1.10.2-0.20180628010248-798f60e20bb2/go.mod h1:TLb2Sg7HQcgGdloNxkrmtgDNR9uVYF3lfdFIN4Ro6Sk=
@@ -605,6 +613,7 @@ github.com/peterbourgon/diskv v2.0.1+incompatible/go.mod h1:uqqh8zWWbv1HBMNONnaR
 github.com/philhofer/fwd v1.0.0 h1:UbZqGr5Y38ApvM/V/jEljVxwocdweyH+vmYvRPBnbqQ=
 github.com/philhofer/fwd v1.0.0/go.mod h1:gk3iGcWd9+svBvR0sR+KPcfE+RNWozjowpeBVG3ZVNU=
 github.com/pierrec/lz4 v1.0.2-0.20190131084431-473cd7ce01a1/go.mod h1:3/3N9NVKO0jef7pBehbT1qWhCMrIgbYNnFAZCqQ5LRc=
+github.com/pierrec/lz4 v2.0.5+incompatible h1:2xWsjqPFWcplujydGg4WmhC/6fZqK42wMM8aXeqhl0I=
 github.com/pierrec/lz4 v2.0.5+incompatible/go.mod h1:pdkljMzZIN41W+lC3N2tnIh5sFi+IEE17M5jbnwPHcY=
 github.com/pkg/errors v0.8.0/go.mod h1:bwawxfHBFNV+L2hUp1rHADufV3IMtnDRdf1r5NINEl0=
 github.com/pkg/errors v0.8.1/go.mod h1:bwawxfHBFNV+L2hUp1rHADufV3IMtnDRdf1r5NINEl0=
@@ -664,6 +673,7 @@ github.com/prometheus/procfs v0.1.3/go.mod h1:lV6e/gmhEcM9IjHGsFOCxxuZ+z1YqCvr4O
 github.com/prometheus/procfs v0.2.0 h1:wH4vA7pcjKuZzjF7lM8awk4fnuJO6idemZXoKnULUx4=
 github.com/prometheus/procfs v0.2.0/go.mod h1:lV6e/gmhEcM9IjHGsFOCxxuZ+z1YqCvr4OA4YeYWdaU=
 github.com/prometheus/tsdb v0.7.1/go.mod h1:qhTCs0VvXwvX/y3TZrWD7rabWM+ijKTux40TwIPHuXU=
+github.com/rcrowley/go-metrics v0.0.0-20181016184325-3113b8401b8a h1:9ZKAASQSHhDYGoxY8uLVpewe1GDZ2vu2Tr/vTdVAkFQ=
 github.com/rcrowley/go-metrics v0.0.0-20181016184325-3113b8401b8a/go.mod h1:bCqnVzQkZxMG4s8nGwiZ5l3QUCyqpo9Y+/ZMZ9VjZe4=
 github.com/renier/xmlrpc v0.0.0-20170708154548-ce4a1a486c03 h1:Wdi9nwnhFNAlseAOekn6B5G/+GMtks9UKbvRU/CMM/o=
 github.com/renier/xmlrpc v0.0.0-20170708154548-ce4a1a486c03/go.mod h1:gRAiPF5C5Nd0eyyRdqIu9qTiFSoZzpTq727b5B8fkkU=

website/content/usage/executors/_index.md

@@ -7,9 +7,9 @@ weight: 30
 
 Executor plugins are the main mechanism of execution in Dkron. They implement different "types" of jobs in the sense that they can perform the most diverse actions on the target nodes.
 
-For example, the built-in `shell` executor, will run the indicated command in the target node.
+For example, the built-in `shell` executor, will run the indicated command on the target node.
 
-New plugins will be added, or you can create new ones, to perform different tasks, as HTTP requests, Docker runs, anything that you can imagine.
+New plugins will be added, or you can create new ones, to perform different tasks, such as HTTP requests, Docker runs, anything that you can imagine.
 
 {{% children  %}}
 

website/content/usage/executors/http.md

@@ -9,18 +9,18 @@ HTTP executor can send a request to an HTTP endpoint
 Params:
 
 ```
-method: Request method in uppercase
-url: Request url
-headers: Json string, such as "[\"Content-Type: application/json\"]"
-body: POST body
-timeout: Request timeout, unit seconds
+method:     Request method in uppercase
+url:        Request url
+headers:    Json string, such as "[\"Content-Type: application/json\"]"
+body:       POST body
+timeout:    Request timeout, unit seconds
 expectCode: Expect response code, such as 200,206
 expectBody: Expect response body, support regexp, such as /success/
-debug: Debug option, will log everything when this option is not empty
-tlsNoVerifyPeer: false (default) or true. If true, disables verification of the remote SSL certificate's validity.
-tlsCertificateFile: Path to the PEM file containing the client certificate. Optional.
+debug:      Debug option, will log everything when this option is not empty
+tlsNoVerifyPeer:       false (default) or true. If true, disables verification of the remote SSL certificate's validity.
+tlsCertificateFile:    Path to the PEM file containing the client certificate. Optional.
 tlsCertificateKeyFile: Path to the PEM file containing the client certificate private key. Optional.
-tlsRootCAsFile: Path to the PEM file containing certificates to use as root CAs. Optional.
+tlsRootCAsFile:        Path to the PEM file containing certificates to use as root CAs. Optional.
 ```
 
 Example