From 0c48cefef211ff5c6841f025cebfb8eb1407e04e Mon Sep 17 00:00:00 2001 From: rosstimothy <39066650+rosstimothy@users.noreply.github.com> Date: Thu, 21 Jul 2022 13:31:50 -0400 Subject: [PATCH] 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 | 3 + lib/backend/dynamo/configure.go | 3 +- lib/backend/dynamo/doc.go | 2 +- lib/backend/dynamo/dynamo.go | 17 --- lib/backend/dynamo/dynamodbbk.go | 19 ++- lib/events/dynamoevents/dynamoevents.go | 36 +++-- lib/observability/metrics/dynamo/api.go | 154 ++++++++++++++++++++ lib/observability/metrics/dynamo/dynamo.go | 75 ++++++++++ lib/observability/metrics/dynamo/streams.go | 73 ++++++++++ 9 files changed, 345 insertions(+), 37 deletions(-) delete mode 100644 lib/backend/dynamo/dynamo.go create mode 100644 lib/observability/metrics/dynamo/api.go create mode 100644 lib/observability/metrics/dynamo/dynamo.go create mode 100644 lib/observability/metrics/dynamo/streams.go 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 +}