diff --git a/docs/pages/setup/reference/metrics.mdx b/docs/pages/setup/reference/metrics.mdx index 199c3ba26a5..a9588bdf8c3 100644 --- a/docs/pages/setup/reference/metrics.mdx +++ b/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. | diff --git a/lib/backend/dynamo/configure.go b/lib/backend/dynamo/configure.go index 4fee0684510..d7695cdc584 100644 --- a/lib/backend/dynamo/configure.go +++ b/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{ diff --git a/lib/backend/dynamo/doc.go b/lib/backend/dynamo/doc.go index 0f04ea36acc..75cd6f91ddc 100644 --- a/lib/backend/dynamo/doc.go +++ b/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 diff --git a/lib/backend/dynamo/dynamo.go b/lib/backend/dynamo/dynamo.go deleted file mode 100644 index 2436ae31cfa..00000000000 --- a/lib/backend/dynamo/dynamo.go +++ /dev/null @@ -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 diff --git a/lib/backend/dynamo/dynamodbbk.go b/lib/backend/dynamo/dynamodbbk.go index 37dda887437..24183e627df 100644 --- a/lib/backend/dynamo/dynamodbbk.go +++ b/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) diff --git a/lib/events/dynamoevents/dynamoevents.go b/lib/events/dynamoevents/dynamoevents.go index 38e9d30fa1d..642d8c4b60c 100644 --- a/lib/events/dynamoevents/dynamoevents.go +++ b/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) } diff --git a/lib/observability/metrics/dynamo/api.go b/lib/observability/metrics/dynamo/api.go new file mode 100644 index 00000000000..86ebfa70bf7 --- /dev/null +++ b/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 +} diff --git a/lib/observability/metrics/dynamo/dynamo.go b/lib/observability/metrics/dynamo/dynamo.go new file mode 100644 index 00000000000..fcfc5c118ed --- /dev/null +++ b/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() +} diff --git a/lib/observability/metrics/dynamo/streams.go b/lib/observability/metrics/dynamo/streams.go new file mode 100644 index 00000000000..364f8dbca2c --- /dev/null +++ b/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 +}