OB-34558 Add Pod "counts" facets

Add a single custom processor that generates 3 facets:
- restarts
- ready_containers
- total_containers

The reasoning behind a single-processor for multiple facets is that all
of these 3 facets are computed by iterating the same slice. Computing
them in 3 different actions means iterating the same slice 3 times.

Author: obs-gh-enricogiorio

components/processors/observek8sattributesprocessor/node.go

@@ -0,0 +1,5 @@
+package observek8sattributesprocessor
+
+func filterNodeEvents(event K8sEvent) bool {
+	return event.Kind == "Node"
+}

components/processors/observek8sattributesprocessor/noderoles.go

@@ -20,8 +20,8 @@ const (
 )
 
 var NodeRolesAction = K8sEventProcessorAction{
-	Key:     NodeRolesAttributeKey,
-	ValueFn: getNodeRoles,
+	Key:       NodeRolesAttributeKey,
+	Transform: getNodeRoles,
 	// Reuse the function to filter events for nodes
 	FilterFn: filterNodeEvents,
 }
@@ -47,5 +47,9 @@ func getNodeRoles(objLog plog.LogRecord) any {
 		}
 	}
 
-	return roles.List()
+	ret := make([]any, 0, roles.Len())
+	for _, role := range roles.List() {
+		ret = append(ret, role)
+	}
+	return ret
 }

components/processors/observek8sattributesprocessor/nodestatus.go

@@ -12,13 +12,9 @@ const (
 )
 
 var NodeStatusAction = K8sEventProcessorAction{
-	Key:      NodeStatusAttributeKey,
-	ValueFn:  getNodeStatus,
-	FilterFn: filterNodeEvents,
-}
-
-func filterNodeEvents(event K8sEvent) bool {
-	return event.Kind == "Node"
+	Key:       NodeStatusAttributeKey,
+	Transform: getNodeStatus,
+	FilterFn:  filterNodeEvents,
 }
 
 // Generates the Node "status" facet. Assumes that objLog is a log from a Node event.

components/processors/observek8sattributesprocessor/pod.go

@@ -0,0 +1,5 @@
+package observek8sattributesprocessor
+
+func filterPodEvents(event K8sEvent) bool {
+	return event.Kind == "Pod"
+}

components/processors/observek8sattributesprocessor/podcounts.go

@@ -0,0 +1,59 @@
+package observek8sattributesprocessor
+
+import (
+	"encoding/json"
+
+	"go.opentelemetry.io/collector/pdata/plog"
+	v1 "k8s.io/api/core/v1"
+)
+
+const (
+	// This action will be ignored and not written in any of the facets, since
+	// we return map[string]any
+	PodCountsActionKey               = "pod_counts"
+	PodContainerRestartsAttributeKey = "restarts"
+	PodTotalContainersAttributeKey   = "total_containers"
+	PodReadyContainersAttributeKey   = "ready_containers"
+)
+
+// This action computes various facets for Pod by aggregating "status" values
+// across all containers of a Pod.
+//
+// We compute more facets into a single action to avoid iterating over the
+// same slice multiple times in different actions.
+var PodContainersCountsAction = K8sEventProcessorAction{
+	Key:                    PodCountsActionKey,
+	Transform:              getPodCounts,
+	FilterFn:               filterPodEvents,
+	ComputesMultipleFacets: true,
+}
+
+func getPodCounts(objLog plog.LogRecord) any {
+	var p v1.Pod
+	err := json.Unmarshal([]byte(objLog.Body().AsString()), &p)
+	if err != nil {
+		return "Unknown"
+	}
+	// we use int32 since containerStatuses.restartCount is int32
+	var restartsCount int32
+	// We don't need to use a hash set on the container ID for these two facets,
+	// since the containerStatuses contain one entry per container.
+	var readyContainers int64
+	var allContainers int64
+
+	for _, stat := range p.Status.ContainerStatuses {
+		restartsCount += stat.RestartCount
+		allContainers++
+		if stat.Ready {
+			readyContainers++
+		}
+	}
+
+	// Returning map[string]any will make the processor add its elements as
+	// separate facets, rather than adding the whole map under the key of this action
+	return map[string]any{
+		PodContainerRestartsAttributeKey: int64(restartsCount),
+		PodTotalContainersAttributeKey:   allContainers,
+		PodReadyContainersAttributeKey:   readyContainers,
+	}
+}

components/processors/observek8sattributesprocessor/podstatus.go

@@ -14,18 +14,10 @@ const (
 	nodeUnreachablePodReason = "NodeLost"
 )
 
-type OTELKubernetesEvent struct {
-	Object v1.Pod `json:"object"`
-}
-
 var PodStatusAction = K8sEventProcessorAction{
-	Key:      PodStatusAttributeKey,
-	ValueFn:  getPodStatus,
-	FilterFn: filterPodEvents,
-}
-
-func filterPodEvents(event K8sEvent) bool {
-	return event.Kind == "Pod"
+	Key:       PodStatusAttributeKey,
+	Transform: getPodStatus,
+	FilterFn:  filterPodEvents,
 }
 
 // Generates the Pod "status" facet. Assumes that objLog is a log from a Pod event.

components/processors/observek8sattributesprocessor/processor.go

@@ -3,6 +3,7 @@ package observek8sattributesprocessor
 import (
 	"context"
 	"encoding/json"
+	"errors"
 
 	"go.opentelemetry.io/collector/component"
 	"go.opentelemetry.io/collector/pdata/pcommon"
@@ -21,18 +22,21 @@ type K8sEventsProcessor struct {
 	actions []K8sEventProcessorAction
 }
 
+type transformFn func(plog.LogRecord) any
+
 type K8sEventProcessorAction struct {
-	Key      string
-	ValueFn  func(plog.LogRecord) any
-	FilterFn func(K8sEvent) bool
+	Key                    string
+	Transform              transformFn
+	FilterFn               func(K8sEvent) bool
+	ComputesMultipleFacets bool
 }
 
 func newK8sEventsProcessor(logger *zap.Logger, cfg component.Config) *K8sEventsProcessor {
 	return &K8sEventsProcessor{
 		cfg:    cfg,
 		logger: logger,
 		actions: []K8sEventProcessorAction{
-			PodStatusAction, NodeStatusAction, NodeRolesAction,
+			PodStatusAction, NodeStatusAction, NodeRolesAction, PodContainersCountsAction,
 		},
 	}
 }
@@ -79,49 +83,103 @@ func (kep *K8sEventsProcessor) processLogs(_ context.Context, logs plog.Logs) (p
 						facetsMap = facets.Map()
 					} else {
 						facetsMap = transformMap.PutEmptyMap("facets")
+						// Make sure we have capacity for at least as many actions as we have defined
+						// Actions could generate more than one facet, that's taken care of afterwards.
+						facetsMap.EnsureCapacity(len(kep.actions))
 					}
 
-					// This is where the custom processor actually computes the value
-					value := action.ValueFn(lr)
-
-					// TODO [eg] we probably want to make this more modular. For
-					// now using FromRaw for complex types is fine, since we
-					// don't plan to generate arbitrarily complex/nested facets.
-					// For some time coming we will produce facets that are at
-					// most slices of simple types of map of simple types,
-					// nothing beyond that.
-					switch typed := value.(type) {
-					case string:
-						facetsMap.PutStr(action.Key, typed)
-					case int64:
-						facetsMap.PutInt(action.Key, typed)
-					case bool:
-						facetsMap.PutBool(action.Key, typed)
-					case float64:
-						facetsMap.PutDouble(action.Key, typed)
-					case []string:
-						// []string can't fallback to using FromRaw([]any), as
-						// the default implementation of FromRaw is not smart
-						// enough to understand that the slice contains all
-						// string, and inserts them as bytes, instead
-						slc := facetsMap.PutEmptySlice(action.Key)
-						slc.EnsureCapacity(len(typed))
-						for _, str := range typed {
-							slc.AppendEmpty().SetStr(str)
+					// This is where the custom processor actually computes the transformed value(s)
+					value := action.Transform(lr)
+
+					// If the action returns multiple facets, it means that:
+					//  - it must generate a value of type map[string]any
+					//  - we put each individual element of such map into the facets
+					//  - we ignore the key associated with the action itself, since each facet has its own key in the map
+					if action.ComputesMultipleFacets {
+						if facets, ok := value.(map[string]any); ok {
+							// Make sure we have enough capacity for the new facets
+							facetsMap.EnsureCapacity(facetsMap.Len() + len(facets))
+							for key, val := range facets {
+								if err := mapPut(facetsMap, key, val); err != nil {
+									kep.logger.Error(err.Error())
+								}
+							}
+						} else {
+							kep.logger.Error("Action %s computes multiple facets and the returned value is not of type map[string]any", zap.String("name", action.Key))
 						}
-					case map[string]string:
-						// Same reasoning as []string
-						mp := facetsMap.PutEmptyMap(action.Key)
-						mp.EnsureCapacity(len(typed))
-						for k, v := range typed {
-							mp.PutStr(k, v)
+					} else {
+						// Otherwise, whatever is generated by the facet is the value of such facet, and the facet's key is the key of the action itself.
+						// This is true even if the action generates map[string]any when ComputesMultipleFacets == false!
+						// This way we allow computing facets typed map[string]any
+						if err := mapPut(facetsMap, action.Key, value); err != nil {
+							kep.logger.Error(err.Error())
 						}
-					default:
-						kep.logger.Error("sending the generated facet to Observe in bytes since no custom serialization logic is implemented", zap.Error(err))
 					}
 				}
 			}
 		}
 	}
 	return logs, nil
 }
+
+func slicePut(theSlice pcommon.Slice, value any) error {
+	elem := theSlice.AppendEmpty()
+	switch typed := value.(type) {
+	case string:
+		elem.SetStr(typed)
+	case int64:
+		elem.SetInt(typed)
+	case bool:
+		elem.SetBool(typed)
+	case float64:
+		elem.SetDouble(typed)
+	// Let's not complicate things and avoid putting maps/slices into slices There's
+	// gotta be an easier way to model the processor's output to avoid this
+	default:
+		return errors.New("sending the generated facet to Observe in bytes since no custom serialization logic is implemented")
+	}
+
+	return nil
+}
+
+// puts "anything" into a map, with some assumptions and intentional
+// limitations:
+//
+//   - No nested slices: can only put "base types" inside slices (although
+//     elements of a slice can be of different [base] types).
+//
+//   - Not all "base types" are covered. For instance, numbers are only int64 and float64.
+//
+//   - No maps with keys of arbitrary types: only string
+func mapPut(theMap pcommon.Map, key string, value any) error {
+	switch typed := value.(type) {
+	case string:
+		theMap.PutStr(key, typed)
+	case int64:
+		theMap.PutInt(key, typed)
+	case bool:
+		theMap.PutBool(key, typed)
+	case float64:
+		theMap.PutDouble(key, typed)
+	case []any:
+		slc := theMap.PutEmptySlice(key)
+		slc.EnsureCapacity(len(typed))
+		for _, elem := range typed {
+			slicePut(slc, elem)
+		}
+	case map[string]any:
+		// This is potentially arbitrarily recursive. We don't care about
+		// checking the nesting level since we will never need to define
+		// processors with more than one nesting level
+		new := theMap.PutEmptyMap(key)
+		new.EnsureCapacity(len(typed))
+		for k, v := range typed {
+			mapPut(new, k, v)
+		}
+	default:
+		return errors.New("sending the generated facet to Observe in bytes since no custom serialization logic is implemented")
+	}
+
+	return nil
+
+}

components/processors/observek8sattributesprocessor/processor_test.go

@@ -45,7 +45,6 @@ func TestK8sEventsProcessor(t *testing.T) {
 			name: "noObserveTransformAttributes",
 			inLogs: createResourceLogs(
 				logWithResource{
-					logName:          "noObserveTransformAttributes",
 					testBodyFilepath: "./testdata/podObjectEvent.json",
 				},
 			),
@@ -57,7 +56,6 @@ func TestK8sEventsProcessor(t *testing.T) {
 			name: "existingObserveTransformAttributes",
 			inLogs: createResourceLogs(
 				logWithResource{
-					logName:          "existingObserveTransformAttributes",
 					testBodyFilepath: "./testdata/podObjectEvent.json",
 					recordAttributes: map[string]any{
 						"observe_transform": map[string]interface{}{
@@ -100,6 +98,19 @@ func TestK8sEventsProcessor(t *testing.T) {
 				{"observe_transform.facets.roles", []any{"anotherRole!", "control-plane"}},
 			},
 		},
+		{
+			name: "Pod container counts",
+			inLogs: createResourceLogs(
+				logWithResource{
+					testBodyFilepath: "./testdata/podObjectEvent.json",
+				},
+			),
+			expectedResults: []queryWithResult{
+				{"observe_transform.facets.restarts", int64(5)},
+				{"observe_transform.facets.total_containers", int64(4)},
+				{"observe_transform.facets.ready_containers", int64(3)},
+			},
+		},
 	} {
 		t.Run(test.name, func(t *testing.T) {
 			kep := newK8sEventsProcessor(zap.NewNop(), &Config{})