grafana/pkg/services/store/entity_events.go

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{}