diff --git a/confgenerator/confgenerator.go b/confgenerator/confgenerator.go
index c3bd8b0f6b..378109d61c 100644
--- a/confgenerator/confgenerator.go
+++ b/confgenerator/confgenerator.go
@@ -20,6 +20,7 @@ import (
 	"crypto/md5"
 	"encoding/hex"
 	"fmt"
+	"log"
 	"path"
 	"regexp"
 	"sort"
@@ -28,6 +29,7 @@ import (
 
 	"github.com/GoogleCloudPlatform/ops-agent/confgenerator/fluentbit"
 	"github.com/GoogleCloudPlatform/ops-agent/confgenerator/otel"
+	"github.com/GoogleCloudPlatform/ops-agent/confgenerator/resourcedetector"
 	"github.com/GoogleCloudPlatform/ops-agent/internal/platform"
 )
 
@@ -293,6 +295,57 @@ func processUserDefinedMultilineParser(i int, pID string, receiver LoggingReceiv
 	return nil
 }
 
+func sliceContains(s []string, v string) bool {
+	for _, e := range s {
+		if e == v {
+			return true
+		}
+	}
+	return false
+}
+
+const (
+	attributeLabelPrefix string = "compute.googleapis.com/attributes/"
+)
+
+// addGceMetadataAttributesComponents annotates logs with labels corresponding
+// to instance attributes from the GCE metadata server.
+func addGceMetadataAttributesComponents(ctx context.Context, attributes []string, tag, uid string) []fluentbit.Component {
+	processorName := fmt.Sprintf("%s.%s.gce_metadata", tag, uid)
+	resource, err := platform.FromContext(ctx).GetResource()
+	if err != nil {
+		log.Printf("can't get resource metadata: %v", err)
+		return nil
+	}
+	gceMetadata, ok := resource.(resourcedetector.GCEResource)
+	if !ok {
+		// Not on GCE; no attributes to detect.
+		log.Printf("ignoring the gce_metadata_attributes processor outside of GCE: %T", resource)
+		return nil
+	}
+	modifications := map[string]*ModifyField{}
+	var attributeKeys []string
+	for k, _ := range gceMetadata.Metadata {
+		attributeKeys = append(attributeKeys, k)
+	}
+	sort.Strings(attributeKeys)
+	for _, k := range attributeKeys {
+		if !sliceContains(attributes, k) {
+			continue
+		}
+		v := gceMetadata.Metadata[k]
+		modifications[fmt.Sprintf(`labels."%s%s"`, attributeLabelPrefix, k)] = &ModifyField{
+			StaticValue: &v,
+		}
+	}
+	if len(modifications) == 0 {
+		return nil
+	}
+	return LoggingProcessorModifyFields{
+		Fields: modifications,
+	}.Components(ctx, tag, processorName)
+}
+
 // generateFluentbitComponents generates a slice of fluentbit config sections to represent l.
 func (uc *UnifiedConfig) generateFluentbitComponents(ctx context.Context, userAgent string) ([]fluentbit.Component, error) {
 	l := uc.Logging
@@ -396,6 +449,11 @@ func (uc *UnifiedConfig) generateFluentbitComponents(ctx context.Context, userAg
 			out = append(out, stackdriverOutputComponent(ctx, strings.Join(tags, "|"), userAgent, "2G"))
 		}
 		out = append(out, uc.generateSelfLogsComponents(ctx, userAgent)...)
+		out = append(out, addGceMetadataAttributesComponents(ctx, []string{
+			"dataproc-cluster-name",
+			"dataproc-cluster-uuid",
+			"dataproc-region",
+		}, "*", "default-dataproc")...)
 	}
 	out = append(out, fluentbit.MetricsOutputComponent())
 
diff --git a/confgenerator/confgenerator_test.go b/confgenerator/confgenerator_test.go
index 7185fbbf1c..45bdf93f2b 100644
--- a/confgenerator/confgenerator_test.go
+++ b/confgenerator/confgenerator_test.go
@@ -57,20 +57,38 @@ var winlogv1channels = []string{
 }
 
 var (
-	testPlatforms = []platformConfig{
-		{
-			name:            "linux",
-			defaultLogsDir:  "/var/log/google-cloud-ops-agent",
-			defaultStateDir: "/var/lib/google-cloud-ops-agent/fluent-bit",
-			platform: platform.Platform{
-				Type: platform.Linux,
-				HostInfo: &host.InfoStat{
-					OS:              "linux",
-					Platform:        "linux_platform",
-					PlatformVersion: "linux_platform_version",
-				},
+	// Set up the test environment with mocked data.
+	testResource = resourcedetector.GCEResource{
+		Project:       "test-project",
+		Zone:          "test-zone",
+		Network:       "test-network",
+		Subnetwork:    "test-subnetwork",
+		PublicIP:      "test-public-ip",
+		PrivateIP:     "test-private-ip",
+		InstanceID:    "test-instance-id",
+		InstanceName:  "test-instance-name",
+		Tags:          "test-tag",
+		MachineType:   "test-machine-type",
+		Metadata:      map[string]string{"test-key": "test-value", "test-escape": "${foo:bar}"},
+		Label:         map[string]string{"test-label-key": "test-label-value"},
+		InterfaceIPv4: map[string]string{"test-interface": "test-interface-ipv4"},
+	}
+	linuxTestPlatform = platformConfig{
+		name:            "linux",
+		defaultLogsDir:  "/var/log/google-cloud-ops-agent",
+		defaultStateDir: "/var/lib/google-cloud-ops-agent/fluent-bit",
+		platform: platform.Platform{
+			Type: platform.Linux,
+			HostInfo: &host.InfoStat{
+				OS:              "linux",
+				Platform:        "linux_platform",
+				PlatformVersion: "linux_platform_version",
 			},
+			TestGCEResourceOverride: testResource,
 		},
+	}
+	testPlatforms = []platformConfig{
+		linuxTestPlatform,
 		{
 			name:            "linux-gpu",
 			defaultLogsDir:  "/var/log/google-cloud-ops-agent",
@@ -82,7 +100,8 @@ var (
 					Platform:        "linux_platform",
 					PlatformVersion: "linux_platform_version",
 				},
-				HasNvidiaGpu: true,
+				TestGCEResourceOverride: testResource,
+				HasNvidiaGpu:            true,
 			},
 		},
 		{
@@ -98,6 +117,7 @@ var (
 					Platform:        "win_platform",
 					PlatformVersion: "win_platform_version",
 				},
+				TestGCEResourceOverride: testResource,
 			},
 		},
 		{
@@ -113,6 +133,7 @@ var (
 					Platform:        "win_platform",
 					PlatformVersion: "win_platform_version",
 				},
+				TestGCEResourceOverride: testResource,
 			},
 		},
 	}
@@ -149,6 +170,42 @@ func TestGoldens(t *testing.T) {
 	}
 }
 
+func TestDataprocDefaults(t *testing.T) {
+	t.Parallel()
+
+	goldensDir := "goldens"
+	testName := "builtin"
+	dataprocMetadata := map[string]string{
+		"dataproc-cluster-name": "test-cluster",
+		"dataproc-cluster-uuid": "test-uuid",
+		"dataproc-region":       "test-region",
+	}
+
+	t.Run(testName, func(t *testing.T) {
+		t.Parallel()
+		pc := linuxTestPlatform
+		// Update mocked resource to include Dataproc labels.
+		dataprocResource := testResource
+		newMetadata := map[string]string{}
+		for k, v := range testResource.Metadata {
+			newMetadata[k] = v
+		}
+		for k, v := range dataprocMetadata {
+			newMetadata[k] = v
+		}
+		dataprocResource.Metadata = newMetadata
+		pc.platform.TestGCEResourceOverride = dataprocResource
+		t.Run(pc.name, func(t *testing.T) {
+			testDir := filepath.Join(goldensDir, testName)
+			got, err := generateConfigs(pc, testDir)
+			assert.NilError(t, err, "Failed to generate configs: %v", err)
+			if err := testGeneratedFiles(t, got, filepath.Join(testDir, goldenDir, "linux-dataproc")); err != nil {
+				t.Errorf("Failed to check generated configs: %v", err)
+			}
+		})
+	})
+}
+
 func getTestsInDir(t *testing.T, testDir string) []string {
 	t.Helper()
 
@@ -310,25 +367,6 @@ func TestMain(m *testing.M) {
 }
 
 func init() {
-	testResource := resourcedetector.GCEResource{
-		Project:       "test-project",
-		Zone:          "test-zone",
-		Network:       "test-network",
-		Subnetwork:    "test-subnetwork",
-		PublicIP:      "test-public-ip",
-		PrivateIP:     "test-private-ip",
-		InstanceID:    "test-instance-id",
-		InstanceName:  "test-instance-name",
-		Tags:          "test-tag",
-		MachineType:   "test-machine-type",
-		Metadata:      map[string]string{"test-key": "test-value", "test-escape": "${foo:bar}"},
-		Label:         map[string]string{"test-label-key": "test-label-value"},
-		InterfaceIPv4: map[string]string{"test-interface": "test-interface-ipv4"},
-	}
-
-	// Set up the test environment with mocked data.
-	confgenerator.MetadataResource = testResource
-
 	// Enable experimental features here by calling:
 	//	 os.Setenv("EXPERIMENTAL_FEATURES", "...(comma-separated feature list)...")
 }
diff --git a/confgenerator/files.go b/confgenerator/files.go
index 9109434aa8..c31958b974 100644
--- a/confgenerator/files.go
+++ b/confgenerator/files.go
@@ -20,8 +20,6 @@ import (
 	"io/ioutil"
 	"os"
 	"path/filepath"
-
-	"github.com/GoogleCloudPlatform/ops-agent/confgenerator/resourcedetector"
 )
 
 // ReadUnifiedConfigFromFile reads the user config file and returns a UnifiedConfig.
@@ -62,13 +60,6 @@ func (uc *UnifiedConfig) GenerateFilesFromConfig(ctx context.Context, service, l
 			}
 		}
 	case "otel":
-		// Fetch resource information from the metadata server.
-		var err error
-		MetadataResource, err = resourcedetector.GetResource()
-		if err != nil {
-			return fmt.Errorf("can't get resource metadata: %w", err)
-		}
-
 		otelConfig, err := uc.GenerateOtelConfig(ctx)
 		if err != nil {
 			return fmt.Errorf("can't parse configuration: %w", err)
diff --git a/confgenerator/prometheus.go b/confgenerator/prometheus.go
index cc49b5f0ad..737d8e47c4 100644
--- a/confgenerator/prometheus.go
+++ b/confgenerator/prometheus.go
@@ -18,6 +18,7 @@ import (
 	"context"
 	"errors"
 	"fmt"
+	"log"
 	"os"
 	"reflect"
 	"sort"
@@ -25,7 +26,6 @@ import (
 	"strings"
 
 	"github.com/GoogleCloudPlatform/ops-agent/confgenerator/otel"
-	"github.com/GoogleCloudPlatform/ops-agent/confgenerator/resourcedetector"
 	"github.com/GoogleCloudPlatform/ops-agent/internal/platform"
 	"github.com/go-playground/validator/v10"
 	yaml "github.com/goccy/go-yaml"
@@ -36,12 +36,6 @@ import (
 	_ "github.com/prometheus/prometheus/discovery/install" // init() of this package registers service discovery impl.
 )
 
-var (
-	// MetadataResource is the resource metadata for the instance we're running on.
-	// Note: This is a global variable so that it can be set in tests.
-	MetadataResource resourcedetector.Resource
-)
-
 type PrometheusMetrics struct {
 	ConfigComponent `yaml:",inline"`
 
@@ -61,9 +55,10 @@ func (r PrometheusMetrics) Type() string {
 }
 
 func (r PrometheusMetrics) Pipelines(ctx context.Context) []otel.ReceiverPipeline {
-	resource := MetadataResource
-	if p := platform.FromContext(ctx).ResourceOverride; p != nil {
-		resource = p
+	resource, err := platform.FromContext(ctx).GetResource()
+	if err != nil {
+		log.Printf("can't get resource metadata: %v", err)
+		return nil
 	}
 	if resource != nil {
 		// Get the resource metadata for the instance we're running on.
diff --git a/confgenerator/resourcedetector/detector.go b/confgenerator/resourcedetector/detector.go
index 0570dd5ce5..6fa0dbfe16 100644
--- a/confgenerator/resourcedetector/detector.go
+++ b/confgenerator/resourcedetector/detector.go
@@ -22,17 +22,36 @@ import (
 // An implementation of the Resource interface will have fields represent
 // available attributes about the current monitoring resource.
 type Resource interface {
+	ProjectName() string
 	MonitoredResource() *monitoredres.MonitoredResource
 	OTelResourceAttributes() map[string]string
-	ProjectName() string
 	PrometheusStyleMetadata() map[string]string
 }
 
+type resourceCache struct {
+	Resource Resource
+	Error    error
+}
+
+var cachedResourceAndError *resourceCache
+
 // Get a resource instance for the current environment;
 // In order to access the attributes of a specific type of resource,
 // needs to cast the returned Resource instance to its underlying type:
 // actual, ok := resource.(GCEResource)
 func GetResource() (Resource, error) {
+	if cachedResourceAndError != nil {
+		return cachedResourceAndError.Resource, cachedResourceAndError.Error
+	}
+	r, err := getUncachedResource()
+	cachedResourceAndError = &resourceCache{
+		Resource: r,
+		Error:    err,
+	}
+	return r, err
+}
+
+func getUncachedResource() (Resource, error) {
 	switch {
 	case gcp_metadata.OnGCE():
 		return GetGCEResource()
@@ -46,10 +65,6 @@ func GetResource() (Resource, error) {
 type UnrecognizedPlatformResource struct {
 }
 
-func (UnrecognizedPlatformResource) OTelResourceAttributes() map[string]string {
-	return nil
-}
-
 func (UnrecognizedPlatformResource) ProjectName() string {
 	return ""
 }
@@ -58,6 +73,10 @@ func (UnrecognizedPlatformResource) MonitoredResource() *monitoredres.MonitoredR
 	return nil
 }
 
+func (UnrecognizedPlatformResource) OTelResourceAttributes() map[string]string {
+	return nil
+}
+
 func (UnrecognizedPlatformResource) PrometheusStyleMetadata() map[string]string {
 	return nil
 }
diff --git a/confgenerator/testdata/goldens/builtin/golden/linux-dataproc/3c47867f5b9bfc4409551ecdaaac4562.lua b/confgenerator/testdata/goldens/builtin/golden/linux-dataproc/3c47867f5b9bfc4409551ecdaaac4562.lua
new file mode 100644
index 0000000000..c4acb5b5fa
--- /dev/null
+++ b/confgenerator/testdata/goldens/builtin/golden/linux-dataproc/3c47867f5b9bfc4409551ecdaaac4562.lua
@@ -0,0 +1,28 @@
+
+function process(tag, timestamp, record)
+local v = "test-cluster";
+(function(value)
+if record["logging.googleapis.com/labels"] == nil
+then
+record["logging.googleapis.com/labels"] = {}
+end
+record["logging.googleapis.com/labels"]["compute.googleapis.com/attributes/dataproc-cluster-name"] = value
+end)(v)
+local v = "test-uuid";
+(function(value)
+if record["logging.googleapis.com/labels"] == nil
+then
+record["logging.googleapis.com/labels"] = {}
+end
+record["logging.googleapis.com/labels"]["compute.googleapis.com/attributes/dataproc-cluster-uuid"] = value
+end)(v)
+local v = "test-region";
+(function(value)
+if record["logging.googleapis.com/labels"] == nil
+then
+record["logging.googleapis.com/labels"] = {}
+end
+record["logging.googleapis.com/labels"]["compute.googleapis.com/attributes/dataproc-region"] = value
+end)(v)
+return 2, timestamp, record
+end
diff --git a/confgenerator/testdata/goldens/builtin/golden/linux-dataproc/4d6012ff003886818fb9b9285b4af962.lua b/confgenerator/testdata/goldens/builtin/golden/linux-dataproc/4d6012ff003886818fb9b9285b4af962.lua
new file mode 100644
index 0000000000..c225be24d5
--- /dev/null
+++ b/confgenerator/testdata/goldens/builtin/golden/linux-dataproc/4d6012ff003886818fb9b9285b4af962.lua
@@ -0,0 +1,19 @@
+
+function process(tag, timestamp, record)
+local __field_0 = (function()
+return record["severity"]
+end)();
+(function(value)
+record["severity"] = value
+end)(nil);
+local v = __field_0;
+if v == "debug" then v = "DEBUG"
+elseif v == "error" then v = "ERROR"
+elseif v == "info" then v = "INFO"
+elseif v == "warn" then v = "WARNING"
+end
+(function(value)
+record["logging.googleapis.com/severity"] = value
+end)(v)
+return 2, timestamp, record
+end
diff --git a/confgenerator/testdata/goldens/builtin/golden/linux-dataproc/5fc5f42c16c9e1ab8292e3d42f74f3be.lua b/confgenerator/testdata/goldens/builtin/golden/linux-dataproc/5fc5f42c16c9e1ab8292e3d42f74f3be.lua
new file mode 100644
index 0000000000..c5465182b5
--- /dev/null
+++ b/confgenerator/testdata/goldens/builtin/golden/linux-dataproc/5fc5f42c16c9e1ab8292e3d42f74f3be.lua
@@ -0,0 +1,49 @@
+
+  function shallow_merge(record, parsedRecord)
+    -- If no exiting record exists
+    if (record == nil) then 
+        return parsedRecord
+    end
+    
+    for k, v in pairs(parsedRecord) do
+        record[k] = v
+    end
+
+    return record
+end
+
+function merge(record, parsedRecord)
+    -- If no exiting record exists
+    if record == nil then 
+        return parsedRecord
+    end
+    
+    -- Potentially overwrite or merge the original records.
+    for k, v in pairs(parsedRecord) do
+        -- If there is no conflict
+        if k == "logging.googleapis.com/logName" then 
+            -- Ignore the parsed payload since the logName is controlled
+            -- by the OpsAgent.
+        elseif k == "logging.googleapis.com/labels" then 
+            -- LogEntry.labels are basically a map[string]string and so only require a
+            -- shallow merge (one level deep merge).
+            record[k] = shallow_merge(record[k], v)
+        else
+            record[k] = v
+        end
+    end
+
+    return record
+end
+
+function parser_merge_record(tag, timestamp, record)
+    originalPayload = record["logging.googleapis.com/__tmp"]
+    if originalPayload == nil then
+        return 0, timestamp, record
+    end
+    
+    -- Remove original payload
+    record["logging.googleapis.com/__tmp"] = nil
+    record = merge(originalPayload, record)
+    return 2, timestamp, record
+end
diff --git a/confgenerator/testdata/goldens/builtin/golden/linux-dataproc/68d6a3a0e8edf37868e1bf94adc737f4.lua b/confgenerator/testdata/goldens/builtin/golden/linux-dataproc/68d6a3a0e8edf37868e1bf94adc737f4.lua
new file mode 100644
index 0000000000..16b933cb7d
--- /dev/null
+++ b/confgenerator/testdata/goldens/builtin/golden/linux-dataproc/68d6a3a0e8edf37868e1bf94adc737f4.lua
@@ -0,0 +1,38 @@
+
+function process(tag, timestamp, record)
+local __field_0 = (function()
+return record["message"]
+end)();
+local v = __field_0;
+if v == "LogParseErr" then v = "Ops Agent failed to parse logs, Code: LogParseErr, Documentation: https://cloud.google.com/stackdriver/docs/solutions/agents/ops-agent/troubleshoot-find-info"
+elseif v == "LogPipelineErr" then v = "Ops Agent logging pipeline failed, Code: LogPipelineErr, Documentation: https://cloud.google.com/stackdriver/docs/solutions/agents/ops-agent/troubleshoot-find-info"
+end
+(function(value)
+record["message"] = value
+end)(v)
+local v = "ops-agent";
+(function(value)
+if record["logging.googleapis.com/labels"] == nil
+then
+record["logging.googleapis.com/labels"] = {}
+end
+record["logging.googleapis.com/labels"]["agent.googleapis.com/health/agentKind"] = value
+end)(v)
+local v = "latest";
+(function(value)
+if record["logging.googleapis.com/labels"] == nil
+then
+record["logging.googleapis.com/labels"] = {}
+end
+record["logging.googleapis.com/labels"]["agent.googleapis.com/health/agentVersion"] = value
+end)(v)
+local v = "v1";
+(function(value)
+if record["logging.googleapis.com/labels"] == nil
+then
+record["logging.googleapis.com/labels"] = {}
+end
+record["logging.googleapis.com/labels"]["agent.googleapis.com/health/schemaVersion"] = value
+end)(v)
+return 2, timestamp, record
+end
diff --git a/confgenerator/testdata/goldens/builtin/golden/linux-dataproc/b4a0dead382dce7b4fe011d3f59fdb6d.lua b/confgenerator/testdata/goldens/builtin/golden/linux-dataproc/b4a0dead382dce7b4fe011d3f59fdb6d.lua
new file mode 100644
index 0000000000..6263563b66
--- /dev/null
+++ b/confgenerator/testdata/goldens/builtin/golden/linux-dataproc/b4a0dead382dce7b4fe011d3f59fdb6d.lua
@@ -0,0 +1,17 @@
+
+function parser_nest(tag, timestamp, record)
+  local nestedRecord = {}
+  local parseKey = "message"
+  for k, v in pairs(record) do
+      if k ~= parseKey then
+          nestedRecord[k] = v
+      end
+  end
+
+  local result = {}
+  result[parseKey] = record[parseKey]
+  result["logging.googleapis.com/__tmp"] = nestedRecord
+
+  return 2, timestamp, result
+end
+
diff --git a/confgenerator/testdata/goldens/builtin/golden/linux-dataproc/builtin_conf.yaml b/confgenerator/testdata/goldens/builtin/golden/linux-dataproc/builtin_conf.yaml
new file mode 100644
index 0000000000..cd84fc39b6
--- /dev/null
+++ b/confgenerator/testdata/goldens/builtin/golden/linux-dataproc/builtin_conf.yaml
@@ -0,0 +1,25 @@
+logging:
+  receivers:
+    syslog:
+      type: files
+      include_paths:
+      - /var/log/messages
+      - /var/log/syslog
+  service:
+    pipelines:
+      default_pipeline:
+        receivers: [syslog]
+metrics:
+  receivers:
+    hostmetrics:
+      type: hostmetrics
+      collection_interval: 60s
+  processors:
+    metrics_filter:
+      type: exclude_metrics
+      metrics_pattern: []
+  service:
+    pipelines:
+      default_pipeline:
+        receivers: [hostmetrics]
+        processors: [metrics_filter]
diff --git a/confgenerator/testdata/goldens/builtin/golden/linux-dataproc/f120d4527bd717cab023dbbe5fbdc332.lua b/confgenerator/testdata/goldens/builtin/golden/linux-dataproc/f120d4527bd717cab023dbbe5fbdc332.lua
new file mode 100644
index 0000000000..c09b885e11
--- /dev/null
+++ b/confgenerator/testdata/goldens/builtin/golden/linux-dataproc/f120d4527bd717cab023dbbe5fbdc332.lua
@@ -0,0 +1,42 @@
+
+function process(tag, timestamp, record)
+local __field_0 = (function()
+return record["agent.googleapis.com/log_file_path"]
+end)();
+local __field_1 = (function()
+if record["logging.googleapis.com/labels"] == nil
+then
+return nil
+end
+return record["logging.googleapis.com/labels"]["compute.googleapis.com/resource_name"]
+end)();
+local __field_2 = (function()
+return record["logging.googleapis.com/logName"]
+end)();
+(function(value)
+record["agent.googleapis.com/log_file_path"] = value
+end)(nil);
+local v = __field_0;
+(function(value)
+if record["logging.googleapis.com/labels"] == nil
+then
+record["logging.googleapis.com/labels"] = {}
+end
+record["logging.googleapis.com/labels"]["agent.googleapis.com/log_file_path"] = value
+end)(v)
+local v = __field_1;
+if v == nil then v = "" end;
+(function(value)
+if record["logging.googleapis.com/labels"] == nil
+then
+record["logging.googleapis.com/labels"] = {}
+end
+record["logging.googleapis.com/labels"]["compute.googleapis.com/resource_name"] = value
+end)(v)
+local v = __field_2;
+if v == nil then v = "syslog" end;
+(function(value)
+record["logging.googleapis.com/logName"] = value
+end)(v)
+return 2, timestamp, record
+end
diff --git a/confgenerator/testdata/goldens/builtin/golden/linux-dataproc/features.yaml b/confgenerator/testdata/goldens/builtin/golden/linux-dataproc/features.yaml
new file mode 100644
index 0000000000..6d4de0791f
--- /dev/null
+++ b/confgenerator/testdata/goldens/builtin/golden/linux-dataproc/features.yaml
@@ -0,0 +1,12 @@
+- module: logging
+  feature: service:pipelines
+  key: default_pipeline_overridden
+  value: "false"
+- module: metrics
+  feature: service:pipelines
+  key: default_pipeline_overridden
+  value: "false"
+- module: global
+  feature: default:self_log
+  key: default_self_log_file_collection
+  value: "true"
diff --git a/confgenerator/testdata/goldens/builtin/golden/linux-dataproc/fluent_bit_main.conf b/confgenerator/testdata/goldens/builtin/golden/linux-dataproc/fluent_bit_main.conf
new file mode 100644
index 0000000000..b803fa3274
--- /dev/null
+++ b/confgenerator/testdata/goldens/builtin/golden/linux-dataproc/fluent_bit_main.conf
@@ -0,0 +1,186 @@
+@SET buffers_dir=/var/lib/google-cloud-ops-agent/fluent-bit/buffers
+@SET logs_dir=/var/log/google-cloud-ops-agent
+
+[SERVICE]
+    Daemon                    off
+    Flush                     1
+    Log_Level                 info
+    dns.resolver              legacy
+    storage.backlog.mem_limit 50M
+    storage.checksum          off
+    storage.max_chunks_up     128
+    storage.metrics           on
+    storage.sync              normal
+
+[INPUT]
+    Name            fluentbit_metrics
+    Scrape_Interval 60
+    Scrape_On_Start True
+
+[INPUT]
+    Buffer_Chunk_Size 512k
+    Buffer_Max_Size   2M
+    DB                ${buffers_dir}/default_pipeline_syslog
+    DB.locking        true
+    Key               message
+    Mem_Buf_Limit     10M
+    Name              tail
+    Path              /var/log/messages,/var/log/syslog
+    Read_from_Head    True
+    Rotate_Wait       30
+    Skip_Long_Lines   On
+    Tag               default_pipeline.syslog
+    storage.type      filesystem
+
+[INPUT]
+    Buffer_Chunk_Size 512k
+    Buffer_Max_Size   2M
+    DB                ${buffers_dir}/ops-agent-fluent-bit
+    DB.locking        true
+    Key               message
+    Mem_Buf_Limit     10M
+    Name              tail
+    Path              ${logs_dir}/subagents/logging-module.log
+    Read_from_Head    True
+    Rotate_Wait       30
+    Skip_Long_Lines   On
+    Tag               ops-agent-fluent-bit
+    storage.type      memory
+
+[INPUT]
+    Buffer_Chunk_Size 512k
+    Buffer_Max_Size   2M
+    DB                ${buffers_dir}/ops-agent-health
+    DB.locking        true
+    Key               message
+    Mem_Buf_Limit     10M
+    Name              tail
+    Path              ${logs_dir}/health-checks.log
+    Read_from_Head    True
+    Rotate_Wait       30
+    Skip_Long_Lines   On
+    Tag               ops-agent-health
+    storage.type      memory
+
+[FILTER]
+    Match  default_pipeline.syslog
+    Name   lua
+    call   process
+    script f120d4527bd717cab023dbbe5fbdc332.lua
+
+[FILTER]
+    Match  ops-agent-fluent-bit
+    Name   lua
+    call   parser_nest
+    script b4a0dead382dce7b4fe011d3f59fdb6d.lua
+
+[FILTER]
+    Key_Name     message
+    Match        ops-agent-fluent-bit
+    Name         parser
+    Preserve_Key True
+    Reserve_Data True
+    Parser       ops-agent-fluent-bit.fluent-bit-self-log-regex-parsing
+
+[FILTER]
+    Match  ops-agent-fluent-bit
+    Name   lua
+    call   parser_merge_record
+    script 5fc5f42c16c9e1ab8292e3d42f74f3be.lua
+
+[FILTER]
+    Match  ops-agent-health
+    Name   lua
+    call   parser_nest
+    script b4a0dead382dce7b4fe011d3f59fdb6d.lua
+
+[FILTER]
+    Key_Name     message
+    Match        ops-agent-health
+    Name         parser
+    Reserve_Data True
+    Parser       ops-agent-health.health-checks-json
+
+[FILTER]
+    Match  ops-agent-health
+    Name   lua
+    call   parser_merge_record
+    script 5fc5f42c16c9e1ab8292e3d42f74f3be.lua
+
+[FILTER]
+    Match ops-agent-health
+    Name  grep
+    Regex severity INFO|ERROR|WARNING|DEBUG|info|error|warning|debug
+
+[FILTER]
+    Match ops-agent-fluent-bit
+    Name  rewrite_tag
+    Rule  message \[error\]\s\[lib\]\sbackend\sfailed ops-agent-health true
+
+[FILTER]
+    Name      modify
+    Match     ops-agent-health
+    Condition Key_value_matches message \[error\]\s\[lib\]\sbackend\sfailed
+    Set       message LogPipelineErr
+    Set       code LogPipelineErr
+
+[FILTER]
+    Match ops-agent-fluent-bit
+    Name  rewrite_tag
+    Rule  message \[error\]\s\[parser\]\scannot\sparse ops-agent-health true
+
+[FILTER]
+    Name      modify
+    Match     ops-agent-health
+    Condition Key_value_matches message \[error\]\s\[parser\]\scannot\sparse
+    Set       message LogParseErr
+    Set       code LogParseErr
+
+[FILTER]
+    Match  ops-agent-health
+    Name   lua
+    call   process
+    script 68d6a3a0e8edf37868e1bf94adc737f4.lua
+
+[FILTER]
+    Match  ops-agent-*
+    Name   lua
+    call   process
+    script 4d6012ff003886818fb9b9285b4af962.lua
+
+[FILTER]
+    Match  *
+    Name   lua
+    call   process
+    script 3c47867f5b9bfc4409551ecdaaac4562.lua
+
+[OUTPUT]
+    Match_Regex                   ^(default_pipeline\.syslog)$
+    Name                          stackdriver
+    Retry_Limit                   3
+    http_request_key              logging.googleapis.com/httpRequest
+    net.connect_timeout_log_error False
+    resource                      gce_instance
+    stackdriver_agent             Google-Cloud-Ops-Agent-Logging/latest (BuildDistro=build_distro;Platform=linux;ShortName=linux_platform;ShortVersion=linux_platform_version)
+    storage.total_limit_size      2G
+    tls                           On
+    tls.verify                    Off
+    workers                       8
+
+[OUTPUT]
+    Match_Regex                   ^(ops-agent-health|ops-agent-fluent-bit)$
+    Name                          stackdriver
+    Retry_Limit                   3
+    http_request_key              logging.googleapis.com/httpRequest
+    net.connect_timeout_log_error False
+    resource                      gce_instance
+    stackdriver_agent             Google-Cloud-Ops-Agent-Logging/latest (BuildDistro=build_distro;Platform=linux;ShortName=linux_platform;ShortVersion=linux_platform_version)
+    tls                           On
+    tls.verify                    Off
+    workers                       8
+
+[OUTPUT]
+    Match *
+    Name  prometheus_exporter
+    host  0.0.0.0
+    port  20202
diff --git a/confgenerator/testdata/goldens/builtin/golden/linux-dataproc/fluent_bit_parser.conf b/confgenerator/testdata/goldens/builtin/golden/linux-dataproc/fluent_bit_parser.conf
new file mode 100644
index 0000000000..e6c4f531ea
--- /dev/null
+++ b/confgenerator/testdata/goldens/builtin/golden/linux-dataproc/fluent_bit_parser.conf
@@ -0,0 +1,13 @@
+[PARSER]
+    Format      regex
+    Name        ops-agent-fluent-bit.fluent-bit-self-log-regex-parsing
+    Regex       (?<message>\[[ ]*(?<time>\d+\/\d+\/\d+ \d+:\d+:\d+)] \[[ ]*(?<severity>[a-z]+)\].*)
+    Time_Format %Y/%m/%d %H:%M:%S
+    Time_Key    time
+    Types       severity:string
+
+[PARSER]
+    Format      json
+    Name        ops-agent-health.health-checks-json
+    Time_Format %Y-%m-%dT%H:%M:%S%z
+    Time_Key    time
diff --git a/confgenerator/testdata/goldens/builtin/golden/linux-dataproc/otel.yaml b/confgenerator/testdata/goldens/builtin/golden/linux-dataproc/otel.yaml
new file mode 100644
index 0000000000..437d55de1c
--- /dev/null
+++ b/confgenerator/testdata/goldens/builtin/golden/linux-dataproc/otel.yaml
@@ -0,0 +1,488 @@
+exporters:
+  googlecloud:
+    metric:
+      instrumentation_library_labels: false
+      prefix: ""
+      resource_filters: []
+      service_resource_labels: false
+      skip_create_descriptor: true
+    user_agent: Google-Cloud-Ops-Agent-Metrics/latest (BuildDistro=build_distro;Platform=linux;ShortName=linux_platform;ShortVersion=linux_platform_version)
+processors:
+  agentmetrics/hostmetrics_0:
+    blank_label_metrics:
+    - system.cpu.utilization
+  filter/default__pipeline_hostmetrics_0:
+    metrics:
+      exclude:
+        match_type: regexp
+        metric_names: []
+  filter/fluentbit_0:
+    metrics:
+      include:
+        match_type: strict
+        metric_names:
+        - fluentbit_uptime
+        - fluentbit_stackdriver_requests_total
+        - fluentbit_stackdriver_proc_records_total
+        - fluentbit_stackdriver_retried_records_total
+  filter/hostmetrics_1:
+    metrics:
+      exclude:
+        match_type: strict
+        metric_names:
+        - system.cpu.time
+        - system.network.dropped
+        - system.filesystem.inodes.usage
+        - system.paging.faults
+        - system.disk.operation_time
+  filter/otel_0:
+    metrics:
+      include:
+        match_type: strict
+        metric_names:
+        - otelcol_process_uptime
+        - otelcol_process_memory_rss
+        - otelcol_grpc_io_client_completed_rpcs
+        - otelcol_googlecloudmonitoring_point_count
+  metricstransform/fluentbit_1:
+    transforms:
+    - action: update
+      include: fluentbit_uptime
+      new_name: agent/uptime
+      operations:
+      - action: toggle_scalar_data_type
+      - action: add_label
+        new_label: version
+        new_value: google-cloud-ops-agent-logging/latest-build_distro
+      - action: aggregate_labels
+        aggregation_type: sum
+        label_set:
+        - version
+    - action: update
+      include: fluentbit_stackdriver_requests_total
+      new_name: agent/request_count
+      operations:
+      - action: toggle_scalar_data_type
+      - action: update_label
+        label: status
+        new_label: response_code
+      - action: aggregate_labels
+        aggregation_type: sum
+        label_set:
+        - response_code
+    - action: update
+      include: fluentbit_stackdriver_proc_records_total
+      new_name: agent/log_entry_count
+      operations:
+      - action: toggle_scalar_data_type
+      - action: update_label
+        label: status
+        new_label: response_code
+      - action: aggregate_labels
+        aggregation_type: sum
+        label_set:
+        - response_code
+    - action: update
+      include: fluentbit_stackdriver_retried_records_total
+      new_name: agent/log_entry_retry_count
+      operations:
+      - action: toggle_scalar_data_type
+      - action: update_label
+        label: status
+        new_label: response_code
+      - action: aggregate_labels
+        aggregation_type: sum
+        label_set:
+        - response_code
+    - action: update
+      include: ^(.*)$$
+      match_type: regexp
+      new_name: agent.googleapis.com/$${1}
+  metricstransform/hostmetrics_2:
+    transforms:
+    - action: update
+      include: system.cpu.time
+      new_name: cpu/usage_time
+      operations:
+      - action: toggle_scalar_data_type
+      - action: update_label
+        label: cpu
+        new_label: cpu_number
+      - action: update_label
+        label: state
+        new_label: cpu_state
+    - action: update
+      include: system.cpu.utilization
+      new_name: cpu/utilization
+      operations:
+      - action: aggregate_labels
+        aggregation_type: mean
+        label_set:
+        - state
+        - blank
+      - action: update_label
+        label: blank
+        new_label: cpu_number
+      - action: update_label
+        label: state
+        new_label: cpu_state
+    - action: update
+      include: system.cpu.load_average.1m
+      new_name: cpu/load_1m
+    - action: update
+      include: system.cpu.load_average.5m
+      new_name: cpu/load_5m
+    - action: update
+      include: system.cpu.load_average.15m
+      new_name: cpu/load_15m
+    - action: update
+      include: system.disk.read_io
+      new_name: disk/read_bytes_count
+    - action: update
+      include: system.disk.write_io
+      new_name: disk/write_bytes_count
+    - action: update
+      include: system.disk.operations
+      new_name: disk/operation_count
+    - action: update
+      include: system.disk.io_time
+      new_name: disk/io_time
+      operations:
+      - action: experimental_scale_value
+        experimental_scale: 1000.0
+      - action: toggle_scalar_data_type
+    - action: update
+      include: system.disk.weighted_io_time
+      new_name: disk/weighted_io_time
+      operations:
+      - action: experimental_scale_value
+        experimental_scale: 1000.0
+      - action: toggle_scalar_data_type
+    - action: update
+      include: system.disk.average_operation_time
+      new_name: disk/operation_time
+      operations:
+      - action: experimental_scale_value
+        experimental_scale: 1000.0
+      - action: toggle_scalar_data_type
+    - action: update
+      include: system.disk.pending_operations
+      new_name: disk/pending_operations
+      operations:
+      - action: toggle_scalar_data_type
+    - action: update
+      include: system.disk.merged
+      new_name: disk/merged_operations
+    - action: update
+      include: system.filesystem.usage
+      new_name: disk/bytes_used
+      operations:
+      - action: toggle_scalar_data_type
+      - action: aggregate_labels
+        aggregation_type: max
+        label_set:
+        - device
+        - state
+    - action: update
+      include: system.filesystem.utilization
+      new_name: disk/percent_used
+      operations:
+      - action: aggregate_labels
+        aggregation_type: max
+        label_set:
+        - device
+        - state
+    - action: update
+      include: system.memory.usage
+      new_name: memory/bytes_used
+      operations:
+      - action: toggle_scalar_data_type
+      - action: aggregate_label_values
+        aggregated_values:
+        - slab_reclaimable
+        - slab_unreclaimable
+        aggregation_type: sum
+        label: state
+        new_value: slab
+    - action: update
+      include: system.memory.utilization
+      new_name: memory/percent_used
+      operations:
+      - action: aggregate_label_values
+        aggregated_values:
+        - slab_reclaimable
+        - slab_unreclaimable
+        aggregation_type: sum
+        label: state
+        new_value: slab
+    - action: update
+      include: system.network.io
+      new_name: interface/traffic
+      operations:
+      - action: update_label
+        label: interface
+        new_label: device
+      - action: update_label
+        label: direction
+        value_actions:
+        - new_value: rx
+          value: receive
+        - new_value: tx
+          value: transmit
+    - action: update
+      include: system.network.errors
+      new_name: interface/errors
+      operations:
+      - action: update_label
+        label: interface
+        new_label: device
+      - action: update_label
+        label: direction
+        value_actions:
+        - new_value: rx
+          value: receive
+        - new_value: tx
+          value: transmit
+    - action: update
+      include: system.network.packets
+      new_name: interface/packets
+      operations:
+      - action: update_label
+        label: interface
+        new_label: device
+      - action: update_label
+        label: direction
+        value_actions:
+        - new_value: rx
+          value: receive
+        - new_value: tx
+          value: transmit
+    - action: update
+      include: system.network.connections
+      new_name: network/tcp_connections
+      operations:
+      - action: toggle_scalar_data_type
+      - action: delete_label_value
+        label: protocol
+        label_value: udp
+      - action: update_label
+        label: state
+        new_label: tcp_state
+      - action: aggregate_labels
+        aggregation_type: sum
+        label_set:
+        - tcp_state
+      - action: add_label
+        new_label: port
+        new_value: all
+    - action: update
+      include: system.processes.created
+      new_name: processes/fork_count
+    - action: update
+      include: system.processes.count
+      new_name: processes/count_by_state
+      operations:
+      - action: toggle_scalar_data_type
+      - action: update_label
+        label: status
+        new_label: state
+    - action: update
+      include: system.paging.usage
+      new_name: swap/bytes_used
+      operations:
+      - action: toggle_scalar_data_type
+    - action: update
+      include: system.paging.utilization
+      new_name: swap/percent_used
+    - action: insert
+      include: swap/percent_used
+      new_name: pagefile/percent_used
+      operations:
+      - action: aggregate_labels
+        aggregation_type: sum
+        label_set:
+        - state
+    - action: update
+      include: system.paging.operations
+      new_name: swap/io
+      operations:
+      - action: aggregate_labels
+        aggregation_type: sum
+        label_set:
+        - direction
+      - action: update_label
+        label: direction
+        value_actions:
+        - new_value: in
+          value: page_in
+        - new_value: out
+          value: page_out
+    - action: update
+      include: process.cpu.time
+      new_name: processes/cpu_time
+      operations:
+      - action: experimental_scale_value
+        experimental_scale: 1e+06
+      - action: toggle_scalar_data_type
+      - action: add_label
+        new_label: process
+        new_value: all
+      - action: delete_label_value
+        label: state
+        label_value: wait
+      - action: update_label
+        label: state
+        new_label: user_or_syst
+      - action: update_label
+        label: user_or_syst
+        value_actions:
+        - new_value: syst
+          value: system
+    - action: update
+      include: process.disk.read_io
+      new_name: processes/disk/read_bytes_count
+      operations:
+      - action: add_label
+        new_label: process
+        new_value: all
+    - action: update
+      include: process.disk.write_io
+      new_name: processes/disk/write_bytes_count
+      operations:
+      - action: add_label
+        new_label: process
+        new_value: all
+    - action: update
+      include: process.memory.usage
+      new_name: processes/rss_usage
+      operations:
+      - action: toggle_scalar_data_type
+      - action: add_label
+        new_label: process
+        new_value: all
+    - action: update
+      include: process.memory.virtual
+      new_name: processes/vm_usage
+      operations:
+      - action: toggle_scalar_data_type
+      - action: add_label
+        new_label: process
+        new_value: all
+    - action: update
+      include: ^(.*)$$
+      match_type: regexp
+      new_name: agent.googleapis.com/$${1}
+  metricstransform/otel_1:
+    transforms:
+    - action: update
+      include: otelcol_process_uptime
+      new_name: agent/uptime
+      operations:
+      - action: toggle_scalar_data_type
+      - action: add_label
+        new_label: version
+        new_value: google-cloud-ops-agent-metrics/latest-build_distro
+      - action: aggregate_labels
+        aggregation_type: sum
+        label_set:
+        - version
+    - action: update
+      include: otelcol_process_memory_rss
+      new_name: agent/memory_usage
+      operations:
+      - action: aggregate_labels
+        aggregation_type: sum
+        label_set: []
+    - action: update
+      include: otelcol_grpc_io_client_completed_rpcs
+      new_name: agent/api_request_count
+      operations:
+      - action: toggle_scalar_data_type
+      - action: update_label
+        label: grpc_client_status
+        new_label: state
+      - action: aggregate_labels
+        aggregation_type: sum
+        label_set:
+        - state
+    - action: update
+      include: otelcol_googlecloudmonitoring_point_count
+      new_name: agent/monitoring/point_count
+      operations:
+      - action: toggle_scalar_data_type
+      - action: aggregate_labels
+        aggregation_type: sum
+        label_set:
+        - status
+    - action: update
+      include: ^(.*)$$
+      match_type: regexp
+      new_name: agent.googleapis.com/$${1}
+  resourcedetection/_global_0:
+    detectors:
+    - gcp
+receivers:
+  hostmetrics/hostmetrics:
+    collection_interval: 60s
+    scrapers:
+      cpu: {}
+      disk: {}
+      filesystem: {}
+      load: {}
+      memory: {}
+      network: {}
+      paging: {}
+      process:
+        mute_process_name_error: true
+      processes: {}
+  prometheus/fluentbit:
+    config:
+      scrape_configs:
+      - job_name: logging-collector
+        metrics_path: /metrics
+        scrape_interval: 1m
+        static_configs:
+        - targets:
+          - 0.0.0.0:20202
+  prometheus/otel:
+    config:
+      scrape_configs:
+      - job_name: otel-collector
+        scrape_interval: 1m
+        static_configs:
+        - targets:
+          - 0.0.0.0:20201
+service:
+  pipelines:
+    metrics/default__pipeline_hostmetrics:
+      exporters:
+      - googlecloud
+      processors:
+      - agentmetrics/hostmetrics_0
+      - filter/hostmetrics_1
+      - metricstransform/hostmetrics_2
+      - filter/default__pipeline_hostmetrics_0
+      - resourcedetection/_global_0
+      receivers:
+      - hostmetrics/hostmetrics
+    metrics/fluentbit:
+      exporters:
+      - googlecloud
+      processors:
+      - filter/fluentbit_0
+      - metricstransform/fluentbit_1
+      - resourcedetection/_global_0
+      receivers:
+      - prometheus/fluentbit
+    metrics/otel:
+      exporters:
+      - googlecloud
+      processors:
+      - filter/otel_0
+      - metricstransform/otel_1
+      - resourcedetection/_global_0
+      receivers:
+      - prometheus/otel
+  telemetry:
+    metrics:
+      address: 0.0.0.0:20201
diff --git a/integration_test/agents/agents.go b/integration_test/agents/agents.go
index 60508b1c97..42e3e1a540 100644
--- a/integration_test/agents/agents.go
+++ b/integration_test/agents/agents.go
@@ -839,18 +839,22 @@ func RecommendedMachineType(platform string) string {
 
 // CommonSetup sets up the VM for testing.
 func CommonSetup(t *testing.T, platform string) (context.Context, *logging.DirectoryLogger, *gce.VM) {
-	return CommonSetupWithExtraCreateArguments(t, platform, []string{})
+	return CommonSetupWithExtraCreateArguments(t, platform, nil)
 }
 
 // CommonSetupWithExtraCreateArguments sets up the VM for testing with extra creation arguments for the `gcloud compute instances create` command.
 func CommonSetupWithExtraCreateArguments(t *testing.T, platform string, extraCreateArguments []string) (context.Context, *logging.DirectoryLogger, *gce.VM) {
+	return CommonSetupWithExtraCreateArgumentsAndMetadata(t, platform, extraCreateArguments, nil)
+}
+// CommonSetupWithExtraCreateArgumentsAndMetadata sets up the VM for testing with extra creation arguments for the `gcloud compute instances create` command and additional metadata.
+func CommonSetupWithExtraCreateArgumentsAndMetadata(t *testing.T, platform string, extraCreateArguments []string, additionalMetadata map[string]string) (context.Context, *logging.DirectoryLogger, *gce.VM) {
 	t.Helper()
 	ctx, cancel := context.WithTimeout(context.Background(), gce.SuggestedTimeout)
 	t.Cleanup(cancel)
 
 	logger := gce.SetupLogger(t)
 	logger.ToMainLog().Println("Calling SetupVM(). For details, see VM_initialization.txt.")
-	vm := gce.SetupVM(ctx, t, logger.ToFile("VM_initialization.txt"), gce.VMOptions{Platform: platform, MachineType: RecommendedMachineType(platform), ExtraCreateArguments: extraCreateArguments})
+	vm := gce.SetupVM(ctx, t, logger.ToFile("VM_initialization.txt"), gce.VMOptions{Platform: platform, MachineType: RecommendedMachineType(platform), ExtraCreateArguments: extraCreateArguments, Metadata: additionalMetadata})
 	logger.ToMainLog().Printf("VM is ready: %#v", vm)
 	t.Cleanup(func() {
 		RunOpsAgentDiagnostics(ctx, logger, vm)
diff --git a/integration_test/ops_agent_test.go b/integration_test/ops_agent_test.go
index a9a3d20a1a..8403dd5023 100644
--- a/integration_test/ops_agent_test.go
+++ b/integration_test/ops_agent_test.go
@@ -3751,6 +3751,34 @@ func TestLoggingSelfLogs(t *testing.T) {
 	})
 }
 
+func TestLoggingDataprocAttributes(t *testing.T) {
+	t.Parallel()
+	gce.RunForEachPlatform(t, func(t *testing.T, platform string) {
+		t.Parallel()
+		if gce.IsWindows(platform) {
+			t.Skip("Dataproc tests aren't supported on Windows")
+		}
+		ctx, logger, vm := agents.CommonSetupWithExtraCreateArgumentsAndMetadata(t, platform, nil, map[string]string{
+			"dataproc-cluster-name": "my-test-cluster",
+		})
+
+		if err := agents.SetupOpsAgent(ctx, logger.ToMainLog(), vm, ""); err != nil {
+			t.Fatal(err)
+		}
+
+
+		if err := writeToSystemLog(ctx, logger.ToMainLog(), vm, "123456789"); err != nil {
+			t.Fatal(err)
+		}
+
+		tag := systemLogTagForPlatform(vm.Platform)
+		query := fmt.Sprintf(`labels."compute.googleapis.com/attributes/dataproc-cluster-name"="my-test-cluster" AND %s`, logMessageQueryForPlatform(vm.Platform, "123456789"))
+		if err := gce.WaitForLog(ctx, logger.ToMainLog(), vm, tag, time.Hour, query); err != nil {
+			t.Error(err)
+		}
+	})
+}
+
 func diagnosticsLivenessChecker(ctx context.Context, logger *log.Logger, vm *gce.VM) error {
 	time.Sleep(3 * time.Minute)
 	// Query for a metric sent by the diagnostics service from the last
diff --git a/internal/platform/platform.go b/internal/platform/platform.go
index ed5f668832..48f1d290d6 100644
--- a/internal/platform/platform.go
+++ b/internal/platform/platform.go
@@ -30,6 +30,8 @@ type Platform struct {
 	HostInfo           *host.InfoStat
 	HasNvidiaGpu       bool
 	ResourceOverride   resourcedetector.Resource
+	// Resource override only for GCE metadata unit testing
+	TestGCEResourceOverride resourcedetector.Resource
 }
 
 type Type int
@@ -91,3 +93,13 @@ func (p Platform) Name() string {
 	}
 	panic(fmt.Sprintf("unknown type %v", p.Type))
 }
+
+func (p Platform) GetResource() (resourcedetector.Resource, error) {
+	if p.TestGCEResourceOverride != nil {
+		return p.TestGCEResourceOverride, nil
+	} else if p.ResourceOverride != nil {
+		return p.ResourceOverride, nil
+	}
+	r, err := resourcedetector.GetResource()
+	return r, err
+}