Support SELECT/INSERT/UPDATE/DELETE db operation + span names

.github/workflows/e2e/k8s/sample-job.yml

@@ -38,6 +38,8 @@ spec:
           value: "tracecontext,baggage"
         - name: OTEL_GO_AUTO_INCLUDE_DB_STATEMENT
           value: "true"
+        - name: OTEL_GO_AUTO_INCLUDE_DB_OPERATION
+          value: "true"
         - name: OTEL_BSP_SCHEDULE_DELAY
           value: "60000"
         resources: {}

CHANGELOG.md

@@ -20,6 +20,7 @@ OpenTelemetry Go Automatic Instrumentation adheres to [Semantic Versioning](http
 - Support `google.golang.org/grpc` `1.68.0`. ([#1251](https://github.com/open-telemetry/opentelemetry-go-instrumentation/pull/1251))
 - Support `golang.org/x/net` `0.31.0`. ([#1254](https://github.com/open-telemetry/opentelemetry-go-instrumentation/pull/1254))
 - Support `go.opentelemetry.io/[email protected]`. ([#1302](https://github.com/open-telemetry/opentelemetry-go-instrumentation/pull/1302))
+- Support `SELECT`, `INSERT`, `UPDATE`, and `DELETE` for database span names and `db.operation.name` attribute. ([#1253](https://github.com/open-telemetry/opentelemetry-go-instrumentation/pull/1253))
 
 ### Fixed
 

docs/configuration.md

@@ -27,6 +27,7 @@ Alternatively, you can add support for additional or different configurations by
 | Environment variable                | Description                                            | Default value |
 |-------------------------------------|--------------------------------------------------------|---------------|
 | `OTEL_GO_AUTO_INCLUDE_DB_STATEMENT` | Sets whether to include SQL queries in the trace data. |               |
+| `OTEL_GO_AUTO_INCLUDE_DB_OPERATION` | Sets whether to include SQL operation in the trace data. |               |
 
 ## Traces exporter
 

go.mod

@@ -22,6 +22,7 @@ require (
 	github.com/hashicorp/go-version v1.7.0
 	github.com/pkg/errors v0.9.1
 	github.com/stretchr/testify v1.9.0
+	github.com/xwb1989/sqlparser v0.0.0-20180606152119-120387863bf2
 	go.opentelemetry.io/collector/pdata v1.19.0
 	go.opentelemetry.io/contrib/exporters/autoexport v0.57.0
 	go.opentelemetry.io/otel v1.32.0

go.sum

@@ -74,6 +74,8 @@ github.com/stretchr/objx v0.1.0/go.mod h1:HFkY916IF+rwdDfMAkV7OtwuqBVzrE8GR6GFx+
 github.com/stretchr/testify v1.3.0/go.mod h1:M5WIy9Dh21IEIfnGCwXGc5bZfKNJtfHm1UVUgZn+9EI=
 github.com/stretchr/testify v1.9.0 h1:HtqpIVDClZ4nwg75+f6Lvsy/wHu+3BoSGCbBAcpTsTg=
 github.com/stretchr/testify v1.9.0/go.mod h1:r2ic/lqez/lEtzL7wO/rwa5dbSLXVDPFyf8C91i36aY=
+github.com/xwb1989/sqlparser v0.0.0-20180606152119-120387863bf2 h1:zzrxE1FKn5ryBNl9eKOeqQ58Y/Qpo3Q9QNxKHX5uzzQ=
+github.com/xwb1989/sqlparser v0.0.0-20180606152119-120387863bf2/go.mod h1:hzfGeIUDq/j97IG+FhNqkowIyEcD88LrW6fyU3K3WqY=
 github.com/yuin/goldmark v1.1.27/go.mod h1:3hX8gzYuyVAZsxl0MRgGTJEmQBFcNTphYh9decYSb74=
 github.com/yuin/goldmark v1.2.1/go.mod h1:3hX8gzYuyVAZsxl0MRgGTJEmQBFcNTphYh9decYSb74=
 go.opentelemetry.io/collector/pdata v1.19.0 h1:jmnU5R8TOCbwRr4B8sjdRxM7L5WnEKlQWX1dtLYxIbE=

internal/pkg/instrumentation/bpf/database/sql/probe.go

@@ -8,6 +8,8 @@ import (
 	"os"
 	"strconv"
 
+	sql "github.com/xwb1989/sqlparser"
+
 	"go.opentelemetry.io/collector/pdata/pcommon"
 	"go.opentelemetry.io/collector/pdata/ptrace"
 	semconv "go.opentelemetry.io/otel/semconv/v1.26.0"
@@ -27,6 +29,9 @@ const (
 
 	// IncludeDBStatementEnvVar is the environment variable to opt-in for sql query inclusion in the trace.
 	IncludeDBStatementEnvVar = "OTEL_GO_AUTO_INCLUDE_DB_STATEMENT"
+
+	// IncludeDBOperationEnvVar is the environment variable to opt-in for sql query operation in the trace.
+	IncludeDBOperationEnvVar = "OTEL_GO_AUTO_INCLUDE_DB_OPERATION"
 )
 
 // New returns a new [probe.Probe].
@@ -97,6 +102,32 @@ func processFn(e *event) ptrace.SpanSlice {
 		span.Attributes().PutStr(string(semconv.DBQueryTextKey), query)
 	}
 
+	includeOperationVal := os.Getenv(IncludeDBOperationEnvVar)
+	if includeOperationVal != "" {
+		include, err := strconv.ParseBool(includeOperationVal)
+		if err == nil && include {
+			q, err := sql.Parse(query)
+			if err == nil {
+				operation := ""
+				switch q.(type) {
+				case *sql.Select:
+					operation = "SELECT"
+				case *sql.Update:
+					operation = "UPDATE"
+				case *sql.Insert:
+					operation = "INSERT"
+				case *sql.Delete:
+					operation = "DELETE"
+				}
+
+				if operation != "" {
+					span.Attributes().PutStr(string(semconv.DBOperationNameKey), operation)
+					span.SetName(operation)
+				}
+			}
+		}
+	}
+
 	return spans
 }
 

internal/pkg/instrumentation/bpf/database/sql/probe_test.go

@@ -4,6 +4,7 @@
 package sql
 
 import (
+	"os"
 	"testing"
 	"time"
 
@@ -18,7 +19,60 @@ import (
 	"go.opentelemetry.io/auto/internal/pkg/instrumentation/utils"
 )
 
+func BenchmarkProcessFn(b *testing.B) {
+	tests := []struct {
+		name  string
+		query string
+	}{
+		{
+			name:  "no query (baseline)",
+			query: "",
+		},
+		{
+			name:  "simple query",
+			query: "SELECT * FROM customers",
+		},
+		{
+			name:  "medium query",
+			query: "SELECT * FROM customers WHERE first_name='Mike' AND last_name IN ('Santa', 'Banana')",
+		},
+		{
+			name:  "hard query",
+			query: "WITH (SELECT last_name FROM customers WHERE first_name='Mike' AND country='North Pole') AS test_table SELECT * FROM test_table WHERE first_name='Mike' AND last_name IN ('Santa', 'Banana')",
+		},
+	}
+
+	start := time.Unix(0, time.Now().UnixNano()) // No wall clock.
+	end := start.Add(1 * time.Second)
+
+	startOffset := utils.TimeToBootOffset(start)
+	endOffset := utils.TimeToBootOffset(end)
+
+	traceID := trace.TraceID{1}
+	spanID := trace.SpanID{1}
+
+	for _, t := range tests {
+		b.Run(t.name, func(b *testing.B) {
+			var byteQuery [256]byte
+			copy(byteQuery[:], []byte(t.query))
+			b.ResetTimer()
+			for i := 0; i < b.N; i++ {
+				_ = processFn(&event{
+					BaseSpanProperties: context.BaseSpanProperties{
+						StartTime:   startOffset,
+						EndTime:     endOffset,
+						SpanContext: context.EBPFSpanContext{TraceID: traceID, SpanID: spanID},
+					},
+					Query: byteQuery,
+				})
+			}
+		})
+	}
+}
+
 func TestProbeConvertEvent(t *testing.T) {
+	err := os.Setenv(IncludeDBOperationEnvVar, "true")
+	assert.NoError(t, err)
 	start := time.Unix(0, time.Now().UnixNano()) // No wall clock.
 	end := start.Add(1 * time.Second)
 
@@ -41,14 +95,14 @@ func TestProbeConvertEvent(t *testing.T) {
 	want := func() ptrace.SpanSlice {
 		spans := ptrace.NewSpanSlice()
 		span := spans.AppendEmpty()
-		span.SetName("DB")
+		span.SetName("SELECT")
 		span.SetKind(ptrace.SpanKindClient)
 		span.SetStartTimestamp(utils.BootOffsetToTimestamp(startOffset))
 		span.SetEndTimestamp(utils.BootOffsetToTimestamp(endOffset))
 		span.SetTraceID(pcommon.TraceID(traceID))
 		span.SetSpanID(pcommon.SpanID(spanID))
 		span.SetFlags(uint32(trace.FlagsSampled))
-		utils.Attributes(span.Attributes(), semconv.DBQueryText("SELECT * FROM foo"))
+		utils.Attributes(span.Attributes(), semconv.DBQueryText("SELECT * FROM foo"), semconv.DBOperationName("SELECT"))
 		return spans
 	}()
 	assert.Equal(t, want, got)

internal/test/e2e/databasesql/main.go

@@ -100,6 +100,118 @@ func (s *Server) queryDb(w http.ResponseWriter, req *http.Request) {
 	}
 }
 
+func (s *Server) insert(w http.ResponseWriter, req *http.Request) {
+	ctx := req.Context()
+
+	conn, err := s.db.Conn(ctx)
+	if err != nil {
+		panic(err)
+	}
+
+	rows, err := conn.QueryContext(req.Context(), "INSERT INTO contacts (first_name) VALUES ('Mike')")
+	if err != nil {
+		panic(err)
+	}
+
+	logger.Info("insert called")
+	for rows.Next() {
+		var id int
+		var firstName string
+		var lastName string
+		var email string
+		var phone string
+		err := rows.Scan(&id, &firstName, &lastName, &email, &phone)
+		if err != nil {
+			panic(err)
+		}
+		fmt.Fprintf(w, "ID: %d, firstName: %s, lastName: %s, email: %s, phone: %s\n", id, firstName, lastName, email, phone)
+	}
+}
+
+func (s *Server) update(w http.ResponseWriter, req *http.Request) {
+	ctx := req.Context()
+
+	conn, err := s.db.Conn(ctx)
+	if err != nil {
+		panic(err)
+	}
+
+	rows, err := conn.QueryContext(req.Context(), "UPDATE contacts SET last_name = 'Santa' WHERE first_name = 'Mike'")
+	if err != nil {
+		panic(err)
+	}
+
+	logger.Info("update called")
+	for rows.Next() {
+		var id int
+		var firstName string
+		var lastName string
+		var email string
+		var phone string
+		err := rows.Scan(&id, &firstName, &lastName, &email, &phone)
+		if err != nil {
+			panic(err)
+		}
+		fmt.Fprintf(w, "ID: %d, firstName: %s, lastName: %s, email: %s, phone: %s\n", id, firstName, lastName, email, phone)
+	}
+}
+
+func (s *Server) delete(w http.ResponseWriter, req *http.Request) {
+	ctx := req.Context()
+
+	conn, err := s.db.Conn(ctx)
+	if err != nil {
+		panic(err)
+	}
+
+	rows, err := conn.QueryContext(req.Context(), "DELETE FROM contacts WHERE first_name = 'Mike'")
+	if err != nil {
+		panic(err)
+	}
+
+	logger.Info("delete called")
+	for rows.Next() {
+		var id int
+		var firstName string
+		var lastName string
+		var email string
+		var phone string
+		err := rows.Scan(&id, &firstName, &lastName, &email, &phone)
+		if err != nil {
+			panic(err)
+		}
+		fmt.Fprintf(w, "ID: %d, firstName: %s, lastName: %s, email: %s, phone: %s\n", id, firstName, lastName, email, phone)
+	}
+}
+
+func (s *Server) drop(w http.ResponseWriter, req *http.Request) {
+	ctx := req.Context()
+
+	conn, err := s.db.Conn(ctx)
+	if err != nil {
+		panic(err)
+	}
+
+	rows, err := conn.QueryContext(req.Context(), "DROP TABLE contacts")
+	if err != nil {
+		panic(err)
+	}
+
+	logger.Info("drop called")
+	for rows.Next() {
+		var id int
+		var firstName string
+		var lastName string
+		var email string
+		var phone string
+		err := rows.Scan(&id, &firstName, &lastName, &email, &phone)
+		if err != nil {
+			panic(err)
+		}
+		fmt.Fprintf(w, "ID: %d, firstName: %s, lastName: %s, email: %s, phone: %s\n", id, firstName, lastName, email, phone)
+	}
+}
+
 var logger *zap.Logger
 
 func main() {
@@ -115,6 +227,10 @@ func main() {
 	s := NewServer()
 
 	http.HandleFunc("/query_db", s.queryDb)
+	http.HandleFunc("/insert", s.insert)
+	http.HandleFunc("/update", s.update)
+	http.HandleFunc("/delete", s.delete)
+	http.HandleFunc("/drop", s.drop)
 	go func() {
 		_ = http.ListenAndServe(":8080", nil)
 	}()
@@ -134,6 +250,54 @@ func main() {
 	logger.Info("Body:\n", zap.String("body", string(body[:])))
 	_ = resp.Body.Close()
 
+	resp, err = http.Get("http://localhost:8080/insert")
+	if err != nil {
+		logger.Error("Error performing GET", zap.Error(err))
+	}
+	body, err = io.ReadAll(resp.Body)
+	if err != nil {
+		logger.Error("Error reading http body", zap.Error(err))
+	}
+
+	logger.Info("Body:\n", zap.String("body", string(body[:])))
+	_ = resp.Body.Close()
+
+	resp, err = http.Get("http://localhost:8080/update")
+	if err != nil {
+		logger.Error("Error performing GET", zap.Error(err))
+	}
+	body, err = io.ReadAll(resp.Body)
+	if err != nil {
+		logger.Error("Error reading http body", zap.Error(err))
+	}
+
+	logger.Info("Body:\n", zap.String("body", string(body[:])))
+	_ = resp.Body.Close()
+
+	resp, err = http.Get("http://localhost:8080/delete")
+	if err != nil {
+		logger.Error("Error performing GET", zap.Error(err))
+	}
+	body, err = io.ReadAll(resp.Body)
+	if err != nil {
+		logger.Error("Error reading http body", zap.Error(err))
+	}
+
+	logger.Info("Body:\n", zap.String("body", string(body[:])))
+	_ = resp.Body.Close()
+
+	resp, err = http.Get("http://localhost:8080/drop")
+	if err != nil {
+		logger.Error("Error performing GET", zap.Error(err))
+	}
+	body, err = io.ReadAll(resp.Body)
+	if err != nil {
+		logger.Error("Error reading http body", zap.Error(err))
+	}
+
+	logger.Info("Body:\n", zap.String("body", string(body[:])))
+	_ = resp.Body.Close()
+
 	// give time for auto-instrumentation to report signal
 	time.Sleep(5 * time.Second)
 }