diff --git a/core/services/synchronization/chip_ingress_batch_worker.go b/core/services/synchronization/chip_ingress_batch_worker.go index c59252a79d6..15ebd21a3df 100644 --- a/core/services/synchronization/chip_ingress_batch_worker.go +++ b/core/services/synchronization/chip_ingress_batch_worker.go @@ -70,13 +70,28 @@ func (cw *chipIngressBatchWorker) Send(ctx context.Context) { ctx, cancel := context.WithTimeout(ctx, cw.sendTimeout) defer cancel() - _, err := cw.chipClient.PublishBatch(ctx, batch) + resp, err := cw.chipClient.PublishBatch(ctx, batch) if err != nil { cw.lggr.Warnf("Could not send telemetry via ChIP ingress: %v", err) TelemetryClientMessagesSendErrors.WithLabelValues(chipIngress, string(cw.telemType)).Inc() return } + // Inspect per-event results for partial delivery errors. When transaction_enabled=false + // (the default), the server can accept some events while rejecting others. + // + // This is intentionally not logged: it's a per-event, often persistent server-side + // condition (e.g. missing schema) that would otherwise WARN on every send interval + // (as low as 500ms) for as long as it lasts. ChipIngressPartialDeliveryDropped + // (telemetry_type, error_code) is the intended signal for alerting/investigation. + for _, result := range resp.GetResults() { + if e := result.GetError(); e != nil { + code := e.GetErrorCode().String() + TelemetryClientMessagesDropped.WithLabelValues(chipIngress, string(cw.telemType)).Inc() + ChipIngressPartialDeliveryDropped.WithLabelValues(string(cw.telemType), code).Inc() + } + } + TelemetryClientMessagesSent.WithLabelValues(chipIngress, string(cw.telemType)).Inc() if cw.logging { cw.lggr.Debugw("Successfully sent telemetry to ChIP ingress", "contractID", cw.contractID, "telemType", cw.telemType, "batchSize", len(batch.Events)) diff --git a/core/services/synchronization/chip_ingress_batch_worker_test.go b/core/services/synchronization/chip_ingress_batch_worker_test.go index 8dc68ad5fbd..f3acf2fe642 100644 --- a/core/services/synchronization/chip_ingress_batch_worker_test.go +++ b/core/services/synchronization/chip_ingress_batch_worker_test.go @@ -6,8 +6,10 @@ import ( "time" cepb "github.com/cloudevents/sdk-go/binding/format/protobuf/v2/pb" + "github.com/prometheus/client_golang/prometheus/testutil" "github.com/stretchr/testify/assert" "github.com/stretchr/testify/require" + "go.uber.org/zap" "google.golang.org/grpc" "github.com/smartcontractkit/chainlink-common/pkg/chipingress" @@ -25,6 +27,38 @@ func (noopChipIngressPublisher) PublishBatch(ctx context.Context, batch *pb.Clou return &pb.PublishResponse{}, nil } +type partialChipIngressPublisher struct { + resp *pb.PublishResponse +} + +func (p partialChipIngressPublisher) Publish(ctx context.Context, event *cepb.CloudEvent, opts ...grpc.CallOption) (*pb.PublishResponse, error) { + return &pb.PublishResponse{}, nil +} + +func (p partialChipIngressPublisher) PublishBatch(ctx context.Context, batch *pb.CloudEventBatch, opts ...grpc.CallOption) (*pb.PublishResponse, error) { + return p.resp, nil +} + +func (p partialChipIngressPublisher) Ping(ctx context.Context, req *pb.EmptyRequest, opts ...grpc.CallOption) (*pb.PingResponse, error) { + return &pb.PingResponse{}, nil +} + +func (p partialChipIngressPublisher) StreamEvents(ctx context.Context, opts ...grpc.CallOption) (grpc.BidiStreamingClient[pb.StreamEventsRequest, pb.StreamEventsResponse], error) { + return nil, nil +} + +func (p partialChipIngressPublisher) RegisterSchema(ctx context.Context, req *pb.RegisterSchemaRequest, opts ...grpc.CallOption) (*pb.RegisterSchemaResponse, error) { + return &pb.RegisterSchemaResponse{}, nil +} + +func (p partialChipIngressPublisher) Close() error { + return nil +} + +func (p partialChipIngressPublisher) RegisterSchemas(ctx context.Context, schemas ...*pb.Schema) (map[string]int, error) { + return nil, nil +} + func (noopChipIngressPublisher) Ping(ctx context.Context, req *pb.EmptyRequest, opts ...grpc.CallOption) (*pb.PingResponse, error) { return &pb.PingResponse{}, nil } @@ -45,6 +79,80 @@ func (noopChipIngressPublisher) RegisterSchemas(ctx context.Context, schemas ... return nil, nil } +func TestChipIngressBatchWorker_Send_PartialDelivery(t *testing.T) { + t.Parallel() + // Verify that Send records per-event partial-delivery errors as metrics without + // logging: partial delivery is a per-event, often persistent condition (e.g. missing + // schema), so logging it would spam at fleet-wide volume. The + // ChipIngressPartialDeliveryDropped metric (telemetry_type, error_code) is the + // intended signal. + partialResp := &pb.PublishResponse{ + Results: []*pb.PublishResult{ + { + EventId: "evt-1", + Error: &pb.PublishError{ + ErrorCode: pb.PublishErrorCode_PUBLISH_ERROR_CODE_VALIDATION_FAILED, + Reason: "schema not found", + }, + }, + { + EventId: "evt-2", + Error: &pb.PublishError{ + ErrorCode: pb.PublishErrorCode_PUBLISH_ERROR_CODE_VALIDATION_FAILED, + Reason: "schema not found", + }, + }, + }, + } + publisher := partialChipIngressPublisher{resp: partialResp} + + lggr, observed := logger.TestLoggerObserved(t, zap.WarnLevel) + chTelemetry := make(chan TelemPayload, 5) + worker := NewChipIngressBatchWorker( + 2, + time.Second, + publisher, + chTelemetry, + "0xabc", + OCR, + lggr, + true, + ) + + chTelemetry <- TelemPayload{ + Telemetry: []byte("payload1"), + TelemType: OCR, + ContractID: "0xabc", + Domain: "data-feeds", + Entity: "ocr.v1.telemetry", + ChainSelector: 7700, + Network: "EVM", + ChainID: "1", + } + chTelemetry <- TelemPayload{ + Telemetry: []byte("payload2"), + TelemType: OCR, + ContractID: "0xabc", + Domain: "data-feeds", + Entity: "ocr.v1.telemetry", + ChainSelector: 7700, + Network: "EVM", + ChainID: "1", + } + + code := pb.PublishErrorCode_PUBLISH_ERROR_CODE_VALIDATION_FAILED.String() + before := testutil.ToFloat64(ChipIngressPartialDeliveryDropped.WithLabelValues(string(OCR), code)) + + require.NotPanics(t, func() { worker.Send(t.Context()) }) + assert.Empty(t, chTelemetry) + + after := testutil.ToFloat64(ChipIngressPartialDeliveryDropped.WithLabelValues(string(OCR), code)) + assert.Equal(t, float64(2), after-before, "expected both failed events to increment the metric") + + logs := observed.FilterMessage("chip ingress partial delivery errors") + assert.Zero(t, logs.Len(), "partial delivery drops must not be logged") +} + func TestChipIngressBatchWorker_BuildCloudEventBatch(t *testing.T) { maxBatchSize := 3 chTelemetry := make(chan TelemPayload, 10) diff --git a/core/services/synchronization/metrics.go b/core/services/synchronization/metrics.go index d6bafb7b0d6..329bcf542cb 100644 --- a/core/services/synchronization/metrics.go +++ b/core/services/synchronization/metrics.go @@ -30,4 +30,9 @@ var ( Name: "telemetry_client_workers", Help: "Number of telemetry workers", }, []string{"endpoint", "telemetry_type"}) + + ChipIngressPartialDeliveryDropped = promauto.NewCounterVec(prometheus.CounterOpts{ + Name: "chip_ingress_partial_delivery_dropped", + Help: "Number of chip ingress events dropped due to per-event server rejection (partial delivery)", + }, []string{"telemetry_type", "error_code"}) )