Live: display stream rate, fix duplicate channels in list response (#37365)

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)
+	activeChannels, err := r.frameCache.GetActiveChannels(orgID)
-	for _, v := range r.Streams(orgID) {
+	if err != nil {
-		streamChannels, err := v.ListChannels(orgID)
+		return []*ManagedChannel{}, fmt.Errorf("error getting active managed stream paths: %v", err)
-		if err != nil {
+	}
-			return nil, 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",
+			Channel:    "plugin/testdata/random-2s-stream",
-			Data:    frameJSON,
+			Data:       frameJSON,
+			MinuteRate: 30,
 		}, &ManagedChannel{
-			Channel: "plugin/testdata/random-flakey-stream",
+			Channel:    "plugin/testdata/random-flakey-stream",
-			Data:    frameJSON,
+			Data:       frameJSON,
+			MinuteRate: 150,
 		}, &ManagedChannel{
-			Channel: "plugin/testdata/random-20Hz-stream",
+			Channel:    "plugin/testdata/random-20Hz-stream",
-			Data:    frameJSON,
+			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"`
+	Channel    string          `json:"channel"`
-	Data    json.RawMessage `json:"data"`
+	MinuteRate int64           `json:"minute_rate"`
-}
+	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
 }
 
 // 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) {

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 {
+func (p *testPublisher) publish(_ int64, _ string, _ []byte) error {
-	require.Equal(p.t, p.orgID, orgID)
 	return nil
 }
 
 func TestNewManagedStream(t *testing.T) {
-	publisher := &testPublisher{orgID: 1, t: t}
+	publisher := &testPublisher{t: t}
-	c := NewManagedStream("a", publisher.publish, NewMemoryFrameCache())
+	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.
+}

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)

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

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)
+	channels := map[string]*managedstream.ManagedChannel{}
-	duplicatesCheck := map[string]struct{}{}
 
 	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)
+			channels[ch.Channel] = ch
-			duplicatesCheck[ch.Channel] = struct{}{}
 		}
 	}
 
-	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
 }

public/app/plugins/datasource/grafana/components/QueryEditor.tsx

@@ -54,7 +54,7 @@ export class QueryEditor extends PureComponent<Props, State> {
               }
               return {
                 value: c.channel,
-                label: c.channel,
+                label: c.channel + ' [' + c.minute_rate + ' msg/min]',
               };
             });