Skip to content

feat(pivot): port stdoutexporter off internal pipeline+consumer #202

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
trilamsr merged 1 commit into from
May 31, 2026
Merged

feat(pivot): port stdoutexporter off internal pipeline+consumer #202

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
3 changes: 2 additions & 1 deletion components/exporters/stdoutexporter/config.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -21,5 +21,6 @@ type Config struct {
}

// Validate is a no-op — there are no operator-configurable fields to
// reject. Defined so Config satisfies pipeline.Config.
// reject. Defined so Config satisfies component.Config (the upstream
// interface requires Validate()).
func (*Config) Validate() error { return nil }
79 changes: 48 additions & 31 deletions components/exporters/stdoutexporter/factory.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -7,47 +7,64 @@ import (
"fmt"
"os"

"github.com/tracecoreai/tracecore/internal/pipeline"
"go.opentelemetry.io/collector/component"
"go.opentelemetry.io/collector/exporter"
"go.uber.org/zap"
)

func componentType() pipeline.Type { return pipeline.MustNewType("stdoutexporter") }
// componentType is the kind name registered in components.yaml.
// Wrapped in a function so the MustNewType call is not a top-level side
// effect (mirrors the nccl_fr / clockreceiver pattern).
func componentType() component.Type { return component.MustNewType("stdoutexporter") }

// Factory is the package-scoped ExporterFactory for stdoutexporter.
// Mirrors clockreceiver.Factory in shape — M8+ exporter authors copy
// this pattern.
//
// Only CreateMetrics returns a real Exporter; CreateTraces and
// CreateLogs return pipeline.ErrSignalNotSupported (M1.5 keeps scope
// tight; M8+ can extend by implementing the missing signal paths in
// stdoutexporter.go).
var Factory pipeline.ExporterFactory = &factory{}

// NewFactory returns the package-var Factory. Required by
// tools/components-gen, which generates calls like
// `stdoutexporter.NewFactory()`. Exporter authors should reference
// `stdoutexporter.Factory` directly in code.
func NewFactory() pipeline.ExporterFactory { return Factory }

type factory struct{}
// stability is the OCB-surfaced stability level for stdoutexporter's
// metrics signal. Beta tracks "metrics + label shape pinned; behavior
// may evolve" — same level the exporter has carried since the v0.1.x
// internal/pipeline factory; PR-B (exporter port) preserves it across
// the upstream swap.
const stability = component.StabilityLevelBeta

func (*factory) Type() pipeline.Type { return componentType() }
// NewFactory returns the upstream exporter.Factory for stdoutexporter.
// Mirrors the upstream-contrib pattern (debugexporter, fileexporter) —
// callers construct via `stdoutexporter.NewFactory()` rather than a
// package var, so each OCB-stitched pipeline gets a freshly-built
// factory and the package surface stays a single exported symbol.
//
// Only the metrics signal returns a real Exporter; logs and traces
// surface upstream's "signal not supported" via exporter.NewFactory's
// default unimplemented behavior.
func NewFactory() exporter.Factory {
return exporter.NewFactory(
componentType(),
createDefaultConfig,
exporter.WithMetrics(createMetrics, stability),
)
}

func (*factory) CreateDefaultConfig() pipeline.Config {
// createDefaultConfig matches upstream component.CreateDefaultConfigFunc.
func createDefaultConfig() component.Config {
return &Config{Out: os.Stdout}
}

func (*factory) CreateMetrics(ctx context.Context, set pipeline.CreateSettings, cfg pipeline.Config) (pipeline.Exporter, error) {
// createMetrics is the exporter.CreateMetricsFunc wired by WithMetrics.
func createMetrics(ctx context.Context, set exporter.Settings, cfg component.Config) (exporter.Metrics, error) {
c, ok := cfg.(*Config)
if !ok {
return nil, fmt.Errorf("stdoutexporter: unexpected config type %T", cfg)
}
return newExporter(ctx, set, c), nil
}

func (*factory) CreateTraces(_ context.Context, _ pipeline.CreateSettings, _ pipeline.Config) (pipeline.Exporter, error) {
return nil, pipeline.ErrSignalNotSupported
}

func (*factory) CreateLogs(_ context.Context, _ pipeline.CreateSettings, _ pipeline.Config) (pipeline.Exporter, error) {
return nil, pipeline.ErrSignalNotSupported
e := newExporter(c)
if set.MeterProvider != nil {
if se, err := newSelfExporter(set.ID, set.MeterProvider); err == nil {
e.telemetry = se
} else {
recordInitError(ctx, set.MeterProvider,
"exporter", set.ID.String(), reasonInstrumentRegister)
if set.Logger != nil {
set.Logger.Warn("stdoutexporter self-telemetry init failed; using noop", zap.Error(err))
}
}
} else if set.Logger != nil {
set.Logger.Warn("stdoutexporter: no MeterProvider; self-telemetry using noop")
}
return e, nil
}
7 changes: 3 additions & 4 deletions components/exporters/stdoutexporter/selftel.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -9,7 +9,7 @@
// `module/` in PR-I, the scope name moves with it, matching OTel
// convention.
//
// Mirrors `components/receivers/nccl_fr/selftel.go` (PR-B1).
// Mirrors `components/receivers/nccl_fr/selftel.go` (PR-B1 / PR-B2).

package stdoutexporter

Expand All @@ -18,10 +18,9 @@ import (
"errors"
"fmt"

"go.opentelemetry.io/collector/component"
"go.opentelemetry.io/otel/attribute"
"go.opentelemetry.io/otel/metric"

"github.com/tracecoreai/tracecore/internal/pipeline"
)

// kind is a low-cardinality error-class identifier for exporter failures.
Expand Down Expand Up @@ -99,7 +98,7 @@ var _ selfExporter = noopSelfExporter{}
// on every emission. Metric name + label shape preserved from the
// v0.1.x internal selftelemetry package so dashboards / alerts don't
// regress.
func newSelfExporter(id pipeline.ID, mp metric.MeterProvider) (selfExporter, error) {
func newSelfExporter(id component.ID, mp metric.MeterProvider) (selfExporter, error) {
if mp == nil {
return nil, errNilMeterProvider
}
Expand Down
53 changes: 29 additions & 24 deletions components/exporters/stdoutexporter/selftel_test.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -10,13 +10,14 @@ import (
"strings"
"testing"

"go.opentelemetry.io/collector/component"
"go.opentelemetry.io/collector/exporter"
"go.opentelemetry.io/collector/exporter/exportertest"
"go.opentelemetry.io/otel/attribute"
"go.opentelemetry.io/otel/metric"
"go.opentelemetry.io/otel/metric/embedded"
sdkmetric "go.opentelemetry.io/otel/sdk/metric"
"go.opentelemetry.io/otel/sdk/metric/metricdata"

"github.com/tracecoreai/tracecore/internal/pipeline"
)

// newTestMeterProvider builds an SDK MeterProvider backed by a ManualReader
Expand Down Expand Up @@ -245,12 +246,11 @@ func TestFactory_FallsBackToNoopWhenMeterFails(t *testing.T) {
mp, rdr := newTestMeterProvider(t)
failing := &failingExporterMP{real: mp}

set := pipeline.CreateSettings{
ID: pipeline.MustNewID(pipeline.MustNewType("stdoutexporter"), "test"),
}
set.Telemetry.MeterProvider = failing
set := exportertest.NewNopSettings(componentType())
set.ID = component.NewIDWithName(componentType(), "test")
set.MeterProvider = failing
cfg := &Config{Out: &bytes.Buffer{}}
e, err := Factory.CreateMetrics(context.Background(), set, cfg)
e, err := NewFactory().CreateMetrics(context.Background(), set, cfg)
if err != nil {
t.Fatalf("CreateMetrics: %v", err)
}
Expand Down Expand Up @@ -286,24 +286,23 @@ func TestFactory_FallsBackToNoopWhenMeterFails(t *testing.T) {

// TestFactory_FallsBackToNoopWhenMeterProviderIsNil pins the
// nil-MeterProvider symmetry of the register-failure fallback: when
// `set.Telemetry.MeterProvider` is nil at construction (no telemetry
// wired), the factory MUST (1) return without error, (2) leave the
// exporter with a working noop telemetry field (no nil, no panic on
// hot-path calls), and (3) emit no datapoints anywhere — there's no
// MeterProvider to scrape. The skip-tick semantic for the nil path is
// intentional: `recordInitError` is only meaningful when telemetry is
// wired but instrument registration failed; a nil provider means the
// operator opted out of telemetry entirely, so a phantom counter would
// be noise. Mirrors `TestFactory_FallsBackToNoopWhenMeterFails` minus
// the failingExporterMP wrapper.
// `set.MeterProvider` is nil at construction (no telemetry wired), the
// factory MUST (1) return without error, (2) leave the exporter with a
// working noop telemetry field (no nil, no panic on hot-path calls),
// and (3) emit no datapoints anywhere — there's no MeterProvider to
// scrape. The skip-tick semantic for the nil path is intentional:
// `recordInitError` is only meaningful when telemetry is wired but
// instrument registration failed; a nil provider means the operator
// opted out of telemetry entirely, so a phantom counter would be
// noise. Mirrors `TestFactory_FallsBackToNoopWhenMeterFails` minus the
// failingExporterMP wrapper.
func TestFactory_FallsBackToNoopWhenMeterProviderIsNil(t *testing.T) {
set := pipeline.CreateSettings{
ID: pipeline.MustNewID(pipeline.MustNewType("stdoutexporter"), "test"),
}
set.Telemetry.MeterProvider = nil
set := exportertest.NewNopSettings(componentType())
set.ID = component.NewIDWithName(componentType(), "test")
set.MeterProvider = nil
cfg := &Config{Out: &bytes.Buffer{}}

e, err := Factory.CreateMetrics(context.Background(), set, cfg)
e, err := NewFactory().CreateMetrics(context.Background(), set, cfg)
if err != nil {
t.Fatalf("CreateMetrics: %v", err)
}
Expand Down Expand Up @@ -352,8 +351,8 @@ func TestSelfExporter_SiblingTypesArePackageLocal(t *testing.T) {
}
}

func testID() pipeline.ID {
return pipeline.MustNewID(pipeline.MustNewType("stdoutexporter"), "test")
func testID() component.ID {
return component.NewIDWithName(componentType(), "test")
}

func dumpNames(rm metricdata.ResourceMetrics) string {
Expand Down Expand Up @@ -397,3 +396,9 @@ func (m *failingExporterMeter) Int64Counter(name string, opts ...metric.Int64Cou
}
return c, nil
}

// Compile-time assertion: NewFactory returns an exporter.Factory.
// Pins the upstream-type contract that PR-B established — if a future
// refactor regresses to internal/pipeline.ExporterFactory, this fails
// to compile.
var _ exporter.Factory = NewFactory()
70 changes: 36 additions & 34 deletions components/exporters/stdoutexporter/stdoutexporter.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -6,31 +6,25 @@ import (
"context"
"fmt"
"io"
"log/slog"
"sync"

"go.opentelemetry.io/collector/component"
"go.opentelemetry.io/collector/consumer"
"go.opentelemetry.io/collector/exporter"
"go.opentelemetry.io/collector/pdata/pmetric"

"github.com/tracecoreai/tracecore/internal/consumer"
"github.com/tracecoreai/tracecore/internal/pipeline"
)

// stdoutExporter writes one OTLP/JSON-encoded line per
// ConsumeMetrics call. Embeds ComponentState for lifecycle bookkeeping;
// no goroutines, no I/O beyond synchronous writes (the operator
// running this in production is responsible for stdout going somewhere
// useful).
// ConsumeMetrics call. No goroutines, no I/O beyond synchronous writes
// (the operator running this in production is responsible for stdout
// going somewhere useful).
//
// Self-telemetry types live in `selftel.go` (sibling-scoped), not in
// `internal/selftelemetry` — see RFC-0013 §migration PR-B1; this
// exporter is the M8+ copy-worthy template, so the sibling pattern
// must be visible here.
type stdoutExporter struct {
pipeline.ComponentState

id pipeline.ID
logger *slog.Logger
out io.Writer
out io.Writer

// telemetry records exporter self-telemetry (calls_total +
// per-call success/failure). Always non-nil — factory substitutes
Expand All @@ -46,35 +40,33 @@ type stdoutExporter struct {
marshaler pmetric.JSONMarshaler
}

func newExporter(ctx context.Context, set pipeline.CreateSettings, cfg *Config) *stdoutExporter {
se := newNoopSelfExporter()
if set.Telemetry.MeterProvider == nil {
if set.Telemetry.Logger != nil {
set.Telemetry.Logger.Warn("stdoutexporter: no MeterProvider; self-telemetry using noop")
}
} else if x, err := newSelfExporter(set.ID, set.Telemetry.MeterProvider); err == nil {
se = x
} else {
recordInitError(ctx, set.Telemetry.MeterProvider,
"exporter", set.ID.String(), reasonInstrumentRegister)
if set.Telemetry.Logger != nil {
set.Telemetry.Logger.Warn("stdoutexporter self-telemetry init failed; using noop", "err", err)
}
}
// Compile-time assertion: stdoutExporter satisfies the upstream
// exporter.Metrics interface (component.Component + consumer.Metrics).
// A breaking shape change in upstream surfaces here at build time
// rather than at runtime when OCB stitches the pipeline.
var _ exporter.Metrics = (*stdoutExporter)(nil)

// newExporter constructs a stdoutExporter wired to a noop telemetry
// fallback. The factory replaces telemetry post-construction iff
// set.MeterProvider is non-nil AND instrument registration succeeds
// — see createMetrics in factory.go. Keeping the noop default here
// (rather than requiring the factory to set it) means the hot path
// never nil-checks e.telemetry.
func newExporter(cfg *Config) *stdoutExporter {
return &stdoutExporter{
id: set.ID,
logger: set.Telemetry.Logger,
out: cfg.Out,
telemetry: se,
telemetry: newNoopSelfExporter(),
}
}

// NOTE on ExporterCarrier removal:
//
// v0.1.x stdoutexporter exposed `SelfExporter() selftelemetry.Exporter`
// so `cmd/tracecore/collect.collectFailureRateReaders` could feed
// `tracecore.exporter.failure_rate`. The PR-B1 sibling port drops the
// `selftelemetry.ExporterCarrier` implementation:
// `tracecore.exporter.failure_rate`. The PR-B1 sibling port dropped
// the `selftelemetry.ExporterCarrier` implementation; this port (off
// `internal/pipeline` + `internal/consumer` onto upstream) preserves
// that drop. The rationale that justified PR-B1's drop still holds:
//
// - The runtime's reader-collection path silently skips components
// that don't implement the carrier (documented behavior).
Expand All @@ -87,7 +79,17 @@ func newExporter(ctx context.Context, set pipeline.CreateSettings, cfg *Config)
// evaporates regardless.
//
// `tracecore.exporter.calls_total` continues to surface because the
// sibling impl emits it on `set.Telemetry.MeterProvider` directly.
// sibling impl emits it on `set.MeterProvider` directly.

// Start is a no-op — stdoutexporter has no goroutines, no
// connections, and no resources to acquire at pipeline-start time.
// Defined to satisfy component.Component.
func (*stdoutExporter) Start(context.Context, component.Host) error { return nil }

// Shutdown is a no-op — there's nothing to release. The configured
// io.Writer is owned by the operator (typically os.Stdout) and must
// not be closed by the exporter. Defined to satisfy component.Component.
func (*stdoutExporter) Shutdown(context.Context) error { return nil }

// Capabilities reports MutatesData=false — stdoutexporter only reads
// the incoming pmetric.Metrics. Fan-out can share a read-only payload
Expand Down
Loading
Loading