client method for row ops from performance schema

go.mod

@@ -8,7 +8,9 @@ module github.com/open-telemetry/opentelemetry-collector-contrib
 // For the OpenTelemetry Collector Contrib distribution specifically, see
 // https://github.com/open-telemetry/opentelemetry-collector-releases/tree/main/distributions/otelcol-contrib
 
-go 1.21.0
+go 1.22.2
+
+toolchain go1.22.5
 
 require (
 	github.com/open-telemetry/opentelemetry-collector-contrib/connector/countconnector v0.102.0
@@ -511,6 +513,7 @@ require (
 	github.com/jpillora/backoff v1.0.0 // indirect
 	github.com/json-iterator/go v1.1.12 // indirect
 	github.com/julienschmidt/httprouter v1.3.0 // indirect
+	github.com/k0kubun/pp v3.0.1+incompatible // indirect
 	github.com/kardianos/osext v0.0.0-20190222173326-2bc1f35cddc0 // indirect
 	github.com/karrick/godirwalk v1.17.0 // indirect
 	github.com/klauspost/compress v1.17.8 // indirect
@@ -539,6 +542,7 @@ require (
 	github.com/mattn/go-sqlite3 v1.14.22 // indirect
 	github.com/microsoft/ApplicationInsights-Go v0.4.4 // indirect
 	github.com/microsoft/go-mssqldb v1.7.2 // indirect
+	github.com/middleware-labs/innoParser v0.0.0-20240508090457-8c2fa2246395 // indirect
 	github.com/miekg/dns v1.1.58 // indirect
 	github.com/mistifyio/go-zfs v2.1.2-0.20190413222219-f784269be439+incompatible // indirect
 	github.com/mitchellh/copystructure v1.2.0 // indirect

receiver/mysqlreceiver/client.go

@@ -29,6 +29,7 @@ type client interface {
 	getInnodbStatusStats() (map[string]int64, error, int)
 	getTotalRows() ([]NRows, error)
 	getTotalErrors() (int64, error)
+	getRowOperationStats() (RowOperationStats, error)
 	Close() error
 }
 
@@ -40,6 +41,12 @@ type mySQLClient struct {
 	statementEventsTimeLimit       time.Duration
 }
 
+type RowOperationStats struct {
+	rowsRead     int64
+	rowsUpdated  int64
+	rowsDeleted  int64
+	rowsInserted int64
+}
 type IoWaitsStats struct {
 	schema      string
 	name        string
@@ -233,6 +240,58 @@ func (c *mySQLClient) getVersion() (string, error) {
 	return version, nil
 }
 
+func (c *mySQLClient) getRowOperationStats() (RowOperationStats, error) {
+	query := "SELECT SUBSTRING_INDEX(DIGEST_TEXT, ' ', 1) AS statement_type, " +
+		"SUM(SUM_ROWS_AFFECTED) AS rows_affected, " +
+		"SUM(SUM_ROWS_SENT) AS rows_sent " +
+		"FROM performance_schema.events_statements_summary_by_digest " +
+		"WHERE DIGEST_TEXT LIKE 'SELECT% '" +
+		"OR DIGEST_TEXT LIKE 'INSERT%' " +
+		"OR DIGEST_TEXT LIKE 'UPDATE%' " +
+		"OR DIGEST_TEXT LIKE 'DELETE%' " +
+		"GROUP BY statement_type; "
+
+	rows, err := c.client.Query(query)
+	rowOpsStats := new(RowOperationStats)
+	/*
+		+----------------+-----------------+---------------+-----------+---------------+
+		| statement_type | execution_count | rows_affected | rows_sent | rows_examined |
+		+----------------+-----------------+---------------+-----------+---------------+
+		| SELECT         |            4862 |             0 |    246533 |      43871844 |
+		| UPDATE         |            2800 |           395 |         0 |        480395 |
+		| DELETE         |             400 |             0 |         0 |         79800 |
+		| INSERT         |             800 |           400 |         0 |         80200 |
+		+----------------+-----------------+---------------+-----------+---------------+
+	*/
+	if err != nil {
+		return *rowOpsStats, err
+	}
+
+	defer rows.Close()
+
+	for rows.Next() {
+		var rowsAffected int64
+		var rowsSent int64
+		var statementType string
+		err := rows.Scan(&statementType, &rowsAffected, &rowsSent)
+
+		if err != nil {
+			return *rowOpsStats, err
+		}
+
+		if statementType == "SELECT" {
+			rowOpsStats.rowsRead = rowsSent
+		} else if statementType == "UPDATE" {
+			rowOpsStats.rowsUpdated = rowsAffected
+		} else if statementType == "DELETE" {
+			rowOpsStats.rowsDeleted = rowsAffected
+		} else if statementType == "INSERT" {
+			rowOpsStats.rowsInserted = rowsAffected
+		}
+	}
+	return *rowOpsStats, nil
+}
+
 // getGlobalStats queries the db for global status metrics.
 func (c *mySQLClient) getGlobalStats() (map[string]string, error) {
 	q := "SHOW GLOBAL STATUS;"

receiver/mysqlreceiver/documentation.md

@@ -286,6 +286,38 @@ The number of InnoDB page operations.
 | ---- | ----------- | ------ |
 | operation | The page operation types. | Str: ``created``, ``read``, ``written`` |
 
+### mysql.performance.rows_deleted
+
+The number of rows deleted in the database as per the performance schema.
+
+| Unit | Metric Type | Value Type | Aggregation Temporality | Monotonic |
+| ---- | ----------- | ---------- | ----------------------- | --------- |
+| {row} | Sum | Int | Cumulative | true |
+
+### mysql.performance.rows_inserted
+
+The number of rows inserted in the database as per the performance schema.
+
+| Unit | Metric Type | Value Type | Aggregation Temporality | Monotonic |
+| ---- | ----------- | ---------- | ----------------------- | --------- |
+| {row} | Sum | Int | Cumulative | true |
+
+### mysql.performance.rows_read
+
+The number of rows read in the database as per the performance schema.
+
+| Unit | Metric Type | Value Type | Aggregation Temporality | Monotonic |
+| ---- | ----------- | ---------- | ----------------------- | --------- |
+| {row} | Sum | Int | Cumulative | true |
+
+### mysql.performance.rows_updated
+
+The number of rows updated in the database as per the performance schema.
+
+| Unit | Metric Type | Value Type | Aggregation Temporality | Monotonic |
+| ---- | ----------- | ---------- | ----------------------- | --------- |
+| {row} | Sum | Int | Cumulative | true |
+
 ### mysql.prepared_statements
 
 The number of times each type of prepared statement command has been issued.