Search: in-memory index (#47709)
* #45498: add entity events table
* #45498: add entity events service
* #45498: hook up entity events service to http server
* #45498: use `dashboards.id` rather than `uid` and `org_id` in grn
* Update pkg/services/entityevents/service.go
Co-authored-by: Ryan McKinley <ryantxu@gmail.com>
* #45498: move entityeventsservice to services/store
* #45498: add null check
* #45498: rename
* #45498: fix comment
* #45498: switch grn back to uid
* Search: listen for updates (#47719)
* #45498: wire entity event service with searchv2
* load last event id before building index for org 1
* fix service init in integration tests
* depend on required subset of event store methods
* Update pkg/services/sqlstore/migrations/entity_events_mig.go
Co-authored-by: Alexander Emelin <frvzmb@gmail.com>
* #45498: pointer receiver
* #45498: mockery!
* #45498: add entity events service to background services
* dashboard query pagination, allow queries while re-indexing
* log level cleanups, use rlock, add comments
* fix lint, check feature toggle in search v2 service
* use unix time for event created column
* add missing changes for created column
* fix integration tests init
* log re-index execution times on info level
* #45498: fix entityEventsService tests
* #45498: save events on dashboard delete
* use camel case for log labels
* formatting
* #45498: rename grn to entityid
* #45498: add `IsDisabled` to entityEventsService
* #45498: remove feature flag from migration
* better context usage, fix capacity, comments/cleanups
* replace print with logger
* Revert "#45498: remove feature flag from migration"
This reverts commit ed23968898.
* revert:revert:revert conditional feature flag
Co-authored-by: Ryan McKinley <ryantxu@gmail.com>
Co-authored-by: Alexander Emelin <frvzmb@gmail.com>
This commit is contained in:
Artur Wierzbicki
co-authored by
Ryan McKinley
Alexander Emelin
Ryan McKinley
Alexander Emelin
parent
8764040fe3
commit
25e153e4e7
@@ -0,0 +1,179 @@
|
||||
package store
|
||||
|
||||
import (
|
||||
"context"
|
||||
"fmt"
|
||||
"time"
|
||||
|
||||
"github.com/grafana/grafana/pkg/infra/log"
|
||||
"github.com/grafana/grafana/pkg/registry"
|
||||
"github.com/grafana/grafana/pkg/services/featuremgmt"
|
||||
"github.com/grafana/grafana/pkg/services/sqlstore"
|
||||
"github.com/grafana/grafana/pkg/setting"
|
||||
)
|
||||
|
||||
type EntityEventType string
|
||||
|
||||
const (
|
||||
EntityEventTypeDelete EntityEventType = "delete"
|
||||
EntityEventTypeCreate EntityEventType = "create"
|
||||
EntityEventTypeUpdate EntityEventType = "update"
|
||||
)
|
||||
|
||||
type EntityType string
|
||||
|
||||
const (
|
||||
EntityTypeDashboard EntityType = "dashboard"
|
||||
)
|
||||
|
||||
// CreateDatabaseEntityId creates entityId for entities stored in the existing SQL tables
|
||||
func CreateDatabaseEntityId(internalId interface{}, orgId int64, entityType EntityType) string {
|
||||
var internalIdAsString string
|
||||
switch id := internalId.(type) {
|
||||
case string:
|
||||
internalIdAsString = id
|
||||
default:
|
||||
internalIdAsString = fmt.Sprintf("%#v", internalId)
|
||||
}
|
||||
|
||||
return fmt.Sprintf("database/%d/%s/%s", orgId, entityType, internalIdAsString)
|
||||
}
|
||||
|
||||
type EntityEvent struct {
|
||||
Id int64
|
||||
EventType EntityEventType
|
||||
EntityId string
|
||||
Created int64
|
||||
}
|
||||
|
||||
type SaveEventCmd struct {
|
||||
EntityId string
|
||||
EventType EntityEventType
|
||||
}
|
||||
|
||||
// EntityEventsService is a temporary solution to support change notifications in an HA setup
|
||||
// With this service each system can query for any events that have happened since a fixed time
|
||||
//go:generate mockery --name EntityEventsService --structname MockEntityEventsService --inpackage --filename entity_events_mock.go
|
||||
type EntityEventsService interface {
|
||||
registry.BackgroundService
|
||||
registry.CanBeDisabled
|
||||
SaveEvent(ctx context.Context, cmd SaveEventCmd) error
|
||||
GetLastEvent(ctx context.Context) (*EntityEvent, error)
|
||||
GetAllEventsAfter(ctx context.Context, id int64) ([]*EntityEvent, error)
|
||||
|
||||
deleteEventsOlderThan(ctx context.Context, duration time.Duration) error
|
||||
}
|
||||
|
||||
func ProvideEntityEventsService(cfg *setting.Cfg, sqlStore *sqlstore.SQLStore, features featuremgmt.FeatureToggles) EntityEventsService {
|
||||
if !features.IsEnabled(featuremgmt.FlagPanelTitleSearch) {
|
||||
return &dummyEntityEventsService{}
|
||||
}
|
||||
|
||||
return &entityEventService{
|
||||
sql: sqlStore,
|
||||
features: features,
|
||||
log: log.New("entity-events"),
|
||||
}
|
||||
}
|
||||
|
||||
type entityEventService struct {
|
||||
sql *sqlstore.SQLStore
|
||||
log log.Logger
|
||||
features featuremgmt.FeatureToggles
|
||||
}
|
||||
|
||||
func (e *entityEventService) SaveEvent(ctx context.Context, cmd SaveEventCmd) error {
|
||||
return e.sql.WithDbSession(ctx, func(sess *sqlstore.DBSession) error {
|
||||
_, err := sess.Insert(&EntityEvent{
|
||||
EventType: cmd.EventType,
|
||||
EntityId: cmd.EntityId,
|
||||
Created: time.Now().Unix(),
|
||||
})
|
||||
return err
|
||||
})
|
||||
}
|
||||
|
||||
func (e *entityEventService) GetLastEvent(ctx context.Context) (*EntityEvent, error) {
|
||||
var entityEvent *EntityEvent
|
||||
err := e.sql.WithDbSession(ctx, func(sess *sqlstore.DBSession) error {
|
||||
bean := &EntityEvent{}
|
||||
found, err := sess.OrderBy("id desc").Get(bean)
|
||||
if found {
|
||||
entityEvent = bean
|
||||
}
|
||||
return err
|
||||
})
|
||||
|
||||
return entityEvent, err
|
||||
}
|
||||
|
||||
func (e *entityEventService) GetAllEventsAfter(ctx context.Context, id int64) ([]*EntityEvent, error) {
|
||||
var evs = make([]*EntityEvent, 0)
|
||||
err := e.sql.WithDbSession(ctx, func(sess *sqlstore.DBSession) error {
|
||||
return sess.OrderBy("id asc").Where("id > ?", id).Find(&evs)
|
||||
})
|
||||
|
||||
return evs, err
|
||||
}
|
||||
|
||||
func (e *entityEventService) deleteEventsOlderThan(ctx context.Context, duration time.Duration) error {
|
||||
return e.sql.WithDbSession(ctx, func(sess *sqlstore.DBSession) error {
|
||||
maxCreated := time.Now().Add(-duration)
|
||||
deletedCount, err := sess.Where("created < ?", maxCreated.Unix()).Delete(&EntityEvent{})
|
||||
e.log.Info("deleting old events", "count", deletedCount, "maxCreated", maxCreated)
|
||||
return err
|
||||
})
|
||||
}
|
||||
|
||||
func (e *entityEventService) IsDisabled() bool {
|
||||
return false
|
||||
}
|
||||
|
||||
func (e *entityEventService) Run(ctx context.Context) error {
|
||||
clean := time.NewTicker(1 * time.Hour)
|
||||
|
||||
for {
|
||||
select {
|
||||
case <-clean.C:
|
||||
go func() {
|
||||
err := e.deleteEventsOlderThan(context.Background(), 24*time.Hour)
|
||||
if err != nil {
|
||||
e.log.Info("failed to delete old entity events", "error", err)
|
||||
}
|
||||
}()
|
||||
case <-ctx.Done():
|
||||
e.log.Debug("Grafana is shutting down - stopping entity events service")
|
||||
clean.Stop()
|
||||
return nil
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
type dummyEntityEventsService struct {
|
||||
}
|
||||
|
||||
func (d dummyEntityEventsService) Run(ctx context.Context) error {
|
||||
return nil
|
||||
}
|
||||
|
||||
func (d dummyEntityEventsService) IsDisabled() bool {
|
||||
return false
|
||||
}
|
||||
|
||||
func (d dummyEntityEventsService) SaveEvent(ctx context.Context, cmd SaveEventCmd) error {
|
||||
return nil
|
||||
}
|
||||
|
||||
func (d dummyEntityEventsService) GetLastEvent(ctx context.Context) (*EntityEvent, error) {
|
||||
return nil, nil
|
||||
}
|
||||
|
||||
func (d dummyEntityEventsService) GetAllEventsAfter(ctx context.Context, id int64) ([]*EntityEvent, error) {
|
||||
return make([]*EntityEvent, 0), nil
|
||||
}
|
||||
|
||||
func (d dummyEntityEventsService) deleteEventsOlderThan(ctx context.Context, duration time.Duration) error {
|
||||
return nil
|
||||
}
|
||||
|
||||
var _ EntityEventsService = &dummyEntityEventsService{}
|
||||
@@ -0,0 +1,117 @@
|
||||
// Code generated by mockery v2.10.6. DO NOT EDIT.
|
||||
|
||||
package store
|
||||
|
||||
import (
|
||||
context "context"
|
||||
time "time"
|
||||
|
||||
mock "github.com/stretchr/testify/mock"
|
||||
)
|
||||
|
||||
// MockEntityEventsService is an autogenerated mock type for the EntityEventsService type
|
||||
type MockEntityEventsService struct {
|
||||
mock.Mock
|
||||
}
|
||||
|
||||
// GetAllEventsAfter provides a mock function with given fields: ctx, id
|
||||
func (_m *MockEntityEventsService) GetAllEventsAfter(ctx context.Context, id int64) ([]*EntityEvent, error) {
|
||||
ret := _m.Called(ctx, id)
|
||||
|
||||
var r0 []*EntityEvent
|
||||
if rf, ok := ret.Get(0).(func(context.Context, int64) []*EntityEvent); ok {
|
||||
r0 = rf(ctx, id)
|
||||
} else {
|
||||
if ret.Get(0) != nil {
|
||||
r0 = ret.Get(0).([]*EntityEvent)
|
||||
}
|
||||
}
|
||||
|
||||
var r1 error
|
||||
if rf, ok := ret.Get(1).(func(context.Context, int64) error); ok {
|
||||
r1 = rf(ctx, id)
|
||||
} else {
|
||||
r1 = ret.Error(1)
|
||||
}
|
||||
|
||||
return r0, r1
|
||||
}
|
||||
|
||||
// GetLastEvent provides a mock function with given fields: ctx
|
||||
func (_m *MockEntityEventsService) GetLastEvent(ctx context.Context) (*EntityEvent, error) {
|
||||
ret := _m.Called(ctx)
|
||||
|
||||
var r0 *EntityEvent
|
||||
if rf, ok := ret.Get(0).(func(context.Context) *EntityEvent); ok {
|
||||
r0 = rf(ctx)
|
||||
} else {
|
||||
if ret.Get(0) != nil {
|
||||
r0 = ret.Get(0).(*EntityEvent)
|
||||
}
|
||||
}
|
||||
|
||||
var r1 error
|
||||
if rf, ok := ret.Get(1).(func(context.Context) error); ok {
|
||||
r1 = rf(ctx)
|
||||
} else {
|
||||
r1 = ret.Error(1)
|
||||
}
|
||||
|
||||
return r0, r1
|
||||
}
|
||||
|
||||
// IsDisabled provides a mock function with given fields:
|
||||
func (_m *MockEntityEventsService) IsDisabled() bool {
|
||||
ret := _m.Called()
|
||||
|
||||
var r0 bool
|
||||
if rf, ok := ret.Get(0).(func() bool); ok {
|
||||
r0 = rf()
|
||||
} else {
|
||||
r0 = ret.Get(0).(bool)
|
||||
}
|
||||
|
||||
return r0
|
||||
}
|
||||
|
||||
// Run provides a mock function with given fields: ctx
|
||||
func (_m *MockEntityEventsService) Run(ctx context.Context) error {
|
||||
ret := _m.Called(ctx)
|
||||
|
||||
var r0 error
|
||||
if rf, ok := ret.Get(0).(func(context.Context) error); ok {
|
||||
r0 = rf(ctx)
|
||||
} else {
|
||||
r0 = ret.Error(0)
|
||||
}
|
||||
|
||||
return r0
|
||||
}
|
||||
|
||||
// SaveEvent provides a mock function with given fields: ctx, cmd
|
||||
func (_m *MockEntityEventsService) SaveEvent(ctx context.Context, cmd SaveEventCmd) error {
|
||||
ret := _m.Called(ctx, cmd)
|
||||
|
||||
var r0 error
|
||||
if rf, ok := ret.Get(0).(func(context.Context, SaveEventCmd) error); ok {
|
||||
r0 = rf(ctx, cmd)
|
||||
} else {
|
||||
r0 = ret.Error(0)
|
||||
}
|
||||
|
||||
return r0
|
||||
}
|
||||
|
||||
// deleteEventsOlderThan provides a mock function with given fields: ctx, duration
|
||||
func (_m *MockEntityEventsService) deleteEventsOlderThan(ctx context.Context, duration time.Duration) error {
|
||||
ret := _m.Called(ctx, duration)
|
||||
|
||||
var r0 error
|
||||
if rf, ok := ret.Get(0).(func(context.Context, time.Duration) error); ok {
|
||||
r0 = rf(ctx, duration)
|
||||
} else {
|
||||
r0 = ret.Error(0)
|
||||
}
|
||||
|
||||
return r0
|
||||
}
|
||||
@@ -0,0 +1,182 @@
|
||||
//go:build integration
|
||||
// +build integration
|
||||
|
||||
package store
|
||||
|
||||
import (
|
||||
"context"
|
||||
"testing"
|
||||
"time"
|
||||
|
||||
"github.com/grafana/grafana/pkg/infra/log"
|
||||
"github.com/grafana/grafana/pkg/services/sqlstore"
|
||||
"github.com/stretchr/testify/require"
|
||||
)
|
||||
|
||||
func TestEntityEventsService(t *testing.T) {
|
||||
var ctx context.Context
|
||||
var service EntityEventsService
|
||||
|
||||
setup := func() {
|
||||
service = &entityEventService{
|
||||
sql: sqlstore.InitTestDB(t),
|
||||
log: log.New("entity-event-test"),
|
||||
}
|
||||
ctx = context.Background()
|
||||
}
|
||||
|
||||
t.Run("Should insert an entity event", func(t *testing.T) {
|
||||
setup()
|
||||
|
||||
err := service.SaveEvent(ctx, SaveEventCmd{
|
||||
EntityId: "database/dash/1",
|
||||
EventType: EntityEventTypeCreate,
|
||||
})
|
||||
require.NoError(t, err)
|
||||
})
|
||||
|
||||
t.Run("Should retrieve nil entity if database is empty", func(t *testing.T) {
|
||||
setup()
|
||||
|
||||
ev, err := service.GetLastEvent(ctx)
|
||||
require.NoError(t, err)
|
||||
require.Nil(t, ev)
|
||||
})
|
||||
|
||||
t.Run("Should retrieve last entity event", func(t *testing.T) {
|
||||
setup()
|
||||
lastEventEntityId := "database/dash/1"
|
||||
|
||||
err := service.SaveEvent(ctx, SaveEventCmd{
|
||||
EntityId: "database/dash/3",
|
||||
EventType: EntityEventTypeCreate,
|
||||
})
|
||||
require.NoError(t, err)
|
||||
err = service.SaveEvent(ctx, SaveEventCmd{
|
||||
EntityId: "database/dash/2",
|
||||
EventType: EntityEventTypeCreate,
|
||||
})
|
||||
require.NoError(t, err)
|
||||
err = service.SaveEvent(ctx, SaveEventCmd{
|
||||
EntityId: lastEventEntityId,
|
||||
EventType: EntityEventTypeCreate,
|
||||
})
|
||||
require.NoError(t, err)
|
||||
|
||||
lastEv, err := service.GetLastEvent(ctx)
|
||||
require.NoError(t, err)
|
||||
require.Equal(t, lastEventEntityId, lastEv.EntityId)
|
||||
})
|
||||
|
||||
t.Run("Should retrieve sorted events after an id", func(t *testing.T) {
|
||||
setup()
|
||||
lastEventEntityId := "database/dash/1"
|
||||
|
||||
err := service.SaveEvent(ctx, SaveEventCmd{
|
||||
EntityId: "database/dash/3",
|
||||
EventType: EntityEventTypeCreate,
|
||||
})
|
||||
require.NoError(t, err)
|
||||
firstEv, err := service.GetLastEvent(ctx)
|
||||
firstEvId := firstEv.Id
|
||||
|
||||
err = service.SaveEvent(ctx, SaveEventCmd{
|
||||
EntityId: "database/dash/2",
|
||||
EventType: EntityEventTypeCreate,
|
||||
})
|
||||
require.NoError(t, err)
|
||||
err = service.SaveEvent(ctx, SaveEventCmd{
|
||||
EntityId: lastEventEntityId,
|
||||
EventType: EntityEventTypeCreate,
|
||||
})
|
||||
require.NoError(t, err)
|
||||
|
||||
evs, err := service.GetAllEventsAfter(ctx, firstEvId)
|
||||
require.NoError(t, err)
|
||||
require.Len(t, evs, 2)
|
||||
require.Equal(t, evs[0].EntityId, "database/dash/2")
|
||||
require.Equal(t, evs[1].EntityId, lastEventEntityId)
|
||||
})
|
||||
|
||||
t.Run("Should delete old events", func(t *testing.T) {
|
||||
setup()
|
||||
_ = service.SaveEvent(ctx, SaveEventCmd{
|
||||
EntityId: "database/dash/3",
|
||||
EventType: EntityEventTypeCreate,
|
||||
})
|
||||
_ = service.SaveEvent(ctx, SaveEventCmd{
|
||||
EntityId: "database/dash/2",
|
||||
EventType: EntityEventTypeCreate,
|
||||
})
|
||||
_ = service.SaveEvent(ctx, SaveEventCmd{
|
||||
EntityId: "database/dash/1",
|
||||
EventType: EntityEventTypeCreate,
|
||||
})
|
||||
|
||||
evs, err := service.GetAllEventsAfter(ctx, 0)
|
||||
require.NoError(t, err)
|
||||
require.Len(t, evs, 3)
|
||||
|
||||
err = service.deleteEventsOlderThan(ctx, 24*time.Hour)
|
||||
require.NoError(t, err)
|
||||
|
||||
// did not delete any events
|
||||
evs, err = service.GetAllEventsAfter(ctx, 0)
|
||||
require.NoError(t, err)
|
||||
require.Len(t, evs, 3)
|
||||
|
||||
time.Sleep(2 * time.Second)
|
||||
err = service.deleteEventsOlderThan(ctx, 1*time.Second)
|
||||
require.NoError(t, err)
|
||||
|
||||
// deleted all events
|
||||
evs, err = service.GetAllEventsAfter(ctx, 0)
|
||||
require.NoError(t, err)
|
||||
require.Len(t, evs, 0)
|
||||
})
|
||||
}
|
||||
|
||||
func TestCreateDatabaseEntityId(t *testing.T) {
|
||||
tests := []struct {
|
||||
name string
|
||||
entityType EntityType
|
||||
orgId int64
|
||||
internalId interface{}
|
||||
expected string
|
||||
}{
|
||||
{
|
||||
name: "int64 internal id",
|
||||
entityType: EntityTypeDashboard,
|
||||
orgId: 10,
|
||||
internalId: int64(45),
|
||||
expected: "database/10/dashboard/45",
|
||||
},
|
||||
{
|
||||
name: "big-ish int64 internal id",
|
||||
entityType: EntityTypeDashboard,
|
||||
orgId: 10,
|
||||
internalId: int64(12412421),
|
||||
expected: "database/10/dashboard/12412421",
|
||||
},
|
||||
{
|
||||
name: "int internal id",
|
||||
entityType: EntityTypeDashboard,
|
||||
orgId: 10,
|
||||
internalId: int(1244),
|
||||
expected: "database/10/dashboard/1244",
|
||||
},
|
||||
{
|
||||
name: "string internal id",
|
||||
entityType: EntityTypeDashboard,
|
||||
orgId: 10,
|
||||
internalId: "string-internal-id",
|
||||
expected: "database/10/dashboard/string-internal-id",
|
||||
},
|
||||
}
|
||||
|
||||
for _, tt := range tests {
|
||||
t.Run(tt.name, func(t *testing.T) {
|
||||
require.Equal(t, tt.expected, CreateDatabaseEntityId(tt.internalId, tt.orgId, tt.entityType))
|
||||
})
|
||||
}
|
||||
}
|
||||
Reference in New Issue
Block a user