newInterceptionProcessor returns an [http.HandlerFunc] which is capable of creating a new interceptor and processing a given request using [Provider] p, recording all usage events using [Recorder] rec. If cbs is non-nil, circuit breaker protection is applied per endpoint/model tuple.
(p provider.Provider, cbs *circuitbreaker.ProviderCircuitBreakers, rec recorder.Recorder, mcpProxy mcp.ServerProxier, logger slog.Logger, m *metrics.Metrics, tracer trace.Tracer)
| 218 | // using [Provider] p, recording all usage events using [Recorder] rec. |
| 219 | // If cbs is non-nil, circuit breaker protection is applied per endpoint/model tuple. |
| 220 | func newInterceptionProcessor(p provider.Provider, cbs *circuitbreaker.ProviderCircuitBreakers, rec recorder.Recorder, mcpProxy mcp.ServerProxier, logger slog.Logger, m *metrics.Metrics, tracer trace.Tracer) http.HandlerFunc { |
| 221 | return func(w http.ResponseWriter, r *http.Request) { |
| 222 | ctx, span := tracer.Start(r.Context(), "Intercept") |
| 223 | defer span.End() |
| 224 | |
| 225 | // We execute this before CreateInterceptor since the interceptors |
| 226 | // read the request body and don't reset them. |
| 227 | client := GuessClient(r) |
| 228 | sessionID := GuessSessionID(client, r) |
| 229 | |
| 230 | interceptor, err := p.CreateInterceptor(w, r.WithContext(ctx), tracer) |
| 231 | if err != nil { |
| 232 | span.SetStatus(codes.Error, fmt.Sprintf("failed to create interceptor: %v", err)) |
| 233 | if _, ok := errors.AsType[*http.MaxBytesError](err); ok { |
| 234 | writeRequestBodyTooLarge(w) |
| 235 | } else { |
| 236 | logger.Warn(ctx, "failed to create interceptor", slog.Error(err), slog.F("path", r.URL.Path)) |
| 237 | http.Error(w, fmt.Sprintf("failed to create %q interceptor", r.URL.Path), http.StatusInternalServerError) |
| 238 | } |
| 239 | return |
| 240 | } |
| 241 | |
| 242 | if m != nil { |
| 243 | start := time.Now() |
| 244 | defer func() { |
| 245 | m.InterceptionDuration.WithLabelValues(p.Name(), interceptor.Model()).Observe(time.Since(start).Seconds()) |
| 246 | }() |
| 247 | } |
| 248 | |
| 249 | actor := aibcontext.ActorFromContext(ctx) |
| 250 | if actor == nil { |
| 251 | logger.Warn(ctx, "no actor found in context") |
| 252 | http.Error(w, "no actor found", http.StatusBadRequest) |
| 253 | return |
| 254 | } |
| 255 | |
| 256 | traceAttrs := interceptor.TraceAttributes(r) |
| 257 | span.SetAttributes(traceAttrs...) |
| 258 | ctx = tracing.WithInterceptionAttributesInContext(ctx, traceAttrs) |
| 259 | r = r.WithContext(ctx) |
| 260 | |
| 261 | // Record usage in the background to not block request flow. |
| 262 | asyncRecorder := recorder.NewAsyncRecorder(logger, rec, recordingTimeout) |
| 263 | asyncRecorder.WithMetrics(m) |
| 264 | asyncRecorder.WithProvider(p.Name()) |
| 265 | asyncRecorder.WithModel(interceptor.Model()) |
| 266 | asyncRecorder.WithInitiatorID(actor.ID) |
| 267 | asyncRecorder.WithClient(string(client)) |
| 268 | interceptor.Setup(logger, asyncRecorder, mcpProxy) |
| 269 | |
| 270 | cred := interceptor.Credential() |
| 271 | if err := rec.RecordInterception(ctx, &recorder.InterceptionRecord{ |
| 272 | ID: interceptor.ID().String(), |
| 273 | InitiatorID: actor.ID, |
| 274 | Metadata: actor.Metadata, |
| 275 | Model: interceptor.Model(), |
| 276 | Provider: p.Type(), |
| 277 | ProviderName: p.Name(), |
no test coverage detected