Add podresourcesstore in awscontainerinsightreceiver

receiver/awscontainerinsightreceiver/go.mod

@@ -138,6 +138,7 @@ require (
 	gotest.tools/v3 v3.0.3 // indirect
 	k8s.io/klog/v2 v2.90.1 // indirect
 	k8s.io/kube-openapi v0.0.0-20230501164219-8b0f38b5fd1f // indirect
+	k8s.io/kubelet v0.27.3 // indirect
 	k8s.io/utils v0.0.0-20230209194617-a36077c30491 // indirect
 	sigs.k8s.io/json v0.0.0-20221116044647-bc3834ca7abd // indirect
 	sigs.k8s.io/structured-merge-diff/v4 v4.2.3 // indirect

receiver/awscontainerinsightreceiver/internal/stores/kubeletutil/podresourcesclient.go

@@ -0,0 +1,87 @@
+// Copyright The OpenTelemetry Authors
+// SPDX-License-Identifier: Apache-2.0
+
+package kubeletutil // import "github.com/open-telemetry/opentelemetry-collector-contrib/receiver/awscontainerinsightreceiver/internal/stores/kubeletutil"
+
+import (
+	"context"
+	"fmt"
+	"google.golang.org/grpc"
+	"net"
+	"os"
+	"time"
+
+	"google.golang.org/grpc/credentials/insecure"
+	podresourcesapi "k8s.io/kubelet/pkg/apis/podresources/v1"
+)
+
+const (
+	socketPath        = "/var/lib/kubelet/pod-resources/kubelet.sock"
+	connectionTimeout = 10 * time.Second
+)
+
+type PodResourcesClient struct {
+	delegateClient podresourcesapi.PodResourcesListerClient
+	conn           *grpc.ClientConn
+}
+
+func NewPodResourcesClient() (*PodResourcesClient, error) {
+	podResourcesClient := &PodResourcesClient{}
+
+	conn, err := podResourcesClient.connectToServer(socketPath)
+	podResourcesClient.conn = conn
+	if err != nil {
+		return nil, fmt.Errorf("failed to connect to server: %w", err)
+	}
+
+	podResourcesClient.delegateClient = podresourcesapi.NewPodResourcesListerClient(conn)
+
+	return podResourcesClient, nil
+}
+
+func (p *PodResourcesClient) connectToServer(socket string) (*grpc.ClientConn, error) {
+	_, err := os.Stat(socket)
+	if os.IsNotExist(err) {
+		return nil, fmt.Errorf("socket path does not exist: %s", socket)
+	} else if err != nil {
+		return nil, fmt.Errorf("failed to check socket path: %w", err)
+	}
+
+	ctx, cancel := context.WithTimeout(context.Background(), connectionTimeout)
+	defer cancel()
+
+	conn, err := grpc.DialContext(ctx,
+		socket,
+		grpc.WithTransportCredentials(insecure.NewCredentials()),
+		grpc.WithBlock(),
+		grpc.WithContextDialer(func(ctx context.Context, addr string) (net.Conn, error) {
+			d := net.Dialer{}
+			return d.DialContext(ctx, "unix", addr)
+		}),
+	)
+
+	if err != nil {
+		return nil, fmt.Errorf("failure connecting to '%s': %w", socket, err)
+	}
+
+	return conn, nil
+}
+
+func (p *PodResourcesClient) ListPods() (*podresourcesapi.ListPodResourcesResponse, error) {
+	ctx, cancel := context.WithTimeout(context.Background(), connectionTimeout)
+	defer cancel()
+
+	resp, err := p.delegateClient.List(ctx, &podresourcesapi.ListPodResourcesRequest{})
+	if err != nil {
+		return nil, fmt.Errorf("failure getting pod resources: %w", err)
+	}
+
+	return resp, nil
+}
+
+func (p *PodResourcesClient) Shutdown() {
+	err := p.conn.Close()
+	if err != nil {
+		return
+	}
+}

receiver/awscontainerinsightreceiver/internal/stores/podresourcesstore.go

@@ -0,0 +1,166 @@
+// Copyright The OpenTelemetry Authors
+// SPDX-License-Identifier: Apache-2.0
+
+package stores // import "github.com/open-telemetry/opentelemetry-collector-contrib/receiver/awscontainerinsightreceiver/internal/stores"
+
+import (
+	"context"
+	"fmt"
+	"sync"
+	"time"
+
+	"go.uber.org/zap"
+	v1 "k8s.io/kubelet/pkg/apis/podresources/v1"
+
+	"github.com/open-telemetry/opentelemetry-collector-contrib/receiver/awscontainerinsightreceiver/internal/stores/kubeletutil"
+)
+
+const (
+	taskTimeout = 10 * time.Second
+)
+
+var (
+	instance *PodResourcesStore
+	once     sync.Once
+)
+
+type ContainerInfo struct {
+	PodName       string
+	ContainerName string
+	Namespace     string
+}
+
+type ResourceInfo struct {
+	ResourceName string
+	DeviceID     string
+}
+
+type PodResourcesClientInterface interface {
+	ListPods() (*v1.ListPodResourcesResponse, error)
+	Shutdown()
+}
+
+type PodResourcesStore struct {
+	containerInfoToResourcesMap map[ContainerInfo][]ResourceInfo
+	resourceToPodContainerMap   map[ResourceInfo]ContainerInfo
+	resourceNameSet             map[string]struct{}
+	lastRefreshed               time.Time
+	ctx                         context.Context
+	cancel                      context.CancelFunc
+	logger                      *zap.Logger
+	podResourcesClient          PodResourcesClientInterface
+}
+
+func NewPodResourcesStore(logger *zap.Logger) *PodResourcesStore {
+	once.Do(func() {
+		podResourcesClient, _ := kubeletutil.NewPodResourcesClient()
+		ctx, cancel := context.WithCancel(context.Background())
+		instance = &PodResourcesStore{
+			containerInfoToResourcesMap: make(map[ContainerInfo][]ResourceInfo),
+			resourceToPodContainerMap:   make(map[ResourceInfo]ContainerInfo),
+			resourceNameSet:             make(map[string]struct{}),
+			lastRefreshed:               time.Now(),
+			ctx:                         ctx,
+			cancel:                      cancel,
+			logger:                      logger,
+			podResourcesClient:          podResourcesClient,
+		}
+
+		go func() {
+			refreshTicker := time.NewTicker(time.Second)
+			for {
+				select {
+				case <-refreshTicker.C:
+					instance.refreshTick()
+				case <-instance.ctx.Done():
+					refreshTicker.Stop()
+					return
+				}
+			}
+		}()
+	})
+	return instance
+}
+
+func (p *PodResourcesStore) refreshTick() {
+	now := time.Now()
+	if now.Sub(p.lastRefreshed) >= taskTimeout {
+		p.refresh()
+		p.lastRefreshed = now
+	}
+}
+
+func (p *PodResourcesStore) refresh() {
+	doRefresh := func() {
+		p.updateMaps()
+	}
+
+	refreshWithTimeout(p.ctx, doRefresh, taskTimeout)
+}
+
+func (p *PodResourcesStore) updateMaps() {
+	p.containerInfoToResourcesMap = make(map[ContainerInfo][]ResourceInfo)
+	p.resourceToPodContainerMap = make(map[ResourceInfo]ContainerInfo)
+
+	if len(p.resourceNameSet) == 0 {
+		p.logger.Warn("No resource names allowlisted thus skipping updating of maps.")
+		return
+	}
+
+	devicePods, err := p.podResourcesClient.ListPods()
+
+	if err != nil {
+		p.logger.Error(fmt.Sprintf("Error getting pod resources: %v", err))
+		return
+	}
+
+	for _, pod := range devicePods.GetPodResources() {
+		for _, container := range pod.GetContainers() {
+			for _, device := range container.GetDevices() {
+
+				containerInfo := ContainerInfo{
+					PodName:       pod.GetName(),
+					Namespace:     pod.GetNamespace(),
+					ContainerName: container.GetName(),
+				}
+
+				for _, deviceID := range device.GetDeviceIds() {
+					resourceInfo := ResourceInfo{
+						ResourceName: device.GetResourceName(),
+						DeviceID:     deviceID,
+					}
+					_, found := p.resourceNameSet[resourceInfo.ResourceName]
+					if found {
+						p.containerInfoToResourcesMap[containerInfo] = append(p.containerInfoToResourcesMap[containerInfo], resourceInfo)
+						p.resourceToPodContainerMap[resourceInfo] = containerInfo
+					}
+				}
+			}
+		}
+	}
+}
+
+func (p *PodResourcesStore) GetContainerInfo(deviceID string, resourceName string) *ContainerInfo {
+	key := ResourceInfo{DeviceID: deviceID, ResourceName: resourceName}
+	if containerInfo, ok := p.resourceToPodContainerMap[key]; ok {
+		return &containerInfo
+	}
+	return nil
+}
+
+func (p *PodResourcesStore) GetResourcesInfo(podName string, containerName string, namespace string) *[]ResourceInfo {
+	key := ContainerInfo{PodName: podName, ContainerName: containerName, Namespace: namespace}
+	if resourceInfo, ok := p.containerInfoToResourcesMap[key]; ok {
+		return &resourceInfo
+	}
+	return nil
+}
+
+func (p *PodResourcesStore) AddResourceName(resourceName string) {
+	p.resourceNameSet[resourceName] = struct{}{}
+}
+
+func (p *PodResourcesStore) Shutdown() {
+	p.cancel()
+	p.podResourcesClient.Shutdown()
+}