ClusterCockpit/cc-metric-collector
1
0
Fork 0
mirror of https://github.com/ClusterCockpit/cc-metric-collector.git synced 2024-12-25 23:19:06 +01:00
Code Issues Packages Projects Releases Wiki Activity

Fix: influx sink ignores config batch_size.

Browse Source 
Feature: Add Redfish receiver
This commit is contained in:
Holger Obermaier 2022-05-04 13:00:56 +02:00
commit 186a62a86b
12 changed files with 528 additions and 113 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

4
collectors/diskstatMetric.go
View File

@ -3,7 +3,6 @@ package collectors
import (
"bufio"
"encoding/json"
"fmt"
"os"
"strings"
"syscall"
@ -81,8 +80,7 @@ func (m *DiskstatCollector) Read(interval time.Duration, output chan lp.CCMetric
stat := syscall.Statfs_t{}
err := syscall.Statfs(path, &stat)
if err != nil {
fmt.Println(err.Error())
return
continue
}
tags := map[string]string{"type": "node", "device": linefields[0]}
total := (stat.Blocks * uint64(stat.Bsize)) / uint64(1000000000)

14
collectors/gpfsMetric.go
View File

@ -70,6 +70,7 @@ func (m *GpfsCollector) Init(config json.RawMessage) error {
for _, fs := range m.config.ExcludeFilesystem {
m.skipFS[fs] = struct{}{}
}
m.lastState = make(map[string]GpfsCollectorLastState)
// GPFS / IBM Spectrum Scale file system statistics can only be queried by user root
user, err := user.Current()
@ -162,11 +163,16 @@ func (m *GpfsCollector) Read(interval time.Duration, output chan lp.CCMetric) {
continue
}
// Add filesystem tag
m.tags["filesystem"] = filesystem
if _, ok := m.lastState[filesystem]; !ok {
m.lastState[filesystem] = GpfsCollectorLastState{
bytesRead: -1,
bytesWritten: -1,
// Create initial last state
if m.config.SendBandwidths {
if _, ok := m.lastState[filesystem]; !ok {
m.lastState[filesystem] = GpfsCollectorLastState{
bytesRead: -1,
bytesWritten: -1,
}
}
}

28
collectors/likwidMetric.go
View File

@ -329,7 +329,11 @@ func (m *LikwidCollector) calcEventsetMetrics(evset LikwidEventsetConfig, interv
gctr := C.GoString(counter)
for _, tid := range m.cpu2tid {
res := C.perfmon_getLastResult(evset.gid, C.int(eidx), C.int(tid))
evset.results[tid][gctr] = float64(res)
fres := float64(res)
if m.config.InvalidToZero && (math.IsNaN(fres) || math.IsInf(fres, 0)) {
fres = 0.0
}
evset.results[tid][gctr] = fres
evset.results[tid]["time"] = interval.Seconds()
evset.results[tid]["inverseClock"] = invClock
}
@ -348,15 +352,12 @@ func (m *LikwidCollector) calcEventsetMetrics(evset LikwidEventsetConfig, interv
value, err := agg.EvalFloat64Condition(metric.Calc, evset.results[tid])
if err != nil {
cclog.ComponentError(m.name, "Calculation for metric", metric.Name, "failed:", err.Error())
continue
value = 0.0
}
if m.config.InvalidToZero && (math.IsNaN(value) || math.IsInf(value, 0)) {
value = 0.0
}
evset.metrics[tid][metric.Name] = value
if m.config.InvalidToZero && math.IsNaN(value) {
value = 0.0
}
if m.config.InvalidToZero && math.IsInf(value, 0) {
value = 0.0
}
// Now we have the result, send it with the proper tags
if !math.IsNaN(value) {
if metric.Publish {
@ -400,15 +401,12 @@ func (m *LikwidCollector) calcGlobalMetrics(interval time.Duration, output chan
value, err := agg.EvalFloat64Condition(metric.Calc, params)
if err != nil {
cclog.ComponentError(m.name, "Calculation for metric", metric.Name, "failed:", err.Error())
continue
value = 0.0
}
if m.config.InvalidToZero && (math.IsNaN(value) || math.IsInf(value, 0)) {
value = 0.0
}
m.gmresults[tid][metric.Name] = value
if m.config.InvalidToZero && math.IsNaN(value) {
value = 0.0
}
if m.config.InvalidToZero && math.IsInf(value, 0) {
value = 0.0
}
// Now we have the result, send it with the proper tags
if !math.IsNaN(value) {
if metric.Publish {

99
collectors/likwidMetric.md
View File

@ -3,32 +3,63 @@
The `likwid` collector is probably the most complicated collector. The LIKWID library is included as static library with *direct* access mode. The *direct* access mode is suitable if the daemon is executed by a root user. The static library does not contain the performance groups, so all information needs to be provided in the configuration.
The `likwid` configuration consists of two parts, the "eventsets" and "globalmetrics":
- An event set list itself has two parts, the "events" and a set of derivable "metrics". Each of the "events" is a counter:event pair in LIKWID's syntax. The "metrics" are a list of formulas to derive the metric value from the measurements of the "events". Each metric has a name, the formula, a scope and a publish flag. Counter names can be used like variables in the formulas, so `PMC0+PMC1` sums the measurements for the both events configured in the counters `PMC0` and `PMC1`. The scope tells the Collector whether it is a metric for each hardware thread (`cpu`) or each CPU socket (`socket`). You may specify a unit for the metric with `unit`. The last one is the publishing flag. It tells the collector whether a metric should be sent to the router.
- The global metrics are metrics which require data from all event set measurements to be derived. The inputs are the metrics in the event sets. Similar to the metrics in the event sets, the global metrics are defined by a name, a formula, a scope and a publish flag. See event set metrics for details. The only difference is that there is no access to the raw event measurements anymore but only to the metrics. So, the idea is to derive a metric in the "eventsets" section and reuse it in the "globalmetrics" part. If you need a metric only for deriving the global metrics, disable forwarding of the event set metrics (`publish=false`). **Be aware** that the combination might be misleading because the "behavior" of a metric changes over time and the multiple measurements might count different computing phases. Similar to the metrics in the eventset, you can specify a metric unit with the `unit` field.
```json
"likwid": {
"force_overwrite" : false,
"invalid_to_zero" : false,
"eventsets": [
{
"events" : {
"COUNTER0": "EVENT0",
"COUNTER1": "EVENT1",
},
"metrics" : [
{
"name": "sum_01",
"calc": "COUNTER0 + COUNTER1",
"publish": false,
"unit": "myunit",
"type": "cpu"
}
]
}
]
"globalmetrics" : [
{
"name": "global_sum",
"calc": "sum_01",
"publish": true,
"unit": "myunit",
"type": "cpu"
}
]
}
```
The `likwid` configuration consists of two parts, the `eventsets` and `globalmetrics`:
- An event set list itself has two parts, the `events` and a set of derivable `metrics`. Each of the `events` is a `counter:event` pair in LIKWID's syntax. The `metrics` are a list of formulas to derive the metric value from the measurements of the `events`' values. Each metric has a name, the formula, a type and a publish flag. There is an optional `unit` field. Counter names can be used like variables in the formulas, so `PMC0+PMC1` sums the measurements for the both events configured in the counters `PMC0` and `PMC1`. You can optionally use `time` for the measurement time and `inverseClock` for `1.0/baseCpuFrequency`. The type tells the LikwidCollector whether it is a metric for each hardware thread (`cpu`) or each CPU socket (`socket`). You may specify a unit for the metric with `unit`. The last one is the publishing flag. It tells the LikwidCollector whether a metric should be sent to the router or is only used internally to compute a global metric.
- The `globalmetrics` are metrics which require data from multiple event set measurements to be derived. The inputs are the metrics in the event sets. Similar to the metrics in the event sets, the global metrics are defined by a name, a formula, a scope and a publish flag. See event set metrics for details. The only difference is that there is no access to the raw event measurements anymore but only to the metrics. Also `time` and `inverseClock` cannot be used anymore. So, the idea is to derive a metric in the `eventsets` section and reuse it in the `globalmetrics` part. If you need a metric only for deriving the global metrics, disable forwarding of the event set metrics (`"publish": false`). **Be aware** that the combination might be misleading because the "behavior" of a metric changes over time and the multiple measurements might count different computing phases. Similar to the metrics in the eventset, you can specify a metric unit with the `unit` field.
Additional options:
- `access_mode` : Method to use for hardware performance monitoring (`direct` access as root user, `accessdaemon` for the daemon mode)
- `accessdaemon_path`: Folder with the access daemon `likwid-accessD`, commonly `$LIKWID_INSTALL_LOC/sbin`
- `force_overwrite`: Same as setting `LIKWID_FORCE=1`. In case counters are already in-use, LIKWID overwrites their configuration to do its measurements
- `invalid_to_zero`: In some cases, the calculations result in `NaN` or `Inf`. With this option, all `NaN` and `Inf` values are replaces with `0.0`.
- `access_mode`: Specify LIKWID access mode: `direct` for direct register access as root user or `accessdaemon`
- `accessdaemon_path`: Folder of the accessDaemon `likwid-accessD`
- `liblikwid_path`: Location of `liblikwid.so`
- `invalid_to_zero`: In some cases, the calculations result in `NaN` or `Inf`. With this option, all `NaN` and `Inf` values are replaces with `0.0`. See below in [seperate section](./likwidMetric.md#invalid_to_zero-option)
- `access_mode`: Specify LIKWID access mode: `direct` for direct register access as root user or `accessdaemon`. The access mode `perf_event` is current untested.
- `accessdaemon_path`: Folder of the accessDaemon `likwid-accessD` (like `/usr/local/sbin`)
- `liblikwid_path`: Location of `liblikwid.so` including file name like `/usr/local/lib/liblikwid.so`
### Available metric scopes
Hardware performance counters are scattered all over the system nowadays. A counter coveres a specific part of the system. While there are hardware thread specific counter for CPU cycles, instructions and so on, some others are specific for a whole CPU socket/package. To address that, the collector provides the specification of a 'scope' for each metric.
Hardware performance counters are scattered all over the system nowadays. A counter coveres a specific part of the system. While there are hardware thread specific counter for CPU cycles, instructions and so on, some others are specific for a whole CPU socket/package. To address that, the LikwidCollector provides the specification of a `type` for each metric.
- `cpu` : One metric per CPU hardware thread with the tags `"type" : "cpu"` and `"type-id" : "$cpu_id"`
- `socket` : One metric per CPU socket/package with the tags `"type" : "socket"` and `"type-id" : "$socket_id"`
**Note:** You cannot specify `socket` scope for a metric that is measured at `cpu` scope, so some kind of expert knowledge or lookup work in the [Likwid Wiki](https://github.com/RRZE-HPC/likwid/wiki) is required. Get the scope of each counter from the *Architecture* pages and as soon as one counter in a metric is socket-specific, the whole metric is socket-specific.
**Note:** You should not specify the `socket` type for a metric that is measured at `cpu` scope and vice versa, so some kind of expert knowledge or lookup work in the [Likwid Wiki](https://github.com/RRZE-HPC/likwid/wiki) is required. Get the scope of each counter from the *Architecture* pages and as soon as one counter in a metric is socket-specific, the whole metric is socket-specific.
As a guideline:
- All counters `FIXCx`, `PMCy` and `TMAz` have the scope `cpu`
- All counters names containing `BOX` have the scope `socket`
- All `PWRx` counters have scope `socket`, except `"PWR1" : "RAPL_CORE_ENERGY"` has `cpu` scope
- All `PWRx` counters have scope `socket`, except `"PWR1" : "RAPL_CORE_ENERGY"` has `cpu` scope (AMD Zen)
- All `DFCx` counters have scope `socket`
### Help with the configuration
@ -50,6 +81,7 @@ $ scripts/likwid_perfgroup_to_cc_config.py ICX MEM_DP
{
"events": {
"FIXC0": "INSTR_RETIRED_ANY",
"FIXC1": "CPU_CLK_UNHALTED_CORE",
"..." : "..."
},
"metrics" : [
@ -75,21 +107,28 @@ LIKWID checks the file `/var/run/likwid.lock` before performing any interfering
Before (SLURM prolog, ...)
```
$ chwon $JOBUSER /var/run/likwid.lock
$ chown $JOBUSER /var/run/likwid.lock
```
After (SLURM epilog, ...)
```
$ chwon $CCUSER /var/run/likwid.lock
$ chown $CCUSER /var/run/likwid.lock
```
### `invalid_to_zero` option
In some cases LIKWID returns `0.0` for some events that are further used in processing and maybe used as divisor in a calculation. After evaluation of a metric, the result might be `NaN` or `+-Inf`. These resulting metrics are commonly not created and forwarded to the router because the [InfluxDB line protocol](https://docs.influxdata.com/influxdb/cloud/reference/syntax/line-protocol/#float) does not support these special floating-point values. If you want to have them sent, this option forces these metric values to be `0.0` instead.
One might think this does not happen often but often used metrics in the world of performance engineering like Instructions-per-Cycle (IPC) or more frequently the actual CPU clock are derived with events like `CPU_CLK_UNHALTED_CORE` (Intel) which do not increment in halted state (as the name implies). In there are different power management systems in a chip which can cause a hardware thread to go in such a state. Moreover, if no cycles are executed by the core, also many other events are not incremented as well (like `INSTR_RETIRED_ANY` for retired instructions and part of IPC).
### Example configuration
#### AMD Zen3
```json
"likwid": {
"force_overwrite" : false,
"nan_to_zero" : false,
"invalid_to_zero" : false,
"eventsets": [
{
"events": {
@ -180,33 +219,3 @@ $ chwon $CCUSER /var/run/likwid.lock
}
```
### How to get the eventsets and metrics from LIKWID
The `likwid` collector reads hardware performance counters at a **cpu** and **socket** level. The configuration looks quite complicated but it is basically copy&paste from [LIKWID's performance groups](https://github.com/RRZE-HPC/likwid/tree/master/groups). The collector made multiple iterations and tried to use the performance groups but it lacked flexibility. The current way of configuration provides most flexibility.
The logic is as following: There are multiple eventsets, each consisting of a list of counters+events and a list of metrics. If you compare a common performance group with the example setting above, there is not much difference:
```
EVENTSET -> "events": {
FIXC1 ACTUAL_CPU_CLOCK -> "FIXC1": "ACTUAL_CPU_CLOCK",
FIXC2 MAX_CPU_CLOCK -> "FIXC2": "MAX_CPU_CLOCK",
PMC0 RETIRED_INSTRUCTIONS -> "PMC0" : "RETIRED_INSTRUCTIONS",
PMC1 CPU_CLOCKS_UNHALTED -> "PMC1" : "CPU_CLOCKS_UNHALTED",
PMC2 RETIRED_SSE_AVX_FLOPS_ALL -> "PMC2": "RETIRED_SSE_AVX_FLOPS_ALL",
PMC3 MERGE -> "PMC3": "MERGE",
-> }
```
The metrics are following the same procedure:
```
METRICS -> "metrics": [
IPC PMC0/PMC1 -> {
-> "name" : "IPC",
-> "calc" : "PMC0/PMC1",
-> "scope": "cpu",
-> "publish": true
-> }
-> ]
```
The script `scripts/likwid_perfgroup_to_cc_config.py` might help you.

21
go.mod
View File

@ -3,17 +3,14 @@ module github.com/ClusterCockpit/cc-metric-collector
go 1.16
require (
github.com/NVIDIA/go-nvml v0.11.1-0
github.com/influxdata/influxdb-client-go/v2 v2.7.0
github.com/influxdata/line-protocol v0.0.0-20210922203350-b1ad95c89adf
github.com/nats-io/nats.go v1.13.1-0.20211122170419-d7c1d78a50fc
golang.org/x/sys v0.0.0-20220114195835-da31bd327af9
gopkg.in/Knetic/govaluate.v2 v2.3.0
)
require (
github.com/NVIDIA/go-nvml v0.11.6-0
github.com/PaesslerAG/gval v1.1.2
github.com/golang/protobuf v1.5.2 // indirect
github.com/nats-io/nats-server/v2 v2.7.0 // indirect
google.golang.org/protobuf v1.27.1 // indirect
github.com/gorilla/mux v1.8.0
github.com/influxdata/influxdb-client-go/v2 v2.8.1
github.com/influxdata/line-protocol v0.0.0-20210922203350-b1ad95c89adf
github.com/nats-io/nats-server/v2 v2.8.0 // indirect
github.com/nats-io/nats.go v1.14.0
github.com/prometheus/client_golang v1.12.1
github.com/stmcginnis/gofish v0.13.0
golang.org/x/sys v0.0.0-20220412211240-33da011f77ad
)

17
receivers.json
View File

@ -4,5 +4,22 @@
"address": "nats://my-url",
"port" : "4222",
"database": "testcluster"
},
"redfish_recv": {
"type": "redfish",
"client_config": [
{
"hostname": "my-host-1",
"username": "username-1",
"password": "password-1",
"endpoint": "https://my-endpoint-1"
},
{
"hostname": "my-host-2",
"username": "username-2",
"password": "password-2",
"endpoint": "https://my-endpoint-2"
}
]
}
}

18
receivers/receiveManager.go
View File

@ -10,14 +10,13 @@ import (
)
var AvailableReceivers = map[string]func(name string, config json.RawMessage) (Receiver, error){
"nats": NewNatsReceiver,
"nats": NewNatsReceiver,
"redfish": NewRedfishReceiver,
}
type receiveManager struct {
inputs []Receiver
output chan lp.CCMetric
done chan bool
wg *sync.WaitGroup
config []json.RawMessage
}
@ -33,8 +32,6 @@ func (rm *receiveManager) Init(wg *sync.WaitGroup, receiverConfigFile string) er
// Initialize struct fields
rm.inputs = make([]Receiver, 0)
rm.output = nil
rm.done = make(chan bool)
rm.wg = wg
rm.config = make([]json.RawMessage, 0)
configFile, err := os.Open(receiverConfigFile)
@ -58,7 +55,7 @@ func (rm *receiveManager) Init(wg *sync.WaitGroup, receiverConfigFile string) er
}
func (rm *receiveManager) Start() {
rm.wg.Add(1)
cclog.ComponentDebug("ReceiveManager", "START")
for _, r := range rm.inputs {
cclog.ComponentDebug("ReceiveManager", "START", r.Name())
@ -97,16 +94,19 @@ func (rm *receiveManager) AddOutput(output chan lp.CCMetric) {
}
func (rm *receiveManager) Close() {
cclog.ComponentDebug("ReceiveManager", "CLOSE")
// Close all receivers
for _, r := range rm.inputs {
cclog.ComponentDebug("ReceiveManager", "CLOSE", r.Name())
r.Close()
}
rm.wg.Done()
cclog.ComponentDebug("ReceiveManager", "CLOSE")
cclog.ComponentDebug("ReceiveManager", "DONE")
}
func New(wg *sync.WaitGroup, receiverConfigFile string) (ReceiveManager, error) {
r := &receiveManager{}
r := new(receiveManager)
err := r.Init(wg, receiverConfigFile)
if err != nil {
return nil, err

324
receivers/redfishReceiver.go Normal file
View File

@ -0,0 +1,324 @@
package receivers
import (
"encoding/json"
"fmt"
"strconv"
"sync"
"time"
cclog "github.com/ClusterCockpit/cc-metric-collector/internal/ccLogger"
lp "github.com/ClusterCockpit/cc-metric-collector/internal/ccMetric"
// See: https://pkg.go.dev/github.com/stmcginnis/gofish
"github.com/stmcginnis/gofish"
)
// RedfishReceiver configuration:
type RedfishReceiver struct {
receiver
config struct {
Type string `json:"type"`
Fanout int `json:"fanout,omitempty"` // Default fanout: 64
Interval int `json:"interval,omitempty"` // Default interval: 30s
// Client config for each redfish service
ClientConfigs []struct {
Hostname *string `json:"hostname"`
Username *string `json:"username"`
Password *string `json:"password"`
Endpoint *string `json:"endpoint"`
Insecure *bool `json:"insecure,omitempty"`
ExcludeMetrics []string `json:"exclude_metrics,omitempty"`
gofish gofish.ClientConfig
} `json:"client_config"`
}
done chan bool // channel to finish / stop redfish receiver
wg sync.WaitGroup // wait group for redfish receiver
}
// Start starts the redfish receiver
func (r *RedfishReceiver) Start() {
cclog.ComponentDebug(r.name, "START")
// readPowerMetric reads readfish power metric from the endpoint configured in conf
readPowerMetric := func(clientConfigIndex int) error {
clientConfig := &r.config.ClientConfigs[clientConfigIndex]
// Connect to redfish service
c, err := gofish.Connect(clientConfig.gofish)
if err != nil {
c := struct {
Username string
Endpoint string
BasicAuth bool
Insecure bool
}{
Username: clientConfig.gofish.Username,
Endpoint: clientConfig.gofish.Endpoint,
BasicAuth: clientConfig.gofish.BasicAuth,
Insecure: clientConfig.gofish.Insecure,
}
return fmt.Errorf("readPowerMetric: gofish.Connect(%+v) failed: %v", c, err)
}
defer c.Logout()
// Get all chassis managed by this service
chassis_list, err := c.Service.Chassis()
if err != nil {
return fmt.Errorf("readPowerMetric: c.Service.Chassis() failed: %v", err)
}
for _, chassis := range chassis_list {
timestamp := time.Now()
// Get power information for each chassis
power, err := chassis.Power()
if err != nil {
return fmt.Errorf("readPowerMetric: chassis.Power() failed: %v", err)
}
if power == nil {
continue
}
// Read min, max and average consumed watts for each power control
for _, pc := range power.PowerControl {
// Map of collected metrics
metrics := map[string]float32{
// PowerConsumedWatts shall represent the actual power being consumed (in
// Watts) by the chassis
"consumed_watts": pc.PowerConsumedWatts,
// AverageConsumedWatts shall represent the
// average power level that occurred averaged over the last IntervalInMin
// minutes.
"average_consumed_watts": pc.PowerMetrics.AverageConsumedWatts,
// MinConsumedWatts shall represent the
// minimum power level in watts that occurred within the last
// IntervalInMin minutes.
"min_consumed_watts": pc.PowerMetrics.MinConsumedWatts,
// MaxConsumedWatts shall represent the
// maximum power level in watts that occurred within the last
// IntervalInMin minutes
"max_consumed_watts": pc.PowerMetrics.MaxConsumedWatts,
}
intervalInMin := strconv.FormatFloat(float64(pc.PowerMetrics.IntervalInMin), 'f', -1, 32)
// Metrics to exclude
for _, key := range clientConfig.ExcludeMetrics {
delete(metrics, key)
}
// Set tags
tags := map[string]string{
"hostname": *clientConfig.Hostname,
"type": "node",
// ID uniquely identifies the resource
"id": pc.ID,
// MemberID shall uniquely identify the member within the collection. For
// services supporting Redfish v1.6 or higher, this value shall be the
// zero-based array index.
"member_id": pc.MemberID,
// PhysicalContext shall be a description of the affected device(s) or region
// within the chassis to which this power control applies.
"physical_context": string(pc.PhysicalContext),
// Name
"power_control_name": pc.Name,
}
// Delete empty tags
for key, value := range tags {
if value == "" {
delete(tags, key)
}
}
// Set meta data tags
meta := map[string]string{
"source": r.name,
"group": "Energy",
"interval_in_minutes": intervalInMin,
"unit": "watts",
}
// Delete empty meta data tags
for key, value := range meta {
if value == "" {
delete(meta, key)
}
}
for name, value := range metrics {
y, err := lp.New(name, tags, meta,
map[string]interface{}{
"value": value,
},
timestamp)
if err == nil {
r.sink <- y
}
}
}
}
return nil
}
// doReadPowerMetric read power metrics for all configure redfish services.
// To compensate latencies of the Redfish services a fanout is used.
doReadPowerMetric := func() {
// Compute fanout to use
realFanout := r.config.Fanout
if len(r.config.ClientConfigs) < realFanout {
realFanout = len(r.config.ClientConfigs)
}
// Create wait group and input channel for workers
var workerWaitGroup sync.WaitGroup
workerInput := make(chan int, realFanout)
// Create worker go routines
for i := 0; i < realFanout; i++ {
// Increment worker wait group counter
workerWaitGroup.Add(1)
go func() {
// Decrement worker wait group counter
defer workerWaitGroup.Done()
// Read power metrics for each client config
for clientConfigIndex := range workerInput {
err := readPowerMetric(clientConfigIndex)
if err != nil {
cclog.ComponentError(r.name, err)
}
}
}()
}
// Distribute client configs to workers
for i := range r.config.ClientConfigs {
// Check done channel status
select {
case workerInput <- i:
case <-r.done:
// process done event
// Stop workers, clear channel and wait for all workers to finish
close(workerInput)
for range workerInput {
}
workerWaitGroup.Wait()
return
}
}
// Stop workers and wait for all workers to finish
close(workerInput)
workerWaitGroup.Wait()
}
// Start redfish receiver
r.wg.Add(1)
go func() {
defer r.wg.Done()
// Create ticker
ticker := time.NewTicker(time.Duration(r.config.Interval) * time.Second)
defer ticker.Stop()
for {
doReadPowerMetric()
select {
case <-ticker.C:
// process ticker event -> continue
continue
case <-r.done:
// process done event
return
}
}
}()
cclog.ComponentDebug(r.name, "STARTED")
}
// Close redfish receiver
func (r *RedfishReceiver) Close() {
cclog.ComponentDebug(r.name, "CLOSE")
// Send the signal and wait
close(r.done)
r.wg.Wait()
cclog.ComponentDebug(r.name, "DONE")
}
// New function to create a new instance of the receiver
// Initialize the receiver by giving it a name and reading in the config JSON
func NewRedfishReceiver(name string, config json.RawMessage) (Receiver, error) {
r := new(RedfishReceiver)
// Set name
r.name = fmt.Sprintf("RedfishReceiver(%s)", name)
// Create done channel
r.done = make(chan bool)
// Set defaults in r.config
// Allow overwriting these defaults by reading config JSON
r.config.Fanout = 64
r.config.Interval = 30
// Read the redfish receiver specific JSON config
if len(config) > 0 {
err := json.Unmarshal(config, &r.config)
if err != nil {
cclog.ComponentError(r.name, "Error reading config:", err.Error())
return nil, err
}
}
// Create gofish client config
for i := range r.config.ClientConfigs {
clientConfig := &r.config.ClientConfigs[i]
gofishConfig := &clientConfig.gofish
if clientConfig.Hostname == nil {
err := fmt.Errorf("client config number %v requires hostname", i)
cclog.ComponentError(r.name, err)
return nil, err
}
if clientConfig.Endpoint == nil {
err := fmt.Errorf("client config number %v requires endpoint", i)
cclog.ComponentError(r.name, err)
return nil, err
}
gofishConfig.Endpoint = *clientConfig.Endpoint
if clientConfig.Username == nil {
err := fmt.Errorf("client config number %v requires username", i)
cclog.ComponentError(r.name, err)
return nil, err
}
gofishConfig.Username = *clientConfig.Username
if clientConfig.Password == nil {
err := fmt.Errorf("client config number %v requires password", i)
cclog.ComponentError(r.name, err)
return nil, err
}
gofishConfig.Password = *clientConfig.Password
gofishConfig.Insecure = true
if clientConfig.Insecure != nil {
gofishConfig.Insecure = *clientConfig.Insecure
}
}
return r, nil
}

26
receivers/sampleReceiver.go
View File

@ -36,16 +36,26 @@ func (r *SampleReceiver) Start() {
// or use own go routine but always make sure it exits
// as soon as it gets the signal of the r.done channel
//
// r.done = make(chan bool)
// r.wg.Add(1)
// go func() {
// for {
// select {
// case <-r.done:
// r.wg.Done()
// return
// }
// }
// r.wg.Done()
// defer r.wg.Done()
//
// // Create ticker
// ticker := time.NewTicker(30 * time.Second)
// defer ticker.Stop()
//
// for {
// readMetric()
// select {
// case <-ticker.C:
// // process ticker event -> continue
// continue
// case <-r.done:
// return
// }
// }
// }()
}

6
scripts/cc-metric-collector.config
View File

@ -15,3 +15,9 @@ CONF_DIR=/etc/cc-metric-collector
CONF_FILE=/etc/cc-metric-collector/cc-metric-collector.json
RESTART_ON_UPGRADE=true
# Golang runtime debugging. (see: https://pkg.go.dev/runtime)
# GODEBUG=gctrace=1
# Golang garbage collection target percentage
# GOGC=100

8
sinks.json
View File

@ -1,6 +1,8 @@
{
"mystdout" : {
"type" : "stdout",
"meta_as_tags" : true
"mystdout": {
"type": "stdout",
"meta_as_tags": [
"unit"
]
}
}

74
sinks/influxSink.go
View File

@ -49,20 +49,31 @@ type InfluxSink struct {
//influxMaxRetryDelay uint
}
// connect connects to the InfluxDB server
func (s *InfluxSink) connect() error {
var auth string
// URI options:
// * http://host:port
// * https://host:port
var uri string
if s.config.SSL {
uri = fmt.Sprintf("https://%s:%s", s.config.Host, s.config.Port)
} else {
uri = fmt.Sprintf("http://%s:%s", s.config.Host, s.config.Port)
}
// Authentication options:
// * token
// * username:password
var auth string
if len(s.config.User) == 0 {
auth = s.config.Password
} else {
auth = fmt.Sprintf("%s:%s", s.config.User, s.config.Password)
}
cclog.ComponentDebug(s.name, "Using URI", uri, "Org", s.config.Organization, "Bucket", s.config.Database)
// Set influxDB client options
clientOptions := influxdb2.DefaultOptions()
// if s.influxRetryInterval != 0 {
@ -82,6 +93,7 @@ func (s *InfluxSink) connect() error {
// clientOptions.SetMaxRetries(s.config.InfluxMaxRetries)
// }
// Do not check InfluxDB certificate
clientOptions.SetTLSConfig(
&tls.Config{
InsecureSkipVerify: true,
@ -90,8 +102,11 @@ func (s *InfluxSink) connect() error {
clientOptions.SetPrecision(time.Second)
// Create new writeAPI
s.client = influxdb2.NewClientWithOptions(uri, auth, clientOptions)
s.writeApi = s.client.WriteAPIBlocking(s.config.Organization, s.config.Database)
// Check InfluxDB server accessibility
ok, err := s.client.Ping(context.Background())
if err != nil {
return err
@ -103,24 +118,22 @@ func (s *InfluxSink) connect() error {
}
func (s *InfluxSink) Write(m lp.CCMetric) error {
// err :=
// s.writeApi.WritePoint(
// context.Background(),
// m.ToPoint(s.meta_as_tags),
// )
if len(s.batch) == 0 && s.flushDelay != 0 {
// This is the first write since the last flush, start the flushTimer!
if s.flushTimer != nil && s.flushTimer.Stop() {
cclog.ComponentDebug(s.name, "unexpected: the flushTimer was already running?")
}
// Run a batched flush for all lines that have arrived in the last second
// Run a batched flush for all lines that have arrived in the last flush delay interval
s.flushTimer = time.AfterFunc(s.flushDelay, func() {
if err := s.Flush(); err != nil {
cclog.ComponentError(s.name, "flush failed:", err.Error())
}
})
}
// Append metric to batch slice
p := m.ToPoint(s.meta_as_tags)
s.lock.Lock()
s.batch = append(s.batch, p)
@ -131,21 +144,39 @@ func (s *InfluxSink) Write(m lp.CCMetric) error {
return s.Flush()
}
// Flush if batch size is reached
if len(s.batch) == s.config.BatchSize {
return s.Flush()
}
return nil
}
// Flush sends all metrics buffered in batch slice to InfluxDB server
func (s *InfluxSink) Flush() error {
// Lock access to batch slice
s.lock.Lock()
defer s.lock.Unlock()
// Nothing to do, batch slice is empty
if len(s.batch) == 0 {
return nil
}
// Send metrics from batch slice
err := s.writeApi.WritePoint(context.Background(), s.batch...)
if err != nil {
cclog.ComponentError(s.name, "flush failed:", err.Error())
return err
}
// Clear batch slice
for i := range s.batch {
s.batch[i] = nil
}
s.batch = s.batch[:0]
return nil
}
@ -156,11 +187,16 @@ func (s *InfluxSink) Close() {
s.client.Close()
}
// NewInfluxSink create a new InfluxDB sink
func NewInfluxSink(name string, config json.RawMessage) (Sink, error) {
s := new(InfluxSink)
s.name = fmt.Sprintf("InfluxSink(%s)", name)
// Set config default values
s.config.BatchSize = 100
s.config.FlushDelay = "1s"
// Read config
if len(config) > 0 {
err := json.Unmarshal(config, &s.config)
if err != nil {
@ -174,13 +210,22 @@ func NewInfluxSink(name string, config json.RawMessage) (Sink, error) {
// s.config.InfluxMaxRetries = 0
// s.config.InfluxExponentialBase = 0
if len(s.config.Host) == 0 ||
len(s.config.Port) == 0 ||
len(s.config.Database) == 0 ||
len(s.config.Organization) == 0 ||
len(s.config.Password) == 0 {
return nil, errors.New("not all configuration variables set required by InfluxSink")
if len(s.config.Host) == 0 {
return nil, errors.New("Missing host configuration required by InfluxSink")
}
if len(s.config.Port) == 0 {
return nil, errors.New("Missing port configuration required by InfluxSink")
}
if len(s.config.Database) == 0 {
return nil, errors.New("Missing database configuration required by InfluxSink")
}
if len(s.config.Organization) == 0 {
return nil, errors.New("Missing organization configuration required by InfluxSink")
}
if len(s.config.Password) == 0 {
return nil, errors.New("Missing password configuration required by InfluxSink")
}
// 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 {
@ -199,12 +244,15 @@ func NewInfluxSink(name string, config json.RawMessage) (Sink, error) {
// s.influxRetryInterval = toUint(s.config.InfluxRetryInterval, s.influxRetryInterval)
// s.influxMaxRetryTime = toUint(s.config.InfluxMaxRetryTime, s.influxMaxRetryTime)
// Configure flush delay duration
if len(s.config.FlushDelay) > 0 {
t, err := time.ParseDuration(s.config.FlushDelay)
if err == nil {
s.flushDelay = t
}
}
// allocate batch slice
s.batch = make([]*write.Point, 0, s.config.BatchSize)
// Connect to InfluxDB server