CloudMigrations: Implement migrations API (#85348)

* Implement run migration endpoint * Refactor RunMigration method into separate methods * Save migration runs fix lint * Minor changes * Refactor how to use cms endpoint * fix interface * complete merge * add individual items * adds tracing to getMigration * linter * updated swagger definition with the latest changes * CloudMigrations: Implement core API handlers for cloud migrations and migration runs (#85407) * implement delete * add auth token encryption * implement token validation * call token validation during migration creation * implement get migration status * implement list migration runs * fix bug * finish parse domain func * fix urls * fix typo * fix encoding and decoding * remove double decryption * add missing slash * fix id returned by create function * inject missing services * finish implementing (as far as I can tell right now) data migration and response handling * comment out broken test, needs a rewrite * add a few final touches * get dashboard migration to work properly * changed runMigration to a POST * swagger * swagger * swagger --------- Co-authored-by: Michael Mandrus <michael.mandrus@grafana.com> Co-authored-by: Leonard Gram <leo@xlson.com> Co-authored-by: Michael Mandrus <41969079+mmandrus@users.noreply.github.com>
2024-04-03 13:36:13 +02:00
parent 89638238e5
commit b885da09da
17 changed files with 966 additions and 333 deletions
@@ -1,6 +1,10 @@
 package api

 import (
+	"bytes"
+	"encoding/json"
+	"fmt"
+	"io"
 	"net/http"
 	"strconv"

@@ -15,10 +19,10 @@ import (
 )

 type CloudMigrationAPI struct {
-	cloudMigrationsService cloudmigration.Service
-	routeRegister          routing.RouteRegister
-	log                    log.Logger
-	tracer                 tracing.Tracer
+	cloudMigrationService cloudmigration.Service
+	routeRegister         routing.RouteRegister
+	log                   log.Logger
+	tracer                tracing.Tracer
 }

 func RegisterApi(
@@ -27,10 +31,10 @@ func RegisterApi(
 	tracer tracing.Tracer,
 ) *CloudMigrationAPI {
 	api := &CloudMigrationAPI{
-		log:                    log.New("cloudmigrations.api"),
-		routeRegister:          rr,
-		cloudMigrationsService: cms,
-		tracer:                 tracer,
+		log:                   log.New("cloudmigrations.api"),
+		routeRegister:         rr,
+		cloudMigrationService: cms,
+		tracer:                tracer,
 	}
 	api.registerEndpoints()
 	return api
@@ -43,7 +47,7 @@ func (cma *CloudMigrationAPI) registerEndpoints() {
 		cloudMigrationRoute.Get("/migration", routing.Wrap(cma.GetMigrationList))
 		cloudMigrationRoute.Post("/migration", routing.Wrap(cma.CreateMigration))
 		cloudMigrationRoute.Get("/migration/:id", routing.Wrap(cma.GetMigration))
-		cloudMigrationRoute.Delete("migration/:id", routing.Wrap(cma.DeleteMigration))
+		cloudMigrationRoute.Delete("/migration/:id", routing.Wrap(cma.DeleteMigration))
 		cloudMigrationRoute.Post("/migration/:id/run", routing.Wrap(cma.RunMigration))
 		cloudMigrationRoute.Get("/migration/:id/run", routing.Wrap(cma.GetMigrationRunList))
 		cloudMigrationRoute.Get("/migration/:id/run/:runID", routing.Wrap(cma.GetMigrationRun))
@@ -66,7 +70,7 @@ func (cma *CloudMigrationAPI) CreateToken(c *contextmodel.ReqContext) response.R

 	logger := cma.log.FromContext(ctx)

-	resp, err := cma.cloudMigrationsService.CreateToken(ctx)
+	resp, err := cma.cloudMigrationService.CreateToken(ctx)
 	if err != nil {
 		logger.Error("creating gcom access token", "err", err.Error())
 		return response.Error(http.StatusInternalServerError, "creating gcom access token", err)
@@ -85,7 +89,10 @@ func (cma *CloudMigrationAPI) CreateToken(c *contextmodel.ReqContext) response.R
 // 403: forbiddenError
 // 500: internalServerError
 func (cma *CloudMigrationAPI) GetMigrationList(c *contextmodel.ReqContext) response.Response {
-	cloudMigrations, err := cma.cloudMigrationsService.GetMigrationList(c.Req.Context())
+	ctx, span := cma.tracer.Start(c.Req.Context(), "MigrationAPI.GetMigrationList")
+	defer span.End()
+
+	cloudMigrations, err := cma.cloudMigrationService.GetMigrationList(ctx)
 	if err != nil {
 		return response.Error(http.StatusInternalServerError, "migration list error", err)
 	}
@@ -105,11 +112,14 @@ func (cma *CloudMigrationAPI) GetMigrationList(c *contextmodel.ReqContext) respo
 // 403: forbiddenError
 // 500: internalServerError
 func (cma *CloudMigrationAPI) GetMigration(c *contextmodel.ReqContext) response.Response {
+	ctx, span := cma.tracer.Start(c.Req.Context(), "MigrationAPI.GetMigration")
+	defer span.End()
+
 	id, err := strconv.ParseInt(web.Params(c.Req)[":id"], 10, 64)
 	if err != nil {
 		return response.Error(http.StatusBadRequest, "id is invalid", err)
 	}
-	cloudMigration, err := cma.cloudMigrationsService.GetMigration(c.Req.Context(), id)
+	cloudMigration, err := cma.cloudMigrationService.GetMigration(ctx, id)
 	if err != nil {
 		return response.Error(http.StatusNotFound, "migration not found", err)
 	}
@@ -134,18 +144,21 @@ type GetCloudMigrationRequest struct {
 // 403: forbiddenError
 // 500: internalServerError
 func (cma *CloudMigrationAPI) CreateMigration(c *contextmodel.ReqContext) response.Response {
+	ctx, span := cma.tracer.Start(c.Req.Context(), "MigrationAPI.CreateMigration")
+	defer span.End()
+
 	cmd := cloudmigration.CloudMigrationRequest{}
 	if err := web.Bind(c.Req, &cmd); err != nil {
 		return response.Error(http.StatusBadRequest, "bad request data", err)
 	}
-	cloudMigration, err := cma.cloudMigrationsService.CreateMigration(c.Req.Context(), cmd)
+	cloudMigration, err := cma.cloudMigrationService.CreateMigration(ctx, cmd)
 	if err != nil {
 		return response.Error(http.StatusInternalServerError, "migration creation error", err)
 	}
 	return response.JSON(http.StatusOK, cloudMigration)
 }

-// swagger:route GET /cloudmigration/migration/{id}/run migrations runCloudMigration
+// swagger:route POST /cloudmigration/migration/{id}/run migrations runCloudMigration
 //
 // Trigger the run of a migration to the Grafana Cloud.
 //
@@ -157,11 +170,80 @@ func (cma *CloudMigrationAPI) CreateMigration(c *contextmodel.ReqContext) respon
 // 403: forbiddenError
 // 500: internalServerError
 func (cma *CloudMigrationAPI) RunMigration(c *contextmodel.ReqContext) response.Response {
-	cloudMigrationRun, err := cma.cloudMigrationsService.RunMigration(c.Req.Context(), web.Params(c.Req)[":id"])
+	ctx, span := cma.tracer.Start(c.Req.Context(), "MigrationAPI.RunMigration")
+	defer span.End()
+	logger := cma.log.FromContext(ctx)
+
+	stringID := web.Params(c.Req)[":id"]
+	id, err := strconv.ParseInt(stringID, 10, 64)
 	if err != nil {
-		return response.Error(http.StatusInternalServerError, "migration run error", err)
+		return response.Error(http.StatusBadRequest, "id is invalid", err)
 	}
-	return response.JSON(http.StatusOK, cloudMigrationRun)
+
+	// Get migration to read the auth token
+	migration, err := cma.cloudMigrationService.GetMigration(ctx, id)
+	if err != nil {
+		return response.Error(http.StatusInternalServerError, "migration get error", err)
+	}
+	// get CMS path from the config
+	domain, err := cma.cloudMigrationService.ParseCloudMigrationConfig()
+	if err != nil {
+		return response.Error(http.StatusInternalServerError, "config parse error", err)
+	}
+	path := fmt.Sprintf("https://cms-dev-%s.%s/cloud-migrations/api/v1/migrate-data", migration.ClusterSlug, domain)
+
+	// Get migration data JSON
+	body, err := cma.cloudMigrationService.GetMigrationDataJSON(ctx, id)
+	if err != nil {
+		cma.log.Error("error getting the json request body for migration run", "err", err.Error())
+		return response.Error(http.StatusInternalServerError, "migration data get error", err)
+	}
+
+	req, err := http.NewRequest("POST", path, bytes.NewReader(body))
+	if err != nil {
+		cma.log.Error("error creating http request for cloud migration run", "err", err.Error())
+		return response.Error(http.StatusInternalServerError, "http request error", err)
+	}
+	req.Header.Set("Content-Type", "application/json")
+	req.Header.Set("Authorization", fmt.Sprintf("Bearer %d:%s", migration.StackID, migration.AuthToken))
+
+	client := &http.Client{}
+	resp, err := client.Do(req)
+	if err != nil {
+		cma.log.Error("error sending http request for cloud migration run", "err", err.Error())
+		return response.Error(http.StatusInternalServerError, "http request error", err)
+	} else if resp.StatusCode >= 400 {
+		cma.log.Error("received error response for cloud migration run", "statusCode", resp.StatusCode)
+		return response.Error(http.StatusInternalServerError, "http request error", fmt.Errorf("http request error while migrating data"))
+	}
+
+	defer func() {
+		if err := resp.Body.Close(); err != nil {
+			logger.Error("closing request body: %w", err)
+		}
+	}()
+
+	// read response so we can unmarshal it
+	respData, err := io.ReadAll(resp.Body)
+	if err != nil {
+		logger.Error("reading response body: %w", err)
+		return response.Error(http.StatusInternalServerError, "reading migration run response", err)
+	}
+	var result cloudmigration.MigrateDataResponseDTO
+	if err := json.Unmarshal(respData, &result); err != nil {
+		logger.Error("unmarshalling response body: %w", err)
+		return response.Error(http.StatusInternalServerError, "unmarshalling migration run response", err)
+	}
+
+	_, err = cma.cloudMigrationService.SaveMigrationRun(ctx, &cloudmigration.CloudMigrationRun{
+		CloudMigrationUID: stringID,
+		Result:            respData,
+	})
+	if err != nil {
+		response.Error(http.StatusInternalServerError, "migration run save error", err)
+	}
+
+	return response.JSON(http.StatusOK, result)
 }

 // swagger:parameters runCloudMigration
@@ -182,7 +264,10 @@ type RunCloudMigrationRequest struct {
 // 403: forbiddenError
 // 500: internalServerError
 func (cma *CloudMigrationAPI) GetMigrationRun(c *contextmodel.ReqContext) response.Response {
-	migrationStatus, err := cma.cloudMigrationsService.GetMigrationStatus(c.Req.Context(), web.Params(c.Req)[":id"], web.Params(c.Req)[":runID"])
+	ctx, span := cma.tracer.Start(c.Req.Context(), "MigrationAPI.GetMigrationRun")
+	defer span.End()
+
+	migrationStatus, err := cma.cloudMigrationService.GetMigrationStatus(ctx, web.Params(c.Req)[":id"], web.Params(c.Req)[":runID"])
 	if err != nil {
 		return response.Error(http.StatusInternalServerError, "migration status error", err)
 	}
@@ -212,12 +297,27 @@ type GetMigrationRunParams struct {
 // 403: forbiddenError
 // 500: internalServerError
 func (cma *CloudMigrationAPI) GetMigrationRunList(c *contextmodel.ReqContext) response.Response {
-	migrationStatus, err := cma.cloudMigrationsService.GetMigrationStatusList(c.Req.Context(), web.Params(c.Req)[":id"])
+	ctx, span := cma.tracer.Start(c.Req.Context(), "MigrationAPI.GetMigrationRunList")
+	defer span.End()
+
+	migrationStatuses, err := cma.cloudMigrationService.GetMigrationStatusList(ctx, web.Params(c.Req)[":id"])
 	if err != nil {
 		return response.Error(http.StatusInternalServerError, "migration status error", err)
 	}

-	runList := cloudmigration.CloudMigrationRunList{Runs: migrationStatus}
+	runList := cloudmigration.CloudMigrationRunList{Runs: []cloudmigration.MigrateDataResponseDTO{}}
+	for _, s := range migrationStatuses {
+		// attempt to bind the raw result to a list of response item DTOs
+		r := cloudmigration.MigrateDataResponseDTO{
+			Items: []cloudmigration.MigrateDataResponseItemDTO{},
+		}
+		if err := json.Unmarshal(s.Result, &r); err != nil {
+			return response.Error(http.StatusInternalServerError, "error unmarshalling migration response items", err)
+		}
+		r.RunID = s.ID
+		runList.Runs = append(runList.Runs, r)
+	}
+
 	return response.JSON(http.StatusOK, runList)
 }

@@ -239,7 +339,18 @@ type GetCloudMigrationRunList struct {
 // 403: forbiddenError
 // 500: internalServerError
 func (cma *CloudMigrationAPI) DeleteMigration(c *contextmodel.ReqContext) response.Response {
-	err := cma.cloudMigrationsService.DeleteMigration(c.Req.Context(), web.Params(c.Req)[":id"])
+	ctx, span := cma.tracer.Start(c.Req.Context(), "MigrationAPI.DeleteMigration")
+	defer span.End()
+
+	idStr := web.Params(c.Req)[":id"]
+	if idStr == "" {
+		return response.Error(http.StatusBadRequest, "missing migration id", fmt.Errorf("missing migration id"))
+	}
+	id, err := strconv.ParseInt(idStr, 10, 64)
+	if err != nil {
+		return response.Error(http.StatusBadRequest, "migration id should be numeric", fmt.Errorf("migration id should be numeric"))
+	}
+	_, err = cma.cloudMigrationService.DeleteMigration(ctx, id)
 	if err != nil {
 		return response.Error(http.StatusInternalServerError, "migration delete error", err)
 	}
@@ -257,7 +368,7 @@ type DeleteMigrationRequest struct {
 // swagger:response cloudMigrationRunResponse
 type CloudMigrationRunResponse struct {
 	// in: body
-	Body cloudmigration.CloudMigrationRun
+	Body cloudmigration.MigrateDataResponseDTO
 }

 // swagger:response cloudMigrationListResponse