test(e2e): test monitoring data

tests/e2e/greptimedbcluster/test_cluster_enable_monitoring.go

@@ -18,11 +18,14 @@ import (
 	"context"
 	"fmt"
 	"net"
+	"sort"
 	"time"
 
 	. "github.com/onsi/ginkgo/v2"
 	. "github.com/onsi/gomega"
 
+	"github.com/google/go-cmp/cmp"
+	"github.com/jackc/pgx/v5"
 	"sigs.k8s.io/controller-runtime/pkg/client"
 
 	greptimev1alpha1 "github.com/GreptimeTeam/greptimedb-operator/apis/v1alpha1"
@@ -62,6 +65,7 @@ func TestClusterEnableMonitoring(ctx context.Context, h *helper.Helper) {
 
 	err = h.Get(ctx, client.ObjectKey{Name: testCluster.Name, Namespace: testCluster.Namespace}, testCluster)
 	Expect(err).NotTo(HaveOccurred(), "failed to get cluster")
+
 	By("Execute distributed SQL test")
 	frontendAddr, err := h.PortForward(ctx, testCluster.Namespace, common.ResourceName(testCluster.Name, greptimev1alpha1.FrontendComponentKind), int(testCluster.Spec.PostgreSQLPort))
 	Expect(err).NotTo(HaveOccurred(), "failed to port forward frontend service")
@@ -77,6 +81,19 @@ func TestClusterEnableMonitoring(ctx context.Context, h *helper.Helper) {
 	err = h.RunSQLTest(ctx, frontendAddr, testSQLFile)
 	Expect(err).NotTo(HaveOccurred(), "failed to run sql test")
 
+	monitoringAddr, err := h.PortForward(ctx, testCluster.Namespace, common.ResourceName(common.MonitoringServiceName(testCluster.Name), greptimev1alpha1.StandaloneKind), int(testCluster.Spec.PostgreSQLPort))
+	Expect(err).NotTo(HaveOccurred(), "failed to port forward monitoring service")
+	Eventually(func() error {
+		conn, err := net.Dial("tcp", monitoringAddr)
+		if err != nil {
+			return err
+		}
+		conn.Close()
+		return nil
+	}, helper.DefaultTimeout, time.Second).ShouldNot(HaveOccurred())
+	err = testMonitoringStandalone(ctx, monitoringAddr)
+	Expect(err).NotTo(HaveOccurred(), "failed to test monitoring")
+
 	By("Kill the port forwarding process")
 	h.KillPortForwardProcess()
 
@@ -99,3 +116,62 @@ func TestClusterEnableMonitoring(ctx context.Context, h *helper.Helper) {
 		Expect(err).NotTo(HaveOccurred(), "failed to delete datanode PVC")
 	}
 }
+
+func testMonitoringStandalone(ctx context.Context, addr string) error {
+	// connect public database by default.
+	url := fmt.Sprintf("postgres://postgres@%s/public?sslmode=disable", addr)
+
+	fmt.Printf("Connecting to %s\n", url)
+	conn, err := pgx.Connect(ctx, url)
+	if err != nil {
+		return err
+	}
+	defer conn.Close(ctx)
+
+	// Check metrics table.
+	if err := checkCollectedRoles(ctx, conn, "SELECT DISTINCT app FROM greptime_app_version", []string{"greptime-datanode", "greptime-frontend", "greptime-metasrv"}); err != nil {
+		return err
+	}
+
+	// Check logs table.
+	if err := checkCollectedRoles(ctx, conn, "SELECT DISTINCT role FROM gtlogs", []string{"datanode", "frontend", "meta"}); err != nil {
+		return err
+	}
+
+	var count int
+	if err = conn.QueryRow(context.Background(), "SELECT COUNT(*) FROM gtlogs").Scan(&count); err != nil {
+		return err
+	}
+	// The number of logs should be greater than 0.
+	if count == 0 {
+		return fmt.Errorf("no logs found")
+	}
+
+	return nil
+}
+
+func checkCollectedRoles(ctx context.Context, conn *pgx.Conn, query string, expected []string) error {
+	rows, err := conn.Query(ctx, query)
+	if err != nil {
+		return err
+	}
+	defer rows.Close()
+
+	var result []string
+	for rows.Next() {
+		var role string
+		err = rows.Scan(&role)
+		if err != nil {
+			return err
+		}
+		result = append(result, role)
+	}
+	sort.Strings(result)
+	sort.Strings(expected)
+
+	if !cmp.Equal(result, expected) {
+		return fmt.Errorf("results mismatch, got %v, expect %v", result, expected)
+	}
+
+	return nil
+}