fix(objectcache): synchronize container status fields and add cloud metadata annotations

pkg/containerprofilemanager/v1/lifecycle_control_test.go

@@ -3,11 +3,21 @@ package containerprofilemanager
 import (
 	"testing"
 	"testing/synctest"
+	"time"
 
 	containercollection "github.com/inspektor-gadget/inspektor-gadget/pkg/container-collection"
+	eventtypes "github.com/inspektor-gadget/inspektor-gadget/pkg/types"
+	containerinstance "github.com/kubescape/k8s-interface/instanceidhandler/v1/containerinstance"
 	"github.com/kubescape/node-agent/pkg/config"
+	"github.com/kubescape/node-agent/pkg/containerprofilemanager/v1/queue"
+	"github.com/kubescape/node-agent/pkg/dnsmanager"
+	"github.com/kubescape/node-agent/pkg/k8sclient"
 	"github.com/kubescape/node-agent/pkg/objectcache"
+	"github.com/kubescape/node-agent/pkg/otelsetup"
+	"github.com/kubescape/node-agent/pkg/seccompmanager"
+	"github.com/kubescape/node-agent/pkg/storage"
 	"github.com/stretchr/testify/require"
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 )
 
 func TestLifecycleSignalRejectionUnblocksFullControlChannel(t *testing.T) {
@@ -86,3 +96,67 @@ func TestProfileSplitCoalescesFullControlChannel(t *testing.T) {
 		require.Zero(t, data.size.Load())
 	})
 }
+
+func TestTickerDoesNotRegressTerminalStatus(t *testing.T) {
+	container := &containercollection.Container{
+		Runtime: containercollection.RuntimeMetadata{
+			BasicRuntimeMetadata: eventtypes.BasicRuntimeMetadata{
+				ContainerID:   "test-container",
+				ContainerName: "test-container",
+				ContainerPID:  2,
+			},
+		},
+	}
+	id := container.Runtime.ContainerID
+
+	tickerChan := make(chan time.Time)
+	watched := &objectcache.WatchedContainerData{
+		ContainerID: id, SyncChannel: make(chan error, 1),
+		UpdateDataTicker: &time.Ticker{C: tickerChan}, AckChan: make(chan struct{}, 1),
+		InstanceID: &containerinstance.InstanceID{
+			ApiVersion: "apps/v1", Namespace: "default", Kind: "Pod", Name: "pod", ContainerName: "test-container",
+		},
+		ContainerType: objectcache.Container,
+		ContainerInfos: map[objectcache.ContainerType][]objectcache.ContainerInfo{
+			objectcache.Container: {{Name: "test-container"}},
+		},
+		ParentWorkloadSelector: &metav1.LabelSelector{},
+	}
+	watched.SetStatus(objectcache.WatchedContainerStatusCompleted)
+	ready := make(chan struct{})
+	close(ready)
+	entry := &ContainerEntry{ready: ready, data: &containerData{watchedContainerData: watched, queueErrors: make(chan error, 1), monitorDone: make(chan struct{})}}
+	tempDir := t.TempDir()
+	t.Setenv("QUEUE_DIR", tempDir)
+	qData, err := queue.NewQueueData(t.Context(), &storage.StorageHttpClientMock{}, queue.QueueConfig{
+		QueueDir:     tempDir,
+		MaxQueueSize: 10,
+	})
+	require.NoError(t, err)
+	defer qData.Close()
+
+	cpm := &ContainerProfileManager{
+		containers:        map[string]*ContainerEntry{id: entry},
+		lifecycleTracker:  otelsetup.NewProfileLifecycleTracker(),
+		seccompManager:    &seccompmanager.SeccompManagerMock{},
+		storageClient:     &storage.StorageHttpClientMock{},
+		k8sClient:         &k8sclient.K8sClientMock{},
+		dnsResolverClient: &dnsmanager.DNSManagerMock{},
+		queueData:         qData,
+	}
+
+	monitoringDone := make(chan error, 1)
+	go func() {
+		monitoringDone <- cpm.monitorContainer(container, watched, entry.data)
+	}()
+
+	// Fire a tick while status is Completed
+	tickerChan <- time.Now()
+
+	// Then send termination error to stop monitorContainer
+	watched.SyncChannel <- ContainerHasTerminatedError
+
+	require.Equal(t, ContainerHasTerminatedError, <-monitoringDone)
+	// Status must have remained Completed, not regressed to Ready
+	require.Equal(t, objectcache.WatchedContainerStatusCompleted, watched.GetStatus())
+}

pkg/containerprofilemanager/v1/monitoring.go

@@ -98,8 +98,12 @@ func (cpm *ContainerProfileManager) monitorContainer(container *containercollect
 					// keep monitoring after reaching Completed).
 					cpm.notifyCompleted(watchedContainer.ContainerID)
 				}
-			} else {
-				watchedContainer.SetStatus(objectcache.WatchedContainerStatusReady)
+			} else if !watchedContainer.SetReadyUnlessTerminal() {
+				// Container already reached a terminal state (e.g. Completed or Failed
+				// set by concurrent deleteContainerWithReason). Do not overwrite it
+				// with Ready or save an un-flushed partial profile; the upcoming signal
+				// on SyncChannel will handle final flush and save.
+				continue
 			}
 			if err := cpm.saveProfile(watchedContainer, container, false); err != nil {
 				if handledErr := cpm.handleSaveProfileError(err, watchedContainer, container, data); handledErr != nil {

pkg/objectcache/cloud_metadata_test.go

@@ -0,0 +1,32 @@
+package objectcache
+
+import (
+	"testing"
+
+	"github.com/armosec/armoapi-go/armotypes"
+	helpersv1 "github.com/kubescape/k8s-interface/instanceidhandler/v1/helpers"
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+)
+
+func TestCloudMetadataAnnotations(t *testing.T) {
+	cloud := &armotypes.CloudMetadata{
+		HostType:    "gcp",
+		MachineID:   "m-1",
+		ClusterName: "kind-kind",
+		AccountID:   "acct",
+		Region:      "us-central1",
+	}
+
+	ann := CloudMetadataAnnotations(cloud)
+	require.Equal(t, map[string]string{
+		helpersv1.HostTypeMetadataKey:               "gcp",
+		helpersv1.HostIDMetadataKey:                 "m-1",
+		helpersv1.ClusterMetadataKey:                "kind-kind",
+		helpersv1.CloudAccountIdentifierMetadataKey: "acct",
+		helpersv1.RegionMetadataKey:                 "us-central1",
+	}, ann)
+
+	assert.Equal(t, map[string]string{helpersv1.HostTypeMetadataKey: "gcp"}, CloudMetadataAnnotations(&armotypes.CloudMetadata{HostType: "gcp"}), "empty fields are not written")
+	assert.Empty(t, CloudMetadataAnnotations(nil))
+}

pkg/objectcache/shared_container_data.go

@@ -5,6 +5,7 @@ import (
 	"iter"
 	"maps"
 	"strings"
+	"sync"
 	"time"
 
 	"github.com/armosec/armoapi-go/armotypes"
@@ -75,6 +76,15 @@ type WatchedContainerData struct {
 	ContainerInfos          map[ContainerType][]ContainerInfo
 	NsMntId                 uint64
 	InitialDelayExpired     bool
+	// statusMu guards status and completionStatus. Both are read and written
+	// from several goroutines at once — the monitoring loop sets them on every
+	// tick and at termination, while deleteContainer sets them from the
+	// container callback — and they are string-typed, so an unsynchronised
+	// write is not a single store. A torn read yields a status that matches
+	// nothing, and this is the field that decides whether a profile is stamped
+	// completed. Caught by the race detector, intermittently: two of eight
+	// runs.
+	statusMu                sync.RWMutex
 	status                  WatchedContainerStatus
 	completionStatus        WatchedContainerCompletionStatus
 	ParentWorkloadSelector  *metav1.LabelSelector
@@ -102,6 +112,25 @@ func formatDuration(d time.Duration) string {
 	return s
 }
 
+func CloudMetadataAnnotations(cloudMetadata *armotypes.CloudMetadata) map[string]string {
+	out := map[string]string{}
+	if cloudMetadata == nil {
+		return out
+	}
+	for k, v := range map[string]string{
+		helpersv1.HostTypeMetadataKey:               string(cloudMetadata.HostType),
+		helpersv1.HostIDMetadataKey:                 cloudMetadata.MachineID,
+		helpersv1.ClusterMetadataKey:                cloudMetadata.ClusterName,
+		helpersv1.CloudAccountIdentifierMetadataKey: cloudMetadata.AccountID,
+		helpersv1.RegionMetadataKey:                 cloudMetadata.Region,
+	} {
+		if v != "" {
+			out[k] = v
+		}
+	}
+	return out
+}
+
 func GetLabels(cloudMetadata *armotypes.CloudMetadata, watchedContainer *WatchedContainerData, stripContainer bool) map[string]string {
 	labels := watchedContainer.InstanceID.GetLabels()
 	labels[helpersv1.LearningPeriodMetadataKey] = formatDuration(watchedContainer.LearningPeriod)
@@ -142,24 +171,62 @@ func GetLabels(cloudMetadata *armotypes.CloudMetadata, watchedContainer *Watched
 	return labels
 }
 
+// IsTerminal reports whether the status represents a terminal lifecycle state.
+func (s WatchedContainerStatus) IsTerminal() bool {
+	switch s {
+	case WatchedContainerStatusCompleted,
+		WatchedContainerStatusFailed,
+		WatchedContainerStatusMissingRuntime,
+		WatchedContainerStatusTooLarge,
+		WatchedContainerStatusRejected:
+		return true
+	default:
+		return false
+	}
+}
+
 func (watchedContainer *WatchedContainerData) GetStatus() WatchedContainerStatus {
+	watchedContainer.statusMu.RLock()
+	defer watchedContainer.statusMu.RUnlock()
 	return watchedContainer.status
 }
 
+// IsTerminal reports whether the container has reached a terminal status.
+func (watchedContainer *WatchedContainerData) IsTerminal() bool {
+	watchedContainer.statusMu.RLock()
+	defer watchedContainer.statusMu.RUnlock()
+	return watchedContainer.status.IsTerminal()
+}
+
 func (watchedContainer *WatchedContainerData) GetCompletionStatus() WatchedContainerCompletionStatus {
+	watchedContainer.statusMu.RLock()
+	defer watchedContainer.statusMu.RUnlock()
 	return watchedContainer.completionStatus
 }
 
 func (watchedContainer *WatchedContainerData) SetStatus(newStatus WatchedContainerStatus) {
-	if newStatus != watchedContainer.status {
-		watchedContainer.status = newStatus
+	watchedContainer.statusMu.Lock()
+	defer watchedContainer.statusMu.Unlock()
+	watchedContainer.status = newStatus
+}
+
+// SetReadyUnlessTerminal sets the container status to Ready unless it has already
+// reached a terminal state (Completed, Failed, MissingRuntime, TooLarge, Rejected).
+// It reports whether the status was updated to Ready.
+func (watchedContainer *WatchedContainerData) SetReadyUnlessTerminal() bool {
+	watchedContainer.statusMu.Lock()
+	defer watchedContainer.statusMu.Unlock()
+	if watchedContainer.status.IsTerminal() {
+		return false
 	}
+	watchedContainer.status = WatchedContainerStatusReady
+	return true
 }
 
 func (watchedContainer *WatchedContainerData) SetCompletionStatus(newStatus WatchedContainerCompletionStatus) {
-	if newStatus != watchedContainer.completionStatus {
-		watchedContainer.completionStatus = newStatus
-	}
+	watchedContainer.statusMu.Lock()
+	defer watchedContainer.statusMu.Unlock()
+	watchedContainer.completionStatus = newStatus
 }
 
 func (watchedContainer *WatchedContainerData) SetContainerInfo(wl workloadinterface.IWorkload, containerName string) error {

pkg/objectcache/shared_container_data_test.go

@@ -1,6 +1,7 @@
 package objectcache
 
 import (
+	"sync"
 	"testing"
 	"time"
 
@@ -115,3 +116,100 @@ func Test_formatDuration(t *testing.T) {
 		})
 	}
 }
+
+func TestWatchedContainerData_StatusConcurrency(t *testing.T) {
+	data := &WatchedContainerData{}
+	var wg sync.WaitGroup
+
+	for i := 0; i < 50; i++ {
+		wg.Add(4)
+		go func() {
+			defer wg.Done()
+			data.SetStatus(WatchedContainerStatusReady)
+			_ = data.GetStatus()
+		}()
+		go func() {
+			defer wg.Done()
+			data.SetStatus(WatchedContainerStatusCompleted)
+			_ = data.GetStatus()
+		}()
+		go func() {
+			defer wg.Done()
+			data.SetCompletionStatus(WatchedContainerCompletionStatusPartial)
+			_ = data.GetCompletionStatus()
+		}()
+		go func() {
+			defer wg.Done()
+			data.SetCompletionStatus(WatchedContainerCompletionStatusFull)
+			_ = data.GetCompletionStatus()
+		}()
+	}
+
+	wg.Wait()
+}
+
+func TestWatchedContainerData_SetReadyUnlessTerminal(t *testing.T) {
+	// Non-terminal states can transition to Ready
+	data := &WatchedContainerData{}
+	assert.False(t, data.IsTerminal())
+	assert.True(t, data.SetReadyUnlessTerminal())
+	assert.Equal(t, WatchedContainerStatusReady, data.GetStatus())
+
+	data.SetStatus(WatchedContainerStatusInitializing)
+	assert.False(t, data.IsTerminal())
+	assert.True(t, data.SetReadyUnlessTerminal())
+	assert.Equal(t, WatchedContainerStatusReady, data.GetStatus())
+
+	// Terminal states cannot transition to Ready
+	terminalStates := []WatchedContainerStatus{
+		WatchedContainerStatusCompleted,
+		WatchedContainerStatusFailed,
+		WatchedContainerStatusMissingRuntime,
+		WatchedContainerStatusTooLarge,
+		WatchedContainerStatusRejected,
+	}
+
+	for _, state := range terminalStates {
+		t.Run(string(state), func(t *testing.T) {
+			assert.True(t, state.IsTerminal())
+			data.SetStatus(state)
+			assert.True(t, data.IsTerminal())
+			assert.False(t, data.SetReadyUnlessTerminal())
+			assert.Equal(t, state, data.GetStatus())
+		})
+	}
+}
+
+func TestWatchedContainerData_SetReadyUnlessTerminal_ConcurrentTermination(t *testing.T) {
+	for iter := 0; iter < 100; iter++ {
+		data := &WatchedContainerData{}
+		data.SetStatus(WatchedContainerStatusReady)
+
+		var wg sync.WaitGroup
+		start := make(chan struct{})
+
+		// Ticker simulator
+		wg.Add(1)
+		go func() {
+			defer wg.Done()
+			<-start
+			for i := 0; i < 50; i++ {
+				data.SetReadyUnlessTerminal()
+			}
+		}()
+
+		// Deletion simulator
+		wg.Add(1)
+		go func() {
+			defer wg.Done()
+			<-start
+			data.SetStatus(WatchedContainerStatusCompleted)
+		}()
+
+		close(start)
+		wg.Wait()
+
+		// Once completed, it must not regress to Ready
+		assert.Equal(t, WatchedContainerStatusCompleted, data.GetStatus())
+	}
+}