mirror of
https://github.com/zitadel/zitadel.git
synced 2025-08-12 07:47:32 +00:00
perf: project quotas and usages (#6441)
* project quota added
* project quota removed
* add periods table
* make log record generic
* accumulate usage
* query usage
* count action run seconds
* fix filter in ReportQuotaUsage
* fix existing tests
* fix logstore tests
* fix typo
* fix: add quota unit tests command side
* fix: add quota unit tests command side
* fix: add quota unit tests command side
* move notifications into debouncer and improve limit querying
* cleanup
* comment
* fix: add quota unit tests command side
* fix remaining quota usage query
* implement InmemLogStorage
* cleanup and linting
* improve test
* fix: add quota unit tests command side
* fix: add quota unit tests command side
* fix: add quota unit tests command side
* fix: add quota unit tests command side
* action notifications and fixes for notifications query
* revert console prefix
* fix: add quota unit tests command side
* fix: add quota integration tests
* improve accountable requests
* improve accountable requests
* fix: add quota integration tests
* fix: add quota integration tests
* fix: add quota integration tests
* comment
* remove ability to store logs in db and other changes requested from review
* changes requested from review
* changes requested from review
* Update internal/api/http/middleware/access_interceptor.go
Co-authored-by: Silvan <silvan.reusser@gmail.com>
* tests: fix quotas integration tests
* improve incrementUsageStatement
* linting
* fix: delete e2e tests as intergation tests cover functionality
* Update internal/api/http/middleware/access_interceptor.go
Co-authored-by: Silvan <silvan.reusser@gmail.com>
* backup
* fix conflict
* create rc
* create prerelease
* remove issue release labeling
* fix tracing
---------
Co-authored-by: Livio Spring <livio.a@gmail.com>
Co-authored-by: Stefan Benz <stefan@caos.ch>
Co-authored-by: adlerhurst <silvan.reusser@gmail.com>
(cherry picked from commit 1a49b7d298)
This commit is contained in:
Elio Bischof
committed by
Livio Spring
Livio Spring
parent
b688d6f842
commit
5823fdbef9
@@ -3,142 +3,84 @@ package execution
|
||||
import (
|
||||
"context"
|
||||
"database/sql"
|
||||
"fmt"
|
||||
"errors"
|
||||
"math"
|
||||
"time"
|
||||
|
||||
"github.com/Masterminds/squirrel"
|
||||
"github.com/zitadel/logging"
|
||||
|
||||
"github.com/zitadel/zitadel/internal/api/call"
|
||||
"github.com/zitadel/zitadel/internal/api/authz"
|
||||
"github.com/zitadel/zitadel/internal/command"
|
||||
"github.com/zitadel/zitadel/internal/database"
|
||||
caos_errors "github.com/zitadel/zitadel/internal/errors"
|
||||
"github.com/zitadel/zitadel/internal/logstore"
|
||||
"github.com/zitadel/zitadel/internal/logstore/record"
|
||||
"github.com/zitadel/zitadel/internal/query"
|
||||
"github.com/zitadel/zitadel/internal/query/projection"
|
||||
"github.com/zitadel/zitadel/internal/repository/quota"
|
||||
)
|
||||
|
||||
const (
|
||||
executionLogsTable = "logstore.execution"
|
||||
executionTimestampCol = "log_date"
|
||||
executionTookCol = "took"
|
||||
executionMessageCol = "message"
|
||||
executionLogLevelCol = "loglevel"
|
||||
executionInstanceIdCol = "instance_id"
|
||||
executionActionIdCol = "action_id"
|
||||
executionMetadataCol = "metadata"
|
||||
)
|
||||
|
||||
var _ logstore.UsageQuerier = (*databaseLogStorage)(nil)
|
||||
var _ logstore.LogCleanupper = (*databaseLogStorage)(nil)
|
||||
var _ logstore.UsageStorer[*record.ExecutionLog] = (*databaseLogStorage)(nil)
|
||||
|
||||
type databaseLogStorage struct {
|
||||
dbClient *database.DB
|
||||
commands *command.Commands
|
||||
queries *query.Queries
|
||||
}
|
||||
|
||||
func NewDatabaseLogStorage(dbClient *database.DB) *databaseLogStorage {
|
||||
return &databaseLogStorage{dbClient: dbClient}
|
||||
func NewDatabaseLogStorage(dbClient *database.DB, commands *command.Commands, queries *query.Queries) *databaseLogStorage {
|
||||
return &databaseLogStorage{dbClient: dbClient, commands: commands, queries: queries}
|
||||
}
|
||||
|
||||
func (l *databaseLogStorage) QuotaUnit() quota.Unit {
|
||||
return quota.ActionsAllRunsSeconds
|
||||
}
|
||||
|
||||
func (l *databaseLogStorage) Emit(ctx context.Context, bulk []logstore.LogRecord) error {
|
||||
func (l *databaseLogStorage) Emit(ctx context.Context, bulk []*record.ExecutionLog) error {
|
||||
if len(bulk) == 0 {
|
||||
return nil
|
||||
}
|
||||
builder := squirrel.Insert(executionLogsTable).
|
||||
Columns(
|
||||
executionTimestampCol,
|
||||
executionTookCol,
|
||||
executionMessageCol,
|
||||
executionLogLevelCol,
|
||||
executionInstanceIdCol,
|
||||
executionActionIdCol,
|
||||
executionMetadataCol,
|
||||
).
|
||||
PlaceholderFormat(squirrel.Dollar)
|
||||
return l.incrementUsage(ctx, bulk)
|
||||
}
|
||||
|
||||
for idx := range bulk {
|
||||
item := bulk[idx].(*Record)
|
||||
|
||||
var took interface{}
|
||||
if item.Took > 0 {
|
||||
took = item.Took
|
||||
func (l *databaseLogStorage) incrementUsage(ctx context.Context, bulk []*record.ExecutionLog) (err error) {
|
||||
byInstance := make(map[string][]*record.ExecutionLog)
|
||||
for _, r := range bulk {
|
||||
if r.InstanceID != "" {
|
||||
byInstance[r.InstanceID] = append(byInstance[r.InstanceID], r)
|
||||
}
|
||||
}
|
||||
for instanceID, instanceBulk := range byInstance {
|
||||
q, getQuotaErr := l.queries.GetQuota(ctx, instanceID, quota.ActionsAllRunsSeconds)
|
||||
if errors.Is(getQuotaErr, sql.ErrNoRows) {
|
||||
continue
|
||||
}
|
||||
err = errors.Join(err, getQuotaErr)
|
||||
if getQuotaErr != nil {
|
||||
continue
|
||||
}
|
||||
sum, incrementErr := l.incrementUsageFromExecutionLogs(ctx, instanceID, q.CurrentPeriodStart, instanceBulk)
|
||||
err = errors.Join(err, incrementErr)
|
||||
if incrementErr != nil {
|
||||
continue
|
||||
}
|
||||
|
||||
builder = builder.Values(
|
||||
item.LogDate,
|
||||
took,
|
||||
item.Message,
|
||||
item.LogLevel,
|
||||
item.InstanceID,
|
||||
item.ActionID,
|
||||
item.Metadata,
|
||||
)
|
||||
notifications, getNotificationErr := l.queries.GetDueQuotaNotifications(ctx, instanceID, quota.ActionsAllRunsSeconds, q, q.CurrentPeriodStart, sum)
|
||||
err = errors.Join(err, getNotificationErr)
|
||||
if getNotificationErr != nil || len(notifications) == 0 {
|
||||
continue
|
||||
}
|
||||
ctx = authz.WithInstanceID(ctx, instanceID)
|
||||
reportErr := l.commands.ReportQuotaUsage(ctx, notifications)
|
||||
err = errors.Join(err, reportErr)
|
||||
if reportErr != nil {
|
||||
continue
|
||||
}
|
||||
}
|
||||
|
||||
stmt, args, err := builder.ToSql()
|
||||
if err != nil {
|
||||
return caos_errors.ThrowInternal(err, "EXEC-KOS7I", "Errors.Internal")
|
||||
}
|
||||
|
||||
result, err := l.dbClient.ExecContext(ctx, stmt, args...)
|
||||
if err != nil {
|
||||
return caos_errors.ThrowInternal(err, "EXEC-0j6i5", "Errors.Access.StorageFailed")
|
||||
}
|
||||
|
||||
rows, err := result.RowsAffected()
|
||||
if err != nil {
|
||||
return caos_errors.ThrowInternal(err, "EXEC-MGchJ", "Errors.Internal")
|
||||
}
|
||||
|
||||
logging.WithFields("rows", rows).Debug("successfully stored execution logs")
|
||||
return nil
|
||||
}
|
||||
|
||||
func (l *databaseLogStorage) QueryUsage(ctx context.Context, instanceId string, start time.Time) (uint64, error) {
|
||||
stmt, args, err := squirrel.Select(
|
||||
fmt.Sprintf("COALESCE(SUM(%s)::INT,0)", executionTookCol),
|
||||
).
|
||||
From(executionLogsTable + l.dbClient.Timetravel(call.Took(ctx))).
|
||||
Where(squirrel.And{
|
||||
squirrel.Eq{executionInstanceIdCol: instanceId},
|
||||
squirrel.GtOrEq{executionTimestampCol: start},
|
||||
squirrel.NotEq{executionTookCol: nil},
|
||||
}).
|
||||
PlaceholderFormat(squirrel.Dollar).
|
||||
ToSql()
|
||||
|
||||
if err != nil {
|
||||
return 0, caos_errors.ThrowInternal(err, "EXEC-DXtzg", "Errors.Internal")
|
||||
}
|
||||
|
||||
var durationSeconds uint64
|
||||
err = l.dbClient.
|
||||
QueryRowContext(ctx,
|
||||
func(row *sql.Row) error {
|
||||
return row.Scan(&durationSeconds)
|
||||
},
|
||||
stmt, args...,
|
||||
)
|
||||
if err != nil {
|
||||
return 0, caos_errors.ThrowInternal(err, "EXEC-Ad8nP", "Errors.Logstore.Execution.ScanFailed")
|
||||
}
|
||||
return durationSeconds, nil
|
||||
}
|
||||
|
||||
func (l *databaseLogStorage) Cleanup(ctx context.Context, keep time.Duration) error {
|
||||
stmt, args, err := squirrel.Delete(executionLogsTable).
|
||||
Where(squirrel.LtOrEq{executionTimestampCol: time.Now().Add(-keep)}).
|
||||
PlaceholderFormat(squirrel.Dollar).
|
||||
ToSql()
|
||||
|
||||
if err != nil {
|
||||
return caos_errors.ThrowInternal(err, "EXEC-Bja8V", "Errors.Internal")
|
||||
}
|
||||
|
||||
execCtx, cancel := context.WithTimeout(ctx, 10*time.Second)
|
||||
defer cancel()
|
||||
_, err = l.dbClient.ExecContext(execCtx, stmt, args...)
|
||||
return err
|
||||
}
|
||||
|
||||
func (l *databaseLogStorage) incrementUsageFromExecutionLogs(ctx context.Context, instanceID string, periodStart time.Time, records []*record.ExecutionLog) (sum uint64, err error) {
|
||||
var total time.Duration
|
||||
for _, r := range records {
|
||||
total += r.Took
|
||||
}
|
||||
return projection.QuotaProjection.IncrementUsage(ctx, quota.ActionsAllRunsSeconds, instanceID, periodStart, uint64(math.Floor(total.Seconds())))
|
||||
}
|
||||
|
||||
Reference in New Issue
Block a user