diff --git a/pkg/containerprofilemanager/v1/lifecycle_control_test.go b/pkg/containerprofilemanager/v1/lifecycle_control_test.go index 9528f74429..eb82afde02 100644 --- a/pkg/containerprofilemanager/v1/lifecycle_control_test.go +++ b/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()) +} diff --git a/pkg/containerprofilemanager/v1/monitoring.go b/pkg/containerprofilemanager/v1/monitoring.go index 96ecbb78ee..d51babb4b9 100644 --- a/pkg/containerprofilemanager/v1/monitoring.go +++ b/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 { diff --git a/pkg/objectcache/cloud_metadata_test.go b/pkg/objectcache/cloud_metadata_test.go new file mode 100644 index 0000000000..1ba0435d96 --- /dev/null +++ b/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)) +} diff --git a/pkg/objectcache/shared_container_data.go b/pkg/objectcache/shared_container_data.go index a5addd358b..8629898f73 100644 --- a/pkg/objectcache/shared_container_data.go +++ b/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 { diff --git a/pkg/objectcache/shared_container_data_test.go b/pkg/objectcache/shared_container_data_test.go index ff1cd4752c..1f4ccdf0f1 100644 --- a/pkg/objectcache/shared_container_data_test.go +++ b/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()) + } +}