Anon: Scaffold anon service (#74744)
* remove API tagging method and authed tagging * add anonstore move debug to after cache change test order fix issue where mysql trims to second * add old device cleanup lint utc-ize everything trim whitespace * remove dangling setting * Add delete devices * Move anonymous authnclient to anonimpl * Add simple post login hook * move registration of Background Service cleanup * add updated_at index * do not untag device if login err * add delete device integration test
This commit is contained in:
Jo
@@ -2,144 +2,76 @@ package anonimpl
|
||||
|
||||
import (
|
||||
"context"
|
||||
"encoding/hex"
|
||||
"encoding/json"
|
||||
"fmt"
|
||||
"hash/fnv"
|
||||
"net/http"
|
||||
"strings"
|
||||
"time"
|
||||
|
||||
"github.com/grafana/grafana/pkg/infra/localcache"
|
||||
"github.com/grafana/grafana/pkg/infra/log"
|
||||
"github.com/grafana/grafana/pkg/infra/network"
|
||||
"github.com/grafana/grafana/pkg/infra/remotecache"
|
||||
"github.com/grafana/grafana/pkg/infra/serverlock"
|
||||
"github.com/grafana/grafana/pkg/infra/usagestats"
|
||||
"github.com/grafana/grafana/pkg/services/anonymous"
|
||||
"github.com/grafana/grafana/pkg/services/anonymous/anonimpl/anonstore"
|
||||
"github.com/grafana/grafana/pkg/services/authn"
|
||||
"github.com/grafana/grafana/pkg/services/org"
|
||||
"github.com/grafana/grafana/pkg/setting"
|
||||
"github.com/grafana/grafana/pkg/web"
|
||||
)
|
||||
|
||||
const thirtyDays = 30 * 24 * time.Hour
|
||||
const deviceIDHeader = "X-Grafana-Device-Id"
|
||||
|
||||
type Device struct {
|
||||
Kind anonymous.DeviceKind `json:"kind"`
|
||||
IP string `json:"ip"`
|
||||
UserAgent string `json:"user_agent"`
|
||||
LastSeen time.Time `json:"last_seen"`
|
||||
}
|
||||
|
||||
func (a *Device) Key() (string, error) {
|
||||
key := strings.Builder{}
|
||||
key.WriteString(a.IP)
|
||||
key.WriteString(a.UserAgent)
|
||||
|
||||
hash := fnv.New128a()
|
||||
if _, err := hash.Write([]byte(key.String())); err != nil {
|
||||
return "", fmt.Errorf("failed to write to hash: %w", err)
|
||||
}
|
||||
|
||||
return strings.Join([]string{string(a.Kind), hex.EncodeToString(hash.Sum(nil))}, ":"), nil
|
||||
}
|
||||
|
||||
func (a *Device) UIKey(deviceID string) (string, error) {
|
||||
return strings.Join([]string{string(a.Kind), deviceID}, ":"), nil
|
||||
}
|
||||
const keepFor = time.Hour * 24 * 61
|
||||
|
||||
type AnonDeviceService struct {
|
||||
remoteCache remotecache.CacheStorage
|
||||
log log.Logger
|
||||
localCache *localcache.CacheService
|
||||
log log.Logger
|
||||
localCache *localcache.CacheService
|
||||
anonStore anonstore.AnonStore
|
||||
serverLock *serverlock.ServerLockService
|
||||
}
|
||||
|
||||
func ProvideAnonymousDeviceService(remoteCache remotecache.CacheStorage, usageStats usagestats.Service) *AnonDeviceService {
|
||||
func ProvideAnonymousDeviceService(usageStats usagestats.Service, authBroker authn.Service,
|
||||
anonStore anonstore.AnonStore, cfg *setting.Cfg, orgService org.Service,
|
||||
serverLockService *serverlock.ServerLockService,
|
||||
) *AnonDeviceService {
|
||||
a := &AnonDeviceService{
|
||||
remoteCache: remoteCache,
|
||||
log: log.New("anonymous-session-service"),
|
||||
localCache: localcache.New(29*time.Minute, 15*time.Minute),
|
||||
log: log.New("anonymous-session-service"),
|
||||
localCache: localcache.New(29*time.Minute, 15*time.Minute),
|
||||
anonStore: anonStore,
|
||||
serverLock: serverLockService,
|
||||
}
|
||||
|
||||
usageStats.RegisterMetricsFunc(a.usageStatFn)
|
||||
|
||||
anonClient := &Anonymous{
|
||||
cfg: cfg,
|
||||
log: log.New("authn.anonymous"),
|
||||
orgService: orgService,
|
||||
anonDeviceService: a,
|
||||
}
|
||||
|
||||
if anonClient.cfg.AnonymousEnabled {
|
||||
authBroker.RegisterClient(anonClient)
|
||||
authBroker.RegisterPostLoginHook(a.untagDevice, 100)
|
||||
}
|
||||
|
||||
return a
|
||||
}
|
||||
|
||||
func (a *AnonDeviceService) usageStatFn(ctx context.Context) (map[string]any, error) {
|
||||
anonDeviceCount, err := a.remoteCache.Count(ctx, string(anonymous.AnonDevice))
|
||||
// Count the number of unique devices that have been updated in the last 30 days.
|
||||
// One minute is added to the end time as mysql has a precision of seconds and it will break tests that write too fast.
|
||||
anonUIDeviceCount, err := a.anonStore.CountDevices(ctx, time.Now().Add(-thirtyDays), time.Now().Add(time.Minute))
|
||||
if err != nil {
|
||||
return nil, nil
|
||||
}
|
||||
|
||||
authedDeviceCount, err := a.remoteCache.Count(ctx, string(anonymous.AuthedDevice))
|
||||
if err != nil {
|
||||
return nil, nil
|
||||
}
|
||||
|
||||
anonUIDeviceCount, err := a.remoteCache.Count(ctx, string(anonymous.AnonDeviceUI))
|
||||
if err != nil {
|
||||
return nil, nil
|
||||
}
|
||||
|
||||
authedUIDeviceCount, err := a.remoteCache.Count(ctx, string(anonymous.AuthedDeviceUI))
|
||||
if err != nil {
|
||||
return nil, nil
|
||||
return nil, err
|
||||
}
|
||||
|
||||
return map[string]any{
|
||||
"stats.anonymous.session.count": anonDeviceCount, // keep session for legacy data
|
||||
"stats.users.device.count": authedDeviceCount,
|
||||
"stats.anonymous.device.ui.count": anonUIDeviceCount,
|
||||
"stats.users.device.ui.count": authedUIDeviceCount,
|
||||
}, nil
|
||||
}
|
||||
|
||||
func (a *AnonDeviceService) untagDevice(ctx context.Context, device *Device) error {
|
||||
key, err := device.Key()
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
|
||||
if err := a.remoteCache.Delete(ctx, key); err != nil {
|
||||
return err
|
||||
}
|
||||
|
||||
return nil
|
||||
}
|
||||
|
||||
func (a *AnonDeviceService) untagUIDevice(ctx context.Context, deviceID string, device *Device) error {
|
||||
key, err := device.UIKey(deviceID)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
|
||||
if err := a.remoteCache.Delete(ctx, key); err != nil {
|
||||
return err
|
||||
}
|
||||
|
||||
return nil
|
||||
}
|
||||
|
||||
func (a *AnonDeviceService) tagDeviceUI(ctx context.Context, httpReq *http.Request, device Device) error {
|
||||
deviceID := httpReq.Header.Get(deviceIDHeader)
|
||||
if deviceID == "" {
|
||||
return nil
|
||||
}
|
||||
|
||||
if device.Kind == anonymous.AnonDevice {
|
||||
device.Kind = anonymous.AnonDeviceUI
|
||||
} else if device.Kind == anonymous.AuthedDevice {
|
||||
device.Kind = anonymous.AuthedDeviceUI
|
||||
}
|
||||
|
||||
key, err := device.UIKey(deviceID)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
|
||||
if setting.Env == setting.Dev {
|
||||
a.log.Debug("Tagging device for UI", "deviceID", deviceID, "device", device, "key", key)
|
||||
}
|
||||
func (a *AnonDeviceService) tagDeviceUI(ctx context.Context, httpReq *http.Request, device *anonstore.Device) error {
|
||||
key := device.CacheKey()
|
||||
|
||||
if _, ok := a.localCache.Get(key); ok {
|
||||
return nil
|
||||
@@ -147,33 +79,41 @@ func (a *AnonDeviceService) tagDeviceUI(ctx context.Context, httpReq *http.Reque
|
||||
|
||||
a.localCache.SetDefault(key, struct{}{})
|
||||
|
||||
deviceJSON, err := json.Marshal(device)
|
||||
if err != nil {
|
||||
return err
|
||||
if setting.Env == setting.Dev {
|
||||
a.log.Debug("Tagging device for UI", "deviceID", device.DeviceID, "device", device, "key", key)
|
||||
}
|
||||
|
||||
if err := a.remoteCache.Set(ctx, key, deviceJSON, thirtyDays); err != nil {
|
||||
return err
|
||||
}
|
||||
|
||||
// remove existing tag when device switches to another kind
|
||||
untagKind := anonymous.AnonDeviceUI
|
||||
if device.Kind == anonymous.AnonDeviceUI {
|
||||
untagKind = anonymous.AuthedDeviceUI
|
||||
}
|
||||
|
||||
if err := a.untagUIDevice(ctx, deviceID, &Device{
|
||||
Kind: untagKind,
|
||||
IP: device.IP,
|
||||
UserAgent: device.UserAgent,
|
||||
}); err != nil {
|
||||
if err := a.anonStore.CreateOrUpdateDevice(ctx, device); err != nil {
|
||||
return err
|
||||
}
|
||||
|
||||
return nil
|
||||
}
|
||||
|
||||
func (a *AnonDeviceService) untagDevice(ctx context.Context,
|
||||
identity *authn.Identity, r *authn.Request, err error) {
|
||||
if err != nil {
|
||||
return
|
||||
}
|
||||
|
||||
deviceID := r.HTTPRequest.Header.Get(deviceIDHeader)
|
||||
if deviceID == "" {
|
||||
return
|
||||
}
|
||||
|
||||
errD := a.anonStore.DeleteDevice(ctx, deviceID)
|
||||
if errD != nil {
|
||||
a.log.Debug("Failed to untag device", "error", err)
|
||||
}
|
||||
}
|
||||
|
||||
// FIXME: Unexport and remove interface
|
||||
func (a *AnonDeviceService) TagDevice(ctx context.Context, httpReq *http.Request, kind anonymous.DeviceKind) error {
|
||||
deviceID := httpReq.Header.Get(deviceIDHeader)
|
||||
if deviceID == "" {
|
||||
return nil
|
||||
}
|
||||
|
||||
addr := web.RemoteAddr(httpReq)
|
||||
ip, err := network.GetIPFromAddress(addr)
|
||||
if err != nil {
|
||||
@@ -186,54 +126,39 @@ func (a *AnonDeviceService) TagDevice(ctx context.Context, httpReq *http.Request
|
||||
clientIPStr = ""
|
||||
}
|
||||
|
||||
taggedDevice := &Device{
|
||||
Kind: kind,
|
||||
IP: clientIPStr,
|
||||
taggedDevice := &anonstore.Device{
|
||||
DeviceID: deviceID,
|
||||
ClientIP: clientIPStr,
|
||||
UserAgent: httpReq.UserAgent(),
|
||||
LastSeen: time.Now().UTC(),
|
||||
CreatedAt: time.Now(),
|
||||
UpdatedAt: time.Now(),
|
||||
}
|
||||
|
||||
err = a.tagDeviceUI(ctx, httpReq, *taggedDevice)
|
||||
err = a.tagDeviceUI(ctx, httpReq, taggedDevice)
|
||||
if err != nil {
|
||||
a.log.Debug("Failed to tag device for UI", "error", err)
|
||||
}
|
||||
|
||||
key, err := taggedDevice.Key()
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
|
||||
if setting.Env == setting.Dev {
|
||||
a.log.Debug("Tagging device", "device", taggedDevice, "key", key)
|
||||
}
|
||||
|
||||
if _, ok := a.localCache.Get(key); ok {
|
||||
return nil
|
||||
}
|
||||
|
||||
a.localCache.SetDefault(key, struct{}{})
|
||||
|
||||
deviceJSON, err := json.Marshal(taggedDevice)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
|
||||
if err := a.remoteCache.Set(ctx, key, deviceJSON, thirtyDays); err != nil {
|
||||
return err
|
||||
}
|
||||
|
||||
// remove existing tag when device switches to another kind
|
||||
untagKind := anonymous.AnonDevice
|
||||
if kind == anonymous.AnonDevice {
|
||||
untagKind = anonymous.AuthedDevice
|
||||
}
|
||||
if err := a.untagDevice(ctx, &Device{
|
||||
Kind: untagKind,
|
||||
IP: taggedDevice.IP,
|
||||
UserAgent: taggedDevice.UserAgent,
|
||||
}); err != nil {
|
||||
return err
|
||||
}
|
||||
|
||||
return nil
|
||||
}
|
||||
|
||||
func (a *AnonDeviceService) Run(ctx context.Context) error {
|
||||
ticker := time.NewTicker(2 * time.Hour)
|
||||
|
||||
for {
|
||||
select {
|
||||
case <-ticker.C:
|
||||
err := a.serverLock.LockAndExecute(ctx, "cleanup old anon devices", time.Hour*10, func(context.Context) {
|
||||
if err := a.anonStore.DeleteDevicesOlderThan(ctx, time.Now().Add(-keepFor)); err != nil {
|
||||
a.log.Error("An error occurred while deleting old anon devices", "err", err)
|
||||
}
|
||||
})
|
||||
if err != nil {
|
||||
a.log.Error("Failed to lock and execute cleanup old anon devices", "error", err)
|
||||
}
|
||||
|
||||
case <-ctx.Done():
|
||||
return ctx.Err()
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
Reference in New Issue
Block a user