Better handle missing containerids

There are cases when we are notified of a new container and there is a
ContainerStatus object that matches the container, but the container is
not running yet so it does not have a containerID property within it
yet. This messes up our state machine where we attempt to tail the wrong
file without a containerID as part of the file path.

collector/logs/sources/tail/pod_discovery.go

@@ -206,8 +206,10 @@ func (i *PodDiscovery) tombstoneTarget(existingCurrentTarget *currenttarget) {
 
 	// We don't want to just tell tailsource to stop tailing when the file reaches EOF because the file
 	// may be written to again in the "future" and perhaps even rotated.
-	logger.Infof("Tombstoning target %s", existingCurrentTarget.CurrentTarget.FilePath)
-	existingCurrentTarget.ExpireTime = i.nowFunc().Add(1 * time.Minute)
+	if existingCurrentTarget.ExpireTime.IsZero() {
+		logger.Infof("Tombstoning target %s", existingCurrentTarget.CurrentTarget.FilePath)
+		existingCurrentTarget.ExpireTime = i.nowFunc().Add(1 * time.Minute)
+	}
 }
 
 func (i *PodDiscovery) cleanTombstonedTargets() {

collector/logs/sources/tail/pod_discovery_test.go

@@ -39,6 +39,25 @@ func TestPodDiscoveryLifecycle(t *testing.T) {
 		require.Equal(t, 0, len(tailSource.targetRemoves))
 	})
 
+	t.Run("Add Pod Without ContainerId", func(t *testing.T) {
+		_, tailSource, podDiscovery := create()
+		defer tailSource.Close()
+		err := podDiscovery.Open(context.Background())
+		require.NoError(t, err)
+
+		pod := scrapedPod("pod1")
+		pod.Status.ContainerStatuses[0].ContainerID = ""
+
+		podDiscovery.OnAdd(pod, false)
+
+		cleanTombstonedTargets(podDiscovery)
+		err = podDiscovery.Close()
+		require.NoError(t, err)
+		// No containerID yet, so no target added.
+		require.Equal(t, 0, len(tailSource.targetAdds))
+		require.Equal(t, 0, len(tailSource.targetRemoves))
+	})
+
 	t.Run("Add Tolerant of Failure", func(t *testing.T) {
 		// Failure tailing first container, but ok to tail the second container
 		failureFunc := func(target FileTailTarget) error {

collector/logs/sources/tail/pod_target.go

@@ -172,7 +172,8 @@ func isTargetChanged(old, new FileTailTarget) bool {
 func getContainerID(containerStatuses []v1.ContainerStatus, containerName string) (string, bool) {
 	for _, container := range containerStatuses {
 		if container.Name == containerName {
-			return container.ContainerID, true
+			containerIdPopulated := container.ContainerID != ""
+			return container.ContainerID, containerIdPopulated
 		}
 	}
 	return "", false