otel-collector/exporter/zapexporter/exporter.go

package zapexporter

import (
	"context"
	"errors"
	"fmt"
	"math/rand/v2"
	"os"
	"strconv"
	"strings"
	"sync"
	"sync/atomic"
	"time"

	luxzap "github.com/luxfi/zap"
	"go.opentelemetry.io/collector/component"
	"go.opentelemetry.io/collector/consumer/consumererror"
	"go.opentelemetry.io/collector/exporter"
	"go.opentelemetry.io/collector/pdata/pcommon"
	"go.opentelemetry.io/collector/pdata/plog"
	"go.opentelemetry.io/collector/pdata/pmetric"
	"go.opentelemetry.io/collector/pdata/ptrace"
	"go.uber.org/zap"
)

// zapExporter ships pdata to an o11y ZAP receiver as a JSON batch inside a
// luxfi/zap envelope. See wire.go for why the previous OTLP-protobuf-over-
// zap-proto/http approach could not work and what it cost.
//
// The receiver answers NOTHING — its handler returns a nil message on every path
// and its own doc says "the sender doesn't wait on the response". So this uses
// Send, not Call. Waiting for a reply is what produced two days of `i/o timeout`
// against a receiver behaving exactly as designed.
type zapExporter struct {
	cfg      *Config
	settings exporter.Settings

	// mu guards node and peerID and is NEVER held across network I/O. It used
	// to wrap the whole of send(), which meant one call parked in the kernel
	// held every other consumer behind it: the exporter went to zero
	// throughput and zero errors at the same time, which is the one
	// combination no queue and no retry budget can survive. Measured on the
	// busiest node in the fleet, 135s after a cloud restart: 9,854 records
	// accepted, 0 sent, 0 failed, queue climbing toward its ceiling.
	mu     sync.Mutex
	node   *luxzap.Node
	peerID string

	// dialing keeps at most one dial in flight. A second caller is turned
	// away rather than queued, because queueing behind a dial is the same
	// trap as the mutex: the dial's handshake read has no deadline.
	dialing atomic.Bool

	// stalls counts consecutive attempts that made no progress — abandoned on the
	// caller's deadline, or refused outright. See resetNode for what it is for.
	stalls atomic.Int64

	// lastReset is when the node was last rebuilt, as UnixNano. It is the floor
	// under rebuild frequency: see resetNode.
	lastReset atomic.Int64
}

// stallsBeforeReset is how many consecutive deadline-abandoned attempts mean
// the connection is not slow but wedged. At the exporter's 5s timeout this is
// ~15s of no progress plus backoff — far above any normal slow write, far
// below the tens of minutes a wedged connection was observed to hold.
const stallsBeforeReset = 3

// initialDialTimeout bounds the best-effort dial in Start. See Start.
const initialDialTimeout = 5 * time.Second

// minResetInterval is the floor under how often the node may be rebuilt.
//
// A rebuild is the right answer to a connection that will never work again, and
// the wrong answer to a receiver that is merely down: while cloud is restarting,
// EVERY dial fails, so an unfloored budget would mint a fresh identity several
// times a second across every agent in the fleet and hand the receiver a crowd
// of dead registrations to reap the moment it came back. Thirty seconds is far
// longer than a wedge takes to detect (three failures at a 5s timeout) and far
// shorter than an operator would take to notice, so a real refusal costs one
// rebuild and an outage costs two a minute.
const minResetInterval = 30 * time.Second

// errDialInFlight and errNoNode are the two failures that mean "ask again",
// not "this connection is finished". They must never count toward the stall
// budget: the first is ordinary concurrency between the three signal pipelines
// sharing this exporter, and counting it would let healthy parallelism trigger
// rebuilds.
var (
	errDialInFlight = errors.New("dial already in flight")
	errNoNode       = errors.New("zap node not started")
)

// stalled reports whether an attempt made no progress for a reason a new node
// could fix.
//
// The budget's original reading was context.DeadlineExceeded only, which covers
// a peer that accepts the socket and stops answering. It does not cover the
// other half, and the other half is the one with no way out: luxfi/zap admits
// exactly ONE connection per peer NodeID, so an agent whose registration the
// receiver still holds is answered with EOF at handshake — the dial fails
// immediately, cleanly, and forever, since redialing presents the same refused
// identity. A budget that ignores that error waits for a deadline that never
// comes, which is why a wedged agent stayed wedged with its queue at the
// ceiling until it was deleted by hand.
//
// So: everything counts EXCEPT the two "ask again" sentinels above.
func stalled(err error) bool {
	if err == nil {
		return false
	}
	return !errors.Is(err, errDialInFlight) && !errors.Is(err, errNoNode)
}

func newExporter(set exporter.Settings, cfg *Config) *zapExporter {
	return &zapExporter{cfg: cfg, settings: set}
}

// within runs fn and stops waiting when ctx does.
//
// luxfi/zap's TCP path takes no deadline anywhere: ConnectDirectID dials with
// a 5s timeout and then does a plain blocking handshake read, Conn.Send writes
// with no write deadline, and Node.Send accepts a context it consults only on
// the QUIC path. So a peer that accepts the connection and then stops reading
// — exactly what cloud looks like while it restarts — parks the caller
// indefinitely, and the exporter's own timeout setting cannot touch it.
//
// This is the only place that can bound it. fn keeps running (there is no way
// to interrupt a blocking read on a socket nobody set a deadline on), but the
// exporter stops waiting and reports a retryable failure instead of going
// silent. The goroutine ends when the kernel finally gives up on the socket.
func within(ctx context.Context, fn func() error) error {
	done := make(chan error, 1)
	go func() { done <- fn() }()
	select {
	case err := <-done:
		return err
	case <-ctx.Done():
		return ctx.Err()
	}
}

// hostname is the per-process half of the ZAP node identity. A var so a test can
// stand two "hosts" up inside one process; nothing in production reassigns it.
var hostname = os.Hostname

// nodeID is this exporter's identity on the ZAP wire, and it MUST be unique
// across every process that dials the same receiver.
//
// luxfi/zap keys its connection table by peer NodeID and admits exactly ONE
// connection per ID: handleConn sees the duplicate and returns WITHOUT writing a
// handshake reply ("Don't send handshake - outgoing side will get EOF"), so the
// loser reads EOF mid-handshake. An ID derived only from the component ID is the
// same string in every copy of one config, so N senders sharing a config get one
// winner and N-1 permanently locked out — they are not slow or throttled, they
// are refused, forever, with a message that reads like a network fault.
//
// Measured 2026-08-01 before this fix: of 25 otel-agent DaemonSet pods all
// claiming "otel-agent-zap", exactly ONE had zero errors; the other 24 logged
// `zapexporter: connect cloud.hanzo.svc:4318: EOF` ~26x/min each until their
// sending queues overflowed, discarding ~88% of the fleet's logs. The queue was
// the symptom; this identity was the cause.
//
// Hostname is the cross-process half — in Kubernetes it is the pod name, unique
// per node for a DaemonSet. The component ID is the in-process half, so two zap
// exporters in one collector aimed at one receiver stay distinct. Both halves
// are DERIVED, never configured: a config knob here would just be a way to set
// the same value on every node, which is precisely the bug it would exist to
// prevent.
func (e *zapExporter) nodeID(incarnation string) string {
	host, err := hostname()
	if err != nil || host == "" {
		// A missing hostname must not silently re-collide. A random suffix keeps
		// this process distinguishable; the cost is a new identity per restart,
		// which is strictly better than being refused by the receiver.
		host = "unknown-" + strconv.FormatUint(rand.Uint64(), 36)
	}
	id := "otel-agent-" + host
	if cid := e.settings.ID.String(); cid != "" {
		id += "-" + strings.ReplaceAll(cid, "/", "-")
	}
	// The RESTART half. Hostname is unique per pod but NOT per process: a
	// container that restarts inside its pod comes back with the same name and
	// reclaims an identity the receiver may still be holding — and the duplicate
	// rule above then refuses the new process with EOF. That refusal never clears
	// from the sender's side, so the pod crash-loops and the only cure is
	// deleting it so a new name is minted.
	//
	// Measured 2026-08-08: otel-gateway logged
	// `connect 10.124.0.71:4319: EOF` 56 times across 10 restarts and delivered
	// nothing until it was deleted by hand — four separate incidents in one day,
	// each ended by the same manual delete.
	//
	// The incarnation is minted per node instance and never reused, so both a
	// process restart and an in-place rebuild arrive as a NEW peer rather than a
	// duplicate of a corpse. The cost is a short-lived stale entry the receiver
	// reaps on its own, which is strictly better than a sender that can never
	// reconnect.
	return id + "-" + incarnation
}

// newIncarnation identifies one NODE INSTANCE — not this host, and not this
// process either.
//
// Per-process was already right for the restart case nodeID describes. It is
// not enough once the exporter can rebuild its node in place to escape a
// wedged connection: a rebuild inside one process would reuse the process's
// nonce, the receiver would still be holding the old registration, and the
// duplicate rule above would refuse the new node with EOF — permanently,
// trading a stall that ends when TCP gives up for one that never ends. Minting
// the nonce per node instance keeps every rebuild a genuinely new peer, and a
// process that never rebuilds is unaffected: its first node mints one nonce
// and keeps it.
func newIncarnation() string { return strconv.FormatUint(rand.Uint64(), 36) }

// Start brings up the local ZAP node. A failed dial is deliberately NOT fatal:
// this runs as a DaemonSet on every node and the collector may be unreachable at
// boot, so it must not take the agent down — send() reconnects.
func (e *zapExporter) Start(ctx context.Context, _ component.Host) error {
	node, err := e.startNode()
	if err != nil {
		return err
	}
	e.mu.Lock()
	e.node = node
	e.mu.Unlock()
	// The initial dial is best-effort and must not hold up startup. It is not
	// merely slow when the peer is unreachable: a peer that accepts the socket
	// and never answers the handshake parks this call forever (luxfi/zap sets no
	// read deadline), so an unbounded dial here means a collector that never
	// reports ready and never tails a single file. Bounded, a boot-time
	// unreachable collector costs one timeout and send() picks it up later,
	// which is what the comment above has always promised.
	dialCtx, cancel := context.WithTimeout(ctx, initialDialTimeout)
	defer cancel()
	if err := e.connect(dialCtx); err != nil {
		e.settings.Logger.Debug("OTLZ exporter: initial connect failed; will retry on export",
			zap.String("endpoint", e.cfg.Endpoint), zap.Error(err))
	}
	e.settings.Logger.Info("OTLZ exporter ready (wire=luxfi/zap envelope, JSON batch)",
		zap.String("endpoint", e.cfg.Endpoint),
		zap.String("transport", e.transportName()))
	return nil
}

// startNode builds and starts one ZAP node with a fresh identity. Both the
// initial Start and a post-stall rebuild go through here, so the two can never
// drift on transport selection or identity.
func (e *zapExporter) startNode() (*luxzap.Node, error) {
	cfg := luxzap.NodeConfig{
		NodeID:      e.nodeID(newIncarnation()),
		ServiceType: "_o11y._tcp",
		Port:        0,
		NoDiscovery: true,
	}
	// A UDS endpoint needs no transport selection — luxfi/zap's Network() reads
	// the address shape, so a socket path dials unix on both ends. QUIC is the
	// remote case, and its TLS 1.3 negotiates X25519MLKEM768 (X-Wing) by default
	// on Go 1.26, so a cross-machine hop is quantum-secure without this exporter
	// configuring any crypto.
	if e.cfg.Transport == "quic" {
		cfg.Transport = luxzap.TransportQUIC
	}
	node := luxzap.NewNode(cfg)
	if err := node.Start(); err != nil {
		return nil, fmt.Errorf("zapexporter: zap node start (transport=%s): %w", e.transportName(), err)
	}
	return node, nil
}

// transportName reports the transport actually in use, for the one log line an
// operator reads to confirm a hop is not silently plaintext across machines.
func (e *zapExporter) transportName() string {
	if isSocketPath(e.cfg.Endpoint) {
		return "uds"
	}
	if e.cfg.Transport == "quic" {
		return "quic+x25519mlkem768"
	}
	return "tcp"
}

func (e *zapExporter) Shutdown(context.Context) error {
	e.mu.Lock()
	node := e.node
	e.node, e.peerID = nil, ""
	e.mu.Unlock()
	if node != nil {
		// Outside the lock: Stop closes conns, and closing a wedged one can
		// block for as long as its socket does.
		node.Stop()
	}
	return nil
}

// connect dials once at startup. Failure is not fatal; send() dials on demand.
func (e *zapExporter) connect(ctx context.Context) error {
	_, err := e.peer(ctx)
	return err
}

// peer returns a peer ID to send to, dialing if there is not one cached.
//
// At most one dial runs at a time and it is bounded by ctx. A caller that
// arrives while a dial is in flight is told to retry rather than made to wait:
// the dial can block for as long as the peer's TCP stack feels like it, and
// making every other consumer share that fate is what took the exporter to
// zero throughput.
func (e *zapExporter) peer(ctx context.Context) (string, error) {
	e.mu.Lock()
	id, node := e.peerID, e.node
	e.mu.Unlock()
	if id != "" {
		return id, nil
	}
	if node == nil {
		return "", errNoNode
	}
	if !e.dialing.CompareAndSwap(false, true) {
		return "", errDialInFlight
	}
	err := within(ctx, func() error {
		defer e.dialing.Store(false)
		if err := node.ConnectDirect(e.cfg.Endpoint); err != nil {
			return err
		}
		peers := node.Peers()
		if len(peers) == 0 {
			return fmt.Errorf("connected to %s but no peer ID", e.cfg.Endpoint)
		}
		e.mu.Lock()
		if e.node == node {
			e.peerID = peers[0]
		}
		e.mu.Unlock()
		return nil
	})
	if err != nil {
		// The dial goroutine owns the flag on the paths it reaches; if we
		// abandoned it on ctx, it will clear the flag when it finally returns.
		return "", err
	}
	e.mu.Lock()
	defer e.mu.Unlock()
	return e.peerID, nil
}

// forget drops the cached peer if it is still the one that just failed.
func (e *zapExporter) forget(id string) {
	e.mu.Lock()
	defer e.mu.Unlock()
	if e.peerID == id {
		e.peerID = ""
	}
}

// resetNode abandons the current ZAP node and builds a new one.
//
// luxfi/zap has no way to drop a single peer: Node exposes Start, Stop, Send,
// Peers and ConnectDirect, and ConnectDirect is documented idempotent — it
// "returns the existing peer's NodeID and drops the duplicate dial" — so
// redialing a wedged peer hands back the same wedged Conn, whose per-conn
// mutex every subsequent send then queues on. Rebuilding the node is the only
// way to let go of a connection that has stopped draining.
//
// That is not a theoretical state. After a cloud restart an agent logged
// "Connected to peer" and then sent zero records for 34 minutes with no error
// on any path, while its queue filled toward the ceiling where it would begin
// dropping. Only the process restarting cleared it. This does that much, and
// only that much, without taking the agent down.
func (e *zapExporter) resetNode() {
	// The floor, and the serialiser. Three signal pipelines share this exporter
	// and all of them can be failing at once, so the compare-and-swap is what
	// makes concurrent stalls produce ONE rebuild rather than three.
	now := time.Now().UnixNano()
	last := e.lastReset.Load()
	if last != 0 && now-last < int64(minResetInterval) {
		return
	}
	if !e.lastReset.CompareAndSwap(last, now) {
		return
	}

	e.mu.Lock()
	old := e.node
	e.peerID = ""
	e.mu.Unlock()

	if old != nil {
		go old.Stop() // may block on the same wedged socket; do not wait on it
	}
	node, err := e.startNode()
	if err != nil {
		e.settings.Logger.Warn("OTLZ exporter: could not rebuild the ZAP node after a stall",
			zap.String("endpoint", e.cfg.Endpoint), zap.Error(err))
		return
	}
	e.mu.Lock()
	e.node = node
	e.mu.Unlock()
	e.stalls.Store(0)
	e.settings.Logger.Info("OTLZ exporter: rebuilt the ZAP node after a stalled connection",
		zap.String("endpoint", e.cfg.Endpoint))
}

// noteStall records a deadline-abandoned attempt and rebuilds the node once
// enough of them have happened in a row to mean wedged rather than slow.
func (e *zapExporter) noteStall() {
	if e.stalls.Add(1) >= stallsBeforeReset {
		e.resetNode()
	}
}

// send delivers one envelope, redialing if the cached peer is stale.
//
// A failure is returned as RETRYABLE (exporterhelper backs off and retries with
// the batch intact) rather than swallowed. Dropping a node's logs quietly is how
// a fleet loses two days of telemetry without one error surfacing — and a send
// that never returns is the same loss wearing a healthier face, because the
// queue behind it fills and starts refusing while every counter reads zero.
func (e *zapExporter) send(ctx context.Context, msg *luxzap.Message) error {
	peerID, err := e.peer(ctx)
	if err != nil {
		if stalled(err) {
			e.noteStall()
		}
		return fmt.Errorf("zapexporter: connect %s: %w", e.cfg.Endpoint, err)
	}
	e.mu.Lock()
	node := e.node
	e.mu.Unlock()
	if node == nil {
		return errNoNode
	}
	if err := within(ctx, func() error { return node.Send(ctx, peerID, msg) }); err != nil {
		e.forget(peerID) // stale connection; next call redials
		if stalled(err) {
			e.noteStall()
		}
		return fmt.Errorf("zapexporter: send to %s: %w", e.cfg.Endpoint, err)
	}
	e.stalls.Store(0)
	return nil
}

func (e *zapExporter) pushTraces(ctx context.Context, td ptrace.Traces) error {
	for _, batch := range tracesToBatches(td) {
		msg, err := encode(msgSpanBatch, batch)
		if err != nil {
			return consumererror.NewPermanent(err)
		}
		if err := e.send(ctx, msg); err != nil {
			return err
		}
	}
	return nil
}

func (e *zapExporter) pushLogs(ctx context.Context, ld plog.Logs) error {
	for _, batch := range logsToBatches(ld) {
		msg, err := encode(msgLogBatch, batch)
		if err != nil {
			return consumererror.NewPermanent(err)
		}
		if err := e.send(ctx, msg); err != nil {
			return err
		}
	}
	return nil
}

func (e *zapExporter) pushMetrics(ctx context.Context, md pmetric.Metrics) error {
	batches, skipped := metricsToBatches(md)
	if skipped > 0 {
		// Named rather than silently dropped: the wire is Prometheus-shaped
		// (counter/gauge/histogram/summary), so an OTLP type with no counterpart
		// has nowhere to go. Approximating one would corrupt the series.
		e.settings.Logger.Warn("OTLZ exporter: metric types with no wire counterpart were skipped",
			zap.Int("skipped", skipped))
	}
	for _, batch := range batches {
		msg, err := encode(msgMetricBatch, batch)
		if err != nil {
			return consumererror.NewPermanent(err)
		}
		if err := e.send(ctx, msg); err != nil {
			return err
		}
	}
	return nil
}

// ── pdata -> wire ───────────────────────────────────────────────────────────
//
// One ResourceSpans/ResourceLogs becomes one batch, because a batch carries
// exactly one service's identity (appName + resource). Instrumentation scope is
// dropped: the wire has no field for it, and inventing one would put a name on
// the wire that the emitter never sent.

// resourceOf splits a resource into the batch's service identity and the rest of
// its attributes. service.name and service.version are lifted out because the
// wire carries them as their own fields.
func resourceOf(attrs pcommon.Map) (appName, version string, rest map[string]string) {
	rest = map[string]string{}
	attrs.Range(func(k string, v pcommon.Value) bool {
		switch k {
		case "service.name":
			appName = v.AsString()
		case "service.version":
			version = v.AsString()
		default:
			rest[k] = v.AsString()
		}
		return true
	})
	if len(rest) == 0 {
		rest = nil
	}
	return appName, version, rest
}

func attrsOf(attrs pcommon.Map) map[string]any {
	if attrs.Len() == 0 {
		return nil
	}
	out := make(map[string]any, attrs.Len())
	attrs.Range(func(k string, v pcommon.Value) bool {
		out[k] = v.AsRaw()
		return true
	})
	return out
}

func tracesToBatches(td ptrace.Traces) []*spanBatch {
	var out []*spanBatch
	rss := td.ResourceSpans()
	for i := 0; i < rss.Len(); i++ {
		rs := rss.At(i)
		app, ver, res := resourceOf(rs.Resource().Attributes())
		b := &spanBatch{AppName: app, Version: ver, Resource: res, Spans: []span{}}
		sss := rs.ScopeSpans()
		for j := 0; j < sss.Len(); j++ {
			spans := sss.At(j).Spans()
			for k := 0; k < spans.Len(); k++ {
				b.Spans = append(b.Spans, translateSpan(spans.At(k)))
			}
		}
		if len(b.Spans) > 0 {
			out = append(out, b)
		}
	}
	return out
}

func translateSpan(s ptrace.Span) span {
	tid := s.TraceID()
	sid := s.SpanID()
	dst := span{
		TraceID: fmt.Sprintf("%x", tid[:]),
		SpanID:  fmt.Sprintf("%x", sid[:]),
		Name:    s.Name(),
		// LOWERCASE, because that is the vocabulary the receiver reads.
		//
		// pdata renders SpanKind title-cased ("Server"), and the o11y receiver
		// matches "server" or "SPAN_KIND_SERVER" and defaults everything else to
		// "internal". "Server" is neither, so every span this exporter sent —
		// server, client, producer, consumer alike — landed in event.span as
		// kind='internal'. Verified end-to-end 2026-08-02: a span sent with
		// kind=SPAN_KIND_SERVER stored as internal.
		//
		// Nothing errors on that, which is why it survived: the rows arrive, the
		// trace renders, and only the entry/exit distinction is gone — so no
		// latency question that starts "at the edge" can be asked of the data.
		Kind:        strings.ToLower(s.Kind().String()),
		StartUnixNs: int64(s.StartTimestamp()),
		EndUnixNs:   int64(s.EndTimestamp()),
		Attributes:  attrsOf(s.Attributes()),
		StatusMsg:   s.Status().Message(),
	}
	if psid := s.ParentSpanID(); !psid.IsEmpty() {
		dst.ParentSpanID = fmt.Sprintf("%x", psid[:])
	}
	if c := s.Status().Code(); c != ptrace.StatusCodeUnset {
		dst.StatusCode = c.String()
	}
	evs := s.Events()
	for i := 0; i < evs.Len(); i++ {
		ev := evs.At(i)
		dst.Events = append(dst.Events, spanEvent{
			Name:       ev.Name(),
			TimeUnixNs: int64(ev.Timestamp()),
			Attributes: attrsOf(ev.Attributes()),
		})
	}
	return dst
}

func logsToBatches(ld plog.Logs) []*logBatch {
	var out []*logBatch
	rls := ld.ResourceLogs()
	for i := 0; i < rls.Len(); i++ {
		rl := rls.At(i)
		app, ver, res := resourceOf(rl.Resource().Attributes())
		b := &logBatch{AppName: app, Version: ver, Resource: res, Records: []logRecord{}}
		sls := rl.ScopeLogs()
		for j := 0; j < sls.Len(); j++ {
			recs := sls.At(j).LogRecords()
			for k := 0; k < recs.Len(); k++ {
				b.Records = append(b.Records, translateLog(recs.At(k)))
			}
		}
		if len(b.Records) > 0 {
			out = append(out, b)
		}
	}
	return out
}

func translateLog(r plog.LogRecord) logRecord {
	dst := logRecord{
		TimeUnixNs:         int64(r.Timestamp()),
		ObservedTimeUnixNs: int64(r.ObservedTimestamp()),
		Severity:           int(r.SeverityNumber()),
		SeverityText:       r.SeverityText(),
		Body:               r.Body().AsString(),
		Attributes:         attrsOf(r.Attributes()),
		EventName:          r.EventName(),
	}
	if tid := r.TraceID(); !tid.IsEmpty() {
		dst.TraceID = fmt.Sprintf("%x", tid[:])
	}
	if sid := r.SpanID(); !sid.IsEmpty() {
		dst.SpanID = fmt.Sprintf("%x", sid[:])
	}
	return dst
}

// metricsToBatches converts what maps and COUNTS what does not.
//
// The wire is Prometheus-shaped, so Gauge and Sum become gauge/counter and
// Histogram becomes histogram with cumulative buckets. ExponentialHistogram and
// Summary have no faithful representation here and are reported as skipped
// rather than approximated.
func metricsToBatches(md pmetric.Metrics) (out []*metricBatch, skipped int) {
	rms := md.ResourceMetrics()
	for i := 0; i < rms.Len(); i++ {
		rm := rms.At(i)
		app, ver, res := resourceOf(rm.Resource().Attributes())
		b := &metricBatch{AppName: app, Version: ver, Resource: res, Families: []metricFamily{}}
		sms := rm.ScopeMetrics()
		for j := 0; j < sms.Len(); j++ {
			ms := sms.At(j).Metrics()
			for k := 0; k < ms.Len(); k++ {
				fam, ts, ok := translateMetric(ms.At(k))
				if !ok {
					skipped++
					continue
				}
				if ts > b.TimestampNs {
					b.TimestampNs = ts
				}
				b.Families = append(b.Families, fam)
			}
		}
		if len(b.Families) > 0 {
			out = append(out, b)
		}
	}
	return out, skipped
}

func translateMetric(m pmetric.Metric) (fam metricFamily, latestNs int64, ok bool) {
	fam = metricFamily{Name: m.Name(), Help: m.Description()}
	switch m.Type() {
	case pmetric.MetricTypeGauge:
		fam.Type = "gauge"
		dps := m.Gauge().DataPoints()
		for i := 0; i < dps.Len(); i++ {
			dp := dps.At(i)
			v := numberValue(dp)
			fam.Metrics = append(fam.Metrics, metric{Labels: labelsOf(dp.Attributes()), Value: &v})
			latestNs = maxNs(latestNs, int64(dp.Timestamp()))
		}
	case pmetric.MetricTypeSum:
		fam.Type = "counter"
		dps := m.Sum().DataPoints()
		for i := 0; i < dps.Len(); i++ {
			dp := dps.At(i)
			v := numberValue(dp)
			fam.Metrics = append(fam.Metrics, metric{Labels: labelsOf(dp.Attributes()), Value: &v})
			latestNs = maxNs(latestNs, int64(dp.Timestamp()))
		}
	case pmetric.MetricTypeHistogram:
		fam.Type = "histogram"
		dps := m.Histogram().DataPoints()
		for i := 0; i < dps.Len(); i++ {
			dp := dps.At(i)
			count := dp.Count()
			sum := dp.Sum()
			mm := metric{Labels: labelsOf(dp.Attributes()), SampleCount: &count, SampleSum: &sum}
			// OTLP bucket counts are per-bucket; this wire wants cumulative.
			var cum uint64
			bounds := dp.ExplicitBounds()
			counts := dp.BucketCounts()
			for bi := 0; bi < bounds.Len() && bi < counts.Len(); bi++ {
				cum += counts.At(bi)
				mm.Buckets = append(mm.Buckets, bucket{UpperBound: bounds.At(bi), CumulativeCount: cum})
			}
			fam.Metrics = append(fam.Metrics, mm)
			latestNs = maxNs(latestNs, int64(dp.Timestamp()))
		}
	default:
		return metricFamily{}, 0, false
	}
	if len(fam.Metrics) == 0 {
		return metricFamily{}, 0, false
	}
	return fam, latestNs, true
}

func numberValue(dp pmetric.NumberDataPoint) float64 {
	if dp.ValueType() == pmetric.NumberDataPointValueTypeInt {
		return float64(dp.IntValue())
	}
	return dp.DoubleValue()
}

func labelsOf(attrs pcommon.Map) map[string]string {
	if attrs.Len() == 0 {
		return nil
	}
	out := make(map[string]string, attrs.Len())
	attrs.Range(func(k string, v pcommon.Value) bool {
		out[k] = v.AsString()
		return true
	})
	return out
}

func maxNs(a, b int64) int64 {
	if b > a {
		return b
	}
	return a
}