Partialwritetestcase

integration_test/ops_agent_test/main_test.go

@@ -4981,6 +4981,83 @@ traces:
 	)
 }
 
+func TestOTLPMetricsPartialSuccess(t *testing.T) {
+	t.Parallel()
+	RunForEachImageAndFeatureFlag(t, []string{agents.OtlpHttpExporterFeatureFlag}, func(t *testing.T, imageSpec string, feature string) {
+		t.Parallel()
+		ctx, logger, vm := setupMainLogAndVM(t, imageSpec)
+		otlpConfig := `
+combined:
+  receivers:
+    otlp:
+      type: otlp
+      grpc_endpoint: 0.0.0.0:4317
+      metrics_mode: googlecloudmonitoring
+metrics:
+  service:
+    pipelines:
+      otlp:
+        receivers:
+        - otlp
+traces:
+  service:
+    pipelines:
+`
+		if err := agents.SetupOpsAgentWithFeatureFlag(ctx, logger, vm, otlpConfig, feature); err != nil {
+			t.Fatal(err)
+		}
+
+		time.Sleep(2 * time.Minute)
+
+		// Generate metric traffic with dummy app
+		metricFile, err := testdataDir.Open(path.Join("testdata", "otlp", "metrics_partial_success.go"))
+		if err != nil {
+			t.Fatal(err)
+		}
+		defer metricFile.Close()
+		if err := installGolang(ctx, logger, vm); err != nil {
+			t.Fatal(err)
+		}
+		serviceName := "otlp-metric-partial-success-test"
+		if err = runGoCode(
+			ctx, logger, vm, metricFile,
+			"-service_name", serviceName,
+		); err != nil {
+			t.Fatal(err)
+		}
+
+		// 1. Verify good metric arrives
+		expectedMetricType := "workload.googleapis.com/otlp.test.gauge_good"
+		_, err = gce.WaitForMetricSeries(ctx, logger, vm, expectedMetricType, time.Hour, nil, false, 1)
+		if err != nil {
+			t.Error(err)
+		}
+
+		// 2. Verify self-metric for failed points
+		// Wait for point_count with status != "OK" to prove failed points are recorded
+		// failedFilter := []string{`metric.labels.status != "OK"`}
+		failedMetric := metadata.ExpectedMetric{
+			MetricSpec: metadata.MetricSpec{
+				Type:               "agent.googleapis.com/agent/monitoring/point_count",
+				Kind:               metric.MetricDescriptor_CUMULATIVE.String(),
+				ValueType:          metric.MetricDescriptor_INT64.String(),
+				MonitoredResources: []string{"gce_instance"},
+				Labels: []*metadata.MetricLabel{
+					{
+						Name:       "status",
+						ValueRegex: "INVALID_ARGUMENT",
+					},
+				},
+			},
+		}
+		failedFilter := []string{`metric.labels.status != "OK"`}
+		if err := waitForAndAssertMetric(ctx, logger, vm, time.Hour, &failedMetric, failedFilter, false); err != nil {
+			t.Error(err)
+		}
+	})
+}
+
+
 func TestOTLPMetricsOTLP(t *testing.T) {
 	t.Parallel()
 	gce.RunForEachImage(t, func(t *testing.T, imageSpec string) {

integration_test/ops_agent_test/testdata/otlp/metrics_partial_success.go

@@ -0,0 +1,77 @@
+package main
+
+import (
+	"context"
+	"flag"
+	"log"
+	"time"
+
+	"go.opentelemetry.io/collector/pdata/pcommon"
+	"go.opentelemetry.io/collector/pdata/pmetric"
+	"go.opentelemetry.io/collector/pdata/pmetric/pmetricotlp"
+	"google.golang.org/grpc"
+	"google.golang.org/grpc/credentials/insecure"
+)
+
+var (
+	flagServiceName       = flag.String("service_name", "", "service.name attribute value")
+	flagServiceNamespace  = flag.String("service_namespace", "", "service.namespace attribute value")
+	flagServiceInstanceID = flag.String("service_instance_id", "", "service.instance.id attribute value")
+)
+
+func main() {
+	flag.Parse()
+	ctx := context.Background()
+
+	conn, err := grpc.DialContext(ctx, "localhost:4317", grpc.WithTransportCredentials(insecure.NewCredentials()), grpc.WithBlock())
+	if err != nil {
+		log.Fatalf("did not connect: %v", err)
+	}
+	defer conn.Close()
+
+	client := pmetricotlp.NewGRPCClient(conn)
+
+	md := pmetric.NewMetrics()
+	rl := md.ResourceMetrics().AppendEmpty()
+	if *flagServiceName != "" {
+		rl.Resource().Attributes().PutStr("service.name", *flagServiceName)
+	}
+	if *flagServiceNamespace != "" {
+		rl.Resource().Attributes().PutStr("service.namespace", *flagServiceNamespace)
+	}
+	if *flagServiceInstanceID != "" {
+		rl.Resource().Attributes().PutStr("service.instance.id", *flagServiceInstanceID)
+	}
+
+	sl := rl.ScopeMetrics().AppendEmpty()
+	sl.Scope().SetName("foo")
+
+	// 1. Good metric
+	m1 := sl.Metrics().AppendEmpty()
+	m1.SetName("otlp.test.gauge_good")
+	m1.SetEmptyGauge()
+	dp1 := m1.Gauge().DataPoints().AppendEmpty()
+	ts := pcommon.NewTimestampFromTime(time.Now())
+	dp1.SetTimestamp(ts)
+	dp1.SetDoubleValue(33)
+
+	// 2. Metric with duplicate points (triggers partial success)
+	m2 := sl.Metrics().AppendEmpty()
+	m2.SetName("otlp.test.gauge_duplicate")
+	m2.SetEmptyGauge()
+
+	dp2 := m2.Gauge().DataPoints().AppendEmpty()
+	dp2.SetTimestamp(ts)
+	dp2.SetDoubleValue(5)
+
+	dp3 := m2.Gauge().DataPoints().AppendEmpty()
+	dp3.SetTimestamp(ts) // Same timestamp!
+	dp3.SetDoubleValue(10)
+
+	req := pmetricotlp.NewExportRequestFromMetrics(md)
+	_, err = client.Export(ctx, req)
+	if err != nil {
+		log.Fatalf("could not export: %v", err)
+	}
+	log.Println("Successfully exported metrics")
+}