Added implementation of Google Cloud Spanner receiver - 3rd part(stat…

…s readers). (open-telemetry#5684)

This part of implementation contains stats readers implementations.
This stuff will be used for reading data from Google Cloud spanner tables and transforming it to metrics using parsed metrics metadata.
Final integration of all components into receiver will be introduced as separate MR.

receiver/googlecloudspannerreceiver/internal/statsreader/currentstatsreader.go

@@ -0,0 +1,98 @@
+// Copyright  The OpenTelemetry Authors
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//      http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package statsreader
+
+import (
+	"context"
+	"fmt"
+	"time"
+
+	"cloud.google.com/go/spanner"
+	"go.opentelemetry.io/collector/model/pdata"
+	"go.uber.org/zap"
+	"google.golang.org/api/iterator"
+
+	"github.com/open-telemetry/opentelemetry-collector-contrib/receiver/googlecloudspannerreceiver/internal/datasource"
+	"github.com/open-telemetry/opentelemetry-collector-contrib/receiver/googlecloudspannerreceiver/internal/metadata"
+)
+
+type currentStatsReader struct {
+	logger                 *zap.Logger
+	database               *datasource.Database
+	metricsMetadata        *metadata.MetricsMetadata
+	topMetricsQueryMaxRows int
+	statement              func(args statementArgs) spanner.Statement
+}
+
+func newCurrentStatsReader(
+	logger *zap.Logger,
+	database *datasource.Database,
+	metricsMetadata *metadata.MetricsMetadata,
+	config ReaderConfig) *currentStatsReader {
+
+	return &currentStatsReader{
+		logger:                 logger,
+		database:               database,
+		metricsMetadata:        metricsMetadata,
+		statement:              currentStatsStatement,
+		topMetricsQueryMaxRows: config.TopMetricsQueryMaxRows,
+	}
+}
+
+func (reader *currentStatsReader) Name() string {
+	return fmt.Sprintf("%v %v::%v::%v", reader.metricsMetadata.Name, reader.database.DatabaseID().ProjectID(),
+		reader.database.DatabaseID().InstanceID(), reader.database.DatabaseID().DatabaseName())
+}
+
+func (reader *currentStatsReader) Read(ctx context.Context) ([]pdata.Metrics, error) {
+	reader.logger.Debug("Executing read method", zap.String("reader", reader.Name()))
+
+	stmt := reader.newPullStatement()
+
+	return reader.pull(ctx, stmt)
+}
+
+func (reader *currentStatsReader) newPullStatement() spanner.Statement {
+	args := statementArgs{
+		query:                  reader.metricsMetadata.Query,
+		topMetricsQueryMaxRows: reader.topMetricsQueryMaxRows,
+	}
+
+	return reader.statement(args)
+}
+
+func (reader *currentStatsReader) pull(ctx context.Context, stmt spanner.Statement) ([]pdata.Metrics, error) {
+	rowsIterator := reader.database.Client().Single().WithTimestampBound(spanner.ExactStaleness(15*time.Second)).Query(ctx, stmt)
+	defer rowsIterator.Stop()
+
+	var collectedMetrics []pdata.Metrics
+
+	for {
+		row, err := rowsIterator.Next()
+		if err != nil {
+			if err == iterator.Done {
+				return collectedMetrics, nil
+			}
+			return nil, fmt.Errorf("query %q failed with error: %w", stmt.SQL, err)
+		}
+
+		rowMetrics, err := reader.metricsMetadata.RowToMetrics(reader.database.DatabaseID(), row)
+		if err != nil {
+			return nil, fmt.Errorf("query %q failed with error: %w", stmt.SQL, err)
+		}
+
+		collectedMetrics = append(collectedMetrics, rowMetrics...)
+	}
+}

receiver/googlecloudspannerreceiver/internal/statsreader/currentstatsreader_test.go

@@ -0,0 +1,88 @@
+// Copyright  The OpenTelemetry Authors
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//      http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package statsreader
+
+import (
+	"context"
+	"testing"
+
+	"cloud.google.com/go/spanner"
+	"github.com/stretchr/testify/assert"
+	"go.uber.org/zap"
+
+	"github.com/open-telemetry/opentelemetry-collector-contrib/receiver/googlecloudspannerreceiver/internal/datasource"
+	"github.com/open-telemetry/opentelemetry-collector-contrib/receiver/googlecloudspannerreceiver/internal/metadata"
+)
+
+const (
+	projectID    = "ProjectID"
+	instanceID   = "InstanceID"
+	databaseName = "DatabaseName"
+
+	name = "name"
+)
+
+func TestCurrentStatsReader_Name(t *testing.T) {
+	databaseID := datasource.NewDatabaseID(projectID, instanceID, databaseName)
+	ctx := context.Background()
+	client, _ := spanner.NewClient(ctx, "")
+	database := datasource.NewDatabaseFromClient(client, databaseID)
+	metricsMetadata := &metadata.MetricsMetadata{
+		Name: name,
+	}
+
+	reader := currentStatsReader{
+		database:        database,
+		metricsMetadata: metricsMetadata,
+	}
+
+	assert.Equal(t, reader.metricsMetadata.Name+" "+databaseID.ProjectID()+"::"+
+		databaseID.InstanceID()+"::"+databaseID.DatabaseName(), reader.Name())
+}
+
+func TestNewCurrentStatsReader(t *testing.T) {
+	databaseID := datasource.NewDatabaseID(projectID, instanceID, databaseName)
+	ctx := context.Background()
+	client, _ := spanner.NewClient(ctx, "")
+	database := datasource.NewDatabaseFromClient(client, databaseID)
+	metricsMetadata := &metadata.MetricsMetadata{
+		Name: name,
+	}
+	logger := zap.NewNop()
+	config := ReaderConfig{
+		TopMetricsQueryMaxRows: topMetricsQueryMaxRows,
+	}
+
+	reader := newCurrentStatsReader(logger, database, metricsMetadata, config)
+
+	assert.Equal(t, database, reader.database)
+	assert.Equal(t, logger, reader.logger)
+	assert.Equal(t, metricsMetadata, reader.metricsMetadata)
+	assert.Equal(t, topMetricsQueryMaxRows, reader.topMetricsQueryMaxRows)
+}
+
+func TestCurrentStatsReader_NewPullStatement(t *testing.T) {
+	metricsMetadata := &metadata.MetricsMetadata{
+		Query: query,
+	}
+
+	reader := currentStatsReader{
+		metricsMetadata:        metricsMetadata,
+		topMetricsQueryMaxRows: topMetricsQueryMaxRows,
+		statement:              currentStatsStatement,
+	}
+
+	assert.NotZero(t, reader.newPullStatement())
+}

receiver/googlecloudspannerreceiver/internal/statsreader/databasereader.go

@@ -0,0 +1,97 @@
+// Copyright  The OpenTelemetry Authors
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//      http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package statsreader
+
+import (
+	"context"
+	"fmt"
+
+	"go.opentelemetry.io/collector/model/pdata"
+	"go.uber.org/zap"
+
+	"github.com/open-telemetry/opentelemetry-collector-contrib/receiver/googlecloudspannerreceiver/internal/datasource"
+	"github.com/open-telemetry/opentelemetry-collector-contrib/receiver/googlecloudspannerreceiver/internal/metadata"
+)
+
+type DatabaseReader struct {
+	database *datasource.Database
+	logger   *zap.Logger
+	readers  []Reader
+}
+
+func NewDatabaseReader(ctx context.Context,
+	parsedMetadata []*metadata.MetricsMetadata,
+	databaseID *datasource.DatabaseID,
+	serviceAccountPath string,
+	readerConfig ReaderConfig,
+	logger *zap.Logger) (*DatabaseReader, error) {
+
+	database, err := datasource.NewDatabase(ctx, databaseID, serviceAccountPath)
+	if err != nil {
+		return nil, fmt.Errorf("error occurred during client instantiation for database %q: %w", databaseID.ID(), err)
+	}
+
+	readers := initializeReaders(logger, parsedMetadata, database, readerConfig)
+
+	return &DatabaseReader{
+		database: database,
+		logger:   logger,
+		readers:  readers,
+	}, nil
+}
+
+func initializeReaders(logger *zap.Logger, parsedMetadata []*metadata.MetricsMetadata,
+	database *datasource.Database, readerConfig ReaderConfig) []Reader {
+	readers := make([]Reader, len(parsedMetadata))
+
+	for i, mData := range parsedMetadata {
+		switch mData.MetadataType() {
+		case metadata.MetricsMetadataTypeCurrentStats:
+			readers[i] = newCurrentStatsReader(logger, database, mData, readerConfig)
+		case metadata.MetricsMetadataTypeIntervalStats:
+			readers[i] = newIntervalStatsReader(logger, database, mData, readerConfig)
+		}
+	}
+
+	return readers
+}
+
+func (databaseReader *DatabaseReader) Name() string {
+	return databaseReader.database.DatabaseID().ID()
+}
+
+func (databaseReader *DatabaseReader) Shutdown() {
+	databaseReader.logger.Debug("Closing connection to database",
+		zap.String("database", databaseReader.database.DatabaseID().ID()))
+	databaseReader.database.Client().Close()
+}
+
+func (databaseReader *DatabaseReader) Read(ctx context.Context) []pdata.Metrics {
+	databaseReader.logger.Debug("Executing read method for database",
+		zap.String("database", databaseReader.database.DatabaseID().ID()))
+
+	var result []pdata.Metrics
+
+	for _, reader := range databaseReader.readers {
+		if metrics, err := reader.Read(ctx); err != nil {
+			databaseReader.logger.Error("Cannot read data for metrics databaseReader because of error",
+				zap.String("reader", reader.Name()), zap.Error(err))
+		} else {
+			result = append(result, metrics...)
+		}
+	}
+
+	return result
+}