forked from influxdata/telegraf
-
Notifications
You must be signed in to change notification settings - Fork 0
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
feat(serializers): Add CloudEvents serializer (influxdata#13224)
- Loading branch information
Showing
28 changed files
with
2,082 additions
and
0 deletions.
There are no files selected for viewing
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,7 @@ | ||
| //go:build !custom || serializers || serializers.cloudevents | ||
|
|
||
| package all | ||
|
|
||
| import ( | ||
| _ "github.com/influxdata/telegraf/plugins/serializers/cloudevents" // register plugin | ||
| ) |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,64 @@ | ||
| # CloudEvents Serializer | ||
|
|
||
| The `cloudevents` data format outputs metrics as [CloudEvents][CloudEvents] in | ||
| [JSON format][JSON Spec]. Currently, versions v1.0 and v0.3 of the specification | ||
| are supported with the former being the default. | ||
|
|
||
| [CloudEvents]: https://cloudevents.io | ||
| [JSON Spec]: https://github.com/cloudevents/spec/blob/v1.0/json-format.md | ||
|
|
||
| ## Configuration | ||
|
|
||
| ```toml | ||
| [[outputs.file]] | ||
| ## Files to write to, "stdout" is a specially handled file | ||
| files = ["stdout", "/tmp/metrics.out"] | ||
|
|
||
| ## Data format to output | ||
| ## Each data format has its own unique set of configuration options, read | ||
| ## more about them here: | ||
| ## https://github.com/influxdata/telegraf/blob/master/docs/DATA_FORMATS_OUTPUT.md | ||
| data_format = "cloudevents" | ||
|
|
||
| ## Specification version to use for events | ||
| ## Currently versions "0.3" and "1.0" are supported. | ||
| # cloudevents_version = "1.0" | ||
|
|
||
| ## Event source specifier | ||
| ## This allows to overwrite the source header-field with the given value. | ||
| # cloudevents_source = "telegraf" | ||
|
|
||
| ## Tag to use as event source specifier | ||
| ## This allows to overwrite the source header-field with the value of the | ||
| ## specified tag. If both 'cloudevents_source' and 'cloudevents_source_tag' | ||
| ## are set, the this setting will take precedence. In case the specified tag | ||
| ## value does not exist for a metric, the serializer will fallback to | ||
| ## 'cloudevents_source'. | ||
| # cloudevents_source_tag = "" | ||
|
|
||
| ## Event-type specifier to overwrite the default value | ||
| ## By default, events (and event batches) containing a single metric will | ||
| ## set the event-type to 'com.influxdata.telegraf.metric' while events | ||
| ## containing a batch of metrics will set the event-type to | ||
| ## 'com.influxdata.telegraf.metric' (plural). | ||
| # cloudevents_event_type = "" | ||
|
|
||
| ## Set time header of the event | ||
| ## Supported values are: | ||
| ## none -- do not set event time | ||
| ## earliest -- use timestamp of the earliest metric | ||
| ## latest -- use timestamp of the latest metric | ||
| ## creation -- use timestamp of event creation | ||
| ## For events containing only a single metric, earliest and latest are | ||
| ## equivalent. | ||
| # cloudevents_event_time = "latest" | ||
|
|
||
| ## Batch format of the output when running in batch mode | ||
| ## If set to 'events' the resulting output will contain a list of events, | ||
| ## each with a single metric according to the JSON Batch Format of the | ||
| ## specification. Use 'application/cloudevents-batch+json' for this format. | ||
| ## | ||
| ## When set to 'metrics', a single event will be generated containing a list | ||
| ## of metrics as payload. Use 'application/cloudevents+json' for this format. | ||
| # cloudevents_batch_format = "events" | ||
| ``` |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,199 @@ | ||
| package cloudevents | ||
|
|
||
| import ( | ||
| "encoding/json" | ||
| "errors" | ||
| "fmt" | ||
| "time" | ||
|
|
||
| cloudevents "github.com/cloudevents/sdk-go/v2" | ||
| "github.com/cloudevents/sdk-go/v2/event" | ||
| "github.com/gofrs/uuid/v5" | ||
|
|
||
| "github.com/influxdata/telegraf" | ||
| "github.com/influxdata/telegraf/plugins/serializers" | ||
| ) | ||
|
|
||
| const ( | ||
| EventTypeSingle = "com.influxdata.telegraf.metric" | ||
| EventTypeBatch = "com.influxdata.telegraf.metrics" | ||
| ) | ||
|
|
||
| type Serializer struct { | ||
| Version string `toml:"cloudevents_version"` | ||
| Source string `toml:"cloudevents_source"` | ||
| SourceTag string `toml:"cloudevents_source_tag"` | ||
| EventType string `toml:"cloudevents_event_type"` | ||
| EventTime string `toml:"cloudevents_event_time"` | ||
| BatchFormat string `toml:"cloudevents_batch_format"` | ||
| Log telegraf.Logger `toml:"-"` | ||
|
|
||
| idgen uuid.Generator | ||
| } | ||
|
|
||
| func (s *Serializer) Init() error { | ||
| switch s.Version { | ||
| case "": | ||
| s.Version = event.CloudEventsVersionV1 | ||
| case event.CloudEventsVersionV03, event.CloudEventsVersionV1: | ||
| default: | ||
| return errors.New("invalid 'cloudevents_version'") | ||
| } | ||
|
|
||
| switch s.EventTime { | ||
| case "": | ||
| s.EventTime = "latest" | ||
| case "none", "earliest", "latest", "creation": | ||
| default: | ||
| return errors.New("invalid 'cloudevents_event_time'") | ||
| } | ||
|
|
||
| switch s.BatchFormat { | ||
| case "": | ||
| s.BatchFormat = "events" | ||
| case "metrics", "events": | ||
| default: | ||
| return errors.New("invalid 'cloudevents_batch_format'") | ||
| } | ||
|
|
||
| if s.Source == "" { | ||
| s.Source = "telegraf" | ||
| } | ||
|
|
||
| s.idgen = uuid.NewGen() | ||
|
|
||
| return nil | ||
| } | ||
|
|
||
| func (s *Serializer) Serialize(m telegraf.Metric) ([]byte, error) { | ||
| // Create the event that forms the envelop around the metric | ||
| evt, err := s.createEvent(m) | ||
| if err != nil { | ||
| return nil, err | ||
| } | ||
| return evt.MarshalJSON() | ||
| } | ||
|
|
||
| func (s *Serializer) SerializeBatch(metrics []telegraf.Metric) ([]byte, error) { | ||
| switch s.BatchFormat { | ||
| case "metrics": | ||
| return s.batchMetrics(metrics) | ||
| case "events": | ||
| return s.batchEvents(metrics) | ||
| } | ||
| return nil, fmt.Errorf("unexpected batch-format %q", s.BatchFormat) | ||
| } | ||
|
|
||
| func (s *Serializer) batchMetrics(metrics []telegraf.Metric) ([]byte, error) { | ||
| // Determine the necessary information | ||
| eventType := EventTypeBatch | ||
| if s.EventType != "" { | ||
| eventType = s.EventType | ||
| } | ||
| id, err := s.idgen.NewV1() | ||
| if err != nil { | ||
| return nil, fmt.Errorf("generating ID failed: %w", err) | ||
| } | ||
|
|
||
| // Serialize the metrics | ||
| var earliest, latest time.Time | ||
| data := make([]map[string]interface{}, 0, len(metrics)) | ||
| for _, m := range metrics { | ||
| ts := m.Time() | ||
| data = append(data, map[string]interface{}{ | ||
| "name": m.Name(), | ||
| "tags": m.Tags(), | ||
| "fields": m.Fields(), | ||
| "timestamp": ts.UnixNano(), | ||
| }) | ||
| if ts.Before(earliest) { | ||
| earliest = ts | ||
| } | ||
| if ts.After(latest) { | ||
| latest = ts | ||
| } | ||
| } | ||
|
|
||
| // Create the event that forms the envelop around the metric | ||
| evt := cloudevents.NewEvent(s.Version) | ||
| evt.SetSource(s.Source) | ||
| evt.SetID(id.String()) | ||
| evt.SetType(eventType) | ||
| if err := evt.SetData(cloudevents.ApplicationJSON, data); err != nil { | ||
| return nil, fmt.Errorf("setting data failed: %w", err) | ||
| } | ||
| switch s.EventTime { | ||
| case "creation": | ||
| evt.SetTime(time.Now()) | ||
| case "earliest": | ||
| evt.SetTime(earliest) | ||
| case "latest": | ||
| evt.SetTime(latest) | ||
| } | ||
|
|
||
| return json.Marshal(evt) | ||
| } | ||
|
|
||
| func (s *Serializer) batchEvents(metrics []telegraf.Metric) ([]byte, error) { | ||
| events := make([]*cloudevents.Event, 0, len(metrics)) | ||
| for _, m := range metrics { | ||
| e, err := s.createEvent(m) | ||
| if err != nil { | ||
| s.Log.Errorf("creating event for %v failed: %w", m, err) | ||
| continue | ||
| } | ||
| events = append(events, e) | ||
| } | ||
| return json.Marshal(events) | ||
| } | ||
|
|
||
| func (s *Serializer) createEvent(m telegraf.Metric) (*cloudevents.Event, error) { | ||
| // Determine the necessary information | ||
| source := s.Source | ||
| if s.SourceTag != "" { | ||
| if v, ok := m.GetTag(s.SourceTag); ok { | ||
| source = v | ||
| } | ||
| } | ||
| eventType := EventTypeSingle | ||
| if s.EventType != "" { | ||
| eventType = s.EventType | ||
| } | ||
| id, err := s.idgen.NewV1() | ||
| if err != nil { | ||
| return nil, fmt.Errorf("generating ID failed: %w", err) | ||
| } | ||
|
|
||
| // Serialize the metric | ||
| data := map[string]interface{}{ | ||
| "name": m.Name(), | ||
| "tags": m.Tags(), | ||
| "fields": m.Fields(), | ||
| "timestamp": m.Time().UnixNano(), | ||
| } | ||
|
|
||
| // Create the event that forms the envelop around the metric | ||
| evt := cloudevents.NewEvent(s.Version) | ||
| evt.SetSource(source) | ||
| evt.SetID(id.String()) | ||
| evt.SetType(eventType) | ||
| if err := evt.SetData(cloudevents.ApplicationJSON, data); err != nil { | ||
| return nil, fmt.Errorf("setting data failed: %w", err) | ||
| } | ||
| switch s.EventTime { | ||
| case "creation": | ||
| evt.SetTime(time.Now()) | ||
| case "earliest", "latest": | ||
| evt.SetTime(m.Time()) | ||
| } | ||
|
|
||
| return &evt, nil | ||
| } | ||
|
|
||
| func init() { | ||
| serializers.Add("cloudevents", | ||
| func() serializers.Serializer { | ||
| return &Serializer{} | ||
| }, | ||
| ) | ||
| } |
Oops, something went wrong.