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

CloudMigrations: Fix traceability & HTTP Client initialisation (#94141)

Browse Source 
* Add traceability to Migration Assistant feature

* Fix some compilation errors

* Fix lint issues

* Use async context

* Add trace for LibraryElements
This commit is contained in:
Roberto Jiménez Sánchez
2024-10-07 11:31:45 +02:00
committed by GitHub
parent 9680722b78
commit 19c77eaae1
8 changed files with 257 additions and 62 deletions
Download Patch File Download Diff File Expand all files Collapse all files
pkg/services/cloudmigration/cloudmigrationimpl/cloudmigration.go
+73 -20
View File
@@ -12,6 +12,7 @@ import (
"time"
"github.com/google/uuid"
"github.com/grafana/grafana-plugin-sdk-go/backend/httpclient"
"github.com/grafana/grafana/pkg/api/routing"
"github.com/grafana/grafana/pkg/infra/db"
"github.com/grafana/grafana/pkg/infra/kvstore"
@@ -35,6 +36,7 @@ import (
"github.com/grafana/grafana/pkg/util"
"github.com/prometheus/client_golang/prometheus"
"go.opentelemetry.io/otel/attribute"
"go.opentelemetry.io/otel/codes"
"go.opentelemetry.io/otel/trace"
)
@@ -83,6 +85,7 @@ var _ cloudmigration.Service = (*Service)(nil)
// builds the service, and api, and configures routes
func ProvideService(
cfg *setting.Cfg,
httpClientProvider *httpclient.Provider,
features featuremgmt.FeatureToggles,
db db.DB,
dsService datasources.DataSourceService,
@@ -118,15 +121,29 @@ func ProvideService(
}
s.api = api.RegisterApi(routeRegister, s, tracer)
s.objectStorage = objectstorage.NewS3()
httpClientS3, err := httpClientProvider.New()
if err != nil {
return nil, fmt.Errorf("creating http client for S3: %w", err)
}
s.objectStorage = objectstorage.NewS3(httpClientS3, tracer)
if !cfg.CloudMigration.IsDeveloperMode {
c, err := gmsclient.NewGMSClient(cfg)
httpClientGMS, err := httpClientProvider.New()
if err != nil {
return nil, fmt.Errorf("creating http client for GMS: %w", err)
}
c, err := gmsclient.NewGMSClient(cfg, httpClientGMS)
if err != nil {
return nil, fmt.Errorf("initializing GMS client: %w", err)
}
s.gmsClient = c
s.gcomService = gcom.New(gcom.Config{ApiURL: cfg.GrafanaComAPIURL, Token: cfg.CloudMigration.GcomAPIToken})
httpClientGcom, err := httpClientProvider.New()
if err != nil {
return nil, fmt.Errorf("creating http client for GCOM: %w", err)
}
s.gcomService = gcom.New(gcom.Config{ApiURL: cfg.GrafanaComAPIURL, Token: cfg.CloudMigration.GcomAPIToken}, httpClientGcom)
} else {
s.gmsClient = gmsclient.NewInMemoryClient()
s.gcomService = &gcomStub{policies: map[string]gcom.AccessPolicy{}, token: nil}
@@ -169,7 +186,8 @@ func (s *Service) GetToken(ctx context.Context) (gcom.TokenView, error) {
RequestID: requestID,
Region: instance.RegionSlug,
AccessPolicyName: accessPolicyName,
TokenName: accessTokenName})
TokenName: accessTokenName,
})
if err != nil {
return gcom.TokenView{}, fmt.Errorf("listing tokens: %w", err)
}
@@ -279,9 +297,6 @@ func (s *Service) CreateToken(ctx context.Context) (cloudmigration.CreateAccessT
}
func (s *Service) findAccessPolicyByName(ctx context.Context, regionSlug, accessPolicyName string) (*gcom.AccessPolicy, error) {
ctx, span := s.tracer.Start(ctx, "CloudMigrationService.findAccessPolicyByName")
defer span.End()
accessPolicies, err := s.gcomService.ListAccessPolicies(ctx, gcom.ListAccessPoliciesParams{
RequestID: tracing.TraceIDFromContext(ctx, false),
Region: regionSlug,
@@ -341,7 +356,7 @@ func (s *Service) DeleteToken(ctx context.Context, tokenID string) error {
}
func (s *Service) GetSession(ctx context.Context, uid string) (*cloudmigration.CloudMigrationSession, error) {
ctx, span := s.tracer.Start(ctx, "CloudMigrationService.GetMigration")
ctx, span := s.tracer.Start(ctx, "CloudMigrationService.GetSession")
defer span.End()
migration, err := s.store.GetMigrationSessionByUID(ctx, uid)
if err != nil {
@@ -352,6 +367,9 @@ func (s *Service) GetSession(ctx context.Context, uid string) (*cloudmigration.C
}
func (s *Service) GetSessionList(ctx context.Context) (*cloudmigration.CloudMigrationSessionListResponse, error) {
ctx, span := s.tracer.Start(ctx, "CloudMigrationService.GetSessionList")
defer span.End()
values, err := s.store.GetCloudMigrationSessionList(ctx)
if err != nil {
return nil, fmt.Errorf("retrieving session list from store: %w", err)
@@ -370,7 +388,7 @@ func (s *Service) GetSessionList(ctx context.Context) (*cloudmigration.CloudMigr
}
func (s *Service) CreateSession(ctx context.Context, cmd cloudmigration.CloudMigrationSessionRequest) (*cloudmigration.CloudMigrationSessionResponse, error) {
ctx, span := s.tracer.Start(ctx, "CloudMigrationService.createMigration")
ctx, span := s.tracer.Start(ctx, "CloudMigrationService.CreateSession")
defer span.End()
base64Token := cmd.AuthToken
@@ -405,6 +423,9 @@ func (s *Service) CreateSession(ctx context.Context, cmd cloudmigration.CloudMig
}
func (s *Service) DeleteSession(ctx context.Context, sessionUID string) (*cloudmigration.CloudMigrationSession, error) {
ctx, span := s.tracer.Start(ctx, "CloudMigrationService.DeleteSession")
defer span.End()
session, snapshots, err := s.store.DeleteMigrationSessionByUID(ctx, sessionUID)
if err != nil {
s.report(ctx, session, gmsclient.EventDisconnect, 0, err)
@@ -470,26 +491,36 @@ func (s *Service) CreateSnapshot(ctx context.Context, signedInUser *user.SignedI
s.cancelMutex.Unlock()
}()
ctx, cancelFunc := context.WithCancel(context.Background())
// Create context out the span context to ensure the trace is propagated
asyncCtx := trace.ContextWithSpanContext(context.Background(), span.SpanContext())
asyncCtx, asyncSpan := s.tracer.Start(asyncCtx, "CloudMigrationService.CreateSnapshotAsync")
defer asyncSpan.End()
asyncCtx, cancelFunc := context.WithCancel(asyncCtx)
s.cancelFunc = cancelFunc
s.report(ctx, session, gmsclient.EventStartBuildingSnapshot, 0, nil)
s.report(asyncCtx, session, gmsclient.EventStartBuildingSnapshot, 0, nil)
start := time.Now()
err := s.buildSnapshot(ctx, signedInUser, initResp.MaxItemsPerPartition, initResp.Metadata, snapshot)
err := s.buildSnapshot(asyncCtx, signedInUser, initResp.MaxItemsPerPartition, initResp.Metadata, snapshot)
if err != nil {
asyncSpan.SetStatus(codes.Error, "error building snapshot")
asyncSpan.RecordError(err)
s.log.Error("building snapshot", "err", err.Error())
// Update status to error with retries
if err := s.updateSnapshotWithRetries(context.Background(), cloudmigration.UpdateSnapshotCmd{
if err := s.updateSnapshotWithRetries(asyncCtx, cloudmigration.UpdateSnapshotCmd{
UID: snapshot.UID,
SessionID: sessionUid,
Status: cloudmigration.SnapshotStatusError,
}); err != nil {
s.log.Error("critical failure during snapshot creation - please report any error logs")
asyncSpan.RecordError(err)
}
}
s.report(ctx, session, gmsclient.EventDoneBuildingSnapshot, time.Since(start), err)
span.SetStatus(codes.Ok, "snapshot built")
s.report(asyncCtx, session, gmsclient.EventDoneBuildingSnapshot, time.Since(start), err)
}()
return &snapshot, nil
@@ -624,32 +655,48 @@ func (s *Service) UploadSnapshot(ctx context.Context, sessionUid string, snapsho
s.cancelMutex.Unlock()
}()
ctx, cancelFunc := context.WithCancel(context.Background())
s.cancelFunc = cancelFunc
// Create context out the span context to ensure the trace is propagated
asyncCtx := trace.ContextWithSpanContext(context.Background(), span.SpanContext())
asyncCtx, asyncSpan := s.tracer.Start(asyncCtx, "CloudMigrationService.UploadSnapshot")
defer asyncSpan.End()
s.report(ctx, session, gmsclient.EventStartUploadingSnapshot, 0, nil)
asyncCtx, s.cancelFunc = context.WithCancel(asyncCtx)
s.report(asyncCtx, session, gmsclient.EventStartUploadingSnapshot, 0, nil)
start := time.Now()
err := s.uploadSnapshot(ctx, session, snapshot, uploadUrl)
err := s.uploadSnapshot(asyncCtx, session, snapshot, uploadUrl)
if err != nil {
asyncSpan.SetStatus(codes.Error, "error uploading snapshot")
asyncSpan.RecordError(err)
s.log.Error("uploading snapshot", "err", err.Error())
// Update status to error with retries
if err := s.updateSnapshotWithRetries(context.Background(), cloudmigration.UpdateSnapshotCmd{
if err := s.updateSnapshotWithRetries(asyncCtx, cloudmigration.UpdateSnapshotCmd{
UID: snapshot.UID,
SessionID: sessionUid,
Status: cloudmigration.SnapshotStatusError,
}); err != nil {
asyncSpan.RecordError(err)
s.log.Error("critical failure during snapshot upload - please report any error logs")
}
}
s.report(ctx, session, gmsclient.EventDoneUploadingSnapshot, time.Since(start), err)
s.report(asyncCtx, session, gmsclient.EventDoneUploadingSnapshot, time.Since(start), err)
}()
return nil
}
func (s *Service) CancelSnapshot(ctx context.Context, sessionUid string, snapshotUid string) (err error) {
ctx, span := s.tracer.Start(ctx, "CloudMigrationService.CancelSnapshot",
trace.WithAttributes(
attribute.String("sessionUid", sessionUid),
attribute.String("snapshotUid", snapshotUid),
),
)
defer span.End()
// The cancel func itself is protected by a mutex in the async threads, so it may or may not be set by the time CancelSnapshot is called
// Attempt to cancel and recover from the panic if the cancel function is nil
defer func() {
@@ -684,6 +731,9 @@ func (s *Service) report(
d time.Duration,
evtErr error,
) {
ctx, span := s.tracer.Start(ctx, "CloudMigrationService.report")
defer span.End()
id, err := s.getLocalEventId(ctx)
if err != nil {
s.log.Error("failed to report event", "type", t, "error", err.Error())
@@ -738,6 +788,9 @@ func (s *Service) getLocalEventId(ctx context.Context) (string, error) {
}
func (s *Service) deleteLocalFiles(snapshots []cloudmigration.CloudMigrationSnapshot) error {
_, span := s.tracer.Start(context.Background(), "CloudMigrationService.deleteLocalFiles")
defer span.End()
var err error
for _, snapshot := range snapshots {
err = os.RemoveAll(snapshot.LocalDir)
pkg/services/cloudmigration/cloudmigrationimpl/cloudmigration_test.go
+2
View File
@@ -8,6 +8,7 @@ import (
"time"
"github.com/google/uuid"
"github.com/grafana/grafana-plugin-sdk-go/backend/httpclient"
"github.com/grafana/grafana/pkg/api/routing"
"github.com/grafana/grafana/pkg/components/simplejson"
"github.com/grafana/grafana/pkg/infra/db"
@@ -644,6 +645,7 @@ func setUpServiceTest(t *testing.T, withDashboardMock bool) cloudmigration.Servi
s, err := ProvideService(
cfg,
httpclient.NewProvider(),
featuremgmt.WithFeatures(
featuremgmt.FlagOnPremToCloudMigrations,
featuremgmt.FlagDashboardRestore),
pkg/services/cloudmigration/cloudmigrationimpl/snapshot_mgmt.go
+49 -2
View File
@@ -23,9 +23,14 @@ import (
"github.com/grafana/grafana/pkg/services/user"
"github.com/grafana/grafana/pkg/util/retryer"
"golang.org/x/crypto/nacl/box"
"go.opentelemetry.io/otel/codes"
)
func (s *Service) getMigrationDataJSON(ctx context.Context, signedInUser *user.SignedInUser) (*cloudmigration.MigrateDataRequest, error) {
ctx, span := s.tracer.Start(ctx, "CloudMigrationService.getMigrationDataJSON")
defer span.End()
// Data sources
dataSources, err := s.getDataSourceCommands(ctx)
if err != nil {
@@ -103,6 +108,9 @@ func (s *Service) getMigrationDataJSON(ctx context.Context, signedInUser *user.S
}
func (s *Service) getDataSourceCommands(ctx context.Context) ([]datasources.AddDataSourceCommand, error) {
ctx, span := s.tracer.Start(ctx, "CloudMigrationService.getDataSourceCommands")
defer span.End()
dataSources, err := s.dsService.GetAllDataSources(ctx, &datasources.GetAllDataSourcesQuery{})
if err != nil {
s.log.Error("Failed to get all datasources", "err", err)
@@ -141,6 +149,9 @@ func (s *Service) getDataSourceCommands(ctx context.Context) ([]datasources.AddD
// getDashboardAndFolderCommands returns the json payloads required by the dashboard and folder creation APIs
func (s *Service) getDashboardAndFolderCommands(ctx context.Context, signedInUser *user.SignedInUser) ([]dashboards.Dashboard, []folder.CreateFolderCommand, error) {
ctx, span := s.tracer.Start(ctx, "CloudMigrationService.getDashboardAndFolderCommands")
defer span.End()
dashs, err := s.dashboardService.GetAllDashboards(ctx)
if err != nil {
return nil, nil, err
@@ -196,6 +207,9 @@ type libraryElement struct {
// getLibraryElementsCommands returns the json payloads required by the library elements creation API
func (s *Service) getLibraryElementsCommands(ctx context.Context, signedInUser *user.SignedInUser) ([]libraryElement, error) {
ctx, span := s.tracer.Start(ctx, "CloudMigrationService.getLibraryElementsCommands")
defer span.End()
const perPage = 100
cmds := make([]libraryElement, 0)
@@ -242,6 +256,9 @@ func (s *Service) getLibraryElementsCommands(ctx context.Context, signedInUser *
// asynchronous process for writing the snapshot to the filesystem and updating the snapshot status
func (s *Service) buildSnapshot(ctx context.Context, signedInUser *user.SignedInUser, maxItemsPerPartition uint32, metadata []byte, snapshotMeta cloudmigration.CloudMigrationSnapshot) error {
ctx, span := s.tracer.Start(ctx, "CloudMigrationService.buildSnapshot")
defer span.End()
// TODO -- make sure we can only build one snapshot at a time
s.buildSnapshotMutex.Lock()
defer s.buildSnapshotMutex.Unlock()
@@ -339,6 +356,9 @@ func (s *Service) buildSnapshot(ctx context.Context, signedInUser *user.SignedIn
// asynchronous process for and updating the snapshot status
func (s *Service) uploadSnapshot(ctx context.Context, session *cloudmigration.CloudMigrationSession, snapshotMeta *cloudmigration.CloudMigrationSnapshot, uploadUrl string) (err error) {
ctx, span := s.tracer.Start(ctx, "CloudMigrationService.uploadSnapshot")
defer span.End()
// TODO -- make sure we can only upload one snapshot at a time
s.buildSnapshotMutex.Lock()
defer s.buildSnapshotMutex.Unlock()
@@ -361,37 +381,61 @@ func (s *Service) uploadSnapshot(ctx context.Context, session *cloudmigration.Cl
}
}()
_, readIndexSpan := s.tracer.Start(ctx, "CloudMigrationService.uploadSnapshot.readIndex")
index, err := snapshot.ReadIndex(indexFile)
if err != nil {
readIndexSpan.SetStatus(codes.Error, "reading index from file")
readIndexSpan.RecordError(err)
readIndexSpan.End()
return fmt.Errorf("reading index from file: %w", err)
}
readIndexSpan.End()
s.log.Debug(fmt.Sprintf("uploadSnapshot: read index file in %d ms", time.Since(start).Milliseconds()))
uploadCtx, uploadSpan := s.tracer.Start(ctx, "CloudMigrationService.uploadSnapshot.uploadDataFiles")
// Upload the data files.
for _, fileNames := range index.Items {
for _, fileName := range fileNames {
filePath := filepath.Join(snapshotMeta.LocalDir, fileName)
key := fmt.Sprintf("%d/snapshots/%s/%s", session.StackID, snapshotMeta.GMSSnapshotUID, fileName)
if err := s.uploadUsingPresignedURL(ctx, uploadUrl, key, filePath); err != nil {
if err := s.uploadUsingPresignedURL(uploadCtx, uploadUrl, key, filePath); err != nil {
uploadSpan.SetStatus(codes.Error, "uploading snapshot data file using presigned url")
uploadSpan.RecordError(err)
uploadSpan.End()
return fmt.Errorf("uploading snapshot file using presigned url: %w", err)
}
s.log.Debug(fmt.Sprintf("uploadSnapshot: uploaded %s in %d ms", fileName, time.Since(start).Milliseconds()))
}
}
uploadSpan.End()
s.log.Debug(fmt.Sprintf("uploadSnapshot: uploaded all data files in %d ms", time.Since(start).Milliseconds()))
uploadCtx, uploadSpan = s.tracer.Start(ctx, "CloudMigrationService.uploadSnapshot.uploadIndex")
// Upload the index file. Must be done after uploading the data files.
key := fmt.Sprintf("%d/snapshots/%s/%s", session.StackID, snapshotMeta.GMSSnapshotUID, "index.json")
if _, err := indexFile.Seek(0, 0); err != nil {
uploadSpan.SetStatus(codes.Error, "seeking to beginning of index file")
uploadSpan.RecordError(err)
uploadSpan.End()
return fmt.Errorf("seeking to beginning of index file: %w", err)
}
if err := s.objectStorage.PresignedURLUpload(ctx, uploadUrl, key, indexFile); err != nil {
if err := s.objectStorage.PresignedURLUpload(uploadCtx, uploadUrl, key, indexFile); err != nil {
uploadSpan.SetStatus(codes.Error, "uploading index file using presigned url")
uploadSpan.RecordError(err)
uploadSpan.End()
return fmt.Errorf("uploading file using presigned url: %w", err)
}
uploadSpan.End()
s.log.Debug(fmt.Sprintf("uploadSnapshot: uploaded index file in %d ms", time.Since(start).Milliseconds()))
s.log.Info("successfully uploaded snapshot", "snapshotUid", snapshotMeta.UID, "cloud_snapshotUid", snapshotMeta.GMSSnapshotUID)
@@ -408,6 +452,9 @@ func (s *Service) uploadSnapshot(ctx context.Context, session *cloudmigration.Cl
}
func (s *Service) uploadUsingPresignedURL(ctx context.Context, uploadURL, key string, filePath string) (err error) {
ctx, span := s.tracer.Start(ctx, "CloudMigrationService.uploadUsingPresignedURL")
defer span.End()
// The directory that contains the file can set in the configuration, therefore the directory can be any directory.
// nolint:gosec
file, err := os.Open(filePath)