Merge pull request #50 from abhinav1602/queryAnalysis-OHI

Query analysis ohi

src/args/argument_list.go

@@ -28,10 +28,10 @@ type ArgumentList struct {
 	ShowVersion                  bool   `default:"false" help:"Print build information and exit"`
 	ExtraConnectionURLArgs       string `default:"" help:"Appends additional parameters to connection url. Ex. 'applicationintent=readonly&foo=bar'"`
 	EnableDiskMetricsInBytes     bool   `default:"true" help:"Enable collection of instance.diskInBytes."`
-	EnableQueryPerformance       bool   `default:"false"`
-	QueryResponseTimeThreshold   int    `default:"0"`
-	QueryCountThreshold          int    `default:"20"`
-	FetchInterval                int    `default:"15"`
+	EnableQueryPerformance       bool   `default:"false" help:"Enable collection of detailed query performance metrics."`
+	QueryResponseTimeThreshold   int    `default:"0" help:"Threshold in milliseconds for query response time. If response time exceeds this threshold, the query will be considered slow."`
+	QueryCountThreshold          int    `default:"20" help:"Maximum number of queries returned in query analysis results."`
+	FetchInterval                int    `default:"15" help:"Interval in seconds for fetching grouped slow queries; Should always be same as mysql-config interval."`
 }
 
 // Validate validates SQL specific arguments

src/queryanalysis/config/query_config.go

@@ -122,8 +122,7 @@ var Queries = []models.QueryDetailsDto{
 	},
 	{
 		Name: "MSSQLWaitTimeAnalysis",
-		Query: `DECLARE @IntervalSeconds INT = %d;     	-- Define the interval in seconds
-				DECLARE @TopN INT = %d; 				-- Number of results to retrieve
+		Query: `DECLARE @TopN INT = %d; 				-- Number of results to retrieve
 				DECLARE @TextTruncateLimit INT = %d; 	-- Truncate limit for query_text
 				DECLARE @sql NVARCHAR(MAX) = '';
 				DECLARE @dbName NVARCHAR(128);
@@ -173,7 +172,6 @@ var Queries = []models.QueryDetailsDto{
 					  qsqt.query_sql_text NOT LIKE ''%%WITH%%''
 					  AND qsqt.query_sql_text NOT LIKE ''%%sys.%%''
 					  AND qsqt.query_sql_text NOT LIKE ''%%INFORMATION_SCHEMA%%''
-					  AND qsq.last_execution_time > DATEADD(second, -' + CAST(@IntervalSeconds AS NVARCHAR(10)) + ', GETUTCDATE())
 					GROUP BY 
 					  qsqt.query_sql_text 
 				  ),
@@ -209,7 +207,6 @@ var Queries = []models.QueryDetailsDto{
 					  qsqt.query_sql_text NOT LIKE ''%%WITH%%''
 					  AND qsqt.query_sql_text NOT LIKE ''%%sys.%%''
 					  AND qsqt.query_sql_text NOT LIKE ''%%INFORMATION_SCHEMA%%''
-					  AND qsq.last_execution_time > DATEADD(second, -' + CAST(@IntervalSeconds AS NVARCHAR(10)) + ', GETUTCDATE())
 				  )
 				  SELECT
 					query_id,
@@ -229,12 +226,10 @@ var Queries = []models.QueryDetailsDto{
 				
 				  FETCH NEXT FROM db_cursor INTO @dbName;
 				END
-				
 				CLOSE db_cursor;
 				DEALLOCATE db_cursor;
-				
 				SELECT TOP (@TopN) * FROM @resultTable 
-				ORDER BY last_execution_time DESC;`,
+				ORDER BY total_wait_time_ms DESC;`,
 		Type: "waitAnalysis",
 	},
 	{
@@ -313,7 +308,7 @@ TopPlans AS (
     CROSS APPLY sys.dm_exec_sql_text(qs.sql_handle) AS st
     CROSS APPLY sys.dm_exec_query_plan(qs.plan_handle) AS qp
     WHERE CONVERT(NVARCHAR(50), qs.query_hash) = @QueryID 
-    AND qs.last_execution_time BETWEEN DATEADD(SECOND, -@IntervalSeconds, GETDATE()) AND GETDATE() 
+    AND qs.last_execution_time BETWEEN DATEADD(SECOND, -@IntervalSeconds, GETUTCDATE()) AND GETUTCDATE() 
     AND COALESCE((qs.total_elapsed_time / NULLIF(qs.execution_count, 0)) / 1000, 0) > @ElapsedTimeThreshold
     ORDER BY avg_elapsed_time_ms DESC
 ),
@@ -353,4 +348,5 @@ var (
 	SlowQueryCountThresholdDefault    = 20
 	IndividualQueryCountMax           = 10
 	GroupedQueryCountMax              = 30
+	MaxSystemDatabaseID               = 4
 )

src/queryanalysis/models/blocking_session_query_details.go

@@ -1,15 +1,15 @@
 package models
 
 type BlockingSessionQueryDetails struct {
-	BlockingSPID          *int64   `db:"blocking_spid"`
-	BlockingStatus        *string  `db:"blocking_status"`
-	BlockedSPID           *int64   `db:"blocked_spid"`
-	BlockedStatus         *string  `db:"blocked_status"`
-	WaitType              *string  `db:"wait_type"`
-	WaitTimeInSeconds     *float64 `db:"wait_time_in_seconds"`
-	CommandType           *string  `db:"command_type"`
-	DatabaseName          *string  `db:"database_name"`
-	BlockingQueryText     *string  `db:"blocking_query_text"`
-	BlockedQueryText      *string  `db:"blocked_query_text"`
-	BlockedQueryStartTime *string  `db:"blocked_query_start_time"`
+	BlockingSPID          *int64   `db:"blocking_spid" metric_name:"blocking_spid" sourceType:"gauge"`
+	BlockingStatus        *string  `db:"blocking_status" metric_name:"blocking_status" sourceType:"attribute"`
+	BlockedSPID           *int64   `db:"blocked_spid" metric_name:"blocked_spid" sourceType:"gauge"`
+	BlockedStatus         *string  `db:"blocked_status" metric_name:"blocked_status" sourceType:"attribute"`
+	WaitType              *string  `db:"wait_type" metric_name:"wait_type" sourceType:"attribute"`
+	WaitTimeInSeconds     *float64 `db:"wait_time_in_seconds" metric_name:"wait_time_in_seconds" sourceType:"gauge"`
+	CommandType           *string  `db:"command_type" metric_name:"command_type" sourceType:"attribute"`
+	DatabaseName          *string  `db:"database_name" metric_name:"database_name" sourceType:"attribute"`
+	BlockingQueryText     *string  `db:"blocking_query_text" metric_name:"blocking_query_text" sourceType:"attribute"`
+	BlockedQueryText      *string  `db:"blocked_query_text" metric_name:"blocked_query_text" sourceType:"attribute"`
+	BlockedQueryStartTime *string  `db:"blocked_query_start_time" metric_name:"blocked_query_start_time" sourceType:"attribute"`
 }

src/queryanalysis/models/cutomType_test.go

@@ -0,0 +1,101 @@
+package models
+
+import (
+	"fmt"
+	"testing"
+)
+
+func runScanTests(t *testing.T, tests []struct {
+	name    string
+	input   interface{}
+	want    string
+	wantErr error
+}, scanFunc func(interface{}) (string, error)) {
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			got, err := scanFunc(tt.input)
+
+			if tt.wantErr != nil {
+				if err == nil || err.Error() != tt.wantErr.Error() {
+					t.Errorf("Scan() error = %v, wantErr %v", err, tt.wantErr)
+				}
+			} else {
+				if err != nil {
+					t.Errorf("Scan() unexpected error = %v", err)
+				}
+				if got != tt.want {
+					t.Errorf("Scan() = %v, want %v", got, tt.want)
+				}
+			}
+		})
+	}
+}
+
+func TestHexString_Scan(t *testing.T) {
+	tests := []struct {
+		name    string
+		input   interface{}
+		want    string
+		wantErr error
+	}{
+		{
+			name:    "valid byte slice",
+			input:   []uint8{0x12, 0x34, 0xab, 0xcd},
+			want:    "0x1234abcd",
+			wantErr: nil,
+		},
+		{
+			name:    "empty byte slice",
+			input:   []uint8{},
+			want:    "0x",
+			wantErr: nil,
+		},
+		{
+			name:    "invalid type",
+			input:   "not a byte slice",
+			want:    "",
+			wantErr: fmt.Errorf("%w, got %T", ErrExpectedUint8Slice, "not a byte slice"),
+		},
+		{
+			name:    "nil input",
+			input:   nil,
+			want:    "",
+			wantErr: fmt.Errorf("%w, got %T", ErrExpectedUint8Slice, nil),
+		},
+	}
+
+	runScanTests(t, tests, func(input interface{}) (string, error) {
+		var hex HexString
+		err := hex.Scan(input)
+		return string(hex), err
+	})
+}
+
+func TestVarBinary64_Scan(t *testing.T) {
+	tests := []struct {
+		name    string
+		input   interface{}
+		want    string
+		wantErr error
+	}{
+		{
+			name:    "VarBinary64: valid byte slice",
+			input:   []byte{0x12, 0x34, 0xab, 0xcd},
+			want:    "0x1234abcd",
+			wantErr: nil,
+		},
+
+		{
+			name:    "input value is nil",
+			input:   nil,
+			want:    "",
+			wantErr: fmt.Errorf("%w, got %T", ErrExpectedByteSlice, nil),
+		},
+	}
+
+	runScanTests(t, tests, func(input interface{}) (string, error) {
+		var vb VarBinary64
+		err := vb.Scan(input)
+		return string(vb), err
+	})
+}

src/queryanalysis/models/database_details_dto.go

@@ -4,4 +4,5 @@ type DatabaseDetailsDto struct {
 	Name           string `db:"name"`
 	IsQueryStoreOn bool   `db:"is_query_store_on"`
 	Compatibility  int    `db:"compatibility_level"`
+	DatabaseID     int    `db:"database_id"`
 }

src/queryanalysis/models/query_execution_plan.go

@@ -1,27 +1,27 @@
 package models
 
 type ExecutionPlanResult struct {
-	SQLText                *string      `db:"sql_text"`
-	QueryID                *HexString   `db:"query_id"`
-	QueryPlanID            *HexString   `db:"query_plan_id"`
-	NodeID                 *int         `db:"NodeId"`
-	PhysicalOp             *string      `db:"PhysicalOp"`
-	LogicalOp              *string      `db:"LogicalOp"`
-	EstimateRows           *float64     `db:"EstimateRows"`
-	EstimateIO             *float64     `db:"EstimateIO"`
-	EstimateCPU            *float64     `db:"EstimateCPU"`
-	AvgRowSize             *float64     `db:"AvgRowSize"`
-	TotalSubtreeCost       *float64     `db:"TotalSubtreeCost"`
-	EstimatedOperatorCost  *float64     `db:"EstimatedOperatorCost"`
-	EstimatedExecutionMode *string      `db:"EstimatedExecutionMode"`
-	GrantedMemoryKb        *int         `db:"GrantedMemoryKb"`
-	SpillOccurred          *bool        `db:"SpillOccurred"`
-	NoJoinPredicate        *bool        `db:"NoJoinPredicate"`
-	TotalWorkerTime        *int64       `db:"total_worker_time"`
-	TotalElapsedTime       *int64       `db:"total_elapsed_time"`
-	TotalLogicalReads      *int64       `db:"total_logical_reads"`
-	TotalLogicalWrites     *int64       `db:"total_logical_writes"`
-	ExecutionCount         *int64       `db:"execution_count"`
-	PlanHandle             *VarBinary64 `db:"plan_handle"`
-	AvgElapsedTimeMs       *float64     `db:"avg_elapsed_time_ms"`
+	SQLText                *string      `db:"sql_text" metric_name:"sql_text" sourceType:"attribute"`
+	QueryID                *HexString   `db:"query_id" metric_name:"query_id" sourceType:"attribute"`
+	QueryPlanID            *HexString   `db:"query_plan_id" metric_name:"query_plan_id" sourceType:"attribute"`
+	NodeID                 *int         `db:"NodeId" metric_name:"NodeId" sourceType:"gauge"`
+	PhysicalOp             *string      `db:"PhysicalOp" metric_name:"PhysicalOp" sourceType:"attribute"`
+	LogicalOp              *string      `db:"LogicalOp" metric_name:"LogicalOp" sourceType:"attribute"`
+	EstimateRows           *float64     `db:"EstimateRows" metric_name:"EstimateRows" sourceType:"gauge"`
+	EstimateIO             *float64     `db:"EstimateIO" metric_name:"EstimateIO" sourceType:"gauge"`
+	EstimateCPU            *float64     `db:"EstimateCPU" metric_name:"EstimateCPU" sourceType:"gauge"`
+	AvgRowSize             *float64     `db:"AvgRowSize" metric_name:"AvgRowSize" sourceType:"gauge"`
+	TotalSubtreeCost       *float64     `db:"TotalSubtreeCost" metric_name:"TotalSubtreeCost" sourceType:"gauge"`
+	EstimatedOperatorCost  *float64     `db:"EstimatedOperatorCost" metric_name:"EstimatedOperatorCost" sourceType:"gauge"`
+	EstimatedExecutionMode *string      `db:"EstimatedExecutionMode" metric_name:"EstimatedExecutionMode" sourceType:"attribute"`
+	GrantedMemoryKb        *int         `db:"GrantedMemoryKb" metric_name:"GrantedMemoryKb" sourceType:"gauge"`
+	SpillOccurred          *bool        `db:"SpillOccurred" metric_name:"SpillOccurred" sourceType:"attribute"`
+	NoJoinPredicate        *bool        `db:"NoJoinPredicate" metric_name:"NoJoinPredicate" sourceType:"attribute"`
+	TotalWorkerTime        *int64       `db:"total_worker_time" metric_name:"total_worker_time" sourceType:"gauge"`
+	TotalElapsedTime       *int64       `db:"total_elapsed_time" metric_name:"total_elapsed_time" sourceType:"gauge"`
+	TotalLogicalReads      *int64       `db:"total_logical_reads" metric_name:"total_logical_reads" sourceType:"gauge"`
+	TotalLogicalWrites     *int64       `db:"total_logical_writes" metric_name:"total_logical_writes" sourceType:"gauge"`
+	ExecutionCount         *int64       `db:"execution_count" metric_name:"execution_count" sourceType:"gauge"`
+	PlanHandle             *VarBinary64 `db:"plan_handle" metric_name:"plan_handle" sourceType:"attribute"`
+	AvgElapsedTimeMs       *float64     `db:"avg_elapsed_time_ms" metric_name:"avg_elapsed_time_ms" sourceType:"gauge"`
 }

src/queryanalysis/models/top_n_slow_query_details.go

@@ -1,16 +1,16 @@
 package models
 
 type TopNSlowQueryDetails struct {
-	QueryID                *HexString `db:"query_id"`
-	QueryText              *string    `db:"query_text"`
-	DatabaseName           *string    `db:"database_name"`
-	SchemaName             *string    `db:"schema_name"`
-	LastExecutionTimestamp *string    `db:"last_execution_timestamp"`
-	ExecutionCount         *int64     `db:"execution_count"`
-	AvgCPUTimeMS           *float64   `db:"avg_cpu_time_ms"`
-	AvgElapsedTimeMS       *float64   `db:"avg_elapsed_time_ms"`
-	AvgDiskReads           *float64   `db:"avg_disk_reads"`
-	AvgDiskWrites          *float64   `db:"avg_disk_writes"`
-	StatementType          *string    `db:"statement_type"`
-	CollectionTimestamp    *string    `db:"collection_timestamp"`
+	QueryID                *HexString `db:"query_id" metric_name:"query_id" sourceType:"attribute"`
+	QueryText              *string    `db:"query_text" metric_name:"query_text" sourceType:"attribute"`
+	DatabaseName           *string    `db:"database_name" metric_name:"database_name" sourceType:"attribute"`
+	SchemaName             *string    `db:"schema_name" metric_name:"schema_name" sourceType:"attribute"`
+	LastExecutionTimestamp *string    `db:"last_execution_timestamp" metric_name:"last_execution_timestamp" sourceType:"attribute"`
+	ExecutionCount         *int64     `db:"execution_count" metric_name:"execution_count" sourceType:"gauge"`
+	AvgCPUTimeMS           *float64   `db:"avg_cpu_time_ms" metric_name:"avg_cpu_time_ms" sourceType:"gauge"`
+	AvgElapsedTimeMS       *float64   `db:"avg_elapsed_time_ms" metric_name:"avg_elapsed_time_ms" sourceType:"gauge"`
+	AvgDiskReads           *float64   `db:"avg_disk_reads" metric_name:"avg_disk_reads" sourceType:"gauge"`
+	AvgDiskWrites          *float64   `db:"avg_disk_writes" metric_name:"avg_disk_writes" sourceType:"gauge"`
+	StatementType          *string    `db:"statement_type" metric_name:"statement_type" sourceType:"attribute"`
+	CollectionTimestamp    *string    `db:"collection_timestamp" metric_name:"collection_timestamp" sourceType:"attribute"`
 }

src/queryanalysis/models/wait_time_analysis.go

@@ -8,13 +8,13 @@ import (
 
 type WaitTimeAnalysis struct {
 	Connection          *sqlx.DB
-	QueryID             *HexString `db:"query_id" json:"query_id"`
-	DatabaseName        *string    `db:"database_name" json:"database_name"`
-	QueryText           *string    `db:"query_text" json:"query_text"`
-	WaitCategory        *string    `db:"wait_category" json:"wait_category"`
-	TotalWaitTimeMs     *float64   `db:"total_wait_time_ms" json:"total_wait_time_ms"`
-	AvgWaitTimeMs       *float64   `db:"avg_wait_time_ms" json:"avg_wait_time_ms"`
-	WaitEventCount      *int64     `db:"wait_event_count" json:"wait_event_count"`
-	LastExecutionTime   *time.Time `db:"last_execution_time" json:"last_execution_time"`
-	CollectionTimestamp time.Time  `db:"collection_timestamp"`
+	QueryID             *HexString `db:"query_id" json:"query_id" metric_name:"query_id" sourceType:"attribute"`
+	DatabaseName        *string    `db:"database_name" json:"database_name" metric_name:"database_name" sourceType:"attribute"`
+	QueryText           *string    `db:"query_text" json:"query_text" metric_name:"query_text" sourceType:"attribute"`
+	WaitCategory        *string    `db:"wait_category" json:"wait_category" metric_name:"wait_category" sourceType:"attribute"`
+	TotalWaitTimeMs     *float64   `db:"total_wait_time_ms" json:"total_wait_time_ms" metric_name:"total_wait_time_ms" sourceType:"gauge"`
+	AvgWaitTimeMs       *float64   `db:"avg_wait_time_ms" json:"avg_wait_time_ms" metric_name:"avg_wait_time_ms" sourceType:"gauge"`
+	WaitEventCount      *int64     `db:"wait_event_count" json:"wait_event_count" metric_name:"wait_event_count" sourceType:"gauge"`
+	LastExecutionTime   *time.Time `db:"last_execution_time" json:"last_execution_time" metric_name:"last_execution_time" sourceType:"attribute"`
+	CollectionTimestamp time.Time  `db:"collection_timestamp" metric_name:"collection_timestamp" sourceType:"attribute"`
 }

src/queryanalysis/utils/utils.go

@@ -33,7 +33,7 @@ func LoadQueries(arguments args.ArgumentList) ([]models.QueryDetailsDto, error)
 			queries[i].Query = fmt.Sprintf(queries[i].Query, arguments.FetchInterval, arguments.QueryCountThreshold,
 				arguments.QueryResponseTimeThreshold, config.TextTruncateLimit)
 		case "waitAnalysis":
-			queries[i].Query = fmt.Sprintf(queries[i].Query, arguments.FetchInterval, arguments.QueryCountThreshold, config.TextTruncateLimit)
+			queries[i].Query = fmt.Sprintf(queries[i].Query, arguments.QueryCountThreshold, config.TextTruncateLimit)
 		case "blockingSessions":
 			queries[i].Query = fmt.Sprintf(queries[i].Query, arguments.QueryCountThreshold, config.TextTruncateLimit)
 		default:
@@ -49,7 +49,7 @@ func ExecuteQuery(arguments args.ArgumentList, queryDetailsDto models.QueryDetai
 	if err != nil {
 		return nil, fmt.Errorf("failed to execute query: %w", err)
 	}
-
+	defer rows.Close()
 	return BindQueryResults(arguments, rows, queryDetailsDto, integration, sqlConnection)
 }
 
@@ -59,8 +59,6 @@ func BindQueryResults(arguments args.ArgumentList,
 	queryDetailsDto models.QueryDetailsDto,
 	integration *integration.Integration,
 	sqlConnection *connection.SQLConnection) ([]interface{}, error) {
-	defer rows.Close()
-
 	results := make([]interface{}, 0)
 
 	for rows.Next() {
@@ -76,7 +74,9 @@ func BindQueryResults(arguments args.ArgumentList,
 			results = append(results, model)
 
 			// fetch and generate execution plan
-			GenerateAndInjestExecutionPlan(arguments, integration, sqlConnection, *model.QueryID)
+			if model.QueryID != nil {
+				GenerateAndIngestExecutionPlan(arguments, integration, sqlConnection, *model.QueryID)
+			}
 
 		case "waitAnalysis":
 			var model models.WaitTimeAnalysis
@@ -103,7 +103,7 @@ func BindQueryResults(arguments args.ArgumentList,
 	return results, nil
 }
 
-func GenerateAndInjestExecutionPlan(arguments args.ArgumentList,
+func GenerateAndIngestExecutionPlan(arguments args.ArgumentList,
 	integration *integration.Integration,
 	sqlConnection *connection.SQLConnection,
 	queryID models.HexString) {
@@ -240,8 +240,12 @@ func DetectMetricType(value string) metric.SourceType {
 	return metric.GAUGE
 }
 
+var re = regexp.MustCompile(`'[^']*'|\d+|".*?"`)
+
 func AnonymizeQueryText(query *string) {
-	re := regexp.MustCompile(`'[^']*'|\d+|".*?"`)
+	if query == nil {
+		return
+	}
 	anonymizedQuery := re.ReplaceAllString(*query, "?")
 	*query = anonymizedQuery
 }