Skip to content

Commit

Permalink
add timeout and context to queue metrics
Browse files Browse the repository at this point in the history
  • Loading branch information
abelanger5 committed Sep 27, 2024
1 parent 5f5e1e8 commit 5e7e127
Show file tree
Hide file tree
Showing 3 changed files with 11 additions and 7 deletions.
2 changes: 1 addition & 1 deletion api/v1/server/handlers/tenants/get_queue_metrics.go
Original file line number Diff line number Diff line change
Expand Up @@ -38,7 +38,7 @@ func (t *TenantService) TenantGetQueueMetrics(ctx echo.Context, request gen.Tena
opts.WorkflowIds = *request.Params.Workflows
}

metrics, err := t.config.APIRepository.Tenant().GetQueueMetrics(tenant.ID, &opts)
metrics, err := t.config.APIRepository.Tenant().GetQueueMetrics(ctx.Request().Context(), tenant.ID, &opts)

if err != nil {
return nil, err
Expand Down
14 changes: 9 additions & 5 deletions pkg/repository/prisma/tenant.go
Original file line number Diff line number Diff line change
Expand Up @@ -197,7 +197,7 @@ func (r *tenantAPIRepository) DeleteTenantMember(memberId string) (*db.TenantMem
).Delete().Exec(context.Background())
}

func (r *tenantAPIRepository) GetQueueMetrics(tenantId string, opts *repository.GetQueueMetricsOpts) (*repository.GetQueueMetricsResponse, error) {
func (r *tenantAPIRepository) GetQueueMetrics(ctx context.Context, tenantId string, opts *repository.GetQueueMetricsOpts) (*repository.GetQueueMetricsResponse, error) {
if err := r.v.Validate(opts); err != nil {
return nil, err
}
Expand Down Expand Up @@ -231,28 +231,32 @@ func (r *tenantAPIRepository) GetQueueMetrics(tenantId string, opts *repository.
totalParams.WorkflowIds = uuids
}

tx, err := r.pool.Begin(context.Background())
tx, commit, rollback, err := prepareTx(ctx, r.pool, r.l, 60*1000)

if err != nil {
return nil, err
}

defer deferRollback(context.Background(), r.l, tx.Rollback)
defer rollback()

// get the totals
total, err := r.queries.GetTenantTotalQueueMetrics(context.Background(), tx, totalParams)
total, err := r.queries.GetTenantTotalQueueMetrics(ctx, tx, totalParams)

if err != nil {
return nil, err
}

// get the workflow metrics
workflowMetrics, err := r.queries.GetTenantWorkflowQueueMetrics(context.Background(), tx, workflowParams)
workflowMetrics, err := r.queries.GetTenantWorkflowQueueMetrics(ctx, tx, workflowParams)

if err != nil {
return nil, err
}

if err := commit(ctx); err != nil {
return nil, err
}

workflowMetricsMap := make(map[string]repository.QueueMetric)

for _, metric := range workflowMetrics {
Expand Down
2 changes: 1 addition & 1 deletion pkg/repository/tenant.go
Original file line number Diff line number Diff line change
Expand Up @@ -98,7 +98,7 @@ type TenantAPIRepository interface {
DeleteTenantMember(memberId string) (*db.TenantMemberModel, error)

// GetQueueMetrics returns the queue metrics for the given tenant
GetQueueMetrics(tenantId string, opts *GetQueueMetricsOpts) (*GetQueueMetricsResponse, error)
GetQueueMetrics(ctx context.Context, tenantId string, opts *GetQueueMetricsOpts) (*GetQueueMetricsResponse, error)
}

type TenantEngineRepository interface {
Expand Down

0 comments on commit 5e7e127

Please sign in to comment.