Skip to content

Commit 5fba26a

Browse files
authored
NETOBSERV-1935: enable metrics from list/nested fields (#739)
* NETOBSERV-1935: enable metrics from list/nested fields This makes it possible to generate metrics with labels/filters set on list fields (e.g. interfaces) and nested fields (e.g. soon-coming structured network events) In metrics API, user needs to configure explicitly which list field needs to be "flattened", in order to be consumable as filters/labels. Nested fields can be consumed as filters/labels with the ">" character; E.g: `flatten: [networkEvents], filters: [{key: "networkEvents>type", value: "acl"}], labels: [networkEvents>name]` This is a sample config to filter a metric for ACL events and label it by name * linter
1 parent 5bb2146 commit 5fba26a

File tree

15 files changed

+569
-118
lines changed

15 files changed

+569
-118
lines changed

docs/api.md

Lines changed: 2 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -30,6 +30,7 @@ Following is the supported API format for prometheus encode:
3030
valueKey: entry key from which to resolve metric value
3131
labels: labels to be associated with the metric
3232
remap: optional remapping of labels
33+
flatten: list fields to be flattened
3334
buckets: histogram buckets
3435
valueScale: scale factor of the value (MetricVal := FlowVal / Scale)
3536
prefix: prefix added to each metric name
@@ -444,6 +445,7 @@ Following is the supported API format for writing metrics to an OpenTelemetry co
444445
valueKey: entry key from which to resolve metric value
445446
labels: labels to be associated with the metric
446447
remap: optional remapping of labels
448+
flatten: list fields to be flattened
447449
buckets: histogram buckets
448450
valueScale: scale factor of the value (MetricVal := FlowVal / Scale)
449451
pushTimeInterval: how often should metrics be sent to collector:

pkg/api/encode_prom.go

Lines changed: 1 addition & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -53,6 +53,7 @@ type MetricsItem struct {
5353
ValueKey string `yaml:"valueKey" json:"valueKey" doc:"entry key from which to resolve metric value"`
5454
Labels []string `yaml:"labels" json:"labels" doc:"labels to be associated with the metric"`
5555
Remap map[string]string `yaml:"remap" json:"remap" doc:"optional remapping of labels"`
56+
Flatten []string `yaml:"flatten" json:"flatten" doc:"list fields to be flattened"`
5657
Buckets []float64 `yaml:"buckets" json:"buckets" doc:"histogram buckets"`
5758
ValueScale float64 `yaml:"valueScale,omitempty" json:"valueScale,omitempty" doc:"scale factor of the value (MetricVal := FlowVal / Scale)"`
5859
}

pkg/confgen/confgen_test.go

Lines changed: 4 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -152,6 +152,7 @@ func Test_RunShortConfGen(t *testing.T) {
152152
ValueKey: "test_aggregates_value",
153153
Labels: []string{"groupByKeys", "aggregate"},
154154
Remap: map[string]string{},
155+
Flatten: []string{},
155156
Buckets: []float64{},
156157
}},
157158
}, out.Parameters[3].Encode.Prom)
@@ -234,6 +235,7 @@ func Test_RunConfGenNoAgg(t *testing.T) {
234235
ValueKey: "Bytes",
235236
Labels: []string{"service"},
236237
Remap: map[string]string{},
238+
Flatten: []string{},
237239
Buckets: []float64{},
238240
}},
239241
}, out.Parameters[2].Encode.Prom)
@@ -339,6 +341,7 @@ func Test_RunLongConfGen(t *testing.T) {
339341
ValueKey: "test_aggregates_value",
340342
Labels: []string{"groupByKeys", "aggregate"},
341343
Remap: map[string]string{},
344+
Flatten: []string{},
342345
Buckets: []float64{},
343346
}, {
344347
Name: "test_histo",
@@ -347,6 +350,7 @@ func Test_RunLongConfGen(t *testing.T) {
347350
ValueKey: "test_aggregates_value",
348351
Labels: []string{"groupByKeys", "aggregate"},
349352
Remap: map[string]string{},
353+
Flatten: []string{},
350354
Buckets: []float64{},
351355
}},
352356
}, out.Parameters[4].Encode.Prom)

pkg/config/pipeline_builder_test.go

Lines changed: 2 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -138,6 +138,7 @@ func TestKafkaPromPipeline(t *testing.T) {
138138
ValueKey: "recent_count",
139139
Labels: []string{"by", "aggregate"},
140140
Remap: map[string]string{},
141+
Flatten: []string{},
141142
Buckets: []float64{},
142143
}},
143144
Prefix: "flp_",
@@ -171,7 +172,7 @@ func TestKafkaPromPipeline(t *testing.T) {
171172

172173
b, err = json.Marshal(params[4])
173174
require.NoError(t, err)
174-
require.JSONEq(t, `{"name":"prom","encode":{"type":"prom","prom":{"expiryTime":"50s", "metrics":[{"name":"connections_per_source_as","type":"counter","filters":[{"key":"name","value":"src_as_connection_count"}],"valueKey":"recent_count","labels":["by","aggregate"],"remap":{},"buckets":[]}],"prefix":"flp_"}}}`, string(b))
175+
require.JSONEq(t, `{"name":"prom","encode":{"type":"prom","prom":{"expiryTime":"50s", "metrics":[{"name":"connections_per_source_as","type":"counter","filters":[{"key":"name","value":"src_as_connection_count"}],"valueKey":"recent_count","labels":["by","aggregate"],"flatten":[],"remap":{},"buckets":[]}],"prefix":"flp_"}}}`, string(b))
175176
}
176177

177178
func TestForkPipeline(t *testing.T) {

pkg/pipeline/encode/encode_prom.go

Lines changed: 8 additions & 7 deletions
Original file line numberDiff line numberDiff line change
@@ -25,6 +25,7 @@ import (
2525
"github.com/netobserv/flowlogs-pipeline/pkg/api"
2626
"github.com/netobserv/flowlogs-pipeline/pkg/config"
2727
"github.com/netobserv/flowlogs-pipeline/pkg/operational"
28+
"github.com/netobserv/flowlogs-pipeline/pkg/pipeline/encode/metrics"
2829
promserver "github.com/netobserv/flowlogs-pipeline/pkg/prometheus"
2930
"github.com/prometheus/client_golang/prometheus"
3031
"github.com/sirupsen/logrus"
@@ -114,25 +115,25 @@ func (e *EncodeProm) Cleanup(cleanupFunc interface{}) {
114115
cleanupFunc.(func())()
115116
}
116117

117-
func (e *EncodeProm) addCounter(fullMetricName string, mInfo *MetricInfo) prometheus.Collector {
118+
func (e *EncodeProm) addCounter(fullMetricName string, mInfo *metrics.Preprocessed) prometheus.Collector {
118119
counter := prometheus.NewCounterVec(prometheus.CounterOpts{Name: fullMetricName, Help: ""}, mInfo.TargetLabels())
119120
e.metricCommon.AddCounter(fullMetricName, counter, mInfo)
120121
return counter
121122
}
122123

123-
func (e *EncodeProm) addGauge(fullMetricName string, mInfo *MetricInfo) prometheus.Collector {
124+
func (e *EncodeProm) addGauge(fullMetricName string, mInfo *metrics.Preprocessed) prometheus.Collector {
124125
gauge := prometheus.NewGaugeVec(prometheus.GaugeOpts{Name: fullMetricName, Help: ""}, mInfo.TargetLabels())
125126
e.metricCommon.AddGauge(fullMetricName, gauge, mInfo)
126127
return gauge
127128
}
128129

129-
func (e *EncodeProm) addHistogram(fullMetricName string, mInfo *MetricInfo) prometheus.Collector {
130+
func (e *EncodeProm) addHistogram(fullMetricName string, mInfo *metrics.Preprocessed) prometheus.Collector {
130131
histogram := prometheus.NewHistogramVec(prometheus.HistogramOpts{Name: fullMetricName, Help: ""}, mInfo.TargetLabels())
131132
e.metricCommon.AddHist(fullMetricName, histogram, mInfo)
132133
return histogram
133134
}
134135

135-
func (e *EncodeProm) addAgghistogram(fullMetricName string, mInfo *MetricInfo) prometheus.Collector {
136+
func (e *EncodeProm) addAgghistogram(fullMetricName string, mInfo *metrics.Preprocessed) prometheus.Collector {
136137
agghistogram := prometheus.NewHistogramVec(prometheus.HistogramOpts{Name: fullMetricName, Help: ""}, mInfo.TargetLabels())
137138
e.metricCommon.AddAggHist(fullMetricName, agghistogram, mInfo)
138139
return agghistogram
@@ -176,10 +177,10 @@ func (e *EncodeProm) cleanDeletedMetrics(newCfg api.PromEncode) {
176177
}
177178

178179
// returns true if a registry restart is needed
179-
func (e *EncodeProm) checkMetricUpdate(prefix string, apiItem *api.MetricsItem, store map[string]mInfoStruct, createMetric func(string, *MetricInfo) prometheus.Collector) bool {
180+
func (e *EncodeProm) checkMetricUpdate(prefix string, apiItem *api.MetricsItem, store map[string]mInfoStruct, createMetric func(string, *metrics.Preprocessed) prometheus.Collector) bool {
180181
fullMetricName := prefix + apiItem.Name
181182
plog.Debugf("Checking metric: %s", fullMetricName)
182-
mInfo := CreateMetricInfo(apiItem)
183+
mInfo := metrics.Preprocess(apiItem)
183184
if oldMetric, ok := store[fullMetricName]; ok {
184185
if !reflect.DeepEqual(mInfo.TargetLabels(), oldMetric.info.TargetLabels()) {
185186
plog.Debug("Changes detected in labels")
@@ -257,7 +258,7 @@ func (e *EncodeProm) resetRegistry() {
257258
for i := range e.cfg.Metrics {
258259
mCfg := &e.cfg.Metrics[i]
259260
fullMetricName := e.cfg.Prefix + mCfg.Name
260-
mInfo := CreateMetricInfo(mCfg)
261+
mInfo := metrics.Preprocess(mCfg)
261262
plog.Debugf("Create metric: %s, Labels: %v", fullMetricName, mInfo.TargetLabels())
262263
var m prometheus.Collector
263264
switch mCfg.Type {

pkg/pipeline/encode/encode_prom_test.go

Lines changed: 198 additions & 15 deletions
Original file line numberDiff line numberDiff line change
@@ -543,9 +543,6 @@ func Test_MissingLabels(t *testing.T) {
543543
},
544544
}
545545
params := api.PromEncode{
546-
ExpiryTime: api.Duration{
547-
Duration: time.Duration(60 * time.Second),
548-
},
549546
Metrics: []api.MetricsItem{
550547
{
551548
Name: "my_counter",
@@ -588,9 +585,6 @@ func Test_Remap(t *testing.T) {
588585
},
589586
}
590587
params := api.PromEncode{
591-
ExpiryTime: api.Duration{
592-
Duration: time.Duration(60 * time.Second),
593-
},
594588
Metrics: []api.MetricsItem{
595589
{
596590
Name: "my_counter",
@@ -616,6 +610,204 @@ func Test_Remap(t *testing.T) {
616610
require.Contains(t, exposed, `my_counter{ip="10.0.0.3",namespace="B"} 4`)
617611
}
618612

613+
func Test_WithListField(t *testing.T) {
614+
metrics := []config.GenericMap{
615+
{
616+
"namespace": "A",
617+
"interfaces": []string{"eth0", "123456"},
618+
"bytes": 7,
619+
},
620+
{
621+
"namespace": "A",
622+
"interfaces": []any{"eth0", "abcdef"},
623+
"bytes": 1,
624+
},
625+
{
626+
"namespace": "A",
627+
"interfaces": []any{"eth0", "xyz"},
628+
"bytes": 10,
629+
},
630+
{
631+
"namespace": "B",
632+
"bytes": 2,
633+
},
634+
{
635+
"namespace": "C",
636+
"interfaces": []string{},
637+
"bytes": 3,
638+
},
639+
}
640+
params := api.PromEncode{
641+
Metrics: []api.MetricsItem{
642+
{
643+
Name: "my_counter",
644+
Type: "counter",
645+
ValueKey: "bytes",
646+
Filters: []api.MetricsFilter{
647+
{Key: "interfaces", Value: "xyz", Type: api.MetricFilterNotEqual},
648+
},
649+
Flatten: []string{"interfaces"},
650+
Labels: []string{"namespace", "interfaces"},
651+
Remap: map[string]string{"interfaces": "interface"},
652+
},
653+
},
654+
}
655+
656+
encodeProm, err := initProm(&params)
657+
require.NoError(t, err)
658+
for _, metric := range metrics {
659+
encodeProm.Encode(metric)
660+
}
661+
time.Sleep(100 * time.Millisecond)
662+
663+
exposed := test.ReadExposedMetrics(t, encodeProm.server)
664+
665+
require.Contains(t, exposed, `my_counter{interface="eth0",namespace="A"} 18`)
666+
require.Contains(t, exposed, `my_counter{interface="123456",namespace="A"} 7`)
667+
require.Contains(t, exposed, `my_counter{interface="abcdef",namespace="A"} 1`)
668+
require.Contains(t, exposed, `my_counter{interface="",namespace="B"} 2`)
669+
require.Contains(t, exposed, `my_counter{interface="",namespace="C"} 3`)
670+
require.NotContains(t, exposed, `"xyz"`)
671+
}
672+
673+
func Test_WithObjectListField(t *testing.T) {
674+
metrics := []config.GenericMap{
675+
{
676+
"namespace": "A",
677+
"events": []any{
678+
config.GenericMap{"type": "acl", "name": "my_policy"},
679+
},
680+
"bytes": 1,
681+
},
682+
{
683+
"namespace": "A",
684+
"events": []any{
685+
config.GenericMap{"type": "egress", "name": "my_egress"},
686+
config.GenericMap{"type": "acl", "name": "my_policy"},
687+
},
688+
"bytes": 10,
689+
},
690+
{
691+
"namespace": "B",
692+
"bytes": 2,
693+
},
694+
{
695+
"namespace": "C",
696+
"events": []string{},
697+
"bytes": 3,
698+
},
699+
}
700+
params := api.PromEncode{
701+
Metrics: []api.MetricsItem{
702+
{
703+
Name: "policy_counter",
704+
Type: "counter",
705+
ValueKey: "bytes",
706+
Filters: []api.MetricsFilter{
707+
{Key: "events>type", Value: "acl", Type: api.MetricFilterEqual},
708+
},
709+
Labels: []string{"namespace", "events>name"},
710+
Flatten: []string{"events"},
711+
Remap: map[string]string{"events>name": "name"},
712+
},
713+
},
714+
}
715+
716+
encodeProm, err := initProm(&params)
717+
require.NoError(t, err)
718+
for _, metric := range metrics {
719+
encodeProm.Encode(metric)
720+
}
721+
time.Sleep(100 * time.Millisecond)
722+
723+
exposed := test.ReadExposedMetrics(t, encodeProm.server)
724+
725+
require.Contains(t, exposed, `policy_counter{name="my_policy",namespace="A"} 11`)
726+
require.NotContains(t, exposed, `"my_egress"`)
727+
require.NotContains(t, exposed, `"B"`)
728+
require.NotContains(t, exposed, `"C"`)
729+
}
730+
731+
func Test_WithObjectListField_bis(t *testing.T) {
732+
metrics := []config.GenericMap{
733+
{
734+
"namespace": "A",
735+
"events": []any{
736+
config.GenericMap{"type": "egress", "name": "my_egress"},
737+
config.GenericMap{"type": "acl", "name": "my_policy"},
738+
},
739+
"bytes": 10,
740+
},
741+
}
742+
params := api.PromEncode{
743+
Metrics: []api.MetricsItem{
744+
{
745+
Name: "policy_counter",
746+
Type: "counter",
747+
ValueKey: "bytes",
748+
Filters: []api.MetricsFilter{
749+
{Key: "events>type", Value: "acl", Type: api.MetricFilterEqual},
750+
},
751+
Flatten: []string{"events"},
752+
Labels: []string{"namespace"},
753+
},
754+
},
755+
}
756+
757+
encodeProm, err := initProm(&params)
758+
require.NoError(t, err)
759+
for _, metric := range metrics {
760+
encodeProm.Encode(metric)
761+
}
762+
time.Sleep(100 * time.Millisecond)
763+
764+
exposed := test.ReadExposedMetrics(t, encodeProm.server)
765+
766+
require.Contains(t, exposed, `policy_counter{namespace="A"} 10`)
767+
require.NotContains(t, exposed, `"my_egress"`)
768+
require.NotContains(t, exposed, `"B"`)
769+
require.NotContains(t, exposed, `"C"`)
770+
}
771+
772+
func Test_WithObjectListField_ter(t *testing.T) {
773+
metrics := []config.GenericMap{
774+
{
775+
"namespace": "A",
776+
"events": []any{
777+
config.GenericMap{"type": "egress", "name": "my_egress"},
778+
config.GenericMap{"type": "acl", "name": "my_policy"},
779+
},
780+
"bytes": 10,
781+
},
782+
}
783+
params := api.PromEncode{
784+
Metrics: []api.MetricsItem{
785+
{
786+
Name: "policy_counter",
787+
Type: "counter",
788+
ValueKey: "bytes",
789+
Labels: []string{"namespace", "events>name"},
790+
Flatten: []string{"events"},
791+
Remap: map[string]string{"events>name": "name"},
792+
},
793+
},
794+
}
795+
796+
encodeProm, err := initProm(&params)
797+
require.NoError(t, err)
798+
for _, metric := range metrics {
799+
encodeProm.Encode(metric)
800+
}
801+
time.Sleep(100 * time.Millisecond)
802+
803+
exposed := test.ReadExposedMetrics(t, encodeProm.server)
804+
805+
require.Contains(t, exposed, `policy_counter{name="my_policy",namespace="A"} 10`)
806+
require.Contains(t, exposed, `policy_counter{name="my_egress",namespace="A"} 10`)
807+
require.NotContains(t, exposed, `"B"`)
808+
require.NotContains(t, exposed, `"C"`)
809+
}
810+
619811
func buildFlow() config.GenericMap {
620812
return config.GenericMap{
621813
"srcIP": "10.0.0." + strconv.Itoa(rand.Intn(20)),
@@ -754,12 +946,3 @@ func Test_MultipleProm(t *testing.T) {
754946

755947
// TODO: Add test for different addresses, but need to deal with StartPromServer (ListenAndServe)
756948
}
757-
758-
func Test_Filters_extractVarLookups(t *testing.T) {
759-
variables := extractVarLookups("$(abc)--$(def)")
760-
761-
require.Equal(t, [][]string{{"$(abc)", "abc"}, {"$(def)", "def"}}, variables)
762-
763-
variables = extractVarLookups("")
764-
require.Empty(t, variables)
765-
}
Lines changed: 28 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,28 @@
1+
package metrics
2+
3+
import "github.com/netobserv/flowlogs-pipeline/pkg/config"
4+
5+
func (p *Preprocessed) ApplyFilters(flow config.GenericMap, flatParts []config.GenericMap) (bool, []config.GenericMap) {
6+
filteredParts := flatParts
7+
for _, filter := range p.filters {
8+
if filter.useFlat {
9+
filteredParts = filter.filterFlatParts(filteredParts)
10+
if len(filteredParts) == 0 {
11+
return false, nil
12+
}
13+
} else if !filter.predicate(flow) {
14+
return false, nil
15+
}
16+
}
17+
return true, filteredParts
18+
}
19+
20+
func (pf *preprocessedFilter) filterFlatParts(flatParts []config.GenericMap) []config.GenericMap {
21+
var filteredParts []config.GenericMap
22+
for _, part := range flatParts {
23+
if pf.predicate(part) {
24+
filteredParts = append(filteredParts, part)
25+
}
26+
}
27+
return filteredParts
28+
}

0 commit comments

Comments
 (0)