`开头的请求的最终是由`proxyHandler.ServeHTTP()`来处理，代码如下：
+
+// staging\src\k8s.io\kube-aggregator\pkg\apiserver\handler_proxy.go
+func (r *proxyHandler) ServeHTTP(w http.ResponseWriter, req *http.Request) {
+ value := r.handlingInfo.Load()
+ if value == nil {
+ r.localDelegate.ServeHTTP(w, req)
+ return
+ }
+ handlingInfo := value.(proxyHandlingInfo)
+ if handlingInfo.local {
+ if r.localDelegate == nil {
+ http.Error(w, "", http.StatusNotFound)
+ return
+ }
+ r.localDelegate.ServeHTTP(w, req)
+ return
+ }
+
+ if !handlingInfo.serviceAvailable {
+ proxyError(w, req, "service unavailable", http.StatusServiceUnavailable)
+ return
+ }
+
+ if handlingInfo.transportBuildingError != nil {
+ proxyError(w, req, handlingInfo.transportBuildingError.Error(), http.StatusInternalServerError)
+ return
+ }
+
+ user, ok := genericapirequest.UserFrom(req.Context())
+ if !ok {
+ proxyError(w, req, "missing user", http.StatusInternalServerError)
+ return
+ }
+
+ // write a new location based on the existing request pointed at the target service
+ location := &url.URL{}
+ location.Scheme = "https"
+ rloc, err := r.serviceResolver.ResolveEndpoint(handlingInfo.serviceNamespace, handlingInfo.serviceName, handlingInfo.servicePort)
+ if err != nil {
+ klog.Errorf("error resolving %s/%s: %v", handlingInfo.serviceNamespace, handlingInfo.serviceName, err)
+ proxyError(w, req, "service unavailable", http.StatusServiceUnavailable)
+ return
+ }
+ location.Host = rloc.Host
+ location.Path = req.URL.Path
+ location.RawQuery = req.URL.Query().Encode()
+
+ newReq, cancelFn := apiserverproxyutil.NewRequestForProxy(location, req)
+ defer cancelFn()
+
+ if handlingInfo.proxyRoundTripper == nil {
+ proxyError(w, req, "", http.StatusNotFound)
+ return
+ }
+
+ proxyRoundTripper := handlingInfo.proxyRoundTripper
+ upgrade := httpstream.IsUpgradeRequest(req)
+
+ var userUID string
+ if utilfeature.DefaultFeatureGate.Enabled(genericfeatures.RemoteRequestHeaderUID) {
+ userUID = user.GetUID()
+ }
+
+ proxyRoundTripper = transport.NewAuthProxyRoundTripper(user.GetName(), userUID, user.GetGroups(), user.GetExtra(), proxyRoundTripper)
+
+ if utilfeature.DefaultFeatureGate.Enabled(genericfeatures.APIServerTracing) && !upgrade {
+ tracingWrapper := tracing.WrapperFor(r.tracerProvider)
+ proxyRoundTripper = tracingWrapper(proxyRoundTripper)
+ }
+
+ // If we are upgrading, then the upgrade path tries to use this request with the TLS config we provide, but it does
+ // NOT use the proxyRoundTripper. It's a direct dial that bypasses the proxyRoundTripper. This means that we have to
+ // attach the "correct" user headers to the request ahead of time.
+ if upgrade {
+ transport.SetAuthProxyHeaders(newReq, user.GetName(), userUID, user.GetGroups(), user.GetExtra())
+ }
+
+ handler := proxy.NewUpgradeAwareHandler(location, proxyRoundTripper, true, upgrade, &responder{w: w})
+ if r.rejectForwardingRedirects {
+ handler.RejectForwardingRedirects = true
+ }
+ utilflowcontrol.RequestDelegated(req.Context())
+ handler.ServeHTTP(w, newReq)
+}
+
+
+### PrepareRun
+
+```go
+
+// PrepareRun prepares the aggregator to run, by setting up the OpenAPI spec &
+// aggregated discovery document and calling the generic PrepareRun.
+func (s *APIAggregator) PrepareRun() (preparedAPIAggregator, error) {
+ // 在post start hook中去启动OpenAPI Controller
+ // add post start hook before generic PrepareRun in order to be before /healthz installation
+ if s.openAPIConfig != nil {
+ s.GenericAPIServer.AddPostStartHookOrDie("apiservice-openapi-controller", func(context genericapiserver.PostStartHookContext) error {
+ go s.openAPIAggregationController.Run(context.Done())
+ return nil
+ })
+ }
+
+ if s.openAPIV3Config != nil {
+ s.GenericAPIServer.AddPostStartHookOrDie("apiservice-openapiv3-controller", func(context genericapiserver.PostStartHookContext) error {
+ go s.openAPIV3AggregationController.Run(context.Done())
+ return nil
+ })
+ }
+
+ // 调用s.GenericAPIServer.PrepareRun()方法
+ prepared := s.GenericAPIServer.PrepareRun()
+
+ // // 制作一个OpenAPI Controller,观测各个Aggregated APIServer所支持的API的变化,然后更新AggregatorServer自己的OpenAPI Spec
+ // delay OpenAPI setup until the delegate had a chance to setup their OpenAPI handlers
+ if s.openAPIConfig != nil {
+ specDownloader := openapiaggregator.NewDownloader()
+ openAPIAggregator, err := openapiaggregator.BuildAndRegisterAggregator(
+ &specDownloader,
+ s.GenericAPIServer.NextDelegate(),
+ s.GenericAPIServer.Handler.GoRestfulContainer.RegisteredWebServices(),
+ s.openAPIConfig,
+ s.GenericAPIServer.Handler.NonGoRestfulMux)
+ if err != nil {
+ return preparedAPIAggregator{}, err
+ }
+ s.openAPIAggregationController = openapicontroller.NewAggregationController(&specDownloader, openAPIAggregator)
+ }
+
+ if s.openAPIV3Config != nil {
+ specDownloaderV3 := openapiv3aggregator.NewDownloader()
+ openAPIV3Aggregator, err := openapiv3aggregator.BuildAndRegisterAggregator(
+ specDownloaderV3,
+ s.GenericAPIServer.NextDelegate(),
+ s.GenericAPIServer.Handler.GoRestfulContainer,
+ s.openAPIV3Config,
+ s.GenericAPIServer.Handler.NonGoRestfulMux)
+ if err != nil {
+ return preparedAPIAggregator{}, err
+ }
+ s.openAPIV3AggregationController = openapiv3controller.NewAggregationController(openAPIV3Aggregator)
+ }
+
+ return preparedAPIAggregator{APIAggregator: s, runnable: prepared}, nil
+}
```
.png)
.png)