From ed1176adc8f47a99984cc6d514d59034511c8602 Mon Sep 17 00:00:00 2001 From: Ryan McKinley Date: Wed, 19 Oct 2022 17:05:51 -0700 Subject: [PATCH] Storage: export dashboards + playlists to object store (#57313) --- pkg/cmd/grafana-cli/runner/wire.go | 2 + pkg/services/export/dummy_job.go | 2 +- pkg/services/export/object_store.go | 212 ++++++++++++++++++ pkg/services/export/service.go | 15 +- pkg/services/store/object/dummy/fake_store.go | 38 ++++ public/app/features/storage/ExportView.tsx | 72 +++--- 6 files changed, 306 insertions(+), 35 deletions(-) create mode 100644 pkg/services/export/object_store.go create mode 100644 pkg/services/store/object/dummy/fake_store.go diff --git a/pkg/cmd/grafana-cli/runner/wire.go b/pkg/cmd/grafana-cli/runner/wire.go index 20eb2ce2a09..f3f564874c4 100644 --- a/pkg/cmd/grafana-cli/runner/wire.go +++ b/pkg/cmd/grafana-cli/runner/wire.go @@ -115,6 +115,7 @@ import ( "github.com/grafana/grafana/pkg/services/sqlstore" "github.com/grafana/grafana/pkg/services/star/starimpl" "github.com/grafana/grafana/pkg/services/store" + objectdummyserver "github.com/grafana/grafana/pkg/services/store/object/dummy" "github.com/grafana/grafana/pkg/services/store/sanitizer" "github.com/grafana/grafana/pkg/services/tag" "github.com/grafana/grafana/pkg/services/tag/tagimpl" @@ -324,6 +325,7 @@ var wireSet = wire.NewSet( userauthimpl.ProvideService, ngmetrics.ProvideServiceForTest, notifications.MockNotificationService, + objectdummyserver.ProvideFakeObjectServer, wire.Bind(new(notifications.TempUserStore), new(*dbtest.FakeDB)), wire.Bind(new(notifications.Service), new(*notifications.NotificationServiceMock)), wire.Bind(new(notifications.WebhookSender), new(*notifications.NotificationServiceMock)), diff --git a/pkg/services/export/dummy_job.go b/pkg/services/export/dummy_job.go index eeb3a98865d..5671c742bcb 100644 --- a/pkg/services/export/dummy_job.go +++ b/pkg/services/export/dummy_job.go @@ -30,7 +30,7 @@ func startDummyExportJob(cfg ExportConfig, broadcaster statusBroadcaster) (Job, broadcaster: broadcaster, status: ExportStatus{ Running: true, - Target: "git export", + Target: "dummy export", Started: time.Now().UnixMilli(), Count: make(map[string]int, 10), Index: 0, diff --git a/pkg/services/export/object_store.go b/pkg/services/export/object_store.go new file mode 100644 index 00000000000..293f36f0c81 --- /dev/null +++ b/pkg/services/export/object_store.go @@ -0,0 +1,212 @@ +package export + +import ( + "context" + "fmt" + "sync" + "time" + + "github.com/grafana/grafana/pkg/infra/db" + "github.com/grafana/grafana/pkg/infra/log" + "github.com/grafana/grafana/pkg/models" + "github.com/grafana/grafana/pkg/services/playlist" + "github.com/grafana/grafana/pkg/services/sqlstore/session" + "github.com/grafana/grafana/pkg/services/store" + "github.com/grafana/grafana/pkg/services/store/object" + "github.com/grafana/grafana/pkg/services/user" +) + +var _ Job = new(objectStoreJob) + +type objectStoreJob struct { + logger log.Logger + + statusMu sync.Mutex + status ExportStatus + cfg ExportConfig + broadcaster statusBroadcaster + stopRequested bool + + sess *session.SessionDB + playlistService playlist.Service + store object.ObjectStoreServer +} + +func startObjectStoreJob(cfg ExportConfig, broadcaster statusBroadcaster, db db.DB, playlistService playlist.Service, store object.ObjectStoreServer) (Job, error) { + job := &objectStoreJob{ + logger: log.New("export_to_object_store_job"), + cfg: cfg, + broadcaster: broadcaster, + status: ExportStatus{ + Running: true, + Target: "object store export", + Started: time.Now().UnixMilli(), + Count: make(map[string]int, 10), + Index: 0, + }, + sess: db.GetSqlxSession(), + playlistService: playlistService, + store: store, + } + + broadcaster(job.status) + go job.start() + return job, nil +} + +func (e *objectStoreJob) requestStop() { + e.stopRequested = true +} + +func (e *objectStoreJob) start() { + defer func() { + e.logger.Info("Finished dummy export job") + + e.statusMu.Lock() + defer e.statusMu.Unlock() + s := e.status + if err := recover(); err != nil { + e.logger.Error("export panic", "error", err) + s.Status = fmt.Sprintf("ERROR: %v", err) + } + // Make sure it finishes OK + if s.Finished < 10 { + s.Finished = time.Now().UnixMilli() + } + s.Running = false + if s.Status == "" { + s.Status = "done" + } + e.status = s + e.broadcaster(s) + }() + + e.logger.Info("Starting dummy export job") + // Select all dashboards + rowUser := &user.SignedInUser{ + Login: "?", + OrgID: 0, // gets filled in from each row + UserID: 0, + } + ctx := store.ContextWithUser(context.Background(), rowUser) + + what := models.StandardKindDashboard + e.status.Count[what] = 0 + + // TODO paging etc + // NOTE: doing work inside rows.Next() leads to database locked + dashInfo, err := e.getDashboards(ctx) + if err != nil { + e.status.Status = "error: " + err.Error() + return + } + + for _, dash := range dashInfo { + rowUser.OrgID = dash.OrgID + rowUser.UserID = dash.UpdatedBy + if dash.UpdatedBy < 0 { + rowUser.UserID = 0 // avoid Uint64Val issue???? + } + + _, err = e.store.Write(ctx, &object.WriteObjectRequest{ + UID: fmt.Sprintf("export/%s", dash.UID), + Kind: models.StandardKindDashboard, + Body: dash.Body, + Comment: "export from dashboard table", + }) + if err != nil { + e.status.Status = "error: " + err.Error() + return + } + e.status.Changed = time.Now().UnixMilli() + e.status.Index++ + e.status.Count[what] += 1 + e.status.Last = fmt.Sprintf("ITEM: %s", dash.UID) + e.broadcaster(e.status) + } + + // Playlists + what = models.StandardKindPlaylist + e.status.Count[what] = 0 + rowUser.OrgID = 1 + rowUser.UserID = 1 + res, err := e.playlistService.Search(ctx, &playlist.GetPlaylistsQuery{ + OrgId: rowUser.OrgID, // TODO... all or orgs + Limit: 5000, + }) + if err != nil { + e.status.Status = "error: " + err.Error() + return + } + for _, item := range res { + playlist, err := e.playlistService.Get(ctx, &playlist.GetPlaylistByUidQuery{ + UID: item.UID, + OrgId: rowUser.OrgID, + }) + if err != nil { + e.status.Status = "error: " + err.Error() + return + } + + _, err = e.store.Write(ctx, &object.WriteObjectRequest{ + UID: fmt.Sprintf("export/%s", playlist.Uid), + Kind: models.StandardKindPlaylist, + Body: prettyJSON(playlist), + Comment: "export from playlists", + }) + if err != nil { + e.status.Status = "error: " + err.Error() + return + } + e.status.Changed = time.Now().UnixMilli() + e.status.Index++ + e.status.Count[what] += 1 + e.status.Last = fmt.Sprintf("ITEM: %s", playlist.Uid) + e.broadcaster(e.status) + } +} + +type dashInfo struct { + OrgID int64 + UID string + Body []byte + UpdatedBy int64 +} + +func (e *objectStoreJob) getDashboards(ctx context.Context) ([]dashInfo, error) { + e.status.Last = "find dashbaords...." + e.broadcaster(e.status) + + dash := make([]dashInfo, 0) + rows, err := e.sess.Query(ctx, "SELECT org_id,uid,data,updated_by FROM dashboard WHERE is_folder=0") + if err != nil { + return nil, err + } + for rows.Next() { + if e.stopRequested { + return dash, nil + } + + row := dashInfo{} + err = rows.Scan(&row.OrgID, &row.UID, &row.Body, &row.UpdatedBy) + if err != nil { + return nil, err + } + dash = append(dash, row) + } + return dash, nil +} + +func (e *objectStoreJob) getStatus() ExportStatus { + e.statusMu.Lock() + defer e.statusMu.Unlock() + + return e.status +} + +func (e *objectStoreJob) getConfig() ExportConfig { + e.statusMu.Lock() + defer e.statusMu.Unlock() + + return e.cfg +} diff --git a/pkg/services/export/service.go b/pkg/services/export/service.go index caf85002421..c0112762d4c 100644 --- a/pkg/services/export/service.go +++ b/pkg/services/export/service.go @@ -19,6 +19,7 @@ import ( "github.com/grafana/grafana/pkg/services/live" "github.com/grafana/grafana/pkg/services/org" "github.com/grafana/grafana/pkg/services/playlist" + "github.com/grafana/grafana/pkg/services/store/object" "github.com/grafana/grafana/pkg/setting" ) @@ -151,25 +152,25 @@ type StandardExport struct { dataDir string // Services - sql db.DB + db db.DB dashboardsnapshotsService dashboardsnapshots.Service playlistService playlist.Service orgService org.Service datasourceService datasources.DataSourceService + store object.ObjectStoreServer // updated with mutex exportJob Job } -func ProvideService(sql db.DB, features featuremgmt.FeatureToggles, gl *live.GrafanaLive, cfg *setting.Cfg, +func ProvideService(db db.DB, features featuremgmt.FeatureToggles, gl *live.GrafanaLive, cfg *setting.Cfg, dashboardsnapshotsService dashboardsnapshots.Service, playlistService playlist.Service, orgService org.Service, - datasourceService datasources.DataSourceService) ExportService { + datasourceService datasources.DataSourceService, store object.ObjectStoreServer) ExportService { if !features.IsEnabled(featuremgmt.FlagExport) { return &StubExport{} } return &StandardExport{ - sql: sql, glive: gl, logger: log.New("export_service"), dashboardsnapshotsService: dashboardsnapshotsService, @@ -178,6 +179,8 @@ func ProvideService(sql db.DB, features featuremgmt.FeatureToggles, gl *live.Gra datasourceService: datasourceService, exportJob: &stoppedJob{}, dataDir: cfg.DataPath, + store: store, + db: db, } } @@ -227,12 +230,14 @@ func (ex *StandardExport) HandleRequestExport(c *models.ReqContext) response.Res switch cfg.Format { case "dummy": job, err = startDummyExportJob(cfg, broadcast) + case "objectStore": + job, err = startObjectStoreJob(cfg, broadcast, ex.db, ex.playlistService, ex.store) case "git": dir := filepath.Join(ex.dataDir, "export_git", fmt.Sprintf("git_%d", time.Now().Unix())) if err := os.MkdirAll(dir, os.ModePerm); err != nil { return response.Error(http.StatusBadRequest, "Error creating export folder", nil) } - job, err = startGitExportJob(cfg, ex.sql, ex.dashboardsnapshotsService, dir, c.OrgID, broadcast, ex.playlistService, ex.orgService, ex.datasourceService) + job, err = startGitExportJob(cfg, ex.db, ex.dashboardsnapshotsService, dir, c.OrgID, broadcast, ex.playlistService, ex.orgService, ex.datasourceService) default: return response.Error(http.StatusBadRequest, "Unsupported job format", nil) } diff --git a/pkg/services/store/object/dummy/fake_store.go b/pkg/services/store/object/dummy/fake_store.go new file mode 100644 index 00000000000..092f4ce9f43 --- /dev/null +++ b/pkg/services/store/object/dummy/fake_store.go @@ -0,0 +1,38 @@ +package objectdummyserver + +import ( + "context" + "fmt" + + "github.com/grafana/grafana/pkg/services/store/object" +) + +func ProvideFakeObjectServer() object.ObjectStoreServer { + return &fakeObjectStore{} +} + +type fakeObjectStore struct{} + +func (i fakeObjectStore) Write(ctx context.Context, r *object.WriteObjectRequest) (*object.WriteObjectResponse, error) { + return nil, fmt.Errorf("unimplemented") +} + +func (i fakeObjectStore) Read(ctx context.Context, r *object.ReadObjectRequest) (*object.ReadObjectResponse, error) { + return nil, fmt.Errorf("unimplemented") +} + +func (i fakeObjectStore) BatchRead(ctx context.Context, batchR *object.BatchReadObjectRequest) (*object.BatchReadObjectResponse, error) { + return nil, fmt.Errorf("unimplemented") +} + +func (i fakeObjectStore) Delete(ctx context.Context, r *object.DeleteObjectRequest) (*object.DeleteObjectResponse, error) { + return nil, fmt.Errorf("unimplemented") +} + +func (i fakeObjectStore) History(ctx context.Context, r *object.ObjectHistoryRequest) (*object.ObjectHistoryResponse, error) { + return nil, fmt.Errorf("unimplemented") +} + +func (i fakeObjectStore) Search(ctx context.Context, r *object.ObjectSearchRequest) (*object.ObjectSearchResponse, error) { + return nil, fmt.Errorf("unimplemented") +} diff --git a/public/app/features/storage/ExportView.tsx b/public/app/features/storage/ExportView.tsx index beefbc99f6f..aa9ee5180fb 100644 --- a/public/app/features/storage/ExportView.tsx +++ b/public/app/features/storage/ExportView.tsx @@ -2,7 +2,7 @@ import React, { useEffect, useState, useCallback } from 'react'; import { useAsync, useLocalStorage } from 'react-use'; import { isLiveChannelMessageEvent, isLiveChannelStatusEvent, LiveChannelScope, SelectableValue } from '@grafana/data'; -import { getBackendSrv, getGrafanaLiveSrv } from '@grafana/runtime'; +import { getBackendSrv, getGrafanaLiveSrv, config } from '@grafana/runtime'; import { Button, CodeEditor, @@ -16,6 +16,7 @@ import { LinkButton, Select, Switch, + Alert, } from '@grafana/ui'; import { StorageView } from './types'; @@ -60,6 +61,7 @@ interface ExporterInfo { const formats: Array> = [ { label: 'GIT', value: 'git', description: 'Exports a fresh git repository' }, + { label: 'Object store', value: 'objectStore', description: 'Export to the SQL based object store' }, ]; interface Props { @@ -171,36 +173,48 @@ export const ExportView = ({ onPathChange }: Props) => { onChange={(v) => setBody({ ...body!, format: v.value! })} /> - - setBody({ ...body!, history: v.currentTarget.checked })} /> - - - + {body?.format === 'objectStore' && !config.featureToggles.objectStore && ( +
+ Enable the `objectStore` feature flag +
+ )} + {body?.format === 'git' && ( <> - - - setInclude('*', v.currentTarget.checked)} - /> - - - {serverOptions.value && ( -
- {serverOptions.value.exporters.map((ex) => ( - - - setInclude(ex.key, v.currentTarget.checked)} - /> - - - ))} -
- )} + + setBody({ ...body!, history: v.currentTarget.checked })} + /> + + + + <> + + + setInclude('*', v.currentTarget.checked)} + /> + + + {serverOptions.value && ( +
+ {serverOptions.value.exporters.map((ex) => ( + + + setInclude(ex.key, v.currentTarget.checked)} + /> + + + ))} +
+ )} + + - + )}