public/grafana
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Rendering service - add optional RenderingSession (#44098)

Browse Source 
* rendering service changes:
- make node-renderer request timeout configurable
- introduce optional RenderingSession providing a long-lived session key

* remove console logs

* added comment explaining empty "afterRequest" method

* fix compilation error

* update imports formatting

* Update pkg/services/rendering/interface.go

Co-authored-by: Agnès Toulet <35176601+AgnesToulet@users.noreply.github.com>

* Update pkg/services/rendering/rendering.go

Co-authored-by: Agnès Toulet <35176601+AgnesToulet@users.noreply.github.com>

* review fix: extract renderKey related functions/structs to auth.go

* #44449: private'd `rendering.getRequestTimeout`

Co-authored-by: Agnès Toulet <35176601+AgnesToulet@users.noreply.github.com>
This commit is contained in:
Artur Wierzbicki
2022-01-27 02:02:19 +04:00
committed by GitHub
co-authored by Agnès Toulet
parent 3c334cd8ba
commit 5148250366
10 changed files with 232 additions and 98 deletions
Download Patch File Download Diff File Expand all files Collapse all files
pkg/services/rendering/auth.go
+120
View File
@@ -0,0 +1,120 @@
package rendering
import (
"context"
"fmt"
"time"
"github.com/grafana/grafana/pkg/infra/log"
"github.com/grafana/grafana/pkg/infra/remotecache"
"github.com/grafana/grafana/pkg/util"
)
const renderKeyPrefix = "render-%s"
type RenderUser struct {
OrgID int64
UserID int64
OrgRole string
}
func (rs *RenderingService) GetRenderUser(ctx context.Context, key string) (*RenderUser, bool) {
val, err := rs.RemoteCacheService.Get(ctx, fmt.Sprintf(renderKeyPrefix, key))
if err != nil {
rs.log.Error("Failed to get render key from cache", "error", err)
}
if val != nil {
if user, ok := val.(*RenderUser); ok {
return user, true
}
}
return nil, false
}
func setRenderKey(cache *remotecache.RemoteCache, ctx context.Context, opts AuthOpts, renderKey string, expiry time.Duration) error {
err := cache.Set(ctx, fmt.Sprintf(renderKeyPrefix, renderKey), &RenderUser{
OrgID: opts.OrgID,
UserID: opts.UserID,
OrgRole: string(opts.OrgRole),
}, expiry)
return err
}
func generateAndSetRenderKey(cache *remotecache.RemoteCache, ctx context.Context, opts AuthOpts, expiry time.Duration) (string, error) {
key, err := util.GetRandomString(32)
if err != nil {
return "", err
}
err = setRenderKey(cache, ctx, opts, key, expiry)
if err != nil {
return "", err
}
return key, nil
}
type longLivedRenderKeyProvider struct {
cache *remotecache.RemoteCache
log log.Logger
renderKey string
authOpts AuthOpts
sessionOpts SessionOpts
}
func (rs *RenderingService) CreateRenderingSession(ctx context.Context, opts AuthOpts, sessionOpts SessionOpts) (Session, error) {
renderKey, err := generateAndSetRenderKey(rs.RemoteCacheService, ctx, opts, sessionOpts.Expiry)
if err != nil {
return nil, err
}
return &longLivedRenderKeyProvider{
log: rs.log,
renderKey: renderKey,
cache: rs.RemoteCacheService,
authOpts: opts,
sessionOpts: sessionOpts,
}, nil
}
func deleteRenderKey(cache *remotecache.RemoteCache, log log.Logger, ctx context.Context, renderKey string) {
err := cache.Delete(ctx, fmt.Sprintf(renderKeyPrefix, renderKey))
if err != nil {
log.Error("Failed to delete render key", "error", err)
}
}
type perRequestRenderKeyProvider struct {
cache *remotecache.RemoteCache
log log.Logger
keyExpiry time.Duration
}
func (r *perRequestRenderKeyProvider) get(ctx context.Context, opts AuthOpts) (string, error) {
return generateAndSetRenderKey(r.cache, ctx, opts, r.keyExpiry)
}
func (r *perRequestRenderKeyProvider) afterRequest(ctx context.Context, opts AuthOpts, renderKey string) {
deleteRenderKey(r.cache, r.log, ctx, renderKey)
}
func (r *longLivedRenderKeyProvider) get(ctx context.Context, opts AuthOpts) (string, error) {
if r.sessionOpts.RefreshExpiryOnEachRequest {
err := setRenderKey(r.cache, ctx, opts, r.renderKey, r.sessionOpts.Expiry)
if err != nil {
r.log.Error("Failed to refresh render key", "error", err, "renderKey", r.renderKey)
}
}
return r.renderKey, nil
}
func (r *longLivedRenderKeyProvider) afterRequest(ctx context.Context, opts AuthOpts, renderKey string) {
// do nothing - renderKey from longLivedRenderKeyProvider is deleted only after session expires
// or someone calls session.Dispose()
}
func (r *longLivedRenderKeyProvider) Dispose(ctx context.Context) {
deleteRenderKey(r.cache, r.log, ctx, r.renderKey)
}
pkg/services/rendering/http_mode.go
+2 -2
View File
@@ -58,7 +58,7 @@ func (rs *RenderingService) renderViaHTTP(ctx context.Context, renderKey string,
rendererURL.RawQuery = queryParams.Encode()
// gives service some additional time to timeout and return possible errors.
reqContext, cancel := context.WithTimeout(ctx, opts.Timeout+time.Second*2)
reqContext, cancel := context.WithTimeout(ctx, getRequestTimeout(opts.TimeoutOpts))
defer cancel()
resp, err := rs.doRequest(reqContext, rendererURL, opts.Headers)
@@ -103,7 +103,7 @@ func (rs *RenderingService) renderCSVViaHTTP(ctx context.Context, renderKey stri
rendererURL.RawQuery = queryParams.Encode()
// gives service some additional time to timeout and return possible errors.
reqContext, cancel := context.WithTimeout(ctx, opts.Timeout+time.Second*2)
reqContext, cancel := context.WithTimeout(ctx, getRequestTimeout(opts.TimeoutOpts))
defer cancel()
resp, err := rs.doRequest(reqContext, rendererURL, opts.Headers)
pkg/services/rendering/interface.go
+41 -10
View File
@@ -26,13 +26,30 @@ const (
ThemeDark Theme = "dark"
)
type TimeoutOpts struct {
Timeout time.Duration // Timeout param passed to image-renderer service
RequestTimeoutMultiplier time.Duration // RequestTimeoutMultiplier used for plugin/HTTP request context timeout
}
type AuthOpts struct {
OrgID int64
UserID int64
OrgRole models.RoleType
}
func getRequestTimeout(opt TimeoutOpts) time.Duration {
if opt.RequestTimeoutMultiplier == 0 {
return opt.Timeout * 2 // default
}
return opt.Timeout * opt.RequestTimeoutMultiplier
}
type Opts struct {
TimeoutOpts
AuthOpts
Width int
Height int
Timeout time.Duration
OrgID int64
UserID int64
OrgRole models.RoleType
Path string
Encoding string
Timezone string
@@ -43,10 +60,8 @@ type Opts struct {
}
type CSVOpts struct {
Timeout time.Duration
OrgID int64
UserID int64
OrgRole models.RoleType
TimeoutOpts
AuthOpts
Path string
Encoding string
Timezone string
@@ -66,11 +81,27 @@ type RenderCSVResult struct {
type renderFunc func(ctx context.Context, renderKey string, options Opts) (*RenderResult, error)
type renderCSVFunc func(ctx context.Context, renderKey string, options CSVOpts) (*RenderCSVResult, error)
type renderKeyProvider interface {
get(ctx context.Context, opts AuthOpts) (string, error)
afterRequest(ctx context.Context, opts AuthOpts, renderKey string)
}
type SessionOpts struct {
Expiry time.Duration
RefreshExpiryOnEachRequest bool
}
type Session interface {
renderKeyProvider
Dispose(ctx context.Context)
}
type Service interface {
IsAvailable() bool
Version() string
Render(ctx context.Context, opts Opts) (*RenderResult, error)
RenderCSV(ctx context.Context, opts CSVOpts) (*RenderCSVResult, error)
Render(ctx context.Context, opts Opts, session Session) (*RenderResult, error)
RenderCSV(ctx context.Context, opts CSVOpts, session Session) (*RenderCSVResult, error)
RenderErrorImage(theme Theme, error error) (*RenderResult, error)
GetRenderUser(ctx context.Context, key string) (*RenderUser, bool)
CreateRenderingSession(ctx context.Context, authOpts AuthOpts, sessionOpts SessionOpts) (Session, error)
}
pkg/services/rendering/plugin_mode.go
+2 -3
View File
@@ -4,7 +4,6 @@ import (
"context"
"errors"
"fmt"
"time"
"github.com/grafana/grafana/pkg/plugins/backendplugin/pluginextensionv2"
)
@@ -15,7 +14,7 @@ func (rs *RenderingService) startPlugin(ctx context.Context) error {
func (rs *RenderingService) renderViaPlugin(ctx context.Context, renderKey string, opts Opts) (*RenderResult, error) {
// gives plugin some additional time to timeout and return possible errors.
ctx, cancel := context.WithTimeout(ctx, opts.Timeout+time.Second*2)
ctx, cancel := context.WithTimeout(ctx, getRequestTimeout(opts.TimeoutOpts))
defer cancel()
filePath, err := rs.getNewFilePath(RenderPNG)
@@ -62,7 +61,7 @@ func (rs *RenderingService) renderViaPlugin(ctx context.Context, renderKey strin
func (rs *RenderingService) renderCSVViaPlugin(ctx context.Context, renderKey string, opts CSVOpts) (*RenderCSVResult, error) {
// gives plugin some additional time to timeout and return possible errors.
ctx, cancel := context.WithTimeout(ctx, opts.Timeout+time.Second*2)
ctx, cancel := context.WithTimeout(ctx, getRequestTimeout(opts.TimeoutOpts))
defer cancel()
filePath, err := rs.getNewFilePath(RenderCSV)
pkg/services/rendering/rendering.go
+31 -62
View File
@@ -17,7 +17,6 @@ import (
"github.com/grafana/grafana/pkg/infra/log"
"github.com/grafana/grafana/pkg/infra/metrics"
"github.com/grafana/grafana/pkg/infra/remotecache"
"github.com/grafana/grafana/pkg/models"
"github.com/grafana/grafana/pkg/plugins"
"github.com/grafana/grafana/pkg/setting"
"github.com/grafana/grafana/pkg/util"
@@ -28,13 +27,6 @@ func init() {
}
const ServiceName = "RenderingService"
const renderKeyPrefix = "render-%s"
type RenderUser struct {
OrgID int64
UserID int64
OrgRole string
}
type RenderingService struct {
log log.Logger
@@ -46,9 +38,10 @@ type RenderingService struct {
version string
versionMutex sync.RWMutex
Cfg *setting.Cfg
RemoteCacheService *remotecache.RemoteCache
RendererPluginManager plugins.RendererManager
perRequestRenderKeyProvider renderKeyProvider
Cfg *setting.Cfg
RemoteCacheService *remotecache.RemoteCache
RendererPluginManager plugins.RendererManager
}
func ProvideService(cfg *setting.Cfg, remoteCache *remotecache.RemoteCache, rm plugins.RendererManager) (*RenderingService, error) {
@@ -81,11 +74,17 @@ func ProvideService(cfg *setting.Cfg, remoteCache *remotecache.RemoteCache, rm p
domain = "localhost"
}
logger := log.New("rendering")
s := &RenderingService{
perRequestRenderKeyProvider: &perRequestRenderKeyProvider{
cache: remoteCache,
log: logger,
keyExpiry: 5 * time.Minute,
},
Cfg: cfg,
RemoteCacheService: remoteCache,
RendererPluginManager: rm,
log: log.New("rendering"),
log: logger,
domain: domain,
}
return s, nil
@@ -195,9 +194,14 @@ func (rs *RenderingService) renderUnavailableImage() *RenderResult {
}
}
func (rs *RenderingService) Render(ctx context.Context, opts Opts) (*RenderResult, error) {
func (rs *RenderingService) Render(ctx context.Context, opts Opts, session Session) (*RenderResult, error) {
startTime := time.Now()
result, err := rs.render(ctx, opts)
renderKeyProvider := rs.perRequestRenderKeyProvider
if session != nil {
renderKeyProvider = session
}
result, err := rs.render(ctx, opts, renderKeyProvider)
elapsedTime := time.Since(startTime).Milliseconds()
saveMetrics(elapsedTime, err, RenderPNG)
@@ -205,7 +209,7 @@ func (rs *RenderingService) Render(ctx context.Context, opts Opts) (*RenderResul
return result, err
}
func (rs *RenderingService) render(ctx context.Context, opts Opts) (*RenderResult, error) {
func (rs *RenderingService) render(ctx context.Context, opts Opts, renderKeyProvider renderKeyProvider) (*RenderResult, error) {
if int(atomic.LoadInt32(&rs.inProgressCount)) > opts.ConcurrentLimit {
rs.log.Warn("Could not render image, hit the currency limit", "concurrencyLimit", opts.ConcurrentLimit, "path", opts.Path)
@@ -230,12 +234,12 @@ func (rs *RenderingService) render(ctx context.Context, opts Opts) (*RenderResul
if math.IsInf(opts.DeviceScaleFactor, 0) || math.IsNaN(opts.DeviceScaleFactor) || opts.DeviceScaleFactor == 0 {
opts.DeviceScaleFactor = 1
}
renderKey, err := rs.generateAndStoreRenderKey(ctx, opts.OrgID, opts.UserID, opts.OrgRole)
renderKey, err := renderKeyProvider.get(ctx, opts.AuthOpts)
if err != nil {
return nil, err
}
defer rs.deleteRenderKey(ctx, renderKey)
defer renderKeyProvider.afterRequest(ctx, opts.AuthOpts, renderKey)
defer func() {
metrics.MRenderingQueue.Set(float64(atomic.AddInt32(&rs.inProgressCount, -1)))
@@ -245,9 +249,14 @@ func (rs *RenderingService) render(ctx context.Context, opts Opts) (*RenderResul
return rs.renderAction(ctx, renderKey, opts)
}
func (rs *RenderingService) RenderCSV(ctx context.Context, opts CSVOpts) (*RenderCSVResult, error) {
func (rs *RenderingService) RenderCSV(ctx context.Context, opts CSVOpts, session Session) (*RenderCSVResult, error) {
startTime := time.Now()
result, err := rs.renderCSV(ctx, opts)
renderKeyProvider := rs.perRequestRenderKeyProvider
if session != nil {
renderKeyProvider = session
}
result, err := rs.renderCSV(ctx, opts, renderKeyProvider)
elapsedTime := time.Since(startTime).Milliseconds()
saveMetrics(elapsedTime, err, RenderCSV)
@@ -255,7 +264,7 @@ func (rs *RenderingService) RenderCSV(ctx context.Context, opts CSVOpts) (*Rende
return result, err
}
func (rs *RenderingService) renderCSV(ctx context.Context, opts CSVOpts) (*RenderCSVResult, error) {
func (rs *RenderingService) renderCSV(ctx context.Context, opts CSVOpts, renderKeyProvider renderKeyProvider) (*RenderCSVResult, error) {
if int(atomic.LoadInt32(&rs.inProgressCount)) > opts.ConcurrentLimit {
return nil, ErrConcurrentLimitReached
}
@@ -265,12 +274,12 @@ func (rs *RenderingService) renderCSV(ctx context.Context, opts CSVOpts) (*Rende
}
rs.log.Info("Rendering", "path", opts.Path)
renderKey, err := rs.generateAndStoreRenderKey(ctx, opts.OrgID, opts.UserID, opts.OrgRole)
renderKey, err := renderKeyProvider.get(ctx, opts.AuthOpts)
if err != nil {
return nil, err
}
defer rs.deleteRenderKey(ctx, renderKey)
defer renderKeyProvider.afterRequest(ctx, opts.AuthOpts, renderKey)
defer func() {
metrics.MRenderingQueue.Set(float64(atomic.AddInt32(&rs.inProgressCount, -1)))
@@ -280,21 +289,6 @@ func (rs *RenderingService) renderCSV(ctx context.Context, opts CSVOpts) (*Rende
return rs.renderCSVAction(ctx, renderKey, opts)
}
func (rs *RenderingService) GetRenderUser(ctx context.Context, key string) (*RenderUser, bool) {
val, err := rs.RemoteCacheService.Get(ctx, fmt.Sprintf(renderKeyPrefix, key))
if err != nil {
rs.log.Error("Failed to get render key from cache", "error", err)
}
if val != nil {
if user, ok := val.(*RenderUser); ok {
return user, true
}
}
return nil, false
}
func (rs *RenderingService) getNewFilePath(rt RenderType) (string, error) {
rand, err := util.GetRandomString(20)
if err != nil {
@@ -340,31 +334,6 @@ func (rs *RenderingService) getURL(path string) string {
return fmt.Sprintf("%s://%s:%s%s/%s&render=1", protocol, rs.domain, rs.Cfg.HTTPPort, subPath, path)
}
func (rs *RenderingService) generateAndStoreRenderKey(ctx context.Context, orgId, userId int64, orgRole models.RoleType) (string, error) {
key, err := util.GetRandomString(32)
if err != nil {
return "", err
}
err = rs.RemoteCacheService.Set(ctx, fmt.Sprintf(renderKeyPrefix, key), &RenderUser{
OrgID: orgId,
UserID: userId,
OrgRole: string(orgRole),
}, 5*time.Minute)
if err != nil {
return "", err
}
return key, nil
}
func (rs *RenderingService) deleteRenderKey(ctx context.Context, key string) {
err := rs.RemoteCacheService.Delete(ctx, fmt.Sprintf(renderKeyPrefix, key))
if err != nil {
rs.log.Error("Failed to delete render key", "error", err)
}
}
func isoTimeOffsetToPosixTz(isoOffset string) string {
// invert offset
if strings.HasPrefix(isoOffset, "UTC+") {
pkg/services/rendering/rendering_test.go
+1 -1
View File
@@ -138,7 +138,7 @@ func TestRenderLimitImage(t *testing.T) {
for _, tc := range tests {
t.Run(tc.name, func(t *testing.T) {
opts := Opts{Theme: tc.theme, ConcurrentLimit: 1}
result, err := rs.Render(context.Background(), opts)
result, err := rs.Render(context.Background(), opts, nil)
assert.NoError(t, err)
assert.Equal(t, tc.expected, result.FilePath)
})