Compare commits

..

2 Commits

Author SHA1 Message Date
Holger Obermaier
7aa0e31562 Replace depreceated functions 2026-02-13 11:49:37 +01:00
Holger Obermaier
9908d76aac Cleanup (#197)
* Removed unused code
* Use cclog for logging
* Wrap errors so that they can be unwrapped
* Revert wrong use of slices.Delete()
* Fix derivative values should be float
* Suggestions from the gocritic linter
* Fixed: interface method AddChannel must have all named params (inamedparam)
* Enable linter: errorlint
* Replace fmt.Sprintf("%d", i)) by strconv.Itoa(i) for improved performance
* Correct misspelled words
* Break up very long lines into multiple lines
* lp.NewMessage -> lp.NewMetric
* Preallocate slices of known length
2026-02-13 09:36:14 +01:00

View File

@@ -8,10 +8,11 @@
package collectors package collectors
import ( import (
"bytes"
"encoding/json" "encoding/json"
"errors" "errors"
"fmt" "fmt"
"log" "io"
"os" "os"
"os/exec" "os/exec"
"slices" "slices"
@@ -20,7 +21,9 @@ import (
cclog "github.com/ClusterCockpit/cc-lib/v2/ccLogger" cclog "github.com/ClusterCockpit/cc-lib/v2/ccLogger"
lp "github.com/ClusterCockpit/cc-lib/v2/ccMessage" lp "github.com/ClusterCockpit/cc-lib/v2/ccMessage"
influx "github.com/influxdata/line-protocol"
receivers "github.com/ClusterCockpit/cc-lib/v2/receivers"
lp2 "github.com/influxdata/line-protocol/v2/lineprotocol"
) )
const CUSTOMCMDPATH = `/home/unrz139/Work/cc-metric-collector/collectors/custom` const CUSTOMCMDPATH = `/home/unrz139/Work/cc-metric-collector/collectors/custom`
@@ -33,8 +36,6 @@ type CustomCmdCollectorConfig struct {
type CustomCmdCollector struct { type CustomCmdCollector struct {
metricCollector metricCollector
handler *influx.MetricHandler
parser *influx.Parser
config CustomCmdCollectorConfig config CustomCmdCollectorConfig
commands []string commands []string
files []string files []string
@@ -86,9 +87,6 @@ func (m *CustomCmdCollector) Init(config json.RawMessage) error {
if len(m.files) == 0 && len(m.commands) == 0 { if len(m.files) == 0 && len(m.commands) == 0 {
return errors.New("no metrics to collect") return errors.New("no metrics to collect")
} }
m.handler = influx.NewMetricHandler()
m.parser = influx.NewParser(m.handler)
m.parser.SetTimeFunc(DefaultTime)
m.init = true m.init = true
return nil return nil
} }
@@ -102,46 +100,83 @@ func (m *CustomCmdCollector) Read(interval time.Duration, output chan lp.CCMessa
return return
} }
for _, cmd := range m.commands { for _, cmd := range m.commands {
cmdfields := strings.Fields(cmd) // Execute configured commands
command := exec.Command(cmdfields[0], cmdfields[1:]...) cmdFields := strings.Fields(cmd)
if err := command.Wait(); err != nil { command := exec.Command(cmdFields[0], cmdFields[1:]...)
log.Print(err) stdout, _ := command.StdoutPipe()
continue errBuf := new(bytes.Buffer)
} command.Stderr = errBuf
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) // Start command
} if err := command.Start(); err != nil {
} cclog.ComponentError(
for _, file := range m.files { m.name,
buffer, err := os.ReadFile(file) fmt.Sprintf("Read(): Failed to start command \"%s\": %v", command.String(), err),
if err != nil { )
log.Print(err)
return return
} }
fmetrics, err := m.parser.Parse(buffer)
// Read and decode influxDB line-protocol from command output
d := lp2.NewDecoder(stdout)
for d.Next() {
metric, err := receivers.DecodeInfluxMessage(d)
if err != nil { if err != nil {
log.Print(err) cclog.ComponentError(
m.name,
fmt.Sprintf("Read(): Failed to decode influx Message: %v", err),
)
continue continue
} }
for _, f := range fmetrics { if slices.Contains(m.config.ExcludeMetrics, metric.Name()) {
if slices.Contains(m.config.ExcludeMetrics, f.Name()) {
continue continue
} }
output <- lp.FromInfluxMetric(f) output <- metric
}
// Wait for command end
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
}
}
for _, filename := range m.files {
file, err := os.Open(filename)
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),
)
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),
)
} }
} }
} }