Merge branch 'main' into alternate_storage

This commit is contained in:
Thomas Gruber 2021-10-04 15:49:46 +02:00 committed by GitHub
commit d7ef32de18
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
4 changed files with 20 additions and 6 deletions

View File

@ -12,6 +12,7 @@ import (
) )
const LIDFILE = `/sys/class/infiniband/mlx4_0/ports/1/lid` const LIDFILE = `/sys/class/infiniband/mlx4_0/ports/1/lid`
const PERFQUERY = `/usr/sbin/perfquery`
type InfinibandCollector struct { type InfinibandCollector struct {
MetricCollector MetricCollector
@ -24,7 +25,10 @@ func (m *InfinibandCollector) Init() error {
m.tags = map[string]string{"type": "node"} m.tags = map[string]string{"type": "node"}
_, err := ioutil.ReadFile(string(LIDFILE)) _, err := ioutil.ReadFile(string(LIDFILE))
if err == nil { if err == nil {
_, err = ioutil.ReadFile(string(PERFQUERY))
if err == nil {
m.init = true m.init = true
}
} }
return err return err
} }
@ -39,7 +43,7 @@ func (m *InfinibandCollector) Read(interval time.Duration, out *[]lp.MutableMetr
args := fmt.Sprintf("-r %s 1 0xf000", string(buffer)) args := fmt.Sprintf("-r %s 1 0xf000", string(buffer))
command := exec.Command("/usr/sbin/perfquery", args) command := exec.Command(PERFQUERY, args)
command.Wait() command.Wait()
stdout, err := command.Output() stdout, err := command.Output()
if err != nil { if err != nil {

View File

@ -5,7 +5,7 @@ The base class/configuration is located in `metricSink.go`.
# Sinks # Sinks
* `stdoutSink.go`: Writes all metrics to `stdout` in InfluxDB line protocol. The sink does not use https://github.com/influxdata/line-protocol to reduce the executed code for debugging * `stdoutSink.go`: Writes all metrics to `stdout` in InfluxDB line protocol. The sink does not use https://github.com/influxdata/line-protocol to reduce the executed code for debugging
* `influxSink.go`: Writes all metrics to an InfluxDB database instance using a blocking writer. It uses https://github.com/influxdata/influxdb-client-go . Configuration for the server, port, user, password, database name and organisation are in the global configuration file. It uses the v2 API of Influx. * `influxSink.go`: Writes all metrics to an InfluxDB database instance using a blocking writer. It uses https://github.com/influxdata/influxdb-client-go . Configuration for the server, port, ssl, password, database name and organisation are in the global configuration file. The 'password' is used for the token and the 'database' for the bucket. It uses the v2 API of Influx.
* `natsSink.go`: Sends all metrics to an NATS server using the InfluxDB line protocol as encoding. It uses https://github.com/nats-io/nats.go . Configuration for the server, port, user, password and database name are in the global configuration file. The database name is used as subject for the NATS messages. * `natsSink.go`: Sends all metrics to an NATS server using the InfluxDB line protocol as encoding. It uses https://github.com/nats-io/nats.go . Configuration for the server, port, user, password and database name are in the global configuration file. The database name is used as subject for the NATS messages.
# Installation # Installation

View File

@ -2,6 +2,7 @@ package sinks
import ( import (
"context" "context"
"crypto/tls"
"errors" "errors"
"fmt" "fmt"
influxdb2 "github.com/influxdata/influxdb-client-go/v2" influxdb2 "github.com/influxdata/influxdb-client-go/v2"
@ -19,14 +20,20 @@ type InfluxSink struct {
func (s *InfluxSink) connect() error { func (s *InfluxSink) connect() error {
var auth string var auth string
uri := fmt.Sprintf("http://%s:%s", s.host, s.port) var uri string
if s.ssl {
uri = fmt.Sprintf("https://%s:%s", s.host, s.port)
} else {
uri = fmt.Sprintf("http://%s:%s", s.host, s.port)
}
if len(s.user) == 0 { if len(s.user) == 0 {
auth = s.password auth = s.password
} else { } else {
auth = fmt.Sprintf("%s:%s", s.user, s.password) auth = fmt.Sprintf("%s:%s", s.user, s.password)
} }
log.Print("Using URI ", uri, " Org ", s.organization, " Bucket ", s.database) log.Print("Using URI ", uri, " Org ", s.organization, " Bucket ", s.database)
s.client = influxdb2.NewClient(uri, auth) s.client = influxdb2.NewClientWithOptions(uri, auth,
influxdb2.DefaultOptions().SetTLSConfig(&tls.Config{InsecureSkipVerify: true}))
s.writeApi = s.client.WriteAPIBlocking(s.organization, s.database) s.writeApi = s.client.WriteAPIBlocking(s.organization, s.database)
return nil return nil
} }
@ -45,6 +52,7 @@ func (s *InfluxSink) Init(config SinkConfig) error {
s.organization = config.Organization s.organization = config.Organization
s.user = config.User s.user = config.User
s.password = config.Password s.password = config.Password
s.ssl = config.SSL
return s.connect() return s.connect()
} }

View File

@ -13,6 +13,7 @@ type SinkConfig struct {
Password string `json:"password"` Password string `json:"password"`
Organization string `json:"organization"` Organization string `json:"organization"`
Type string `json:"type"` Type string `json:"type"`
SSL bool `json:"ssl"`
} }
type Sink struct { type Sink struct {
@ -22,6 +23,7 @@ type Sink struct {
password string password string
database string database string
organization string organization string
ssl bool
} }
type SinkFuncs interface { type SinkFuncs interface {