Compare commits

..

11 Commits

Author SHA1 Message Date
Holger Obermaier
d715f7aa07 Preallocate slices of known length 2026-02-12 15:07:39 +01:00
Holger Obermaier
555ba9504a * Replace fmt.Sprintf("%d", i)) by strconv.Itoa(i)
* Correct misspelled words
* Remove unused code
* Break up very long lines into multiple lines
* lp.NewMessage -> lp.NewMetric
2026-02-12 14:31:01 +01:00
Holger Obermaier
309bc32a24 Enable linter: errorlint 2026-02-11 15:59:01 +01:00
Holger Obermaier
0b1f88b8a6 Fixed: interface method AddChannel must have all named params (inamedparam) 2026-02-11 15:55:14 +01:00
Holger Obermaier
3181f81db1 Suggestions from the gocritic linter 2026-02-11 14:56:21 +01:00
Holger Obermaier
18e2518660 Fix derivative values should be float 2026-02-11 14:27:08 +01:00
Holger Obermaier
2cca8d6ac0 Revert wrong use of slices.Delete() 2026-02-11 14:02:47 +01:00
Holger Obermaier
6bfdd6ff17 Wrap errors so that they can be unwrapped 2026-02-11 13:51:58 +01:00
Holger Obermaier
ca95494a83 Use cclog for logging 2026-02-11 12:15:58 +01:00
Holger Obermaier
e512f3255c Use cclog for logging 2026-02-11 12:01:47 +01:00
Holger Obermaier
5b08183d54 Removed unused code 2026-02-11 11:45:08 +01:00

View File

@@ -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))))
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 _, file := range m.files {
buffer, err := os.ReadFile(file)
if err != nil {
log.Print(err)
return
}
}
for _, filename := range m.files {
file, err := os.Open(filename)
fmetrics, err := m.parser.Parse(buffer)
if err != nil {
cclog.ComponentError(
m.name,
fmt.Sprintf("Read(): Failed to open file \"%s\": %v\n", filename, err),
)
}
// 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),
)
log.Print(err)
continue
}
if slices.Contains(m.config.ExcludeMetrics, metric.Name()) {
for _, f := range fmetrics {
if slices.Contains(m.config.ExcludeMetrics, f.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)
}
}
}