mirror of
https://github.com/ClusterCockpit/cc-metric-collector.git
synced 2026-02-13 14:41:45 +01:00
Compare commits
2 Commits
cleanup
...
replace_de
| Author | SHA1 | Date | |
|---|---|---|---|
|
|
7aa0e31562 | ||
|
|
9908d76aac |
@@ -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),
|
||||||
|
)
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|||||||
Reference in New Issue
Block a user