diff --git a/pkg/apigateway/handler/backendproxy.go b/pkg/apigateway/handler/backendproxy.go index 0f200aa02c..48047390e6 100644 --- a/pkg/apigateway/handler/backendproxy.go +++ b/pkg/apigateway/handler/backendproxy.go @@ -65,6 +65,10 @@ func (h *SBackendServiceProxyHandler) requestManipulator(ctx context.Context, r if slashPos <= 0 { return r, httperrors.NewBadRequestError("invalid request URL %s", r.URL.Path) } + serviceName := path[:slashPos] + // Tell upstream (e.g. mcp-server SSE) the external path prefix so endpoint + // events point clients back through the gateway: /api/s//message + r.Header.Set("X-Forwarded-Prefix", "/api/s/"+serviceName) path = path[slashPos:] if strings.HasPrefix(path, "/r/") { path = path[len("/r/"):] diff --git a/pkg/appsrv/response.go b/pkg/appsrv/response.go index fa44d0a945..7f4f5202ab 100644 --- a/pkg/appsrv/response.go +++ b/pkg/appsrv/response.go @@ -18,8 +18,10 @@ import ( "bufio" "context" "fmt" + "mime" "net" "net/http" + "strings" "yunion.io/x/onecloud/pkg/httperrors" ) @@ -36,6 +38,8 @@ type responseWriterChannel struct { bodyResp chan responseWriterResponse statusChan chan int statusResp chan bool + flushChan chan struct{} + flushResp chan struct{} isClosed bool } @@ -47,6 +51,8 @@ func newResponseWriterChannel(backend http.ResponseWriter) responseWriterChannel bodyResp: make(chan responseWriterResponse), statusChan: make(chan int), statusResp: make(chan bool), + flushChan: make(chan struct{}), + flushResp: make(chan struct{}), isClosed: false, } } @@ -76,16 +82,34 @@ func (w *responseWriterChannel) WriteHeader(status int) { <-w.statusResp } -// implent http.Flusher +// Flush implements http.Flusher. Flush must run on the same goroutine as Write +// (wait loop); otherwise SSE via ReverseProxy never reaches the client. func (w *responseWriterChannel) Flush() { if w.isClosed { return } + w.flushChan <- struct{}{} + <-w.flushResp +} + +func (w *responseWriterChannel) flushBackend() { if f, ok := w.backend.(http.Flusher); ok { f.Flush() } } +func (w *responseWriterChannel) isEventStream() bool { + ct := w.backend.Header().Get("Content-Type") + if ct == "" { + return false + } + baseCT, _, err := mime.ParseMediaType(ct) + if err != nil { + return strings.HasPrefix(ct, "text/event-stream") + } + return baseCT == "text/event-stream" +} + // Hijack implements the Hijacker.Hijack method. Our response is both a ResponseWriter // and a Hijacker. func (w *responseWriterChannel) Hijack() (rwc net.Conn, buf *bufio.ReadWriter, err error) { @@ -121,6 +145,11 @@ func (w *responseWriterChannel) wait(ctx context.Context, workerChan chan *SWork // log.Infof("Recive body: %s, more: %v", len(bytes), more) if more { c, e := w.backend.Write(bytes) + // SSE / streaming: flush immediately after each chunk so clients + // see events without waiting for handler completion. + if e == nil && w.isEventStream() { + w.flushBackend() + } w.bodyResp <- responseWriterResponse{count: c, err: e} } else { stop = true @@ -133,6 +162,13 @@ func (w *responseWriterChannel) wait(ctx context.Context, workerChan chan *SWork } else { stop = true } + case _, more := <-w.flushChan: + if more { + w.flushBackend() + w.flushResp <- struct{}{} + } else { + stop = true + } } } return err @@ -148,4 +184,6 @@ func (w *responseWriterChannel) closeChannels() { close(w.bodyResp) close(w.statusChan) close(w.statusResp) + close(w.flushChan) + close(w.flushResp) } diff --git a/pkg/mcp-server/server/access_log.go b/pkg/mcp-server/server/access_log.go new file mode 100644 index 0000000000..15411b667e --- /dev/null +++ b/pkg/mcp-server/server/access_log.go @@ -0,0 +1,109 @@ +// Copyright 2019 Yunion +// +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. + +package server + +import ( + "bufio" + "crypto/sha256" + "encoding/base64" + "fmt" + "net" + "net/http" + "os" + "time" + + "yunion.io/x/log" + "yunion.io/x/pkg/utils" +) + +var accessLogHostId = genAccessLogHostId() + +func genAccessLogHostId() string { + hostname, _ := os.Hostname() + h := sha256.New() + fmt.Fprintf(h, "mcp-server:%s", hostname) + return base64.URLEncoding.EncodeToString(h.Sum(nil)) +} + +type accessLogResponseWriter struct { + http.ResponseWriter + status int +} + +func (w *accessLogResponseWriter) WriteHeader(code int) { + if w.status == 0 { + w.status = code + } + w.ResponseWriter.WriteHeader(code) +} + +func (w *accessLogResponseWriter) Write(b []byte) (int, error) { + if w.status == 0 { + w.status = http.StatusOK + } + return w.ResponseWriter.Write(b) +} + +func (w *accessLogResponseWriter) Flush() { + if f, ok := w.ResponseWriter.(http.Flusher); ok { + f.Flush() + } +} + +func (w *accessLogResponseWriter) Unwrap() http.ResponseWriter { + return w.ResponseWriter +} + +func (w *accessLogResponseWriter) Hijack() (net.Conn, *bufio.ReadWriter, error) { + h, ok := w.ResponseWriter.(http.Hijacker) + if !ok { + return nil, nil, fmt.Errorf("not a hijacker") + } + return h.Hijack() +} + +func genAccessRequestId(w http.ResponseWriter, r *http.Request) string { + rid := r.Header.Get("X-Request-Id") + if len(rid) == 0 { + rid = utils.GenRequestId(3) + } else { + rid = fmt.Sprintf("%s-%s", rid, utils.GenRequestId(3)) + } + w.Header().Set("X-Request-Id", rid) + w.Header().Set("X-Request-Host-Id", accessLogHostId) + return rid +} + +// withAccessLog wraps handler with appsrv-style access logs: +// +// hostId status requestId METHOD /path (remote) durationMs +func withAccessLog(next http.Handler) http.Handler { + return http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + lrw := &accessLogResponseWriter{ResponseWriter: w, status: 0} + rid := genAccessRequestId(lrw, r) + start := time.Now() + next.ServeHTTP(lrw, r) + if lrw.status == 0 { + lrw.status = http.StatusOK + } + durationMs := float64(time.Since(start).Nanoseconds()) / 1e6 + remote := r.RemoteAddr + if peer := r.Header.Get("X-Yunion-Peer-Service-Name"); peer != "" { + remote = fmt.Sprintf("%s:%s", r.RemoteAddr, peer) + } + log.Infof("%s %d %s %s %s (%s) %.2fms", + accessLogHostId, lrw.status, rid, r.Method, r.URL, remote, durationMs) + }) +} diff --git a/pkg/mcp-server/server/server.go b/pkg/mcp-server/server/server.go index 62c14f9afa..6f1dad54b4 100644 --- a/pkg/mcp-server/server/server.go +++ b/pkg/mcp-server/server/server.go @@ -193,7 +193,13 @@ func (s *CloudpodsMCPServer) Start() error { sseServer := server.NewSSEServer( s.mcpServer, server.WithSSEContextFunc(contextFunc), - server.WithHTTPServer(&http.Server{Handler: mux}), + // Gateway proxies /api/s/mcp-server/sse → /sse and sets X-Forwarded-Prefix. + // Clients must POST messages to /api/s/mcp-server/message, not /message. + server.WithDynamicBasePath(func(r *http.Request, _ string) string { + return strings.TrimSuffix(r.Header.Get("X-Forwarded-Prefix"), "/") + }), + server.WithUseFullURLForMessageEndpoint(false), + server.WithHTTPServer(&http.Server{Handler: withAccessLog(mux)}), ) mux.HandleFunc("/version", func(w http.ResponseWriter, r *http.Request) { appsrv.VersionHandler(context.Background(), w, r) diff --git a/pkg/proxy/http.go b/pkg/proxy/http.go index 69072201e0..615b0ac11b 100644 --- a/pkg/proxy/http.go +++ b/pkg/proxy/http.go @@ -20,6 +20,7 @@ import ( "net/http" "net/http/httputil" "net/url" + "time" "yunion.io/x/log" "yunion.io/x/pkg/appctx" @@ -68,15 +69,24 @@ func (p *SReverseProxy) ServeHTTP(ctx context.Context, w http.ResponseWriter, r } log.Debugf("Forwarding to servie: %q, url: %q", p.serviceName, remoteUrl.String()) proxy := httputil.NewSingleHostReverseProxy(remoteUrl) + // SSE / long-polling: flush immediately; never ask upstream for gzip (buffers whole stream). + proxy.FlushInterval = -1 proxy.Transport = &http.Transport{ - Proxy: http.ProxyFromEnvironment, - TLSClientConfig: &tls.Config{InsecureSkipVerify: true}, - DisableKeepAlives: true, + Proxy: http.ProxyFromEnvironment, + TLSClientConfig: &tls.Config{InsecureSkipVerify: true}, + DisableKeepAlives: true, + DisableCompression: true, + IdleConnTimeout: 90 * time.Second, + } + proxy.ModifyResponse = func(resp *http.Response) error { + resp.Header.Set("X-Accel-Buffering", "no") + return nil } r, err = p.manipulator(ctx, r) if err != nil { httperrors.InternalServerError(ctx, w, "%v", err) return } + r.Header.Del("Accept-Encoding") proxy.ServeHTTP(w, r) }