[receiver/oracledb] Oracle top query collection interval (#44505)

<!--Ex. Fixing a bug - Describe the bug and how this fixes the issue.
Ex. Adding a feature - Explain what this achieves.-->
#### Description
At present, only the SQLServer receiver allows an independent collection
interval to be configured for top query metrics. Because the Oracle
receiver lacks this option, users must configure it twice—separately for
top queries and samples—whenever different collection intervals are
required.

This PR is to do a similar implementation in Oracle receiver.

With this change now the `collection_interval` for top_query_collection
can be configured as below.

```
receivers:
  oracledb:
    datasource: "oracle://system:[email protected]:1521/XEPDB1"
    top_query_collection:
      collection_interval: 60s       <--- new config parameter
      max_query_sample_count: 1000
      top_query_count: 200
```

<!-- Issue number (e.g. #1234) or full URL to issue, if applicable. -->
#### Link to tracking issue
Fixes
#44607

<!--Describe what testing was performed and which tests were added.-->
#### Testing
Unit tests added

<!--Describe the documentation added.-->
#### Documentation

<!--Please delete paragraphs that you did not use before submitting.-->

Author: sv-splunk

.chloggen/oracledb_top_query_collection_interval.yaml

@@ -0,0 +1,27 @@
+# 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. receiver/filelog)
+component: receiver/oracledb
+
+# A brief description of the change.  Surround your text with quotes ("") if it needs to start with a backtick (`).
+note: Added independent collection interval config for Oracle top query metrics collection
+
+# Mandatory: One or more tracking issues related to the change. You can use the PR number here if no issue exists.
+issues: [44607]
+
+# (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:
+
+# 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

@@ -94,3 +94,26 @@ receivers:
         enabled: true
 ```
 
+## Enabling events.
+
+
+The following is a generic configuration that can be used for the default logs and metrics scraped
+by the oracledb receiver.
+
+```yaml
+receivers:
+  oracledb:
+    collection_interval: 10s                     # interval for overall collection
+    datasource: "oracle://otel:password@localhost:51521/XE"
+    events:
+      db.server.query_sample:
+        enabled: true
+      db.server.top_query:
+        enabled: true
+    top_query_collection:                        # this collection exports the most expensive queries as logs
+      max_query_sample_count: 1000               # maximum number of samples collected from db to filter the top N
+      top_query_count: 200                       # The maximum number of queries (N) for which the metrics would be reported
+      collection_interval: 60s                   # collection interval for top query collection specifically
+    query_sample_collection:                     # this collection exports the currently (relate to the query time) executing queries as logs
+      max_rows_per_query: 100                     # the maximum number of samples to bre reported.
+```

receiver/oracledbreceiver/config.go

@@ -9,6 +9,7 @@ import (
 	"net"
 	"net/url"
 	"strconv"
+	"time"
 
 	"go.opentelemetry.io/collector/scraper/scraperhelper"
 	"go.uber.org/multierr"
@@ -29,8 +30,9 @@ var (
 )
 
 type TopQueryCollection struct {
-	MaxQuerySampleCount uint `mapstructure:"max_query_sample_count"`
-	TopQueryCount       uint `mapstructure:"top_query_count"`
+	MaxQuerySampleCount uint          `mapstructure:"max_query_sample_count"`
+	TopQueryCount       uint          `mapstructure:"top_query_count"`
+	CollectionInterval  time.Duration `mapstructure:"collection_interval"`
 }
 
 type QuerySample struct {

receiver/oracledbreceiver/config_test.go

@@ -117,7 +117,7 @@ func TestValidateInvalidConfigs(t *testing.T) {
 
 func TestCreateDefaultConfig(t *testing.T) {
 	cfg := createDefaultConfig().(*Config)
-	assert.Equal(t, 10*time.Second, cfg.CollectionInterval)
+	assert.Equal(t, 10*time.Second, cfg.ControllerConfig.CollectionInterval)
 }
 
 func TestParseConfig(t *testing.T) {

receiver/oracledbreceiver/factory.go

@@ -50,6 +50,7 @@ func createDefaultConfig() component.Config {
 		TopQueryCollection: TopQueryCollection{
 			MaxQuerySampleCount: 1000,
 			TopQueryCount:       200,
+			CollectionInterval:  time.Minute,
 		},
 	}
 }

receiver/oracledbreceiver/scraper.go

@@ -11,6 +11,7 @@ import (
 	"encoding/json"
 	"errors"
 	"fmt"
+	"math"
 	"net"
 	"os"
 	"sort"
@@ -133,6 +134,7 @@ type oracleScraper struct {
 	obfuscator                 *obfuscator
 	querySampleCfg             QuerySample
 	serviceInstanceID          string
+	lastExecutionTimestamp     time.Time
 }
 
 func newScraper(metricsBuilder *metadata.MetricsBuilder, metricsBuilderConfig metadata.MetricsBuilderConfig, scrapeCfg scraperhelper.ControllerConfig, logger *zap.Logger, providerFunc dbProviderFunc, clientProviderFunc clientProviderFunc, instanceName, hostName string) (scraper.Metrics, error) {
@@ -529,9 +531,16 @@ func (s *oracleScraper) scrapeLogs(ctx context.Context) (plog.Logs, error) {
 	var scrapeErrors []error
 
 	if s.logsBuilderConfig.Events.DbServerTopQuery.Enabled {
-		topNCollectionErrors := s.collectTopNMetricData(ctx, logs)
-		if topNCollectionErrors != nil {
-			scrapeErrors = append(scrapeErrors, topNCollectionErrors)
+		currentCollectionTime := time.Now()
+		lookbackTimeCounter := s.calculateLookbackSeconds()
+		if lookbackTimeCounter < int(s.topQueryCollectCfg.CollectionInterval.Seconds()) {
+			s.logger.Debug("Skipping the collection of top queries because collection interval has not yet elapsed.")
+		} else {
+			topNCollectionErrors := s.collectTopNMetricData(ctx, logs, currentCollectionTime, lookbackTimeCounter)
+			if topNCollectionErrors != nil {
+				scrapeErrors = append(scrapeErrors, topNCollectionErrors)
+			}
+			s.lastExecutionTimestamp = currentCollectionTime
 		}
 	}
 
@@ -545,13 +554,11 @@ func (s *oracleScraper) scrapeLogs(ctx context.Context) (plog.Logs, error) {
 	return logs, errors.Join(scrapeErrors...)
 }
 
-func (s *oracleScraper) collectTopNMetricData(ctx context.Context, logs plog.Logs) error {
+func (s *oracleScraper) collectTopNMetricData(ctx context.Context, logs plog.Logs, collectionTime time.Time, lookbackTimeSeconds int) error {
 	var errs []error
 	// get metrics and query texts from DB
-	timestamp := pcommon.NewTimestampFromTime(time.Now())
-	intervalSeconds := int(s.scrapeCfg.CollectionInterval.Seconds())
 	s.oracleQueryMetricsClient = s.clientProviderFunc(s.db, oracleQueryMetricsSQL, s.logger)
-	metricRows, metricError := s.oracleQueryMetricsClient.metricRows(ctx, intervalSeconds, s.topQueryCollectCfg.MaxQuerySampleCount)
+	metricRows, metricError := s.oracleQueryMetricsClient.metricRows(ctx, lookbackTimeSeconds, s.topQueryCollectCfg.MaxQuerySampleCount)
 
 	if metricError != nil {
 		return fmt.Errorf("error executing oracleQueryMetricsSQL: %w", metricError)
@@ -646,7 +653,7 @@ func (s *oracleScraper) collectTopNMetricData(ctx context.Context, logs plog.Log
 		planString := string(planBytes)
 
 		s.lb.RecordDbServerTopQueryEvent(context.Background(),
-			timestamp,
+			pcommon.NewTimestampFromTime(collectionTime),
 			dbSystemNameVal,
 			s.hostName,
 			hit.queryText,
@@ -872,3 +879,17 @@ func constructInstanceID(host, port, service string) string {
 	}
 	return fmt.Sprintf("%s:%s", host, port)
 }
+
+func (s *oracleScraper) calculateLookbackSeconds() int {
+	if s.lastExecutionTimestamp.IsZero() {
+		return int(s.topQueryCollectCfg.CollectionInterval.Seconds())
+	}
+
+	// vsqlRefreshLagSec is the buffer to account for v$sql maximum refresh latency (5 seconds) + 5 seconds to offset any collection delays.
+	// PS: https://docs.oracle.com/en/database/oracle/oracle-database/21/refrn/V-SQL.html
+	const vsqlRefreshLagSec = 10 * time.Second
+
+	return int(math.Ceil(time.Now().
+		Add(vsqlRefreshLagSec).
+		Sub(s.lastExecutionTimestamp).Seconds()))
+}

receiver/oracledbreceiver/scraper_test.go

@@ -12,6 +12,7 @@ import (
 	"path/filepath"
 	"strings"
 	"testing"
+	"time"
 
 	lru "github.com/hashicorp/golang-lru/v2"
 	"github.com/stretchr/testify/assert"
@@ -291,6 +292,7 @@ func TestScraper_ScrapeTopNLogs(t *testing.T) {
 			}()
 			require.NoError(t, err)
 			expectedQueryPlanFile := filepath.Join("testdata", "expectedQueryTextAndPlanQuery.yaml")
+			assert.True(t, scrpr.lastExecutionTimestamp.IsZero(), "No value exists on lastExecutionTimestamp before any collection.")
 
 			logs, err := scrpr.scrapeLogs(t.Context())
 
@@ -304,6 +306,8 @@ func TestScraper_ScrapeTopNLogs(t *testing.T) {
 				assert.NoError(t, errs)
 				assert.Equal(t, "db.server.top_query", logs.ResourceLogs().At(0).ScopeLogs().At(0).LogRecords().At(0).EventName())
 				assert.NoError(t, errs)
+
+				assert.False(t, scrpr.lastExecutionTimestamp.IsZero(), "lastExecutionTimestamp hasn't set after a successful collection.")
 			}
 		})
 	}
@@ -426,6 +430,102 @@ func TestGetInstanceId(t *testing.T) {
 	assert.Equal(t, "unknown:1521", localInstanceID)
 }
 
+func TestScrapesTopNLogsOnlyWhenIntervalHasElapsed(t *testing.T) {
+	var metricRowData []metricRow
+	var logRowData []metricRow
+	tests := []struct {
+		name       string
+		dbclientFn func(db *sql.DB, s string, logger *zap.Logger) dbClient
+		errWanted  string
+	}{
+		{
+			name: "valid collection",
+			dbclientFn: func(_ *sql.DB, s string, _ *zap.Logger) dbClient {
+				if strings.Contains(s, "V$SQL_PLAN") {
+					metricRowFile := readFile("oracleQueryPlanData.txt")
+					unmarshalErr := json.Unmarshal(metricRowFile, &logRowData)
+					if unmarshalErr == nil {
+						return &fakeDbClient{
+							Responses: [][]metricRow{
+								logRowData,
+							},
+						}
+					}
+				} else {
+					metricRowFile := readFile("oracleQueryMetricsData.txt")
+					unmarshalErr := json.Unmarshal(metricRowFile, &metricRowData)
+					if unmarshalErr == nil {
+						return &fakeDbClient{
+							Responses: [][]metricRow{
+								metricRowData,
+							},
+						}
+					}
+				}
+				return nil
+			},
+		},
+	}
+
+	for _, test := range tests {
+		t.Run(test.name, func(t *testing.T) {
+			logsCfg := metadata.DefaultLogsBuilderConfig()
+			logsCfg.Events.DbServerTopQuery.Enabled = true
+			metricsCfg := metadata.DefaultMetricsBuilderConfig()
+			lruCache, _ := lru.New[string, map[string]int64](500)
+			lruCache.Add("fxk8aq3nds8aw:0", cacheValue)
+
+			scrpr := oracleScraper{
+				logger: zap.NewNop(),
+				mb:     metadata.NewMetricsBuilder(metricsCfg, receivertest.NewNopSettings(metadata.Type)),
+				lb:     metadata.NewLogsBuilder(logsCfg, receivertest.NewNopSettings(metadata.Type)),
+				dbProviderFunc: func() (*sql.DB, error) {
+					return nil, nil
+				},
+				clientProviderFunc:   test.dbclientFn,
+				metricsBuilderConfig: metadata.DefaultMetricsBuilderConfig(),
+				logsBuilderConfig:    metadata.DefaultLogsBuilderConfig(),
+				metricCache:          lruCache,
+				topQueryCollectCfg:   TopQueryCollection{MaxQuerySampleCount: 5000, TopQueryCount: 200},
+				obfuscator:           newObfuscator(),
+			}
+
+			scrpr.logsBuilderConfig.Events.DbServerTopQuery.Enabled = true
+			scrpr.topQueryCollectCfg.CollectionInterval = 1 * time.Minute
+
+			err := scrpr.start(t.Context(), componenttest.NewNopHost())
+			defer func() {
+				assert.NoError(t, scrpr.shutdown(t.Context()))
+			}()
+			require.NoError(t, err)
+
+			assert.True(t, scrpr.lastExecutionTimestamp.IsZero(), "No value should be set for lastExecutionTimestamp before a successful collection")
+			logsCol1, _ := scrpr.scrapeLogs(t.Context())
+			assert.Equal(t, 1, logsCol1.ResourceLogs().At(0).ScopeLogs().Len(), "Collection should run when lastExecutionTimestamp is not available")
+			assert.False(t, scrpr.lastExecutionTimestamp.IsZero(), "A value should be set for lastExecutionTimestamp after a successful collection")
+
+			scrpr.lastExecutionTimestamp = scrpr.lastExecutionTimestamp.Add(-10 * time.Second)
+			logsCol2, err := scrpr.scrapeLogs(t.Context())
+			assert.Equal(t, 0, logsCol2.ResourceLogs().Len(), "top_query should not be collected until %s elapsed.", scrpr.topQueryCollectCfg.CollectionInterval.String())
+			require.NoError(t, err)
+		})
+	}
+}
+
+func TestCalculateLookbackSeconds(t *testing.T) {
+	collectionInterval := 20 * time.Second
+	vsqlRefreshLagSec := 10 * time.Second
+	expectedMinimumLookbackTime := int((collectionInterval + vsqlRefreshLagSec).Seconds())
+	currentCollectionTime := time.Now()
+
+	scrpr := oracleScraper{
+		lastExecutionTimestamp: currentCollectionTime.Add(-collectionInterval),
+	}
+	lookbackTime := scrpr.calculateLookbackSeconds()
+
+	assert.LessOrEqual(t, expectedMinimumLookbackTime, lookbackTime, "`lookbackTime` should be minimum %d", expectedMinimumLookbackTime)
+}
+
 func readFile(fname string) []byte {
 	file, err := os.ReadFile(filepath.Join("testdata", fname))
 	if err != nil {