opensource/grafana
1
0
Fork 0
mirror of https://github.com/grafana/grafana.git synced 2025-07-30 05:02:12 +08:00
Code Issues Packages Projects Releases Wiki Activity

Chore: Remove context.TODO (#43458)

Browse Source 
* Remove context.TODO() from services

* Fix live test
This commit is contained in:
idafurjes
2021-12-28 10:26:18 +01:00
committed by GitHub
parent 56b3dc5445
commit 56c3875bb9
24 changed files with 81 additions and 77 deletions
Download Patch File Download Diff File Expand all files Collapse all files

2
pkg/api/app_routes.go
View File

@ -34,7 +34,7 @@ func (hs *HTTPServer) initAppPluginRoutes(r *web.Mux) {
TLSHandshakeTimeout: 10 * time.Second, TLSHandshakeTimeout: 10 * time.Second,
} }
for _, plugin := range hs.pluginStore.Plugins(context.TODO(), plugins.App) { for _, plugin := range hs.pluginStore.Plugins(context.Background(), plugins.App) {
for _, route := range plugin.Routes { for _, route := range plugin.Routes {
url := util.JoinURLFragments("/api/plugin-proxy/"+plugin.ID, route.Path) url := util.JoinURLFragments("/api/plugin-proxy/"+plugin.ID, route.Path)
handlers := make([]web.Handler, 0) handlers := make([]web.Handler, 0)

2
pkg/api/ldap_debug.go
View File

@ -177,7 +177,7 @@ func (hs *HTTPServer) PostSyncUserWithLDAP(c *models.ReqContext) response.Respon
authModuleQuery := &models.GetAuthInfoQuery{UserId: query.Result.Id, AuthModule: models.AuthModuleLDAP} authModuleQuery := &models.GetAuthInfoQuery{UserId: query.Result.Id, AuthModule: models.AuthModuleLDAP}
if err := bus.DispatchCtx(context.TODO(), authModuleQuery); err != nil { // validate the userId comes from LDAP if err := bus.DispatchCtx(c.Req.Context(), authModuleQuery); err != nil { // validate the userId comes from LDAP
if errors.Is(err, models.ErrUserNotFound) { if errors.Is(err, models.ErrUserNotFound) {
return response.Error(404, models.ErrUserNotFound.Error(), nil) return response.Error(404, models.ErrUserNotFound.Error(), nil)
} }

4
pkg/infra/remotecache/remotecache.go
View File

@ -28,7 +28,7 @@ const (
) )
func ProvideService(cfg *setting.Cfg, sqlStore *sqlstore.SQLStore) (*RemoteCache, error) { func ProvideService(cfg *setting.Cfg, sqlStore *sqlstore.SQLStore) (*RemoteCache, error) {
client, err := createClient(context.Background(), cfg.RemoteCacheOptions, sqlStore) client, err := createClient(cfg.RemoteCacheOptions, sqlStore)
if err != nil { if err != nil {
return nil, err return nil, err
} }
@ -95,7 +95,7 @@ func (ds *RemoteCache) Run(ctx context.Context) error {
return ctx.Err() return ctx.Err()
} }
func createClient(ctx context.Context, opts *setting.RemoteCacheOptions, sqlstore *sqlstore.SQLStore) (CacheStorage, error) { func createClient(opts *setting.RemoteCacheOptions, sqlstore *sqlstore.SQLStore) (CacheStorage, error) {
if opts.Name == redisCacheType { if opts.Name == redisCacheType {
return newRedisStorage(opts) return newRedisStorage(opts)
} }

2
pkg/infra/remotecache/remotecache_test.go
View File

@ -44,7 +44,7 @@ func TestCachedBasedOnConfig(t *testing.T) {
} }
func TestInvalidCacheTypeReturnsError(t *testing.T) { func TestInvalidCacheTypeReturnsError(t *testing.T) {
_, err := createClient(context.Background(), &setting.RemoteCacheOptions{Name: "invalid"}, nil) _, err := createClient(&setting.RemoteCacheOptions{Name: "invalid"}, nil)
assert.Equal(t, err, ErrInvalidCacheType) assert.Equal(t, err, ErrInvalidCacheType)
} }

42
pkg/services/live/live.go
View File

@ -158,7 +158,7 @@ func ProvideService(plugCtxProvider *plugincontext.Provider, cfg *setting.Cfg, r
redisClient := redis.NewClient(&redis.Options{ redisClient := redis.NewClient(&redis.Options{
Addr: g.Cfg.LiveHAEngineAddress, Addr: g.Cfg.LiveHAEngineAddress,
}) })
cmd := redisClient.Ping(context.TODO()) cmd := redisClient.Ping(context.Background())
if _, err := cmd.Result(); err != nil { if _, err := cmd.Result(); err != nil {
return nil, fmt.Errorf("error pinging Redis: %v", err) return nil, fmt.Errorf("error pinging Redis: %v", err)
} }
@ -206,7 +206,7 @@ func ProvideService(plugCtxProvider *plugincontext.Provider, cfg *setting.Cfg, r
// This can be unreasonable to have in production scenario with many // This can be unreasonable to have in production scenario with many
// organizations. // organizations.
orgQuery := &models.SearchOrgsQuery{} orgQuery := &models.SearchOrgsQuery{}
err := sqlstore.SearchOrgs(context.TODO(), orgQuery) err := sqlstore.SearchOrgs(context.Background(), orgQuery)
if err != nil { if err != nil {
return nil, fmt.Errorf("can't get org list: %w", err) return nil, fmt.Errorf("can't get org list: %w", err)
} }
@ -278,7 +278,7 @@ func ProvideService(plugCtxProvider *plugincontext.Provider, cfg *setting.Cfg, r
// Called when client subscribes to the channel. // Called when client subscribes to the channel.
client.OnSubscribe(func(e centrifuge.SubscribeEvent, cb centrifuge.SubscribeCallback) { client.OnSubscribe(func(e centrifuge.SubscribeEvent, cb centrifuge.SubscribeCallback) {
err := runConcurrentlyIfNeeded(client.Context(), semaphore, func() { err := runConcurrentlyIfNeeded(client.Context(), semaphore, func() {
cb(g.handleOnSubscribe(client, e)) cb(g.handleOnSubscribe(context.Background(), client, e))
}) })
if err != nil { if err != nil {
cb(centrifuge.SubscribeReply{}, err) cb(centrifuge.SubscribeReply{}, err)
@ -290,7 +290,7 @@ func ProvideService(plugCtxProvider *plugincontext.Provider, cfg *setting.Cfg, r
// allows some simple prototypes to work quickly. // allows some simple prototypes to work quickly.
client.OnPublish(func(e centrifuge.PublishEvent, cb centrifuge.PublishCallback) { client.OnPublish(func(e centrifuge.PublishEvent, cb centrifuge.PublishCallback) {
err := runConcurrentlyIfNeeded(client.Context(), semaphore, func() { err := runConcurrentlyIfNeeded(client.Context(), semaphore, func() {
cb(g.handleOnPublish(client, e)) cb(g.handleOnPublish(context.Background(), client, e))
}) })
if err != nil { if err != nil {
cb(centrifuge.PublishReply{}, err) cb(centrifuge.PublishReply{}, err)
@ -429,8 +429,8 @@ type GrafanaLive struct {
usageStats usageStats usageStats usageStats
} }
func (g *GrafanaLive) getStreamPlugin(pluginID string) (backend.StreamHandler, error) { func (g *GrafanaLive) getStreamPlugin(ctx context.Context, pluginID string) (backend.StreamHandler, error) {
plugin, exists := g.pluginStore.Plugin(context.TODO(), pluginID) plugin, exists := g.pluginStore.Plugin(ctx, pluginID)
if !exists { if !exists {
return nil, fmt.Errorf("plugin not found: %s", pluginID) return nil, fmt.Errorf("plugin not found: %s", pluginID)
} }
@ -604,7 +604,7 @@ func (g *GrafanaLive) handleOnRPC(client *centrifuge.Client, e centrifuge.RPCEve
}, nil }, nil
} }
func (g *GrafanaLive) handleOnSubscribe(client *centrifuge.Client, e centrifuge.SubscribeEvent) (centrifuge.SubscribeReply, error) { func (g *GrafanaLive) handleOnSubscribe(ctx context.Context, client *centrifuge.Client, e centrifuge.SubscribeEvent) (centrifuge.SubscribeReply, error) {
logger.Debug("Client wants to subscribe", "user", client.UserID(), "client", client.ID(), "channel", e.Channel) logger.Debug("Client wants to subscribe", "user", client.UserID(), "client", client.ID(), "channel", e.Channel)
user, ok := livecontext.GetContextSignedUser(client.Context()) user, ok := livecontext.GetContextSignedUser(client.Context())
@ -668,7 +668,7 @@ func (g *GrafanaLive) handleOnSubscribe(client *centrifuge.Client, e centrifuge.
} }
} }
if !ruleFound { if !ruleFound {
handler, addr, err := g.GetChannelHandler(user, channel) handler, addr, err := g.GetChannelHandler(ctx, user, channel)
if err != nil { if err != nil {
if errors.Is(err, live.ErrInvalidChannelID) { if errors.Is(err, live.ErrInvalidChannelID) {
logger.Info("Invalid channel ID", "user", client.UserID(), "client", client.ID(), "channel", e.Channel) logger.Info("Invalid channel ID", "user", client.UserID(), "client", client.ID(), "channel", e.Channel)
@ -704,7 +704,7 @@ func (g *GrafanaLive) handleOnSubscribe(client *centrifuge.Client, e centrifuge.
}, nil }, nil
} }
func (g *GrafanaLive) handleOnPublish(client *centrifuge.Client, e centrifuge.PublishEvent) (centrifuge.PublishReply, error) { func (g *GrafanaLive) handleOnPublish(ctx context.Context, client *centrifuge.Client, e centrifuge.PublishEvent) (centrifuge.PublishReply, error) {
logger.Debug("Client wants to publish", "user", client.UserID(), "client", client.ID(), "channel", e.Channel) logger.Debug("Client wants to publish", "user", client.UserID(), "client", client.ID(), "channel", e.Channel)
user, ok := livecontext.GetContextSignedUser(client.Context()) user, ok := livecontext.GetContextSignedUser(client.Context())
@ -761,7 +761,7 @@ func (g *GrafanaLive) handleOnPublish(client *centrifuge.Client, e centrifuge.Pu
} }
} }
handler, addr, err := g.GetChannelHandler(user, channel) handler, addr, err := g.GetChannelHandler(ctx, user, channel)
if err != nil { if err != nil {
if errors.Is(err, live.ErrInvalidChannelID) { if errors.Is(err, live.ErrInvalidChannelID) {
logger.Info("Invalid channel ID", "user", client.UserID(), "client", client.ID(), "channel", e.Channel) logger.Info("Invalid channel ID", "user", client.UserID(), "client", client.ID(), "channel", e.Channel)
@ -831,7 +831,7 @@ func publishStatusToHTTPError(status backend.PublishStreamStatus) (int, string)
} }
// GetChannelHandler gives thread-safe access to the channel. // GetChannelHandler gives thread-safe access to the channel.
func (g *GrafanaLive) GetChannelHandler(user *models.SignedInUser, channel string) (models.ChannelHandler, live.Channel, error) { func (g *GrafanaLive) GetChannelHandler(ctx context.Context, user *models.SignedInUser, channel string) (models.ChannelHandler, live.Channel, error) {
// Parse the identifier ${scope}/${namespace}/${path} // Parse the identifier ${scope}/${namespace}/${path}
addr, err := live.ParseChannel(channel) addr, err := live.ParseChannel(channel)
if err != nil { if err != nil {
@ -854,7 +854,7 @@ func (g *GrafanaLive) GetChannelHandler(user *models.SignedInUser, channel strin
return c, addr, nil return c, addr, nil
} }
getter, err := g.GetChannelHandlerFactory(user, addr.Scope, addr.Namespace) getter, err := g.GetChannelHandlerFactory(ctx, user, addr.Scope, addr.Namespace)
if err != nil { if err != nil {
return nil, addr, fmt.Errorf("error getting channel handler factory: %w", err) return nil, addr, fmt.Errorf("error getting channel handler factory: %w", err)
} }
@ -872,14 +872,14 @@ func (g *GrafanaLive) GetChannelHandler(user *models.SignedInUser, channel strin
// GetChannelHandlerFactory gets a ChannelHandlerFactory for a namespace. // GetChannelHandlerFactory gets a ChannelHandlerFactory for a namespace.
// It gives thread-safe access to the channel. // It gives thread-safe access to the channel.
func (g *GrafanaLive) GetChannelHandlerFactory(user *models.SignedInUser, scope string, namespace string) (models.ChannelHandlerFactory, error) { func (g *GrafanaLive) GetChannelHandlerFactory(ctx context.Context, user *models.SignedInUser, scope string, namespace string) (models.ChannelHandlerFactory, error) {
switch scope { switch scope {
case live.ScopeGrafana: case live.ScopeGrafana:
return g.handleGrafanaScope(user, namespace) return g.handleGrafanaScope(user, namespace)
case live.ScopePlugin: case live.ScopePlugin:
return g.handlePluginScope(user, namespace) return g.handlePluginScope(ctx, user, namespace)
case live.ScopeDatasource: case live.ScopeDatasource:
return g.handleDatasourceScope(user, namespace) return g.handleDatasourceScope(ctx, user, namespace)
case live.ScopeStream: case live.ScopeStream:
return g.handleStreamScope(user, namespace) return g.handleStreamScope(user, namespace)
default: default:
@ -894,7 +894,7 @@ func (g *GrafanaLive) handleGrafanaScope(_ *models.SignedInUser, namespace strin
return nil, fmt.Errorf("unknown feature: %q", namespace) return nil, fmt.Errorf("unknown feature: %q", namespace)
} }
func (g *GrafanaLive) handlePluginScope(_ *models.SignedInUser, namespace string) (models.ChannelHandlerFactory, error) { func (g *GrafanaLive) handlePluginScope(ctx context.Context, _ *models.SignedInUser, namespace string) (models.ChannelHandlerFactory, error) {
// Temporary hack until we have a more generic solution later on // Temporary hack until we have a more generic solution later on
if namespace == "cloudwatch" { if namespace == "cloudwatch" {
return &cloudwatch.LogQueryRunnerSupplier{ return &cloudwatch.LogQueryRunnerSupplier{
@ -902,7 +902,7 @@ func (g *GrafanaLive) handlePluginScope(_ *models.SignedInUser, namespace string
Service: g.LogsService, Service: g.LogsService,
}, nil }, nil
} }
streamHandler, err := g.getStreamPlugin(namespace) streamHandler, err := g.getStreamPlugin(ctx, namespace)
if err != nil { if err != nil {
return nil, fmt.Errorf("can't find stream plugin: %s", namespace) return nil, fmt.Errorf("can't find stream plugin: %s", namespace)
} }
@ -919,12 +919,12 @@ func (g *GrafanaLive) handleStreamScope(u *models.SignedInUser, namespace string
return g.ManagedStreamRunner.GetOrCreateStream(u.OrgId, live.ScopeStream, namespace) return g.ManagedStreamRunner.GetOrCreateStream(u.OrgId, live.ScopeStream, namespace)
} }
func (g *GrafanaLive) handleDatasourceScope(user *models.SignedInUser, namespace string) (models.ChannelHandlerFactory, error) { func (g *GrafanaLive) handleDatasourceScope(ctx context.Context, user *models.SignedInUser, namespace string) (models.ChannelHandlerFactory, error) {
ds, err := g.DataSourceCache.GetDatasourceByUID(context.TODO(), namespace, user, false) ds, err := g.DataSourceCache.GetDatasourceByUID(ctx, namespace, user, false)
if err != nil { if err != nil {
return nil, fmt.Errorf("error getting datasource: %w", err) return nil, fmt.Errorf("error getting datasource: %w", err)
} }
streamHandler, err := g.getStreamPlugin(ds.Type) streamHandler, err := g.getStreamPlugin(ctx, ds.Type)
if err != nil { if err != nil {
return nil, fmt.Errorf("can't find stream plugin: %s", ds.Type) return nil, fmt.Errorf("can't find stream plugin: %s", ds.Type)
} }
@ -996,7 +996,7 @@ func (g *GrafanaLive) HandleHTTPPublish(ctx *models.ReqContext) response.Respons
} }
} }
channelHandler, addr, err := g.GetChannelHandler(ctx.SignedInUser, cmd.Channel) channelHandler, addr, err := g.GetChannelHandler(ctx.Req.Context(), ctx.SignedInUser, cmd.Channel)
if err != nil { if err != nil {
logger.Error("Error getting channels handler", "error", err, "channel", cmd.Channel) logger.Error("Error getting channels handler", "error", err, "channel", cmd.Channel)
return response.Error(http.StatusInternalServerError, http.StatusText(http.StatusInternalServerError), nil) return response.Error(http.StatusInternalServerError, http.StatusText(http.StatusInternalServerError), nil)

5
pkg/services/live/managedstream/cache.go
View File

@ -1,6 +1,7 @@
package managedstream package managedstream
import ( import (
"context"
"encoding/json" "encoding/json"
"github.com/grafana/grafana-plugin-sdk-go/data" "github.com/grafana/grafana-plugin-sdk-go/data"
@ -11,7 +12,7 @@ type FrameCache interface {
// GetActiveChannels returns active managed stream channels with JSON schema. // GetActiveChannels returns active managed stream channels with JSON schema.
GetActiveChannels(orgID int64) (map[string]json.RawMessage, error) GetActiveChannels(orgID int64) (map[string]json.RawMessage, error)
// GetFrame returns full JSON frame for a channel in org. // GetFrame returns full JSON frame for a channel in org.
GetFrame(orgID int64, channel string) (json.RawMessage, bool, error) GetFrame(ctx context.Context, orgID int64, channel string) (json.RawMessage, bool, error)
// Update updates frame cache and returns true if schema changed. // Update updates frame cache and returns true if schema changed.
Update(orgID int64, channel string, frameJson data.FrameJSONCache) (bool, error) Update(ctx context.Context, orgID int64, channel string, frameJson data.FrameJSONCache) (bool, error)
} }

5
pkg/services/live/managedstream/cache_memory.go
View File

@ -1,6 +1,7 @@
package managedstream package managedstream
import ( import (
"context"
"encoding/json" "encoding/json"
"sync" "sync"
@ -34,14 +35,14 @@ func (c *MemoryFrameCache) GetActiveChannels(orgID int64) (map[string]json.RawMe
return info, nil return info, nil
} }
func (c *MemoryFrameCache) GetFrame(orgID int64, channel string) (json.RawMessage, bool, error) { func (c *MemoryFrameCache) GetFrame(ctx context.Context, orgID int64, channel string) (json.RawMessage, bool, error) {
c.mu.RLock() c.mu.RLock()
defer c.mu.RUnlock() defer c.mu.RUnlock()
cachedFrame, ok := c.frames[orgID][channel] cachedFrame, ok := c.frames[orgID][channel]
return cachedFrame.Bytes(data.IncludeAll), ok, nil return cachedFrame.Bytes(data.IncludeAll), ok, nil
} }
func (c *MemoryFrameCache) Update(orgID int64, channel string, jsonFrame data.FrameJSONCache) (bool, error) { func (c *MemoryFrameCache) Update(ctx context.Context, orgID int64, channel string, jsonFrame data.FrameJSONCache) (bool, error) {
c.mu.Lock() c.mu.Lock()
defer c.mu.Unlock() defer c.mu.Unlock()
if _, ok := c.frames[orgID]; !ok { if _, ok := c.frames[orgID]; !ok {

11
pkg/services/live/managedstream/cache_memory_test.go
View File

@ -1,6 +1,7 @@
package managedstream package managedstream
import ( import (
"context"
"encoding/json" "encoding/json"
"testing" "testing"
@ -15,7 +16,7 @@ func testFrameCache(t *testing.T, c FrameCache) {
frameJsonCache, err := data.FrameToJSONCache(frame) frameJsonCache, err := data.FrameToJSONCache(frame)
require.NoError(t, err) require.NoError(t, err)
updated, err := c.Update(1, "test", frameJsonCache) updated, err := c.Update(context.Background(), 1, "test", frameJsonCache)
require.NoError(t, err) require.NoError(t, err)
require.True(t, updated) require.True(t, updated)
@ -27,7 +28,7 @@ func testFrameCache(t *testing.T, c FrameCache) {
require.NotZero(t, schema) require.NotZero(t, schema)
// Make sure the same frame does not update schema. // Make sure the same frame does not update schema.
updated, err = c.Update(1, "test", frameJsonCache) updated, err = c.Update(context.Background(), 1, "test", frameJsonCache)
require.NoError(t, err) require.NoError(t, err)
require.False(t, updated) require.False(t, updated)
@ -37,17 +38,17 @@ func testFrameCache(t *testing.T, c FrameCache) {
require.NoError(t, err) require.NoError(t, err)
// Make sure schema updated. // Make sure schema updated.
updated, err = c.Update(1, "test", frameJsonCache) updated, err = c.Update(context.Background(), 1, "test", frameJsonCache)
require.NoError(t, err) require.NoError(t, err)
require.True(t, updated) require.True(t, updated)
// Add the same with another orgID and make sure schema updated. // Add the same with another orgID and make sure schema updated.
updated, err = c.Update(2, "test", frameJsonCache) updated, err = c.Update(context.Background(), 2, "test", frameJsonCache)
require.NoError(t, err) require.NoError(t, err)
require.True(t, updated) require.True(t, updated)
// Make sure that the last frame successfully saved in cache. // Make sure that the last frame successfully saved in cache.
frameJSON, ok, err := c.GetFrame(1, "test") frameJSON, ok, err := c.GetFrame(context.Background(), 1, "test")
require.NoError(t, err) require.NoError(t, err)
require.True(t, ok) require.True(t, ok)

8
pkg/services/live/managedstream/cache_redis.go
View File

@ -42,9 +42,9 @@ func (c *RedisFrameCache) GetActiveChannels(orgID int64) (map[string]json.RawMes
return info, nil return info, nil
} }
func (c *RedisFrameCache) GetFrame(orgID int64, channel string) (json.RawMessage, bool, error) { func (c *RedisFrameCache) GetFrame(ctx context.Context, orgID int64, channel string) (json.RawMessage, bool, error) {
key := getCacheKey(orgchannel.PrependOrgID(orgID, channel)) key := getCacheKey(orgchannel.PrependOrgID(orgID, channel))
cmd := c.redisClient.HGetAll(context.TODO(), key) cmd := c.redisClient.HGetAll(ctx, key)
result, err := cmd.Result() result, err := cmd.Result()
if err != nil { if err != nil {
return nil, false, err return nil, false, err
@ -59,7 +59,7 @@ const (
frameCacheTTL = 7 * 24 * time.Hour frameCacheTTL = 7 * 24 * time.Hour
) )
func (c *RedisFrameCache) Update(orgID int64, channel string, jsonFrame data.FrameJSONCache) (bool, error) { func (c *RedisFrameCache) Update(ctx context.Context, orgID int64, channel string, jsonFrame data.FrameJSONCache) (bool, error) {
c.mu.Lock() c.mu.Lock()
if _, ok := c.frames[orgID]; !ok { if _, ok := c.frames[orgID]; !ok {
c.frames[orgID] = map[string]data.FrameJSONCache{} c.frames[orgID] = map[string]data.FrameJSONCache{}
@ -74,8 +74,6 @@ func (c *RedisFrameCache) Update(orgID int64, channel string, jsonFrame data.Fra
pipe := c.redisClient.TxPipeline() pipe := c.redisClient.TxPipeline()
defer func() { _ = pipe.Close() }() defer func() { _ = pipe.Close() }()
ctx := context.TODO()
pipe.HGetAll(ctx, key) pipe.HGetAll(ctx, key)
pipe.HMSet(ctx, key, map[string]string{ pipe.HMSet(ctx, key, map[string]string{
"schema": stringSchema, "schema": stringSchema,

8
pkg/services/live/managedstream/runner.go
View File

@ -170,7 +170,7 @@ func NewNamespaceStream(orgID int64, scope string, namespace string, publisher m
// Push sends frame to the stream and saves it for later retrieval by subscribers. // Push sends frame to the stream and saves it for later retrieval by subscribers.
// * Saves the entire frame to cache. // * Saves the entire frame to cache.
// * If schema has been changed sends entire frame to channel, otherwise only data. // * If schema has been changed sends entire frame to channel, otherwise only data.
func (s *NamespaceStream) Push(path string, frame *data.Frame) error { func (s *NamespaceStream) Push(ctx context.Context, path string, frame *data.Frame) error {
jsonFrameCache, err := data.FrameToJSONCache(frame) jsonFrameCache, err := data.FrameToJSONCache(frame)
if err != nil { if err != nil {
return err return err
@ -179,7 +179,7 @@ func (s *NamespaceStream) Push(path string, frame *data.Frame) error {
// The channel this will be posted into. // The channel this will be posted into.
channel := live.Channel{Scope: s.scope, Namespace: s.namespace, Path: path}.String() channel := live.Channel{Scope: s.scope, Namespace: s.namespace, Path: path}.String()
isUpdated, err := s.frameCache.Update(s.orgID, channel, jsonFrameCache) isUpdated, err := s.frameCache.Update(ctx, s.orgID, channel, jsonFrameCache)
if err != nil { if err != nil {
logger.Error("Error updating managed stream schema", "error", err) logger.Error("Error updating managed stream schema", "error", err)
return err return err
@ -238,9 +238,9 @@ func (s *NamespaceStream) GetHandlerForPath(_ string) (models.ChannelHandler, er
return s, nil return s, nil
} }
func (s *NamespaceStream) OnSubscribe(_ context.Context, u *models.SignedInUser, e models.SubscribeEvent) (models.SubscribeReply, backend.SubscribeStreamStatus, error) { func (s *NamespaceStream) OnSubscribe(ctx context.Context, u *models.SignedInUser, e models.SubscribeEvent) (models.SubscribeReply, backend.SubscribeStreamStatus, error) {
reply := models.SubscribeReply{} reply := models.SubscribeReply{}
frameJSON, ok, err := s.frameCache.GetFrame(u.OrgId, e.Channel) frameJSON, ok, err := s.frameCache.GetFrame(ctx, u.OrgId, e.Channel)
if err != nil { if err != nil {
return reply, 0, err return reply, 0, err
} }

9
pkg/services/live/managedstream/runner_test.go
View File

@ -1,6 +1,7 @@
package managedstream package managedstream
import ( import (
"context"
"testing" "testing"
"time" "time"
@ -57,13 +58,13 @@ func TestGetManagedStreams(t *testing.T) {
require.NoError(t, err) require.NoError(t, err)
require.Len(t, managedChannels, 4) // 4 hardcoded testdata streams. require.Len(t, managedChannels, 4) // 4 hardcoded testdata streams.
err = s1.Push("cpu1", data.NewFrame("cpu1")) err = s1.Push(context.Background(), "cpu1", data.NewFrame("cpu1"))
require.NoError(t, err) require.NoError(t, err)
err = s1.Push("cpu2", data.NewFrame("cpu2")) err = s1.Push(context.Background(), "cpu2", data.NewFrame("cpu2"))
require.NoError(t, err) require.NoError(t, err)
err = s2.Push("cpu1", data.NewFrame("cpu1")) err = s2.Push(context.Background(), "cpu1", data.NewFrame("cpu1"))
require.NoError(t, err) require.NoError(t, err)
managedChannels, err = runner.GetManagedChannels(1) managedChannels, err = runner.GetManagedChannels(1)
@ -76,7 +77,7 @@ func TestGetManagedStreams(t *testing.T) {
// Different org. // Different org.
s3, err := runner.GetOrCreateStream(2, "stream", "test1") s3, err := runner.GetOrCreateStream(2, "stream", "test1")
require.NoError(t, err) require.NoError(t, err)
err = s3.Push("cpu1", data.NewFrame("cpu1")) err = s3.Push(context.Background(), "cpu1", data.NewFrame("cpu1"))
require.NoError(t, err) require.NoError(t, err)
managedChannels, err = runner.GetManagedChannels(1) managedChannels, err = runner.GetManagedChannels(1)
require.NoError(t, err) require.NoError(t, err)

2
pkg/services/live/pipeline/data_output_builtin.go
View File

@ -29,7 +29,7 @@ func (s *BuiltinDataOutput) OutputData(ctx context.Context, vars Vars, data []by
if !ok { if !ok {
return nil, errors.New("user not found in context") return nil, errors.New("user not found in context")
} }
handler, _, err := s.channelHandlerGetter.GetChannelHandler(u, vars.Channel) handler, _, err := s.channelHandlerGetter.GetChannelHandler(ctx, u, vars.Channel)
if err != nil { if err != nil {
return nil, err return nil, err
} }

4
pkg/services/live/pipeline/frame_output_managed_stream.go
View File

@ -22,11 +22,11 @@ func (out *ManagedStreamFrameOutput) Type() string {
return FrameOutputTypeManagedStream return FrameOutputTypeManagedStream
} }
func (out *ManagedStreamFrameOutput) OutputFrame(_ context.Context, vars Vars, frame *data.Frame) ([]*ChannelFrame, error) { func (out *ManagedStreamFrameOutput) OutputFrame(ctx context.Context, vars Vars, frame *data.Frame) ([]*ChannelFrame, error) {
stream, err := out.managedStream.GetOrCreateStream(vars.OrgID, vars.Scope, vars.Namespace) stream, err := out.managedStream.GetOrCreateStream(vars.OrgID, vars.Scope, vars.Namespace)
if err != nil { if err != nil {
logger.Error("Error getting stream", "error", err) logger.Error("Error getting stream", "error", err)
return nil, err return nil, err
} }
return nil, stream.Push(vars.Path, frame) return nil, stream.Push(ctx, vars.Path, frame)
} }

4
pkg/services/live/pipeline/subscribe_builtin.go
View File

@ -15,7 +15,7 @@ type BuiltinSubscriber struct {
} }
type ChannelHandlerGetter interface { type ChannelHandlerGetter interface {
GetChannelHandler(user *models.SignedInUser, channel string) (models.ChannelHandler, live.Channel, error) GetChannelHandler(ctx context.Context, user *models.SignedInUser, channel string) (models.ChannelHandler, live.Channel, error)
} }
const SubscriberTypeBuiltin = "builtin" const SubscriberTypeBuiltin = "builtin"
@ -33,7 +33,7 @@ func (s *BuiltinSubscriber) Subscribe(ctx context.Context, vars Vars, data []byt
if !ok { if !ok {
return models.SubscribeReply{}, backend.SubscribeStreamStatusPermissionDenied, nil return models.SubscribeReply{}, backend.SubscribeStreamStatusPermissionDenied, nil
} }
handler, _, err := s.channelHandlerGetter.GetChannelHandler(u, vars.Channel) handler, _, err := s.channelHandlerGetter.GetChannelHandler(ctx, u, vars.Channel)
if err != nil { if err != nil {
return models.SubscribeReply{}, 0, err return models.SubscribeReply{}, 0, err
} }

2
pkg/services/live/pushhttp/push.go
View File

@ -87,7 +87,7 @@ func (g *Gateway) Handle(ctx *models.ReqContext) {
// interval = "1s" vs flush_interval = "5s" // interval = "1s" vs flush_interval = "5s"
for _, mf := range metricFrames { for _, mf := range metricFrames {
err := stream.Push(mf.Key(), mf.Frame()) err := stream.Push(ctx.Req.Context(), mf.Key(), mf.Frame())
if err != nil { if err != nil {
logger.Error("Error pushing frame", "error", err, "data", string(body)) logger.Error("Error pushing frame", "error", err, "data", string(body))
ctx.Resp.WriteHeader(http.StatusInternalServerError) ctx.Resp.WriteHeader(http.StatusInternalServerError)

2
pkg/services/live/pushws/push_stream.go
View File

@ -91,7 +91,7 @@ func (s *Handler) ServeHTTP(rw http.ResponseWriter, r *http.Request) {
} }
for _, mf := range metricFrames { for _, mf := range metricFrames {
err := stream.Push(mf.Key(), mf.Frame()) err := stream.Push(r.Context(), mf.Key(), mf.Frame())
if err != nil { if err != nil {
logger.Error("Error pushing frame", "error", err, "data", string(body)) logger.Error("Error pushing frame", "error", err, "data", string(body))
return return

10
pkg/services/login/authinfoservice/database.go
View File

@ -21,7 +21,7 @@ func (s *Implementation) GetExternalUserInfoByLogin(ctx context.Context, query *
} }
authInfoQuery := &models.GetAuthInfoQuery{UserId: userQuery.Result.Id} authInfoQuery := &models.GetAuthInfoQuery{UserId: userQuery.Result.Id}
if err := s.Bus.DispatchCtx(context.TODO(), authInfoQuery); err != nil { if err := s.Bus.DispatchCtx(ctx, authInfoQuery); err != nil {
return err return err
} }
@ -47,7 +47,7 @@ func (s *Implementation) GetAuthInfo(ctx context.Context, query *models.GetAuthI
var has bool var has bool
var err error var err error
err = s.SQLStore.WithDbSession(context.Background(), func(sess *sqlstore.DBSession) error { err = s.SQLStore.WithDbSession(ctx, func(sess *sqlstore.DBSession) error {
has, err = sess.Desc("created").Get(userAuth) has, err = sess.Desc("created").Get(userAuth)
return err return err
}) })
@ -121,7 +121,7 @@ func (s *Implementation) SetAuthInfo(ctx context.Context, cmd *models.SetAuthInf
authUser.OAuthExpiry = cmd.OAuthToken.Expiry authUser.OAuthExpiry = cmd.OAuthToken.Expiry
} }
return s.SQLStore.WithTransactionalDbSession(context.Background(), func(sess *sqlstore.DBSession) error { return s.SQLStore.WithTransactionalDbSession(ctx, func(sess *sqlstore.DBSession) error {
_, err := sess.Insert(authUser) _, err := sess.Insert(authUser)
return err return err
}) })
@ -169,7 +169,7 @@ func (s *Implementation) UpdateAuthInfo(ctx context.Context, cmd *models.UpdateA
AuthModule: cmd.AuthModule, AuthModule: cmd.AuthModule,
} }
return s.SQLStore.WithTransactionalDbSession(context.Background(), func(sess *sqlstore.DBSession) error { return s.SQLStore.WithTransactionalDbSession(ctx, func(sess *sqlstore.DBSession) error {
upd, err := sess.Update(authUser, cond) upd, err := sess.Update(authUser, cond)
s.logger.Debug("Updated user_auth", "user_id", cmd.UserId, "auth_module", cmd.AuthModule, "rows", upd) s.logger.Debug("Updated user_auth", "user_id", cmd.UserId, "auth_module", cmd.AuthModule, "rows", upd)
return err return err
@ -177,7 +177,7 @@ func (s *Implementation) UpdateAuthInfo(ctx context.Context, cmd *models.UpdateA
} }
func (s *Implementation) DeleteAuthInfo(ctx context.Context, cmd *models.DeleteAuthInfoCommand) error { func (s *Implementation) DeleteAuthInfo(ctx context.Context, cmd *models.DeleteAuthInfoCommand) error {
return s.SQLStore.WithTransactionalDbSession(context.Background(), func(sess *sqlstore.DBSession) error { return s.SQLStore.WithTransactionalDbSession(ctx, func(sess *sqlstore.DBSession) error {
_, err := sess.Delete(cmd.UserAuth) _, err := sess.Delete(cmd.UserAuth)
return err return err
}) })

10
pkg/services/ngalert/notifier/alertmanager.go
View File

@ -127,7 +127,7 @@ type Alertmanager struct {
decryptFn channels.GetDecryptedValueFn decryptFn channels.GetDecryptedValueFn
} }
func newAlertmanager(orgID int64, cfg *setting.Cfg, store store.AlertingStore, kvStore kvstore.KVStore, func newAlertmanager(ctx context.Context, orgID int64, cfg *setting.Cfg, store store.AlertingStore, kvStore kvstore.KVStore,
peer ClusterPeer, decryptFn channels.GetDecryptedValueFn, m *metrics.Alertmanager) (*Alertmanager, error) { peer ClusterPeer, decryptFn channels.GetDecryptedValueFn, m *metrics.Alertmanager) (*Alertmanager, error) {
am := &Alertmanager{ am := &Alertmanager{
Settings: cfg, Settings: cfg,
@ -147,11 +147,11 @@ func newAlertmanager(orgID int64, cfg *setting.Cfg, store store.AlertingStore, k
am.gokitLogger = logging.NewWrapper(am.logger) am.gokitLogger = logging.NewWrapper(am.logger)
am.fileStore = NewFileStore(am.orgID, kvStore, am.WorkingDirPath()) am.fileStore = NewFileStore(am.orgID, kvStore, am.WorkingDirPath())
nflogFilepath, err := am.fileStore.FilepathFor(context.TODO(), notificationLogFilename) nflogFilepath, err := am.fileStore.FilepathFor(ctx, notificationLogFilename)
if err != nil { if err != nil {
return nil, err return nil, err
} }
silencesFilePath, err := am.fileStore.FilepathFor(context.TODO(), silencesFilename) silencesFilePath, err := am.fileStore.FilepathFor(ctx, silencesFilename)
if err != nil { if err != nil {
return nil, err return nil, err
} }
@ -162,7 +162,7 @@ func newAlertmanager(orgID int64, cfg *setting.Cfg, store store.AlertingStore, k
nflog.WithRetention(retentionNotificationsAndSilences), nflog.WithRetention(retentionNotificationsAndSilences),
nflog.WithSnapshot(nflogFilepath), nflog.WithSnapshot(nflogFilepath),
nflog.WithMaintenance(maintenanceNotificationAndSilences, am.stopc, am.wg.Done, func() (int64, error) { nflog.WithMaintenance(maintenanceNotificationAndSilences, am.stopc, am.wg.Done, func() (int64, error) {
return am.fileStore.Persist(context.TODO(), notificationLogFilename, am.notificationLog) return am.fileStore.Persist(ctx, notificationLogFilename, am.notificationLog)
}), }),
) )
if err != nil { if err != nil {
@ -187,7 +187,7 @@ func newAlertmanager(orgID int64, cfg *setting.Cfg, store store.AlertingStore, k
am.wg.Add(1) am.wg.Add(1)
go func() { go func() {
am.silences.Maintenance(15*time.Minute, silencesFilePath, am.stopc, func() (int64, error) { am.silences.Maintenance(15*time.Minute, silencesFilePath, am.stopc, func() (int64, error) {
return am.fileStore.Persist(context.TODO(), silencesFilename, am.silences) return am.fileStore.Persist(ctx, silencesFilename, am.silences)
}) })
am.wg.Done() am.wg.Done()
}() }()

2
pkg/services/ngalert/notifier/alertmanager_test.go
View File

@ -51,7 +51,7 @@ func setupAMTest(t *testing.T) *Alertmanager {
kvStore := NewFakeKVStore(t) kvStore := NewFakeKVStore(t)
secretsService := secretsManager.SetupTestService(t, database.ProvideSecretsStore(sqlStore)) secretsService := secretsManager.SetupTestService(t, database.ProvideSecretsStore(sqlStore))
decryptFn := secretsService.GetDecryptedValue decryptFn := secretsService.GetDecryptedValue
am, err := newAlertmanager(1, cfg, s, kvStore, &NilPeer{}, decryptFn, m) am, err := newAlertmanager(context.Background(), 1, cfg, s, kvStore, &NilPeer{}, decryptFn, m)
require.NoError(t, err) require.NoError(t, err)
return am return am
} }

2
pkg/services/ngalert/notifier/multiorg_alertmanager.go
View File

@ -171,7 +171,7 @@ func (moa *MultiOrgAlertmanager) SyncAlertmanagersForOrgs(ctx context.Context, o
// To export them, we need to translate the metrics from each individual registry and, // To export them, we need to translate the metrics from each individual registry and,
// then aggregate them on the main registry. // then aggregate them on the main registry.
m := metrics.NewAlertmanagerMetrics(moa.metrics.GetOrCreateOrgRegistry(orgID)) m := metrics.NewAlertmanagerMetrics(moa.metrics.GetOrCreateOrgRegistry(orgID))
am, err := newAlertmanager(orgID, moa.settings, moa.configStore, moa.kvStore, moa.peer, moa.decryptFn, m) am, err := newAlertmanager(ctx, orgID, moa.settings, moa.configStore, moa.kvStore, moa.peer, moa.decryptFn, m)
if err != nil { if err != nil {
moa.logger.Error("unable to create Alertmanager for org", "org", orgID, "err", err) moa.logger.Error("unable to create Alertmanager for org", "org", orgID, "err", err)
} }

9
pkg/services/ngalert/state/cache.go
View File

@ -1,6 +1,7 @@
package state package state
import ( import (
"context"
"fmt" "fmt"
"net/url" "net/url"
"strings" "strings"
@ -32,14 +33,14 @@ func newCache(logger log.Logger, metrics *metrics.State, externalURL *url.URL) *
} }
} }
func (c *cache) getOrCreate(alertRule *ngModels.AlertRule, result eval.Result) *State { func (c *cache) getOrCreate(ctx context.Context, alertRule *ngModels.AlertRule, result eval.Result) *State {
c.mtxStates.Lock() c.mtxStates.Lock()
defer c.mtxStates.Unlock() defer c.mtxStates.Unlock()
// clone the labels so we don't change eval.Result // clone the labels so we don't change eval.Result
labels := result.Instance.Copy() labels := result.Instance.Copy()
attachRuleLabels(labels, alertRule) attachRuleLabels(labels, alertRule)
ruleLabels, annotations := c.expandRuleLabelsAndAnnotations(alertRule, labels, result) ruleLabels, annotations := c.expandRuleLabelsAndAnnotations(ctx, alertRule, labels, result)
// if duplicate labels exist, alertRule label will take precedence // if duplicate labels exist, alertRule label will take precedence
lbs := mergeLabels(ruleLabels, result.Instance) lbs := mergeLabels(ruleLabels, result.Instance)
@ -88,11 +89,11 @@ func attachRuleLabels(m map[string]string, alertRule *ngModels.AlertRule) {
m[prometheusModel.AlertNameLabel] = alertRule.Title m[prometheusModel.AlertNameLabel] = alertRule.Title
} }
func (c *cache) expandRuleLabelsAndAnnotations(alertRule *ngModels.AlertRule, labels map[string]string, alertInstance eval.Result) (map[string]string, map[string]string) { func (c *cache) expandRuleLabelsAndAnnotations(ctx context.Context, alertRule *ngModels.AlertRule, labels map[string]string, alertInstance eval.Result) (map[string]string, map[string]string) {
expand := func(original map[string]string) map[string]string { expand := func(original map[string]string) map[string]string {
expanded := make(map[string]string, len(original)) expanded := make(map[string]string, len(original))
for k, v := range original { for k, v := range original {
ev, err := expandTemplate(alertRule.Title, v, labels, alertInstance, c.externalURL) ev, err := expandTemplate(ctx, alertRule.Title, v, labels, alertInstance, c.externalURL)
expanded[k] = ev expanded[k] = ev
if err != nil { if err != nil {
c.log.Error("error in expanding template", "name", k, "value", v, "err", err.Error()) c.log.Error("error in expanding template", "name", k, "value", v, "err", err.Error())

6
pkg/services/ngalert/state/manager.go
View File

@ -116,8 +116,8 @@ func (st *Manager) Warm() {
} }
} }
func (st *Manager) getOrCreate(alertRule *ngModels.AlertRule, result eval.Result) *State { func (st *Manager) getOrCreate(ctx context.Context, alertRule *ngModels.AlertRule, result eval.Result) *State {
return st.cache.getOrCreate(alertRule, result) return st.cache.getOrCreate(ctx, alertRule, result)
} }
func (st *Manager) set(entry *State) { func (st *Manager) set(entry *State) {
@ -153,7 +153,7 @@ func (st *Manager) ProcessEvalResults(ctx context.Context, alertRule *ngModels.A
// Set the current state based on evaluation results // Set the current state based on evaluation results
func (st *Manager) setNextState(ctx context.Context, alertRule *ngModels.AlertRule, result eval.Result) *State { func (st *Manager) setNextState(ctx context.Context, alertRule *ngModels.AlertRule, result eval.Result) *State {
currentState := st.getOrCreate(alertRule, result) currentState := st.getOrCreate(ctx, alertRule, result)
currentState.LastEvaluationTime = result.EvaluatedAt currentState.LastEvaluationTime = result.EvaluatedAt
currentState.EvaluationDuration = result.EvaluationDuration currentState.EvaluationDuration = result.EvaluationDuration

4
pkg/services/ngalert/state/template.go
View File

@ -31,7 +31,7 @@ func (v templateCaptureValue) String() string {
return strconv.FormatFloat(v.Value, 'f', -1, 64) return strconv.FormatFloat(v.Value, 'f', -1, 64)
} }
func expandTemplate(name, text string, labels map[string]string, alertInstance eval.Result, externalURL *url.URL) (result string, resultErr error) { func expandTemplate(ctx context.Context, name, text string, labels map[string]string, alertInstance eval.Result, externalURL *url.URL) (result string, resultErr error) {
name = "__alert_" + name name = "__alert_" + name
text = "{{- $labels := .Labels -}}{{- $values := .Values -}}{{- $value := .Value -}}" + text text = "{{- $labels := .Labels -}}{{- $values := .Values -}}{{- $value := .Value -}}" + text
data := struct { data := struct {
@ -45,7 +45,7 @@ func expandTemplate(name, text string, labels map[string]string, alertInstance e
} }
expander := template.NewTemplateExpander( expander := template.NewTemplateExpander(
context.TODO(), // This context is only used with the `query()` function - which we don't support yet. ctx, // This context is only used with the `query()` function - which we don't support yet.
text, text,
name, name,
data, data,

3
pkg/services/ngalert/state/template_test.go
View File

@ -1,6 +1,7 @@
package state package state
import ( import (
"context"
"errors" "errors"
"net/url" "net/url"
"testing" "testing"
@ -403,7 +404,7 @@ func TestExpandTemplate(t *testing.T) {
for _, c := range cases { for _, c := range cases {
t.Run(c.name, func(t *testing.T) { t.Run(c.name, func(t *testing.T) {
v, err := expandTemplate("test", c.text, c.labels, c.alertInstance, externalURL) v, err := expandTemplate(context.Background(), "test", c.text, c.labels, c.alertInstance, externalURL)
if c.expectedError != nil { if c.expectedError != nil {
require.NotNil(t, err) require.NotNil(t, err)
require.EqualError(t, c.expectedError, err.Error()) require.EqualError(t, c.expectedError, err.Error())