ClusterCockpit/cc-metric-collector
1
0
Fork 0
mirror of https://github.com/ClusterCockpit/cc-metric-collector.git synced 2024-11-14 05:57:25 +01:00
Code Issues Packages Projects Releases Wiki Activity

Add some comments

Browse Source
This commit is contained in:
Holger Obermaier 2023-09-18 15:37:52 +02:00
parent 2d41531b51
commit fd227ed8b3
1 changed files with 19 additions and 5 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

24
receivers/natsReceiver.go
View File

@ -32,11 +32,14 @@ var DefaultTime = func() time.Time {
return time.Unix(42, 0) return time.Unix(42, 0)
} }
// Start subscribes to the configured NATS subject
// Messages wil be handled by r._NatsReceive
func (r *NatsReceiver) Start() { func (r *NatsReceiver) Start() {
cclog.ComponentDebug(r.name, "START") cclog.ComponentDebug(r.name, "START")
r.nc.Subscribe(r.config.Subject, r._NatsReceive) r.nc.Subscribe(r.config.Subject, r._NatsReceive)
} }
// _NatsReceive receives subscribed messages from the NATS server
func (r *NatsReceiver) _NatsReceive(m *nats.Msg) { func (r *NatsReceiver) _NatsReceive(m *nats.Msg) {
metrics, err := r.parser.Parse(m.Data) metrics, err := r.parser.Parse(m.Data)
if err == nil { if err == nil {
@ -52,6 +55,7 @@ func (r *NatsReceiver) _NatsReceive(m *nats.Msg) {
} }
} }
// Close closes the connection to the NATS server
func (r *NatsReceiver) Close() { func (r *NatsReceiver) Close() {
if r.nc != nil { if r.nc != nil {
cclog.ComponentDebug(r.name, "CLOSE") cclog.ComponentDebug(r.name, "CLOSE")
@ -59,10 +63,13 @@ func (r *NatsReceiver) Close() {
} }
} }
// NewNatsReceiver creates a new Receiver which subscribes to messages from a NATS server
func NewNatsReceiver(name string, config json.RawMessage) (Receiver, error) { func NewNatsReceiver(name string, config json.RawMessage) (Receiver, error) {
r := new(NatsReceiver) r := new(NatsReceiver)
r.name = fmt.Sprintf("NatsReceiver(%s)", name) r.name = fmt.Sprintf("NatsReceiver(%s)", name)
r.config.Addr = nats.DefaultURL
// Read configuration file, allow overwriting default config
r.config.Addr = "localhost"
r.config.Port = "4222" r.config.Port = "4222"
if len(config) > 0 { if len(config) > 0 {
err := json.Unmarshal(config, &r.config) err := json.Unmarshal(config, &r.config)
@ -76,15 +83,22 @@ func NewNatsReceiver(name string, config json.RawMessage) (Receiver, error) {
len(r.config.Subject) == 0 { len(r.config.Subject) == 0 {
return nil, errors.New("not all configuration variables set required by NatsReceiver") return nil, errors.New("not all configuration variables set required by NatsReceiver")
} }
r.meta = map[string]string{"source": r.name}
uri := fmt.Sprintf("%s:%s", r.config.Addr, r.config.Port) // Set metadata
cclog.ComponentDebug(r.name, "NewNatsReceiver", uri, "Subject", r.config.Subject) r.meta = map[string]string{
if nc, err := nats.Connect(uri); err == nil { "source": r.name,
}
// Connect to NATS server
url := fmt.Sprintf("nats://%s:%s", r.config.Addr, r.config.Port)
cclog.ComponentDebug(r.name, "NewNatsReceiver", url, "Subject", r.config.Subject)
if nc, err := nats.Connect(url); err == nil {
r.nc = nc r.nc = nc
} else { } else {
r.nc = nil r.nc = nil
return nil, err return nil, err
} }
r.handler = influx.NewMetricHandler() r.handler = influx.NewMetricHandler()
r.parser = influx.NewParser(r.handler) r.parser = influx.NewParser(r.handler)
r.parser.SetTimeFunc(DefaultTime) r.parser.SetTimeFunc(DefaultTime)