diff --git a/pkg/api/api.go b/pkg/api/api.go index 81f7c964200..011797344e0 100644 --- a/pkg/api/api.go +++ b/pkg/api/api.go @@ -435,7 +435,7 @@ func (hs *HTTPServer) registerRoutes() { if hs.Cfg.FeatureToggles["live-pipeline"] { // POST Live data to be processed according to channel rules. - liveRoute.Post("/push/:streamId/:path", hs.LivePushGateway.HandlePath) + liveRoute.Post("/pipeline/push/*", hs.LivePushGateway.HandlePipelinePush) liveRoute.Post("/pipeline-convert-test", routing.Wrap(hs.Live.HandlePipelineConvertTestHTTP), reqOrgAdmin) liveRoute.Get("/pipeline-entities", routing.Wrap(hs.Live.HandlePipelineEntitiesListHTTP), reqOrgAdmin) liveRoute.Get("/channel-rules", routing.Wrap(hs.Live.HandleChannelRulesListHTTP), reqOrgAdmin) diff --git a/pkg/services/live/live.go b/pkg/services/live/live.go index 3a66bdebdc7..70d530104cf 100644 --- a/pkg/services/live/live.go +++ b/pkg/services/live/live.go @@ -321,6 +321,12 @@ func ProvideService(plugCtxProvider *plugincontext.Provider, cfg *setting.Cfg, r CheckOrigin: checkOrigin, }) + pushPipelineWSHandler := pushws.NewPipelinePushHandler(g.Pipeline, pushws.Config{ + ReadBufferSize: 1024, + WriteBufferSize: 1024, + CheckOrigin: checkOrigin, + }) + g.websocketHandler = func(ctx *models.ReqContext) { user := ctx.SignedInUser @@ -342,12 +348,21 @@ func ProvideService(plugCtxProvider *plugincontext.Provider, cfg *setting.Cfg, r pushWSHandler.ServeHTTP(ctx.Resp, r) } + g.pushPipelineWebsocketHandler = func(ctx *models.ReqContext) { + user := ctx.SignedInUser + newCtx := livecontext.SetContextSignedUser(ctx.Req.Context(), user) + newCtx = livecontext.SetContextChannelID(newCtx, web.Params(ctx.Req)["*"]) + r := ctx.Req.WithContext(newCtx) + pushPipelineWSHandler.ServeHTTP(ctx.Resp, r) + } + g.RouteRegister.Group("/api/live", func(group routing.RouteRegister) { group.Get("/ws", g.websocketHandler) }, middleware.ReqSignedIn) g.RouteRegister.Group("/api/live", func(group routing.RouteRegister) { group.Get("/push/:streamId", g.pushWebsocketHandler) + group.Get("/pipeline/push/*", g.pushPipelineWebsocketHandler) }, middleware.ReqOrgAdmin) g.registerUsageMetrics() @@ -375,8 +390,9 @@ type GrafanaLive struct { surveyCaller *survey.Caller // Websocket handlers - websocketHandler interface{} - pushWebsocketHandler interface{} + websocketHandler interface{} + pushWebsocketHandler interface{} + pushPipelineWebsocketHandler interface{} // Full channel handler channels map[string]models.ChannelHandler diff --git a/pkg/services/live/livecontext/context.go b/pkg/services/live/livecontext/context.go index 44503ff2b2c..36120245e34 100644 --- a/pkg/services/live/livecontext/context.go +++ b/pkg/services/live/livecontext/context.go @@ -37,3 +37,18 @@ func GetContextStreamID(ctx context.Context) (string, bool) { } return "", false } + +type channelIDContextKey struct{} + +func SetContextChannelID(ctx context.Context, channelID string) context.Context { + ctx = context.WithValue(ctx, channelIDContextKey{}, channelID) + return ctx +} + +func GetContextChannelID(ctx context.Context) (string, bool) { + if val := ctx.Value(channelIDContextKey{}); val != nil { + values, ok := val.(string) + return values, ok + } + return "", false +} diff --git a/pkg/services/live/pipeline/devdata.go b/pkg/services/live/pipeline/devdata.go index b5a9d3ce925..31e470c60a2 100644 --- a/pkg/services/live/pipeline/devdata.go +++ b/pkg/services/live/pipeline/devdata.go @@ -64,21 +64,21 @@ func postTestData() { jsonData, _ := json.Marshal(d) log.Println(string(jsonData)) - req, _ := http.NewRequest("POST", "http://localhost:3000/api/live/push/json/auto", bytes.NewReader(jsonData)) + req, _ := http.NewRequest("POST", "http://localhost:3000/api/live/pipeline/push/stream/json/auto", bytes.NewReader(jsonData)) req.Header.Set("Authorization", "Bearer "+os.Getenv("GF_TOKEN")) resp, err := http.DefaultClient.Do(req) if err != nil { log.Fatal(err) } _ = resp.Body.Close() - req, _ = http.NewRequest("POST", "http://localhost:3000/api/live/push/json/tip", bytes.NewReader(jsonData)) + req, _ = http.NewRequest("POST", "http://localhost:3000/api/live/push/pipeline/push/stream/json/tip", bytes.NewReader(jsonData)) req.Header.Set("Authorization", "Bearer "+os.Getenv("GF_TOKEN")) resp, err = http.DefaultClient.Do(req) if err != nil { log.Fatal(err) } _ = resp.Body.Close() - req, _ = http.NewRequest("POST", "http://localhost:3000/api/live/push/json/exact", bytes.NewReader(jsonData)) + req, _ = http.NewRequest("POST", "http://localhost:3000/api/live/pipeline/push/stream/json/exact", bytes.NewReader(jsonData)) req.Header.Set("Authorization", "Bearer "+os.Getenv("GF_TOKEN")) resp, err = http.DefaultClient.Do(req) if err != nil { diff --git a/pkg/services/live/pushhttp/push.go b/pkg/services/live/pushhttp/push.go index 777db550d77..a5ed4748d89 100644 --- a/pkg/services/live/pushhttp/push.go +++ b/pkg/services/live/pushhttp/push.go @@ -96,9 +96,8 @@ func (g *Gateway) Handle(ctx *models.ReqContext) { } } -func (g *Gateway) HandlePath(ctx *models.ReqContext) { - streamID := web.Params(ctx.Req)[":streamId"] - path := web.Params(ctx.Req)[":path"] +func (g *Gateway) HandlePipelinePush(ctx *models.ReqContext) { + channelID := web.Params(ctx.Req)["*"] body, err := io.ReadAll(ctx.Req.Body) if err != nil { @@ -108,13 +107,10 @@ func (g *Gateway) HandlePath(ctx *models.ReqContext) { } logger.Debug("Live channel push request", "protocol", "http", - "streamId", streamID, - "path", path, + "channel", channelID, "bodyLength", len(body), ) - channelID := "stream/" + streamID + "/" + path - ruleFound, err := g.GrafanaLive.Pipeline.ProcessInput(ctx.Req.Context(), ctx.OrgId, channelID, body) if err != nil { logger.Error("Pipeline input processing error", "error", err, "body", string(body)) diff --git a/pkg/services/live/pushws/push.go b/pkg/services/live/pushws/push.go deleted file mode 100644 index 90d747ffc2d..00000000000 --- a/pkg/services/live/pushws/push.go +++ /dev/null @@ -1,200 +0,0 @@ -package pushws - -import ( - "fmt" - "net/http" - "net/url" - "strings" - "time" - - "github.com/grafana/grafana/pkg/infra/log" - "github.com/grafana/grafana/pkg/services/live/convert" - "github.com/grafana/grafana/pkg/services/live/livecontext" - "github.com/grafana/grafana/pkg/services/live/managedstream" - "github.com/grafana/grafana/pkg/services/live/pushurl" - - "github.com/gorilla/websocket" - liveDto "github.com/grafana/grafana-plugin-sdk-go/live" -) - -var ( - logger = log.New("live.push_ws") -) - -// Handler handles WebSocket client connections that push data to Live. -type Handler struct { - managedStreamRunner *managedstream.Runner - config Config - upgrade *websocket.Upgrader - converter *convert.Converter -} - -// Config represents config for Handler. -type Config struct { - // ReadBufferSize is a parameter that is used for raw websocket Upgrader. - // If set to zero reasonable default value will be used. - ReadBufferSize int - - // WriteBufferSize is a parameter that is used for raw websocket Upgrader. - // If set to zero reasonable default value will be used. - WriteBufferSize int - - // MessageSizeLimit sets the maximum size in bytes of allowed message from client. - // By default DefaultWebsocketMessageSizeLimit will be used. - MessageSizeLimit int - - // CheckOrigin func to provide custom origin check logic, - // zero value means same host check. - CheckOrigin func(r *http.Request) bool - - // PingInterval sets interval server will send ping messages to clients. - // By default DefaultWebsocketPingInterval will be used. - PingInterval time.Duration -} - -// NewHandler creates new Handler. -func NewHandler(managedStreamRunner *managedstream.Runner, c Config) *Handler { - if c.CheckOrigin == nil { - c.CheckOrigin = sameHostOriginCheck() - } - upgrade := &websocket.Upgrader{ - ReadBufferSize: c.ReadBufferSize, - WriteBufferSize: c.WriteBufferSize, - CheckOrigin: c.CheckOrigin, - } - return &Handler{ - managedStreamRunner: managedStreamRunner, - config: c, - upgrade: upgrade, - converter: convert.NewConverter(), - } -} - -func sameHostOriginCheck() func(r *http.Request) bool { - return func(r *http.Request) bool { - err := checkSameHost(r) - if err != nil { - logger.Warn("Origin check failure", "origin", r.Header.Get("origin"), "error", err) - return false - } - return true - } -} - -func checkSameHost(r *http.Request) error { - origin := r.Header.Get("Origin") - if origin == "" { - return nil - } - u, err := url.Parse(origin) - if err != nil { - return fmt.Errorf("failed to parse Origin header %q: %w", origin, err) - } - if strings.EqualFold(r.Host, u.Host) { - return nil - } - return fmt.Errorf("request Origin %q is not authorized for Host %q", origin, r.Host) -} - -// Defaults. -const ( - DefaultWebsocketPingInterval = 25 * time.Second - DefaultWebsocketMessageSizeLimit = 1024 * 1024 // 1MB -) - -func (s *Handler) ServeHTTP(rw http.ResponseWriter, r *http.Request) { - streamID, ok := livecontext.GetContextStreamID(r.Context()) - if !ok || streamID == "" { - logger.Warn("Push request without stream ID") - rw.WriteHeader(http.StatusInternalServerError) - return - } - - conn, err := s.upgrade.Upgrade(rw, r, nil) - if err != nil { - return - } - - pingInterval := s.config.PingInterval - if pingInterval == 0 { - pingInterval = DefaultWebsocketPingInterval - } - messageSizeLimit := s.config.MessageSizeLimit - if messageSizeLimit == 0 { - messageSizeLimit = DefaultWebsocketMessageSizeLimit - } - - if messageSizeLimit > 0 { - conn.SetReadLimit(int64(messageSizeLimit)) - } - if pingInterval > 0 { - pongWait := pingInterval * 10 / 9 - _ = conn.SetReadDeadline(time.Now().Add(pongWait)) - conn.SetPongHandler(func(string) error { - _ = conn.SetReadDeadline(time.Now().Add(pongWait)) - return nil - }) - } - - go func() { - ticker := time.NewTicker(25 * time.Second) - defer ticker.Stop() - for { - select { - case <-r.Context().Done(): - return - case <-ticker.C: - deadline := time.Now().Add(pingInterval / 2) - err := conn.WriteControl(websocket.PingMessage, nil, deadline) - if err != nil { - return - } - } - } - }() - - user, ok := livecontext.GetContextSignedUser(r.Context()) - if !ok { - logger.Error("No user found in context") - rw.WriteHeader(http.StatusInternalServerError) - return - } - - for { - _, body, err := conn.ReadMessage() - if err != nil { - break - } - - stream, err := s.managedStreamRunner.GetOrCreateStream(user.OrgId, liveDto.ScopeStream, streamID) - if err != nil { - logger.Error("Error getting stream", "error", err) - continue - } - - // TODO Grafana 8: decide which formats to use or keep all. - urlValues := r.URL.Query() - frameFormat := pushurl.FrameFormatFromValues(urlValues) - - logger.Debug("Live Push request", - "protocol", "http", - "streamId", streamID, - "bodyLength", len(body), - "frameFormat", frameFormat, - ) - - metricFrames, err := s.converter.Convert(body, frameFormat) - if err != nil { - logger.Error("Error converting metrics", "error", err, "frameFormat", frameFormat) - continue - } - - for _, mf := range metricFrames { - err := stream.Push(mf.Key(), mf.Frame()) - if err != nil { - logger.Error("Error pushing frame", "error", err, "data", string(body)) - return - } - } - } -} diff --git a/pkg/services/live/pushws/push_pipeline.go b/pkg/services/live/pushws/push_pipeline.go new file mode 100644 index 00000000000..5651a5d4227 --- /dev/null +++ b/pkg/services/live/pushws/push_pipeline.go @@ -0,0 +1,82 @@ +package pushws + +import ( + "net/http" + + "github.com/grafana/grafana/pkg/services/live/convert" + "github.com/grafana/grafana/pkg/services/live/livecontext" + "github.com/grafana/grafana/pkg/services/live/pipeline" + + "github.com/gorilla/websocket" +) + +// PipelinePushHandler handles WebSocket client connections that push data to Live Pipeline. +type PipelinePushHandler struct { + pipeline *pipeline.Pipeline + config Config + upgrade *websocket.Upgrader + converter *convert.Converter +} + +// NewPathHandler creates new PipelinePushHandler. +func NewPipelinePushHandler(pipeline *pipeline.Pipeline, c Config) *PipelinePushHandler { + if c.CheckOrigin == nil { + c.CheckOrigin = sameHostOriginCheck() + } + upgrade := &websocket.Upgrader{ + ReadBufferSize: c.ReadBufferSize, + WriteBufferSize: c.WriteBufferSize, + CheckOrigin: c.CheckOrigin, + } + return &PipelinePushHandler{ + pipeline: pipeline, + config: c, + upgrade: upgrade, + converter: convert.NewConverter(), + } +} + +func (s *PipelinePushHandler) ServeHTTP(rw http.ResponseWriter, r *http.Request) { + channelID, ok := livecontext.GetContextChannelID(r.Context()) + if !ok || channelID == "" { + logger.Warn("Push request without channel") + rw.WriteHeader(http.StatusInternalServerError) + return + } + + user, ok := livecontext.GetContextSignedUser(r.Context()) + if !ok { + logger.Error("No user found in context") + rw.WriteHeader(http.StatusInternalServerError) + return + } + + conn, err := s.upgrade.Upgrade(rw, r, nil) + if err != nil { + return + } + setupWSConn(r.Context(), conn, s.config) + + for { + _, body, err := conn.ReadMessage() + if err != nil { + break + } + + logger.Debug("Live channel push request", + "protocol", "http", + "channel", channelID, + "bodyLength", len(body), + ) + + ruleFound, err := s.pipeline.ProcessInput(r.Context(), user.OrgId, channelID, body) + if err != nil { + logger.Error("Pipeline input processing error", "error", err, "body", string(body)) + return + } + if !ruleFound { + logger.Error("No conversion rule for a channel", "error", err, "channel", channelID) + return + } + } +} diff --git a/pkg/services/live/pushws/push_stream.go b/pkg/services/live/pushws/push_stream.go new file mode 100644 index 00000000000..a7892df2b97 --- /dev/null +++ b/pkg/services/live/pushws/push_stream.go @@ -0,0 +1,99 @@ +package pushws + +import ( + "net/http" + + "github.com/grafana/grafana/pkg/services/live/convert" + "github.com/grafana/grafana/pkg/services/live/livecontext" + "github.com/grafana/grafana/pkg/services/live/managedstream" + "github.com/grafana/grafana/pkg/services/live/pushurl" + + "github.com/gorilla/websocket" + liveDto "github.com/grafana/grafana-plugin-sdk-go/live" +) + +// Handler handles WebSocket client connections that push data to Live. +type Handler struct { + managedStreamRunner *managedstream.Runner + config Config + upgrade *websocket.Upgrader + converter *convert.Converter +} + +// NewHandler creates new Handler. +func NewHandler(managedStreamRunner *managedstream.Runner, c Config) *Handler { + if c.CheckOrigin == nil { + c.CheckOrigin = sameHostOriginCheck() + } + upgrade := &websocket.Upgrader{ + ReadBufferSize: c.ReadBufferSize, + WriteBufferSize: c.WriteBufferSize, + CheckOrigin: c.CheckOrigin, + } + return &Handler{ + managedStreamRunner: managedStreamRunner, + config: c, + upgrade: upgrade, + converter: convert.NewConverter(), + } +} + +func (s *Handler) ServeHTTP(rw http.ResponseWriter, r *http.Request) { + streamID, ok := livecontext.GetContextStreamID(r.Context()) + if !ok || streamID == "" { + logger.Warn("Push request without stream ID") + rw.WriteHeader(http.StatusInternalServerError) + return + } + + user, ok := livecontext.GetContextSignedUser(r.Context()) + if !ok { + logger.Error("No user found in context") + rw.WriteHeader(http.StatusInternalServerError) + return + } + + conn, err := s.upgrade.Upgrade(rw, r, nil) + if err != nil { + return + } + setupWSConn(r.Context(), conn, s.config) + + for { + _, body, err := conn.ReadMessage() + if err != nil { + break + } + + stream, err := s.managedStreamRunner.GetOrCreateStream(user.OrgId, liveDto.ScopeStream, streamID) + if err != nil { + logger.Error("Error getting stream", "error", err) + continue + } + + // TODO Grafana 8: decide which formats to use or keep all. + urlValues := r.URL.Query() + frameFormat := pushurl.FrameFormatFromValues(urlValues) + + logger.Debug("Live Push request", + "protocol", "http", + "streamId", streamID, + "bodyLength", len(body), + "frameFormat", frameFormat, + ) + + metricFrames, err := s.converter.Convert(body, frameFormat) + if err != nil { + logger.Error("Error converting metrics", "error", err, "frameFormat", frameFormat) + continue + } + + for _, mf := range metricFrames { + err := stream.Push(mf.Key(), mf.Frame()) + if err != nil { + logger.Error("Error pushing frame", "error", err, "data", string(body)) + return + } + } + } +} diff --git a/pkg/services/live/pushws/ws.go b/pkg/services/live/pushws/ws.go new file mode 100644 index 00000000000..f7ae55521d2 --- /dev/null +++ b/pkg/services/live/pushws/ws.go @@ -0,0 +1,113 @@ +package pushws + +import ( + "context" + "fmt" + "net/http" + "net/url" + "strings" + "time" + + "github.com/gorilla/websocket" + + "github.com/grafana/grafana/pkg/infra/log" +) + +var ( + logger = log.New("live.push_ws") +) + +// Config represents config for Handler. +type Config struct { + // ReadBufferSize is a parameter that is used for raw websocket Upgrader. + // If set to zero reasonable default value will be used. + ReadBufferSize int + + // WriteBufferSize is a parameter that is used for raw websocket Upgrader. + // If set to zero reasonable default value will be used. + WriteBufferSize int + + // MessageSizeLimit sets the maximum size in bytes of allowed message from client. + // By default DefaultWebsocketMessageSizeLimit will be used. + MessageSizeLimit int + + // CheckOrigin func to provide custom origin check logic, + // zero value means same host check. + CheckOrigin func(r *http.Request) bool + + // PingInterval sets interval server will send ping messages to clients. + // By default DefaultWebsocketPingInterval will be used. + PingInterval time.Duration +} + +func sameHostOriginCheck() func(r *http.Request) bool { + return func(r *http.Request) bool { + err := checkSameHost(r) + if err != nil { + logger.Warn("Origin check failure", "origin", r.Header.Get("origin"), "error", err) + return false + } + return true + } +} + +func checkSameHost(r *http.Request) error { + origin := r.Header.Get("Origin") + if origin == "" { + return nil + } + u, err := url.Parse(origin) + if err != nil { + return fmt.Errorf("failed to parse Origin header %q: %w", origin, err) + } + if strings.EqualFold(r.Host, u.Host) { + return nil + } + return fmt.Errorf("request Origin %q is not authorized for Host %q", origin, r.Host) +} + +// Defaults. +const ( + DefaultWebsocketPingInterval = 25 * time.Second + DefaultWebsocketMessageSizeLimit = 1024 * 1024 // 1MB +) + +func setupWSConn(ctx context.Context, conn *websocket.Conn, config Config) { + pingInterval := config.PingInterval + if pingInterval == 0 { + pingInterval = DefaultWebsocketPingInterval + } + messageSizeLimit := config.MessageSizeLimit + if messageSizeLimit == 0 { + messageSizeLimit = DefaultWebsocketMessageSizeLimit + } + + if messageSizeLimit > 0 { + conn.SetReadLimit(int64(messageSizeLimit)) + } + if pingInterval > 0 { + pongWait := pingInterval * 10 / 9 + _ = conn.SetReadDeadline(time.Now().Add(pongWait)) + conn.SetPongHandler(func(string) error { + _ = conn.SetReadDeadline(time.Now().Add(pongWait)) + return nil + }) + } + + go func() { + ticker := time.NewTicker(25 * time.Second) + defer ticker.Stop() + for { + select { + case <-ctx.Done(): + return + case <-ticker.C: + deadline := time.Now().Add(pingInterval / 2) + err := conn.WriteControl(websocket.PingMessage, nil, deadline) + if err != nil { + return + } + } + } + }() +}