Merge branch '486-se-platform-token' into 'master'

feat(engine): enable SE billing (#486)

Closes #486

See merge request postgres-ai/database-lab!693

Author: agneum

engine/cmd/database-lab/main.go

@@ -13,15 +13,14 @@ import (
 	"fmt"
 	"os"
 	"os/signal"
-	"runtime"
 	"strings"
 	"syscall"
 	"time"
 
 	"github.com/docker/docker/client"
-	"github.com/pbnjay/memory"
 	"github.com/pkg/errors"
 
+	"gitlab.com/postgres-ai/database-lab/v3/internal/billing"
 	"gitlab.com/postgres-ai/database-lab/v3/internal/cloning"
 	"gitlab.com/postgres-ai/database-lab/v3/internal/diagnostic"
 	"gitlab.com/postgres-ai/database-lab/v3/internal/embeddedui"
@@ -56,7 +55,7 @@ func main() {
 	}
 
 	logFilter := log.GetFilter()
-	logFilter.ReloadLogRegExp([]string{cfg.Server.VerificationToken, cfg.Platform.AccessToken})
+	logFilter.ReloadLogRegExp([]string{cfg.Server.VerificationToken, cfg.Platform.AccessToken, cfg.Platform.OrgKey})
 
 	config.ApplyGlobals(cfg)
 
@@ -83,6 +82,13 @@ func main() {
 	log.Msg("Database Lab Instance ID:", engProps.InstanceID)
 	log.Msg("Database Lab Engine version:", version.GetVersion())
 
+	// Create a platform service to make requests to Platform.
+	platformSvc, err := platform.New(ctx, cfg.Platform, engProps.InstanceID)
+	if err != nil {
+		log.Errf(err.Error())
+		return
+	}
+
 	if cfg.Server.VerificationToken == "" {
 		log.Warn("Verification Token is empty. Database Lab Engine is insecure")
 	}
@@ -97,33 +103,22 @@ func main() {
 
 	defer networks.Stop(docker, internalNetworkID, engProps.ContainerName)
 
-	// Create a platform service to make requests to Platform.
-	platformSvc, err := platform.New(ctx, cfg.Platform)
-	if err != nil {
-		log.Errf(errors.WithMessage(err, "failed to create a new platform service").Error())
-		return
-	}
-
 	dbCfg := &resources.DB{
 		Username: cfg.Global.Database.User(),
 		DBName:   cfg.Global.Database.Name(),
 	}
 
-	tm, err := telemetry.New(cfg.Global, engProps)
-	if err != nil {
-		log.Errf(errors.WithMessage(err, "failed to initialize a telemetry service").Error())
-		return
-	}
+	tm := telemetry.New(platformSvc, engProps.InstanceID)
 
 	pm := pool.NewPoolManager(&cfg.PoolManager, runner)
 	if err = pm.ReloadPools(); err != nil {
 		log.Err(err.Error())
 	}
 
 	// Create a new retrieval service to prepare a data directory and start snapshotting.
-	retrievalSvc, err := retrieval.New(cfg, engProps, docker, pm, tm, runner)
+	retrievalSvc, err := retrieval.New(cfg, &engProps, docker, pm, tm, runner)
 	if err != nil {
-		log.Errf(errors.WithMessage(err, `error in the "retrieval" section of the config`).Error())
+		log.Errf(errors.WithMessage(err, `error in the "retrieval" section of config`).Error())
 		return
 	}
 
@@ -160,27 +155,34 @@ func main() {
 
 	go removeObservingClones(observingChan, obs)
 
+	systemMetrics := billing.GetSystemMetrics(pm)
+
 	tm.SendEvent(ctx, telemetry.EngineStartedEvent, telemetry.EngineStarted{
 		EngineVersion: version.GetVersion(),
 		DBEngine:      cfg.Global.Engine,
 		DBVersion:     provisioner.DetectDBVersion(),
 		Pools:         pm.CollectPoolStat(),
 		Restore:       retrievalSvc.ReportState(),
-		System: telemetry.System{
-			CPU:         runtime.NumCPU(),
-			TotalMemory: memory.TotalMemory(),
-		},
+		System:        systemMetrics,
 	})
 
+	billingSvc := billing.New(platformSvc.Client, &engProps, pm)
+
+	if err := billingSvc.RegisterInstance(ctx, systemMetrics); err != nil {
+		log.Msg("Skip registering instance:", err)
+	}
+
+	log.Msg("DLE Edition:", engProps.GetEdition())
+
 	embeddedUI := embeddedui.New(cfg.EmbeddedUI, engProps, runner, docker)
 
 	logCleaner := diagnostic.NewLogCleaner()
 
 	reloadConfigFn := func(server *srv.Server) error {
 		return reloadConfig(
 			ctx,
+			engProps,
 			provisioner,
-			tm,
 			retrievalSvc,
 			pm,
 			cloningSvc,
@@ -192,11 +194,13 @@ func main() {
 		)
 	}
 
-	server := srv.NewServer(&cfg.Server, &cfg.Global, engProps, docker, cloningSvc, provisioner, retrievalSvc, platformSvc,
-		obs, pm, tm, tokenHolder, logFilter, embeddedUI, reloadConfigFn)
+	server := srv.NewServer(&cfg.Server, &cfg.Global, &engProps, docker, cloningSvc, provisioner, retrievalSvc, platformSvc,
+		billingSvc, obs, pm, tm, tokenHolder, logFilter, embeddedUI, reloadConfigFn)
 	shutdownCh := setShutdownListener()
 
-	go setReloadListener(ctx, provisioner, tm, retrievalSvc, pm, cloningSvc, platformSvc, embeddedUI, server,
+	go setReloadListener(ctx, engProps, provisioner,
+		retrievalSvc, pm, cloningSvc, platformSvc,
+		embeddedUI, server,
 		logCleaner, logFilter)
 
 	server.InitHandlers()
@@ -207,6 +211,8 @@ func main() {
 		}
 	}()
 
+	go billingSvc.CollectUsage(ctx, systemMetrics)
+
 	if cfg.EmbeddedUI.Enabled {
 		go func() {
 			if err := embeddedUI.Run(ctx); err != nil {
@@ -245,13 +251,13 @@ func main() {
 	tm.SendEvent(ctxBackground, telemetry.EngineStoppedEvent, telemetry.EngineStopped{Uptime: server.Uptime()})
 }
 
-func getEngineProperties(ctx context.Context, dockerCLI *client.Client, cfg *config.Config) (global.EngineProps, error) {
+func getEngineProperties(ctx context.Context, docker *client.Client, cfg *config.Config) (global.EngineProps, error) {
 	hostname := os.Getenv("HOSTNAME")
 	if hostname == "" {
 		return global.EngineProps{}, errors.New("hostname is empty")
 	}
 
-	dleContainer, err := dockerCLI.ContainerInspect(ctx, hostname)
+	dleContainer, err := docker.ContainerInspect(ctx, hostname)
 	if err != nil {
 		return global.EngineProps{}, fmt.Errorf("failed to inspect DLE container: %w", err)
 	}
@@ -276,15 +282,15 @@ func getEngineProperties(ctx context.Context, dockerCLI *client.Client, cfg *con
 	return engProps, nil
 }
 
-func reloadConfig(ctx context.Context, provisionSvc *provision.Provisioner, tm *telemetry.Agent,
+func reloadConfig(ctx context.Context, engProp global.EngineProps, provisionSvc *provision.Provisioner,
 	retrievalSvc *retrieval.Retrieval, pm *pool.Manager, cloningSvc *cloning.Base, platformSvc *platform.Service,
 	embeddedUI *embeddedui.UIManager, server *srv.Server, cleaner *diagnostic.Cleaner, filtering *log.Filtering) error {
 	cfg, err := config.LoadConfiguration()
 	if err != nil {
 		return err
 	}
 
-	filtering.ReloadLogRegExp([]string{cfg.Server.VerificationToken, cfg.Platform.AccessToken})
+	filtering.ReloadLogRegExp([]string{cfg.Server.VerificationToken, cfg.Platform.AccessToken, cfg.Platform.OrgKey})
 	config.ApplyGlobals(cfg)
 
 	if err := provision.IsValidConfig(cfg.Provision); err != nil {
@@ -296,7 +302,7 @@ func reloadConfig(ctx context.Context, provisionSvc *provision.Provisioner, tm *
 		return err
 	}
 
-	newPlatformSvc, err := platform.New(ctx, cfg.Platform)
+	newPlatformSvc, err := platform.New(ctx, cfg.Platform, engProp.InstanceID)
 	if err != nil {
 		return err
 	}
@@ -319,7 +325,6 @@ func reloadConfig(ctx context.Context, provisionSvc *provision.Provisioner, tm *
 	}
 
 	provisionSvc.Reload(cfg.Provision, dbCfg)
-	tm.Reload(cfg.Global)
 	retrievalSvc.Reload(ctx, newRetrievalConfig)
 	cloningSvc.Reload(cfg.Cloning)
 	platformSvc.Reload(newPlatformSvc)
@@ -328,7 +333,7 @@ func reloadConfig(ctx context.Context, provisionSvc *provision.Provisioner, tm *
 	return nil
 }
 
-func setReloadListener(ctx context.Context, provisionSvc *provision.Provisioner, tm *telemetry.Agent,
+func setReloadListener(ctx context.Context, engProp global.EngineProps, provisionSvc *provision.Provisioner,
 	retrievalSvc *retrieval.Retrieval, pm *pool.Manager, cloningSvc *cloning.Base, platformSvc *platform.Service,
 	embeddedUI *embeddedui.UIManager, server *srv.Server, cleaner *diagnostic.Cleaner, logFilter *log.Filtering) {
 	reloadCh := make(chan os.Signal, 1)
@@ -337,7 +342,11 @@ func setReloadListener(ctx context.Context, provisionSvc *provision.Provisioner,
 	for range reloadCh {
 		log.Msg("Reloading configuration")
 
-		if err := reloadConfig(ctx, provisionSvc, tm, retrievalSvc, pm, cloningSvc, platformSvc, embeddedUI, server,
+		if err := reloadConfig(ctx, engProp,
+			provisionSvc, retrievalSvc,
+			pm, cloningSvc,
+			platformSvc,
+			embeddedUI, server,
 			cleaner, logFilter); err != nil {
 			log.Err("Failed to reload configuration", err)
 		}

engine/cmd/runci/main.go

@@ -65,7 +65,8 @@ func main() {
 	}
 
 	// Create a platform service to make requests to Platform.
-	platformSvc, err := platform.New(ctx, cfg.Platform)
+	// Instance ID is not defined for the Run CI service.
+	platformSvc, err := platform.New(ctx, cfg.Platform, "")
 	if err != nil {
 		log.Errf(errors.WithMessage(err, "failed to create a new platform service").Error())
 		return

engine/configs/config.example.logical_generic.yml

@@ -20,6 +20,7 @@ server:
   # In this case, the DLE API and the UI application will not require any credentials.
   verificationToken: "secret_token"
 
+
   # HTTP server port. Default: 2345.
   port: 2345
 
@@ -58,15 +59,6 @@ global:
     # Default database name.
     dbname: postgres
 
-  # Telemetry: anonymous statistics sent to Postgres.ai.
-  # Used to analyze DLE usage, it helps the DLE maintainers make decisions on product development.
-  # Please leave it enabled if possible – this will contribute to DLE development.
-  # The full list of data points being collected: https://postgres.ai/docs/database-lab/telemetry
-  telemetry:
-    enabled: true
-    # Telemetry API URL. To send anonymous telemetry data, keep it default ("https://postgres.ai/api/general").
-    url: "https://postgres.ai/api/general"
-
 # Manages filesystem pools (in the case of ZFS) or volume groups.
 poolManager:
   # The full path which contains the pool mount directories. mountDir can contain multiple pool directories.
@@ -367,10 +359,21 @@ diagnostic:
 # Postgres.ai Platform integration (provides GUI) – extends the open source offering.
 # Uncomment the following lines if you need GUI, personal tokens, audit logs, more.
 #
-#platform:
-#  # Platform API URL. To work with Postgres.ai SaaS, keep it default
-#  # ("https://postgres.ai/api/general").
-#  url: "https://postgres.ai/api/general"
+platform:
+  # Platform API URL. To work with Postgres.ai SaaS, keep it default
+  # ("https://postgres.ai/api/general").
+  url: "https://postgres.ai/api/general"
+  # Telemetry: anonymous statistics sent to Postgres.ai.
+  # Used to analyze DLE usage, it helps the DLE maintainers make decisions on product development.
+  # Please leave it enabled if possible – this will contribute to DLE development.
+  # The full list of data points being collected: https://postgres.ai/docs/database-lab/telemetry
+  enableTelemetry: true
+#
+#  # Project name
+#  projectName: "project_name"
+#
+#  # Organization key
+#  orgKey: "org_key"
 #
 #  # Token for authorization in Platform API. This token can be obtained on
 #  # the Postgres.ai Console: https://postgres.ai/console/YOUR_ORG_NAME/tokens

engine/configs/config.example.logical_rds_iam.yml

@@ -58,15 +58,6 @@ global:
     # Default database name.
     dbname: postgres
 
-  # Telemetry: anonymous statistics sent to Postgres.ai.
-  # Used to analyze DLE usage, it helps the DLE maintainers make decisions on product development.
-  # Please leave it enabled if possible – this will contribute to DLE development.
-  # The full list of data points being collected: https://postgres.ai/docs/database-lab/telemetry
-  telemetry:
-    enabled: true
-    # Telemetry API URL. To send anonymous telemetry data, keep it default ("https://postgres.ai/api/general").
-    url: "https://postgres.ai/api/general"
-
 # Manages filesystem pools (in the case of ZFS) or volume groups.
 poolManager:
   # The full path which contains the pool mount directories. mountDir can contain multiple pool directories.
@@ -368,19 +359,30 @@ diagnostic:
 # Postgres.ai Platform integration (provides GUI) – extends the open source offering.
 # Uncomment the following lines if you need GUI, personal tokens, audit logs, more.
 #
-#platform:
-#  # Platform API URL. To work with Postgres.ai SaaS, keep it default
-#  # ("https://postgres.ai/api/general").
-#  url: "https://postgres.ai/api/general"
+platform:
+  # Platform API URL. To work with Postgres.ai SaaS, keep it default
+  # ("https://postgres.ai/api/general").
+  url: "https://postgres.ai/api/general"
+  # Telemetry: anonymous statistics sent to Postgres.ai.
+  # Used to analyze DLE usage, it helps the DLE maintainers make decisions on product development.
+  # Please leave it enabled if possible – this will contribute to DLE development.
+  # The full list of data points being collected: https://postgres.ai/docs/database-lab/telemetry
+  enableTelemetry: true
+#
+#  # Project name
+#  projectName: "project_name"
+#
+#  # Organization key
+#  orgKey: "org_key"
 #
 #  # Token for authorization in Platform API. This token can be obtained on
 #  # the Postgres.ai Console: https://postgres.ai/console/YOUR_ORG_NAME/tokens
 #  # This token needs to be kept in secret, known only to the administrator.
 #  accessToken: "platform_access_token"
 #
 #  # Enable authorization with personal tokens of the organization's members.
-#  # If false: all users must use "accessToken" value for any API request
-#  # If true: "accessToken" is known only to admin, users use their own tokens,
+#  # If false: all users must use "verificationToken" value for any API request
+#  # If true: "verificationToken" is known only to admin, users use their own tokens,
 #  #          and any token can be revoked not affecting others
 #  enablePersonalTokens: true
 #