[receiver/oracledb] events for oracledbreceiver - Adding top_query and query_sample collection

Author: sv-splunk

.chloggen/oracledbreceiver-top-n-queries-and-query-samples.yaml

@@ -0,0 +1,31 @@
+# Use this changelog template to create an entry for release notes.
+
+# One of 'breaking', 'deprecation', 'new_component', 'enhancement', 'bug_fix'
+change_type: enhancement
+
+# The name of the component, or a single word describing the area of concern, (e.g. filelogreceiver)
+component: oracledbreceiver
+
+# A brief description of the change.  Surround your text with quotes ("") if it needs to start with a backtick (`).
+note: Support query-level log collection, fetching top N query metrics and details of currently running queries (samples).
+
+# Mandatory: One or more tracking issues related to the change. You can use the PR number here if no issue exists.
+issues: [37478]
+
+# (Optional) One or more lines of additional information to render under the primary note.
+# These lines will be padded with 2 spaces and then inserted directly into the document.
+# Use pipe (|) for multiline entries.
+subtext: |
+  With the introduction of query level log collection, the oracledbreceiver can now send metrics for Top N queries, 
+  filtered based on the highest CPU time consumed. The number of queries can be configured. This helps the user to have
+  a better understanding of the database operations. In addition to this, query samples can be collected for the currently
+  executing queries. This enhancement enables users to proactively monitor and manage the database performance.
+
+# If your change doesn't affect end users or the exported elements of any package,
+# you should instead start your pull request title with [chore] or use the "Skip Changelog" label.
+# Optional: The change log or logs in which this entry should be included.
+# e.g. '[user]' or '[user, api]'
+# Include 'user' if the change is relevant to end users.
+# Include 'api' if there is a change to a library API.
+# Default: '[user]'
+change_logs: [user]

receiver/oracledbreceiver/README.md

@@ -3,12 +3,14 @@
 <!-- status autogenerated section -->
 | Status        |           |
 | ------------- |-----------|
-| Stability     | [alpha]: metrics   |
+| Stability     | [development]: logs   |
+|               | [alpha]: metrics   |
 | Distributions | [contrib] |
 | Issues        | [![Open issues](https://img.shields.io/github/issues-search/open-telemetry/opentelemetry-collector-contrib?query=is%3Aissue%20is%3Aopen%20label%3Areceiver%2Foracledb%20&label=open&color=orange&logo=opentelemetry)](https://github.com/open-telemetry/opentelemetry-collector-contrib/issues?q=is%3Aopen+is%3Aissue+label%3Areceiver%2Foracledb) [![Closed issues](https://img.shields.io/github/issues-search/open-telemetry/opentelemetry-collector-contrib?query=is%3Aissue%20is%3Aclosed%20label%3Areceiver%2Foracledb%20&label=closed&color=blue&logo=opentelemetry)](https://github.com/open-telemetry/opentelemetry-collector-contrib/issues?q=is%3Aclosed+is%3Aissue+label%3Areceiver%2Foracledb) |
 | Code coverage | [![codecov](https://codecov.io/github/open-telemetry/opentelemetry-collector-contrib/graph/main/badge.svg?component=receiver_oracledb)](https://app.codecov.io/gh/open-telemetry/opentelemetry-collector-contrib/tree/main/?components%5B0%5D=receiver_oracledb&displayType=list) |
 | [Code Owners](https://github.com/open-telemetry/opentelemetry-collector-contrib/blob/main/CONTRIBUTING.md#becoming-a-code-owner)    | [@dmitryax](https://www.github.com/dmitryax), [@crobert-1](https://www.github.com/crobert-1), [@atoulme](https://www.github.com/atoulme) |
 
+[development]: https://github.com/open-telemetry/opentelemetry-collector/blob/main/docs/component-stability.md#development
 [alpha]: https://github.com/open-telemetry/opentelemetry-collector/blob/main/docs/component-stability.md#alpha
 [contrib]: https://github.com/open-telemetry/opentelemetry-collector-releases/tree/main/distributions/otelcol-contrib
 <!-- end autogenerated section -->

receiver/oracledbreceiver/config.go

@@ -17,15 +17,24 @@ import (
 )
 
 var (
-	errBadDataSource = errors.New("datasource is invalid")
-	errBadEndpoint   = errors.New("endpoint must be specified as host:port")
-	errBadPort       = errors.New("invalid port in endpoint")
-	errEmptyEndpoint = errors.New("endpoint must be specified")
-	errEmptyPassword = errors.New("password must be set")
-	errEmptyService  = errors.New("service must be specified")
-	errEmptyUsername = errors.New("username must be set")
+	errBadDataSource       = errors.New("datasource is invalid")
+	errBadEndpoint         = errors.New("endpoint must be specified as host:port")
+	errBadPort             = errors.New("invalid port in endpoint")
+	errEmptyEndpoint       = errors.New("endpoint must be specified")
+	errEmptyPassword       = errors.New("password must be set")
+	errEmptyService        = errors.New("service must be specified")
+	errEmptyUsername       = errors.New("username must be set")
+	errMaxQuerySampleCount = errors.New("`max_query_sample_count` must be between 1 and 10000")
+	errTopQueryCount       = errors.New("`top_query_count` must be between 1 and 200 and less than or equal to `max_query_sample_count`")
+	errQueryCacheSize      = errors.New("`query_cache_size` must be strictly positive")
 )
 
+type TopQueryCollection struct {
+	MaxQuerySampleCount uint `mapstructure:"max_query_sample_count"`
+	TopQueryCount       uint `mapstructure:"top_query_count"`
+	QueryCacheSize      int  `mapstructure:"query_cache_size"`
+}
+
 type Config struct {
 	DataSource                     string `mapstructure:"datasource"`
 	Endpoint                       string `mapstructure:"endpoint"`
@@ -34,6 +43,9 @@ type Config struct {
 	Username                       string `mapstructure:"username"`
 	scraperhelper.ControllerConfig `mapstructure:",squash"`
 	metadata.MetricsBuilderConfig  `mapstructure:",squash"`
+	metadata.LogsBuilderConfig     `mapstructure:",squash"`
+
+	TopQueryCollection `mapstructure:"top_query_collection"`
 }
 
 func (c Config) Validate() error {
@@ -79,5 +91,15 @@ func (c Config) Validate() error {
 			allErrs = multierr.Append(allErrs, fmt.Errorf("%w: %s", errBadDataSource, err.Error()))
 		}
 	}
+
+	if c.MaxQuerySampleCount < 1 || c.MaxQuerySampleCount > 10000 {
+		allErrs = multierr.Append(allErrs, errMaxQuerySampleCount)
+	}
+	if c.TopQueryCount < 1 || c.TopQueryCount > 200 || c.TopQueryCount > c.MaxQuerySampleCount {
+		allErrs = multierr.Append(allErrs, errTopQueryCount)
+	}
+	if c.QueryCacheSize <= 0 {
+		allErrs = multierr.Append(allErrs, errQueryCacheSize)
+	}
 	return allErrs
 }

receiver/oracledbreceiver/db_client.go

@@ -16,7 +16,7 @@ import (
 )
 
 type dbClient interface {
-	metricRows(ctx context.Context) ([]metricRow, error)
+	metricRows(ctx context.Context, args ...any) ([]metricRow, error)
 }
 
 type metricRow map[string]string
@@ -35,8 +35,8 @@ func newDbClient(db *sql.DB, sql string, logger *zap.Logger) dbClient {
 	}
 }
 
-func (cl dbSQLClient) metricRows(ctx context.Context) ([]metricRow, error) {
-	sqlRows, err := cl.db.QueryContext(ctx, cl.sql)
+func (cl dbSQLClient) metricRows(ctx context.Context, args ...any) ([]metricRow, error) {
+	sqlRows, err := cl.db.QueryContext(ctx, cl.sql, args...)
 	if err != nil {
 		return nil, err
 	}

receiver/oracledbreceiver/documentation.md

@@ -369,8 +369,90 @@ Number of SELECT statements executed in parallel
 | ---- | ----------- | ---------- | ----------------------- | --------- |
 | {queries} | Sum | Int | Cumulative | true |
 
+## Default Events
+
+The following events are emitted by default. Each of them can be disabled by applying the following configuration:
+
+```yaml
+events:
+  <event_name>:
+    enabled: false
+```
+
+## Optional Events
+
+The following events are not emitted by default. Each of them can be enabled by applying the following configuration:
+
+```yaml
+events:
+  <event_name>:
+    enabled: true
+```
+
+### db.server.query_sample
+
+sample query
+
+#### Attributes
+
+| Name | Description | Values |
+| ---- | ----------- | ------ |
+| db.query.text | The text of the database query being executed. | Any Str |
+| db.system.name | The database management system (DBMS) product as identified by the client instrumentation. | Any Str |
+| user.name | Database user name under which a session is connected to | Any Str |
+| client.address | Identifies the name of the machine (host) from which the database client is connecting to the Oracle database. | Any Str |
+| oracledb.plan_hash_value | Binary hash value calculated on the query execution plan and used to identify similar query execution plans, reported in the HEX format. | Any Str |
+| oracledb.sql_id | The SQL ID of the query. | Any Str |
+| oracledb.child_number | The child number of the query. | Any Str |
+| oracledb.sid | ID of the Oracle Server session. | Any Str |
+| oracledb.serial | Serial number associated with a session. | Any Str |
+| oracledb.process | The operating system process ID (PID) associated with a session. | Any Str |
+| oracledb.schemaname | Oracle schema under which SQL statements are being executed | Any Str |
+| oracledb.program | Name of the client program or tool that initiated the Oracle database session. | Any Str |
+| oracledb.module | Logical module name of the client application that initiated a query or session. | Any Str |
+| oracledb.status | Execution state or result of a database query or session. | Any Str |
+| oracledb.state | Current state of the query or the session executing it. | Any Str |
+| oracledb.wait_class | The category of wait events a query or session is currently experiencing in Oracle Database. | Any Str |
+| oracledb.event | The specific wait event that a query or session is currently experiencing. | Any Str |
+| oracledb.object_name | Name of the database object that a query is accessing. | Any Str |
+| oracledb.object_type | Type of the database object that a query is accessing. | Any Str |
+| oracledb.osuser | Name of the operating system user that initiated or is running the Oracle database session. | Any Str |
+| oracledb.duration_sec | Total time taken by a database query to execute. | Any Double |
+
+### db.server.top_query
+
+top query
+
+#### Attributes
+
+| Name | Description | Values |
+| ---- | ----------- | ------ |
+| db.system.name | The database management system (DBMS) product as identified by the client instrumentation. | Any Str |
+| db.server.name | The name of the server hosting the database. | Any Str |
+| db.query.text | The text of the database query being executed. | Any Str |
+| oracledb.query_plan | The query execution plan used by the SQL Server. | Any Str |
+| oracledb.sql_id | The SQL ID of the query. | Any Str |
+| oracledb.child_number | The child number of the query. | Any Str |
+| oracledb.application_wait_time | The total time (in seconds) a query spent waiting on the application before it could proceed with execution (reporting delta). | Any Double |
+| oracledb.buffer_gets | Number of logical reads (i.e., buffer cache accesses) performed by a query (reporting delta). | Any Int |
+| oracledb.cluster_wait_time | Total time (in seconds) that a query waited due to Oracle Real Application Clusters (RAC) coordination (reporting delta). | Any Double |
+| oracledb.concurrency_wait_time | Total time (in seconds) a query spent waiting on concurrency-related events (reporting delta). | Any Double |
+| oracledb.cpu_time | Total time (in seconds) that the CPU spent actively processing a query, excluding time spent waiting (reporting delta). | Any Double |
+| oracledb.direct_reads | The number of direct path reads performed by a query — i.e., data blocks read directly from disk into the session’s memory (reporting delta). | Any Int |
+| oracledb.direct_writes | The number of direct path write operations, where data is written directly to disk from user memory (reporting delta). | Any Int |
+| oracledb.disk_reads | The number of physical reads a query performs — that is, the number of data blocks read from disk (reporting delta). | Any Int |
+| oracledb.elapsed_time | The total time (in seconds) taken by a query from start to finish, including CPU time and all types of waits (reporting delta). | Any Double |
+| oracledb.executions | The number of times a specific SQL query has been executed (reporting delta). | Any Int |
+| oracledb.physical_read_bytes | The total number of bytes read from disk by a query (reporting delta). | Any Int |
+| oracledb.physical_read_requests | The number of physical I/O read operations performed by a query (reporting delta). | Any Int |
+| oracledb.physical_write_bytes | The total number of bytes written to disk by a query (reporting delta). | Any Int |
+| oracledb.physical_write_requests | The number of times a query requested to write data to disk (reporting delta). | Any Int |
+| oracledb.rows_processed | The total number of rows that a query has read, returned, or affected during its execution (reporting delta). | Any Int |
+| oracledb.user_io_wait_time | The total time (in seconds) a query spent waiting for user I/O operations—such as reading or writing data to disk or network file systems (reporting delta). | Any Double |
+
 ## Resource Attributes
 
 | Name | Description | Values | Enabled |
 | ---- | ----------- | ------ | ------- |
+| host.name | The host name of Oracle Server | Any Str | true |
 | oracledb.instance.name | The name of the instance that data is coming from. | Any Str | true |

receiver/oracledbreceiver/factory.go

@@ -11,11 +11,14 @@ import (
 	"strconv"
 	"time"
 
+	lru "github.com/hashicorp/golang-lru/v2"
 	go_ora "github.com/sijms/go-ora/v2"
 	"go.opentelemetry.io/collector/component"
 	"go.opentelemetry.io/collector/consumer"
 	"go.opentelemetry.io/collector/receiver"
+	"go.opentelemetry.io/collector/scraper"
 	"go.opentelemetry.io/collector/scraper/scraperhelper"
+	"go.uber.org/zap"
 
 	"github.com/open-telemetry/opentelemetry-collector-contrib/receiver/oracledbreceiver/internal/metadata"
 )
@@ -27,7 +30,10 @@ func NewFactory() receiver.Factory {
 		createDefaultConfig,
 		receiver.WithMetrics(createReceiverFunc(func(dataSourceName string) (*sql.DB, error) {
 			return sql.Open("oracle", dataSourceName)
-		}, newDbClient), metadata.MetricsStability))
+		}, newDbClient), metadata.MetricsStability),
+		receiver.WithLogs(createLogsReceiverFunc(func(dataSourceName string) (*sql.DB, error) {
+			return sql.Open("oracle", dataSourceName)
+		}, newDbClient), metadata.LogsStability))
 }
 
 func createDefaultConfig() component.Config {
@@ -37,6 +43,12 @@ func createDefaultConfig() component.Config {
 	return &Config{
 		ControllerConfig:     cfg,
 		MetricsBuilderConfig: metadata.DefaultMetricsBuilderConfig(),
+		LogsBuilderConfig:    metadata.DefaultLogsBuilderConfig(),
+		TopQueryCollection: TopQueryCollection{
+			MaxQuerySampleCount: 1000,
+			TopQueryCount:       200,
+			QueryCacheSize:      5000,
+		},
 	}
 }
 
@@ -56,10 +68,14 @@ func createReceiverFunc(sqlOpenerFunc sqlOpenerFunc, clientProviderFunc clientPr
 		if err != nil {
 			return nil, err
 		}
+		hostName, hostNameErr := getHostName(getDataSource(*sqlCfg))
+		if hostNameErr != nil {
+			return nil, hostNameErr
+		}
 
 		mp, err := newScraper(metricsBuilder, sqlCfg.MetricsBuilderConfig, sqlCfg.ControllerConfig, settings.Logger, func() (*sql.DB, error) {
 			return sqlOpenerFunc(getDataSource(*sqlCfg))
-		}, clientProviderFunc, instanceName)
+		}, clientProviderFunc, instanceName, hostName)
 		if err != nil {
 			return nil, err
 		}
@@ -74,6 +90,56 @@ func createReceiverFunc(sqlOpenerFunc sqlOpenerFunc, clientProviderFunc clientPr
 	}
 }
 
+func createLogsReceiverFunc(sqlOpenerFunc sqlOpenerFunc, clientProviderFunc clientProviderFunc) receiver.CreateLogsFunc {
+	return func(
+		_ context.Context,
+		settings receiver.Settings,
+		cfg component.Config,
+		logsConsumer consumer.Logs,
+	) (receiver.Logs, error) {
+		sqlCfg := cfg.(*Config)
+
+		logsBuilder := metadata.NewLogsBuilder(sqlCfg.LogsBuilderConfig, settings)
+
+		instanceName, err := getInstanceName(getDataSource(*sqlCfg))
+		if err != nil {
+			return nil, err
+		}
+
+		hostName, hostNameErr := getHostName(getDataSource(*sqlCfg))
+		if hostNameErr != nil {
+			return nil, hostNameErr
+		}
+
+		cacheSize := sqlCfg.QueryCacheSize
+		metricCache, err := lru.New[string, map[string]int64](cacheSize)
+		if err != nil {
+			settings.Logger.Error("Failed to create LRU cache, skipping the current scraper", zap.Error(err))
+			return nil, err
+		}
+
+		mp, err := newLogsScraper(logsBuilder, sqlCfg.LogsBuilderConfig, sqlCfg.ControllerConfig, settings.Logger, func() (*sql.DB, error) {
+			return sqlOpenerFunc(getDataSource(*sqlCfg))
+		}, clientProviderFunc, instanceName, metricCache, sqlCfg.TopQueryCollection, hostName)
+		if err != nil {
+			return nil, err
+		}
+
+		f := scraper.NewFactory(metadata.Type, nil,
+			scraper.WithLogs(func(context.Context, scraper.Settings, component.Config) (scraper.Logs, error) {
+				return mp, nil
+			}, component.StabilityLevelAlpha))
+		opt := scraperhelper.AddFactoryWithConfig(f, nil)
+
+		return scraperhelper.NewLogsController(
+			&sqlCfg.ControllerConfig,
+			settings,
+			logsConsumer,
+			opt,
+		)
+	}
+}
+
 func getDataSource(cfg Config) string {
 	if cfg.DataSource != "" {
 		return cfg.DataSource
@@ -95,3 +161,11 @@ func getInstanceName(datasource string) (string, error) {
 	instanceName := datasourceURL.Host + datasourceURL.Path
 	return instanceName, nil
 }
+
+func getHostName(datasource string) (string, error) {
+	datasourceURL, err := url.Parse(datasource)
+	if err != nil {
+		return "", err
+	}
+	return datasourceURL.Host, nil
+}

receiver/oracledbreceiver/factory_test.go

@@ -30,6 +30,19 @@ func TestNewFactory(t *testing.T) {
 		consumertest.NewNop(),
 	)
 	require.NoError(t, err)
+
+	config := factory.CreateDefaultConfig().(*Config)
+	_, logsErr := factory.CreateLogs(
+		context.Background(),
+		receiver.Settings{
+			ID:                component.NewID(metadata.Type),
+			TelemetrySettings: componenttest.NewNopTelemetrySettings(),
+		},
+		config,
+
+		consumertest.NewNop(),
+	)
+	require.NoError(t, logsErr)
 }
 
 func TestGetInstanceName(t *testing.T) {

receiver/oracledbreceiver/fake_db_client.go

@@ -13,7 +13,7 @@ type fakeDbClient struct {
 	RequestCounter int
 }
 
-func (c *fakeDbClient) metricRows(context.Context) ([]metricRow, error) {
+func (c *fakeDbClient) metricRows(context.Context, ...any) ([]metricRow, error) {
 	if c.Err != nil {
 		return nil, c.Err
 	}