Merge pull request #341 from ahadas/syslog

Forwarding to external syslog, based on current fluentd plugin

manifests/4.5/logforwardings.crd.yaml

@@ -29,6 +29,7 @@ spec:
                     enum:
                       - elasticsearch
                       - forward
+                      - syslog
                   name: 
                     description: The name of the output
                     type: string

pkg/apis/logging/v1alpha1/forwarding_types.go

@@ -74,6 +74,9 @@ const (
 
 	//OutputTypeForward configures the pipeline to send messages via Fluent's secure forward
 	OutputTypeForward OutputType = "forward"
+
+	//OutputTypeSyslog configures pipeline to send messages to an external syslog server through docebo/fluent-plugin-remote-syslog
+	OutputTypeSyslog OutputType = "syslog"
 )
 
 //LogForwardingReason The reason for the current state

pkg/generators/factory.go

@@ -13,13 +13,7 @@ type Generator struct {
 
 //New creates an instance of a template engine for a set of templates
 func New(name string, addFunctions *template.FuncMap, templates ...string) (*Generator, error) {
-	allFunctions := funcMap
-	if addFunctions != nil {
-		for name, f := range *addFunctions {
-			allFunctions[name] = f
-		}
-	}
-	tmpl := template.New(name).Funcs(funcMap)
+	tmpl := template.New(name).Funcs(*addFunctions).Funcs(funcMap)
 	var err error
 	for i, s := range templates {
 		tmpl, err = tmpl.Parse(s)

pkg/generators/forwarding/fluentd/fluent_conf.go

@@ -11,6 +11,7 @@ import (
 )
 
 var replacer = strings.NewReplacer(" ", "_", "-", "_", ".", "_")
+var protocolSeparator = "://"
 
 type outputLabelConf struct {
 	Name            string
@@ -44,17 +45,34 @@ func (conf *outputLabelConf) Template() *template.Template {
 	return conf.TemplateContext
 }
 func (conf *outputLabelConf) Host() string {
-	return strings.Split(conf.Target.Endpoint, ":")[0]
+	endpoint := stripProtocol(conf.Target.Endpoint)
+	return strings.Split(endpoint, ":")[0]
 }
 
 func (conf *outputLabelConf) Port() string {
-	parts := strings.Split(conf.Target.Endpoint, ":")
+	endpoint := stripProtocol(conf.Target.Endpoint)
+	parts := strings.Split(endpoint, ":")
 	if len(parts) == 1 {
 		return "9200"
 	}
 	return parts[1]
 }
 
+func (conf *outputLabelConf) Protocol() string {
+	endpoint := conf.Target.Endpoint
+	if index := strings.Index(endpoint, protocolSeparator); index != -1 {
+		return endpoint[:index]
+	}
+	return ""
+}
+
+func stripProtocol(endpoint string) string {
+	if index := strings.Index(endpoint, protocolSeparator); index != -1 {
+		endpoint = endpoint[index+len(protocolSeparator):]
+	}
+	return endpoint
+}
+
 func (conf *outputLabelConf) BufferPath() string {
 	return fmt.Sprintf("/var/lib/fluentd/%s", conf.StoreID())
 }

pkg/generators/forwarding/fluentd/generators.go

@@ -202,6 +202,9 @@ func (engine *ConfigGenerator) generateOutputLabelBlocks(outputs []logforward.Ou
 		case logforward.OutputTypeForward:
 			storeTemplateName = "forward"
 			outputTemplateName = "outputLabelConfNoCopy"
+		case logforward.OutputTypeSyslog:
+			storeTemplateName = "storeSyslog"
+			outputTemplateName = "outputLabelConfNoRetry"
 		default:
 			logger.Warnf("Pipeline targets include an unrecognized type: %q", output.Type)
 			continue

pkg/generators/forwarding/fluentd/output_conf_syslog_test.go

@@ -0,0 +1,108 @@
+package fluentd
+
+import (
+	. "github.com/onsi/ginkgo"
+	. "github.com/onsi/gomega"
+
+	logging "github.com/openshift/cluster-logging-operator/pkg/apis/logging/v1alpha1"
+	test "github.com/openshift/cluster-logging-operator/test"
+)
+
+var _ = Describe("Generating external syslog server output store config blocks", func() {
+
+	var (
+		err       error
+		outputs   []logging.OutputSpec
+		generator *ConfigGenerator
+	)
+	BeforeEach(func() {
+		generator, err = NewConfigGenerator(false, false)
+		Expect(err).To(BeNil())
+	})
+
+	Context("based on syslog plugin", func() {
+		tcpConf := `<label @SYSLOG_RECEIVER>
+		<match **>
+		@type copy
+		<store>
+			@type syslog_buffered
+			@id syslog_receiver
+			remote_syslog sl.svc.messaging.cluster.local
+			port 9654
+			hostname ${hostname}
+			facility user
+			severity debug
+		</store>
+		</match>
+	</label>`
+
+		udpConf := `<label @SYSLOG_RECEIVER>
+		<match **>
+		@type copy
+		<store>
+			@type syslog
+			@id syslog_receiver
+			remote_syslog sl.svc.messaging.cluster.local
+			port 9654
+			hostname ${hostname}
+			facility user
+			severity debug
+		</store>
+		</match>
+	</label>`
+
+		Context("for protocol-less endpoint", func() {
+			BeforeEach(func() {
+				outputs = []logging.OutputSpec{
+					{
+						Type:     logging.OutputTypeSyslog,
+						Name:     "syslog-receiver",
+						Endpoint: "sl.svc.messaging.cluster.local:9654",
+					},
+				}
+			})
+			It("should produce well formed output label config", func() {
+				results, err := generator.generateOutputLabelBlocks(outputs)
+				Expect(err).To(BeNil())
+				Expect(len(results)).To(Equal(1))
+				test.Expect(results[0]).ToEqual(tcpConf)
+			})
+		})
+
+		Context("for tcp endpoint", func() {
+			BeforeEach(func() {
+				outputs = []logging.OutputSpec{
+					{
+						Type:     logging.OutputTypeSyslog,
+						Name:     "syslog-receiver",
+						Endpoint: "tcp://sl.svc.messaging.cluster.local:9654",
+					},
+				}
+			})
+			It("should produce well formed output label config", func() {
+				results, err := generator.generateOutputLabelBlocks(outputs)
+				Expect(err).To(BeNil())
+				Expect(len(results)).To(Equal(1))
+				test.Expect(results[0]).ToEqual(tcpConf)
+			})
+		})
+
+		Context("for udp endpoint", func() {
+			BeforeEach(func() {
+				outputs = []logging.OutputSpec{
+					{
+						Type:     logging.OutputTypeSyslog,
+						Name:     "syslog-receiver",
+						Endpoint: "udp://sl.svc.messaging.cluster.local:9654",
+					},
+				}
+			})
+			It("should produce well formed output label config", func() {
+				results, err := generator.generateOutputLabelBlocks(outputs)
+				Expect(err).To(BeNil())
+				Expect(len(results)).To(Equal(1))
+				test.Expect(results[0]).ToEqual(udpConf)
+			})
+		})
+	})
+})

pkg/generators/forwarding/fluentd/syslog_conf.go

@@ -0,0 +1,8 @@
+package fluentd
+
+func (conf *outputLabelConf) SyslogPlugin() string {
+	if protocol := conf.Protocol(); protocol == "udp" {
+		return "syslog"
+	}
+	return "syslog_buffered"
+}

pkg/generators/forwarding/fluentd/templates.go

@@ -12,8 +12,10 @@ var templateRegistry = []string{
 	sourceToPipelineCopyTemplate,
 	outputLabelConfTemplate,
 	outputLabelConfNocopyTemplate,
+	outputLabelConfNoretryTemplate,
 	storeElasticsearchTemplate,
 	forwardTemplate,
+	storeSyslogTemplate,
 }
 
 const fluentConfTemplate = `{{- define "fluentConf" }}
@@ -484,6 +486,15 @@ const outputLabelConfNocopyTemplate = `{{- define "outputLabelConfNoCopy" }}
 </label>
 {{- end}}`
 
+const outputLabelConfNoretryTemplate = `{{- define "outputLabelConfNoRetry" }}
+<label {{.LabelName}}>
+	<match **>
+		@type copy
+{{include .StoreTemplate . "" | indent 4}}
+	</match>
+</label>
+{{- end}}`
+
 const forwardTemplate = `{{- define "forward" }}
 	# https://docs.fluentd.org/v1.0/articles/in_forward
 	@type forward
@@ -573,3 +584,15 @@ const storeElasticsearchTemplate = `{{- define "storeElasticsearch" }}
 	</buffer>
 </store>
 {{- end}}`
+
+const storeSyslogTemplate = `{{- define "storeSyslog" }}
+<store>
+	@type {{.SyslogPlugin}}
+	@id {{.StoreID}}
+	remote_syslog {{.Host}}
+	port {{.Port}}
+	hostname ${hostname}
+	facility user
+	severity debug
+</store>
+{{- end}}`

pkg/k8shandler/forwarding.go

@@ -26,7 +26,7 @@ const (
 )
 
 var (
-	outputTypes = sets.NewString(string(logforward.OutputTypeElasticsearch), string(logforward.OutputTypeForward))
+	outputTypes = sets.NewString(string(logforward.OutputTypeElasticsearch), string(logforward.OutputTypeForward), string(logforward.OutputTypeSyslog))
 	sourceTypes = sets.NewString(string(logforward.LogSourceTypeApp), string(logforward.LogSourceTypeInfra), string(logforward.LogSourceTypeAudit))
 )