Files
68ce98920a feat(zap): native ZAP-transport OTLP receiver + exporter (#4)
* feat(zap): native ZAP-transport OTLP receiver + exporter

Adds ZAP-native telemetry ingestion/forwarding over github.com/zap-proto/http
(the Hanzo ZAP wire — capnp frames, X-Wing PQ-KEM handshake), NOT OTLP-HTTP
(:4318) or OTLP-gRPC (:4317). The payload stays standard OTLP protobuf so full
OTel fidelity + pipeline interop are preserved; only the transport is ZAP.

- receiver/zapreceiver: runs a zaphttp.Server on :4319, terminates OTLP over
  the ZAP wire on /v1/{traces,logs,metrics}, decodes to pdata, pushes to the
  pipeline (-> ClickHouse exporters -> hanzoai/datastore). ObsReport transport
  label is "zap". One listener serves all three signals.
- exporter/zapexporter: pdata -> OTLP proto -> zaphttp POST to a peer collector's
  ZAP receiver (used by the filelog log-agent to ship node logs over ZAP).
- Registered in components.go alongside otlpreceiver (OTLP retained as an
  interop endpoint for third-party SDKs).

Tests: TestZapReceiverAllSignals (traces+logs+metrics e2e over the real ZAP
wire) and TestExportTracesOverZAP both green; full ./components build clean.

* build(zap): self-building multi-stage Dockerfile for in-cluster Kaniko/BuildKit

The default Dockerfile COPYs a CI-prebuilt binary; this compiles in-image so an
in-cluster BuildKit/Kaniko job (platform.hanzo.ai) can produce the ZAP-enabled
image without the (currently broken) GitHub reusable workflow.

---------

Co-authored-by: hanzo-dev <[email protected]>
2026-07-03 14:56:33 -07:00

75 lines
2.3 KiB
Go

package zapexporter
import (
"context"
"go.opentelemetry.io/collector/component"
"go.opentelemetry.io/collector/config/configoptional"
"go.opentelemetry.io/collector/config/configretry"
"go.opentelemetry.io/collector/exporter"
"go.opentelemetry.io/collector/exporter/exporterhelper"
)
var typeStr = component.MustNewType("zap")
const (
defaultEndpoint = "otel-collector.hanzo.svc:4319"
stability = component.StabilityLevelBeta
)
// NewFactory returns the ZAP-native OTLP exporter factory.
func NewFactory() exporter.Factory {
return exporter.NewFactory(
typeStr,
createDefaultConfig,
exporter.WithTraces(createTraces, stability),
exporter.WithLogs(createLogs, stability),
exporter.WithMetrics(createMetrics, stability),
)
}
func createDefaultConfig() component.Config {
return &Config{
Endpoint: defaultEndpoint,
TimeoutConfig: exporterhelper.NewDefaultTimeoutConfig(),
QueueBatchConfig: configoptional.Some(exporterhelper.NewDefaultQueueConfig()),
BackOffConfig: configretry.NewDefaultBackOffConfig(),
}
}
func createTraces(ctx context.Context, set exporter.Settings, cfg component.Config) (exporter.Traces, error) {
c := cfg.(*Config)
e := newExporter(set, c)
return exporterhelper.NewTraces(ctx, set, cfg, e.pushTraces,
exporterhelper.WithStart(e.Start),
exporterhelper.WithShutdown(e.Shutdown),
exporterhelper.WithTimeout(c.TimeoutConfig),
exporterhelper.WithQueue(c.QueueBatchConfig),
exporterhelper.WithRetry(c.BackOffConfig),
)
}
func createLogs(ctx context.Context, set exporter.Settings, cfg component.Config) (exporter.Logs, error) {
c := cfg.(*Config)
e := newExporter(set, c)
return exporterhelper.NewLogs(ctx, set, cfg, e.pushLogs,
exporterhelper.WithStart(e.Start),
exporterhelper.WithShutdown(e.Shutdown),
exporterhelper.WithTimeout(c.TimeoutConfig),
exporterhelper.WithQueue(c.QueueBatchConfig),
exporterhelper.WithRetry(c.BackOffConfig),
)
}
func createMetrics(ctx context.Context, set exporter.Settings, cfg component.Config) (exporter.Metrics, error) {
c := cfg.(*Config)
e := newExporter(set, c)
return exporterhelper.NewMetrics(ctx, set, cfg, e.pushMetrics,
exporterhelper.WithStart(e.Start),
exporterhelper.WithShutdown(e.Shutdown),
exporterhelper.WithTimeout(c.TimeoutConfig),
exporterhelper.WithQueue(c.QueueBatchConfig),
exporterhelper.WithRetry(c.BackOffConfig),
)
}