Spaces:

Leon4gr45
/

openmeter

Paused

openmeter / pkg /framework /transport /httptransport /handler.go

Upload folder using huggingface_hub (part 9)

fea99b3 verified 4 months ago

5.62 kB

package httptransport

import (
	"context"
	"errors"
	"net/http"
	"sync/atomic"

	"go.opentelemetry.io/otel"
	semconv "go.opentelemetry.io/otel/semconv/v1.27.0"
	"go.opentelemetry.io/otel/trace"

	"github.com/openmeterio/openmeter/pkg/contextx"
	"github.com/openmeterio/openmeter/pkg/framework/commonhttp"
	"github.com/openmeterio/openmeter/pkg/framework/operation"
	"github.com/openmeterio/openmeter/pkg/framework/transport/httptransport/encoder"
	"github.com/openmeterio/openmeter/pkg/models"
)

var defaultHandlerOptions = []HandlerOption{
	WithErrorEncoder(commonhttp.GenericErrorEncoder()),
}

// tracer reads the globally configured TracerProvider (set during telemetry init).
// Used to start an application-level span named after the handler operation, as a
// child of the otelhttp server span.
var tracer = otel.Tracer("github.com/openmeterio/openmeter/pkg/framework/transport/httptransport")

// operationSpansEnabled is a global toggle for the per-operation child span. It is off
// by default — the span is added to every operation request, so enabling it across the
// whole API surface meaningfully increases trace-span volume. Set once at startup via
// EnableOperationSpans.
var operationSpansEnabled atomic.Bool

// EnableOperationSpans globally enables or disables the application-level per-operation
// child span. Intended to be called once during startup from telemetry configuration.
func EnableOperationSpans(enabled bool) {
	operationSpansEnabled.Store(enabled)
}

type Handler[Request any, Response any] interface {
	ServeHTTP(w http.ResponseWriter, r *http.Request)
	Chain(outer operation.Middleware[Request, Response], others ...operation.Middleware[Request, Response]) Handler[Request, Response]
}

// NewHandler returns a new HTTP handler that wraps the given [operation.Operation].
func NewHandler[Request any, Response any](
	requestDecoder RequestDecoder[Request],
	op operation.Operation[Request, Response],
	responseEncoder encoder.ResponseEncoder[Response],
	options ...HandlerOption,
) Handler[Request, Response] {
	return newHandler(requestDecoder, op, responseEncoder, options...)
}

func newHandler[Request any, Response any](
	requestDecoder RequestDecoder[Request],
	op operation.Operation[Request, Response],
	responseEncoder encoder.ResponseEncoder[Response],
	options ...HandlerOption,
) handler[Request, Response] {
	h := handler[Request, Response]{
		operation: op,

		decodeRequest:  requestDecoder,
		encodeResponse: responseEncoder,
	}

	options = append(options, defaultHandlerOptions...)

	h.apply(options)

	return h
}

type handler[Request any, Response any] struct {
	operation         operation.Operation[Request, Response]
	operationNameFunc func(ctx context.Context) string

	decodeRequest  RequestDecoder[Request]
	encodeResponse encoder.ResponseEncoder[Response]
	errorEncoders  []encoder.ErrorEncoder

	errorHandler ErrorHandler
}

type RequestDecoder[Request any] func(ctx context.Context, r *http.Request) (Request, error)

// ErrorHandler receives a transport error to be processed for diagnostic purposes.
// Usually this means logging the error.
type ErrorHandler interface {
	HandleContext(ctx context.Context, err error)
}

func (h handler[Request, Response]) ServeHTTP(w http.ResponseWriter, r *http.Request) {
	ctx := r.Context()

	// TODO: rewrite this as a generic hook
	if h.operationNameFunc != nil {
		name := h.operationNameFunc(ctx)
		ctx = contextx.WithAttr(ctx, string(semconv.HTTPRouteKey), name)

		// When enabled globally (EnableOperationSpans), start an application-level span
		// named after the operation, as a child of the otelhttp server span. The server
		// span stays route-named; this one carries the operation identity (e.g.
		// "query-governance-access") and exposes the handler-vs-middleware timing split.
		// Off by default: it adds a span to every operation request API-wide, so it is
		// gated by a single startup toggle rather than enabled unconditionally.
		if operationSpansEnabled.Load() {
			var span trace.Span
			ctx, span = tracer.Start(ctx, name)
			defer span.End()
		}
	}

	request, err := h.decodeRequest(ctx, r)
	if err != nil {
		// Might be a client error (can be encoded, non-terminal)
		// Might be a server error (terminal)

		handled := h.encodeError(ctx, err, w, r)
		if !handled {
			h.errorHandler.HandleContext(ctx, err)
		}

		return
	}

	response, err := h.operation(ctx, request)
	if err != nil {
		// Might be a client error (can be encoded, non-terminal)
		// Might be a server error (terminal)

		handled := h.encodeError(ctx, err, w, r)
		if !handled {
			h.errorHandler.HandleContext(ctx, err)
		}

		return
	}

	if err := h.encodeResponse(ctx, w, r, response); err != nil {
		// Always a server error (terminal)?

		h.errorHandler.HandleContext(ctx, err)
		return
	}
}

func (h handler[Request, Response]) encodeError(ctx context.Context, err error, w http.ResponseWriter, r *http.Request) bool {
	for _, errorEncoder := range h.errorEncoders {
		if errorEncoder(ctx, err, w, r) {
			return true
		}
	}

	if encoder, ok := err.(SelfEncodingError); ok {
		if encoder.EncodeError(ctx, w) {
			return true
		}
	}

	models.NewStatusProblem(ctx, errors.New("internal server error"), http.StatusInternalServerError).Respond(w)

	return false
}

func (h handler[Request, Response]) Chain(outer operation.Middleware[Request, Response], others ...operation.Middleware[Request, Response]) Handler[Request, Response] {
	h.operation = operation.Chain(outer, others...)(h.operation)
	return h
}

type SelfEncodingError interface {
	EncodeError(ctx context.Context, w http.ResponseWriter) bool
}

	package httptransport

	import (
	"context"
	"errors"
	"net/http"
	"sync/atomic"

	"go.opentelemetry.io/otel"
	semconv "go.opentelemetry.io/otel/semconv/v1.27.0"
	"go.opentelemetry.io/otel/trace"

	"github.com/openmeterio/openmeter/pkg/contextx"
	"github.com/openmeterio/openmeter/pkg/framework/commonhttp"
	"github.com/openmeterio/openmeter/pkg/framework/operation"
	"github.com/openmeterio/openmeter/pkg/framework/transport/httptransport/encoder"
	"github.com/openmeterio/openmeter/pkg/models"
	)

	var defaultHandlerOptions = []HandlerOption{
	WithErrorEncoder(commonhttp.GenericErrorEncoder()),
	}

	// tracer reads the globally configured TracerProvider (set during telemetry init).
	// Used to start an application-level span named after the handler operation, as a
	// child of the otelhttp server span.
	var tracer = otel.Tracer("github.com/openmeterio/openmeter/pkg/framework/transport/httptransport")

	// operationSpansEnabled is a global toggle for the per-operation child span. It is off
	// by default — the span is added to every operation request, so enabling it across the
	// whole API surface meaningfully increases trace-span volume. Set once at startup via
	// EnableOperationSpans.
	var operationSpansEnabled atomic.Bool

	// EnableOperationSpans globally enables or disables the application-level per-operation
	// child span. Intended to be called once during startup from telemetry configuration.
	func EnableOperationSpans(enabled bool) {
	operationSpansEnabled.Store(enabled)
	}

	type Handler[Request any, Response any] interface {
	ServeHTTP(w http.ResponseWriter, r *http.Request)
	Chain(outer operation.Middleware[Request, Response], others ...operation.Middleware[Request, Response]) Handler[Request, Response]
	}

	// NewHandler returns a new HTTP handler that wraps the given [operation.Operation].
	func NewHandler[Request any, Response any](
	requestDecoder RequestDecoder[Request],
	op operation.Operation[Request, Response],
	responseEncoder encoder.ResponseEncoder[Response],

	options ...HandlerOption,
	) Handler[Request, Response] {
	return newHandler(requestDecoder, op, responseEncoder, options...)
	}

	func newHandler[Request any, Response any](
	requestDecoder RequestDecoder[Request],
	op operation.Operation[Request, Response],
	responseEncoder encoder.ResponseEncoder[Response],

	options ...HandlerOption,
	) handler[Request, Response] {
	h := handler[Request, Response]{
	operation: op,

	decodeRequest: requestDecoder,
	encodeResponse: responseEncoder,
	}

	options = append(options, defaultHandlerOptions...)

	h.apply(options)

	return h
	}

	type handler[Request any, Response any] struct {
	operation operation.Operation[Request, Response]
	operationNameFunc func(ctx context.Context) string

	decodeRequest RequestDecoder[Request]
	encodeResponse encoder.ResponseEncoder[Response]
	errorEncoders []encoder.ErrorEncoder

	errorHandler ErrorHandler
	}

	type RequestDecoder[Request any] func(ctx context.Context, r *http.Request) (Request, error)

	// ErrorHandler receives a transport error to be processed for diagnostic purposes.
	// Usually this means logging the error.
	type ErrorHandler interface {
	HandleContext(ctx context.Context, err error)
	}

	func (h handler[Request, Response]) ServeHTTP(w http.ResponseWriter, r *http.Request) {
	ctx := r.Context()

	// TODO: rewrite this as a generic hook
	if h.operationNameFunc != nil {
	name := h.operationNameFunc(ctx)
	ctx = contextx.WithAttr(ctx, string(semconv.HTTPRouteKey), name)

	// When enabled globally (EnableOperationSpans), start an application-level span
	// named after the operation, as a child of the otelhttp server span. The server
	// span stays route-named; this one carries the operation identity (e.g.
	// "query-governance-access") and exposes the handler-vs-middleware timing split.
	// Off by default: it adds a span to every operation request API-wide, so it is
	// gated by a single startup toggle rather than enabled unconditionally.
	if operationSpansEnabled.Load() {
	var span trace.Span
	ctx, span = tracer.Start(ctx, name)
	defer span.End()
	}
	}

	request, err := h.decodeRequest(ctx, r)
	if err != nil {
	// Might be a client error (can be encoded, non-terminal)
	// Might be a server error (terminal)

	handled := h.encodeError(ctx, err, w, r)
	if !handled {
	h.errorHandler.HandleContext(ctx, err)
	}

	return
	}

	response, err := h.operation(ctx, request)
	if err != nil {
	// Might be a client error (can be encoded, non-terminal)
	// Might be a server error (terminal)

	handled := h.encodeError(ctx, err, w, r)
	if !handled {
	h.errorHandler.HandleContext(ctx, err)
	}

	return
	}

	if err := h.encodeResponse(ctx, w, r, response); err != nil {
	// Always a server error (terminal)?

	h.errorHandler.HandleContext(ctx, err)
	return
	}
	}

	func (h handler[Request, Response]) encodeError(ctx context.Context, err error, w http.ResponseWriter, r *http.Request) bool {
	for _, errorEncoder := range h.errorEncoders {
	if errorEncoder(ctx, err, w, r) {
	return true
	}
	}

	if encoder, ok := err.(SelfEncodingError); ok {
	if encoder.EncodeError(ctx, w) {
	return true
	}
	}

	models.NewStatusProblem(ctx, errors.New("internal server error"), http.StatusInternalServerError).Respond(w)

	return false
	}

	func (h handler[Request, Response]) Chain(outer operation.Middleware[Request, Response], others ...operation.Middleware[Request, Response]) Handler[Request, Response] {
	h.operation = operation.Chain(outer, others...)(h.operation)
	return h
	}

	type SelfEncodingError interface {
	EncodeError(ctx context.Context, w http.ResponseWriter) bool
	}