[EXP-704] ecs: v1 metadata is not required for daemon mode on Managed Instances (#54174)

### What does this PR do?

Makes the ECS Metadata v1 introspection endpoint a best-effort dependency for ECS **Managed Instances** daemon mode instead of a hard startup requirement, and adds a task-metadata fallback for ECS cluster metadata.

Two changes:

1. `comp/core/workloadmeta/collectors/internal/ecs/daemon_parser.go` — `initializeDaemonMode` no longer returns early when `ecsmeta.V1()` or `GetInstance` fails. When v1 is unusable and the launch type is Managed Instances with a working v4 client, it routes to `parseTasksFromV4TasksEndpoint`. EC2 still fails, because both of its daemon parsers read the task list from `metaV1.GetTasks` — but it now fails with a retriable error so the collector is retried rather than dropped.
2. `pkg/util/ecs/ecs.go` — `newECSMeta` falls back from `getECSInstanceMetadata` (v1) to `getECSTaskMetadata` (v3/v4) on error, instead of only using task metadata on Fargate.

### Motivation

On Managed Instances, `initializeDaemonMode` called `ecsmeta.V1()` unconditionally before daemon mode could start, but the parser that actually runs there never consumes what v1 provides:

- **ClusterName / ClusterARN** — derived from each v4 task's own `Cluster` field in `util.ParseV4Task`
- **ContainerInstanceARN** — never set by `ParseV4Task` for any launch type, so it is already empty on this path
- **instance.Version** — only drives parser selection, and Managed Instances already has a fallback that keys off `ECS_CONTAINER_METADATA_URI_V4`

So a v1 failure blocked startup entirely, producing zero `ECSTask` entities and no `task_arn`, `task_family`, `task_version`, `ecs_cluster_name` or `ecs_container_name` tags on `container.*` metrics for the whole host — even though the v4 `/tasks` path works fine. 

The second change fixes `GetClusterMeta()` which separate v1 call site that the collector fix does not touch, so the orchestrator ECS check stayed gated off:

```
Failed to get ECS meta: temporary failure in ecsutil-meta-v1 ...
Orchestrator ECS check is missing required information, region: , awsAccountID: , clusterName: , clusterID:
```

All four required values are available from the v4 task payload, and `initClusterID` is a pure `md5(account/region/cluster)` derivation, so the cluster ID is identical either way. This also unblocks the container lifecycle check's cluster ID and the flare's ECS section.

### Describe how you validated your changes

Unit tests (run in the Linux dev container, since the `docker` build tag is excluded on macOS by `DARWIN_EXCLUDED_TAGS`):

- `TestInitializeDaemonModeV1Unavailable` — Managed Instances starts and selects the v4 `/tasks` parser both when the v1 client fails to initialize and when `GetInstance` fails; EC2 fails in both cases; Managed Instances also fails when task collection is disabled or no v4 client is available. Asserts `taskCollectionParser` is never nil when `Start` succeeds, and that every error return satisfies `retry.IsErrWillRetry`.
- `TestInitializeDaemonModeV1Available` — a reachable v1 endpoint still populates `clusterName`/`containerInstanceARN` and selects the per-task v4 parser, unchanged for EC2.
- `TestNewECSMetaTaskMetadataFallback` — instance metadata used when available; falls back to task metadata on failure and produces the same cluster ID; surfaces the original v1 error when both fail.

```
dda inv test --targets=./comp/core/workloadmeta/collectors/internal/ecs   # 69 tests pass
dda inv test --targets=./pkg/util/ecs                                     # 47 tests pass
dda inv test --targets=./pkg/collector/corechecks/orchestrator/ecs,./pkg/collector/corechecks/containerlifecycle,./comp/core/tagger/collectors  # 177 tests pass
dda inv linter.go --targets=./pkg/util/ecs,./comp/core/workloadmeta/collectors/internal/ecs  # 0 issues
```

### Additional Notes

**The daemon-mode startup error has to be a `*retry.Error`.** `workloadmeta.startCandidates` keeps a collector in its candidate set only when `retry.IsErrWillRetry(err)` is true, and `retry.IsRetryError` type-asserts to `*retry.Error` instead of unwrapping. So wrapping with `fmt.Errorf("%w")` makes the error look non-retriable, and the collector is deleted from the candidate set — ECS task collection would then stay off until the Agent restarts. `initializeDaemonMode` returns a `*retry.Error` with `FailWillRetry`, keeping the original error as `LogicError` for context. This also makes the `GetInstance`-failure path retriable, where it previously returned `nil` and left a nil task parser that would be invoked on the first `Pull`. `TestInitializeDaemonModeV1Unavailable` asserts the retriable contract on every error path so it cannot regress silently.

**Launch types other than MI daemon are unaffected.** `metaV1` is referenced in exactly four places in the collector package: `daemon_parser.go` (init) and `v1parser.go` / `v4parser.go` (`GetTasks`). Sidecar mode uses `metaV2` and `metaV4` and never touches v1, so Fargate (always sidecar) and MI sidecar are untouched. EC2 daemon keeps its existing v1 requirement.




Co-authored-by: steve.zou <steve.zou@datadoghq.com>

Author: stzou

comp/core/workloadmeta/collectors/internal/ecs/BUILD.bazel

@@ -29,6 +29,7 @@ go_library(
         "//pkg/util/ecs/metadata/v3or4",
         "//pkg/util/fargate",
         "//pkg/util/log",
+        "//pkg/util/retry",
         "@com_github_patrickmn_go_cache//:go-cache",
         "@io_k8s_client_go//util/workqueue",
         "@org_golang_x_time//rate",
@@ -62,6 +63,7 @@ dd_agent_go_test(
         "//pkg/util/ecs/metadata/v1:metadata",
         "//pkg/util/ecs/metadata/v2:metadata",
         "//pkg/util/ecs/metadata/v3or4",
+        "//pkg/util/retry",
         "@com_github_patrickmn_go_cache//:go-cache",
         "@com_github_stretchr_testify//assert",
         "@com_github_stretchr_testify//require",

comp/core/workloadmeta/collectors/internal/ecs/daemon_parser.go

@@ -10,6 +10,7 @@ package ecs
 
 import (
 	"context"
+	"fmt"
 	"os"
 	"time"
 
@@ -18,14 +19,26 @@ import (
 	ecsmeta "github.com/DataDog/datadog-agent/pkg/util/ecs/metadata"
 	"github.com/DataDog/datadog-agent/pkg/util/ecs/metadata/v3or4"
 	"github.com/DataDog/datadog-agent/pkg/util/log"
+	"github.com/DataDog/datadog-agent/pkg/util/retry"
 
 	"github.com/DataDog/datadog-agent/comp/core/workloadmeta/collectors/util"
 )
 
+// declare these as vars not const to ease testing: the underlying helpers memoize
+// global state and perform network I/O, which unit tests cannot rely on.
+//
+// These are process-wide mutable globals, so tests that swap them must restore the
+// originals and must not call t.Parallel().
+var (
+	ecsMetaV1                = ecsmeta.V1
+	ecsMetaV4FromCurrentTask = ecsmeta.V4FromCurrentTask
+	ecsHasEC2ResourceTags    = ecsutil.HasEC2ResourceTags
+)
+
 // initializeDaemonMode sets up the collector for daemon deployment mode.
 //
 // In daemon mode, the agent runs as a daemon on an ECS instance and monitors all tasks on that instance.
-// This mode requires V1 metadata API access and uses different parsing strategies based on configuration:
+// The parsing strategy depends on configuration and on which metadata endpoints are reachable:
 //
 //   - V1 parsing: Lists all tasks on the instance (basic info)
 //     See: v1parser.go - parseTasksFromV1Endpoint()
@@ -36,15 +49,13 @@ import (
 //   - V4 /tasks parsing: Fetches all host tasks in a single call from the daemon container's v4 endpoint.
 //     Used on ECS Managed Instances where the /tasks endpoint is available.
 //     See: daemon_parser.go - parseTasksFromV4TasksEndpoint()
+//
+// V1 is required by the first two strategies, which read the task list from it. It is not
+// required by the third: on ECS Managed Instances the v4 /tasks payload carries everything
+// the parser needs, and the v1 introspection endpoint is not guaranteed to be reachable
+// from the daemon container. Daemon mode therefore treats v1 as best-effort and only fails
+// when no v1-independent strategy is available.
 func (c *collector) initializeDaemonMode(ctx context.Context) error {
-	var err error
-
-	// Daemon mode requires v1 API access
-	c.metaV1, err = ecsmeta.V1()
-	if err != nil {
-		return err
-	}
-
 	// This only exists to allow overriding for testing
 	c.metaV3or4 = func(metaURI, metaVersion string) v3or4.Client {
 		return v3or4.NewClient(metaURI, metaVersion, v3or4.WithTryOption(
@@ -56,25 +67,54 @@ func (c *collector) initializeDaemonMode(ctx context.Context) error {
 
 	// Attempt to initialize a v4 client for the daemon agent's own container.
 	// This enables the /tasks endpoint on ECS Managed Instances.
-	if v4Client, err := ecsmeta.V4FromCurrentTask(); err == nil {
+	if v4Client, err := ecsMetaV4FromCurrentTask(); err == nil {
 		c.metaV4 = v4Client
 	} else {
 		log.Debugf("ECS daemon: failed to initialize v4 client for current task (may not be available): %v", err)
 	}
 
-	c.hasResourceTags = ecsutil.HasEC2ResourceTags()
+	c.hasResourceTags = ecsHasEC2ResourceTags()
 	c.collectResourceTags = c.config.GetBool("ecs_collect_resource_tags_ec2")
 
-	instance, err := c.metaV1.GetInstance(ctx)
-	if err == nil {
-		c.clusterName = instance.Cluster
-		c.containerInstanceARN = instance.ContainerInstanceARN
-		c.setTaskCollectionParserForDaemon(instance.Version)
+	var v1Err error
+	c.metaV1, v1Err = ecsMetaV1()
+	if v1Err != nil {
+		log.Warnf("ECS daemon: metadata v1 client unavailable: %v", v1Err)
+		// Guard against a non-nil interface wrapping a nil client.
+		c.metaV1 = nil
 	} else {
-		log.Warnf("cannot determine ECS cluster name: %s", err)
+		instance, err := c.metaV1.GetInstance(ctx)
+		if err != nil {
+			v1Err = err
+			log.Warnf("cannot determine ECS cluster name: %s", err)
+		} else {
+			c.clusterName = instance.Cluster
+			c.containerInstanceARN = instance.ContainerInstanceARN
+			c.setTaskCollectionParserForDaemon(instance.Version)
+			return nil
+		}
+	}
+
+	// No usable v1 instance metadata. The v4 /tasks endpoint on Managed Instances derives
+	// cluster identity from each task, so it can run without v1. Select it explicitly
+	// rather than going through setTaskCollectionParserForDaemon, which would otherwise
+	// pick a v1-backed parser and dereference the nil client on the first Pull.
+	if c.taskCollectionEnabled && c.metaV4 != nil &&
+		c.actualLaunchType == workloadmeta.ECSLaunchTypeManagedInstances {
+		log.Infof("ECS daemon: metadata v1 unavailable, using metadata v4 /tasks endpoint for managed instances")
+		c.taskCollectionParser = c.parseTasksFromV4TasksEndpoint
+		return nil
 	}
 
-	return nil
+	// Returning a retriable error leaves the collector in workloadmeta's candidate set, so
+	// Start is retried and picks up the cluster name once the endpoint recovers. This must
+	// be a *retry.Error: workloadmeta gates on retry.IsErrWillRetry, which type-asserts
+	// rather than unwrapping, so wrapping with %w here would drop the collector for good.
+	return &retry.Error{
+		LogicError:    fmt.Errorf("ECS daemon mode requires metadata v1: %w", v1Err),
+		RessourceName: componentName,
+		RetryStatus:   retry.FailWillRetry,
+	}
 }
 
 // setTaskCollectionParserForDaemon sets up the appropriate task parser for daemon deployment mode.

comp/core/workloadmeta/collectors/internal/ecs/daemon_parser_test.go

@@ -8,6 +8,8 @@
 package ecs
 
 import (
+	"context"
+	"errors"
 	"os"
 	"reflect"
 	"runtime"
@@ -16,8 +18,11 @@ import (
 	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/require"
 
+	"github.com/DataDog/datadog-agent/comp/core/config"
 	workloadmeta "github.com/DataDog/datadog-agent/comp/core/workloadmeta/def"
+	v1 "github.com/DataDog/datadog-agent/pkg/util/ecs/metadata/v1"
 	"github.com/DataDog/datadog-agent/pkg/util/ecs/metadata/v3or4"
+	"github.com/DataDog/datadog-agent/pkg/util/retry"
 )
 
 // taskParserName returns the name of the function backing the task parser for assertion.
@@ -166,3 +171,167 @@ func TestSetTaskCollectionParserForDaemon(t *testing.T) {
 		})
 	}
 }
+
+// TestInitializeDaemonModeV1Unavailable covers daemon-mode startup when the metadata v1
+// introspection endpoint cannot be used. ECS Managed Instances must still start by routing
+// to the v4 /tasks endpoint, while EC2 must keep failing because every EC2 daemon parser
+// reads the task list from v1.
+func TestInitializeDaemonModeV1Unavailable(t *testing.T) {
+	getInstanceErr := errors.New("connection refused")
+
+	tests := []struct {
+		name                  string
+		launchType            workloadmeta.ECSLaunchType
+		taskCollectionEnabled bool
+		hasMetaV4             bool
+		// v1Client is returned by the ecsMetaV1 seam; nil means v1 client init fails.
+		v1Client           *fakev1EcsClient
+		expectErr          bool
+		expectParserSuffix string
+	}{
+		{
+			name:                  "managed instances without v1 client uses v4 /tasks",
+			launchType:            workloadmeta.ECSLaunchTypeManagedInstances,
+			taskCollectionEnabled: true,
+			hasMetaV4:             true,
+			v1Client:              nil,
+			expectParserSuffix:    "parseTasksFromV4TasksEndpoint",
+		},
+		{
+			name:                  "managed instances with failing GetInstance uses v4 /tasks",
+			launchType:            workloadmeta.ECSLaunchTypeManagedInstances,
+			taskCollectionEnabled: true,
+			hasMetaV4:             true,
+			v1Client: &fakev1EcsClient{
+				mockGetInstance: func(context.Context) (*v1.Instance, error) { return nil, getInstanceErr },
+			},
+			expectParserSuffix: "parseTasksFromV4TasksEndpoint",
+		},
+		{
+			name:                  "managed instances without v1 and without metaV4 fails",
+			launchType:            workloadmeta.ECSLaunchTypeManagedInstances,
+			taskCollectionEnabled: true,
+			hasMetaV4:             false,
+			v1Client:              nil,
+			expectErr:             true,
+		},
+		{
+			name:                  "managed instances without v1 and task collection disabled fails",
+			launchType:            workloadmeta.ECSLaunchTypeManagedInstances,
+			taskCollectionEnabled: false,
+			hasMetaV4:             true,
+			v1Client:              nil,
+			expectErr:             true,
+		},
+		{
+			name:                  "ec2 without v1 client fails",
+			launchType:            workloadmeta.ECSLaunchTypeEC2,
+			taskCollectionEnabled: true,
+			hasMetaV4:             true,
+			v1Client:              nil,
+			expectErr:             true,
+		},
+		{
+			name:                  "ec2 with failing GetInstance fails instead of leaving a nil parser",
+			launchType:            workloadmeta.ECSLaunchTypeEC2,
+			taskCollectionEnabled: true,
+			hasMetaV4:             true,
+			v1Client: &fakev1EcsClient{
+				mockGetInstance: func(context.Context) (*v1.Instance, error) { return nil, getInstanceErr },
+			},
+			expectErr: true,
+		},
+	}
+
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			// The v4 /tasks route additionally requires the v4 env var to be present.
+			t.Setenv(v3or4.DefaultMetadataURIv4EnvVariable, "http://169.254.170.2/v4")
+
+			restore := stubDaemonMetadataSeams(t, tt.v1Client, tt.hasMetaV4)
+			defer restore()
+
+			c := &collector{
+				config:                config.NewMockWithOverrides(t, map[string]interface{}{}),
+				taskCollectionEnabled: tt.taskCollectionEnabled,
+				actualLaunchType:      tt.launchType,
+			}
+
+			err := c.initializeDaemonMode(context.Background())
+
+			if tt.expectErr {
+				require.Error(t, err)
+				// workloadmeta only keeps a collector in its candidate set when
+				// retry.IsErrWillRetry matches, and that helper type-asserts instead of
+				// unwrapping. A non-retriable error here would drop the ECS collector
+				// permanently instead of retrying Start.
+				assert.True(t, retry.IsErrWillRetry(err),
+					"startup error must be retriable, got %T: %v", err, err)
+				return
+			}
+
+			require.NoError(t, err)
+			require.NotNil(t, c.taskCollectionParser, "a nil parser would panic on the first Pull")
+			assert.Contains(t, taskParserName(c.taskCollectionParser), tt.expectParserSuffix)
+		})
+	}
+
+}
+
+// TestInitializeDaemonModeV1Available verifies that a reachable v1 endpoint still drives
+// parser selection and populates the instance-derived fields, unchanged for EC2.
+func TestInitializeDaemonModeV1Available(t *testing.T) {
+	t.Setenv(v3or4.DefaultMetadataURIv4EnvVariable, "http://169.254.170.2/v4")
+
+	v1Client := &fakev1EcsClient{
+		mockGetInstance: func(context.Context) (*v1.Instance, error) {
+			return &v1.Instance{
+				Cluster:              "my-cluster",
+				ContainerInstanceARN: "arn:aws:ecs:us-east-1:123456789012:container-instance/my-cluster/abc123",
+				Version:              "Amazon ECS Agent - v1.54.0 (abc1234)",
+			}, nil
+		},
+	}
+
+	restore := stubDaemonMetadataSeams(t, v1Client, false)
+	defer restore()
+
+	c := &collector{
+		config:                config.NewMockWithOverrides(t, map[string]interface{}{}),
+		taskCollectionEnabled: true,
+		actualLaunchType:      workloadmeta.ECSLaunchTypeEC2,
+	}
+
+	require.NoError(t, c.initializeDaemonMode(context.Background()))
+	assert.Equal(t, "my-cluster", c.clusterName)
+	assert.Equal(t, "arn:aws:ecs:us-east-1:123456789012:container-instance/my-cluster/abc123", c.containerInstanceARN)
+	require.NotNil(t, c.taskCollectionParser)
+	assert.Contains(t, taskParserName(c.taskCollectionParser), "parseTasksFromV4Endpoint")
+}
+
+// stubDaemonMetadataSeams replaces the metadata helpers that initializeDaemonMode calls so
+// tests do not depend on memoized global clients or network I/O. A nil v1Client makes the
+// v1 client initialization fail.
+func stubDaemonMetadataSeams(t *testing.T, v1Client *fakev1EcsClient, hasMetaV4 bool) func() {
+	t.Helper()
+
+	origV1, origV4, origTags := ecsMetaV1, ecsMetaV4FromCurrentTask, ecsHasEC2ResourceTags
+
+	ecsMetaV1 = func() (v1.Client, error) {
+		if v1Client == nil {
+			return nil, errors.New("temporary failure in ecsutil-meta-v1")
+		}
+		return v1Client, nil
+	}
+	ecsMetaV4FromCurrentTask = func() (v3or4.Client, error) {
+		if !hasMetaV4 {
+			return nil, errors.New("v4 metadata endpoint not available")
+		}
+		return &fakev3or4EcsClient{}, nil
+	}
+	ecsHasEC2ResourceTags = func() bool { return false }
+
+	return func() {
+		ecsMetaV1, ecsMetaV4FromCurrentTask, ecsHasEC2ResourceTags = origV1, origV4, origTags
+	}
+}

comp/core/workloadmeta/collectors/internal/ecs/ecs_test.go

@@ -45,14 +45,18 @@ func (store *fakeWorkloadmetaStore) GetContainer(id string) (*workloadmeta.Conta
 }
 
 type fakev1EcsClient struct {
-	mockGetTasks func(context.Context) ([]v1.Task, error)
+	mockGetTasks    func(context.Context) ([]v1.Task, error)
+	mockGetInstance func(context.Context) (*v1.Instance, error)
 }
 
 func (c *fakev1EcsClient) GetTasks(ctx context.Context) ([]v1.Task, error) {
 	return c.mockGetTasks(ctx)
 }
 
-func (c *fakev1EcsClient) GetInstance(_ context.Context) (*v1.Instance, error) {
+func (c *fakev1EcsClient) GetInstance(ctx context.Context) (*v1.Instance, error) {
+	if c.mockGetInstance != nil {
+		return c.mockGetInstance(ctx)
+	}
 	return nil, errors.New("unimplemented")
 }
 

pkg/util/ecs/ecs.go

@@ -21,6 +21,13 @@ import (
 	"github.com/DataDog/datadog-agent/pkg/config/setup/constants"
 	"github.com/DataDog/datadog-agent/pkg/util/cache"
 	"github.com/DataDog/datadog-agent/pkg/util/ecs/metadata"
+	"github.com/DataDog/datadog-agent/pkg/util/log"
+)
+
+// declare these as vars not const to ease testing
+var (
+	fetchECSInstanceMetadata = getECSInstanceMetadata
+	fetchECSTaskMetadata     = getECSTaskMetadata
 )
 
 // MetaECS stores ECS cluster metadata
@@ -80,9 +87,35 @@ func newECSMeta(ctx context.Context) (*MetaECS, error) {
 
 	if env.IsFeaturePresent(env.ECSFargate) {
 		// There is no instance metadata endpoint on ECS Fargate
-		awsAccountID, region, cluster, version, err = getECSTaskMetadata(ctx)
+		awsAccountID, region, cluster, version, err = fetchECSTaskMetadata(ctx)
 	} else {
-		awsAccountID, region, cluster, version, err = getECSInstanceMetadata(ctx)
+		awsAccountID, region, cluster, version, err = fetchECSInstanceMetadata(ctx)
+		if err != nil {
+			// The v1 introspection endpoint is not guaranteed to be reachable outside of
+			// ECS EC2 (notably on ECS Managed Instances). The agent's own task metadata
+			// carries the same cluster identity — the agent task belongs to the cluster it
+			// runs on, and region/account come from its own task ARN — so fall back to it.
+			// This applies to every non-Fargate launch type, not just Managed Instances: on
+			// EC2 it only takes effect when v1 already failed, where the previous behaviour
+			// was to give up entirely. A non-containerised agent has no task metadata
+			// endpoint, so the fallback fails and the original v1 error is returned.
+			//
+			// Note the returned version is the task revision here rather than the ECS agent
+			// version. MetaECS.ECSAgentVersion is only round-tripped through the cache key
+			// (toCacheValue/fromCacheValue) and never used for behaviour, and the Fargate
+			// branch above already populates it the same way.
+			log.Debugf("could not get ECS instance metadata, falling back to task metadata: %s", err)
+
+			var fallbackErr error
+			awsAccountID, region, cluster, version, fallbackErr = fetchECSTaskMetadata(ctx)
+			if fallbackErr != nil {
+				log.Debugf("could not get ECS task metadata either: %s", fallbackErr)
+				// Surface the original instance metadata error, which is the more
+				// meaningful one outside of Managed Instances.
+				return nil, err
+			}
+			err = nil
+		}
 	}
 
 	if err != nil {