feat(tenant): Build out the onboarding system in tenant and integrate…

… it into launcher (#17558)

We are adding in a setup/user route this is not in swagger at the moment but will be
added once we feel it is stable.

cmd/influxd/launcher/launcher.go

@@ -595,7 +595,6 @@ func (m *Launcher) run(ctx context.Context) (err error) {
 		userLogSvc                platform.UserOperationLogService         = m.kvService
 		bucketLogSvc              platform.BucketOperationLogService       = m.kvService
 		orgLogSvc                 platform.OrganizationOperationLogService = m.kvService
-		onboardingSvc             platform.OnboardingService               = m.kvService
 		scraperTargetSvc          platform.ScraperTargetStoreService       = m.kvService
 		telegrafSvc               platform.TelegrafConfigStore             = m.kvService
 		userResourceSvc           platform.UserResourceMappingService      = m.kvService
@@ -605,6 +604,12 @@ func (m *Launcher) run(ctx context.Context) (err error) {
 		notificationEndpointStore platform.NotificationEndpointService     = m.kvService
 	)
 
+	store, err := tenant.NewStore(m.kvStore)
+	if err != nil {
+		m.log.Error("Failed creating new meta store", zap.Error(err))
+		return err
+	}
+
 	if m.enableNewMetaStore {
 		var ts platform.TenantService
 		if m.newMetaStoreReadOnly {
@@ -617,11 +622,6 @@ func (m *Launcher) run(ctx context.Context) (err error) {
 			newSvc := tenant.NewService(store)
 			ts = tenant.NewDuplicateReadTenantService(m.log, oldSvc, newSvc)
 		} else {
-			store, err := tenant.NewStore(m.kvStore)
-			if err != nil {
-				m.log.Error("Failed creating new meta store", zap.Error(err))
-				return err
-			}
 			ts = tenant.NewService(store)
 		}
 
@@ -888,7 +888,6 @@ func (m *Launcher) run(ctx context.Context) (err error) {
 		SourceService:                   sourceSvc,
 		VariableService:                 variableSvc,
 		PasswordsService:                passwdsSvc,
-		OnboardingService:               onboardingSvc,
 		InfluxQLService:                 storageQueryService,
 		FluxService:                     storageQueryService,
 		TaskService:                     taskSvc,
@@ -943,8 +942,18 @@ func (m *Launcher) run(ctx context.Context) (err error) {
 		pkgHTTPServer = pkger.NewHTTPServer(pkgServerLogger, pkgSVC)
 	}
 
+	var onboardHTTPServer *tenant.OnboardHandler
+	{
+		onboardSvc := tenant.NewOnboardService(store, authSvc)                                            // basic service
+		onboardSvc = tenant.NewAuthedOnboardSvc(onboardSvc)                                               // with auth
+		onboardSvc = tenant.NewOnboardingMetrics(m.reg, onboardSvc, tenant.WithSuffix("new"))             // with metrics
+		onboardSvc = tenant.NewOnboardingLogger(m.log.With(zap.String("handler", "onboard")), onboardSvc) // with logging
+
+		onboardHTTPServer = tenant.NewHTTPOnboardHandler(m.log, onboardSvc)
+	}
+
 	{
-		platformHandler := http.NewPlatformHandler(m.apibackend, http.WithResourceHandler(pkgHTTPServer))
+		platformHandler := http.NewPlatformHandler(m.apibackend, http.WithResourceHandler(pkgHTTPServer), http.WithResourceHandler(onboardHTTPServer))
 
 		httpLogger := m.log.With(zap.String("service", "http"))
 		m.httpServer.Handler = http.NewHandlerFromRegistry(

http/api_handler.go

@@ -69,7 +69,6 @@ type APIBackend struct {
 	SourceService                   influxdb.SourceService
 	VariableService                 influxdb.VariableService
 	PasswordsService                influxdb.PasswordsService
-	OnboardingService               influxdb.OnboardingService
 	InfluxQLService                 query.ProxyQueryService
 	FluxService                     query.ProxyQueryService
 	TaskService                     influxdb.TaskService
@@ -180,9 +179,6 @@ func NewAPIHandler(b *APIBackend, opts ...APIHandlerOptFn) *APIHandler {
 	h.Mount(prefixSignIn, sessionHandler)
 	h.Mount(prefixSignOut, sessionHandler)
 
-	setupBackend := NewSetupBackend(b.Logger.With(zap.String("handler", "setup")), b)
-	h.Mount(prefixSetup, NewSetupHandler(b.Logger, setupBackend))
-
 	sourceBackend := NewSourceBackend(b.Logger.With(zap.String("handler", "source")), b)
 	sourceBackend.SourceService = authorizer.NewSourceService(b.SourceService)
 	sourceBackend.BucketService = authorizer.NewBucketService(b.BucketService, noAuthUserResourceMappingService)

http/onboarding.go

@@ -4,6 +4,7 @@ import (
 	"bytes"
 	"context"
 	"encoding/json"
+	"errors"
 	"fmt"
 	"net/http"
 
@@ -23,9 +24,9 @@ type SetupBackend struct {
 // NewSetupBackend returns a new instance of SetupBackend.
 func NewSetupBackend(log *zap.Logger, b *APIBackend) *SetupBackend {
 	return &SetupBackend{
-		HTTPErrorHandler:  b.HTTPErrorHandler,
-		log:               log,
-		OnboardingService: b.OnboardingService,
+		HTTPErrorHandler: b.HTTPErrorHandler,
+		log:              log,
+		// OnboardingService: b.OnboardingService,
 	}
 }
 
@@ -210,3 +211,7 @@ func (s *SetupService) OnboardInitialUser(ctx context.Context, or *platform.Onbo
 		Bucket: bkt,
 	}, nil
 }
+
+func (s *SetupService) OnboardUser(ctx context.Context, or *platform.OnboardingRequest) (*platform.OnboardingResults, error) {
+	return nil, errors.New("not yet implemented")
+}

kv/onboarding.go

@@ -2,6 +2,7 @@ package kv
 
 import (
 	"context"
+	"errors"
 	"fmt"
 	"time"
 
@@ -171,3 +172,7 @@ func (s *Service) OnboardInitialUser(ctx context.Context, req *influxdb.Onboardi
 		Auth:   auth,
 	}, nil
 }
+
+func (s *Service) OnboardUser(ctx context.Context, req *influxdb.OnboardingRequest) (*influxdb.OnboardingResults, error) {
+	return nil, errors.New("not yet implemented")
+}

mock/onboarding_service.go

@@ -16,15 +16,19 @@ type OnboardingService struct {
 	UserService
 	AuthorizationService
 
-	IsOnboardingFn func(context.Context) (bool, error)
-	GenerateFn     func(context.Context, *platform.OnboardingRequest) (*platform.OnboardingResults, error)
+	IsOnboardingFn       func(context.Context) (bool, error)
+	OnboardInitialUserFn func(context.Context, *platform.OnboardingRequest) (*platform.OnboardingResults, error)
+	OnboardUserFn        func(context.Context, *platform.OnboardingRequest) (*platform.OnboardingResults, error)
 }
 
 // NewOnboardingService returns a mock of OnboardingService where its methods will return zero values.
 func NewOnboardingService() *OnboardingService {
 	return &OnboardingService{
 		IsOnboardingFn: func(context.Context) (bool, error) { return false, nil },
-		GenerateFn: func(context.Context, *platform.OnboardingRequest) (*platform.OnboardingResults, error) {
+		OnboardInitialUserFn: func(context.Context, *platform.OnboardingRequest) (*platform.OnboardingResults, error) {
+			return nil, nil
+		},
+		OnboardUserFn: func(context.Context, *platform.OnboardingRequest) (*platform.OnboardingResults, error) {
 			return nil, nil
 		},
 	}
@@ -37,5 +41,10 @@ func (s *OnboardingService) IsOnboarding(ctx context.Context) (bool, error) {
 
 // OnboardInitialUser OnboardingResults.
 func (s *OnboardingService) OnboardInitialUser(ctx context.Context, req *platform.OnboardingRequest) (*platform.OnboardingResults, error) {
-	return s.GenerateFn(ctx, req)
+	return s.OnboardInitialUserFn(ctx, req)
+}
+
+// OnboardUser OnboardingResults.
+func (s *OnboardingService) OnboardUser(ctx context.Context, req *platform.OnboardingRequest) (*platform.OnboardingResults, error) {
+	return s.OnboardUserFn(ctx, req)
 }

onboarding.go

@@ -4,16 +4,14 @@ import "context"
 
 // OnboardingService represents a service for the first run.
 type OnboardingService interface {
-	PasswordsService
-	BucketService
-	OrganizationService
-	UserService
-	AuthorizationService
-
 	// IsOnboarding determine if onboarding request is allowed.
 	IsOnboarding(ctx context.Context) (bool, error)
+
 	// OnboardInitialUser OnboardingResults.
 	OnboardInitialUser(ctx context.Context, req *OnboardingRequest) (*OnboardingResults, error)
+
+	// OnboardUser creates a new user/org/buckets
+	OnboardUser(ctx context.Context, req *OnboardingRequest) (*OnboardingResults, error)
 }
 
 // OnboardingResults is a group of elements required for first run.

tenant/error.go

@@ -27,6 +27,17 @@ var (
 		Code: influxdb.EInternal,
 		Msg:  "unable to generate valid id",
 	}
+
+	// ErrOnboardingNotAllowed occurs when request to onboard comes in and we are not allowing this request
+	ErrOnboardingNotAllowed = &influxdb.Error{
+		Code: influxdb.EConflict,
+		Msg:  "onboarding has already been completed",
+	}
+
+	ErrOnboardInvalid = &influxdb.Error{
+		Code: influxdb.EEmptyValue,
+		Msg:  "onboard failed, missing value",
+	}
 )
 
 // ErrCorruptID the ID stored in the Store is corrupt.

tenant/http_client_onboarding.go

@@ -0,0 +1,79 @@
+package tenant
+
+import (
+	"context"
+	"path"
+
+	"github.com/influxdata/influxdb/v2"
+	"github.com/influxdata/influxdb/v2/pkg/httpc"
+)
+
+// OnboardClientService connects to Influx via HTTP to perform onboarding operations
+type OnboardClientService struct {
+	Client *httpc.Client
+}
+
+// IsOnboarding determine if onboarding request is allowed.
+func (s *OnboardClientService) IsOnboarding(ctx context.Context) (bool, error) {
+	var resp isOnboardingResponse
+	err := s.Client.
+		Get(prefixOrganizations).
+		DecodeJSON(&resp).
+		Do(ctx)
+
+	if err != nil {
+		return false, err
+	}
+	return resp.Allowed, nil
+}
+
+// OnboardInitialUser OnboardingResults.
+func (s *OnboardClientService) OnboardInitialUser(ctx context.Context, or *influxdb.OnboardingRequest) (*influxdb.OnboardingResults, error) {
+	res := &onboardingResponse{}
+
+	err := s.Client.
+		PostJSON(or, prefixOnboard).
+		DecodeJSON(res).
+		Do(ctx)
+
+	if err != nil {
+		return nil, err
+	}
+
+	bkt, err := res.Bucket.toInfluxDB()
+	if err != nil {
+		return nil, err
+	}
+
+	return &influxdb.OnboardingResults{
+		Org:    &res.Organization.Organization,
+		User:   &res.User.User,
+		Auth:   res.Auth.toPlatform(),
+		Bucket: bkt,
+	}, nil
+}
+
+func (s *OnboardClientService) OnboardUser(ctx context.Context, or *influxdb.OnboardingRequest) (*influxdb.OnboardingResults, error) {
+	res := &onboardingResponse{}
+
+	err := s.Client.
+		PostJSON(or, path.Join(prefixOnboard, "user")).
+		DecodeJSON(res).
+		Do(ctx)
+
+	if err != nil {
+		return nil, err
+	}
+
+	bkt, err := res.Bucket.toInfluxDB()
+	if err != nil {
+		return nil, err
+	}
+
+	return &influxdb.OnboardingResults{
+		Org:    &res.Organization.Organization,
+		User:   &res.User.User,
+		Auth:   res.Auth.toPlatform(),
+		Bucket: bkt,
+	}, nil
+}