-
Notifications
You must be signed in to change notification settings - Fork 8
/
db.go
160 lines (134 loc) · 4.21 KB
/
db.go
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
/**
* Copyright 2020 Comcast Cable Communications Management, LLC
*
* 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 dynamodb
import (
"errors"
"net/http"
"time"
"github.com/aws/aws-sdk-go/aws"
"github.com/aws/aws-sdk-go/aws/awserr"
"github.com/aws/aws-sdk-go/aws/credentials"
"github.com/go-playground/validator/v10"
"github.com/xmidt-org/argus/model"
"github.com/xmidt-org/argus/store"
"github.com/xmidt-org/argus/store/db/metric"
"github.com/xmidt-org/httpaux/erraux"
)
// DynamoDB is the path to the configuration structure
// needed to connect to a dynamo DB instance.
const DynamoDB = "dynamo"
const (
defaultTable = "gifnoc"
defaultMaxRetries = 3
)
var validate *validator.Validate
var errHTTPBadRequest = &erraux.Error{
Err: errors.New("bad request to dynamodb"),
Code: http.StatusBadRequest,
}
func init() {
validate = validator.New()
}
// Config contains all fields needed to establish a connection
// with a dynamoDB instance.
type Config struct {
// Table is the name of the target DB table.
// (Optional) Defaults to 'gifnoc'
Table string
// Endpoint is the HTTP(S) URL to the DB.
// (Optional) Defaults to endpoint generated by the aws sdk.
Endpoint string
// Region is the AWS region of the running DB.
Region string `validate:"required"`
// MaxRetries is the number of times DB operations will be retried on error.
// (Optional) Defaults to 3.
MaxRetries int
// GetAllLimit is the maximum number of items to get at a time.
// (Optional) defaults to no limit
GetAllLimit int
// AccessKey is the AWS AccessKey credential.
AccessKey string `validate:"required"`
// SecretKey is the AWS SecretKey credential.
SecretKey string `validate:"required"`
// DisableDualStack indicates whether the connection to the DB should be
// dual stack (IPv4 and IPv6).
// (Optional) Defaults to False.
DisableDualStack bool
}
// dao adapts the underlying dynamodb data service to match
// the store.DAO (currently named store.S but we should rename it) interface.
type dao struct {
s service
}
func NewDynamoDB(config Config, measures metric.Measures) (store.S, error) {
if config.Table == "" {
config.Table = defaultTable
}
if config.MaxRetries == 0 {
config.MaxRetries = defaultMaxRetries
}
err := validate.Struct(config)
if err != nil {
return nil, err
}
awsConfig := *aws.NewConfig().
WithEndpoint(config.Endpoint).
WithUseDualStack(!config.DisableDualStack).
WithMaxRetries(config.MaxRetries).
WithCredentialsChainVerboseErrors(true).
WithRegion(config.Region).
WithCredentials(credentials.NewStaticCredentialsFromCreds(credentials.Value{
AccessKeyID: config.AccessKey,
SecretAccessKey: config.SecretKey,
}))
svc, err := newService(awsConfig, "", config.Table, int64(config.GetAllLimit), &measures)
if err != nil {
return nil, err
}
svc = newInstrumentingService(&dynamoMeasuresUpdater{measures: &measures}, svc, time.Now)
return &dao{
s: svc,
}, nil
}
func (d dao) Push(key model.Key, item store.OwnableItem) error {
_, err := d.s.Push(key, item)
return sanitizeError(err)
}
func (d dao) Get(key model.Key) (store.OwnableItem, error) {
item, _, err := d.s.Get(key)
return item, sanitizeError(err)
}
func (d *dao) Delete(key model.Key) (store.OwnableItem, error) {
item, _, err := d.s.Delete(key)
return item, sanitizeError(err)
}
func (d *dao) GetAll(bucket string) (map[string]store.OwnableItem, error) {
items, _, err := d.s.GetAll(bucket)
return items, sanitizeError(err)
}
func sanitizeError(err error) error {
if err == nil {
return nil
}
var awsErr awserr.Error
if errors.As(err, &awsErr) {
if awsErr.Code() == "ValidationException" {
return store.SanitizedError{Err: err, ErrHTTP: errHTTPBadRequest}
}
}
return store.SanitizeError(err)
}