1818package encode
1919
2020import (
21- "strings"
22- "time"
23-
24- "github.com/netobserv/flowlogs-pipeline/pkg/api"
2521 "github.com/netobserv/flowlogs-pipeline/pkg/config"
2622 "github.com/netobserv/flowlogs-pipeline/pkg/operational"
2723 "github.com/netobserv/flowlogs-pipeline/pkg/pipeline/encode/metrics"
@@ -41,12 +37,9 @@ type MetricsCommonStruct struct {
4137 counters map [string ]mInfoStruct
4238 histos map [string ]mInfoStruct
4339 aggHistos map [string ]mInfoStruct
44- mCache * putils.TimedCache
45- mChacheLenMetric prometheus.Gauge
4640 metricsProcessed prometheus.Counter
4741 metricsDropped prometheus.Counter
4842 errorsCounter * prometheus.CounterVec
49- expiryTime time.Duration
5043 exitChan <- chan struct {}
5144}
5245
7770 operational .TypeCounter ,
7871 "error" , "metric" , "key" ,
7972 )
80- mChacheLen = operational .DefineMetric (
81- "encode_prom_metrics_reported" ,
82- "Total number of prometheus metrics reported by this stage" ,
83- operational .TypeGauge ,
84- "stage" ,
85- )
8673)
8774
8875func (m * MetricsCommonStruct ) AddCounter (name string , g interface {}, info * metrics.Preprocessed ) {
@@ -110,12 +97,12 @@ func (m *MetricsCommonStruct) MetricCommonEncode(mci MetricsCommonInterface, met
11097
11198 // Process counters
11299 for _ , mInfo := range m .counters {
113- labelSets , value := m .prepareMetric (mci , metricRecord , mInfo .info , mInfo . genericMetric )
100+ labelSets , value := m .prepareMetric (metricRecord , mInfo .info )
114101 if labelSets == nil {
115102 continue
116103 }
117104 for _ , labels := range labelSets {
118- err := mci .ProcessCounter (mInfo .genericMetric , labels . lMap , value )
105+ err := mci .ProcessCounter (mInfo .genericMetric , labels , value )
119106 if err != nil {
120107 log .Errorf ("labels registering error on %s: %v" , mInfo .info .Name , err )
121108 m .errorsCounter .WithLabelValues ("LabelsRegisteringError" , mInfo .info .Name , "" ).Inc ()
@@ -127,12 +114,12 @@ func (m *MetricsCommonStruct) MetricCommonEncode(mci MetricsCommonInterface, met
127114
128115 // Process gauges
129116 for _ , mInfo := range m .gauges {
130- labelSets , value := m .prepareMetric (mci , metricRecord , mInfo .info , mInfo . genericMetric )
117+ labelSets , value := m .prepareMetric (metricRecord , mInfo .info )
131118 if labelSets == nil {
132119 continue
133120 }
134121 for _ , labels := range labelSets {
135- err := mci .ProcessGauge (mInfo .genericMetric , labels . lMap , value , labels . key )
122+ err := mci .ProcessGauge (mInfo .genericMetric , labels , value , "" )
136123 if err != nil {
137124 log .Errorf ("labels registering error on %s: %v" , mInfo .info .Name , err )
138125 m .errorsCounter .WithLabelValues ("LabelsRegisteringError" , mInfo .info .Name , "" ).Inc ()
@@ -144,12 +131,12 @@ func (m *MetricsCommonStruct) MetricCommonEncode(mci MetricsCommonInterface, met
144131
145132 // Process histograms
146133 for _ , mInfo := range m .histos {
147- labelSets , value := m .prepareMetric (mci , metricRecord , mInfo .info , mInfo . genericMetric )
134+ labelSets , value := m .prepareMetric (metricRecord , mInfo .info )
148135 if labelSets == nil {
149136 continue
150137 }
151138 for _ , labels := range labelSets {
152- err := mci .ProcessHist (mInfo .genericMetric , labels . lMap , value )
139+ err := mci .ProcessHist (mInfo .genericMetric , labels , value )
153140 if err != nil {
154141 log .Errorf ("labels registering error on %s: %v" , mInfo .info .Name , err )
155142 m .errorsCounter .WithLabelValues ("LabelsRegisteringError" , mInfo .info .Name , "" ).Inc ()
@@ -161,12 +148,12 @@ func (m *MetricsCommonStruct) MetricCommonEncode(mci MetricsCommonInterface, met
161148
162149 // Process pre-aggregated histograms
163150 for _ , mInfo := range m .aggHistos {
164- labelSets , values := m .prepareAggHisto (mci , metricRecord , mInfo .info , mInfo . genericMetric )
151+ labelSets , values := m .prepareAggHisto (metricRecord , mInfo .info )
165152 if labelSets == nil {
166153 continue
167154 }
168155 for _ , labels := range labelSets {
169- err := mci .ProcessAggHist (mInfo .genericMetric , labels . lMap , values )
156+ err := mci .ProcessAggHist (mInfo .genericMetric , labels , values )
170157 if err != nil {
171158 log .Errorf ("labels registering error on %s: %v" , mInfo .info .Name , err )
172159 m .errorsCounter .WithLabelValues ("LabelsRegisteringError" , mInfo .info .Name , "" ).Inc ()
@@ -177,7 +164,7 @@ func (m *MetricsCommonStruct) MetricCommonEncode(mci MetricsCommonInterface, met
177164 }
178165}
179166
180- func (m * MetricsCommonStruct ) prepareMetric (mci MetricsCommonInterface , flow config.GenericMap , info * metrics.Preprocessed , mv interface {} ) ([]labelsKeyAndMap , float64 ) {
167+ func (m * MetricsCommonStruct ) prepareMetric (flow config.GenericMap , info * metrics.Preprocessed ) ([]map [ string ] string , float64 ) {
181168 flatParts := info .GenerateFlatParts (flow )
182169 ok , flatParts := info .ApplyFilters (flow , flatParts )
183170 if ! ok {
@@ -198,22 +185,15 @@ func (m *MetricsCommonStruct) prepareMetric(mci MetricsCommonInterface, flow con
198185 }
199186
200187 labelSets := extractLabels (flow , flatParts , info )
201- var lkms []labelsKeyAndMap
188+ var lkms []map [ string ] string
202189 for _ , ls := range labelSets {
203190 // Update entry for expiry mechanism (the entry itself is its own cleanup function)
204- lkm := ls .toKeyAndMap (info )
205- lkms = append (lkms , lkm )
206- cacheEntry := mci .GetChacheEntry (lkm .lMap , mv )
207- ok := m .mCache .UpdateCacheEntry (lkm .key , cacheEntry )
208- if ! ok {
209- m .metricsDropped .Inc ()
210- return nil , 0
211- }
191+ lkms = append (lkms , ls .toMap ())
212192 }
213193 return lkms , floatVal
214194}
215195
216- func (m * MetricsCommonStruct ) prepareAggHisto (mci MetricsCommonInterface , flow config.GenericMap , info * metrics.Preprocessed , mc interface {} ) ([]labelsKeyAndMap , []float64 ) {
196+ func (m * MetricsCommonStruct ) prepareAggHisto (flow config.GenericMap , info * metrics.Preprocessed ) ([]map [ string ] string , []float64 ) {
217197 flatParts := info .GenerateFlatParts (flow )
218198 ok , flatParts := info .ApplyFilters (flow , flatParts )
219199 if ! ok {
@@ -231,17 +211,10 @@ func (m *MetricsCommonStruct) prepareAggHisto(mci MetricsCommonInterface, flow c
231211 }
232212
233213 labelSets := extractLabels (flow , flatParts , info )
234- var lkms []labelsKeyAndMap
214+ var lkms []map [ string ] string
235215 for _ , ls := range labelSets {
236216 // Update entry for expiry mechanism (the entry itself is its own cleanup function)
237- lkm := ls .toKeyAndMap (info )
238- lkms = append (lkms , lkm )
239- cacheEntry := mci .GetChacheEntry (lkm .lMap , mc )
240- ok := m .mCache .UpdateCacheEntry (lkm .key , cacheEntry )
241- if ! ok {
242- m .metricsDropped .Inc ()
243- return nil , nil
244- }
217+ lkms = append (lkms , ls .toMap ())
245218 }
246219 return lkms , values
247220}
@@ -266,22 +239,12 @@ type label struct {
266239
267240type labelSet []label
268241
269- type labelsKeyAndMap struct {
270- key string
271- lMap map [string ]string
272- }
273-
274- func (l labelSet ) toKeyAndMap (info * metrics.Preprocessed ) labelsKeyAndMap {
275- key := strings.Builder {}
276- key .WriteString (info .Name )
277- key .WriteRune ('|' )
278- m := map [string ]string {}
242+ func (l labelSet ) toMap () map [string ]string {
243+ m := make (map [string ]string , len (l ))
279244 for _ , kv := range l {
280- key .WriteString (kv .value )
281- key .WriteRune ('|' )
282245 m [kv .key ] = kv .value
283246 }
284- return labelsKeyAndMap { key : key . String (), lMap : m }
247+ return m
285248}
286249
287250// extractLabels takes the flow and a single metric definition as input.
@@ -313,41 +276,23 @@ func newLabelSet(part config.GenericMap, labels []metrics.MappedLabel) labelSet
313276 return ls
314277}
315278
316- func (m * MetricsCommonStruct ) cleanupExpiredEntriesLoop (callback putils.CacheCallback ) {
317- ticker := time .NewTicker (m .expiryTime )
318- for {
319- select {
320- case <- m .exitChan :
321- log .Debugf ("exiting cleanupExpiredEntriesLoop because of signal" )
322- return
323- case <- ticker .C :
324- m .mCache .CleanupExpiredEntries (m .expiryTime , callback )
325- }
326- }
327- }
328-
329279func (m * MetricsCommonStruct ) cleanupInfoStructs () {
330280 m .gauges = map [string ]mInfoStruct {}
331281 m .counters = map [string ]mInfoStruct {}
332282 m .histos = map [string ]mInfoStruct {}
333283 m .aggHistos = map [string ]mInfoStruct {}
334284}
335285
336- func NewMetricsCommonStruct (opMetrics * operational.Metrics , maxCacheEntries int , name string , expiryTime api.Duration , callback putils.CacheCallback ) * MetricsCommonStruct {
337- mChacheLenMetric := opMetrics .NewGauge (& mChacheLen , name )
286+ func NewMetricsCommonStruct (opMetrics * operational.Metrics , name string ) * MetricsCommonStruct {
338287 m := & MetricsCommonStruct {
339- mCache : putils .NewTimedCache (maxCacheEntries , mChacheLenMetric ),
340- mChacheLenMetric : mChacheLenMetric ,
341288 metricsProcessed : opMetrics .NewCounter (& metricsProcessed , name ),
342289 metricsDropped : opMetrics .NewCounter (& metricsDropped , name ),
343290 errorsCounter : opMetrics .NewCounterVec (& encodePromErrors ),
344- expiryTime : expiryTime .Duration ,
345291 exitChan : putils .ExitChannel (),
346292 gauges : map [string ]mInfoStruct {},
347293 counters : map [string ]mInfoStruct {},
348294 histos : map [string ]mInfoStruct {},
349295 aggHistos : map [string ]mInfoStruct {},
350296 }
351- go m .cleanupExpiredEntriesLoop (callback )
352297 return m
353298}
0 commit comments