feat(datastore): PostgreSQL compatibility for Fleet datastore

go.mod

@@ -286,6 +286,10 @@ require (
 	github.com/huandu/xstrings v1.3.2 // indirect
 	github.com/imdario/mergo v0.3.15 // indirect
 	github.com/inconshreveable/mousetrap v1.1.0 // indirect
+	github.com/jackc/pgpassfile v1.0.0 // indirect
+	github.com/jackc/pgservicefile v0.0.0-20240606120523-5a60cdf6a761 // indirect
+	github.com/jackc/pgx/v5 v5.9.1 // indirect
+	github.com/jackc/puddle/v2 v2.2.2 // indirect
 	github.com/jbenet/go-context v0.0.0-20150711004518-d14ea06fba99 // indirect
 	github.com/joeshaw/multierror v0.0.0-20140124173710-69b34d4ec901 // indirect
 	github.com/jonboulle/clockwork v0.5.0 // indirect

go.sum

@@ -543,6 +543,14 @@ github.com/imdario/mergo v0.3.15/go.mod h1:WBLT9ZmE3lPoWsEzCh9LPo3TiwVN+ZKEjmz+h
 github.com/inconshreveable/mousetrap v1.0.0/go.mod h1:PxqpIevigyE2G7u3NXJIT2ANytuPF1OarO4DADm73n8=
 github.com/inconshreveable/mousetrap v1.1.0 h1:wN+x4NVGpMsO7ErUn/mUI3vEoE6Jt13X2s0bqwp9tc8=
 github.com/inconshreveable/mousetrap v1.1.0/go.mod h1:vpF70FUmC8bwa3OWnCshd2FqLfsEA9PFc4w1p2J65bw=
+github.com/jackc/pgpassfile v1.0.0 h1:/6Hmqy13Ss2zCq62VdNG8tM1wchn8zjSGOBJ6icpsIM=
+github.com/jackc/pgpassfile v1.0.0/go.mod h1:CEx0iS5ambNFdcRtxPj5JhEz+xB6uRky5eyVu/W2HEg=
+github.com/jackc/pgservicefile v0.0.0-20240606120523-5a60cdf6a761 h1:iCEnooe7UlwOQYpKFhBabPMi4aNAfoODPEFNiAnClxo=
+github.com/jackc/pgservicefile v0.0.0-20240606120523-5a60cdf6a761/go.mod h1:5TJZWKEWniPve33vlWYSoGYefn3gLQRzjfDlhSJ9ZKM=
+github.com/jackc/pgx/v5 v5.9.1 h1:uwrxJXBnx76nyISkhr33kQLlUqjv7et7b9FjCen/tdc=
+github.com/jackc/pgx/v5 v5.9.1/go.mod h1:mal1tBGAFfLHvZzaYh77YS/eC6IX9OWbRV1QIIM0Jn4=
+github.com/jackc/puddle/v2 v2.2.2 h1:PR8nw+E/1w0GLuRFSmiioY6UooMp6KJv0/61nB7icHo=
+github.com/jackc/puddle/v2 v2.2.2/go.mod h1:vriiEXHvEE654aYKXXjOvZM39qJ0q+azkZFrfEOc3H4=
 github.com/jbenet/go-context v0.0.0-20150711004518-d14ea06fba99 h1:BQSFePA1RWJOlocH6Fxy8MmwDt+yVQYULKfN0RoTN8A=
 github.com/jbenet/go-context v0.0.0-20150711004518-d14ea06fba99/go.mod h1:1lJo3i6rXxKeerYnT8Nvf0QmHCRC1n8sfWVwXF2Frvo=
 github.com/jessevdk/go-flags v1.4.0/go.mod h1:4FA24M0QyGHXBuZZK/XkWh8h0e1EYbRYJSGM75WSRxI=

server/config/config.go

@@ -39,6 +39,9 @@ const (
 
 // MysqlConfig defines configs related to MySQL
 type MysqlConfig struct {
+	// Driver selects the database driver. Only "mysql" is valid in Phase 1.
+	// Future values: "postgres" (Phase 4+).
+	Driver           string `yaml:"driver"`
 	Protocol         string `yaml:"protocol"`
 	Address          string `yaml:"address"`
 	Username         string `yaml:"username"`
@@ -1128,6 +1131,8 @@ func (t *TLS) ToTLSConfig() (*tls.Config, error) {
 // filled into the FleetConfig struct
 func (man Manager) addConfigs() {
 	addMysqlConfig := func(prefix, defaultAddr, usageSuffix string) {
+		man.addConfigString(prefix+".driver", "",
+			"Database driver: mysql (default) or postgres"+usageSuffix)
 		man.addConfigString(prefix+".protocol", "tcp",
 			"MySQL server communication protocol (tcp,unix,...)"+usageSuffix)
 		man.addConfigString(prefix+".address", defaultAddr,
@@ -1637,6 +1642,7 @@ func (man Manager) LoadConfig() FleetConfig {
 
 	loadMysqlConfig := func(prefix string) MysqlConfig {
 		return MysqlConfig{
+			Driver:           man.getConfigString(prefix + ".driver"),
 			Protocol:         man.getConfigString(prefix + ".protocol"),
 			Address:          man.getConfigString(prefix + ".address"),
 			Username:         man.getConfigString(prefix + ".username"),

server/datastore/mysql/activities.go

@@ -60,7 +60,7 @@ func (ds *Datastore) ListHostUpcomingActivities(ctx context.Context, hostID uint
 				'script_name', COALESCE(ses.name, scr.name, ''),
 				'script_execution_id', ua.execution_id,
 				'batch_execution_id', bahr.batch_execution_id,
-				'async', NOT ua.payload->'$.sync_request',
+				'async', COALESCE(ua.payload->>'$.sync_request', '0') != '1',
 				'policy_id', sua.policy_id,
 				'policy_name', p.name
 			) as details,
@@ -104,7 +104,7 @@ func (ds *Datastore) ListHostUpcomingActivities(ctx context.Context, hostID uint
 				'software_package', COALESCE(si.filename, ua.payload->>'$.installer_filename', ''),
 				'install_uuid', ua.execution_id,
 				'status', 'pending_install',
-				'self_service', ua.payload->'$.self_service' IS TRUE,
+				'self_service', COALESCE(ua.payload->>'$.self_service', '0') = '1',
 				'source', COALESCE(st.source, ua.payload->>'$.source'),
 				'policy_id', siua.policy_id,
 				'policy_name', p.name
@@ -146,7 +146,7 @@ func (ds *Datastore) ListHostUpcomingActivities(ctx context.Context, hostID uint
 				'software_title', COALESCE(st.name, ua.payload->>'$.software_title_name', ''),
 				'script_execution_id', ua.execution_id,
 				'status', 'pending_uninstall',
-				'self_service', COALESCE(ua.payload->'$.self_service', FALSE) IS TRUE,
+				'self_service', COALESCE(ua.payload->>'$.self_service', '0') = '1',
 				'source', COALESCE(st.source, ua.payload->>'$.source'),
 				'policy_id', siua.policy_id,
 				'policy_name', p.name
@@ -188,7 +188,7 @@ func (ds *Datastore) ListHostUpcomingActivities(ctx context.Context, hostID uint
 				'software_title', COALESCE(st.name, ''),
 				'app_store_id', vaua.adam_id,
 				'command_uuid', ua.execution_id,
-				'self_service', ua.payload->'$.self_service' IS TRUE,
+				'self_service', COALESCE(ua.payload->>'$.self_service', '0') = '1',
 				'status', 'pending_install',
 				'host_platform', h.platform
 			) AS details,
@@ -228,7 +228,7 @@ func (ds *Datastore) ListHostUpcomingActivities(ctx context.Context, hostID uint
 				'host_display_name', COALESCE(hdn.display_name, ''),
 				'software_title', COALESCE(st.name, ''),
 				'command_uuid', ua.execution_id,
-				'self_service', ua.payload->'$.self_service' IS TRUE,
+				'self_service', COALESCE(ua.payload->>'$.self_service', '0') = '1',
 				'status', 'pending_install'
 			) AS details,
 			IF(ua.activated_at IS NULL, 0, 1) as topmost,
@@ -809,10 +809,10 @@ func (ds *Datastore) GetHostUpcomingActivityMeta(ctx context.Context, hostID uin
 		ua.activated_at,
 		ua.activity_type,
 		CASE
-			WHEN hma.lock_ref = :execution_id THEN :lock_action
-			WHEN hma.unlock_ref = :execution_id THEN :unlock_action
-			WHEN hma.wipe_ref = :execution_id THEN :wipe_action
-			ELSE :none_action
+			WHEN hma.lock_ref = :execution_id THEN CAST(:lock_action AS UNSIGNED)
+			WHEN hma.unlock_ref = :execution_id THEN CAST(:unlock_action AS UNSIGNED)
+			WHEN hma.wipe_ref = :execution_id THEN CAST(:wipe_action AS UNSIGNED)
+			ELSE CAST(:none_action AS UNSIGNED)
 		END AS well_known_action
 	FROM
 		upcoming_activities ua
@@ -1070,9 +1070,9 @@ SELECT
 	sua.script_id,
 	sua.policy_id,
 	ua.user_id,
-	COALESCE(ua.payload->'$.sync_request', 0),
+	COALESCE(ua.payload->>'$.sync_request', '0') = '1',
 	sua.setup_experience_script_id,
-	COALESCE(ua.payload->'$.is_internal', 0)
+	COALESCE(ua.payload->>'$.is_internal', '0') = '1'
 FROM
 	upcoming_activities ua
 	INNER JOIN script_upcoming_activities sua
@@ -1108,7 +1108,7 @@ SELECT
 	ua.host_id,
 	siua.software_installer_id,
 	ua.user_id,
-	COALESCE(ua.payload->'$.self_service', 0),
+	COALESCE(ua.payload->>'$.self_service', '0') = '1',
 	siua.policy_id,
 	COALESCE(si.filename, ua.payload->>'$.installer_filename', '[deleted installer]'),
 	COALESCE(si.version, ua.payload->>'$.version', 'unknown'),
@@ -1120,7 +1120,7 @@ SELECT
 	-- the number of prior tries. +1 makes this the next attempt in sequence:
 	-- first install = 1, first retry = 2, second retry = 3, etc.
 	CASE
-		WHEN siua.policy_id IS NULL AND COALESCE(ua.payload->'$.with_retries', 0) = 1 THEN (
+		WHEN siua.policy_id IS NULL AND COALESCE(ua.payload->>'$.with_retries', '0') = '1' THEN (
 			SELECT COUNT(*) + 1
 			FROM host_software_installs hsi2
 			WHERE hsi2.host_id = ua.host_id
@@ -1171,7 +1171,7 @@ SELECT
 	si.uninstall_script_content_id,
 	'',
 	ua.user_id,
-	1
+	TRUE
 FROM
 	upcoming_activities ua
 	INNER JOIN software_install_upcoming_activities siua
@@ -1195,11 +1195,11 @@ SELECT
 	ua.host_id,
 	siua.software_installer_id,
 	ua.user_id,
-	1,  -- uninstall
+	TRUE,  -- uninstall
 	'', -- no installer_filename for uninstalls
 	COALESCE(si.title_id, siua.software_title_id),
 	COALESCE(st.name, ua.payload->>'$.software_title_name', '[deleted title]'),
-	COALESCE(ua.payload->>'$.self_service', FALSE),
+	COALESCE(ua.payload->>'$.self_service', '0') = '1',
 	'unknown'
 FROM
 	upcoming_activities ua
@@ -1255,7 +1255,7 @@ SELECT
 	ua.execution_id,
 	ua.user_id,
 	ua.payload->>'$.associated_event_id',
-	COALESCE(ua.payload->'$.self_service', 0),
+	COALESCE(ua.payload->>'$.self_service', '0') = '1',
 	vaua.policy_id
 FROM
 	upcoming_activities ua
@@ -1291,7 +1291,7 @@ SELECT
 	ua.execution_id,
 	ua.user_id,
 	iha.platform,
-	COALESCE(ua.payload->'$.self_service', 0)
+	COALESCE(ua.payload->>'$.self_service', '0') = '1'
 FROM
 	upcoming_activities ua
 	INNER JOIN in_house_app_upcoming_activities ihua

server/datastore/mysql/aggregated_stats.go

@@ -44,32 +44,51 @@ FROM (SELECT (@rownum := @rownum + 1) AS row_number_value, sum1.*
             GROUP BY d.host_id) as sum2) AS t2
 WHERE t1.row_number_value = FLOOR(total_rows * %[2]s) + 1`
 
+const scheduledQueryPercentileQueryPG = `
+SELECT COALESCE((t1.%[1]s_total / t1.executions_total), 0)
+FROM (SELECT ROW_NUMBER() OVER (ORDER BY (SUM(d.%[1]s) / SUM(d.executions))) AS row_number_value,
+            SUM(d.%[1]s) as %[1]s_total, SUM(d.executions) as executions_total
+      FROM scheduled_query_stats d
+      WHERE d.scheduled_query_id = ?
+        AND d.executions > 0
+      GROUP BY d.host_id) AS t1,
+     (SELECT COUNT(*) AS total_rows
+      FROM (SELECT 1
+            FROM scheduled_query_stats d
+            WHERE d.scheduled_query_id = ?
+              AND d.executions > 0
+            GROUP BY d.host_id) as sum2) AS t2
+WHERE t1.row_number_value = FLOOR(total_rows * %[2]s) + 1`
+
 const (
 	scheduledQueryTotalExecutions = `SELECT coalesce(sum(executions), 0) FROM scheduled_query_stats WHERE scheduled_query_id=?`
 )
 
-func getPercentileQuery(aggregate fleet.AggregatedStatsType, time string, percentile string) string {
+func getPercentileQuery(aggregate fleet.AggregatedStatsType, time string, percentile string, isPG bool) string {
 	switch aggregate { //nolint:gocritic // ignore singleCaseSwitch
 	case fleet.AggregatedStatsTypeScheduledQuery:
+		if isPG {
+			return fmt.Sprintf(scheduledQueryPercentileQueryPG, time, percentile)
+		}
 		return fmt.Sprintf(scheduledQueryPercentileQuery, time, percentile)
 	}
 	return ""
 }
 
 func setP50AndP95Map(
-	ctx context.Context, tx sqlx.QueryerContext, aggregate fleet.AggregatedStatsType, time string, id uint, statsMap map[string]interface{},
+	ctx context.Context, tx sqlx.QueryerContext, aggregate fleet.AggregatedStatsType, time string, id uint, statsMap map[string]any, isPG bool,
 ) error {
 	var p50, p95 float64
 
-	err := sqlx.GetContext(ctx, tx, &p50, getPercentileQuery(aggregate, time, "0.5"), id, id)
+	err := sqlx.GetContext(ctx, tx, &p50, getPercentileQuery(aggregate, time, "0.5", isPG), id, id)
 	if err != nil {
 		if err == sql.ErrNoRows {
 			return nil
 		}
 		return ctxerr.Wrapf(ctx, err, "getting %s p50 for %s %d", time, aggregate, id)
 	}
 	statsMap[time+"_p50"] = p50
-	err = sqlx.GetContext(ctx, tx, &p95, getPercentileQuery(aggregate, time, "0.95"), id, id)
+	err = sqlx.GetContext(ctx, tx, &p95, getPercentileQuery(aggregate, time, "0.95", isPG), id, id)
 	if err != nil {
 		if err == sql.ErrNoRows {
 			return nil
@@ -99,14 +118,15 @@ func (ds *Datastore) CalculateAggregatedPerfStatsPercentiles(ctx context.Context
 	// We are using the reader because the below SELECT queries are expensive, and we don't want to impact the performance of the writer.
 	reader := ds.reader(ctx)
 	var totalExecutions int
-	statsMap := make(map[string]interface{})
+	statsMap := make(map[string]any)
 
 	// many queries is not ideal, but getting both values and totals in the same query was a bit more complicated
 	// so I went for the simpler approach first, we can optimize later
-	if err := setP50AndP95Map(ctx, reader, aggregate, "user_time", queryID, statsMap); err != nil {
+	_, isPG := ds.dialect.(postgresDialect)
+	if err := setP50AndP95Map(ctx, reader, aggregate, "user_time", queryID, statsMap, isPG); err != nil {
 		return err
 	}
-	if err := setP50AndP95Map(ctx, reader, aggregate, "system_time", queryID, statsMap); err != nil {
+	if err := setP50AndP95Map(ctx, reader, aggregate, "system_time", queryID, statsMap, isPG); err != nil {
 		return err
 	}
 
@@ -128,9 +148,8 @@ func (ds *Datastore) CalculateAggregatedPerfStatsPercentiles(ctx context.Context
 		ctx,
 		`
 		INSERT INTO aggregated_stats(id, type, global_stats, json_value)
-		VALUES (?, ?, 0, ?)
-		ON DUPLICATE KEY UPDATE json_value=VALUES(json_value)
-		`,
+		VALUES (?, ?, false, ?)
+		`+ds.dialect.OnDuplicateKey("id,type,global_stats", `json_value=VALUES(json_value)`),
 		queryID, aggregate, statsJson,
 	)
 	if err != nil {