Kafka and nats executer

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

@@ -0,0 +1,96 @@
+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": {
+//     "url": "http://example.com", // kafka server 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 do http request
+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["url"] == "" {
+
+		return output.Bytes(), errors.New("url 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{"192.168.59.103:9092"}
+	brokers := []string{args.Config["url"]}
+	producer, err := sarama.NewSyncProducer(brokers, config)
+	if err != nil {
+		// Should not reach here
+
+		if debug {
+			log.Printf("request  %#v\n\n", config)
+		}
+		return output.Bytes(), err
+	}
+
+	topic := args.Config["topic"]
+	msg := &sarama.ProducerMessage{
+		Topic: topic,
+		Value: sarama.StringEncoder(args.Config["message"]),
+	}
+
+	_, _, err = producer.SendMessage(msg)
+
+	if err != nil {
+		return output.Bytes(), err
+	}
+	defer func() {
+		producer.Close()
+	}()
+
+	output.Write([]byte("Result: success to publish data\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 TestPublishExecute(t *testing.T) {
+	pa := &dktypes.ExecuteRequest{
+		JobName: "testJob",
+		Config: map[string]string{
+			"topic":   "test",
+			"url":     "tesr",
+			"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": "http://example.com", // nats server url
+//     "message": "",                  //
+//     "topic": "publishTopic",             //
+//     "userName":"[email protected]",
+//     "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["topic"] == "" {
+		return output.Bytes(), errors.New("topic 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 At Nats Connection")
+	}
+
+	nc.Publish(args.Config["topic"], []byte(args.Config["message"]))
+
+	output.Write([]byte("Result: success to publish data\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{
+			"topic":   "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=