Method MetricsSubscribeHandler

engine/server/telemetry.go:273–307

nolint:dupl

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

//nolint:dupl
func (ps *PubSub) MetricsSubscribeHandler(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)
			}
		}
		metrics, err := db.SelectMetricsSince(ctx, clientdb.SelectMetricsSinceParams{
			ID:    since,
			Limit: otlpBatchSize,
		})
		if err != nil {
			return nil, false, fmt.Errorf("select metrics: %w", err)
		}

		if len(metrics) == 0 {
			return nil, false, nil
		}
		since = metrics[len(metrics)-1].ID
		// Marshal the metrics to OTLP.
		payload, err := protojson.Marshal(&colmetricspb.ExportMetricsServiceRequest{
			ResourceMetrics: clientdb.MetricsToPB(metrics),
		})
		if err != nil {
			return nil, false, fmt.Errorf("marshal metrics: %w", err)
		}
		return &sse.Event{
			Name: "metrics",
			ID:   fmt.Sprintf("%d", since),
			Data: payload,
		}, true, nil
	})
}

type clientSpans struct {
	*PubSub