Skip to content

Commit 15e1eeb

Browse files
serprexserprex
authored
Merge branch 'main' into lees-slot-polling
2 parents 537d570 + 49f9ec5 commit 15e1eeb

26 files changed

Lines changed: 520 additions & 511 deletions

‎flow/.golangci.yml‎

Lines changed: 10 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -1,12 +1,19 @@
11
version: "2"
22
linters:
33
enable:
4+
- asasalint
5+
- asciicheck
6+
- bidichk
7+
- canonicalheader
48
- containedctx
59
- copyloopvar
610
- dogsled
711
- durationcheck
12+
- errchkjson
13+
- errname
814
- exptostd
915
- forbidigo
16+
- funcorder
1017
- gocritic
1118
- gosec
1219
- iface
@@ -28,10 +35,13 @@ linters:
2835
- thelper
2936
- unconvert
3037
- unparam
38+
- usestdlibvars
3139
- usetesting
3240
- wastedassign
3341
- whitespace
3442
settings:
43+
funcorder:
44+
struct-method: false
3545
gocritic:
3646
disabled-checks:
3747
- ifElseChain

‎flow/alerting/alerting.go‎

Lines changed: 41 additions & 41 deletions
Original file line numberDiff line numberDiff line change
@@ -48,6 +48,47 @@ type AlertKeys struct {
4848
SlotName string
4949
}
5050

51+
// doesn't take care of closing pool, needs to be done externally.
52+
func NewAlerter(ctx context.Context, catalogPool shared.CatalogPool, otelManager *otel_metrics.OtelManager) *Alerter {
53+
if catalogPool.Pool == nil {
54+
panic("catalog pool is nil for Alerter")
55+
}
56+
snsTopic := internal.PeerDBTelemetryAWSSNSTopicArn()
57+
var snsMessageSender telemetry.Sender
58+
if snsTopic != "" {
59+
var err error
60+
snsMessageSender, err = telemetry.NewSNSMessageSenderWithNewClient(ctx, &telemetry.SNSMessageSenderConfig{
61+
Topic: snsTopic,
62+
})
63+
internal.LoggerFromCtx(ctx).Info("Successfully registered sns telemetry sender")
64+
if err != nil {
65+
panic(fmt.Sprintf("unable to setup telemetry is nil for Alerter %+v", err))
66+
}
67+
}
68+
69+
incidentIoURL := internal.PeerDBGetIncidentIoUrl()
70+
incidentIoAuth := internal.PeerDBGetIncidentIoToken()
71+
var incidentIoTelemetrySender telemetry.Sender
72+
if incidentIoURL != "" && incidentIoAuth != "" {
73+
var err error
74+
incidentIoTelemetrySender, err = telemetry.NewIncidentIoMessageSender(ctx, telemetry.IncidentIoMessageSenderConfig{
75+
URL: incidentIoURL,
76+
Token: incidentIoAuth,
77+
})
78+
internal.LoggerFromCtx(ctx).Info("Successfully registered incident.io telemetry sender")
79+
if err != nil {
80+
panic(fmt.Sprintf("unable to setup incident.io telemetry is nil for Alerter %+v", err))
81+
}
82+
}
83+
84+
return &Alerter{
85+
CatalogPool: catalogPool,
86+
snsTelemetrySender: snsMessageSender,
87+
incidentIoTelemetrySender: incidentIoTelemetrySender,
88+
otelManager: otelManager,
89+
}
90+
}
91+
5192
func (a *Alerter) registerSendersFromPool(ctx context.Context) ([]AlertSenderConfig, error) {
5293
rows, err := a.CatalogPool.Query(ctx,
5394
`SELECT id, service_type, service_config, enc_key_id, alert_for_mirrors
@@ -120,47 +161,6 @@ func (a *Alerter) registerSendersFromPool(ctx context.Context) ([]AlertSenderCon
120161
})
121162
}
122163

123-
// doesn't take care of closing pool, needs to be done externally.
124-
func NewAlerter(ctx context.Context, catalogPool shared.CatalogPool, otelManager *otel_metrics.OtelManager) *Alerter {
125-
if catalogPool.Pool == nil {
126-
panic("catalog pool is nil for Alerter")
127-
}
128-
snsTopic := internal.PeerDBTelemetryAWSSNSTopicArn()
129-
var snsMessageSender telemetry.Sender
130-
if snsTopic != "" {
131-
var err error
132-
snsMessageSender, err = telemetry.NewSNSMessageSenderWithNewClient(ctx, &telemetry.SNSMessageSenderConfig{
133-
Topic: snsTopic,
134-
})
135-
internal.LoggerFromCtx(ctx).Info("Successfully registered sns telemetry sender")
136-
if err != nil {
137-
panic(fmt.Sprintf("unable to setup telemetry is nil for Alerter %+v", err))
138-
}
139-
}
140-
141-
incidentIoURL := internal.PeerDBGetIncidentIoUrl()
142-
incidentIoAuth := internal.PeerDBGetIncidentIoToken()
143-
var incidentIoTelemetrySender telemetry.Sender
144-
if incidentIoURL != "" && incidentIoAuth != "" {
145-
var err error
146-
incidentIoTelemetrySender, err = telemetry.NewIncidentIoMessageSender(ctx, telemetry.IncidentIoMessageSenderConfig{
147-
URL: incidentIoURL,
148-
Token: incidentIoAuth,
149-
})
150-
internal.LoggerFromCtx(ctx).Info("Successfully registered incident.io telemetry sender")
151-
if err != nil {
152-
panic(fmt.Sprintf("unable to setup incident.io telemetry is nil for Alerter %+v", err))
153-
}
154-
}
155-
156-
return &Alerter{
157-
CatalogPool: catalogPool,
158-
snsTelemetrySender: snsMessageSender,
159-
incidentIoTelemetrySender: incidentIoTelemetrySender,
160-
otelManager: otelManager,
161-
}
162-
}
163-
164164
func (a *Alerter) AlertIfSlotLag(ctx context.Context, alertKeys *AlertKeys, slotInfo *protos.SlotInfo) {
165165
alertSenderConfigs, err := a.registerSendersFromPool(ctx)
166166
if err != nil {

‎flow/alerting/email_alert_sender.go‎

Lines changed: 29 additions & 29 deletions
Original file line numberDiff line numberDiff line change
@@ -12,6 +12,15 @@ import (
1212
"github.com/PeerDB-io/peerdb/flow/shared/aws_common"
1313
)
1414

15+
type EmailAlertSenderConfig struct {
16+
sourceEmail string
17+
configurationSetName string
18+
replyToAddresses []string
19+
EmailAddresses []string `json:"email_addresses"`
20+
SlotLagMBAlertThreshold uint32 `json:"slot_lag_mb_alert_threshold"`
21+
OpenConnectionsAlertThreshold uint32 `json:"open_connections_alert_threshold"`
22+
}
23+
1524
type EmailAlertSender struct {
1625
AlertSender
1726
client *ses.Client
@@ -23,6 +32,26 @@ type EmailAlertSender struct {
2332
openConnectionsAlertThreshold uint32
2433
}
2534

35+
func NewEmailAlertSender(client *ses.Client, config *EmailAlertSenderConfig) *EmailAlertSender {
36+
return &EmailAlertSender{
37+
client: client,
38+
sourceEmail: config.sourceEmail,
39+
configurationSetName: config.configurationSetName,
40+
replyToAddresses: config.replyToAddresses,
41+
emailAddresses: config.EmailAddresses,
42+
slotLagMBAlertThreshold: config.SlotLagMBAlertThreshold,
43+
openConnectionsAlertThreshold: config.OpenConnectionsAlertThreshold,
44+
}
45+
}
46+
47+
func NewEmailAlertSenderWithNewClient(ctx context.Context, region *string, config *EmailAlertSenderConfig) (*EmailAlertSender, error) {
48+
client, err := newSesClient(ctx, region)
49+
if err != nil {
50+
return nil, err
51+
}
52+
return NewEmailAlertSender(client, config), nil
53+
}
54+
2655
func (e *EmailAlertSender) getSlotLagMBAlertThreshold() uint32 {
2756
return e.slotLagMBAlertThreshold
2857
}
@@ -31,15 +60,6 @@ func (e *EmailAlertSender) getOpenConnectionsAlertThreshold() uint32 {
3160
return e.openConnectionsAlertThreshold
3261
}
3362

34-
type EmailAlertSenderConfig struct {
35-
sourceEmail string
36-
configurationSetName string
37-
replyToAddresses []string
38-
EmailAddresses []string `json:"email_addresses"`
39-
SlotLagMBAlertThreshold uint32 `json:"slot_lag_mb_alert_threshold"`
40-
OpenConnectionsAlertThreshold uint32 `json:"open_connections_alert_threshold"`
41-
}
42-
4363
func (e *EmailAlertSender) sendAlert(ctx context.Context, alertTitle string, alertMessage string) error {
4464
_, err := e.client.SendEmail(ctx, &ses.SendEmailInput{
4565
Destination: &types.Destination{
@@ -73,26 +93,6 @@ func (e *EmailAlertSender) sendAlert(ctx context.Context, alertTitle string, ale
7393
return nil
7494
}
7595

76-
func NewEmailAlertSenderWithNewClient(ctx context.Context, region *string, config *EmailAlertSenderConfig) (*EmailAlertSender, error) {
77-
client, err := newSesClient(ctx, region)
78-
if err != nil {
79-
return nil, err
80-
}
81-
return NewEmailAlertSender(client, config), nil
82-
}
83-
84-
func NewEmailAlertSender(client *ses.Client, config *EmailAlertSenderConfig) *EmailAlertSender {
85-
return &EmailAlertSender{
86-
client: client,
87-
sourceEmail: config.sourceEmail,
88-
configurationSetName: config.configurationSetName,
89-
replyToAddresses: config.replyToAddresses,
90-
emailAddresses: config.EmailAddresses,
91-
slotLagMBAlertThreshold: config.SlotLagMBAlertThreshold,
92-
openConnectionsAlertThreshold: config.OpenConnectionsAlertThreshold,
93-
}
94-
}
95-
9696
func newSesClient(ctx context.Context, region *string) (*ses.Client, error) {
9797
sdkConfig, err := aws_common.LoadSdkConfig(ctx, region)
9898
if err != nil {

‎flow/cmd/handler.go‎

Lines changed: 2 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -329,7 +329,7 @@ func (h *FlowRequestHandler) FlowStateChange(
329329
return nil, fmt.Errorf("unable to load dynamic config: %w", err)
330330
} else if underMaintenance {
331331
slog.Warn("Flow state change request denied due to maintenance", logs)
332-
return nil, exceptions.UnderMaintenanceError
332+
return nil, exceptions.ErrUnderMaintenance
333333
}
334334

335335
workflowID, err := h.getWorkflowID(ctx, req.FlowJobName)
@@ -517,7 +517,7 @@ func (h *FlowRequestHandler) resyncMirror(
517517
if underMaintenance, err := internal.PeerDBMaintenanceModeEnabled(ctx, nil); err != nil {
518518
return fmt.Errorf("unable to get maintenance mode status: %w", err)
519519
} else if underMaintenance {
520-
return exceptions.UnderMaintenanceError
520+
return exceptions.ErrUnderMaintenance
521521
}
522522

523523
isCDC, err := h.isCDCFlow(ctx, flowName)

‎flow/cmd/logged_interceptor.go‎

Lines changed: 6 additions & 6 deletions
Original file line numberDiff line numberDiff line change
@@ -55,6 +55,12 @@ type LoggedWorkerInterceptor struct {
5555
interceptor.WorkerInterceptorBase
5656
}
5757

58+
func NewLoggedWorkerInterceptor() *LoggedWorkerInterceptor {
59+
return &LoggedWorkerInterceptor{
60+
WorkerInterceptorBase: interceptor.WorkerInterceptorBase{},
61+
}
62+
}
63+
5864
func (c LoggedWorkerInterceptor) InterceptActivity(
5965
ctx context.Context,
6066
next interceptor.ActivityInboundInterceptor,
@@ -71,9 +77,3 @@ func (c LoggedWorkerInterceptor) InterceptWorkflow(
7177
// Workflow intercepting ends here
7278
return intercepted
7379
}
74-
75-
func NewLoggedWorkerInterceptor() *LoggedWorkerInterceptor {
76-
return &LoggedWorkerInterceptor{
77-
WorkerInterceptorBase: interceptor.WorkerInterceptorBase{},
78-
}
79-
}

‎flow/cmd/validate_mirror.go‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -32,7 +32,7 @@ func (h *FlowRequestHandler) ValidateCDCMirror(
3232

3333
if underMaintenance {
3434
slog.Warn("Validate request denied due to maintenance", "flowName", req.ConnectionConfigs.FlowJobName)
35-
return nil, exceptions.UnderMaintenanceError
35+
return nil, exceptions.ErrUnderMaintenance
3636
}
3737

3838
if !req.ConnectionConfigs.Resync {

‎flow/connectors/bigquery/bigquery.go‎

Lines changed: 54 additions & 54 deletions
Original file line numberDiff line numberDiff line change
@@ -27,17 +27,6 @@ const (
2727
SyncRecordsBatchSize = 1024
2828
)
2929

30-
type BigQueryConnector struct {
31-
*metadataStore.PostgresMetadata
32-
logger log.Logger
33-
bqConfig *protos.BigqueryConfig
34-
client *bigquery.Client
35-
storageClient *storage.Client
36-
catalogPool shared.CatalogPool
37-
datasetID string
38-
projectID string
39-
}
40-
4130
func NewBigQueryServiceAccount(bqConfig *protos.BigqueryConfig) (*utils.GcpServiceAccount, error) {
4231
var serviceAccount utils.GcpServiceAccount
4332
serviceAccount.Type = bqConfig.AuthType
@@ -58,49 +47,15 @@ func NewBigQueryServiceAccount(bqConfig *protos.BigqueryConfig) (*utils.GcpServi
5847
return &serviceAccount, nil
5948
}
6049

61-
// ValidateCheck:
62-
// 1. Creates a table
63-
// 2. Inserts one row into the table
64-
// 3. Deletes the table
65-
func (c *BigQueryConnector) ValidateCheck(ctx context.Context) error {
66-
dummyTable := "peerdb_validate_dummy_" + shared.RandomString(4)
67-
68-
newTable := c.client.DatasetInProject(c.projectID, c.datasetID).Table(dummyTable)
69-
70-
createErr := newTable.Create(ctx, &bigquery.TableMetadata{
71-
Schema: []*bigquery.FieldSchema{
72-
{
73-
Name: "dummy",
74-
Type: bigquery.BooleanFieldType,
75-
Repeated: false,
76-
},
77-
},
78-
})
79-
if createErr != nil {
80-
return fmt.Errorf("unable to validate table creation within dataset: %w. "+
81-
"Please check if bigquery.tables.create permission has been granted", createErr)
82-
}
83-
84-
var errs []error
85-
insertQuery := c.client.Query(fmt.Sprintf("INSERT INTO %s VALUES(true)", dummyTable))
86-
insertQuery.DefaultDatasetID = c.datasetID
87-
insertQuery.DefaultProjectID = c.projectID
88-
_, insertErr := insertQuery.Run(ctx)
89-
if insertErr != nil {
90-
errs = append(errs, fmt.Errorf("unable to validate insertion into table: %w. ", insertErr))
91-
}
92-
93-
// Drop the table
94-
deleteErr := newTable.Delete(ctx)
95-
if deleteErr != nil {
96-
errs = append(errs, fmt.Errorf("unable to delete table :%w. ", deleteErr))
97-
}
98-
99-
if len(errs) > 0 {
100-
return errors.Join(errs...)
101-
}
102-
103-
return nil
50+
type BigQueryConnector struct {
51+
*metadataStore.PostgresMetadata
52+
logger log.Logger
53+
bqConfig *protos.BigqueryConfig
54+
client *bigquery.Client
55+
storageClient *storage.Client
56+
catalogPool shared.CatalogPool
57+
datasetID string
58+
projectID string
10459
}
10560

10661
func NewBigQueryConnector(ctx context.Context, config *protos.BigqueryConfig) (*BigQueryConnector, error) {
@@ -156,6 +111,51 @@ func NewBigQueryConnector(ctx context.Context, config *protos.BigqueryConfig) (*
156111
}, nil
157112
}
158113

114+
// ValidateCheck:
115+
// 1. Creates a table
116+
// 2. Inserts one row into the table
117+
// 3. Deletes the table
118+
func (c *BigQueryConnector) ValidateCheck(ctx context.Context) error {
119+
dummyTable := "peerdb_validate_dummy_" + shared.RandomString(4)
120+
121+
newTable := c.client.DatasetInProject(c.projectID, c.datasetID).Table(dummyTable)
122+
123+
createErr := newTable.Create(ctx, &bigquery.TableMetadata{
124+
Schema: []*bigquery.FieldSchema{
125+
{
126+
Name: "dummy",
127+
Type: bigquery.BooleanFieldType,
128+
Repeated: false,
129+
},
130+
},
131+
})
132+
if createErr != nil {
133+
return fmt.Errorf("unable to validate table creation within dataset: %w. "+
134+
"Please check if bigquery.tables.create permission has been granted", createErr)
135+
}
136+
137+
var errs []error
138+
insertQuery := c.client.Query(fmt.Sprintf("INSERT INTO %s VALUES(true)", dummyTable))
139+
insertQuery.DefaultDatasetID = c.datasetID
140+
insertQuery.DefaultProjectID = c.projectID
141+
_, insertErr := insertQuery.Run(ctx)
142+
if insertErr != nil {
143+
errs = append(errs, fmt.Errorf("unable to validate insertion into table: %w. ", insertErr))
144+
}
145+
146+
// Drop the table
147+
deleteErr := newTable.Delete(ctx)
148+
if deleteErr != nil {
149+
errs = append(errs, fmt.Errorf("unable to delete table :%w. ", deleteErr))
150+
}
151+
152+
if len(errs) > 0 {
153+
return errors.Join(errs...)
154+
}
155+
156+
return nil
157+
}
158+
159159
// Close closes the BigQuery driver.
160160
func (c *BigQueryConnector) Close() error {
161161
if c != nil {

0 commit comments

Comments
 (0)