Add dynamodb metrics (#14518)

* Add dynamodb metrics

Adds promethues counters for dynamo api requests, requests failed,
and latency of requests. Each counter is labeled with the table type,
either backend or events, and the operation being requested.

docs/pages/setup/reference/metrics.mdx

@@ -102,6 +102,9 @@ Teleport Cloud does not expose monitoring endpoints for the Auth Service and Pro
 | `backend_write_requests_total` | counter | cache | Number of write requests to the backend. |
 | `backend_write_seconds` | histogram | cache | Latency for backend write operations. |
 | `cluster_name_not_found_total` | counter | Teleport Auth | Number of times a cluster was not found. |
+| `dynamo_requests_total` | counter | dynamodb | Number of requests to the DynamoDB api. |
+| `dynamo_requests` | counter | dynamodb |  Number of failed requests to the DynamoDB api. |
+| `dynamo_requests_seconds` | histogram | dynamodb | Latency of DynamoDB api requests. |
 | `etcd_backend_batch_read_requests` | counter | etcd | Number of read requests to the etcd database. |
 | `etcd_backend_batch_read_seconds` | histogram | etcd | Latency for etcd read operations. |
 | `etcd_backend_read_requests` | counter | etcd | Number of read requests to the etcd database. |

lib/backend/dynamo/configure.go

@@ -24,10 +24,11 @@ import (
 	"github.com/aws/aws-sdk-go/aws"
 	"github.com/aws/aws-sdk-go/service/applicationautoscaling"
 	"github.com/aws/aws-sdk-go/service/dynamodb"
+	"github.com/aws/aws-sdk-go/service/dynamodb/dynamodbiface"
 )
 
 // SetContinuousBackups enables continuous backups.
-func SetContinuousBackups(ctx context.Context, svc *dynamodb.DynamoDB, tableName string) error {
+func SetContinuousBackups(ctx context.Context, svc dynamodbiface.DynamoDBAPI, tableName string) error {
 	// Make request to AWS to update continuous backups settings.
 	_, err := svc.UpdateContinuousBackupsWithContext(ctx, &dynamodb.UpdateContinuousBackupsInput{
 		PointInTimeRecoverySpecification: &dynamodb.PointInTimeRecoverySpecification{

lib/backend/dynamo/doc.go

@@ -13,7 +13,7 @@
 // limitations under the License.
 
 /*
-Package dynamodbDynamoDBBackend implements DynamoDB storage backend
+Package dynamo implements DynamoDB storage backend
 for Teleport auth service, similar to etcd backend.
 
 dynamo package implements the DynamoDB storage back-end for the

lib/backend/dynamo/dynamo.go

@@ -1,17 +0,0 @@
-/*
-Copyright 2015-2020 Gravitational, Inc.
-
-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 dynamo

lib/backend/dynamo/dynamodbbk.go

@@ -29,6 +29,7 @@ import (
 	"github.com/gravitational/teleport/api/utils"
 	"github.com/gravitational/teleport/lib/backend"
 	"github.com/gravitational/teleport/lib/defaults"
+	dynamometrics "github.com/gravitational/teleport/lib/observability/metrics/dynamo"
 
 	"github.com/aws/aws-sdk-go/aws"
 	"github.com/aws/aws-sdk-go/aws/awserr"
@@ -37,7 +38,9 @@ import (
 	"github.com/aws/aws-sdk-go/service/applicationautoscaling"
 	"github.com/aws/aws-sdk-go/service/dynamodb"
 	"github.com/aws/aws-sdk-go/service/dynamodb/dynamodbattribute"
+	"github.com/aws/aws-sdk-go/service/dynamodb/dynamodbiface"
 	"github.com/aws/aws-sdk-go/service/dynamodbstreams"
+	"github.com/aws/aws-sdk-go/service/dynamodbstreams/dynamodbstreamsiface"
 	"github.com/gravitational/trace"
 	"github.com/jonboulle/clockwork"
 	log "github.com/sirupsen/logrus"
@@ -122,8 +125,8 @@ func (cfg *Config) CheckAndSetDefaults() error {
 type Backend struct {
 	*log.Entry
 	Config
-	svc              *dynamodb.DynamoDB
-	streams          *dynamodbstreams.DynamoDBStreams
+	svc              dynamodbiface.DynamoDBAPI
+	streams          dynamodbstreamsiface.DynamoDBStreamsAPI
 	clock            clockwork.Clock
 	buf              *backend.CircularBuffer
 	ctx              context.Context
@@ -257,8 +260,16 @@ func New(ctx context.Context, params backend.Params) (*Backend, error) {
 	b.session.Config.HTTPClient = httpClient
 
 	// create DynamoDB service:
-	b.svc = dynamodb.New(b.session)
-	b.streams = dynamodbstreams.New(b.session)
+	svc, err := dynamometrics.NewAPIMetrics(dynamometrics.Backend, dynamodb.New(b.session))
+	if err != nil {
+		return nil, trace.Wrap(err)
+	}
+	b.svc = svc
+	streams, err := dynamometrics.NewStreamsMetricsAPI(dynamometrics.Backend, dynamodbstreams.New(b.session))
+	if err != nil {
+		return nil, trace.Wrap(err)
+	}
+	b.streams = streams
 
 	// check if the table exists?
 	ts, err := b.getTableStatus(ctx, b.TableName)

lib/events/dynamoevents/dynamoevents.go

@@ -30,6 +30,8 @@ import (
 	"strings"
 	"time"
 
+	"github.com/aws/aws-sdk-go/aws/request"
+
 	"github.com/gravitational/teleport"
 	apidefaults "github.com/gravitational/teleport/api/defaults"
 	"github.com/gravitational/teleport/api/types"
@@ -37,6 +39,7 @@ import (
 	"github.com/gravitational/teleport/lib/backend"
 	"github.com/gravitational/teleport/lib/backend/dynamo"
 	"github.com/gravitational/teleport/lib/events"
+	dynamometrics "github.com/gravitational/teleport/lib/observability/metrics/dynamo"
 	"github.com/gravitational/teleport/lib/session"
 	"github.com/gravitational/teleport/lib/utils"
 
@@ -46,6 +49,7 @@ import (
 	"github.com/aws/aws-sdk-go/service/applicationautoscaling"
 	"github.com/aws/aws-sdk-go/service/dynamodb"
 	"github.com/aws/aws-sdk-go/service/dynamodb/dynamodbattribute"
+	"github.com/aws/aws-sdk-go/service/dynamodb/dynamodbiface"
 	"github.com/google/uuid"
 	"github.com/gravitational/trace"
 	"github.com/jonboulle/clockwork"
@@ -193,7 +197,7 @@ type Log struct {
 	*log.Entry
 	// Config is a backend configuration
 	Config
-	svc *dynamodb.DynamoDB
+	svc dynamodbiface.DynamoDBAPI
 
 	// session holds the AWS client.
 	session *awssession.Session
@@ -291,7 +295,11 @@ func New(ctx context.Context, cfg Config, backend backend.Backend) (*Log, error)
 	b.session.Config.UseFIPSEndpoint = events.FIPSProtoStateToAWSState(cfg.UseFIPSEndpoint)
 
 	// create DynamoDB service:
-	b.svc = dynamodb.New(b.session)
+	svc, err := dynamometrics.NewAPIMetrics(dynamometrics.Events, dynamodb.New(b.session))
+	if err != nil {
+		return nil, trace.Wrap(err)
+	}
+	b.svc = svc
 
 	// check if the table exists?
 	ts, err := b.getTableStatus(ctx, b.Tablename)
@@ -548,11 +556,11 @@ type checkpointKey struct {
 //
 // This function may never return more than 1 MiB of event data.
 func (l *Log) SearchEvents(fromUTC, toUTC time.Time, namespace string, eventTypes []string, limit int, order types.EventOrder, startKey string) ([]apievents.AuditEvent, string, error) {
-	return l.searchEventsWithFilter(fromUTC, toUTC, namespace, limit, order, startKey, searchEventsFilter{eventTypes: eventTypes}, "")
+	return l.searchEventsWithFilter(context.TODO(), fromUTC, toUTC, namespace, limit, order, startKey, searchEventsFilter{eventTypes: eventTypes}, "")
 }
 
-func (l *Log) searchEventsWithFilter(fromUTC, toUTC time.Time, namespace string, limit int, order types.EventOrder, startKey string, filter searchEventsFilter, sessionID string) ([]apievents.AuditEvent, string, error) {
-	rawEvents, lastKey, err := l.searchEventsRaw(fromUTC, toUTC, namespace, limit, order, startKey, filter, sessionID)
+func (l *Log) searchEventsWithFilter(ctx context.Context, fromUTC, toUTC time.Time, namespace string, limit int, order types.EventOrder, startKey string, filter searchEventsFilter, sessionID string) ([]apievents.AuditEvent, string, error) {
+	rawEvents, lastKey, err := l.searchEventsRaw(ctx, fromUTC, toUTC, namespace, limit, order, startKey, filter, sessionID)
 	if err != nil {
 		return nil, "", trace.Wrap(err)
 	}
@@ -628,7 +636,7 @@ func reverseStrings(slice []string) []string {
 
 // searchEventsRaw is a low level function for searching for events. This is kept
 // separate from the SearchEvents function in order to allow tests to grab more metadata.
-func (l *Log) searchEventsRaw(fromUTC, toUTC time.Time, namespace string, limit int, order types.EventOrder, startKey string, filter searchEventsFilter, sessionID string) ([]event, string, error) {
+func (l *Log) searchEventsRaw(ctx context.Context, fromUTC, toUTC time.Time, namespace string, limit int, order types.EventOrder, startKey string, filter searchEventsFilter, sessionID string) ([]event, string, error) {
 	checkpoint, err := getCheckpointFromStartKey(startKey)
 	if err != nil {
 		return nil, "", trace.Wrap(err)
@@ -701,12 +709,12 @@ func (l *Log) searchEventsRaw(fromUTC, toUTC time.Time, namespace string, limit
 
 	var values []event
 	if fromUTC.IsZero() && sessionID != "" {
-		values, err = ef.QueryBySessionIDIndex(sessionID, filterExpr)
+		values, err = ef.QueryBySessionIDIndex(ctx, sessionID, filterExpr)
 		if err != nil {
 			return nil, "", trace.Wrap(err)
 		}
 	} else {
-		values, err = ef.QueryByDateIndex(filterExpr)
+		values, err = ef.QueryByDateIndex(ctx, filterExpr)
 		if err != nil {
 			return nil, "", trace.Wrap(err)
 		}
@@ -774,7 +782,7 @@ func (l *Log) SearchSessionEvents(fromUTC, toUTC time.Time, limit int, order typ
 		filter.condExpr = expr
 		filter.condParams = params
 	}
-	return l.searchEventsWithFilter(fromUTC, toUTC, apidefaults.Namespace, limit, order, startKey, filter, sessionID)
+	return l.searchEventsWithFilter(context.TODO(), fromUTC, toUTC, apidefaults.Namespace, limit, order, startKey, filter, sessionID)
 }
 
 type searchEventsFilter struct {
@@ -1091,7 +1099,7 @@ func (l *Log) StreamSessionEvents(ctx context.Context, sessionID session.ID, sta
 }
 
 type query interface {
-	Query(input *dynamodb.QueryInput) (*dynamodb.QueryOutput, error)
+	QueryWithContext(ctx context.Context, input *dynamodb.QueryInput, opts ...request.Option) (*dynamodb.QueryOutput, error)
 }
 
 type eventsFetcher struct {
@@ -1174,7 +1182,7 @@ func (l *eventsFetcher) processQueryOutput(output *dynamodb.QueryOutput, hasLeft
 	return out, false, nil
 }
 
-func (l *eventsFetcher) QueryByDateIndex(filterExpr *string) (values []event, err error) {
+func (l *eventsFetcher) QueryByDateIndex(ctx context.Context, filterExpr *string) (values []event, err error) {
 	query := "CreatedAtDate = :date AND CreatedAt BETWEEN :start and :end"
 	var attributeNames map[string]*string
 	if len(l.filter.condParams.attrNames) > 0 {
@@ -1213,7 +1221,7 @@ dateLoop:
 				ScanIndexForward:          aws.Bool(l.forward),
 			}
 			start := time.Now()
-			out, err := l.api.Query(&input)
+			out, err := l.api.QueryWithContext(ctx, &input)
 			if err != nil {
 				return nil, trace.Wrap(err)
 			}
@@ -1243,7 +1251,7 @@ dateLoop:
 	return values, nil
 }
 
-func (l *eventsFetcher) QueryBySessionIDIndex(sessionID string, filterExpr *string) (values []event, err error) {
+func (l *eventsFetcher) QueryBySessionIDIndex(ctx context.Context, sessionID string, filterExpr *string) (values []event, err error) {
 	query := "SessionID = :id"
 	var attributeNames map[string]*string
 	if len(l.filter.condParams.attrNames) > 0 {
@@ -1275,7 +1283,7 @@ func (l *eventsFetcher) QueryBySessionIDIndex(sessionID string, filterExpr *stri
 		ScanIndexForward:          aws.Bool(l.forward),
 	}
 	start := time.Now()
-	out, err := l.api.Query(&input)
+	out, err := l.api.QueryWithContext(ctx, &input)
 	if err != nil {
 		return nil, trace.Wrap(err)
 	}

lib/observability/metrics/dynamo/api.go

@@ -0,0 +1,154 @@
+// Copyright 2022 Gravitational, Inc
+//
+// 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 dynamo
+
+import (
+	"context"
+	"time"
+
+	"github.com/aws/aws-sdk-go/aws/request"
+	"github.com/aws/aws-sdk-go/service/dynamodb"
+	"github.com/aws/aws-sdk-go/service/dynamodb/dynamodbiface"
+	"github.com/gravitational/trace"
+
+	"github.com/gravitational/teleport/lib/utils"
+)
+
+// APIMetrics wraps a dynamodbiface.DynamoDBAPI implementation and
+// reports statistics about the dynamo api operations
+type APIMetrics struct {
+	dynamodbiface.DynamoDBAPI
+	tableType TableType
+}
+
+// NewAPIMetrics returns a new APIMetrics for the provided TableType
+func NewAPIMetrics(tableType TableType, api dynamodbiface.DynamoDBAPI) (*APIMetrics, error) {
+	if err := utils.RegisterPrometheusCollectors(dynamoCollectors...); err != nil {
+		return nil, trace.Wrap(err)
+	}
+
+	return &APIMetrics{
+		DynamoDBAPI: api,
+		tableType:   tableType,
+	}, nil
+}
+
+func (m *APIMetrics) DescribeTimeToLiveWithContext(ctx context.Context, input *dynamodb.DescribeTimeToLiveInput, opts ...request.Option) (*dynamodb.DescribeTimeToLiveOutput, error) {
+	start := time.Now()
+	output, err := m.DynamoDBAPI.DescribeTimeToLiveWithContext(ctx, input, opts...)
+
+	recordMetrics(m.tableType, "describe_ttl", err, time.Since(start).Seconds())
+
+	return output, err
+}
+
+func (m *APIMetrics) UpdateTimeToLiveWithContext(ctx context.Context, input *dynamodb.UpdateTimeToLiveInput, opts ...request.Option) (*dynamodb.UpdateTimeToLiveOutput, error) {
+	start := time.Now()
+	output, err := m.DynamoDBAPI.UpdateTimeToLiveWithContext(ctx, input, opts...)
+
+	recordMetrics(m.tableType, "update_ttl", err, time.Since(start).Seconds())
+
+	return output, err
+}
+
+func (m *APIMetrics) DeleteItemWithContext(ctx context.Context, input *dynamodb.DeleteItemInput, opts ...request.Option) (*dynamodb.DeleteItemOutput, error) {
+	start := time.Now()
+	output, err := m.DynamoDBAPI.DeleteItemWithContext(ctx, input, opts...)
+
+	recordMetrics(m.tableType, "delete_item", err, time.Since(start).Seconds())
+
+	return output, err
+}
+
+func (m *APIMetrics) GetItemWithContext(ctx context.Context, input *dynamodb.GetItemInput, opts ...request.Option) (*dynamodb.GetItemOutput, error) {
+	start := time.Now()
+	output, err := m.DynamoDBAPI.GetItemWithContext(ctx, input, opts...)
+
+	recordMetrics(m.tableType, "get_item", err, time.Since(start).Seconds())
+
+	return output, err
+}
+
+func (m *APIMetrics) PutItemWithContext(ctx context.Context, input *dynamodb.PutItemInput, opts ...request.Option) (*dynamodb.PutItemOutput, error) {
+	start := time.Now()
+	output, err := m.DynamoDBAPI.PutItemWithContext(ctx, input, opts...)
+
+	recordMetrics(m.tableType, "put_item", err, time.Since(start).Seconds())
+
+	return output, err
+}
+
+func (m *APIMetrics) UpdateItemWithContext(ctx context.Context, input *dynamodb.UpdateItemInput, opts ...request.Option) (*dynamodb.UpdateItemOutput, error) {
+	start := time.Now()
+	output, err := m.DynamoDBAPI.UpdateItemWithContext(ctx, input, opts...)
+
+	recordMetrics(m.tableType, "update_item", err, time.Since(start).Seconds())
+
+	return output, err
+}
+
+func (m *APIMetrics) DeleteTableWithContext(ctx context.Context, input *dynamodb.DeleteTableInput, opts ...request.Option) (*dynamodb.DeleteTableOutput, error) {
+	start := time.Now()
+	output, err := m.DynamoDBAPI.DeleteTableWithContext(ctx, input, opts...)
+
+	recordMetrics(m.tableType, "delete_table", err, time.Since(start).Seconds())
+
+	return output, err
+}
+
+func (m *APIMetrics) BatchWriteItemWithContext(ctx context.Context, input *dynamodb.BatchWriteItemInput, opts ...request.Option) (*dynamodb.BatchWriteItemOutput, error) {
+	start := time.Now()
+	output, err := m.DynamoDBAPI.BatchWriteItemWithContext(ctx, input, opts...)
+
+	recordMetrics(m.tableType, "batch_write_item", err, time.Since(start).Seconds())
+
+	return output, err
+}
+
+func (m *APIMetrics) ScanWithContext(ctx context.Context, input *dynamodb.ScanInput, opts ...request.Option) (*dynamodb.ScanOutput, error) {
+	start := time.Now()
+	output, err := m.DynamoDBAPI.ScanWithContext(ctx, input, opts...)
+
+	recordMetrics(m.tableType, "scan", err, time.Since(start).Seconds())
+
+	return output, err
+}
+
+func (m *APIMetrics) CreateTableWithContext(ctx context.Context, input *dynamodb.CreateTableInput, opts ...request.Option) (*dynamodb.CreateTableOutput, error) {
+	start := time.Now()
+	output, err := m.DynamoDBAPI.CreateTableWithContext(ctx, input, opts...)
+
+	recordMetrics(m.tableType, "create_table", err, time.Since(start).Seconds())
+
+	return output, err
+}
+
+func (m *APIMetrics) DescribeTableWithContext(ctx context.Context, input *dynamodb.DescribeTableInput, opts ...request.Option) (*dynamodb.DescribeTableOutput, error) {
+	start := time.Now()
+	output, err := m.DynamoDBAPI.DescribeTableWithContext(ctx, input, opts...)
+
+	recordMetrics(m.tableType, "describe_table", err, time.Since(start).Seconds())
+
+	return output, err
+}
+
+func (m *APIMetrics) QueryWithContext(ctx context.Context, input *dynamodb.QueryInput, opts ...request.Option) (*dynamodb.QueryOutput, error) {
+	start := time.Now()
+	output, err := m.DynamoDBAPI.QueryWithContext(ctx, input, opts...)
+
+	recordMetrics(m.tableType, "query", err, time.Since(start).Seconds())
+
+	return output, err
+}

lib/observability/metrics/dynamo/dynamo.go

@@ -0,0 +1,75 @@
+// Copyright 2022 Gravitational, Inc
+//
+// 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 dynamo
+
+import (
+	"github.com/prometheus/client_golang/prometheus"
+)
+
+var (
+	apiRequestsTotal = prometheus.NewCounterVec(
+		prometheus.CounterOpts{
+			Name: "dynamo_requests_total",
+			Help: "Total number of requests to the DynamoDB API",
+		},
+		[]string{"type", "operation"},
+	)
+	apiRequests = prometheus.NewCounterVec(
+		prometheus.CounterOpts{
+			Name: "dynamo_requests",
+			Help: "Number of failed requests to the DynamoDB API by result",
+		},
+		[]string{"type", "operation", "result"},
+	)
+	apiRequestLatencies = prometheus.NewHistogramVec(
+		prometheus.HistogramOpts{
+			Name: "dynamo_requests_seconds",
+			Help: "Request latency for the DynamoDB API",
+			// lowest bucket start of upper bound 0.001 sec (1 ms) with factor 2
+			// highest bucket start of 0.001 sec * 2^15 == 32.768 sec
+			Buckets: prometheus.ExponentialBuckets(0.001, 2, 16),
+		},
+		[]string{"type", "operation"},
+	)
+
+	dynamoCollectors = []prometheus.Collector{
+		apiRequests,
+		apiRequestsTotal,
+		apiRequestLatencies,
+	}
+)
+
+// TableType indicates which type of table metrics are being calculated for
+type TableType string
+
+const (
+	// Backend is a table used to store backend data.
+	Backend TableType = "backend"
+	// Events is a table used to store audit events.
+	Events TableType = "events"
+)
+
+// recordMetrics updates the set of dynamo api metrics
+func recordMetrics(tableType TableType, operation string, err error, latency float64) {
+	labels := []string{string(tableType), operation}
+	apiRequestsTotal.WithLabelValues(labels...).Inc()
+	apiRequestLatencies.WithLabelValues(labels...).Observe(latency)
+
+	result := "success"
+	if err != nil {
+		result = "error"
+	}
+	apiRequests.WithLabelValues(append(labels, result)...).Inc()
+}

lib/observability/metrics/dynamo/streams.go

@@ -0,0 +1,73 @@
+// Copyright 2022 Gravitational, Inc
+//
+// 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 dynamo
+
+import (
+	"context"
+	"time"
+
+	"github.com/aws/aws-sdk-go/aws/request"
+	"github.com/aws/aws-sdk-go/service/dynamodbstreams"
+	"github.com/aws/aws-sdk-go/service/dynamodbstreams/dynamodbstreamsiface"
+	"github.com/gravitational/trace"
+
+	"github.com/gravitational/teleport/lib/utils"
+)
+
+// StreamsMetricsAPI wraps a dynamodbstreamsiface.DynamoDBStreamsAPI implementation and
+// reports statistics about the dynamo api operations
+type StreamsMetricsAPI struct {
+	dynamodbstreamsiface.DynamoDBStreamsAPI
+	tableType TableType
+}
+
+// NewStreamsMetricsAPI returns a new StreamsMetricsAPI for the provided TableType
+func NewStreamsMetricsAPI(tableType TableType, api dynamodbstreamsiface.DynamoDBStreamsAPI) (*StreamsMetricsAPI, error) {
+	if err := utils.RegisterPrometheusCollectors(dynamoCollectors...); err != nil {
+		return nil, trace.Wrap(err)
+	}
+
+	return &StreamsMetricsAPI{
+		DynamoDBStreamsAPI: api,
+		tableType:          tableType,
+	}, nil
+}
+
+func (m *StreamsMetricsAPI) DescribeStreamWithContext(ctx context.Context, input *dynamodbstreams.DescribeStreamInput, opts ...request.Option) (*dynamodbstreams.DescribeStreamOutput, error) {
+	start := time.Now()
+	output, err := m.DynamoDBStreamsAPI.DescribeStreamWithContext(ctx, input, opts...)
+
+	recordMetrics(m.tableType, "describe_stream", err, time.Since(start).Seconds())
+
+	return output, err
+}
+
+func (m *StreamsMetricsAPI) GetShardIteratorWithContext(ctx context.Context, input *dynamodbstreams.GetShardIteratorInput, opts ...request.Option) (*dynamodbstreams.GetShardIteratorOutput, error) {
+	start := time.Now()
+	output, err := m.DynamoDBStreamsAPI.GetShardIteratorWithContext(ctx, input, opts...)
+
+	recordMetrics(m.tableType, "get_shard_iterator", err, time.Since(start).Seconds())
+
+	return output, err
+}
+
+func (m *StreamsMetricsAPI) GetRecordsWithContext(ctx context.Context, input *dynamodbstreams.GetRecordsInput, opts ...request.Option) (*dynamodbstreams.GetRecordsOutput, error) {
+	start := time.Now()
+	output, err := m.DynamoDBStreamsAPI.GetRecordsWithContext(ctx, input, opts...)
+
+	recordMetrics(m.tableType, "get_records", err, time.Since(start).Seconds())
+
+	return output, err
+}