ClusterCockpit/cc-metric-collector
1
0
Fork 0
mirror of https://github.com/ClusterCockpit/cc-metric-collector.git synced 2025-01-15 00:29:09 +01:00
Code Issues Packages Projects Releases Wiki Activity

Use CCMessage FromBytes instead of Influx's decoder

Browse Source
This commit is contained in:
Thomas Roehl 2024-12-27 15:22:59 +00:00
parent 7d3180b526
commit 7b343d0bab
2 changed files with 21 additions and 134 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

71
receivers/httpReceiver.go
View File

@ -13,7 +13,6 @@ import (
lp "github.com/ClusterCockpit/cc-energy-manager/pkg/cc-message"
cclog "github.com/ClusterCockpit/cc-metric-collector/pkg/ccLogger"
mp "github.com/ClusterCockpit/cc-metric-collector/pkg/messageProcessor"
influx "github.com/influxdata/line-protocol/v2/lineprotocol"
)
const HTTP_RECEIVER_PORT = "8080"
@ -151,80 +150,22 @@ func (r *HttpReceiver) ServerHttp(w http.ResponseWriter, req *http.Request) {
}
}
if r.sink != nil {
d := influx.NewDecoder(req.Body)
for d.Next() {
// Decode measurement name
measurement, err := d.Measurement()
buf := make([]byte, 0, req.ContentLength)
len, err := req.Body.Read(buf)
if err == nil && len > 0 {
messages, err := lp.FromBytes(buf)
if err != nil {
msg := "ServerHttp: Failed to decode measurement: " + err.Error()
msg := "ServerHttp: Failed to decode messages: " + err.Error()
cclog.ComponentError(r.name, msg)
http.Error(w, msg, http.StatusInternalServerError)
return
}
// Decode tags
tags := make(map[string]string)
for {
key, value, err := d.NextTag()
if err != nil {
msg := "ServerHttp: Failed to decode tag: " + err.Error()
cclog.ComponentError(r.name, msg)
http.Error(w, msg, http.StatusInternalServerError)
return
}
if key == nil {
break
}
tags[string(key)] = string(value)
}
// Decode fields
fields := make(map[string]interface{})
for {
key, value, err := d.NextField()
if err != nil {
msg := "ServerHttp: Failed to decode field: " + err.Error()
cclog.ComponentError(r.name, msg)
http.Error(w, msg, http.StatusInternalServerError)
return
}
if key == nil {
break
}
fields[string(key)] = value.Interface()
}
// Decode time stamp
t, err := d.Time(influx.Nanosecond, time.Time{})
if err != nil {
msg := "ServerHttp: Failed to decode time stamp: " + err.Error()
cclog.ComponentError(r.name, msg)
http.Error(w, msg, http.StatusInternalServerError)
return
}
y, _ := lp.NewMessage(
string(measurement),
tags,
nil,
fields,
t,
)
for _, y := range messages {
m, err := r.mp.ProcessMessage(y)
if err == nil && m != nil {
r.sink <- m
}
}
// Check for IO errors
err := d.Err()
if err != nil {
msg := "ServerHttp: Failed to decode: " + err.Error()
cclog.ComponentError(r.name, msg)
http.Error(w, msg, http.StatusInternalServerError)
return
}
}

60
receivers/natsReceiver.go
View File

@ -5,12 +5,10 @@ import (
"errors"
"fmt"
"os"
"time"
lp "github.com/ClusterCockpit/cc-energy-manager/pkg/cc-message"
cclog "github.com/ClusterCockpit/cc-metric-collector/pkg/ccLogger"
mp "github.com/ClusterCockpit/cc-metric-collector/pkg/messageProcessor"
influx "github.com/influxdata/line-protocol/v2/lineprotocol"
nats "github.com/nats-io/nats.go"
)
@ -42,63 +40,12 @@ func (r *NatsReceiver) Start() {
func (r *NatsReceiver) _NatsReceive(m *nats.Msg) {
if r.sink != nil {
d := influx.NewDecoderWithBytes(m.Data)
for d.Next() {
// Decode measurement name
measurement, err := d.Measurement()
messages, err := lp.FromBytes(m.Data)
if err != nil {
msg := "_NatsReceive: Failed to decode measurement: " + err.Error()
msg := "_NatsReceive: Failed to decode messages: " + err.Error()
cclog.ComponentError(r.name, msg)
return
}
// Decode tags
tags := make(map[string]string)
for {
key, value, err := d.NextTag()
if err != nil {
msg := "_NatsReceive: Failed to decode tag: " + err.Error()
cclog.ComponentError(r.name, msg)
return
}
if key == nil {
break
}
tags[string(key)] = string(value)
}
// Decode fields
fields := make(map[string]interface{})
for {
key, value, err := d.NextField()
if err != nil {
msg := "_NatsReceive: Failed to decode field: " + err.Error()
cclog.ComponentError(r.name, msg)
return
}
if key == nil {
break
}
fields[string(key)] = value.Interface()
}
// Decode time stamp
t, err := d.Time(influx.Nanosecond, time.Time{})
if err != nil {
msg := "_NatsReceive: Failed to decode time: " + err.Error()
cclog.ComponentError(r.name, msg)
return
}
y, err := lp.NewMessage(
string(measurement),
tags,
nil,
fields,
t,
)
if err == nil {
for _, y := range messages {
m, err := r.mp.ProcessMessage(y)
if err == nil && m != nil && r.sink != nil {
r.sink <- m
@ -106,7 +53,6 @@ func (r *NatsReceiver) _NatsReceive(m *nats.Msg) {
}
}
}
}
// Close closes the connection to the NATS server
func (r *NatsReceiver) Close() {