250 lines
8.2 KiB
Go
250 lines
8.2 KiB
Go
package service
|
|
|
|
import (
|
|
"errors"
|
|
"fmt"
|
|
"sort"
|
|
"strings"
|
|
|
|
"browser.local/platform/domain"
|
|
"browser.local/platform/repo"
|
|
"browser.local/platform/validator"
|
|
)
|
|
|
|
const (
|
|
maxMetricSamplesPerServer = 1000
|
|
maxBackupsPerServer = 100
|
|
maxBackupBytesPerServer = int64(4 * 1024 * 1024 * 1024)
|
|
)
|
|
|
|
func (svc *CoreService) IngestMetricBatch(batch domain.MetricBatchIngest) (domain.MetricBatchIngestResult, error) {
|
|
batch = domain.CopyMetricBatchIngest(batch)
|
|
if len(batch.Samples) == 0 || len(batch.Samples) > 256 {
|
|
return domain.MetricBatchIngestResult{}, validationError("metric batch must contain between 1 and 256 samples")
|
|
}
|
|
if err := svc.validateRunSession(batch.RunEndpointID, batch.SessionToken); err != nil {
|
|
return domain.MetricBatchIngestResult{}, err
|
|
}
|
|
latest := svc.now()
|
|
for index, sample := range batch.Samples {
|
|
instance, err := svc.store.ServerInstances().Get(sample.ServerInstanceID)
|
|
if err != nil {
|
|
return domain.MetricBatchIngestResult{}, err
|
|
}
|
|
if instance.RunEndpointID != batch.RunEndpointID {
|
|
return domain.MetricBatchIngestResult{}, validationError("metric sample server must belong to runEndpointId")
|
|
}
|
|
sample.RunEndpointID = batch.RunEndpointID
|
|
if sample.CollectedAt.IsZero() {
|
|
sample.CollectedAt = svc.now()
|
|
}
|
|
if sample.ID == "" {
|
|
sample.ID = fmt.Sprintf("metric:%s:%d:%d", sample.ServerInstanceID, sample.CollectedAt.UnixNano(), index)
|
|
}
|
|
if err := validator.ValidateMetricSample(sample); err != nil {
|
|
return domain.MetricBatchIngestResult{}, err
|
|
}
|
|
if err := svc.store.MetricSamples().Create(sample); err != nil {
|
|
if !errors.Is(err, repo.ErrDuplicate) {
|
|
return domain.MetricBatchIngestResult{}, err
|
|
}
|
|
existing, getErr := svc.store.MetricSamples().Get(sample.ID)
|
|
if getErr != nil || existing.ServerInstanceID != sample.ServerInstanceID || existing.CollectedAt != sample.CollectedAt {
|
|
return domain.MetricBatchIngestResult{}, validationError("metric sample id conflicts with persisted sample")
|
|
}
|
|
}
|
|
if sample.CollectedAt.After(latest) {
|
|
latest = sample.CollectedAt
|
|
}
|
|
if err := svc.pruneMetricSamples(sample.ServerInstanceID); err != nil {
|
|
return domain.MetricBatchIngestResult{}, err
|
|
}
|
|
}
|
|
return domain.MetricBatchIngestResult{Accepted: true, AcceptedCount: len(batch.Samples), LatestAt: latest, ServerTime: svc.now()}, nil
|
|
}
|
|
|
|
func (svc *CoreService) ListMetricSamplesForSession(sessionID string, filter domain.MetricSampleFilter) ([]domain.MetricSample, error) {
|
|
if err := validator.ValidateMetricSampleFilter(filter); err != nil {
|
|
return nil, err
|
|
}
|
|
if strings.TrimSpace(filter.ServerInstanceID) == "" {
|
|
return nil, validationError("serverInstanceId is required")
|
|
}
|
|
instance, err := svc.GetServerInstanceForSession(sessionID, filter.ServerInstanceID)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
items, err := svc.store.MetricSamples().List(filter)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
sort.SliceStable(items, func(i, j int) bool { return items[i].CollectedAt.Before(items[j].CollectedAt) })
|
|
limit := filter.Limit
|
|
if limit == 0 {
|
|
limit = 100
|
|
}
|
|
if len(items) > limit {
|
|
items = items[len(items)-limit:]
|
|
}
|
|
for _, sample := range items {
|
|
if sample.ServerInstanceID != instance.ID || sample.RunEndpointID != instance.RunEndpointID {
|
|
return nil, ErrForbidden
|
|
}
|
|
}
|
|
return domain.CopyMetricSamples(items), nil
|
|
}
|
|
|
|
func (svc *CoreService) CreateBackupForSession(sessionID string, record domain.BackupRecord) (domain.BackupRecord, error) {
|
|
instance, err := svc.GetServerInstanceForSession(sessionID, record.ServerInstanceID)
|
|
if err != nil {
|
|
return domain.BackupRecord{}, err
|
|
}
|
|
artifact, err := svc.store.Artifacts().Get(record.ArtifactID)
|
|
if err != nil {
|
|
return domain.BackupRecord{}, err
|
|
}
|
|
if err := svc.validateBackupArtifactOwner(instance, artifact); err != nil {
|
|
return domain.BackupRecord{}, err
|
|
}
|
|
stamp := svc.now()
|
|
if record.ID == "" {
|
|
record.ID = fmt.Sprintf("backup:%s:%d", instance.ID, stamp.UnixNano())
|
|
}
|
|
if record.State == "" {
|
|
record.State = domain.BackupStatePending
|
|
}
|
|
if record.Checksum == "" {
|
|
record.Checksum = artifact.Checksum
|
|
}
|
|
if record.SizeBytes == 0 {
|
|
record.SizeBytes = artifact.SizeBytes
|
|
}
|
|
if record.CreatedAt.IsZero() {
|
|
record.CreatedAt = stamp
|
|
}
|
|
record.UpdatedAt = stamp
|
|
if record.State == domain.BackupStateAvailable && artifact.State != domain.ArtifactStateAvailable {
|
|
return domain.BackupRecord{}, validationError("backup artifact must be available")
|
|
}
|
|
if err := validator.ValidateBackupRecord(record); err != nil {
|
|
return domain.BackupRecord{}, err
|
|
}
|
|
if err := svc.store.Backups().Create(record); err != nil {
|
|
return domain.BackupRecord{}, err
|
|
}
|
|
if err := svc.pruneBackups(instance.ID); err != nil {
|
|
return domain.BackupRecord{}, err
|
|
}
|
|
return domain.CopyBackupRecord(record), nil
|
|
}
|
|
|
|
func (svc *CoreService) GetBackupForSession(sessionID string, backupID string) (domain.BackupRecord, error) {
|
|
record, err := svc.store.Backups().Get(strings.TrimSpace(backupID))
|
|
if err != nil {
|
|
return domain.BackupRecord{}, err
|
|
}
|
|
if _, err := svc.GetServerInstanceForSession(sessionID, record.ServerInstanceID); err != nil {
|
|
return domain.BackupRecord{}, err
|
|
}
|
|
return domain.CopyBackupRecord(record), nil
|
|
}
|
|
|
|
func (svc *CoreService) ListBackupsForSession(sessionID string, filter domain.BackupFilter) ([]domain.BackupRecord, error) {
|
|
if err := validator.ValidateBackupFilter(filter); err != nil {
|
|
return nil, err
|
|
}
|
|
if strings.TrimSpace(filter.ServerInstanceID) == "" {
|
|
return nil, validationError("serverInstanceId is required")
|
|
}
|
|
if _, err := svc.GetServerInstanceForSession(sessionID, filter.ServerInstanceID); err != nil {
|
|
return nil, err
|
|
}
|
|
items, err := svc.store.Backups().List(filter)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
sort.SliceStable(items, func(i, j int) bool { return items[i].CreatedAt.After(items[j].CreatedAt) })
|
|
if len(items) > validator.MaxBackupRecordsPerQuery {
|
|
items = items[:validator.MaxBackupRecordsPerQuery]
|
|
}
|
|
return domain.CopyBackupRecords(items), nil
|
|
}
|
|
|
|
func (svc *CoreService) RecoverIncompleteBackups() error {
|
|
items, err := svc.store.Backups().List(domain.BackupFilter{State: domain.BackupStatePending})
|
|
if err != nil {
|
|
return err
|
|
}
|
|
for _, record := range items {
|
|
record.State = domain.BackupStateFailed
|
|
record.RecoveryStatus = "recoverable-after-interrupted-transfer"
|
|
record.UpdatedAt = svc.now()
|
|
if err := svc.store.Backups().Update(record); err != nil {
|
|
return err
|
|
}
|
|
}
|
|
return nil
|
|
}
|
|
|
|
func (svc *CoreService) pruneMetricSamples(serverInstanceID string) error {
|
|
items, err := svc.store.MetricSamples().List(domain.MetricSampleFilter{ServerInstanceID: serverInstanceID})
|
|
if err != nil || len(items) <= maxMetricSamplesPerServer {
|
|
return err
|
|
}
|
|
sort.SliceStable(items, func(i, j int) bool { return items[i].CollectedAt.Before(items[j].CollectedAt) })
|
|
for _, sample := range items[:len(items)-maxMetricSamplesPerServer] {
|
|
if err := svc.store.MetricSamples().Delete(sample.ID); err != nil {
|
|
return err
|
|
}
|
|
}
|
|
return nil
|
|
}
|
|
|
|
func (svc *CoreService) pruneBackups(serverInstanceID string) error {
|
|
items, err := svc.store.Backups().List(domain.BackupFilter{ServerInstanceID: serverInstanceID})
|
|
if err != nil {
|
|
return err
|
|
}
|
|
sort.SliceStable(items, func(i, j int) bool { return items[i].CreatedAt.Before(items[j].CreatedAt) })
|
|
total := int64(0)
|
|
for _, record := range items {
|
|
if record.State != domain.BackupStateExpired {
|
|
total += record.SizeBytes
|
|
}
|
|
}
|
|
for len(items) > maxBackupsPerServer || total > maxBackupBytesPerServer {
|
|
record := items[0]
|
|
items = items[1:]
|
|
if record.State != domain.BackupStateExpired {
|
|
total -= record.SizeBytes
|
|
}
|
|
record.State = domain.BackupStateExpired
|
|
record.RecoveryStatus = "retention-expired"
|
|
record.UpdatedAt = svc.now()
|
|
if err := svc.store.Backups().Update(record); err != nil {
|
|
return err
|
|
}
|
|
}
|
|
return nil
|
|
}
|
|
|
|
func (svc *CoreService) validateBackupArtifactOwner(instance domain.ServerInstance, artifact domain.Artifact) error {
|
|
switch artifact.OwnerKind {
|
|
case domain.ArtifactOwnerKindServerInstance:
|
|
if artifact.OwnerID != instance.ID {
|
|
return ErrForbidden
|
|
}
|
|
case domain.ArtifactOwnerKindJob:
|
|
job, err := svc.store.Jobs().Get(artifact.OwnerID)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
if job.ServerInstanceID != instance.ID || job.RunEndpointID != instance.RunEndpointID {
|
|
return ErrForbidden
|
|
}
|
|
default:
|
|
return ErrForbidden
|
|
}
|
|
return nil
|
|
}
|