JITSU-138: fix Datadog reserved-status collision; synthesize otlp message lines (#1475)

Follow-up to `JITSU-138`, from end-to-end testing against Datadog's
agentless OTLP intake (which merges the structured body's top-level keys
into root log attributes).

**Architecture (reworked per review): the otlp destination stays fully
payload-agnostic** — `otlp.go` is untouched relative to `newjitsu`. Body
adaptation happens where envelopes are produced to the Kafka topic, and
**only for bodies the pipeline constructs itself**:

1. **Go producer** (`eventslog/kafka_events_log.go`), for `bulker_batch`
/ `bulker_stream` records (bulker-built `bulker.State` / stream-status
shapes):
- top-level `status` → `record_status`: `COMPLETED`/`FAILED` values
collide with Datadog's reserved status attribute and made **every**
record render as `critical` regardless of severity (reproduced in
isolation with controlled probe payloads)
- synthesized `message` when absent/empty/null: `bulker_batch COMPLETED:
2 rows → events39`, with a rune-safe-truncated error preview on failures
- the eventId hash is computed from the unadapted body — ids and billing
dedup unchanged
2. **Rotor producer** (`kafka-events-store.ts`): dead-letter bodies
(rotor-built `{payload, error}`) get their `message` line at
construction.

**Function-log bodies are user data and are never touched** — no key
renames, no synthesized fields.

Body stays structured (kvlist) — no stringification. Covered by
`TestKafkaEventsLogAdaptsOwnedBodies` (rename, synthesis, truncation,
function-body-untouched) and rotor `deadLetterMessage` tests (incl.
surrogate-pair safety).

Docs note: jitsucom/websites#60.

🤖 Generated with [Claude Code](https://claude.com/claude-code)

Author: absorbb

bulker/eventslog/kafka_events_log.go

@@ -3,11 +3,14 @@ package eventslog
 import (
 	"crypto/sha256"
 	"encoding/hex"
+	"encoding/json"
 	"fmt"
+	"strings"
 	"time"
 
 	"github.com/jitsucom/bulker/jitsubase/jsonorder"
 	"github.com/jitsucom/bulker/jitsubase/logging"
+	"github.com/jitsucom/bulker/jitsubase/types"
 )
 
 // Live Events observability export fan-out (JITSU-138): a KafkaEventsLogService
@@ -156,6 +159,15 @@ func (k *KafkaEventsLogService) PostAsync(event *ActorEvent) {
 		envelope.ConnectionId = event.ActorId
 	case EventTypeBatch, EventTypeProcessed:
 		envelope.DestinationId = event.ActorId
+		// bulker_batch / bulker_stream bodies are constructed by bulker itself
+		// (bulker.State / stream status shapes), so we own their top-level keys
+		// and adapt them for observability backends here, at the producer —
+		// the otlp destination stays payload-agnostic. Function-log bodies are
+		// user data and are never touched. The eventId hash above is computed
+		// from the unadapted body, so the id is unaffected
+		if adapted := adaptOwnedBody(body, string(event.EventType)); adapted != nil {
+			envelope.Body = adapted
+		}
 	}
 	payload, err := jsonorder.Marshal(&envelope)
 	if err != nil {
@@ -174,6 +186,71 @@ func (k *KafkaEventsLogService) PostAsync(event *ActorEvent) {
 	k.produceBillingRecord(&envelope, timestamp)
 }
 
+// adaptOwnedBody adapts a bulker-owned record body for observability
+// backends: `status` is renamed to `record_status` (Datadog's agentless OTLP
+// intake merges top-level body keys into root log attributes, where `status`
+// is reserved for severity — COMPLETED/FAILED values made every record render
+// as `critical`), and a short human-readable `message` is synthesized when
+// absent or empty so log list views show a line instead of a blank. Returns
+// nil (caller keeps the original body) when the body can't be re-parsed
+func adaptOwnedBody(bodyJSON []byte, eventType string) types.Json {
+	obj := types.NewJson(0)
+	if err := jsonorder.Unmarshal(bodyJSON, &obj); err != nil || obj == nil {
+		return nil
+	}
+	if v, ok := obj.Get("status"); ok {
+		obj.Delete("status")
+		obj.Set("record_status", v)
+	}
+	if v, ok := obj.Get("message"); !ok || v == "" || v == nil {
+		obj.Set("message", stateMessage(eventType, obj))
+	}
+	return obj
+}
+
+// stateMessage: e.g. "bulker_batch COMPLETED: 2 rows → events39"
+func stateMessage(eventType string, body types.Json) string {
+	var b strings.Builder
+	b.WriteString(eventType)
+	if s := body.GetS("record_status"); s != "" {
+		b.WriteString(" " + s)
+	}
+	if v, ok := body.Get("processedRows"); ok {
+		if n, isNum := asInt64(v); isNum {
+			fmt.Fprintf(&b, ": %d rows", n)
+		}
+	}
+	if rep, ok := body.Get("representation"); ok {
+		if repObj, isObj := rep.(types.Json); isObj {
+			if name := repObj.GetS("name"); name != "" {
+				b.WriteString(" → " + name)
+			}
+		}
+	}
+	if e := body.GetS("error"); e != "" {
+		if r := []rune(e); len(r) > 140 {
+			e = string(r[:140]) + "…"
+		}
+		b.WriteString(" — " + e)
+	}
+	return b.String()
+}
+
+func asInt64(v any) (int64, bool) {
+	switch n := v.(type) {
+	case json.Number:
+		i, err := n.Int64()
+		return i, err == nil
+	case float64:
+		return int64(n), true
+	case int64:
+		return n, true
+	case int:
+		return int64(n), true
+	}
+	return 0, false
+}
+
 // produceBillingRecord emits one active_incoming record per exported envelope.
 // The composed key {eventId}_0_{secondsWithinHour} with an hour-truncated
 // timestamp follows ingest's buildSyncMetrics; dedup happens downstream via

bulker/eventslog/kafka_events_log_test.go

@@ -3,6 +3,7 @@ package eventslog
 import (
 	"encoding/json"
 	"fmt"
+	"strings"
 	"testing"
 	"time"
 
@@ -203,6 +204,54 @@ func TestKafkaEventsLogOrderedMapBodyFidelity(t *testing.T) {
 	require.Contains(t, payload, `"lastMappedRow":{"message_id":"m1","nested":{"a":1}}`)
 }
 
+func TestKafkaEventsLogAdaptsOwnedBodies(t *testing.T) {
+	service, produced := newTestKafkaService("", map[string]bool{"ws1": true})
+
+	// bulker_batch: status renamed (Datadog reserved-attribute collision) and a
+	// message line synthesized from status/rows/target table
+	batch := testActorEvent()
+	batch.EventType = EventTypeBatch
+	batch.ActorId = "dst1"
+	rep := types.NewJson(1)
+	rep.Set("name", "events39")
+	body := types.NewJson(3)
+	body.Set("status", "COMPLETED")
+	body.Set("processedRows", 2)
+	body.Set("representation", rep)
+	batch.Event = body
+	service.PostAsync(batch)
+	require.Len(t, *produced, 2)
+	envelope := map[string]any{}
+	require.NoError(t, json.Unmarshal((*produced)[0].payload, &envelope))
+	exported := envelope["body"].(map[string]any)
+	require.NotContains(t, exported, "status")
+	require.Equal(t, "COMPLETED", exported["record_status"])
+	require.Equal(t, "bulker_batch COMPLETED: 2 rows → events39", exported["message"])
+
+	// failed stream: error preview appended, long errors truncated
+	stream := testActorEvent()
+	stream.EventType = EventTypeProcessed
+	stream.ActorId = "dst1"
+	stream.Event = map[string]any{"status": "FAILED", "error": strings.Repeat("x", 200)}
+	service.PostAsync(stream)
+	require.NoError(t, json.Unmarshal((*produced)[2].payload, &envelope))
+	exported = envelope["body"].(map[string]any)
+	msg := exported["message"].(string)
+	require.True(t, strings.HasPrefix(msg, "bulker_stream FAILED — "))
+	require.True(t, strings.HasSuffix(msg, "…"))
+	require.Less(t, len(msg), 200)
+
+	// function logs are user data: status untouched, no message synthesized
+	fn := testActorEvent()
+	fn.Event = map[string]any{"type": "log-info", "status": "custom-user-value"}
+	service.PostAsync(fn)
+	require.NoError(t, json.Unmarshal((*produced)[4].payload, &envelope))
+	exported = envelope["body"].(map[string]any)
+	require.Equal(t, "custom-user-value", exported["status"])
+	require.NotContains(t, exported, "record_status")
+	require.NotContains(t, exported, "message")
+}
+
 func TestKafkaEventsLogNoBillingOnEnqueueFailure(t *testing.T) {
 	attempted := &[]string{}
 	service := NewKafkaEventsLogService(KafkaEventsLogConfig{

services/rotor/__tests__/kafka-events-store.test.ts

@@ -1,5 +1,25 @@
 import { describe, expect, it } from "vitest";
-import { exportEventId } from "../src/lib/kafka-events-store";
+import { deadLetterMessage, exportEventId } from "../src/lib/kafka-events-store";
+
+describe("dead-letter message synthesis", () => {
+  it("uses string errors and Error-like objects, falls back to bare label", () => {
+    expect(deadLetterMessage("boom")).toBe("dead-letter — boom");
+    expect(deadLetterMessage(new Error("kaput"))).toBe("dead-letter — kaput");
+    expect(deadLetterMessage({ message: "obj msg" })).toBe("dead-letter — obj msg");
+    expect(deadLetterMessage(null)).toBe("dead-letter");
+    expect(deadLetterMessage({ code: 42 })).toBe("dead-letter");
+  });
+
+  it("truncates long errors rune-safely", () => {
+    const msg = deadLetterMessage("x".repeat(200));
+    expect(msg).toMatch(/…$/);
+    expect(Array.from(msg).length).toBeLessThan(200 - 140 + 160);
+    // astral characters must not be split into surrogate halves
+    const emoji = deadLetterMessage("💥".repeat(150));
+    expect(emoji.includes("�")).toBe(false);
+    expect(Array.from(emoji).slice(-2).join("")).toBe("💥…");
+  });
+});
 
 describe("export eventId content hash", () => {
   it("matches the pinned cross-language vector (same recipe as Go eventslog.ExportEventId)", () => {

services/rotor/src/lib/kafka-events-store.ts

@@ -54,6 +54,15 @@ type ExportRecord = {
   body: Record<string, any>;
 };
 
+// mirrors the Go producer's stateMessage: short human line for backends that
+// surface a message field, with a rune-safe 140-char error preview
+export function deadLetterMessage(error: any): string {
+  const errText = typeof error === "string" ? error : error && typeof error.message === "string" ? error.message : "";
+  const runes = Array.from(errText);
+  const preview = runes.length > 140 ? runes.slice(0, 140).join("") + "…" : errText;
+  return preview ? `dead-letter — ${preview}` : "dead-letter";
+}
+
 export function createKafkaEventsStore(): EventsStore {
   const prefix = serverEnv.KAFKA_TOPIC_PREFIX || "";
   // unset = billing/metrics emission disabled for this deployment, matching
@@ -189,7 +198,11 @@ export function createKafkaEventsStore(): EventsStore {
         level: "error",
         actorId: connectionId,
         connectionId,
-        body: { payload, error },
+        // dead-letter bodies are rotor-owned, so the human-readable message
+        // line for observability backends is synthesized here at the producer —
+        // the otlp destination stays payload-agnostic. Function-log bodies
+        // already carry their own message and are never touched
+        body: { payload, error, message: deadLetterMessage(error) },
       });
     },
     close() {