Export: introduce export plumbing (behind dev feature flag) (#48091)
This commit is contained in:
Ryan McKinley
@@ -0,0 +1,103 @@
|
||||
package export
|
||||
|
||||
import (
|
||||
"errors"
|
||||
"fmt"
|
||||
"math"
|
||||
"math/rand"
|
||||
"sync"
|
||||
"time"
|
||||
|
||||
"github.com/grafana/grafana/pkg/infra/log"
|
||||
)
|
||||
|
||||
var _ Job = new(dummyExportJob)
|
||||
|
||||
type dummyExportJob struct {
|
||||
logger log.Logger
|
||||
|
||||
statusMu sync.Mutex
|
||||
status ExportStatus
|
||||
cfg ExportConfig
|
||||
broadcaster statusBroadcaster
|
||||
}
|
||||
|
||||
func startDummyExportJob(cfg ExportConfig, broadcaster statusBroadcaster) (Job, error) {
|
||||
if cfg.Format != "git" {
|
||||
return nil, errors.New("only git format is supported")
|
||||
}
|
||||
|
||||
job := &dummyExportJob{
|
||||
logger: log.New("dummy_export_job"),
|
||||
cfg: cfg,
|
||||
broadcaster: broadcaster,
|
||||
status: ExportStatus{
|
||||
Running: true,
|
||||
Target: "git export",
|
||||
Started: time.Now().UnixMilli(),
|
||||
Count: int64(math.Round(10 + rand.Float64()*20)),
|
||||
Current: 0,
|
||||
},
|
||||
}
|
||||
|
||||
broadcaster(job.status)
|
||||
go job.start()
|
||||
return job, nil
|
||||
}
|
||||
|
||||
func (e *dummyExportJob) 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")
|
||||
|
||||
ticker := time.NewTicker(1 * time.Second)
|
||||
for t := range ticker.C {
|
||||
e.statusMu.Lock()
|
||||
e.status.Changed = t.UnixMilli()
|
||||
e.status.Current++
|
||||
e.status.Last = fmt.Sprintf("ITEM: %d", e.status.Current)
|
||||
e.statusMu.Unlock()
|
||||
|
||||
// Wait till we are done
|
||||
shouldStop := e.status.Current >= e.status.Count
|
||||
e.broadcaster(e.status)
|
||||
|
||||
if shouldStop {
|
||||
break
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
func (e *dummyExportJob) getStatus() ExportStatus {
|
||||
e.statusMu.Lock()
|
||||
defer e.statusMu.Unlock()
|
||||
|
||||
return e.status
|
||||
}
|
||||
|
||||
func (e *dummyExportJob) getConfig() ExportConfig {
|
||||
e.statusMu.Lock()
|
||||
defer e.statusMu.Unlock()
|
||||
|
||||
return e.cfg
|
||||
}
|
||||
@@ -0,0 +1,93 @@
|
||||
package export
|
||||
|
||||
import (
|
||||
"encoding/json"
|
||||
"net/http"
|
||||
"sync"
|
||||
|
||||
"github.com/grafana/grafana/pkg/api/response"
|
||||
"github.com/grafana/grafana/pkg/infra/log"
|
||||
"github.com/grafana/grafana/pkg/models"
|
||||
"github.com/grafana/grafana/pkg/services/featuremgmt"
|
||||
"github.com/grafana/grafana/pkg/services/live"
|
||||
"github.com/grafana/grafana/pkg/services/sqlstore"
|
||||
)
|
||||
|
||||
type ExportService interface {
|
||||
// List folder contents
|
||||
HandleGetStatus(c *models.ReqContext) response.Response
|
||||
|
||||
// Read raw file contents out of the store
|
||||
HandleRequestExport(c *models.ReqContext) response.Response
|
||||
}
|
||||
|
||||
type StandardExport struct {
|
||||
logger log.Logger
|
||||
sql *sqlstore.SQLStore
|
||||
glive *live.GrafanaLive
|
||||
mutex sync.Mutex
|
||||
|
||||
// updated with mutex
|
||||
exportJob Job
|
||||
}
|
||||
|
||||
func ProvideService(sql *sqlstore.SQLStore, features featuremgmt.FeatureToggles, gl *live.GrafanaLive) ExportService {
|
||||
if !features.IsEnabled(featuremgmt.FlagExport) {
|
||||
return &StubExport{}
|
||||
}
|
||||
|
||||
return &StandardExport{
|
||||
sql: sql,
|
||||
glive: gl,
|
||||
logger: log.New("export_service"),
|
||||
exportJob: &stoppedJob{},
|
||||
}
|
||||
}
|
||||
|
||||
func (ex *StandardExport) HandleGetStatus(c *models.ReqContext) response.Response {
|
||||
ex.mutex.Lock()
|
||||
defer ex.mutex.Unlock()
|
||||
|
||||
return response.JSON(http.StatusOK, ex.exportJob.getStatus())
|
||||
}
|
||||
|
||||
func (ex *StandardExport) HandleRequestExport(c *models.ReqContext) response.Response {
|
||||
var cfg ExportConfig
|
||||
err := json.NewDecoder(c.Req.Body).Decode(&cfg)
|
||||
if err != nil {
|
||||
return response.Error(http.StatusBadRequest, "unable to read config", err)
|
||||
}
|
||||
|
||||
ex.mutex.Lock()
|
||||
defer ex.mutex.Unlock()
|
||||
|
||||
status := ex.exportJob.getStatus()
|
||||
if status.Running {
|
||||
ex.logger.Error("export already running")
|
||||
return response.Error(http.StatusLocked, "export already running", nil)
|
||||
}
|
||||
|
||||
job, err := startDummyExportJob(cfg, func(s ExportStatus) {
|
||||
ex.broadcastStatus(c.OrgId, s)
|
||||
})
|
||||
if err != nil {
|
||||
ex.logger.Error("failed to start export job", "err", err)
|
||||
return response.Error(http.StatusBadRequest, "failed to start export job", err)
|
||||
}
|
||||
|
||||
ex.exportJob = job
|
||||
return response.JSON(http.StatusOK, ex.exportJob.getStatus())
|
||||
}
|
||||
|
||||
func (ex *StandardExport) broadcastStatus(orgID int64, s ExportStatus) {
|
||||
msg, err := json.Marshal(s)
|
||||
if err != nil {
|
||||
ex.logger.Warn("Error making message", "err", err)
|
||||
return
|
||||
}
|
||||
err = ex.glive.Publish(orgID, "grafana/broadcast/export", msg)
|
||||
if err != nil {
|
||||
ex.logger.Warn("Error Publish message", "err", err)
|
||||
return
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,19 @@
|
||||
package export
|
||||
|
||||
import "time"
|
||||
|
||||
var _ Job = new(stoppedJob)
|
||||
|
||||
type stoppedJob struct {
|
||||
}
|
||||
|
||||
func (e *stoppedJob) getStatus() ExportStatus {
|
||||
return ExportStatus{
|
||||
Running: false,
|
||||
Changed: time.Now().UnixMilli(),
|
||||
}
|
||||
}
|
||||
|
||||
func (e *stoppedJob) getConfig() ExportConfig {
|
||||
return ExportConfig{}
|
||||
}
|
||||
@@ -0,0 +1,20 @@
|
||||
package export
|
||||
|
||||
import (
|
||||
"net/http"
|
||||
|
||||
"github.com/grafana/grafana/pkg/api/response"
|
||||
"github.com/grafana/grafana/pkg/models"
|
||||
)
|
||||
|
||||
var _ ExportService = new(StubExport)
|
||||
|
||||
type StubExport struct{}
|
||||
|
||||
func (ex *StubExport) HandleGetStatus(c *models.ReqContext) response.Response {
|
||||
return response.Error(http.StatusForbidden, "feature not enabled", nil)
|
||||
}
|
||||
|
||||
func (ex *StubExport) HandleRequestExport(c *models.ReqContext) response.Response {
|
||||
return response.Error(http.StatusForbidden, "feature not enabled", nil)
|
||||
}
|
||||
@@ -0,0 +1,36 @@
|
||||
package export
|
||||
|
||||
// Export status. Only one running at a time
|
||||
type ExportStatus struct {
|
||||
Running bool `json:"running"`
|
||||
Target string `json:"target"` // description of where it is going (no secrets)
|
||||
Started int64 `json:"started,omitempty"`
|
||||
Finished int64 `json:"finished,omitempty"`
|
||||
Changed int64 `json:"update,omitempty"`
|
||||
Count int64 `json:"count,omitempty"`
|
||||
Current int64 `json:"current,omitempty"`
|
||||
Last string `json:"last,omitempty"`
|
||||
Status string `json:"status"` // ERROR, SUCCESS, ETC
|
||||
}
|
||||
|
||||
// Basic export config (for now)
|
||||
type ExportConfig struct {
|
||||
Format string `json:"format"`
|
||||
Git GitExportConfig `json:"git"`
|
||||
}
|
||||
|
||||
type GitExportConfig struct {
|
||||
// General folder is either at the root or as a subfolder
|
||||
GeneralAtRoot bool `json:"generalAtRoot"`
|
||||
|
||||
// Keeping all history is nice, but much slower
|
||||
ExcludeHistory bool `json:"excludeHistory"`
|
||||
}
|
||||
|
||||
type Job interface {
|
||||
getStatus() ExportStatus
|
||||
getConfig() ExportConfig
|
||||
}
|
||||
|
||||
// Will broadcast the live status
|
||||
type statusBroadcaster func(s ExportStatus)
|
||||
Reference in New Issue
Block a user