@@ -3,6 +3,7 @@ package observek8sattributesprocessor
33import (
44 "context"
55 "encoding/json"
6+ "errors"
67
78 "go.opentelemetry.io/collector/component"
89 "go.opentelemetry.io/collector/pdata/pcommon"
@@ -21,18 +22,20 @@ type K8sEventsProcessor struct {
2122 actions []K8sEventProcessorAction
2223}
2324
25+ type transformFn func (plog.LogRecord ) any
26+
2427type K8sEventProcessorAction struct {
25- Key string
26- ValueFn func (plog. LogRecord ) any
27- FilterFn func (K8sEvent ) bool
28+ Key string
29+ Transform transformFn
30+ FilterFn func (K8sEvent ) bool
2831}
2932
3033func newK8sEventsProcessor (logger * zap.Logger , cfg component.Config ) * K8sEventsProcessor {
3134 return & K8sEventsProcessor {
3235 cfg : cfg ,
3336 logger : logger ,
3437 actions : []K8sEventProcessorAction {
35- PodStatusAction , NodeStatusAction , NodeRolesAction ,
38+ PodStatusAction , NodeStatusAction , NodeRolesAction , PodContainersCountsAction ,
3639 },
3740 }
3841}
@@ -81,47 +84,85 @@ func (kep *K8sEventsProcessor) processLogs(_ context.Context, logs plog.Logs) (p
8184 facetsMap = transformMap .PutEmptyMap ("facets" )
8285 }
8386
84- // This is where the custom processor actually computes the value
85- value := action .ValueFn (lr )
86-
87- // TODO [eg] we probably want to make this more modular. For
88- // now using FromRaw for complex types is fine, since we
89- // don't plan to generate arbitrarily complex/nested facets.
90- // For some time coming we will produce facets that are at
91- // most slices of simple types of map of simple types,
92- // nothing beyond that.
93- switch typed := value .(type ) {
94- case string :
95- facetsMap .PutStr (action .Key , typed )
96- case int64 :
97- facetsMap .PutInt (action .Key , typed )
98- case bool :
99- facetsMap .PutBool (action .Key , typed )
100- case float64 :
101- facetsMap .PutDouble (action .Key , typed )
102- case []string :
103- // []string can't fallback to using FromRaw([]any), as
104- // the default implementation of FromRaw is not smart
105- // enough to understand that the slice contains all
106- // string, and inserts them as bytes, instead
107- slc := facetsMap .PutEmptySlice (action .Key )
108- slc .EnsureCapacity (len (typed ))
109- for _ , str := range typed {
110- slc .AppendEmpty ().SetStr (str )
111- }
112- case map [string ]string :
113- // Same reasoning as []string
114- mp := facetsMap .PutEmptyMap (action .Key )
115- mp .EnsureCapacity (len (typed ))
116- for k , v := range typed {
117- mp .PutStr (k , v )
118- }
119- default :
120- kep .logger .Error ("sending the generated facet to Observe in bytes since no custom serialization logic is implemented" , zap .Error (err ))
87+ // This is where the custom processor actually computes the transformed value(s)
88+ value := action .Transform (lr )
89+ if err := mapPut (facetsMap , action .Key , value ); err != nil {
90+ kep .logger .Error (err .Error ())
12191 }
12292 }
12393 }
12494 }
12595 }
12696 return logs , nil
12797}
98+
99+ func slicePut (theSlice pcommon.Slice , value any ) error {
100+ elem := theSlice .AppendEmpty ()
101+ switch typed := value .(type ) {
102+ case string :
103+ elem .SetStr (typed )
104+ case int64 :
105+ elem .SetInt (typed )
106+ case bool :
107+ elem .SetBool (typed )
108+ case float64 :
109+ elem .SetDouble (typed )
110+ // Let's not complicate things and avoid putting maps/slices into slices There's
111+ // gotta be an easier way to model the processor's output to avoid this
112+ default :
113+ return errors .New ("sending the generated facet to Observe in bytes since no custom serialization logic is implemented" )
114+ }
115+
116+ return nil
117+ }
118+
119+ // puts "anything" into a map, with some assumptions and intentional
120+ // limitations:
121+ //
122+ // - No nested maps: can only pass maps with signature map[string]any as
123+ // value. In this case, we will unwrap the inner map and put all elements of
124+ // "value" as single elements directly into theMap.
125+ //
126+ // - No nested slices: can only put "base types" inside slices (although
127+ // elements of a slice can be of different [base] types).
128+ //
129+ // - Not all "base types" are covered. For instance, numbers are only int64 and float64.
130+ func mapPut (theMap pcommon.Map , key string , value any ) error {
131+ switch typed := value .(type ) {
132+ case string :
133+ theMap .PutStr (key , typed )
134+ case int64 :
135+ theMap .PutInt (key , typed )
136+ case bool :
137+ theMap .PutBool (key , typed )
138+ case float64 :
139+ theMap .PutDouble (key , typed )
140+ case []any :
141+ // []string can't fallback to using FromRaw([]any), as
142+ // the default implementation of FromRaw is not smart
143+ // enough to understand that the slice contains all
144+ // string, and inserts them as bytes, instead
145+ slc := theMap .PutEmptySlice (key )
146+ slc .EnsureCapacity (len (typed ))
147+ for _ , elem := range typed {
148+ slicePut (slc , elem )
149+ }
150+ case map [string ]any :
151+ // For now, we treat this as a special case to allow single transforms
152+ // generating multiple facets with a single function (most likely
153+ // because they are trivial to compute and to avoid iterating over the
154+ // same array more than once, see Node restarts, total_containers,
155+ // ready_containers). This means that we IGNORE the "key" associated
156+ // with the whole map and insert the contents of the map directly into
157+ // theMap.
158+ theMap .EnsureCapacity (theMap .Len () + len (typed ))
159+ for k , v := range typed {
160+ mapPut (theMap , k , v )
161+ }
162+ default :
163+ return errors .New ("sending the generated facet to Observe in bytes since no custom serialization logic is implemented" )
164+ }
165+
166+ return nil
167+
168+ }
0 commit comments