Basic process that sends job output to fluent destination

builtin/bins/dkron-processor-fluent/fluent_output.go

@@ -0,0 +1,47 @@
+package main
+
+import (
+	"strconv"
+
+	"github.com/distribworks/dkron/v2/plugin"
+	"github.com/distribworks/dkron/v2/plugin/types"
+	log "github.com/sirupsen/logrus"
+)
+
+// Process sends log to Fluent
+func (l *FluentOutput) Process(args *plugin.ProcessorArgs) types.Execution {
+
+	l.parseConfig(args.Config)
+
+	var data = map[string]interface{}{
+		"host":     args.Execution.NodeName,
+		"job_name": args.Execution.JobName,
+		"message":  args.Execution.Output,
+	}
+
+	go l.sendLog(data)
+
+	if !l.forward {
+		args.Execution.Output = []byte("Output sent to Fluent")
+	}
+
+	return args.Execution
+}
+
+func (l *FluentOutput) parseConfig(config plugin.Config) {
+	forward, err := strconv.ParseBool(config["forward"])
+	if err != nil {
+		l.forward = false
+		log.WithField("param", "forward").Warning("Incorrect format or param not found.")
+	} else {
+		l.forward = forward
+		log.Infof("Forwarding set to: %t", forward)
+	}
+}
+
+func (l *FluentOutput) sendLog(data map[string]interface{}) {
+	err := l.fluent.Post(l.tag, data)
+	if err != nil {
+		log.WithError(err).Error("Error sending to Fluent")
+	}
+}

builtin/bins/dkron-processor-fluent/main.go

@@ -0,0 +1,59 @@
+package main
+
+import (
+	"os"
+	"strconv"
+
+	"github.com/distribworks/dkron/v2/plugin"
+	"github.com/fluent/fluent-logger-golang/fluent"
+	log "github.com/sirupsen/logrus"
+)
+
+// FluentOutput is good
+type FluentOutput struct {
+	forward bool
+	tag     string
+	fluent  *fluent.Fluent
+}
+
+func main() {
+
+	fh := getEnv("FLUENTBIT_HOST")
+	fp, _ := strconv.Atoi(getEnv("FLUENTBIT_PORT"))
+	ft := getEnv("FLUENTBIT_TAG")
+
+	logger, _ := fluent.New(fluent.Config{
+		FluentHost: fh,
+		FluentPort: fp,
+		RequestAck: false,
+	})
+
+	fo := FluentOutput{fluent: logger, tag: ft}
+
+	plugin.Serve(&plugin.ServeOpts{
+		Processor: &fo,
+	})
+}
+
+func getEnv(key string) string {
+	v, ok := os.LookupEnv(key)
+
+	if key == "FLUENT_PORT" {
+		r, e := strconv.Atoi(v)
+		if e != nil {
+			return strconv.Itoa(r)
+		}
+		log.Warningf("non integer value '%s' for FLUENT_PORT variable. Using port 24224 as replacement", v)
+		return "24224"
+	}
+
+	if v == "" {
+		log.Warningf("empty value for environment variable %s", key)
+		return "set_my_env_var"
+	}
+	if !ok {
+		log.Warningf("environment variable %s is not set", key)
+		return "var_is_empty"
+	}
+	return v
+}