From 31903778aecf9b2c5362f9483ee6c2bae1eaa742 Mon Sep 17 00:00:00 2001 From: Alexander Emelin Date: Fri, 30 Jul 2021 21:05:39 +0300 Subject: [PATCH] Live: display stream rate, fix duplicate channels in list response (#37365) --- pkg/services/live/managedstream/runner.go | 119 ++++++++++++------ .../live/managedstream/runner_test.go | 73 ++++++++++- pkg/services/live/pushhttp/push.go | 2 +- pkg/services/live/pushws/push.go | 2 +- pkg/services/live/survey/survey.go | 27 +++- .../grafana/components/QueryEditor.tsx | 2 +- 6 files changed, 174 insertions(+), 51 deletions(-) diff --git a/pkg/services/live/managedstream/runner.go b/pkg/services/live/managedstream/runner.go index b98a63c36d7..3069cdff022 100644 --- a/pkg/services/live/managedstream/runner.go +++ b/pkg/services/live/managedstream/runner.go @@ -4,6 +4,7 @@ import ( "context" "encoding/json" "fmt" + "sort" "sync" "time" @@ -36,13 +37,23 @@ func NewRunner(publisher models.ChannelPublisher, frameCache FrameCache) *Runner } func (r *Runner) GetManagedChannels(orgID int64) ([]*ManagedChannel, error) { - channels := make([]*ManagedChannel, 0) - for _, v := range r.Streams(orgID) { - streamChannels, err := v.ListChannels(orgID) - if err != nil { - return nil, err + activeChannels, err := r.frameCache.GetActiveChannels(orgID) + if err != nil { + return []*ManagedChannel{}, fmt.Errorf("error getting active managed stream paths: %v", err) + } + channels := make([]*ManagedChannel, 0, len(activeChannels)) + for ch, schema := range activeChannels { + managedChannel := &ManagedChannel{ + Channel: ch, + Data: schema, } - channels = append(channels, streamChannels...) + // Enrich with minute rate. + channel, _ := live.ParseChannel(managedChannel.Channel) + namespaceStream, ok := r.streams[orgID][channel.Namespace] + if ok { + managedChannel.MinuteRate = namespaceStream.minuteRate(channel.Path) + } + channels = append(channels, managedChannel) } // Hardcode sample streams @@ -54,16 +65,24 @@ func (r *Runner) GetManagedChannels(orgID int64) ([]*ManagedChannel, error) { ), data.IncludeSchemaOnly) if err == nil { channels = append(channels, &ManagedChannel{ - Channel: "plugin/testdata/random-2s-stream", - Data: frameJSON, + Channel: "plugin/testdata/random-2s-stream", + Data: frameJSON, + MinuteRate: 30, }, &ManagedChannel{ - Channel: "plugin/testdata/random-flakey-stream", - Data: frameJSON, + Channel: "plugin/testdata/random-flakey-stream", + Data: frameJSON, + MinuteRate: 150, }, &ManagedChannel{ - Channel: "plugin/testdata/random-20Hz-stream", - Data: frameJSON, + Channel: "plugin/testdata/random-20Hz-stream", + Data: frameJSON, + MinuteRate: 1200, }) } + + sort.Slice(channels, func(i, j int) bool { + return channels[i].Channel < channels[j].Channel + }) + return channels, nil } @@ -92,7 +111,7 @@ func (r *Runner) GetOrCreateStream(orgID int64, streamID string) (*ManagedStream } s, ok := r.streams[orgID][streamID] if !ok { - s = NewManagedStream(streamID, r.publisher, r.frameCache) + s = NewManagedStream(streamID, orgID, r.publisher, r.frameCache) r.streams[orgID][streamID] = s } return s, nil @@ -101,47 +120,41 @@ func (r *Runner) GetOrCreateStream(orgID int64, streamID string) (*ManagedStream // ManagedStream holds the state of a managed stream. type ManagedStream struct { id string + orgID int64 start time.Time publisher models.ChannelPublisher frameCache FrameCache + rateMu sync.RWMutex + rates map[string][60]rateEntry +} + +type rateEntry struct { + time uint32 + count int32 } // NewManagedStream creates new ManagedStream. -func NewManagedStream(id string, publisher models.ChannelPublisher, schemaUpdater FrameCache) *ManagedStream { +func NewManagedStream(id string, orgID int64, publisher models.ChannelPublisher, schemaUpdater FrameCache) *ManagedStream { return &ManagedStream{ id: id, + orgID: orgID, start: time.Now(), publisher: publisher, frameCache: schemaUpdater, + rates: map[string][60]rateEntry{}, } } // ManagedChannel represents a managed stream. type ManagedChannel struct { - Channel string `json:"channel"` - Data json.RawMessage `json:"data"` -} - -// ListChannels returns info for the UI about this stream. -func (s *ManagedStream) ListChannels(orgID int64) ([]*ManagedChannel, error) { - paths, err := s.frameCache.GetActiveChannels(orgID) - if err != nil { - return []*ManagedChannel{}, fmt.Errorf("error getting active managed stream paths: %v", err) - } - info := make([]*ManagedChannel, 0, len(paths)) - for k, v := range paths { - managedChannel := &ManagedChannel{ - Channel: k, - Data: v, - } - info = append(info, managedChannel) - } - return info, nil + Channel string `json:"channel"` + MinuteRate int64 `json:"minute_rate"` + Data json.RawMessage `json:"data"` } // Push sends frame to the stream and saves it for later retrieval by subscribers. // unstableSchema flag can be set to disable schema caching for a path. -func (s *ManagedStream) Push(orgID int64, path string, frame *data.Frame) error { +func (s *ManagedStream) Push(path string, frame *data.Frame) error { jsonFrameCache, err := data.FrameToJSONCache(frame) if err != nil { return err @@ -150,7 +163,7 @@ func (s *ManagedStream) Push(orgID int64, path string, frame *data.Frame) error // The channel this will be posted into. channel := live.Channel{Scope: live.ScopeStream, Namespace: s.id, Path: path}.String() - isUpdated, err := s.frameCache.Update(orgID, channel, jsonFrameCache) + isUpdated, err := s.frameCache.Update(s.orgID, channel, jsonFrameCache) if err != nil { logger.Error("Error updating managed stream schema", "error", err) return err @@ -165,7 +178,41 @@ func (s *ManagedStream) Push(orgID int64, path string, frame *data.Frame) error frameJSON := jsonFrameCache.Bytes(include) logger.Debug("Publish data to channel", "channel", channel, "dataLength", len(frameJSON)) - return s.publisher(orgID, channel, frameJSON) + s.incRate(path, time.Now().Unix()) + return s.publisher(s.orgID, channel, frameJSON) +} + +func (s *ManagedStream) incRate(path string, nowUnix int64) { + s.rateMu.Lock() + pathRate, ok := s.rates[path] + if !ok { + pathRate = [60]rateEntry{} + } + now := time.Unix(nowUnix, 0) + slot := now.Second() % 60 + if pathRate[slot].time != uint32(nowUnix) { + pathRate[slot].count = 0 + } + pathRate[slot].time = uint32(nowUnix) + pathRate[slot].count += 1 + s.rates[path] = pathRate + s.rateMu.Unlock() +} + +func (s *ManagedStream) minuteRate(path string) int64 { + var total int64 + s.rateMu.RLock() + defer s.rateMu.RUnlock() + pathRate, ok := s.rates[path] + if !ok { + return 0 + } + for _, val := range pathRate { + if val.time > uint32(time.Now().Unix()-60) { + total += int64(val.count) + } + } + return total } func (s *ManagedStream) GetHandlerForPath(_ string) (models.ChannelHandler, error) { diff --git a/pkg/services/live/managedstream/runner_test.go b/pkg/services/live/managedstream/runner_test.go index 1eca4fe7004..69153a5a9b4 100644 --- a/pkg/services/live/managedstream/runner_test.go +++ b/pkg/services/live/managedstream/runner_test.go @@ -2,22 +2,83 @@ package managedstream import ( "testing" + "time" + "github.com/grafana/grafana-plugin-sdk-go/data" "github.com/stretchr/testify/require" ) type testPublisher struct { - orgID int64 - t *testing.T + t *testing.T } -func (p *testPublisher) publish(orgID int64, _ string, _ []byte) error { - require.Equal(p.t, p.orgID, orgID) +func (p *testPublisher) publish(_ int64, _ string, _ []byte) error { return nil } func TestNewManagedStream(t *testing.T) { - publisher := &testPublisher{orgID: 1, t: t} - c := NewManagedStream("a", publisher.publish, NewMemoryFrameCache()) + publisher := &testPublisher{t: t} + c := NewManagedStream("a", 1, publisher.publish, NewMemoryFrameCache()) require.NotNil(t, c) } + +func TestManagedStreamMinuteRate(t *testing.T) { + publisher := &testPublisher{t: t} + c := NewManagedStream("a", 1, publisher.publish, NewMemoryFrameCache()) + require.NotNil(t, c) + + c.incRate("test1", time.Now().Unix()) + require.Equal(t, int64(1), c.minuteRate("test1")) + require.Equal(t, int64(0), c.minuteRate("test2")) + c.incRate("test1", time.Now().Unix()) + require.Equal(t, int64(2), c.minuteRate("test1")) + + nowUnix := time.Now().Unix() + for i := 0; i < 1000; i++ { + unixTime := nowUnix + int64(i) + c.incRate("test3", unixTime) + } + require.Equal(t, int64(60), c.minuteRate("test3")) + + c.incRate("test3", nowUnix+999) + require.Equal(t, int64(61), c.minuteRate("test3")) +} + +func TestGetManagedStreams(t *testing.T) { + publisher := &testPublisher{t: t} + frameCache := NewMemoryFrameCache() + runner := NewRunner(publisher.publish, frameCache) + s1, err := runner.GetOrCreateStream(1, "test1") + require.NoError(t, err) + s2, err := runner.GetOrCreateStream(1, "test2") + require.NoError(t, err) + + managedChannels, err := runner.GetManagedChannels(1) + require.NoError(t, err) + require.Len(t, managedChannels, 3) // 3 hardcoded testdata streams. + + err = s1.Push("cpu1", data.NewFrame("cpu1")) + require.NoError(t, err) + + err = s1.Push("cpu2", data.NewFrame("cpu2")) + require.NoError(t, err) + + err = s2.Push("cpu1", data.NewFrame("cpu1")) + require.NoError(t, err) + + managedChannels, err = runner.GetManagedChannels(1) + require.NoError(t, err) + require.Len(t, managedChannels, 6) // 3 hardcoded testdata streams + 3 test channels. + require.Equal(t, "stream/test1/cpu1", managedChannels[3].Channel) + require.Equal(t, "stream/test1/cpu2", managedChannels[4].Channel) + require.Equal(t, "stream/test2/cpu1", managedChannels[5].Channel) + + // Different org. + s3, err := runner.GetOrCreateStream(2, "test1") + require.NoError(t, err) + err = s3.Push("cpu1", data.NewFrame("cpu1")) + require.NoError(t, err) + managedChannels, err = runner.GetManagedChannels(1) + require.NoError(t, err) + require.Len(t, managedChannels, 6) // Not affected by other org. +} diff --git a/pkg/services/live/pushhttp/push.go b/pkg/services/live/pushhttp/push.go index 8ccf828f77f..60b50141df5 100644 --- a/pkg/services/live/pushhttp/push.go +++ b/pkg/services/live/pushhttp/push.go @@ -87,7 +87,7 @@ func (g *Gateway) Handle(ctx *models.ReqContext) { // interval = "1s" vs flush_interval = "5s" for _, mf := range metricFrames { - err := stream.Push(ctx.SignedInUser.OrgId, mf.Key(), mf.Frame()) + err := stream.Push(mf.Key(), mf.Frame()) if err != nil { logger.Error("Error pushing frame", "error", err, "data", string(body)) ctx.Resp.WriteHeader(http.StatusInternalServerError) diff --git a/pkg/services/live/pushws/push.go b/pkg/services/live/pushws/push.go index c41092b0cb9..1f513633c04 100644 --- a/pkg/services/live/pushws/push.go +++ b/pkg/services/live/pushws/push.go @@ -189,7 +189,7 @@ func (s *Handler) ServeHTTP(rw http.ResponseWriter, r *http.Request) { } for _, mf := range metricFrames { - err := stream.Push(user.OrgId, mf.Key(), mf.Frame()) + 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/survey/survey.go b/pkg/services/live/survey/survey.go index e4743aab623..e3d98fc1fc4 100644 --- a/pkg/services/live/survey/survey.go +++ b/pkg/services/live/survey/survey.go @@ -5,6 +5,8 @@ import ( "encoding/json" "errors" "fmt" + "sort" + "strings" "time" "github.com/centrifugal/centrifuge" @@ -92,8 +94,7 @@ func (c *Caller) CallManagedStreams(orgID int64) ([]*managedstream.ManagedChanne return nil, err } - channels := make([]*managedstream.ManagedChannel, 0) - duplicatesCheck := map[string]struct{}{} + channels := map[string]*managedstream.ManagedChannel{} for _, result := range resp { if result.Code != 0 { @@ -105,13 +106,27 @@ func (c *Caller) CallManagedStreams(orgID int64) ([]*managedstream.ManagedChanne return nil, err } for _, ch := range res.Channels { - if _, ok := duplicatesCheck[ch.Channel]; ok { + if _, ok := channels[ch.Channel]; ok { + if strings.HasPrefix(ch.Channel, "plugin/testdata/") { + // Skip adding testdata rates since it works over different + // mechanism (plugin stream) and the minute rate is hardcoded. + continue + } + channels[ch.Channel].MinuteRate += ch.MinuteRate continue } - channels = append(channels, ch) - duplicatesCheck[ch.Channel] = struct{}{} + channels[ch.Channel] = ch } } - return channels, nil + result := make([]*managedstream.ManagedChannel, 0, len(channels)) + for _, v := range channels { + result = append(result, v) + } + + sort.Slice(result, func(i, j int) bool { + return result[i].Channel < result[j].Channel + }) + + return result, nil } diff --git a/public/app/plugins/datasource/grafana/components/QueryEditor.tsx b/public/app/plugins/datasource/grafana/components/QueryEditor.tsx index bae1730c094..2772cb01f2b 100644 --- a/public/app/plugins/datasource/grafana/components/QueryEditor.tsx +++ b/public/app/plugins/datasource/grafana/components/QueryEditor.tsx @@ -54,7 +54,7 @@ export class QueryEditor extends PureComponent { } return { value: c.channel, - label: c.channel, + label: c.channel + ' [' + c.minute_rate + ' msg/min]', }; });