mirror of
https://github.com/zitadel/zitadel.git
synced 2024-12-22 07:47:29 +00:00
5823fdbef9
* 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 1a49b7d298690ce64846727f1fcf5a325f77c76e)
94 lines
2.9 KiB
Go
94 lines
2.9 KiB
Go
package access
|
|
|
|
import (
|
|
"context"
|
|
"database/sql"
|
|
"errors"
|
|
"time"
|
|
|
|
"github.com/zitadel/zitadel/internal/api/authz"
|
|
"github.com/zitadel/zitadel/internal/command"
|
|
"github.com/zitadel/zitadel/internal/database"
|
|
"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"
|
|
"github.com/zitadel/zitadel/internal/telemetry/tracing"
|
|
)
|
|
|
|
var _ logstore.UsageStorer[*record.AccessLog] = (*databaseLogStorage)(nil)
|
|
|
|
type databaseLogStorage struct {
|
|
dbClient *database.DB
|
|
commands *command.Commands
|
|
queries *query.Queries
|
|
}
|
|
|
|
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.RequestsAllAuthenticated
|
|
}
|
|
|
|
func (l *databaseLogStorage) Emit(ctx context.Context, bulk []*record.AccessLog) error {
|
|
if len(bulk) == 0 {
|
|
return nil
|
|
}
|
|
return l.incrementUsage(ctx, bulk)
|
|
}
|
|
|
|
func (l *databaseLogStorage) incrementUsage(ctx context.Context, bulk []*record.AccessLog) (err error) {
|
|
ctx, span := tracing.NewSpan(ctx)
|
|
defer func() { span.EndWithError(err) }()
|
|
|
|
byInstance := make(map[string][]*record.AccessLog)
|
|
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.RequestsAllAuthenticated)
|
|
if errors.Is(getQuotaErr, sql.ErrNoRows) {
|
|
continue
|
|
}
|
|
err = errors.Join(err, getQuotaErr)
|
|
if getQuotaErr != nil {
|
|
continue
|
|
}
|
|
sum, incrementErr := l.incrementUsageFromAccessLogs(ctx, instanceID, q.CurrentPeriodStart, instanceBulk)
|
|
err = errors.Join(err, incrementErr)
|
|
if incrementErr != nil {
|
|
continue
|
|
}
|
|
notifications, getNotificationErr := l.queries.GetDueQuotaNotifications(ctx, instanceID, quota.RequestsAllAuthenticated, 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
|
|
}
|
|
}
|
|
return err
|
|
}
|
|
|
|
func (l *databaseLogStorage) incrementUsageFromAccessLogs(ctx context.Context, instanceID string, periodStart time.Time, records []*record.AccessLog) (sum uint64, err error) {
|
|
ctx, span := tracing.NewSpan(ctx)
|
|
defer func() { span.EndWithError(err) }()
|
|
|
|
var count uint64
|
|
for _, r := range records {
|
|
if r.IsAuthenticated() {
|
|
count++
|
|
}
|
|
}
|
|
return projection.QuotaProjection.IncrementUsage(ctx, quota.RequestsAllAuthenticated, instanceID, periodStart, count)
|
|
}
|