Method TracesSubscribeHandler

engine/server/telemetry.go:197–234

(w http.ResponseWriter, r *http.Request, client *daggerClient)

const otlpBatchSize = 1000

func (ps *PubSub) TracesSubscribeHandler(w http.ResponseWriter, r *http.Request, client *daggerClient) error {
	return ps.sseHandler(w, r, client, func(ctx context.Context, db *clientdb.DB, lastID string) (*sse.Event, bool, error) {
		var since int64
		if lastID != "" {
			_, err := fmt.Sscanf(lastID, "%d", &since)
			if err != nil {
				return nil, false, fmt.Errorf("invalid last ID: %w", err)
			}
		}
		spans, err := db.SelectSpansSince(ctx, clientdb.SelectSpansSinceParams{
			ID:    since,
			Limit: otlpBatchSize,
		})
		if err != nil {
			return nil, false, fmt.Errorf("select spans: %w", err)
		}
		if len(spans) == 0 {
			return nil, false, nil
		}
		roSpans := make([]sdktrace.ReadOnlySpan, len(spans))
		for i, span := range spans {
			roSpans[i] = span.ReadOnly()
			since = span.ID
		}
		// Marshal the spans to OTLP.
		payload, err := protojson.Marshal(&coltracepb.ExportTraceServiceRequest{
			ResourceSpans: telemetry.SpansToPB(roSpans),
		})
		if err != nil {
			return nil, false, fmt.Errorf("marshal spans: %w", err)
		}
		return &sse.Event{
			Name: "spans",
			ID:   fmt.Sprintf("%d", since),
			Data: payload,
		}, true, nil
	})
}

//nolint:dupl
func (ps *PubSub) LogsSubscribeHandler(w http.ResponseWriter, r *http.Request, client *daggerClient) error {