cluster: add new option to set VM as a remote writer

embed/templates/config/datasource.yml.tpl

@@ -1,12 +1,14 @@
 apiVersion: 1
 datasources:
-  - name: {{.ClusterName}}
-    type: prometheus
+{{- range .Datasources}}
+  - name: {{.Name}}
+    type: {{.Type}}
     access: proxy
     url: {{.URL}}
     withCredentials: false
-    isDefault: false
+    isDefault: {{.IsDefault}}
     tlsAuth: false
     tlsAuthWithCACert: false
     version: 1
-    editable: true
+    editable: true
+{{- end}}

embed/templates/config/prometheus.yml.tpl

@@ -474,4 +474,4 @@ scrape_configs:
 
 {{- if .RemoteConfig}}
 {{.RemoteConfig}}
-{{- end}}
+{{- end}}

pkg/cluster/spec/grafana.go

@@ -14,16 +14,21 @@
 package spec
 
 import (
+	"bytes"
 	"context"
 	"crypto/tls"
 	"fmt"
 	"os"
+	"path"
 	"path/filepath"
 	"reflect"
 	"strings"
 	"time"
 
+	"text/template"
+
 	"github.com/pingcap/errors"
+	"github.com/pingcap/tiup/embed"
 	"github.com/pingcap/tiup/pkg/cluster/ctxt"
 	"github.com/pingcap/tiup/pkg/cluster/template/config"
 	"github.com/pingcap/tiup/pkg/cluster/template/scripts"
@@ -273,14 +278,56 @@ func (i *GrafanaInstance) InitConfig(
 	if len(monitors) == 0 {
 		return errors.New("no prometheus found in topology")
 	}
+
+	// Create datasources configuration
+	datasources := make([]*config.DatasourceConfig, 0)
+
+	// Add default Prometheus datasource
+	defaultDatasource := config.NewDatasourceConfig(
+		clusterName,
+		// not support tls
+		fmt.Sprintf("http://%s", utils.JoinHostPort(monitors[0].Host, monitors[0].Port)),
+	)
+	datasources = append(datasources, defaultDatasource)
+
+	// Add VM datasource if enabled
+	if monitors[0].EnableVMRemoteWrite {
+		vmDatasource := config.NewDatasourceConfig(
+			fmt.Sprintf("%s-vm", clusterName),
+			// not support tls
+			fmt.Sprintf("http://%s", utils.JoinHostPort(monitors[0].Host, monitors[0].NgPort)),
+		).WithIsDefault(false)
+		datasources = append(datasources, vmDatasource)
+	}
+
+	// Write datasources configuration
 	fp = filepath.Join(paths.Cache, fmt.Sprintf("datasource_%s.yml", i.GetHost()))
-	datasourceCfg := &config.DatasourceConfig{
-		ClusterName: clusterName,
-		URL:         fmt.Sprintf("http://%s", utils.JoinHostPort(monitors[0].Host, monitors[0].Port)),
+	content := bytes.NewBuffer(nil)
+
+	// Create a map to hold all datasources
+	datasourceMap := map[string]any{
+		"Datasources": datasources,
+	}
+
+	// Create a template for the datasource configuration
+	tpl, err := embed.ReadTemplate(path.Join("templates", "config", "datasource.yml.tpl"))
+	if err != nil {
+		return err
+	}
+
+	tmpl, err := template.New("Datasource").Parse(string(tpl))
+	if err != nil {
+		return err
 	}
-	if err := datasourceCfg.ConfigToFile(fp); err != nil {
+
+	if err := tmpl.Execute(content, datasourceMap); err != nil {
 		return err
 	}
+
+	if err := utils.WriteFile(fp, content.Bytes(), 0644); err != nil {
+		return err
+	}
+
 	dst = filepath.Join(paths.Deploy, "provisioning", "datasources", "datasource.yml")
 	return i.TransferLocalConfigFile(ctx, e, fp, dst)
 }

pkg/cluster/spec/grafana_test.go

@@ -21,13 +21,15 @@ import (
 	"path"
 	"path/filepath"
 	"testing"
+	"time"
 
 	"github.com/google/uuid"
 	"github.com/pingcap/tiup/pkg/cluster/ctxt"
 	"github.com/pingcap/tiup/pkg/cluster/executor"
 	logprinter "github.com/pingcap/tiup/pkg/logger/printer"
 	"github.com/pingcap/tiup/pkg/meta"
 	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
 )
 
 func TestLocalDashboards(t *testing.T) {
@@ -157,3 +159,90 @@ level = warning
 
 	assert.Equal(t, expected, string(result))
 }
+
+type mockExecutor struct{}
+
+func (e *mockExecutor) Execute(ctx context.Context, cmd string, sudo bool, timeouts ...time.Duration) (stdout []byte, stderr []byte, err error) {
+	return nil, nil, nil
+}
+
+func (e *mockExecutor) Transfer(ctx context.Context, src, dst string, download bool, limit int, compress bool) error {
+	// Copy the file for testing
+	if !download {
+		err := os.MkdirAll(filepath.Dir(dst), 0755)
+		if err != nil {
+			return err
+		}
+		content, err := os.ReadFile(src)
+		if err != nil {
+			return err
+		}
+		return os.WriteFile(dst, content, 0644)
+	}
+	return nil
+}
+
+func TestGrafanaDatasourceConfig(t *testing.T) {
+	ctx := context.Background()
+	deployDir := t.TempDir()
+	cacheDir := t.TempDir()
+
+	// Create paths structure
+	paths := meta.DirPaths{
+		Deploy: deployDir,
+		Cache:  cacheDir,
+	}
+
+	// Create test topology
+	topo := new(Specification)
+	topo.Monitors = []*PrometheusSpec{
+		{
+			Host:                "127.0.0.1",
+			Port:                9090,
+			EnableVMRemoteWrite: true,
+		},
+	}
+	topo.Grafanas = []*GrafanaSpec{
+		{
+			Host: "127.0.0.1",
+			Port: 3000,
+		},
+	}
+
+	// Create mock executor
+	mockExec := &mockExecutor{}
+
+	// Create Grafana component
+	comp := GrafanaComponent{topo}
+	grafanaInstance := comp.Instances()[0].(*GrafanaInstance)
+
+	// Test datasource configuration
+	clusterName := "test-cluster"
+	err := grafanaInstance.InitConfig(ctxt.New(ctx, 0, logprinter.NewLogger("")), mockExec, clusterName, "v5.4.0", "tidb", paths)
+	require.NoError(t, err)
+
+	// Verify the datasource configuration file
+	dsContent, err := os.ReadFile(filepath.Join(deployDir, "provisioning", "datasources", "datasource.yml"))
+	require.NoError(t, err)
+
+	// Check if the content contains both Prometheus and VM datasources
+	assert.Contains(t, string(dsContent), fmt.Sprintf("name: %s", clusterName))
+	assert.Contains(t, string(dsContent), fmt.Sprintf("name: %s-vm", clusterName))
+	assert.Contains(t, string(dsContent), "type: prometheus")
+	assert.Contains(t, string(dsContent), "url: http://127.0.0.1:9090")
+
+	// Test without VM remote write enabled
+	topo.Monitors[0].EnableVMRemoteWrite = false
+	err = grafanaInstance.InitConfig(ctxt.New(ctx, 0, logprinter.NewLogger("")), mockExec, clusterName, "v5.4.0", "tidb", paths)
+	require.NoError(t, err)
+
+	// Verify the datasource configuration file again
+	dsContent, err = os.ReadFile(filepath.Join(deployDir, "provisioning", "datasources", "datasource.yml"))
+	require.NoError(t, err)
+
+	// Check if the content contains only Prometheus datasource
+	assert.Contains(t, string(dsContent), fmt.Sprintf("name: %s", clusterName))
+	assert.NotContains(t, string(dsContent), fmt.Sprintf("name: %s-vm", clusterName))
+	assert.Contains(t, string(dsContent), "type: prometheus")
+	assert.Contains(t, string(dsContent), "url: http://127.0.0.1:9090")
+}

pkg/cluster/spec/monitoring.go

@@ -44,7 +44,8 @@ type PrometheusSpec struct {
 	Patched               bool                   `yaml:"patched,omitempty"`
 	IgnoreExporter        bool                   `yaml:"ignore_exporter,omitempty"`
 	Port                  int                    `yaml:"port" default:"9090"`
-	NgPort                int                    `yaml:"ng_port,omitempty" validate:"ng_port:editable"` // ng_port is usable since v5.3.0 and default as 12020 since v5.4.0, so the default value is set in spec.go/AdjustByVersion
+	NgPort                int                    `yaml:"ng_port,omitempty" validate:"ng_port:editable"`                               // ng_port is usable since v5.3.0 and default as 12020 since v5.4.0, so the default value is set in spec.go/AdjustByVersion
+	EnableVMRemoteWrite   bool                   `yaml:"enable_vm_remote_write,omitempty" validate:"enable_vm_remote_write:editable"` // Enable remote write to ng-monitoring
 	DeployDir             string                 `yaml:"deploy_dir,omitempty"`
 	DataDir               string                 `yaml:"data_dir,omitempty"`
 	LogDir                string                 `yaml:"log_dir,omitempty"`
@@ -193,6 +194,38 @@ type MonitorInstance struct {
 	topo Topology
 }
 
+// handleRemoteWrite handles remote write configuration for NG monitoring
+func (i *MonitorInstance) handleRemoteWrite(spec *PrometheusSpec, monitoring *PrometheusSpec) {
+	if !spec.EnableVMRemoteWrite || monitoring.NgPort <= 0 {
+		return
+	}
+
+	// monitor do not support tls for itself
+	remoteWriteURL := fmt.Sprintf("http://%s/api/v1/write", utils.JoinHostPort(monitoring.Host, monitoring.NgPort))
+
+	// Check if this URL already exists in remote write configs
+	urlExists := false
+	if spec.RemoteConfig.RemoteWrite != nil {
+		for _, rw := range spec.RemoteConfig.RemoteWrite {
+			if url, ok := rw["url"].(string); ok && url == remoteWriteURL {
+				urlExists = true
+				break
+			}
+		}
+	}
+
+	if !urlExists {
+		remoteWrite := map[string]any{
+			"url": remoteWriteURL,
+		}
+		if spec.RemoteConfig.RemoteWrite == nil {
+			spec.RemoteConfig.RemoteWrite = []map[string]any{remoteWrite}
+		} else {
+			spec.RemoteConfig.RemoteWrite = append(spec.RemoteConfig.RemoteWrite, remoteWrite)
+		}
+	}
+}
+
 // InitConfig implement Instance interface
 func (i *MonitorInstance) InitConfig(
 	ctx context.Context,
@@ -333,8 +366,8 @@ func (i *MonitorInstance) InitConfig(
 		}
 	}
 	if servers, found := topoHasField("Monitors"); found {
-		for i := 0; i < servers.Len(); i++ {
-			monitoring := servers.Index(i).Interface().(*PrometheusSpec)
+		for idx := 0; idx < servers.Len(); idx++ {
+			monitoring := servers.Index(idx).Interface().(*PrometheusSpec)
 			uniqueHosts.Insert(monitoring.Host)
 		}
 	}
@@ -378,12 +411,6 @@ func (i *MonitorInstance) InitConfig(
 		}
 	}
 
-	remoteCfg, err := encodeRemoteCfg2Yaml(spec.RemoteConfig)
-	if err != nil {
-		return err
-	}
-	cfig.SetRemoteConfig(string(remoteCfg))
-
 	// doesn't work
 	if _, err := i.setTLSConfig(ctx, false, nil, paths); err != nil {
 		return err
@@ -434,11 +461,13 @@ func (i *MonitorInstance) InitConfig(
 		}
 
 		if servers, found := topoHasField("Monitors"); found {
-			for i := 0; i < servers.Len(); i++ {
-				monitoring := servers.Index(i).Interface().(*PrometheusSpec)
+			for idx := 0; idx < servers.Len(); idx++ {
+				monitoring := servers.Index(idx).Interface().(*PrometheusSpec)
 				cfig.AddNGMonitoring(monitoring.Host, uint64(monitoring.NgPort))
+				i.handleRemoteWrite(spec, monitoring)
 			}
 		}
+
 		fp = filepath.Join(paths.Cache, fmt.Sprintf("ngmonitoring_%s_%d.toml", i.GetHost(), i.GetPort()))
 		if err := ngcfg.ConfigToFile(fp); err != nil {
 			return err
@@ -448,6 +477,12 @@ func (i *MonitorInstance) InitConfig(
 			return err
 		}
 	}
+	// set remote config
+	remoteCfg, err := encodeRemoteCfg2Yaml(spec.RemoteConfig)
+	if err != nil {
+		return err
+	}
+	cfig.SetRemoteConfig(string(remoteCfg))
 
 	fp = filepath.Join(paths.Cache, fmt.Sprintf("prometheus_%s_%d.yml", i.GetHost(), i.GetPort()))
 	if err := cfig.ConfigToFile(fp); err != nil {

pkg/cluster/template/config/datasource.go

@@ -24,27 +24,68 @@ import (
 
 // DatasourceConfig represent the data to generate Datasource config
 type DatasourceConfig struct {
-	ClusterName string
-	URL         string
+	Name      string
+	Type      string
+	URL       string
+	IsDefault bool
+}
+
+// NewDatasourceConfig returns a DatasourceConfig
+func NewDatasourceConfig(clusterName, url string) *DatasourceConfig {
+	return &DatasourceConfig{
+		Name:      clusterName,
+		Type:      "prometheus",
+		URL:       url,
+		IsDefault: true,
+	}
+}
+
+// WithName sets name of datasource
+func (c *DatasourceConfig) WithName(name string) *DatasourceConfig {
+	c.Name = name
+	return c
+}
+
+// WithType sets type of datasource
+func (c *DatasourceConfig) WithType(typ string) *DatasourceConfig {
+	c.Type = typ
+	return c
+}
+
+// WithIsDefault sets if datasource is default
+func (c *DatasourceConfig) WithIsDefault(isDefault bool) *DatasourceConfig {
+	c.IsDefault = isDefault
+	return c
 }
 
 // ConfigToFile write config content to specific path
 func (c *DatasourceConfig) ConfigToFile(file string) error {
+	config, err := c.Config()
+	if err != nil {
+		return err
+	}
+	return utils.WriteFile(file, config, 0755)
+}
+
+// Config generate the config file data.
+func (c *DatasourceConfig) Config() ([]byte, error) {
 	fp := path.Join("templates", "config", "datasource.yml.tpl")
 	tpl, err := embed.ReadTemplate(fp)
 	if err != nil {
-		return err
+		return nil, err
 	}
 
 	tmpl, err := template.New("Datasource").Parse(string(tpl))
 	if err != nil {
-		return err
+		return nil, err
 	}
 
 	content := bytes.NewBufferString("")
-	if err := tmpl.Execute(content, c); err != nil {
-		return err
+	if err := tmpl.Execute(content, map[string]any{
+		"Datasources": []any{c},
+	}); err != nil {
+		return nil, err
 	}
 
-	return utils.WriteFile(file, content.Bytes(), 0755)
+	return content.Bytes(), nil
 }