add flag for reading pods from kube-apiserver cache

flags/flags.go

@@ -279,6 +279,7 @@ type FlagsProfiling struct {
 type FlagsMetadata struct {
 	ExternalLabels             map[string]string `help:"Label(s) to attach to all profiles."`
 	ContainerRuntimeSocketPath string            `help:"The filesystem path to the container runtimes socket. Leave this empty to use the defaults."`
+	ListFromCache              bool              `default:"true" help:"listing pods from apiserver cache instead of etcd"`
 
 	DisableCaching       bool `default:"false" help:"[deprecated] Disable caching of metadata."`
 	EnableProcessCmdline bool `default:"false" help:"[deprecated] Add /proc/[pid]/cmdline as a label, which may expose sensitive information like secrets in profiling data."`

main.go

@@ -402,6 +402,7 @@ func mainWithExitCode() flags.ExitCode {
 		f.Debuginfo.UploadQueueSize,
 		f.Debuginfo.TempDir,
 		f.Node,
+		f.Metadata.ListFromCache,
 		relabelConfigs,
 		buildInfo.VcsRevision,
 		reg,

reporter/metadata/containermetadata.go

@@ -112,6 +112,9 @@ type containerMetadataProvider struct {
 	// the kubernetes node name used to retrieve the pod information.
 	nodeName string
 
+	// used to list pods.
+	listFromCache bool
+
 	// containerIDCache stores per process container ID information.
 	containerIDCache *lru.SyncedLRU[libpf.PID, containerIDEntry]
 
@@ -162,7 +165,7 @@ type containerIDEntry struct {
 }
 
 // NewContainerMetadataProvider creates a new container metadata provider.
-func NewContainerMetadataProvider(ctx context.Context, nodeName string) (MetadataProvider, error) {
+func NewContainerMetadataProvider(ctx context.Context, nodeName string, listFromCache bool) (MetadataProvider, error) {
 	containerIDCache, err := lru.NewSynced[libpf.PID, containerIDEntry](
 		containerIDCacheSize, libpf.PID.Hash32)
 	if err != nil {
@@ -175,6 +178,7 @@ func NewContainerMetadataProvider(ctx context.Context, nodeName string) (Metadat
 		dockerClient:     getDockerClient(),
 		containerdClient: getContainerdClient(),
 		nodeName:         nodeName,
+		listFromCache:    listFromCache,
 	}
 
 	p.deferredPID, err = lru.NewSynced[libpf.PID, libpf.Void](deferredLRUSize,
@@ -601,9 +605,13 @@ func (p *containerMetadataProvider) getKubernetesPodMetadata(pidContainerID stri
 	log.Debugf("Get kubernetes pod metadata for container id %v", pidContainerID)
 
 	p.kubernetesClientQueryCount.Add(1)
-	pods, err := p.kubeClientSet.CoreV1().Pods("").List(context.TODO(), v1.ListOptions{
+	listOption := v1.ListOptions{
 		FieldSelector: "spec.nodeName=" + p.nodeName,
-	})
+	}
+	if p.listFromCache {
+		listOption.ResourceVersion = "0"
+	}
+	pods, err := p.kubeClientSet.CoreV1().Pods("").List(context.TODO(), listOption)
 	if err != nil {
 		return nil, fmt.Errorf("failed to retrieve kubernetes pods, %v", err)
 	}

reporter/parca_reporter.go

@@ -573,6 +573,7 @@ func New(
 	uploaderQueueSize uint32,
 	cacheDir string,
 	nodeName string,
+	listFromCache bool,
 	relabelConfigs []*relabel.Config,
 	agentRevision string,
 	reg prometheus.Registerer,
@@ -613,7 +614,7 @@ func New(
 		}
 	}
 
-	cmp, err := metadata.NewContainerMetadataProvider(context.TODO(), nodeName)
+	cmp, err := metadata.NewContainerMetadataProvider(context.TODO(), nodeName, listFromCache)
 	if err != nil {
 		return nil, err
 	}