PMM-14567: Add MongoDB Realtime Agent Type (Option 3)

managed/models/agent_helpers.go

@@ -340,6 +340,7 @@
 		agentTypes = []AgentType{
 			MongoDBExporterType,
 			QANMongoDBProfilerAgentType,
+			MongoDBRealtimeAgentType,
 		}
 	case ExternalServiceType, HAProxyServiceType, ProxySQLServiceType:
 		fallthrough
@@ -772,6 +773,7 @@
 	Disabled          bool
 	ExporterOptions   ExporterOptions
 	QANOptions        QANOptions
+	RTAOptions        RTAOptions
 	AWSOptions        AWSOptions
 	AzureOptions      AzureOptions
 	MongoDBOptions    MongoDBOptions
@@ -911,6 +913,7 @@
 		TLSSkipVerify:     params.TLSSkipVerify,
 		ExporterOptions:   params.ExporterOptions,
 		QANOptions:        params.QANOptions,
+		RTAOptions:        params.RTAOptions,
 		AWSOptions:        params.AWSOptions,
 		AzureOptions:      params.AzureOptions,
 		MongoDBOptions:    params.MongoDBOptions,
@@ -1097,3 +1100,58 @@
 func IsPushMetricsSupported(pmmAgentVersion *string) bool {
 	return true
 }
+
+// CreateMongoDBRealtimeAgent creates a MongoDB Realtime Analytics agent.
+// It retrieves credentials from existing MongoDB agents for the service as per Option 3 requirements.
+func CreateMongoDBRealtimeAgent(q *reform.Querier, pmmAgentID, serviceID string, customLabels map[string]string, disabled bool) (*Agent, error) {
+	// Verify service exists and is MongoDB type
+	service, err := FindServiceByID(q, serviceID)
+	if err != nil {
+		return nil, err
+	}
+	if service.ServiceType != MongoDBServiceType {
+		return nil, status.Errorf(codes.InvalidArgument, "Service must be MongoDB type, got %s", service.ServiceType)
+	}
+
+	// Retrieve credentials from existing MongoDB agents for this service
+	// Try to find credentials from QAN or exporter agents
+	agentTypes := []AgentType{
+		QANMongoDBProfilerAgentType,
+		QANMongoDBMongologAgentType,
+		MongoDBExporterType,
+	}
+
+	var existingAgent *Agent
+	for _, agentType := range agentTypes {
+		agents, err := FindAgents(q, AgentFilters{
+			ServiceID: serviceID,
+			AgentType: &agentType,
+		})
+		if err != nil {
+			return nil, err
+		}
+		if len(agents) > 0 {
+			existingAgent = agents[0]
+			break
+		}
+	}
+
+	if existingAgent == nil {
+		return nil, status.Errorf(codes.FailedPrecondition, "No existing MongoDB agent found for service %s to retrieve credentials", serviceID)
+	}
+
+	// Create the MongoDB realtime agent with credentials from existing agent
+	params := &CreateAgentParams{
+		PMMAgentID:     pmmAgentID,
+		ServiceID:      serviceID,
+		Username:       pointer.GetString(existingAgent.Username),
+		Password:       pointer.GetString(existingAgent.Password),
+		CustomLabels:   customLabels,
+		TLS:            existingAgent.TLS,
+		TLSSkipVerify:  existingAgent.TLSSkipVerify,
+		MongoDBOptions: existingAgent.MongoDBOptions,
+		Disabled:       disabled,
+	}
+
+	return CreateAgent(q, MongoDBRealtimeAgentType, params)
+}

managed/models/agent_model.go

@@ -77,6 +77,7 @@ const (
 	VMAgentType                         AgentType = "vmagent"
 	NomadAgentType                      AgentType = "nomad-agent"
 	ValkeyExporterType                  AgentType = "valkey_exporter"
+	MongoDBRealtimeAgentType            AgentType = "mongodb-realtime-agent"
 )
 
 var v2_42 = version.MustParse("2.42.0-0")
@@ -291,6 +292,21 @@ func (c ValkeyOptions) IsEmpty() bool {
 		c.SSLKey == ""
 }
 
+// RTAOptions represents structure for Real-Time Analytics options.
+// Currently empty but reserved for future RTA-specific configuration.
+type RTAOptions struct{}
+
+// Value implements database/sql/driver.Valuer interface. Should be defined on the value.
+func (c RTAOptions) Value() (driver.Value, error) { return jsonValue(c) }
+
+// Scan implements database/sql.Scanner interface. Should be defined on the pointer.
+func (c *RTAOptions) Scan(src interface{}) error { return jsonScan(c, src) }
+
+// IsEmpty returns true if all RTAOptions fields are unset or have zero values, otherwise returns false.
+func (c RTAOptions) IsEmpty() bool {
+	return true
+}
+
 // Agent represents Agent as stored in database.
 //
 //reform:agents
@@ -321,6 +337,7 @@ type Agent struct {
 
 	ExporterOptions ExporterOptions `reform:"exporter_options"`
 	QANOptions      QANOptions      `reform:"qan_options"`
+	RTAOptions      RTAOptions      `reform:"rta_options"`
 
 	AWSOptions        AWSOptions        `reform:"aws_options"`
 	AzureOptions      AzureOptions      `reform:"azure_options"`
@@ -568,7 +585,7 @@ func (s *Agent) DSN(service *Service, dsnParams DSNParams, tdp *DelimiterPair, p
 
 		return cfg.FormatDSN()
 
-	case QANMongoDBProfilerAgentType, QANMongoDBMongologAgentType, MongoDBExporterType:
+	case QANMongoDBProfilerAgentType, QANMongoDBMongologAgentType, MongoDBExporterType, MongoDBRealtimeAgentType:
 		q := make(url.Values)
 		if dsnParams.DialTimeout != 0 {
 			q.Set("connectTimeoutMS", strconv.Itoa(int(dsnParams.DialTimeout/time.Millisecond)))
@@ -811,7 +828,7 @@ func (s Agent) Files() map[string]string {
 		return nil
 	case ProxySQLExporterType:
 		return nil
-	case QANMongoDBProfilerAgentType, QANMongoDBMongologAgentType, MongoDBExporterType:
+	case QANMongoDBProfilerAgentType, QANMongoDBMongologAgentType, MongoDBExporterType, MongoDBRealtimeAgentType:
 		files := make(map[string]string)
 		if s.MongoDBOptions.TLSCa != "" {
 			files[caFilePlaceholder] = s.MongoDBOptions.TLSCa

managed/models/database.go

@@ -1165,6 +1165,10 @@ var databaseSchema = [][]string{
 			WHERE settings->'updates' IS NULL
 			OR settings->'updates'->'snooze_duration' IS NULL`,
 	},
+	114: {
+		`ALTER TABLE agents ADD COLUMN rta_options JSONB`,
+		`UPDATE agents SET rta_options = '{}'::jsonb`,
+	},
 }
 
 // ^^^ Avoid default values in schema definition. ^^^

managed/models/dsn_helpers.go

@@ -71,7 +71,8 @@ func FindDSNByServiceIDandPMMAgentID(q *reform.Querier, serviceID, pmmAgentID, d
 			agentTypes,
 			QANMongoDBProfilerAgentType,
 			QANMongoDBMongologAgentType,
-			MongoDBExporterType)
+			MongoDBExporterType,
+			MongoDBRealtimeAgentType)
 	default:
 		return "", nil, status.Errorf(codes.FailedPrecondition, "Couldn't resolve dsn, as service is unsupported")
 	}