feat: save last occurrence of failed events and fix instance filtering (#4710)

* fix: filter failed events and current sequence correctly

* fix failed events sorting column

* feat: save last occurrence of failed event

* fix failedEvents query and update sql statements

* change sql statement to only create index

* fix linting

* fix linting

* Update internal/query/failed_events.go

Co-authored-by: Silvan <silvan.reusser@gmail.com>

* update job name on test-docs to match the one from test-code

Co-authored-by: Silvan <silvan.reusser@gmail.com>
This commit is contained in:
Livio Spring
2022-11-18 13:49:38 +01:00
committed by GitHub
parent 6d787bfd62
commit 29441ce4b6
41 changed files with 291 additions and 90 deletions

View File

@@ -14,7 +14,7 @@ on:
- '**.md' - '**.md'
jobs: jobs:
Test: Build-ZITADEL:
runs-on: ubuntu-20.04 runs-on: ubuntu-20.04
steps: steps:
- run: 'echo "No tests for docs are implemented, yet"' - run: 'echo "No tests for docs are implemented, yet"'

View File

@@ -295,3 +295,9 @@ linters:
# Whitespace Linter - Forces you to use empty lines! [fast: true, auto-fix: false] # Whitespace Linter - Forces you to use empty lines! [fast: true, auto-fix: false]
# FUTURE: improves code quality by allowing and blocking line breaks # FUTURE: improves code quality by allowing and blocking line breaks
- wsl - wsl
linters-settings:
gci:
sections:
- standard # Standard section: captures all standard packages.
- default # Default section: contains all imports that could not be matched to another section type.
- prefix(github.com/zitadel/zitadel) # Custom section: groups all imports with the specified Prefix.

25
cmd/setup/05.go Normal file
View File

@@ -0,0 +1,25 @@
package setup
import (
"context"
"database/sql"
_ "embed"
)
var (
//go:embed 05.sql
lastFailedStmts string
)
type LastFailed struct {
dbClient *sql.DB
}
func (mig *LastFailed) Execute(ctx context.Context) error {
_, err := mig.dbClient.ExecContext(ctx, lastFailedStmts)
return err
}
func (mig *LastFailed) String() string {
return "05_last_failed"
}

11
cmd/setup/05.sql Normal file
View File

@@ -0,0 +1,11 @@
CREATE INDEX instance_id_idx ON adminapi.current_sequences (instance_id);
CREATE INDEX instance_id_idx ON auth.current_sequences (instance_id);
CREATE INDEX instance_id_idx ON projections.current_sequences (instance_id);
CREATE INDEX instance_id_idx ON adminapi.failed_events (instance_id);
CREATE INDEX instance_id_idx ON auth.failed_events (instance_id);
CREATE INDEX instance_id_idx ON projections.failed_events (instance_id);
ALTER TABLE adminapi.failed_events ADD COLUMN last_failed TIMESTAMPTZ;
ALTER TABLE auth.failed_events ADD COLUMN last_failed TIMESTAMPTZ;
ALTER TABLE projections.failed_events ADD COLUMN last_failed TIMESTAMPTZ;

View File

@@ -58,6 +58,7 @@ type Steps struct {
s2AssetsTable *AssetTable s2AssetsTable *AssetTable
FirstInstance *FirstInstance FirstInstance *FirstInstance
s4EventstoreIndexes *EventstoreIndexes s4EventstoreIndexes *EventstoreIndexes
s5LastFailed *LastFailed
} }
type encryptionKeyConfig struct { type encryptionKeyConfig struct {

View File

@@ -82,6 +82,7 @@ func Setup(config *Config, steps *Steps, masterKey string) {
steps.FirstInstance.externalPort = config.ExternalPort steps.FirstInstance.externalPort = config.ExternalPort
steps.s4EventstoreIndexes = &EventstoreIndexes{dbClient: dbClient, dbType: config.Database.Type()} steps.s4EventstoreIndexes = &EventstoreIndexes{dbClient: dbClient, dbType: config.Database.Type()}
steps.s5LastFailed = &LastFailed{dbClient: dbClient}
err = projection.Create(ctx, dbClient, eventstoreClient, config.Projections, nil, nil) err = projection.Create(ctx, dbClient, eventstoreClient, config.Projections, nil, nil)
logging.OnError(err).Fatal("unable to start projections") logging.OnError(err).Fatal("unable to start projections")
@@ -107,6 +108,8 @@ func Setup(config *Config, steps *Steps, masterKey string) {
logging.OnError(err).Fatal("unable to migrate step 3") logging.OnError(err).Fatal("unable to migrate step 3")
err = migration.Migrate(ctx, eventstoreClient, steps.s4EventstoreIndexes) err = migration.Migrate(ctx, eventstoreClient, steps.s4EventstoreIndexes)
logging.OnError(err).Fatal("unable to migrate step 4") logging.OnError(err).Fatal("unable to migrate step 4")
err = migration.Migrate(ctx, eventstoreClient, steps.s5LastFailed)
logging.OnError(err).Fatal("unable to migrate step 5")
for _, repeatableStep := range repeatableSteps { for _, repeatableStep := range repeatableSteps {
err = migration.Migrate(ctx, eventstoreClient, repeatableStep) err = migration.Migrate(ctx, eventstoreClient, repeatableStep)

View File

@@ -29,6 +29,13 @@
</td> </td>
</ng-container> </ng-container>
<ng-container matColumnDef="lastFailed">
<th mat-header-cell *matHeaderCellDef>{{ 'IAM.FAILEDEVENTS.LASTFAILED' | translate }}</th>
<td mat-cell *matCellDef="let event">
<span>{{ event?.lastFailed | timestampToDate | localizedDate: 'EEE dd. MMM, HH:mm' }}</span>
</td>
</ng-container>
<ng-container matColumnDef="errorMessage"> <ng-container matColumnDef="errorMessage">
<th mat-header-cell *matHeaderCellDef>{{ 'IAM.FAILEDEVENTS.ERRORMESSAGE' | translate }}</th> <th mat-header-cell *matHeaderCellDef>{{ 'IAM.FAILEDEVENTS.ERRORMESSAGE' | translate }}</th>
<td mat-cell *matCellDef="let event"> <td mat-cell *matCellDef="let event">

View File

@@ -22,6 +22,7 @@ export class FailedEventsComponent implements AfterViewInit {
'database', 'database',
'failedSequence', 'failedSequence',
'failureCount', 'failureCount',
'lastFailed',
'errorMessage', 'errorMessage',
'actions', 'actions',
]; ];

View File

@@ -743,6 +743,7 @@
"DATABASE": "Datenbank", "DATABASE": "Datenbank",
"FAILEDSEQUENCE": "betroffene Sequenz", "FAILEDSEQUENCE": "betroffene Sequenz",
"FAILURECOUNT": "Fehleranzahl", "FAILURECOUNT": "Fehleranzahl",
"LASTFAILED": "Zuletzt gescheitert um",
"ERRORMESSAGE": "Meldung", "ERRORMESSAGE": "Meldung",
"ACTIONS": "Aktionen", "ACTIONS": "Aktionen",
"DELETE": "Entfernen", "DELETE": "Entfernen",

View File

@@ -743,6 +743,7 @@
"DATABASE": "Database", "DATABASE": "Database",
"FAILEDSEQUENCE": "Failed Sequence", "FAILEDSEQUENCE": "Failed Sequence",
"FAILURECOUNT": "Failure Count", "FAILURECOUNT": "Failure Count",
"LASTFAILED": "Last failed at",
"ERRORMESSAGE": "Error Message", "ERRORMESSAGE": "Error Message",
"ACTIONS": "Actions", "ACTIONS": "Actions",
"DELETE": "Remove", "DELETE": "Remove",

View File

@@ -743,6 +743,7 @@
"DATABASE": "Base de données", "DATABASE": "Base de données",
"FAILEDSEQUENCE": "Séquence échouée", "FAILEDSEQUENCE": "Séquence échouée",
"FAILURECOUNT": "Nombre d'échecs", "FAILURECOUNT": "Nombre d'échecs",
"LASTFAILED": "Dernier échec à",
"ERRORMESSAGE": "Message d'erreur", "ERRORMESSAGE": "Message d'erreur",
"ACTIONS": "Actions", "ACTIONS": "Actions",
"DELETE": "Supprimer", "DELETE": "Supprimer",

View File

@@ -743,6 +743,7 @@
"DATABASE": "Database", "DATABASE": "Database",
"FAILEDSEQUENCE": "Sequenza fallita", "FAILEDSEQUENCE": "Sequenza fallita",
"FAILURECOUNT": "Conteggio dei fallimenti", "FAILURECOUNT": "Conteggio dei fallimenti",
"LASTFAILED": "L'ultimo fallimento a",
"ERRORMESSAGE": "Messaggio di errore", "ERRORMESSAGE": "Messaggio di errore",
"ACTIONS": "Azioni", "ACTIONS": "Azioni",
"DELETE": "Rimuovi", "DELETE": "Rimuovi",

View File

@@ -743,6 +743,7 @@
"DATABASE": "数据库", "DATABASE": "数据库",
"FAILEDSEQUENCE": "失败的序列", "FAILEDSEQUENCE": "失败的序列",
"FAILURECOUNT": "失败计数", "FAILURECOUNT": "失败计数",
"LASTFAILED": "最后一次失败是在",
"ERRORMESSAGE": "错误消息", "ERRORMESSAGE": "错误消息",
"ACTIONS": "操作", "ACTIONS": "操作",
"DELETE": "删除", "DELETE": "删除",

View File

@@ -2020,6 +2020,7 @@ This is an empty request
| failed_sequence | uint64 | - | | | failed_sequence | uint64 | - | |
| failure_count | uint64 | - | | | failure_count | uint64 | - | |
| error_message | string | - | | | error_message | string | - | |
| last_failed | google.protobuf.Timestamp | - | |
@@ -4633,7 +4634,7 @@ this is en empty request
| database | string | - | | | database | string | - | |
| view_name | string | - | | | view_name | string | - | |
| processed_sequence | uint64 | - | | | processed_sequence | uint64 | - | |
| event_timestamp | google.protobuf.Timestamp | The timestamp the event occured | | | event_timestamp | google.protobuf.Timestamp | The timestamp the event occurred | |
| last_successful_spooler_run | google.protobuf.Timestamp | - | | | last_successful_spooler_run | google.protobuf.Timestamp | - | |

View File

@@ -375,6 +375,7 @@ This is an empty response
| failed_sequence | uint64 | - | | | failed_sequence | uint64 | - | |
| failure_count | uint64 | - | | | failure_count | uint64 | - | |
| error_message | string | - | | | error_message | string | - | |
| last_failed | google.protobuf.Timestamp | - | |
@@ -556,6 +557,7 @@ This is an empty request
| database | string | - | string.min_len: 1<br /> string.max_len: 200<br /> | | database | string | - | string.min_len: 1<br /> string.max_len: 200<br /> |
| view_name | string | - | string.min_len: 1<br /> string.max_len: 200<br /> | | view_name | string | - | string.min_len: 1<br /> string.max_len: 200<br /> |
| failed_sequence | uint64 | - | | | failed_sequence | uint64 | - | |
| instance_id | string | - | |

View File

@@ -7,8 +7,8 @@ import (
) )
type AdministratorRepository interface { type AdministratorRepository interface {
GetFailedEvents(context.Context) ([]*model.FailedEvent, error) GetFailedEvents(ctx context.Context, instanceID string) ([]*model.FailedEvent, error)
RemoveFailedEvent(context.Context, *model.FailedEvent) error RemoveFailedEvent(context.Context, *model.FailedEvent) error
GetViews() ([]*model.View, error) GetViews(instanceID string) ([]*model.View, error)
ClearView(ctx context.Context, db, viewName string) error ClearView(ctx context.Context, db, viewName string) error
} }

View File

@@ -14,10 +14,10 @@ type AdministratorRepo struct {
View *view.View View *view.View
} }
func (repo *AdministratorRepo) GetFailedEvents(ctx context.Context) ([]*view_model.FailedEvent, error) { func (repo *AdministratorRepo) GetFailedEvents(ctx context.Context, instanceID string) ([]*view_model.FailedEvent, error) {
allFailedEvents := make([]*view_model.FailedEvent, 0) allFailedEvents := make([]*view_model.FailedEvent, 0)
for _, db := range dbList { for _, db := range dbList {
failedEvents, err := repo.View.AllFailedEvents(db) failedEvents, err := repo.View.AllFailedEvents(db, instanceID)
if err != nil { if err != nil {
return nil, err return nil, err
} }
@@ -32,10 +32,10 @@ func (repo *AdministratorRepo) RemoveFailedEvent(ctx context.Context, failedEven
return repo.View.RemoveFailedEvent(failedEvent.Database, repository.FailedEventFromModel(failedEvent)) return repo.View.RemoveFailedEvent(failedEvent.Database, repository.FailedEventFromModel(failedEvent))
} }
func (repo *AdministratorRepo) GetViews() ([]*view_model.View, error) { func (repo *AdministratorRepo) GetViews(instanceID string) ([]*view_model.View, error) {
views := make([]*view_model.View, 0) views := make([]*view_model.View, 0)
for _, db := range dbList { for _, db := range dbList {
sequences, err := repo.View.AllCurrentSequences(db) sequences, err := repo.View.AllCurrentSequences(db, instanceID)
if err != nil { if err != nil {
return nil, err return nil, err
} }

View File

@@ -21,6 +21,6 @@ func (v *View) latestFailedEvent(viewName, instanceID string, sequence uint64) (
return repository.LatestFailedEvent(v.Db, errTable, viewName, instanceID, sequence) return repository.LatestFailedEvent(v.Db, errTable, viewName, instanceID, sequence)
} }
func (v *View) AllFailedEvents(db string) ([]*repository.FailedEvent, error) { func (v *View) AllFailedEvents(db, instanceID string) ([]*repository.FailedEvent, error) {
return repository.AllFailedEvents(v.Db, db+"."+errColumn) return repository.AllFailedEvents(v.Db, db+"."+errColumn, instanceID)
} }

View File

@@ -23,8 +23,8 @@ func (v *View) latestSequences(viewName string, instanceIDs ...string) ([]*repos
return repository.LatestSequences(v.Db, sequencesTable, viewName, instanceIDs...) return repository.LatestSequences(v.Db, sequencesTable, viewName, instanceIDs...)
} }
func (v *View) AllCurrentSequences(db string) ([]*repository.CurrentSequence, error) { func (v *View) AllCurrentSequences(db, instanceID string) ([]*repository.CurrentSequence, error) {
return repository.AllCurrentSequences(v.Db, db+".current_sequences") return repository.AllCurrentSequences(v.Db, db+".current_sequences", instanceID)
} }
func (v *View) updateSpoolerRunSequence(viewName string) error { func (v *View) updateSpoolerRunSequence(viewName string) error {

View File

@@ -3,18 +3,25 @@ package admin
import ( import (
"context" "context"
"github.com/zitadel/zitadel/internal/api/authz"
"github.com/zitadel/zitadel/internal/query" "github.com/zitadel/zitadel/internal/query"
admin_pb "github.com/zitadel/zitadel/pkg/grpc/admin" admin_pb "github.com/zitadel/zitadel/pkg/grpc/admin"
) )
func (s *Server) ListFailedEvents(ctx context.Context, req *admin_pb.ListFailedEventsRequest) (*admin_pb.ListFailedEventsResponse, error) { func (s *Server) ListFailedEvents(ctx context.Context, _ *admin_pb.ListFailedEventsRequest) (*admin_pb.ListFailedEventsResponse, error) {
failedEventsOld, err := s.administrator.GetFailedEvents(ctx) instanceID := authz.GetInstance(ctx).InstanceID()
failedEventsOld, err := s.administrator.GetFailedEvents(ctx, instanceID)
if err != nil { if err != nil {
return nil, err return nil, err
} }
convertedOld := FailedEventsViewToPb(failedEventsOld) convertedOld := FailedEventsViewToPb(failedEventsOld)
instanceIDQuery, err := query.NewFailedEventInstanceIDSearchQuery(instanceID)
failedEvents, err := s.query.SearchFailedEvents(ctx, new(query.FailedEventSearchQueries)) if err != nil {
return nil, err
}
failedEvents, err := s.query.SearchFailedEvents(ctx, &query.FailedEventSearchQueries{
Queries: []query.SearchQuery{instanceIDQuery},
})
if err != nil { if err != nil {
return nil, err return nil, err
} }
@@ -25,9 +32,9 @@ func (s *Server) ListFailedEvents(ctx context.Context, req *admin_pb.ListFailedE
func (s *Server) RemoveFailedEvent(ctx context.Context, req *admin_pb.RemoveFailedEventRequest) (*admin_pb.RemoveFailedEventResponse, error) { func (s *Server) RemoveFailedEvent(ctx context.Context, req *admin_pb.RemoveFailedEventRequest) (*admin_pb.RemoveFailedEventResponse, error) {
var err error var err error
if req.Database != s.database { if req.Database != s.database {
err = s.administrator.RemoveFailedEvent(ctx, RemoveFailedEventRequestToModel(req)) err = s.administrator.RemoveFailedEvent(ctx, RemoveFailedEventRequestToModel(ctx, req))
} else { } else {
err = s.query.RemoveFailedEvent(ctx, req.ViewName, req.FailedSequence) err = s.query.RemoveFailedEvent(ctx, req.ViewName, authz.GetInstance(ctx).InstanceID(), req.FailedSequence)
} }
if err != nil { if err != nil {
return nil, err return nil, err

View File

@@ -1,6 +1,11 @@
package admin package admin
import ( import (
"context"
"google.golang.org/protobuf/types/known/timestamppb"
"github.com/zitadel/zitadel/internal/api/authz"
"github.com/zitadel/zitadel/internal/query" "github.com/zitadel/zitadel/internal/query"
"github.com/zitadel/zitadel/internal/view/model" "github.com/zitadel/zitadel/internal/view/model"
admin_pb "github.com/zitadel/zitadel/pkg/grpc/admin" admin_pb "github.com/zitadel/zitadel/pkg/grpc/admin"
@@ -15,12 +20,17 @@ func FailedEventsViewToPb(failedEvents []*model.FailedEvent) []*admin_pb.FailedE
} }
func FailedEventViewToPb(failedEvent *model.FailedEvent) *admin_pb.FailedEvent { func FailedEventViewToPb(failedEvent *model.FailedEvent) *admin_pb.FailedEvent {
var lastFailed *timestamppb.Timestamp
if !failedEvent.LastFailed.IsZero() {
lastFailed = timestamppb.New(failedEvent.LastFailed)
}
return &admin_pb.FailedEvent{ return &admin_pb.FailedEvent{
Database: failedEvent.Database, Database: failedEvent.Database,
ViewName: failedEvent.ViewName, ViewName: failedEvent.ViewName,
FailedSequence: failedEvent.FailedSequence, FailedSequence: failedEvent.FailedSequence,
FailureCount: failedEvent.FailureCount, FailureCount: failedEvent.FailureCount,
ErrorMessage: failedEvent.ErrMsg, ErrorMessage: failedEvent.ErrMsg,
LastFailed: lastFailed,
} }
} }
@@ -33,19 +43,25 @@ func FailedEventsToPb(database string, failedEvents *query.FailedEvents) []*admi
} }
func FailedEventToPb(database string, failedEvent *query.FailedEvent) *admin_pb.FailedEvent { func FailedEventToPb(database string, failedEvent *query.FailedEvent) *admin_pb.FailedEvent {
var lastFailed *timestamppb.Timestamp
if !failedEvent.LastFailed.IsZero() {
lastFailed = timestamppb.New(failedEvent.LastFailed)
}
return &admin_pb.FailedEvent{ return &admin_pb.FailedEvent{
Database: database, Database: database,
ViewName: failedEvent.ProjectionName, ViewName: failedEvent.ProjectionName,
FailedSequence: failedEvent.FailedSequence, FailedSequence: failedEvent.FailedSequence,
FailureCount: failedEvent.FailureCount, FailureCount: failedEvent.FailureCount,
ErrorMessage: failedEvent.Error, ErrorMessage: failedEvent.Error,
LastFailed: lastFailed,
} }
} }
func RemoveFailedEventRequestToModel(req *admin_pb.RemoveFailedEventRequest) *model.FailedEvent { func RemoveFailedEventRequestToModel(ctx context.Context, req *admin_pb.RemoveFailedEventRequest) *model.FailedEvent {
return &model.FailedEvent{ return &model.FailedEvent{
Database: req.Database, Database: req.Database,
ViewName: req.ViewName, ViewName: req.ViewName,
FailedSequence: req.FailedSequence, FailedSequence: req.FailedSequence,
InstanceID: authz.GetInstance(ctx).InstanceID(),
} }
} }

View File

@@ -1,8 +1,11 @@
package admin package admin
import ( import (
"context"
"testing" "testing"
"time"
"github.com/zitadel/zitadel/internal/api/authz"
"github.com/zitadel/zitadel/internal/test" "github.com/zitadel/zitadel/internal/test"
"github.com/zitadel/zitadel/internal/view/model" "github.com/zitadel/zitadel/internal/view/model"
admin_pb "github.com/zitadel/zitadel/pkg/grpc/admin" admin_pb "github.com/zitadel/zitadel/pkg/grpc/admin"
@@ -25,6 +28,7 @@ func TestFailedEventsToPbFields(t *testing.T) {
ViewName: "users", ViewName: "users",
FailedSequence: 456, FailedSequence: 456,
FailureCount: 5, FailureCount: 5,
LastFailed: time.Now(),
ErrMsg: "some error", ErrMsg: "some error",
}, },
}, },
@@ -57,6 +61,7 @@ func TestFailedEventToPbFields(t *testing.T) {
ViewName: "users", ViewName: "users",
FailedSequence: 456, FailedSequence: 456,
FailureCount: 5, FailureCount: 5,
LastFailed: time.Now(),
ErrMsg: "some error", ErrMsg: "some error",
}, },
}, },
@@ -70,6 +75,7 @@ func TestFailedEventToPbFields(t *testing.T) {
func TestRemoveFailedEventRequestToModelFields(t *testing.T) { func TestRemoveFailedEventRequestToModelFields(t *testing.T) {
type args struct { type args struct {
ctx context.Context
req *admin_pb.RemoveFailedEventRequest req *admin_pb.RemoveFailedEventRequest
} }
tests := []struct { tests := []struct {
@@ -79,6 +85,7 @@ func TestRemoveFailedEventRequestToModelFields(t *testing.T) {
{ {
"all fields", "all fields",
args{ args{
ctx: authz.WithInstanceID(context.Background(), "instanceID"),
req: &admin_pb.RemoveFailedEventRequest{ req: &admin_pb.RemoveFailedEventRequest{
Database: "admin", Database: "admin",
ViewName: "users", ViewName: "users",
@@ -88,7 +95,7 @@ func TestRemoveFailedEventRequestToModelFields(t *testing.T) {
}, },
} }
for _, tt := range tests { for _, tt := range tests {
converted := RemoveFailedEventRequestToModel(tt.args.req) converted := RemoveFailedEventRequestToModel(tt.args.ctx, tt.args.req)
test.AssertFieldsMapped(t, converted, "FailureCount", "ErrMsg") test.AssertFieldsMapped(t, converted, "FailureCount", "LastFailed", "ErrMsg")
} }
} }

View File

@@ -3,17 +3,25 @@ package admin
import ( import (
"context" "context"
"github.com/zitadel/zitadel/internal/api/authz"
"github.com/zitadel/zitadel/internal/query" "github.com/zitadel/zitadel/internal/query"
admin_pb "github.com/zitadel/zitadel/pkg/grpc/admin" admin_pb "github.com/zitadel/zitadel/pkg/grpc/admin"
) )
func (s *Server) ListViews(ctx context.Context, _ *admin_pb.ListViewsRequest) (*admin_pb.ListViewsResponse, error) { func (s *Server) ListViews(ctx context.Context, _ *admin_pb.ListViewsRequest) (*admin_pb.ListViewsResponse, error) {
currentSequences, err := s.query.SearchCurrentSequences(ctx, new(query.CurrentSequencesSearchQueries)) instanceID := authz.GetInstance(ctx).InstanceID()
instanceIDQuery, err := query.NewCurrentSequencesInstanceIDSearchQuery(instanceID)
if err != nil {
return nil, err
}
currentSequences, err := s.query.SearchCurrentSequences(ctx, &query.CurrentSequencesSearchQueries{
Queries: []query.SearchQuery{instanceIDQuery},
})
if err != nil { if err != nil {
return nil, err return nil, err
} }
convertedCurrentSequences := CurrentSequencesToPb(s.database, currentSequences) convertedCurrentSequences := CurrentSequencesToPb(s.database, currentSequences)
views, err := s.administrator.GetViews() views, err := s.administrator.GetViews(instanceID)
if err != nil { if err != nil {
return nil, err return nil, err
} }

View File

@@ -1,10 +1,11 @@
package admin package admin
import ( import (
"google.golang.org/protobuf/types/known/timestamppb"
"github.com/zitadel/zitadel/internal/query" "github.com/zitadel/zitadel/internal/query"
"github.com/zitadel/zitadel/internal/view/model" "github.com/zitadel/zitadel/internal/view/model"
admin_pb "github.com/zitadel/zitadel/pkg/grpc/admin" admin_pb "github.com/zitadel/zitadel/pkg/grpc/admin"
"google.golang.org/protobuf/types/known/timestamppb"
) )
func ViewsToPb(views []*model.View) []*admin_pb.View { func ViewsToPb(views []*model.View) []*admin_pb.View {
@@ -35,9 +36,9 @@ func CurrentSequencesToPb(database string, currentSequences *query.CurrentSequen
func CurrentSequenceToPb(database string, currentSequence *query.CurrentSequence) *admin_pb.View { func CurrentSequenceToPb(database string, currentSequence *query.CurrentSequence) *admin_pb.View {
return &admin_pb.View{ return &admin_pb.View{
Database: database, Database: database,
ViewName: currentSequence.ProjectionName, ViewName: currentSequence.ProjectionName,
ProcessedSequence: currentSequence.CurrentSequence, ProcessedSequence: currentSequence.CurrentSequence,
EventTimestamp: timestamppb.New(currentSequence.Timestamp), LastSuccessfulSpoolerRun: timestamppb.New(currentSequence.Timestamp),
} }
} }

View File

@@ -7,8 +7,8 @@ import (
system_pb "github.com/zitadel/zitadel/pkg/grpc/system" system_pb "github.com/zitadel/zitadel/pkg/grpc/system"
) )
func (s *Server) ListFailedEvents(ctx context.Context, req *system_pb.ListFailedEventsRequest) (*system_pb.ListFailedEventsResponse, error) { func (s *Server) ListFailedEvents(ctx context.Context, _ *system_pb.ListFailedEventsRequest) (*system_pb.ListFailedEventsResponse, error) {
failedEventsOld, err := s.administrator.GetFailedEvents(ctx) failedEventsOld, err := s.administrator.GetFailedEvents(ctx, "")
if err != nil { if err != nil {
return nil, err return nil, err
} }
@@ -27,7 +27,7 @@ func (s *Server) RemoveFailedEvent(ctx context.Context, req *system_pb.RemoveFai
if req.Database != s.database { if req.Database != s.database {
err = s.administrator.RemoveFailedEvent(ctx, RemoveFailedEventRequestToModel(req)) err = s.administrator.RemoveFailedEvent(ctx, RemoveFailedEventRequestToModel(req))
} else { } else {
err = s.query.RemoveFailedEvent(ctx, req.ViewName, req.FailedSequence) err = s.query.RemoveFailedEvent(ctx, req.ViewName, req.InstanceId, req.FailedSequence)
} }
if err != nil { if err != nil {
return nil, err return nil, err

View File

@@ -1,6 +1,8 @@
package system package system
import ( import (
"google.golang.org/protobuf/types/known/timestamppb"
"github.com/zitadel/zitadel/internal/query" "github.com/zitadel/zitadel/internal/query"
"github.com/zitadel/zitadel/internal/view/model" "github.com/zitadel/zitadel/internal/view/model"
system_pb "github.com/zitadel/zitadel/pkg/grpc/system" system_pb "github.com/zitadel/zitadel/pkg/grpc/system"
@@ -15,12 +17,17 @@ func FailedEventsViewToPb(failedEvents []*model.FailedEvent) []*system_pb.Failed
} }
func FailedEventViewToPb(failedEvent *model.FailedEvent) *system_pb.FailedEvent { func FailedEventViewToPb(failedEvent *model.FailedEvent) *system_pb.FailedEvent {
var lastFailed *timestamppb.Timestamp
if !failedEvent.LastFailed.IsZero() {
lastFailed = timestamppb.New(failedEvent.LastFailed)
}
return &system_pb.FailedEvent{ return &system_pb.FailedEvent{
Database: failedEvent.Database, Database: failedEvent.Database,
ViewName: failedEvent.ViewName, ViewName: failedEvent.ViewName,
FailedSequence: failedEvent.FailedSequence, FailedSequence: failedEvent.FailedSequence,
FailureCount: failedEvent.FailureCount, FailureCount: failedEvent.FailureCount,
ErrorMessage: failedEvent.ErrMsg, ErrorMessage: failedEvent.ErrMsg,
LastFailed: lastFailed,
} }
} }
@@ -33,12 +40,17 @@ func FailedEventsToPb(database string, failedEvents *query.FailedEvents) []*syst
} }
func FailedEventToPb(database string, failedEvent *query.FailedEvent) *system_pb.FailedEvent { func FailedEventToPb(database string, failedEvent *query.FailedEvent) *system_pb.FailedEvent {
var lastFailed *timestamppb.Timestamp
if !failedEvent.LastFailed.IsZero() {
lastFailed = timestamppb.New(failedEvent.LastFailed)
}
return &system_pb.FailedEvent{ return &system_pb.FailedEvent{
Database: database, Database: database,
ViewName: failedEvent.ProjectionName, ViewName: failedEvent.ProjectionName,
FailedSequence: failedEvent.FailedSequence, FailedSequence: failedEvent.FailedSequence,
FailureCount: failedEvent.FailureCount, FailureCount: failedEvent.FailureCount,
ErrorMessage: failedEvent.Error, ErrorMessage: failedEvent.Error,
LastFailed: lastFailed,
} }
} }
@@ -47,5 +59,6 @@ func RemoveFailedEventRequestToModel(req *system_pb.RemoveFailedEventRequest) *m
Database: req.Database, Database: req.Database,
ViewName: req.ViewName, ViewName: req.ViewName,
FailedSequence: req.FailedSequence, FailedSequence: req.FailedSequence,
InstanceID: req.InstanceId,
} }
} }

View File

@@ -2,6 +2,7 @@ package system_test
import ( import (
"testing" "testing"
"time"
system_grpc "github.com/zitadel/zitadel/internal/api/grpc/system" system_grpc "github.com/zitadel/zitadel/internal/api/grpc/system"
"github.com/zitadel/zitadel/internal/test" "github.com/zitadel/zitadel/internal/test"
@@ -26,6 +27,7 @@ func TestFailedEventsToPbFields(t *testing.T) {
ViewName: "users", ViewName: "users",
FailedSequence: 456, FailedSequence: 456,
FailureCount: 5, FailureCount: 5,
LastFailed: time.Now(),
ErrMsg: "some error", ErrMsg: "some error",
}, },
}, },
@@ -58,6 +60,7 @@ func TestFailedEventToPbFields(t *testing.T) {
ViewName: "users", ViewName: "users",
FailedSequence: 456, FailedSequence: 456,
FailureCount: 5, FailureCount: 5,
LastFailed: time.Now(),
ErrMsg: "some error", ErrMsg: "some error",
}, },
}, },
@@ -84,12 +87,13 @@ func TestRemoveFailedEventRequestToModelFields(t *testing.T) {
Database: "admin", Database: "admin",
ViewName: "users", ViewName: "users",
FailedSequence: 456, FailedSequence: 456,
InstanceId: "instanceID",
}, },
}, },
}, },
} }
for _, tt := range tests { for _, tt := range tests {
converted := system_grpc.RemoveFailedEventRequestToModel(tt.args.req) converted := system_grpc.RemoveFailedEventRequestToModel(tt.args.req)
test.AssertFieldsMapped(t, converted, "FailureCount", "ErrMsg") test.AssertFieldsMapped(t, converted, "FailureCount", "LastFailed", "ErrMsg")
} }
} }

View File

@@ -13,7 +13,7 @@ func (s *Server) ListViews(ctx context.Context, _ *system_pb.ListViewsRequest) (
return nil, err return nil, err
} }
convertedCurrentSequences := CurrentSequencesToPb(s.database, currentSequences) convertedCurrentSequences := CurrentSequencesToPb(s.database, currentSequences)
views, err := s.administrator.GetViews() views, err := s.administrator.GetViews("")
if err != nil { if err != nil {
return nil, err return nil, err
} }

View File

@@ -1,10 +1,11 @@
package system package system
import ( import (
"google.golang.org/protobuf/types/known/timestamppb"
"github.com/zitadel/zitadel/internal/query" "github.com/zitadel/zitadel/internal/query"
"github.com/zitadel/zitadel/internal/view/model" "github.com/zitadel/zitadel/internal/view/model"
system_pb "github.com/zitadel/zitadel/pkg/grpc/system" system_pb "github.com/zitadel/zitadel/pkg/grpc/system"
"google.golang.org/protobuf/types/known/timestamppb"
) )
func ViewsToPb(views []*model.View) []*system_pb.View { func ViewsToPb(views []*model.View) []*system_pb.View {
@@ -35,9 +36,9 @@ func CurrentSequencesToPb(database string, currentSequences *query.CurrentSequen
func CurrentSequenceToPb(database string, currentSequence *query.CurrentSequence) *system_pb.View { func CurrentSequenceToPb(database string, currentSequence *query.CurrentSequence) *system_pb.View {
return &system_pb.View{ return &system_pb.View{
Database: database, Database: database,
ViewName: currentSequence.ProjectionName, ViewName: currentSequence.ProjectionName,
ProcessedSequence: currentSequence.CurrentSequence, ProcessedSequence: currentSequence.CurrentSequence,
EventTimestamp: timestamppb.New(currentSequence.Timestamp), LastSuccessfulSpoolerRun: timestamppb.New(currentSequence.Timestamp),
} }
} }

View File

@@ -28,8 +28,8 @@ func expectFailureCount(tableName string, projectionName, instanceID string, fai
func expectUpdateFailureCount(tableName string, projectionName, instanceID string, seq, failureCount uint64) func(sqlmock.Sqlmock) { func expectUpdateFailureCount(tableName string, projectionName, instanceID string, seq, failureCount uint64) func(sqlmock.Sqlmock) {
return func(m sqlmock.Sqlmock) { return func(m sqlmock.Sqlmock) {
m.ExpectExec(`INSERT INTO `+tableName+` \(projection_name, failed_sequence, failure_count, error, instance_id\) VALUES \(\$1, \$2, \$3, \$4\, \$5\) ON CONFLICT \(projection_name, failed_sequence, instance_id\) DO UPDATE SET failure_count = EXCLUDED\.failure_count, error = EXCLUDED\.error`). m.ExpectExec(`INSERT INTO `+tableName+` \(projection_name, failed_sequence, failure_count, error, instance_id, last_failed\) VALUES \(\$1, \$2, \$3, \$4\, \$5\, \$6\) ON CONFLICT \(projection_name, failed_sequence, instance_id\) DO UPDATE SET failure_count = EXCLUDED\.failure_count, error = EXCLUDED\.error, last_failed = EXCLUDED\.last_failed`).
WithArgs(projectionName, seq, failureCount, sqlmock.AnyArg(), instanceID).WillReturnResult(sqlmock.NewResult(1, 1)) WithArgs(projectionName, seq, failureCount, sqlmock.AnyArg(), instanceID, "NOW()").WillReturnResult(sqlmock.NewResult(1, 1))
} }
} }

View File

@@ -11,9 +11,9 @@ import (
const ( const (
setFailureCountStmtFormat = "INSERT INTO %s" + setFailureCountStmtFormat = "INSERT INTO %s" +
" (projection_name, failed_sequence, failure_count, error, instance_id)" + " (projection_name, failed_sequence, failure_count, error, instance_id, last_failed)" +
" VALUES ($1, $2, $3, $4, $5) ON CONFLICT (projection_name, failed_sequence, instance_id)" + " VALUES ($1, $2, $3, $4, $5, $6) ON CONFLICT (projection_name, failed_sequence, instance_id)" +
" DO UPDATE SET failure_count = EXCLUDED.failure_count, error = EXCLUDED.error" " DO UPDATE SET failure_count = EXCLUDED.failure_count, error = EXCLUDED.error, last_failed = EXCLUDED.last_failed"
failureCountStmtFormat = "WITH failures AS (SELECT failure_count FROM %s WHERE projection_name = $1 AND failed_sequence = $2 AND instance_id = $3)" + failureCountStmtFormat = "WITH failures AS (SELECT failure_count FROM %s WHERE projection_name = $1 AND failed_sequence = $2 AND instance_id = $3)" +
" SELECT COALESCE((SELECT failure_count FROM failures), 0) AS failure_count" " SELECT COALESCE((SELECT failure_count FROM failures), 0) AS failure_count"
) )
@@ -43,7 +43,7 @@ func (h *StatementHandler) failureCount(tx *sql.Tx, seq uint64, instanceID strin
} }
func (h *StatementHandler) setFailureCount(tx *sql.Tx, seq uint64, count uint, err error, instanceID string) error { func (h *StatementHandler) setFailureCount(tx *sql.Tx, seq uint64, count uint, err error, instanceID string) error {
_, dbErr := tx.Exec(h.setFailureCountStmt, h.ProjectionName, seq, count, err.Error(), instanceID) _, dbErr := tx.Exec(h.setFailureCountStmt, h.ProjectionName, seq, count, err.Error(), instanceID, "NOW()")
if dbErr != nil { if dbErr != nil {
return errors.ThrowInternal(dbErr, "CRDB-4Ht4x", "set failure count failed") return errors.ThrowInternal(dbErr, "CRDB-4Ht4x", "set failure count failed")
} }

View File

@@ -135,7 +135,7 @@ func (s *spooledHandler) awaitError(cancel func(), errs chan error, workerID str
select { select {
case err := <-errs: case err := <-errs:
cancel() cancel()
logging.Log("SPOOL-OT8di").OnError(err).WithField("view", s.ViewModel()).WithField("worker", workerID).Debug("load canceled") logging.OnError(err).WithField("view", s.ViewModel()).WithField("worker", workerID).Debug("load canceled")
} }
} }
@@ -216,6 +216,7 @@ func HandleError(event *models.Event, failedErr error,
failedEvent.FailureCount++ failedEvent.FailureCount++
failedEvent.ErrMsg = failedErr.Error() failedEvent.ErrMsg = failedErr.Error()
failedEvent.InstanceID = event.InstanceID failedEvent.InstanceID = event.InstanceID
failedEvent.LastFailed = time.Now()
err = processFailedEvent(failedEvent) err = processFailedEvent(failedEvent)
if err != nil { if err != nil {
return err return err

View File

@@ -17,6 +17,10 @@ import (
"github.com/zitadel/zitadel/internal/view/repository" "github.com/zitadel/zitadel/internal/view/repository"
) )
var (
testNow = time.Now()
)
type testHandler struct { type testHandler struct {
cycleDuration time.Duration cycleDuration time.Duration
processSleep time.Duration processSleep time.Duration
@@ -429,6 +433,7 @@ func TestHandleError(t *testing.T) {
FailureCount: 6, FailureCount: 6,
ViewName: "super.table", ViewName: "super.table",
InstanceID: instanceID, InstanceID: instanceID,
LastFailed: testNow,
}, nil }, nil
}, },
errorCountUntilSkip: 5, errorCountUntilSkip: 5,
@@ -449,6 +454,7 @@ func TestHandleError(t *testing.T) {
FailureCount: 5, FailureCount: 5,
ViewName: "super.table", ViewName: "super.table",
InstanceID: instanceID, InstanceID: instanceID,
LastFailed: testNow,
}, nil }, nil
}, },
errorCountUntilSkip: 6, errorCountUntilSkip: 6,
@@ -469,6 +475,7 @@ func TestHandleError(t *testing.T) {
FailureCount: 3, FailureCount: 3,
ViewName: "super.table", ViewName: "super.table",
InstanceID: instanceID, InstanceID: instanceID,
LastFailed: testNow,
}, nil }, nil
}, },
errorCountUntilSkip: 5, errorCountUntilSkip: 5,

View File

@@ -42,6 +42,10 @@ type CurrentSequencesSearchQueries struct {
Queries []SearchQuery Queries []SearchQuery
} }
func NewCurrentSequencesInstanceIDSearchQuery(instanceID string) (SearchQuery, error) {
return NewTextQuery(CurrentSequenceColInstanceID, instanceID, TextEquals)
}
func (q *CurrentSequencesSearchQueries) toQuery(query sq.SelectBuilder) sq.SelectBuilder { func (q *CurrentSequencesSearchQueries) toQuery(query sq.SelectBuilder) sq.SelectBuilder {
query = q.SearchRequest.toQuery(query) query = q.SearchRequest.toQuery(query)
for _, q := range q.Queries { for _, q := range q.Queries {

View File

@@ -3,7 +3,7 @@ package query
import ( import (
"context" "context"
"database/sql" "database/sql"
errs "errors" "time"
sq "github.com/Masterminds/squirrel" sq "github.com/Masterminds/squirrel"
@@ -15,6 +15,7 @@ const (
failedEventsColumnProjectionName = "projection_name" failedEventsColumnProjectionName = "projection_name"
failedEventsColumnFailedSequence = "failed_sequence" failedEventsColumnFailedSequence = "failed_sequence"
failedEventsColumnFailureCount = "failure_count" failedEventsColumnFailureCount = "failure_count"
failedEventsColumnLastFailed = "last_failed"
failedEventsColumnError = "error" failedEventsColumnError = "error"
failedEventsColumnInstanceID = "instance_id" failedEventsColumnInstanceID = "instance_id"
) )
@@ -36,10 +37,18 @@ var (
name: failedEventsColumnFailureCount, name: failedEventsColumnFailureCount,
table: failedEventsTable, table: failedEventsTable,
} }
FailedEventsColumnLastFailed = Column{
name: failedEventsColumnLastFailed,
table: failedEventsTable,
}
FailedEventsColumnError = Column{ FailedEventsColumnError = Column{
name: failedEventsColumnError, name: failedEventsColumnError,
table: failedEventsTable, table: failedEventsTable,
} }
FailedEventsColumnInstanceID = Column{
name: failedEventsColumnInstanceID,
table: failedEventsTable,
}
) )
type FailedEvents struct { type FailedEvents struct {
@@ -52,6 +61,7 @@ type FailedEvent struct {
FailedSequence uint64 FailedSequence uint64
FailureCount uint64 FailureCount uint64
Error string Error string
LastFailed time.Time
} }
type FailedEventSearchQueries struct { type FailedEventSearchQueries struct {
@@ -73,26 +83,27 @@ func (q *Queries) SearchFailedEvents(ctx context.Context, queries *FailedEventSe
return scan(rows) return scan(rows)
} }
func (q *Queries) RemoveFailedEvent(ctx context.Context, projectionName string, sequence uint64) (err error) { func (q *Queries) RemoveFailedEvent(ctx context.Context, projectionName, instanceID string, sequence uint64) (err error) {
stmt, args, err := sq.Delete(projection.FailedEventsTable). stmt, args, err := sq.Delete(projection.FailedEventsTable).
Where(sq.Eq{ Where(sq.Eq{
failedEventsColumnProjectionName: projectionName, failedEventsColumnProjectionName: projectionName,
failedEventsColumnFailedSequence: sequence, failedEventsColumnFailedSequence: sequence,
failedEventsColumnInstanceID: instanceID,
}). }).
PlaceholderFormat(sq.Dollar). PlaceholderFormat(sq.Dollar).
ToSql() ToSql()
if err != nil { if err != nil {
return errors.ThrowInternal(err, "QUERY-DGgh3", "Errors.RemoveFailed") return errors.ThrowInternal(err, "QUERY-DGgh3", "Errors.RemoveFailed")
} }
_, err = q.client.Exec(stmt, args...) _, err = q.client.ExecContext(ctx, stmt, args...)
if err != nil { if err != nil {
return errors.ThrowInternal(err, "QUERY-0kbFF", "Errors.RemoveFailed") return errors.ThrowInternal(err, "QUERY-0kbFF", "Errors.RemoveFailed")
} }
return nil return nil
} }
func NewFailedEventProjectionNameSearchQuery(method TextComparison, value string) (SearchQuery, error) { func NewFailedEventInstanceIDSearchQuery(instanceID string) (SearchQuery, error) {
return NewTextQuery(FailedEventsColumnProjectionName, value, method) return NewTextQuery(FailedEventsColumnInstanceID, instanceID, TextEquals)
} }
func (r *ProjectSearchQueries) AppendProjectionNameQuery(projectionName string) error { func (r *ProjectSearchQueries) AppendProjectionNameQuery(projectionName string) error {
@@ -112,36 +123,12 @@ func (q *FailedEventSearchQueries) toQuery(query sq.SelectBuilder) sq.SelectBuil
return query return query
} }
func prepareFailedEventQuery(instanceIDs ...string) (sq.SelectBuilder, func(*sql.Row) (*FailedEvent, error)) {
return sq.Select(
FailedEventsColumnProjectionName.identifier(),
FailedEventsColumnFailedSequence.identifier(),
FailedEventsColumnFailureCount.identifier(),
FailedEventsColumnError.identifier()).
From(failedEventsTable.identifier()).PlaceholderFormat(sq.Dollar),
func(row *sql.Row) (*FailedEvent, error) {
p := new(FailedEvent)
err := row.Scan(
&p.ProjectionName,
&p.FailedSequence,
&p.FailureCount,
&p.Error,
)
if err != nil {
if errs.Is(err, sql.ErrNoRows) {
return nil, errors.ThrowNotFound(err, "QUERY-5N00f", "Errors.FailedEvents.NotFound")
}
return nil, errors.ThrowInternal(err, "QUERY-0oJf3", "Errors.Internal")
}
return p, nil
}
}
func prepareFailedEventsQuery() (sq.SelectBuilder, func(*sql.Rows) (*FailedEvents, error)) { func prepareFailedEventsQuery() (sq.SelectBuilder, func(*sql.Rows) (*FailedEvents, error)) {
return sq.Select( return sq.Select(
FailedEventsColumnProjectionName.identifier(), FailedEventsColumnProjectionName.identifier(),
FailedEventsColumnFailedSequence.identifier(), FailedEventsColumnFailedSequence.identifier(),
FailedEventsColumnFailureCount.identifier(), FailedEventsColumnFailureCount.identifier(),
FailedEventsColumnLastFailed.identifier(),
FailedEventsColumnError.identifier(), FailedEventsColumnError.identifier(),
countColumn.identifier()). countColumn.identifier()).
From(failedEventsTable.identifier()).PlaceholderFormat(sq.Dollar), From(failedEventsTable.identifier()).PlaceholderFormat(sq.Dollar),
@@ -150,16 +137,19 @@ func prepareFailedEventsQuery() (sq.SelectBuilder, func(*sql.Rows) (*FailedEvent
var count uint64 var count uint64
for rows.Next() { for rows.Next() {
failedEvent := new(FailedEvent) failedEvent := new(FailedEvent)
var lastFailed sql.NullTime
err := rows.Scan( err := rows.Scan(
&failedEvent.ProjectionName, &failedEvent.ProjectionName,
&failedEvent.FailedSequence, &failedEvent.FailedSequence,
&failedEvent.FailureCount, &failedEvent.FailureCount,
&lastFailed,
&failedEvent.Error, &failedEvent.Error,
&count, &count,
) )
if err != nil { if err != nil {
return nil, err return nil, err
} }
failedEvent.LastFailed = lastFailed.Time
failedEvents = append(failedEvents, failedEvent) failedEvents = append(failedEvents, failedEvent)
} }

View File

@@ -28,6 +28,7 @@ func Test_FailedEventsPrepares(t *testing.T) {
regexp.QuoteMeta(`SELECT projections.failed_events.projection_name,`+ regexp.QuoteMeta(`SELECT projections.failed_events.projection_name,`+
` projections.failed_events.failed_sequence,`+ ` projections.failed_events.failed_sequence,`+
` projections.failed_events.failure_count,`+ ` projections.failed_events.failure_count,`+
` projections.failed_events.last_failed,`+
` projections.failed_events.error,`+ ` projections.failed_events.error,`+
` COUNT(*) OVER ()`+ ` COUNT(*) OVER ()`+
` FROM projections.failed_events`), ` FROM projections.failed_events`),
@@ -45,6 +46,7 @@ func Test_FailedEventsPrepares(t *testing.T) {
regexp.QuoteMeta(`SELECT projections.failed_events.projection_name,`+ regexp.QuoteMeta(`SELECT projections.failed_events.projection_name,`+
` projections.failed_events.failed_sequence,`+ ` projections.failed_events.failed_sequence,`+
` projections.failed_events.failure_count,`+ ` projections.failed_events.failure_count,`+
` projections.failed_events.last_failed,`+
` projections.failed_events.error,`+ ` projections.failed_events.error,`+
` COUNT(*) OVER ()`+ ` COUNT(*) OVER ()`+
` FROM projections.failed_events`), ` FROM projections.failed_events`),
@@ -52,6 +54,7 @@ func Test_FailedEventsPrepares(t *testing.T) {
"projection_name", "projection_name",
"failed_sequence", "failed_sequence",
"failure_count", "failure_count",
"last_failed",
"error", "error",
"count", "count",
}, },
@@ -60,6 +63,7 @@ func Test_FailedEventsPrepares(t *testing.T) {
"projection-name", "projection-name",
uint64(20211108), uint64(20211108),
uint64(2), uint64(2),
testNow,
"error", "error",
}, },
}, },
@@ -74,6 +78,7 @@ func Test_FailedEventsPrepares(t *testing.T) {
ProjectionName: "projection-name", ProjectionName: "projection-name",
FailedSequence: 20211108, FailedSequence: 20211108,
FailureCount: 2, FailureCount: 2,
LastFailed: testNow,
Error: "error", Error: "error",
}, },
}, },
@@ -87,6 +92,7 @@ func Test_FailedEventsPrepares(t *testing.T) {
regexp.QuoteMeta(`SELECT projections.failed_events.projection_name,`+ regexp.QuoteMeta(`SELECT projections.failed_events.projection_name,`+
` projections.failed_events.failed_sequence,`+ ` projections.failed_events.failed_sequence,`+
` projections.failed_events.failure_count,`+ ` projections.failed_events.failure_count,`+
` projections.failed_events.last_failed,`+
` projections.failed_events.error,`+ ` projections.failed_events.error,`+
` COUNT(*) OVER ()`+ ` COUNT(*) OVER ()`+
` FROM projections.failed_events`), ` FROM projections.failed_events`),
@@ -94,6 +100,7 @@ func Test_FailedEventsPrepares(t *testing.T) {
"projection_name", "projection_name",
"failed_sequence", "failed_sequence",
"failure_count", "failure_count",
"last_failed",
"error", "error",
"count", "count",
}, },
@@ -102,12 +109,14 @@ func Test_FailedEventsPrepares(t *testing.T) {
"projection-name", "projection-name",
uint64(20211108), uint64(20211108),
2, 2,
testNow,
"error", "error",
}, },
{ {
"projection-name-2", "projection-name-2",
uint64(20211108), uint64(20211108),
2, 2,
nil,
"error", "error",
}, },
}, },
@@ -122,6 +131,7 @@ func Test_FailedEventsPrepares(t *testing.T) {
ProjectionName: "projection-name", ProjectionName: "projection-name",
FailedSequence: 20211108, FailedSequence: 20211108,
FailureCount: 2, FailureCount: 2,
LastFailed: testNow,
Error: "error", Error: "error",
}, },
{ {
@@ -141,6 +151,7 @@ func Test_FailedEventsPrepares(t *testing.T) {
regexp.QuoteMeta(`SELECT projections.failed_events.projection_name,`+ regexp.QuoteMeta(`SELECT projections.failed_events.projection_name,`+
` projections.failed_events.failed_sequence,`+ ` projections.failed_events.failed_sequence,`+
` projections.failed_events.failure_count,`+ ` projections.failed_events.failure_count,`+
` projections.failed_events.last_failed,`+
` projections.failed_events.error,`+ ` projections.failed_events.error,`+
` COUNT(*) OVER ()`+ ` COUNT(*) OVER ()`+
` FROM projections.failed_events`), ` FROM projections.failed_events`),

View File

@@ -1,9 +1,13 @@
package model package model
import "time"
type FailedEvent struct { type FailedEvent struct {
Database string Database string
ViewName string ViewName string
FailedSequence uint64 FailedSequence uint64
FailureCount uint64 FailureCount uint64
ErrMsg string ErrMsg string
InstanceID string
LastFailed time.Time
} }

View File

@@ -2,6 +2,7 @@ package repository
import ( import (
"strings" "strings"
"time"
"github.com/jinzhu/gorm" "github.com/jinzhu/gorm"
@@ -11,11 +12,47 @@ import (
) )
type FailedEvent struct { type FailedEvent struct {
ViewName string `gorm:"column:view_name;primary_key"` ViewName string `gorm:"column:view_name;primary_key"`
FailedSequence uint64 `gorm:"column:failed_sequence;primary_key"` FailedSequence uint64 `gorm:"column:failed_sequence;primary_key"`
FailureCount uint64 `gorm:"column:failure_count"` FailureCount uint64 `gorm:"column:failure_count"`
ErrMsg string `gorm:"column:err_msg"` ErrMsg string `gorm:"column:err_msg"`
InstanceID string `gorm:"column:instance_id"` InstanceID string `gorm:"column:instance_id"`
LastFailed time.Time `gorm:"column:last_failed"`
}
type failedEventSearchRequest struct {
Offset uint64
Limit uint64
SortingColumn failedEventSearchKey
Asc bool
Queries []*FailedEventSearchQuery
}
func (f failedEventSearchRequest) GetLimit() uint64 {
return f.Limit
}
func (f failedEventSearchRequest) GetOffset() uint64 {
return f.Offset
}
func (f failedEventSearchRequest) GetSortingColumn() ColumnKey {
if f.SortingColumn == failedEventSearchKey(FailedEventKeyUndefined) {
return nil
}
return f.SortingColumn
}
func (f failedEventSearchRequest) GetAsc() bool {
return f.Asc
}
func (f failedEventSearchRequest) GetQueries() []SearchQuery {
result := make([]SearchQuery, len(f.Queries))
for i, q := range f.Queries {
result[i] = q
}
return result
} }
type FailedEventSearchQuery struct { type FailedEventSearchQuery struct {
@@ -43,6 +80,7 @@ const (
FailedEventKeyViewName FailedEventKeyViewName
FailedEventKeyFailedSequence FailedEventKeyFailedSequence
FailedEventKeyInstanceID FailedEventKeyInstanceID
FailedEventKeyLastFailed
) )
type failedEventSearchKey FailedEventSearchKey type failedEventSearchKey FailedEventSearchKey
@@ -55,6 +93,8 @@ func (key failedEventSearchKey) ToColumnName() string {
return "failed_sequence" return "failed_sequence"
case FailedEventKeyInstanceID: case FailedEventKeyInstanceID:
return "instance_id" return "instance_id"
case FailedEventKeyLastFailed:
return "last_failed"
default: default:
return "" return ""
} }
@@ -65,6 +105,7 @@ func FailedEventFromModel(failedEvent *view_model.FailedEvent) *FailedEvent {
ViewName: failedEvent.Database + "." + failedEvent.ViewName, ViewName: failedEvent.Database + "." + failedEvent.ViewName,
FailureCount: failedEvent.FailureCount, FailureCount: failedEvent.FailureCount,
FailedSequence: failedEvent.FailedSequence, FailedSequence: failedEvent.FailedSequence,
InstanceID: failedEvent.InstanceID,
ErrMsg: failedEvent.ErrMsg, ErrMsg: failedEvent.ErrMsg,
} }
} }
@@ -76,6 +117,7 @@ func FailedEventToModel(failedEvent *FailedEvent) *view_model.FailedEvent {
FailureCount: failedEvent.FailureCount, FailureCount: failedEvent.FailureCount,
FailedSequence: failedEvent.FailedSequence, FailedSequence: failedEvent.FailedSequence,
ErrMsg: failedEvent.ErrMsg, ErrMsg: failedEvent.ErrMsg,
LastFailed: failedEvent.LastFailed,
} }
} }
@@ -123,9 +165,13 @@ func LatestFailedEvent(db *gorm.DB, table, viewName, instanceID string, sequence
} }
func AllFailedEvents(db *gorm.DB, table string) ([]*FailedEvent, error) { func AllFailedEvents(db *gorm.DB, table, instanceID string) ([]*FailedEvent, error) {
queries := make([]*FailedEventSearchQuery, 0, 1)
if instanceID != "" {
queries = append(queries, &FailedEventSearchQuery{Key: FailedEventKeyInstanceID, Method: domain.SearchMethodEquals, Value: instanceID})
}
failedEvents := make([]*FailedEvent, 0) failedEvents := make([]*FailedEvent, 0)
query := PrepareSearchQuery(table, GeneralSearchRequest{}) query := PrepareSearchQuery(table, &failedEventSearchRequest{SortingColumn: failedEventSearchKey(FailedEventKeyLastFailed), Queries: queries})
_, err := query(db, &failedEvents) _, err := query(db, &failedEvents)
if err != nil { if err != nil {
return nil, err return nil, err

View File

@@ -192,9 +192,13 @@ func LatestSequences(db *gorm.DB, table, viewName string, instanceIDs ...string)
return sequences, nil return sequences, nil
} }
func AllCurrentSequences(db *gorm.DB, table string) ([]*CurrentSequence, error) { func AllCurrentSequences(db *gorm.DB, table, instanceID string) ([]*CurrentSequence, error) {
queries := make([]sequenceSearchQuery, 0, 1)
if instanceID != "" {
queries = append(queries, sequenceSearchQuery{key: sequenceSearchKey(SequenceSearchKeyInstanceID), value: instanceID})
}
sequences := make([]*CurrentSequence, 0) sequences := make([]*CurrentSequence, 0)
query := PrepareSearchQuery(table, GeneralSearchRequest{}) query := PrepareSearchQuery(table, &sequenceSearchRequest{queries: queries})
_, err := query(db, &sequences) _, err := query(db, &sequences)
if err != nil { if err != nil {
return nil, err return nil, err

View File

@@ -4648,12 +4648,12 @@ message View {
google.protobuf.Timestamp event_timestamp = 4 [ google.protobuf.Timestamp event_timestamp = 4 [
(grpc.gateway.protoc_gen_openapiv2.options.openapiv2_field) = { (grpc.gateway.protoc_gen_openapiv2.options.openapiv2_field) = {
example: "\"2019-04-01T08:45:00.000000Z\""; example: "\"2019-04-01T08:45:00.000000Z\"";
description: "The timestamp the event occured"; description: "The timestamp the event occurred";
} }
]; // The timestamp the event occured ]; // The timestamp the event occurred
google.protobuf.Timestamp last_successful_spooler_run = 5 [ google.protobuf.Timestamp last_successful_spooler_run = 5 [
(grpc.gateway.protoc_gen_openapiv2.options.openapiv2_field) = { (grpc.gateway.protoc_gen_openapiv2.options.openapiv2_field) = {
description: "The timestamp the event occured"; description: "The timestamp the event occurred";
} }
]; ];
} }
@@ -4684,6 +4684,11 @@ message FailedEvent {
example: "\"ID=EXAMP-ID3ER Message=Example message\""; example: "\"ID=EXAMP-ID3ER Message=Example message\"";
} }
]; ];
google.protobuf.Timestamp last_failed = 6 [
(grpc.gateway.protoc_gen_openapiv2.options.openapiv2_field) = {
description: "The timestamp the failure last occurred";
}
];
} }
message ImportDataRequest { message ImportDataRequest {

View File

@@ -569,6 +569,11 @@ message RemoveFailedEventRequest {
example: "\"9823758\""; example: "\"9823758\"";
} }
]; ];
string instance_id = 4 [
(grpc.gateway.protoc_gen_openapiv2.options.openapiv2_field) = {
example: "\"840498034930840\"";
}
];
} }
//This is an empty response //This is an empty response
@@ -634,4 +639,9 @@ message FailedEvent {
example: "\"ID=EXAMP-ID3ER Message=Example message\""; example: "\"ID=EXAMP-ID3ER Message=Example message\"";
} }
]; ];
google.protobuf.Timestamp last_failed = 6 [
(grpc.gateway.protoc_gen_openapiv2.options.openapiv2_field) = {
description: "The timestamp the failure last occurred";
}
];
} }