Active connections in mysql scrapper

receiver/mysqlreceiver/client.go

@@ -7,6 +7,7 @@ import (
 	"context"
 	"database/sql"
 	"fmt"
+	"strconv"
 	"strings"
 	"time"
 
@@ -30,6 +31,7 @@ type client interface {
 	getTotalRows() ([]NRows, error)
 	getTotalErrors() (int64, error)
 	getRowOperationStats() (RowOperationStats, error)
+	getActiveConnections() (int64, error)
 	Close() error
 }
 
@@ -570,6 +572,25 @@ func (c *mySQLClient) getTotalErrors() (int64, error) {
 	return nerrors, nil
 }
 
+func (c *mySQLClient) getActiveConnections() (int64, error) {
+	query := "SHOW STATUS WHERE `variable_name` = 'Threads_connected'"
+
+	var varName string
+	var value string
+
+	err := c.client.QueryRow(query).Scan(&varName, &value)
+	if err != nil {
+		return -1, fmt.Errorf("failed to scan active connections: %w", err)
+	}
+
+	connections, err := strconv.ParseInt(value, 10, 64)
+	if err != nil {
+		return -1, fmt.Errorf("failed to parse active connections count: %w", err)
+	}
+
+	return connections, nil
+}
+
 func (c *mySQLClient) getTableLockWaitEventStats() ([]tableLockWaitEventStats, error) {
 	query := "SELECT OBJECT_SCHEMA, OBJECT_NAME, COUNT_READ_NORMAL, COUNT_READ_WITH_SHARED_LOCKS," +
 		"COUNT_READ_HIGH_PRIORITY, COUNT_READ_NO_INSERT, COUNT_READ_EXTERNAL, COUNT_WRITE_ALLOW_WRITE," +

receiver/mysqlreceiver/documentation.md

@@ -98,6 +98,14 @@ The number of times each type of command has been executed.
 | ---- | ----------- | ------ |
 | command | The command types. | Str: ``delete``, ``insert``, ``select``, ``update`` |
 
+### mysql.connection.active.count
+
+The numner of active connections to the MySQL server
+
+| Unit | Metric Type | Value Type |
+| ---- | ----------- | ---------- |
+| 1 | Gauge | Int |
+
 ### mysql.connection.count
 
 The number of connection attempts (successful or not) to the MySQL server.

receiver/mysqlreceiver/internal/metadata/testdata/config.yaml

@@ -17,6 +17,8 @@ all_set:
       enabled: true
     mysql.commands:
       enabled: true
+    mysql.connection.active.count:
+      enabled: true
     mysql.connection.count:
       enabled: true
     mysql.connection.errors:
@@ -140,6 +142,8 @@ none_set:
       enabled: false
     mysql.commands:
       enabled: false
+    mysql.connection.active.count:
+      enabled: false
     mysql.connection.count:
       enabled: false
     mysql.connection.errors:

receiver/mysqlreceiver/metadata.yaml

@@ -486,6 +486,14 @@ metrics:
       monotonic: false
       aggregation_temporality: cumulative
     attributes: [schema, table_name, write_lock_type]
+
+  mysql.connection.active.count:
+    enabled: true
+    description: The numner of active connections to the MySQL server
+    unit: 1
+    gauge:
+      value_type: int
+
   mysql.connection.count:
     enabled: true 
     description: The number of connection attempts (successful or not) to the MySQL server.

receiver/mysqlreceiver/scraper.go

@@ -16,6 +16,7 @@ import (
 	"go.opentelemetry.io/collector/receiver/scrapererror"
 	"go.uber.org/zap"
 
+	"github.com/k0kubun/pp"
 	"github.com/open-telemetry/opentelemetry-collector-contrib/receiver/mysqlreceiver/internal/metadata"
 )
 
@@ -118,6 +119,8 @@ func (m *mySQLScraper) scrape(context.Context) (pmetric.Metrics, error) {
 
 	m.scraperInnodbMetricsForDBM(now, errs)
 
+	m.scrapeActiveConnections(now, errs)
+
 	rb := m.mb.NewResourceBuilder()
 
 	version, err := m.sqlclient.getVersion()
@@ -677,6 +680,16 @@ func (m *mySQLScraper) scrapeReplicaStatusStats(now pcommon.Timestamp) {
 	}
 }
 
+func (m *mySQLScraper) scrapeActiveConnections(now pcommon.Timestamp, errs *scrapererror.ScrapeErrors) {
+	activeConnections, err := m.sqlclient.getActiveConnections()
+	pp.Println("Trying to scrape active connections")
+	if err != nil {
+		m.logger.Info("Failed to fetch active connections", zap.Error(err))
+		return
+	}
+	m.mb.RecordMysqlConnectionActiveCountDataPoint(now, activeConnections)
+}
+
 func addPartialIfError(errors *scrapererror.ScrapeErrors, err error) {
 	if err != nil {
 		errors.AddPartial(1, err)

receiver/mysqlreceiver/scraper_test.go

@@ -73,6 +73,7 @@ func TestScrape(t *testing.T) {
 			totalRowsFile:               "total_rows_stats",
 			totalErrorsFile:             "total_error_stats",
 			rowOperationsStatsFile:      "row_operations_status",
+			activeConnectionsFile:       "active_connections",
 		}
 
 		scraper.renameCommands = true
@@ -121,6 +122,7 @@ func TestScrape(t *testing.T) {
 			totalRowsFile:               "total_rows_empty",
 			totalErrorsFile:             "total_errors_empty",
 			rowOperationsStatsFile:      "row_operations_status_empty",
+			activeConnectionsFile:       "active_connections_empty",
 		}
 
 		actualMetrics, scrapeErr := scraper.scrape(context.Background())
@@ -163,6 +165,7 @@ type mockClient struct {
 	totalRowsFile               string
 	totalErrorsFile             string
 	rowOperationsStatsFile      string
+	activeConnectionsFile       string
 }
 
 func readFile(fname string) (map[string]string, error) {
@@ -197,6 +200,42 @@ func (c *mockClient) getInnodbStats() (map[string]string, error) {
 	return readFile(c.innodbStatsFile)
 }
 
+// getActiveConnections implements client.
+func (c *mockClient) getActiveConnections() (int64, error) {
+	// Open test data file
+	file, err := os.Open(filepath.Join("testdata", "scraper", c.rowOperationsStatsFile+".txt"))
+	if err != nil {
+		return -1, fmt.Errorf("failed to open test data file: %w", err)
+	}
+	defer file.Close()
+
+	// Create scanner to read test data
+	scanner := bufio.NewScanner(file)
+
+	// Find the Threads_connected line
+	for scanner.Scan() {
+		line := scanner.Text()
+		if strings.Contains(line, "Threads_connected") {
+			// Split the line by whitespace and get the value
+			fields := strings.Fields(line)
+			if len(fields) >= 2 {
+				// Parse the value to int64
+				connections, err := strconv.ParseInt(fields[len(fields)-1], 10, 64)
+				if err != nil {
+					return -1, fmt.Errorf("failed to parse connection count from test data: %w", err)
+				}
+				return connections, nil
+			}
+		}
+	}
+
+	if err := scanner.Err(); err != nil {
+		return -1, fmt.Errorf("error reading test data: %w", err)
+	}
+
+	return -1, fmt.Errorf("Threads_connected value not found in test data")
+}
+
 func (c *mockClient) getRowOperationStats() (RowOperationStats, error) {
 	rowOpsStats := new(RowOperationStats)
 	file, err := os.Open(filepath.Join("testdata", "scraper", c.rowOperationsStatsFile+".txt"))

receiver/mysqlreceiver/testdata/scraper/active_connections.txt

@@ -0,0 +1 @@
+Threads_connected 1