From 034f76df9e006511c23e62e4372a1d7d2cb6213d Mon Sep 17 00:00:00 2001 From: Paul Lorenz Date: Tue, 23 Feb 2021 13:24:21 -0500 Subject: [PATCH] Split metrics event to be per metric (meter/histogram, etc level). Extract ids for links, control channels --- events/formatter.go | 39 +++++---- events/metrics.go | 176 +++++++++++++++++++++++++++++------------ events/metrics_test.go | 35 ++++++++ 3 files changed, 186 insertions(+), 64 deletions(-) create mode 100644 events/metrics_test.go diff --git a/events/formatter.go b/events/formatter.go index 8597c9377..b7b8e0869 100644 --- a/events/formatter.go +++ b/events/formatter.go @@ -2,10 +2,12 @@ package events import ( "encoding/json" + "github.com/golang/protobuf/ptypes" "github.com/michaelquigley/pfxlog" "github.com/openziti/foundation/util/iomonad" "io" "reflect" + "time" ) type LoggingHandlerFactory interface { @@ -50,17 +52,26 @@ type JsonMetricsEvent MetricsEvent func (event *JsonMetricsEvent) WriteTo(output io.WriteCloser) error { jsonRep := map[string]interface{}{} jsonRep["namespace"] = event.Namespace - jsonRep["source_id"] = event.SourceId - jsonRep["timestamp"] = event.Timestamp - jsonRep["tags"] = event.Tags + jsonRep["metric"] = event.Metric + jsonRep["source_id"] = event.SourceAppId + jsonRep["source_event_id"] = event.SourceEventId + if event.SourceEntityId != "" { + jsonRep["source_entity_id"] = event.SourceEntityId + } + + ts, err := ptypes.Timestamp(event.Timestamp) + if err != nil { + return err + } + + jsonRep["timestamp"] = ts.Format(time.RFC3339Nano) + if len(event.Tags) > 0 { + jsonRep["tags"] = event.Tags + } metrics := map[string]interface{}{} - for name, val := range event.IntMetrics { - metrics[name] = val - } - - for name, val := range event.FloatMetrics { + for name, val := range event.Metrics { metrics[name] = val } @@ -136,12 +147,12 @@ type PlainTextMetricsEvent MetricsEvent func (event *PlainTextMetricsEvent) WriteTo(output io.WriteCloser) error { w := iomonad.Wrap(output) - for name, val := range event.IntMetrics { - w.Printf("%v: %9d\n", name, val) - } - - for name, val := range event.FloatMetrics { - w.Printf("%s: %v\n", name, val) + for name, val := range event.Metrics { + if intVal, ok := val.(int64); ok { + w.Printf("%v: %9d\n", name, intVal) + } else { + w.Printf("%s: %v\n", name, val) + } } return w.GetError() diff --git a/events/metrics.go b/events/metrics.go index e22fb4f18..47d1ff925 100644 --- a/events/metrics.go +++ b/events/metrics.go @@ -2,13 +2,42 @@ package events import ( "github.com/golang/protobuf/ptypes/timestamp" + "github.com/google/uuid" "github.com/openziti/foundation/events" "github.com/openziti/foundation/metrics/metrics_pb" "github.com/pkg/errors" "reflect" "regexp" + "strings" ) +func init() { + AddMetricsNameMapper(mapLinkIds) + AddMetricsNameMapper(mapCtrlIds) +} + +func mapLinkIds(name string) (string, string, bool) { + if strings.HasPrefix(name, "link.") { + return ExtractId(name, "link.", 2) + } + return "", "", false +} + +func mapCtrlIds(name string) (string, string, bool) { + if strings.HasPrefix(name, "ctrl.") { + return ExtractId(name, "ctrl.", 2) + } + return "", "", false +} + +type MetricsNameMapper func(name string) (string, string, bool) + +var metricsNameMappers []MetricsNameMapper + +func AddMetricsNameMapper(mapper MetricsNameMapper) { + metricsNameMappers = append(metricsNameMappers, mapper) +} + func registerMetricsEventHandler(val interface{}, config map[interface{}]interface{}) error { handler, ok := val.(MetricsEventHandler) if !ok { @@ -62,72 +91,110 @@ type metricsAdapter struct { handler MetricsEventHandler } +func (adapter *metricsAdapter) newMetricEvent(msg *metrics_pb.MetricsMessage, name string, id string) *MetricsEvent { + result := &MetricsEvent{ + Namespace: "metrics", + SourceAppId: msg.SourceId, + Timestamp: msg.Timestamp, + Metric: name, + Tags: msg.Tags, + SourceEventId: id, + } + + for _, mapper := range metricsNameMappers { + if mappedName, entityId, mapped := mapper(name); mapped { + result.Metric = mappedName + result.SourceEntityId = entityId + break + } + } + + return result +} + +func (adapter *metricsAdapter) finishEvent(event *MetricsEvent) { + if len(event.Metrics) > 0 { + adapter.handler.AcceptMetricsEvent(event) + } +} + func (adapter *metricsAdapter) AcceptMetrics(msg *metrics_pb.MetricsMessage) { if adapter.sourceFilter != nil && !adapter.sourceFilter.Match([]byte(msg.SourceId)) { return } - event := &MetricsEvent{ - Namespace: "metrics", - SourceId: msg.SourceId, - Timestamp: msg.Timestamp, - Tags: msg.Tags, - MetricGroup: map[string]string{}, - } + parentEventId := uuid.NewString() for name, value := range msg.IntValues { - adapter.filterIntMetric(name, value, event, name) + event := adapter.newMetricEvent(msg, name, parentEventId) + adapter.filterMetric("", value, event) + adapter.finishEvent(event) } for name, value := range msg.FloatValues { - adapter.filterFloatMetric(name, value, event, name) + event := adapter.newMetricEvent(msg, name, parentEventId) + adapter.filterMetric("", value, event) + adapter.finishEvent(event) } for name, value := range msg.Meters { - adapter.filterIntMetric(name+".count", value.Count, event, name) - adapter.filterFloatMetric(name+".mean_rate", value.MeanRate, event, name) - adapter.filterFloatMetric(name+".m1_rate", value.M1Rate, event, name) - adapter.filterFloatMetric(name+".m5_rate", value.M5Rate, event, name) - adapter.filterFloatMetric(name+".m15_rate", value.M15Rate, event, name) + event := adapter.newMetricEvent(msg, name, parentEventId) + adapter.filterMetric(".count", value.Count, event) + adapter.filterMetric(".mean_rate", value.MeanRate, event) + adapter.filterMetric(".m1_rate", value.M1Rate, event) + adapter.filterMetric(".m5_rate", value.M5Rate, event) + adapter.filterMetric(".m15_rate", value.M15Rate, event) + adapter.finishEvent(event) } for name, value := range msg.Histograms { - adapter.filterIntMetric(name+".count", value.Count, event, name) - adapter.filterIntMetric(name+".min", value.Min, event, name) - adapter.filterIntMetric(name+".max", value.Max, event, name) - adapter.filterFloatMetric(name+".mean", value.Mean, event, name) - adapter.filterFloatMetric(name+".std_dev", value.StdDev, event, name) - adapter.filterFloatMetric(name+".variance", value.Variance, event, name) - adapter.filterFloatMetric(name+".p50", value.P50, event, name) - adapter.filterFloatMetric(name+".p75", value.P75, event, name) - adapter.filterFloatMetric(name+".p95", value.P95, event, name) - adapter.filterFloatMetric(name+".p99", value.P99, event, name) - adapter.filterFloatMetric(name+".p999", value.P999, event, name) - adapter.filterFloatMetric(name+".p9999", value.P9999, event, name) + event := adapter.newMetricEvent(msg, name, parentEventId) + adapter.filterMetric(".count", value.Count, event) + adapter.filterMetric(".min", value.Min, event) + adapter.filterMetric(".max", value.Max, event) + adapter.filterMetric(".mean", value.Mean, event) + adapter.filterMetric(".std_dev", value.StdDev, event) + adapter.filterMetric(".variance", value.Variance, event) + adapter.filterMetric(".p50", value.P50, event) + adapter.filterMetric(".p75", value.P75, event) + adapter.filterMetric(".p95", value.P95, event) + adapter.filterMetric(".p99", value.P99, event) + adapter.filterMetric(".p999", value.P999, event) + adapter.filterMetric(".p9999", value.P9999, event) + adapter.finishEvent(event) } - if len(event.IntMetrics) > 0 || len(event.FloatMetrics) > 0 { - adapter.handler.AcceptMetricsEvent(event) + for name, value := range msg.Timers { + event := adapter.newMetricEvent(msg, name, parentEventId) + adapter.filterMetric(".count", value.Count, event) + + adapter.filterMetric(".mean_rate", value.MeanRate, event) + adapter.filterMetric(".m1_rate", value.M1Rate, event) + adapter.filterMetric(".m5_rate", value.M5Rate, event) + adapter.filterMetric(".m15_rate", value.M15Rate, event) + + adapter.filterMetric(".min", value.Min, event) + adapter.filterMetric(".max", value.Max, event) + adapter.filterMetric(".mean", value.Mean, event) + adapter.filterMetric(".std_dev", value.StdDev, event) + adapter.filterMetric(".variance", value.Variance, event) + adapter.filterMetric(".p50", value.P50, event) + adapter.filterMetric(".p75", value.P75, event) + adapter.filterMetric(".p95", value.P95, event) + adapter.filterMetric(".p99", value.P99, event) + adapter.filterMetric(".p999", value.P999, event) + adapter.filterMetric(".p9999", value.P9999, event) + adapter.finishEvent(event) } } -func (adapter *metricsAdapter) filterIntMetric(name string, value int64, event *MetricsEvent, group string) { +func (adapter *metricsAdapter) filterMetric(suffix string, value interface{}, event *MetricsEvent) { + name := event.Metric + suffix if adapter.nameMatches(name) { - if event.IntMetrics == nil { - event.IntMetrics = make(map[string]int64) + if event.Metrics == nil { + event.Metrics = make(map[string]interface{}) } - event.IntMetrics[name] = value - event.MetricGroup[name] = group - } -} - -func (adapter *metricsAdapter) filterFloatMetric(name string, value float64, event *MetricsEvent, group string) { - if adapter.nameMatches(name) { - if event.FloatMetrics == nil { - event.FloatMetrics = make(map[string]float64) - } - event.FloatMetrics[name] = value - event.MetricGroup[name] = group + event.Metrics[name] = value } } @@ -136,15 +203,24 @@ func (adapter *metricsAdapter) nameMatches(name string) bool { } type MetricsEvent struct { - Namespace string - SourceId string - Timestamp *timestamp.Timestamp - Tags map[string]string - IntMetrics map[string]int64 - FloatMetrics map[string]float64 - MetricGroup map[string]string + Namespace string + SourceAppId string + SourceEntityId string + Timestamp *timestamp.Timestamp + Metric string + Metrics map[string]interface{} + Tags map[string]string + SourceEventId string } type MetricsEventHandler interface { AcceptMetricsEvent(event *MetricsEvent) } + +func ExtractId(name string, prefix string, suffixLen int) (string, string, bool) { + rest := strings.TrimPrefix(name, prefix) + vals := strings.Split(rest, ".") + idVals := vals[:len(vals)-suffixLen] + entityId := strings.Join(idVals, ".") + return prefix + rest[len(entityId)+1:], entityId, true +} diff --git a/events/metrics_test.go b/events/metrics_test.go new file mode 100644 index 000000000..41132b575 --- /dev/null +++ b/events/metrics_test.go @@ -0,0 +1,35 @@ +package events + +import ( + "github.com/stretchr/testify/require" + "testing" +) + +func Test_ExtractId(t *testing.T) { + name := "ctrl.3tOOkKfDn.tx.bytesrate" + + req := require.New(t) + name, entityId, _ := ExtractId(name, "ctrl.", 2) + req.Equal(name, "ctrl.tx.bytesrate") + req.Equal(entityId, "3tOOkKfDn") + + name = "ctrl.3tO.kKfDn.tx.bytesrate" + name, entityId, _ = ExtractId(name, "ctrl.", 2) + req.Equal(name, "ctrl.tx.bytesrate") + req.Equal(entityId, "3tO.kKfDn") + + name = "ctrl.3tO.kK.Dn.tx.bytesrate" + name, entityId, _ = ExtractId(name, "ctrl.", 2) + req.Equal(name, "ctrl.tx.bytesrate") + req.Equal(entityId, "3tO.kK.Dn") + + name = "ctrl..tO.kK.Dn.tx.bytesrate" + name, entityId, _ = ExtractId(name, "ctrl.", 2) + req.Equal(name, "ctrl.tx.bytesrate") + req.Equal(entityId, ".tO.kK.Dn") + + name = "ctrl..tO.kK.D..tx.bytesrate" + name, entityId, _ = ExtractId(name, "ctrl.", 2) + req.Equal(name, "ctrl.tx.bytesrate") + req.Equal(entityId, ".tO.kK.D.") +}