mirror of
https://github.com/ClusterCockpit/cc-metric-collector.git
synced 2026-02-13 14:41:45 +01:00
Compare commits
11 Commits
replace_de
...
cleanup
| Author | SHA1 | Date | |
|---|---|---|---|
|
|
d715f7aa07 | ||
|
|
555ba9504a | ||
|
|
309bc32a24 | ||
|
|
0b1f88b8a6 | ||
|
|
3181f81db1 | ||
|
|
18e2518660 | ||
|
|
2cca8d6ac0 | ||
|
|
6bfdd6ff17 | ||
|
|
ca95494a83 | ||
|
|
e512f3255c | ||
|
|
5b08183d54 |
@@ -8,11 +8,10 @@
|
||||
package collectors
|
||||
|
||||
import (
|
||||
"bytes"
|
||||
"encoding/json"
|
||||
"errors"
|
||||
"fmt"
|
||||
"io"
|
||||
"log"
|
||||
"os"
|
||||
"os/exec"
|
||||
"slices"
|
||||
@@ -21,9 +20,7 @@ import (
|
||||
|
||||
cclog "github.com/ClusterCockpit/cc-lib/v2/ccLogger"
|
||||
lp "github.com/ClusterCockpit/cc-lib/v2/ccMessage"
|
||||
|
||||
receivers "github.com/ClusterCockpit/cc-lib/v2/receivers"
|
||||
lp2 "github.com/influxdata/line-protocol/v2/lineprotocol"
|
||||
influx "github.com/influxdata/line-protocol"
|
||||
)
|
||||
|
||||
const CUSTOMCMDPATH = `/home/unrz139/Work/cc-metric-collector/collectors/custom`
|
||||
@@ -36,6 +33,8 @@ type CustomCmdCollectorConfig struct {
|
||||
|
||||
type CustomCmdCollector struct {
|
||||
metricCollector
|
||||
handler *influx.MetricHandler
|
||||
parser *influx.Parser
|
||||
config CustomCmdCollectorConfig
|
||||
commands []string
|
||||
files []string
|
||||
@@ -87,6 +86,9 @@ func (m *CustomCmdCollector) Init(config json.RawMessage) error {
|
||||
if len(m.files) == 0 && len(m.commands) == 0 {
|
||||
return errors.New("no metrics to collect")
|
||||
}
|
||||
m.handler = influx.NewMetricHandler()
|
||||
m.parser = influx.NewParser(m.handler)
|
||||
m.parser.SetTimeFunc(DefaultTime)
|
||||
m.init = true
|
||||
return nil
|
||||
}
|
||||
@@ -100,83 +102,46 @@ func (m *CustomCmdCollector) Read(interval time.Duration, output chan lp.CCMessa
|
||||
return
|
||||
}
|
||||
for _, cmd := range m.commands {
|
||||
// Execute configured commands
|
||||
cmdFields := strings.Fields(cmd)
|
||||
command := exec.Command(cmdFields[0], cmdFields[1:]...)
|
||||
stdout, _ := command.StdoutPipe()
|
||||
errBuf := new(bytes.Buffer)
|
||||
command.Stderr = errBuf
|
||||
|
||||
// Start command
|
||||
if err := command.Start(); err != nil {
|
||||
cclog.ComponentError(
|
||||
m.name,
|
||||
fmt.Sprintf("Read(): Failed to start command \"%s\": %v", command.String(), err),
|
||||
)
|
||||
return
|
||||
}
|
||||
|
||||
// Read and decode influxDB line-protocol from command output
|
||||
d := lp2.NewDecoder(stdout)
|
||||
for d.Next() {
|
||||
metric, err := receivers.DecodeInfluxMessage(d)
|
||||
if err != nil {
|
||||
cclog.ComponentError(
|
||||
m.name,
|
||||
fmt.Sprintf("Read(): Failed to decode influx Message: %v", err),
|
||||
)
|
||||
continue
|
||||
}
|
||||
if slices.Contains(m.config.ExcludeMetrics, metric.Name()) {
|
||||
continue
|
||||
}
|
||||
output <- metric
|
||||
}
|
||||
|
||||
// Wait for command end
|
||||
cmdfields := strings.Fields(cmd)
|
||||
command := exec.Command(cmdfields[0], cmdfields[1:]...)
|
||||
if err := command.Wait(); err != nil {
|
||||
errMsg, _ := io.ReadAll(errBuf)
|
||||
cclog.ComponentError(
|
||||
m.name,
|
||||
fmt.Sprintf("Read(): Failed to wait for the end of command \"%s\": %v\n", command.String(), err),
|
||||
)
|
||||
cclog.ComponentError(
|
||||
m.name,
|
||||
fmt.Sprintf("Read(): command stderr: \"%s\"\n", strings.TrimSpace(string(errMsg))))
|
||||
return
|
||||
log.Print(err)
|
||||
continue
|
||||
}
|
||||
stdout, err := command.Output()
|
||||
if err != nil {
|
||||
log.Print(err)
|
||||
continue
|
||||
}
|
||||
cmdmetrics, err := m.parser.Parse(stdout)
|
||||
if err != nil {
|
||||
log.Print(err)
|
||||
continue
|
||||
}
|
||||
for _, c := range cmdmetrics {
|
||||
if slices.Contains(m.config.ExcludeMetrics, c.Name()) {
|
||||
continue
|
||||
}
|
||||
|
||||
output <- lp.FromInfluxMetric(c)
|
||||
}
|
||||
}
|
||||
for _, filename := range m.files {
|
||||
file, err := os.Open(filename)
|
||||
for _, file := range m.files {
|
||||
buffer, err := os.ReadFile(file)
|
||||
if err != nil {
|
||||
cclog.ComponentError(
|
||||
m.name,
|
||||
fmt.Sprintf("Read(): Failed to open file \"%s\": %v\n", filename, err),
|
||||
)
|
||||
log.Print(err)
|
||||
return
|
||||
}
|
||||
|
||||
// Read and decode influxDB line-protocol from file
|
||||
d := lp2.NewDecoder(file)
|
||||
for d.Next() {
|
||||
metric, err := receivers.DecodeInfluxMessage(d)
|
||||
if err != nil {
|
||||
cclog.ComponentError(
|
||||
m.name,
|
||||
fmt.Sprintf("Read(): Failed to decode influx Message: %v", err),
|
||||
)
|
||||
fmetrics, err := m.parser.Parse(buffer)
|
||||
if err != nil {
|
||||
log.Print(err)
|
||||
continue
|
||||
}
|
||||
for _, f := range fmetrics {
|
||||
if slices.Contains(m.config.ExcludeMetrics, f.Name()) {
|
||||
continue
|
||||
}
|
||||
if slices.Contains(m.config.ExcludeMetrics, metric.Name()) {
|
||||
continue
|
||||
}
|
||||
output <- metric
|
||||
}
|
||||
|
||||
if err := file.Close(); err != nil {
|
||||
cclog.ComponentError(
|
||||
m.name,
|
||||
fmt.Sprintf("Read(): Failed to close file \"%s\": %v\n", filename, err),
|
||||
)
|
||||
output <- lp.FromInfluxMetric(f)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
Reference in New Issue
Block a user