Add more validation of stats received from Docker

agent/stats/container.go

@@ -85,7 +85,11 @@ func (container *StatsContainer) collect() {
 		default:
 			if err != nil {
 				d := backoff.Duration()
-				seelog.Debugf("Container [%s]: Error processing stats stream of container, backing off %s before reopening", dockerID, d)
+				logger.Debug(fmt.Sprintf(
+					"Error processing stats stream of container, backing off %s before reopening", d), logger.Fields{
+					loggerfield.DockerId: dockerID,
+					loggerfield.Error:    err,
+				})
 				time.Sleep(d)
 			}
 			// We were disconnected from the stats stream.
@@ -143,8 +147,12 @@ func (container *StatsContainer) processStatsStream() error {
 	dockerStats, errC := container.client.Stats(container.ctx, dockerID, dockerclient.StatsInactivityTimeout)
 
 	apiContainer, err := container.getApiContainer(dockerID)
-	if err != nil {
-		seelog.Errorf("Will not be able to get restart stats set, error: %s", err)
+	if apiContainer == nil && err != nil {
+		logger.Error("apiContainer is nil - will not be able to get restart stats set or determine "+
+			"if container has restart policy enabled", logger.Fields{
+			loggerfield.DockerId: dockerID,
+			loggerfield.Error:    err,
+		})
 	}
 
 	returnError := false
@@ -167,7 +175,11 @@ func (container *StatsContainer) processStatsStream() error {
 				}
 				return nil
 			}
-			err := validateDockerStats(rawStat)
+			containerEnabledRestartPolicy := false // default to false if apiContainer is nil
+			if apiContainer != nil {
+				containerEnabledRestartPolicy = apiContainer.RestartPolicyEnabled()
+			}
+			err := validateDockerStats(rawStat, containerEnabledRestartPolicy)
 			if err != nil {
 				return err
 			}

agent/stats/queue.go

@@ -584,12 +584,15 @@ func getAggregatedDockerStatAcrossRestarts(dockerStat, lastStatBeforeLastRestart
 	dockerStat = aggregateOSIndependentStats(dockerStat, lastStatBeforeLastRestart)
 	dockerStat = aggregateOSDependentStats(dockerStat, lastStatBeforeLastRestart)
 
-	// PreCPU stats.
+	// Stats relevant to PreCPU.
 	preCPUStats := types.CPUStats{}
+	preRead := time.Time{}
 	if lastStatInStatsQueue != nil {
 		preCPUStats = lastStatInStatsQueue.CPUStats
+		preRead = lastStatInStatsQueue.Read
 	}
 	dockerStat.PreCPUStats = preCPUStats
+	dockerStat.PreRead = preRead
 
 	logger.Debug("Aggregated Docker stat across restart(s)", logger.Fields{
 		loggerfield.DockerId: dockerStat.ID,

agent/stats/unix_test_stats.json

@@ -1,4 +1,5 @@
 {
+    "read": "1969-12-31T23:59:59.99999999Z",
     "blkio_stats": {
         "io_service_bytes_recursive": [
             {

agent/stats/utils.go

@@ -20,6 +20,10 @@ import (
 	"github.com/docker/docker/api/types"
 )
 
+const (
+	invalidStatZeroValueReadTimeMsg = "read time of stat is 0001-01-01T00:00:00Z"
+)
+
 var numCores = uint64(eautils.GetNumCPU())
 
 // nan32 returns a 32bit NaN.

agent/stats/utils_unix.go

@@ -55,7 +55,12 @@ func getMemUsage(mem types.MemoryStats) uint64 {
 	return mem.Usage
 }
 
-func validateDockerStats(dockerStats *types.StatsJSON) error {
+func validateDockerStats(dockerStats *types.StatsJSON, containerEnabledRestartPolicy bool) error {
+	if containerEnabledRestartPolicy && dockerStats.Read.IsZero() {
+		return fmt.Errorf("invalid container statistics reported for container with restart policy enabled, %s",
+			invalidStatZeroValueReadTimeMsg)
+	}
+
 	if config.CgroupV2 {
 		// PercpuUsage is not available in cgroupv2
 		if numCores == uint64(0) {

agent/stats/utils_unix_test.go

@@ -18,8 +18,10 @@ package stats
 import (
 	"encoding/json"
 	"io/ioutil"
+	"os"
 	"path/filepath"
 	"testing"
+	"time"
 
 	"github.com/docker/docker/api/types"
 	"github.com/stretchr/testify/assert"
@@ -55,7 +57,42 @@ func TestDockerStatsToContainerStatsEmptyCpuUsageGeneratesError(t *testing.T) {
 	json.Unmarshal([]byte(jsonBytes), dockerStat)
 	prevNumCores := numCores
 	numCores = uint64(0)
-	err := validateDockerStats(dockerStat)
+	err := validateDockerStats(dockerStat, false)
 	assert.Error(t, err, "expected error converting container stats with numCores=0")
 	numCores = prevNumCores
 }
+
+func TestValidateDockerStatsZeroValueReadTime(t *testing.T) {
+	testCases := []struct {
+		name                          string
+		containerEnabledRestartPolicy bool
+	}{
+		{
+			name:                          "container does not have restart policy enabled",
+			containerEnabledRestartPolicy: false,
+		},
+		{
+			name:                          "container has restart policy enabled",
+			containerEnabledRestartPolicy: true,
+		},
+	}
+	inputJsonFile, _ := filepath.Abs("./unix_test_stats.json")
+	jsonBytes, err := os.ReadFile(inputJsonFile)
+	assert.NoError(t, err)
+	dockerStat := &types.StatsJSON{}
+	json.Unmarshal(jsonBytes, dockerStat)
+
+	for _, tc := range testCases {
+		t.Run(tc.name, func(t *testing.T) {
+			// Set read time of docker stat to zero value of time.Time.
+			dockerStat.Read = time.Time{}
+			err = validateDockerStats(dockerStat, tc.containerEnabledRestartPolicy)
+			if tc.containerEnabledRestartPolicy {
+				assert.Error(t, err)
+				assert.ErrorContains(t, err, invalidStatZeroValueReadTimeMsg)
+			} else {
+				assert.NoError(t, nil)
+			}
+		})
+	}
+}

agent/stats/utils_windows.go

@@ -39,7 +39,12 @@ func dockerStatsToContainerStats(dockerStats *types.StatsJSON) (*ContainerStats,
 	}, nil
 }
 
-func validateDockerStats(dockerStats *types.StatsJSON) error {
+func validateDockerStats(dockerStats *types.StatsJSON, containerEnabledRestartPolicy bool) error {
+	if containerEnabledRestartPolicy && dockerStats.Read.IsZero() {
+		return fmt.Errorf("invalid container statistics reported for container with restart policy enabled, %s",
+			invalidStatZeroValueReadTimeMsg)
+	}
+
 	if numCores == uint64(0) {
 		return fmt.Errorf("invalid container statistics reported, no cpu core usage reported")
 	}

agent/stats/utils_windows_test.go

@@ -19,8 +19,10 @@ import (
 	"encoding/json"
 	"fmt"
 	"io/ioutil"
+	"os"
 	"path/filepath"
 	"testing"
+	"time"
 
 	"github.com/docker/docker/api/types"
 	"github.com/stretchr/testify/assert"
@@ -40,10 +42,45 @@ func TestDockerStatsToContainerStatsZeroCoresGeneratesError(t *testing.T) {
 		}`, 100)
 	dockerStat := &types.StatsJSON{}
 	json.Unmarshal([]byte(jsonStat), dockerStat)
-	err := validateDockerStats(dockerStat)
+	err := validateDockerStats(dockerStat, false)
 	assert.Error(t, err, "expected error converting container stats with zero cpu cores")
 }
 
+func TestValidateDockerStatsZeroValueReadTime(t *testing.T) {
+	testCases := []struct {
+		name                          string
+		containerEnabledRestartPolicy bool
+	}{
+		{
+			name:                          "container does not have restart policy enabled",
+			containerEnabledRestartPolicy: false,
+		},
+		{
+			name:                          "container has restart policy enabled",
+			containerEnabledRestartPolicy: true,
+		},
+	}
+	inputJsonFile, _ := filepath.Abs("./windows_test_stats.json")
+	jsonBytes, err := os.ReadFile(inputJsonFile)
+	assert.NoError(t, err)
+	dockerStat := &types.StatsJSON{}
+	json.Unmarshal(jsonBytes, dockerStat)
+
+	for _, tc := range testCases {
+		t.Run(tc.name, func(t *testing.T) {
+			// Set read time of docker stat to zero value of time.Time.
+			dockerStat.Read = time.Time{}
+			err = validateDockerStats(dockerStat, tc.containerEnabledRestartPolicy)
+			if tc.containerEnabledRestartPolicy {
+				assert.Error(t, err)
+				assert.ErrorContains(t, err, invalidStatZeroValueReadTimeMsg)
+			} else {
+				assert.NoError(t, nil)
+			}
+		})
+	}
+}
+
 func TestDockerStatsToContainerStats(t *testing.T) {
 	numCores = 4
 	inputJsonFile, _ := filepath.Abs("./windows_test_stats.json")

agent/stats/windows_test_stats.json

@@ -1,4 +1,5 @@
 {
+    "read": "1969-12-31T23:59:59.99999999Z",
     "blkio_stats": {
         "io_service_bytes_recursive": null,
         "io_serviced_recursive": null,