Sink to send metrics to Syslog

.github/workflows/Release.yml

@@ -44,16 +44,16 @@ jobs:
     # Use dnf to install build dependencies
     - name: Install build dependencies
       run: |
-          wget -q https://go.dev/dl/go1.22.4.linux-amd64.tar.gz --output-document=- | \
-          tar --directory=/usr/local --extract --gzip
-          export PATH=/usr/local/go/bin:/usr/local/go/pkg/tool/linux_amd64:$PATH
-          go version
+          dnf --assumeyes install \
+              http://mirror.centos.org/centos/8-stream/AppStream/x86_64/os/Packages/golang-1.21.7-1.module_el8+960+4060efbe.x86_64.rpm \
+              http://mirror.centos.org/centos/8-stream/AppStream/x86_64/os/Packages/golang-bin-1.21.7-1.module_el8+960+4060efbe.x86_64.rpm \
+              http://mirror.centos.org/centos/8-stream/AppStream/x86_64/os/Packages/golang-src-1.21.7-1.module_el8+960+4060efbe.noarch.rpm \
+              http://mirror.centos.org/centos/8-stream/AppStream/x86_64/os/Packages/go-toolset-1.21.7-1.module_el8+960+4060efbe.x86_64.rpm
 
     - name: RPM build MetricCollector
       id: rpmbuild
       run: |
           git config --global --add safe.directory /__w/cc-metric-collector/cc-metric-collector
-          export PATH=/usr/local/go/bin:/usr/local/go/pkg/tool/linux_amd64:$PATH
           make RPM
 
     # AlmaLinux 8 is a derivate of RedHat Enterprise Linux 8 (UBI8),
@@ -114,16 +114,16 @@ jobs:
     # Use dnf to install build dependencies
     - name: Install build dependencies
       run: |
-          wget -q https://go.dev/dl/go1.22.4.linux-amd64.tar.gz --output-document=- | \
-          tar --directory=/usr/local --extract --gzip
-          export PATH=/usr/local/go/bin:/usr/local/go/pkg/tool/linux_amd64:$PATH
-          go version
+          dnf --assumeyes --disableplugin=subscription-manager install \
+              http://mirror.centos.org/centos/8-stream/AppStream/x86_64/os/Packages/golang-1.21.7-1.module_el8+960+4060efbe.x86_64.rpm \
+              http://mirror.centos.org/centos/8-stream/AppStream/x86_64/os/Packages/golang-bin-1.21.7-1.module_el8+960+4060efbe.x86_64.rpm \
+              http://mirror.centos.org/centos/8-stream/AppStream/x86_64/os/Packages/golang-src-1.21.7-1.module_el8+960+4060efbe.noarch.rpm \
+              http://mirror.centos.org/centos/8-stream/AppStream/x86_64/os/Packages/go-toolset-1.21.7-1.module_el8+960+4060efbe.x86_64.rpm
 
     - name: RPM build MetricCollector
       id: rpmbuild
       run: |
           git config --global --add safe.directory /__w/cc-metric-collector/cc-metric-collector
-          export PATH=/usr/local/go/bin:/usr/local/go/pkg/tool/linux_amd64:$PATH
           make RPM
 
     # See: https://github.com/actions/upload-artifact
@@ -165,7 +165,7 @@ jobs:
     # Use official golang package
     - name: Install Golang
       run: |
-          wget -q https://go.dev/dl/go1.22.4.linux-amd64.tar.gz --output-document=- | \
+          wget -q https://go.dev/dl/go1.21.1.linux-amd64.tar.gz --output-document=- | \
           tar --directory=/usr/local --extract --gzip
           export PATH=/usr/local/go/bin:/usr/local/go/pkg/tool/linux_amd64:$PATH
           go version

.github/workflows/runonce.yml

@@ -91,16 +91,16 @@ jobs:
     # Use dnf to install build dependencies
     - name: Install build dependencies
       run: |
-          wget -q https://go.dev/dl/go1.22.4.linux-amd64.tar.gz --output-document=- | \
-          tar --directory=/usr/local --extract --gzip
-          export PATH=/usr/local/go/bin:/usr/local/go/pkg/tool/linux_amd64:$PATH
-          go version
+          dnf --assumeyes install \
+              http://mirror.centos.org/centos/8-stream/AppStream/x86_64/os/Packages/golang-1.21.7-1.module_el8+960+4060efbe.x86_64.rpm \
+              http://mirror.centos.org/centos/8-stream/AppStream/x86_64/os/Packages/golang-bin-1.21.7-1.module_el8+960+4060efbe.x86_64.rpm \
+              http://mirror.centos.org/centos/8-stream/AppStream/x86_64/os/Packages/golang-src-1.21.7-1.module_el8+960+4060efbe.noarch.rpm \
+              http://mirror.centos.org/centos/8-stream/AppStream/x86_64/os/Packages/go-toolset-1.21.7-1.module_el8+960+4060efbe.x86_64.rpm
 
     - name: RPM build MetricCollector
       id: rpmbuild
       run: |
           git config --global --add safe.directory /__w/cc-metric-collector/cc-metric-collector
-          export PATH=/usr/local/go/bin:/usr/local/go/pkg/tool/linux_amd64:$PATH
           make RPM
 
   #
@@ -129,16 +129,16 @@ jobs:
     # Use dnf to install build dependencies
     - name: Install build dependencies
       run: |
-          wget -q https://go.dev/dl/go1.22.4.linux-amd64.tar.gz --output-document=- | \
-          tar --directory=/usr/local --extract --gzip
-          export PATH=/usr/local/go/bin:/usr/local/go/pkg/tool/linux_amd64:$PATH
-          go version
+          dnf --assumeyes --disableplugin=subscription-manager install \
+              http://mirror.centos.org/centos/8-stream/AppStream/x86_64/os/Packages/golang-1.21.7-1.module_el8+960+4060efbe.x86_64.rpm \
+              http://mirror.centos.org/centos/8-stream/AppStream/x86_64/os/Packages/golang-bin-1.21.7-1.module_el8+960+4060efbe.x86_64.rpm \
+              http://mirror.centos.org/centos/8-stream/AppStream/x86_64/os/Packages/golang-src-1.21.7-1.module_el8+960+4060efbe.noarch.rpm \
+              http://mirror.centos.org/centos/8-stream/AppStream/x86_64/os/Packages/go-toolset-1.21.7-1.module_el8+960+4060efbe.x86_64.rpm
 
     - name: RPM build MetricCollector
       id: rpmbuild
       run: |
           git config --global --add safe.directory /__w/cc-metric-collector/cc-metric-collector
-          export PATH=/usr/local/go/bin:/usr/local/go/pkg/tool/linux_amd64:$PATH
           make RPM
 
   #
@@ -165,7 +165,7 @@ jobs:
     # Use official golang package
     - name: Install Golang
       run: |
-          wget -q https://go.dev/dl/go1.22.4.linux-amd64.tar.gz --output-document=- | \
+          wget -q https://go.dev/dl/go1.21.1.linux-amd64.tar.gz --output-document=- | \
           tar --directory=/usr/local --extract --gzip
           export PATH=/usr/local/go/bin:/usr/local/go/pkg/tool/linux_amd64:$PATH
           go version

collectors/customCmdMetric.go

@@ -99,7 +99,10 @@ func (m *CustomCmdCollector) Read(interval time.Duration, output chan lp.CCMetri
 				continue
 			}
 
-			output <- lp.FromInfluxMetric(c)
+			y := lp.FromInfluxMetric(c)
+			if err == nil {
+				output <- y
+			}
 		}
 	}
 	for _, file := range m.files {
@@ -118,7 +121,10 @@ func (m *CustomCmdCollector) Read(interval time.Duration, output chan lp.CCMetri
 			if skip {
 				continue
 			}
-			output <- lp.FromInfluxMetric(f)
+			y := lp.FromInfluxMetric(f)
+			if err == nil {
+				output <- y
+			}
 		}
 	}
 }

collectors/ipmiMetric.go

@@ -12,7 +12,6 @@ import (
 	"strconv"
 	"strings"
 	"time"
-
 	cclog "github.com/ClusterCockpit/cc-metric-collector/pkg/ccLogger"
 	lp "github.com/ClusterCockpit/cc-metric-collector/pkg/ccMetric"
 )
@@ -55,30 +54,15 @@ func (m *IpmiCollector) Init(config json.RawMessage) error {
 	// Check if executables ipmitool or ipmisensors are found
 	p, err := exec.LookPath(m.config.IpmitoolPath)
 	if err == nil {
-		command := exec.Command(p)
-		err := command.Run()
-		if err != nil {
-			cclog.ComponentError(m.name, fmt.Sprintf("Failed to execute %s: %v", p, err.Error()))
-			m.ipmitool = ""
-		} else {
-			m.ipmitool = p
-		}
+		m.ipmitool = p
 	}
 	p, err = exec.LookPath(m.config.IpmisensorsPath)
 	if err == nil {
-		command := exec.Command(p)
-		err := command.Run()
-		if err != nil {
-			cclog.ComponentError(m.name, fmt.Sprintf("Failed to execute %s: %v", p, err.Error()))
-			m.ipmisensors = ""
-		} else {
-			m.ipmisensors = p
-		}
+		m.ipmisensors = p
 	}
 	if len(m.ipmitool) == 0 && len(m.ipmisensors) == 0 {
-		return errors.New("no usable IPMI reader found")
+		return errors.New("no IPMI reader found")
 	}
-
 	m.init = true
 	return nil
 }
@@ -135,8 +119,8 @@ func (m *IpmiCollector) readIpmiTool(cmd string, output chan lp.CCMetric) {
 		cclog.ComponentError(
 			m.name,
 			fmt.Sprintf("readIpmiTool(): Failed to wait for the end of command \"%s\": %v\n", command.String(), err),
+			fmt.Sprintf("readIpmiTool(): command stderr: \"%s\"\n", string(errMsg)),
 		)
-		cclog.ComponentError(m.name, fmt.Sprintf("readIpmiTool(): command stderr: \"%s\"\n", strings.TrimSpace(string(errMsg))))
 		return
 	}
 }

docs/configuration.md

@@ -12,8 +12,8 @@ The global file contains the paths to the other four files and some global optio
   "collectors" : "collectors.json",
   "receivers" : "receivers.json",
   "router" : "router.json",
-  "interval": "10s",
-  "duration": "1s"
+  "interval": 10,
+  "duration": 1
 }
 ```
 

receivers/natsReceiver.go

@@ -4,7 +4,6 @@ import (
 	"encoding/json"
 	"errors"
 	"fmt"
-	"os"
 	"time"
 
 	cclog "github.com/ClusterCockpit/cc-metric-collector/pkg/ccLogger"
@@ -14,13 +13,10 @@ import (
 )
 
 type NatsReceiverConfig struct {
-	Type     string `json:"type"`
-	Addr     string `json:"address"`
-	Port     string `json:"port"`
-	Subject  string `json:"subject"`
-	User     string `json:"user,omitempty"`
-	Password string `json:"password,omitempty"`
-	NkeyFile string `json:"nkey_file,omitempty"`
+	Type    string `json:"type"`
+	Addr    string `json:"address"`
+	Port    string `json:"port"`
+	Subject string `json:"subject"`
 }
 
 type NatsReceiver struct {
@@ -113,7 +109,6 @@ func (r *NatsReceiver) Close() {
 
 // NewNatsReceiver creates a new Receiver which subscribes to messages from a NATS server
 func NewNatsReceiver(name string, config json.RawMessage) (Receiver, error) {
-	var uinfo nats.Option = nil
 	r := new(NatsReceiver)
 	r.name = fmt.Sprintf("NatsReceiver(%s)", name)
 
@@ -138,22 +133,10 @@ func NewNatsReceiver(name string, config json.RawMessage) (Receiver, error) {
 		"source": r.name,
 	}
 
-	if len(r.config.User) > 0 && len(r.config.Password) > 0 {
-		uinfo = nats.UserInfo(r.config.User, r.config.Password)
-	} else if len(r.config.NkeyFile) > 0 {
-		_, err := os.Stat(r.config.NkeyFile)
-		if err == nil {
-			uinfo = nats.UserCredentials(r.config.NkeyFile)
-		} else {
-			cclog.ComponentError(r.name, "NKEY file", r.config.NkeyFile, "does not exist: %v", err.Error())
-			return nil, err
-		}
-	}
-
 	// Connect to NATS server
 	url := fmt.Sprintf("nats://%s:%s", r.config.Addr, r.config.Port)
 	cclog.ComponentDebug(r.name, "NewNatsReceiver", url, "Subject", r.config.Subject)
-	if nc, err := nats.Connect(url, uinfo); err == nil {
+	if nc, err := nats.Connect(url); err == nil {
 		r.nc = nc
 	} else {
 		r.nc = nil

receivers/natsReceiver.md

@@ -10,10 +10,7 @@ The `nats` receiver can be used receive metrics from the NATS network. The `nats
     "type": "nats",
     "address" : "nats-server.example.org",
     "port" : "4222",
-    "subject" : "subject",
-    "user": "natsuser",
-    "password": "natssecret",
-    "nkey_file": "/path/to/nkey_file"
+    "subject" : "subject"
   }
 }
 ```
@@ -22,9 +19,6 @@ The `nats` receiver can be used receive metrics from the NATS network. The `nats
 - `address`: Address of the NATS control server
 - `port`: Port of the NATS control server
 - `subject`: Subscribes to this subject and receive metrics
-- `user`: Connect to nats using this user
-- `password`: Connect to nats using this password
-- `nkey_file`: Path to credentials file with NKEY
 
 ### Debugging
 

sinks/README.md

@@ -17,7 +17,7 @@ This folder contains the SinkManager and sink implementations for the cc-metric-
 The configuration file for the sinks is a list of configurations. The `type` field in each specifies which sink to initialize.
 
 ```json
-{
+[
   "mystdout" : {
     "type" : "stdout",
     "meta_as_tags" : [
@@ -31,7 +31,7 @@ The configuration file for the sinks is a list of configurations. The `type` fie
     "database" : "ccmetric",
     "password" : "<jwt token>"
   }
-}
+]
 ```
 
 

sinks/natsSink.go

@@ -5,7 +5,6 @@ import (
 	"encoding/json"
 	"errors"
 	"fmt"
-	"os"
 	"sync"
 	"time"
 
@@ -23,7 +22,6 @@ type NatsSinkConfig struct {
 	User       string `json:"user,omitempty"`
 	Password   string `json:"password,omitempty"`
 	FlushDelay string `json:"flush_delay,omitempty"`
-	NkeyFile   string `json:"nkey_file,omitempty"`
 }
 
 type NatsSink struct {
@@ -44,13 +42,6 @@ func (s *NatsSink) connect() error {
 	var nc *nats.Conn
 	if len(s.config.User) > 0 && len(s.config.Password) > 0 {
 		uinfo = nats.UserInfo(s.config.User, s.config.Password)
-	} else if len(s.config.NkeyFile) > 0 {
-		if _, err := os.Stat(s.config.NkeyFile); err == nil {
-			uinfo = nats.UserCredentials(s.config.NkeyFile)
-		} else {
-			cclog.ComponentError(s.name, "NKEY file", s.config.NkeyFile, "does not exist: %v", err.Error())
-			return err
-		}
 	}
 	uri := fmt.Sprintf("nats://%s:%s", s.config.Host, s.config.Port)
 	cclog.ComponentDebug(s.name, "Connect to", uri)

sinks/natsSink.md

@@ -13,7 +13,6 @@ The `nats` sink publishes all metrics into a NATS network. The publishing key is
     "port": "4222",
     "user": "exampleuser",
     "password" : "examplepw",
-    "nkey_file": "/path/to/nkey_file",
     "meta_as_tags" : [],
   }
 }
@@ -26,4 +25,3 @@ The `nats` sink publishes all metrics into a NATS network. The publishing key is
 - `user`: Username for basic authentication
 - `password`: Password for basic authentication
 - `meta_as_tags`: print all meta information as tags in the output (optional)
-- `nkey_file`: Path to credentials file with NKEY

sinks/syslogSink.go

@@ -0,0 +1,159 @@
+package sinks
+
+import (
+	"bytes"
+	"encoding/json"
+	"errors"
+	"fmt"
+	"log/syslog"
+
+	cclog "github.com/ClusterCockpit/cc-metric-collector/pkg/ccLogger"
+	lp "github.com/ClusterCockpit/cc-metric-collector/pkg/ccMetric"
+)
+
+var AvailableSyslogPriorities map[string]syslog.Priority = map[string]syslog.Priority{
+	"LOG_EMERG":    syslog.LOG_EMERG,
+	"LOG_ALERT":    syslog.LOG_ALERT,
+	"LOG_CRIT":     syslog.LOG_CRIT,
+	"LOG_ERR":      syslog.LOG_ERR,
+	"LOG_NOTICE":   syslog.LOG_NOTICE,
+	"LOG_INFO":     syslog.LOG_INFO,
+	"LOG_DEBUG":    syslog.LOG_DEBUG,
+	"LOG_USER":     syslog.LOG_USER,
+	"LOG_MAIL":     syslog.LOG_MAIL,
+	"LOG_DAEMON":   syslog.LOG_DAEMON,
+	"LOG_AUTH":     syslog.LOG_AUTH,
+	"LOG_SYSLOG":   syslog.LOG_SYSLOG,
+	"LOG_LPR":      syslog.LOG_LPR,
+	"LOG_NEWS":     syslog.LOG_NEWS,
+	"LOG_UUCP":     syslog.LOG_UUCP,
+	"LOG_CRON":     syslog.LOG_CRON,
+	"LOG_AUTHPRIV": syslog.LOG_AUTHPRIV,
+	"LOG_FTP":      syslog.LOG_FTP,
+	"LOG_LOCAL0":   syslog.LOG_LOCAL0,
+	"LOG_LOCAL1":   syslog.LOG_LOCAL1,
+	"LOG_LOCAL2":   syslog.LOG_LOCAL2,
+	"LOG_LOCAL3":   syslog.LOG_LOCAL3,
+	"LOG_LOCAL4":   syslog.LOG_LOCAL4,
+	"LOG_LOCAL5":   syslog.LOG_LOCAL5,
+	"LOG_LOCAL6":   syslog.LOG_LOCAL6,
+	"LOG_LOCAL7":   syslog.LOG_LOCAL7,
+}
+
+type SyslogSinkConfig struct {
+	defaultSinkConfig
+	SyslogTag      string   `json:"syslog_tag"`
+	SyslogPriories []string `json:"syslog_priorities"`
+	SyslogWriter   string   `json:"syslog_writer"`
+}
+
+type SyslogSink struct {
+	sink
+	config      SyslogSinkConfig // entry point to the SyslogSinkConfig
+	syslogPrios syslog.Priority
+	out         *syslog.Writer
+}
+
+func (s *SyslogSink) Write(point lp.CCMetric) error {
+	var err error = nil
+	p := point.ToLineProtocol(s.meta_as_tags)
+	switch s.config.SyslogWriter {
+	case "info":
+		err = s.out.Info(p)
+	case "debug":
+		err = s.out.Debug(p)
+	case "alert":
+		err = s.out.Alert(p)
+	case "emerg":
+		err = s.out.Emerg(p)
+	case "err":
+		err = s.out.Err(p)
+	case "notice":
+		err = s.out.Notice(p)
+	case "warning":
+		err = s.out.Warning(p)
+	}
+
+	return err
+}
+
+func (s *SyslogSink) Flush() error {
+	return nil
+}
+
+func (s *SyslogSink) Close() {
+	s.out.Close()
+	cclog.ComponentDebug(s.name, "CLOSE")
+}
+
+func NewSyslogSink(name string, config json.RawMessage) (Sink, error) {
+	var err error = nil
+	s := new(SyslogSink)
+
+	s.name = fmt.Sprintf("SyslogSink(%s)", name)
+
+	// Read in the config JSON
+	if len(config) > 0 {
+		d := json.NewDecoder(bytes.NewReader(config))
+		d.DisallowUnknownFields()
+		if err := d.Decode(&s.config); err != nil {
+			cclog.ComponentError(s.name, "Error reading config:", err.Error())
+			return nil, err
+		}
+	}
+
+	// Create lookup map to use meta infos as tags in the output metric
+	s.meta_as_tags = make(map[string]bool)
+	for _, k := range s.config.MetaAsTags {
+		s.meta_as_tags[k] = true
+	}
+
+	if len(s.config.SyslogTag) == 0 {
+		err := errors.New("syslog tag must be non-empty")
+		cclog.ComponentError(s.name, err.Error())
+		return nil, err
+	}
+
+	if len(s.config.SyslogPriories) == 0 {
+		err := errors.New("syslog prioritiey must be non-empty")
+		cclog.ComponentError(s.name, err.Error())
+		return nil, err
+	}
+	for _, p := range s.config.SyslogPriories {
+		if prio, ok := AvailableSyslogPriorities[p]; !ok {
+			err := fmt.Errorf("invalid syslog priority '%s'", p)
+			cclog.ComponentError(s.name, err.Error())
+			return nil, err
+		} else {
+			s.syslogPrios |= prio
+		}
+	}
+
+	mywriter := ""
+	for _, w := range []string{
+		"info",
+		"debug",
+		"alert",
+		"emerg",
+		"err",
+		"notice",
+		"warning",
+	} {
+		if s.config.SyslogWriter == w {
+			mywriter = w
+		}
+	}
+	if len(mywriter) == 0 {
+		err := fmt.Errorf("invalid syslog writer '%s'", s.config.SyslogWriter)
+		cclog.ComponentError(s.name, err.Error())
+		return nil, err
+	}
+
+	s.out, err = syslog.New(s.syslogPrios, s.config.SyslogTag)
+	if err != nil {
+		cclog.ComponentError(s.name, err.Error())
+		return nil, err
+	}
+
+	return s, nil
+}

sinks/syslogSink.md

@@ -0,0 +1,64 @@
+## `syslog` sink
+
+The `syslog` sink provides an easy way to submit metrics and events to the Syslog logging system.
+
+### Configuration structure
+
+```json
+{
+  "<name>": {
+    "type": "syslog",
+    "meta_as_tags" : [],
+    "syslog_tag" : "mytag",
+    "syslog_priorities" : [
+        "LOG_INFO",
+        "LOG_LOCAL7",
+    ],
+    "syslog_writer" : "info"
+  }
+}
+```
+
+- `type`: makes the sink an `syslog` sink
+- `meta_as_tags`: print meta information as tags in the output (optional)
+- `syslog_tag`: The tag submitted with each message
+- `syslog_priorities`: List of syslog priorities. Will be OR'd together
+- `syslog_writer`: Submit metrics with this write level
+
+
+#### Possible priorities for `syslog_priorities`
+- `LOG_EMERG`
+- `LOG_ALERT`
+- `LOG_CRIT`
+- `LOG_ERR`
+- `LOG_NOTICE`
+- `LOG_INFO`
+- `LOG_DEBUG`
+- `LOG_USER`
+- `LOG_MAIL`
+- `LOG_DAEMON`
+- `LOG_AUTH`
+- `LOG_SYSLOG`
+- `LOG_LPR`
+- `LOG_NEWS`
+- `LOG_UUCP`
+- `LOG_CRON`
+- `LOG_AUTHPRIV`
+- `LOG_FTP`
+- `LOG_LOCAL0`
+- `LOG_LOCAL1`
+- `LOG_LOCAL2`
+- `LOG_LOCAL3`
+- `LOG_LOCAL4`
+- `LOG_LOCAL5`
+- `LOG_LOCAL6`
+- `LOG_LOCAL7`
+
+#### Possible writers for `syslog_writer`
+- `info`
+- `debug`
+- `alert`
+- `emerg`
+- `err`
+- `notice`
+- `warning`