Remove pre-1.0 audit and protected request scaffolding
This commit is contained in:
npc0-hue
@@ -14,297 +14,9 @@ import (
|
||||
)
|
||||
|
||||
const (
|
||||
capacityHeartbeatStaleAfter = 2 * time.Minute
|
||||
capacityRetryAfterSeconds = 30
|
||||
capacityLogBacklogLimit = 256
|
||||
capacityArtifactBacklogLimit = 128
|
||||
aiConfigDiffTTL = 30 * time.Minute
|
||||
aiConfigDiffTTL = 30 * time.Minute
|
||||
)
|
||||
|
||||
func (svc *CoreService) GetProductionCapacityForSession(sessionID string) (domain.ProductionCapacitySummary, error) {
|
||||
user, err := svc.GetCurrentUser(sessionID)
|
||||
if err != nil {
|
||||
return domain.ProductionCapacitySummary{}, err
|
||||
}
|
||||
endpoints, err := svc.store.RunEndpoints().List(domain.RunEndpointFilter{})
|
||||
if err != nil {
|
||||
return domain.ProductionCapacitySummary{}, err
|
||||
}
|
||||
visibleEndpointIDs, err := svc.visibleEndpointIDs(user)
|
||||
if err != nil {
|
||||
return domain.ProductionCapacitySummary{}, err
|
||||
}
|
||||
alerts, err := svc.store.Alerts().List(domain.AlertFilter{})
|
||||
if err != nil {
|
||||
return domain.ProductionCapacitySummary{}, err
|
||||
}
|
||||
|
||||
summary := domain.ProductionCapacitySummary{GeneratedAt: svc.now()}
|
||||
for _, endpoint := range endpoints {
|
||||
if !isPlatformAdmin(user) {
|
||||
if _, visible := visibleEndpointIDs[endpoint.ID]; !visible {
|
||||
continue
|
||||
}
|
||||
}
|
||||
running, queued, err := svc.capacityJobCounts(endpoint.ID)
|
||||
if err != nil {
|
||||
return domain.ProductionCapacitySummary{}, err
|
||||
}
|
||||
projection := svc.capacityProjection(endpoint, running, queued)
|
||||
for _, alert := range alerts {
|
||||
if alert.SourceKind == "run-endpoint" && alert.SourceID == endpoint.ID && alert.RuleKey == "capacity.pressure" && alert.State != domain.AlertStateResolved {
|
||||
projection.LastAdmissionDecision = domain.CapacityAdmissionDeferred
|
||||
projection.LastAdmissionReason = alert.Message
|
||||
projection.LastAdmissionCheckedAt = alert.LastSeenAt
|
||||
}
|
||||
}
|
||||
summary.Endpoints = append(summary.Endpoints, projection)
|
||||
summary.TotalMaxJobs += projection.MaxJobs
|
||||
summary.TotalRunningJobs += projection.RunningJobs
|
||||
summary.TotalQueuedJobs += projection.QueuedJobs
|
||||
}
|
||||
for _, alert := range alerts {
|
||||
if alert.State != domain.AlertStateResolved && svc.canAccessAlert(user, alert) {
|
||||
summary.ActiveAlerts++
|
||||
}
|
||||
}
|
||||
sort.Slice(summary.Endpoints, func(i, j int) bool { return summary.Endpoints[i].RunEndpointID < summary.Endpoints[j].RunEndpointID })
|
||||
return domain.CopyProductionCapacitySummary(summary), nil
|
||||
}
|
||||
|
||||
func (svc *CoreService) CheckCapacityAdmissionForSession(sessionID string, request domain.CapacityAdmissionRequest) (domain.CapacityAdmissionDecision, error) {
|
||||
if err := validator.ValidateCapacityAdmissionRequest(request); err != nil {
|
||||
return domain.CapacityAdmissionDecision{}, err
|
||||
}
|
||||
user, err := svc.GetCurrentUser(sessionID)
|
||||
if err != nil {
|
||||
return domain.CapacityAdmissionDecision{}, err
|
||||
}
|
||||
request, err = svc.authorizeCapacityRequest(user, request)
|
||||
if err != nil {
|
||||
return domain.CapacityAdmissionDecision{}, err
|
||||
}
|
||||
svc.productionMu.Lock()
|
||||
defer svc.productionMu.Unlock()
|
||||
return svc.checkCapacityAdmission(user.ID, request)
|
||||
}
|
||||
|
||||
func (svc *CoreService) checkCapacityAdmission(actorID string, request domain.CapacityAdmissionRequest) (domain.CapacityAdmissionDecision, error) {
|
||||
endpoint, err := svc.store.RunEndpoints().Get(request.RunEndpointID)
|
||||
if err != nil {
|
||||
return domain.CapacityAdmissionDecision{}, err
|
||||
}
|
||||
running, queued, err := svc.capacityJobCounts(endpoint.ID)
|
||||
if err != nil {
|
||||
return domain.CapacityAdmissionDecision{}, err
|
||||
}
|
||||
projection := svc.capacityProjection(endpoint, running, queued)
|
||||
decision := domain.CapacityAdmissionDecision{
|
||||
Accepted: true, State: domain.CapacityAdmissionAccepted, Reason: "capacity available",
|
||||
ServerInstanceID: request.ServerInstanceID, RunEndpointID: endpoint.ID, Capability: request.Capability,
|
||||
TargetKey: request.TargetKey, MaxJobs: projection.MaxJobs, RunningJobs: projection.RunningJobs,
|
||||
QueuedJobs: projection.QueuedJobs, CheckedAt: svc.now(),
|
||||
}
|
||||
pressure := append([]domain.CapacityPressureCode(nil), projection.PressureCodes...)
|
||||
if len(validator.MissingCapabilities(endpoint.Capabilities, []string{request.Capability})) > 0 {
|
||||
pressure = appendCapacityPressure(pressure, domain.CapacityPressureCapabilityGap)
|
||||
}
|
||||
decision.PressureCodes = pressure
|
||||
|
||||
hardDenied := containsCapacityPressure(pressure, domain.CapacityPressureEndpointOffline) || containsCapacityPressure(pressure, domain.CapacityPressureCapabilityGap)
|
||||
if hardDenied {
|
||||
decision.Accepted = false
|
||||
decision.State = domain.CapacityAdmissionDenied
|
||||
decision.Reason = "endpoint is unavailable or missing the required capability"
|
||||
} else if len(pressure) > 0 {
|
||||
decision.Accepted = false
|
||||
decision.State = domain.CapacityAdmissionDeferred
|
||||
decision.Reason = "endpoint capacity is temporarily under pressure"
|
||||
decision.RetryAfterSeconds = capacityRetryAfterSeconds
|
||||
}
|
||||
|
||||
auditResult := domain.AuditResultSuccess
|
||||
auditAction := "capacity.admission.accepted"
|
||||
if !decision.Accepted {
|
||||
auditResult = domain.AuditResultDenied
|
||||
auditAction = "capacity.admission.denied"
|
||||
}
|
||||
auditID, err := svc.recordAuditEventWithID(actorID, auditAction, "run-endpoint", endpoint.ID, auditResult, decision.Reason)
|
||||
if err != nil {
|
||||
return domain.CapacityAdmissionDecision{}, err
|
||||
}
|
||||
decision.AuditEventID = auditID
|
||||
if !decision.Accepted {
|
||||
severity := domain.AlertSeverityWarning
|
||||
if hardDenied {
|
||||
severity = domain.AlertSeverityCritical
|
||||
}
|
||||
alert, err := svc.upsertAlert(domain.AlertRecord{
|
||||
SourceKind: "run-endpoint", SourceID: endpoint.ID, RuleKey: "capacity.pressure", Severity: severity,
|
||||
Title: "Run endpoint capacity admission blocked", Message: decision.Reason, Retryable: true,
|
||||
RetryAfterSeconds: decision.RetryAfterSeconds, LastAuditEventID: auditID,
|
||||
})
|
||||
if err != nil {
|
||||
return domain.CapacityAdmissionDecision{}, err
|
||||
}
|
||||
decision.AlertID = alert.ID
|
||||
} else if err := svc.resolveAlertForSource("run-endpoint", endpoint.ID, "capacity.pressure", actorID, "capacity returned to an admissible state", auditID); err != nil {
|
||||
return domain.CapacityAdmissionDecision{}, err
|
||||
}
|
||||
if err := validator.ValidateCapacityAdmissionDecision(decision); err != nil {
|
||||
return domain.CapacityAdmissionDecision{}, err
|
||||
}
|
||||
return domain.CopyCapacityAdmissionDecision(decision), nil
|
||||
}
|
||||
|
||||
func (svc *CoreService) ListAlertsForSession(sessionID string, filter domain.AlertFilter) ([]domain.AlertRecord, error) {
|
||||
user, err := svc.GetCurrentUser(sessionID)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
alerts, err := svc.store.Alerts().List(filter)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
visible := make([]domain.AlertRecord, 0, len(alerts))
|
||||
for _, alert := range alerts {
|
||||
if svc.canAccessAlert(user, alert) {
|
||||
visible = append(visible, alert)
|
||||
}
|
||||
}
|
||||
sort.Slice(visible, func(i, j int) bool { return visible[i].UpdatedAt.After(visible[j].UpdatedAt) })
|
||||
return domain.CopyAlertRecords(visible), nil
|
||||
}
|
||||
|
||||
func (svc *CoreService) AcknowledgeAlertForSession(sessionID string, request domain.AlertAcknowledgeRequest) (domain.AlertRecord, error) {
|
||||
if err := validator.ValidateAlertAcknowledgeRequest(request); err != nil {
|
||||
return domain.AlertRecord{}, err
|
||||
}
|
||||
user, err := svc.GetCurrentUser(sessionID)
|
||||
if err != nil {
|
||||
return domain.AlertRecord{}, err
|
||||
}
|
||||
svc.productionMu.Lock()
|
||||
defer svc.productionMu.Unlock()
|
||||
alert, err := svc.store.Alerts().Get(request.AlertID)
|
||||
if err != nil {
|
||||
return domain.AlertRecord{}, err
|
||||
}
|
||||
if !svc.canAccessAlert(user, alert) {
|
||||
return domain.AlertRecord{}, ErrForbidden
|
||||
}
|
||||
if alert.State == domain.AlertStateResolved {
|
||||
return domain.AlertRecord{}, validationError("resolved alerts cannot be acknowledged")
|
||||
}
|
||||
stamp := svc.now()
|
||||
auditID, err := svc.recordAuditEventWithID(user.ID, "alert.acknowledge", "alert", alert.ID, domain.AuditResultSuccess, defaultAlertNote(request.Note, "alert acknowledged"))
|
||||
if err != nil {
|
||||
return domain.AlertRecord{}, err
|
||||
}
|
||||
alert.State = domain.AlertStateAcknowledged
|
||||
alert.AcknowledgedBy = user.ID
|
||||
alert.AcknowledgedAt = stamp
|
||||
alert.LastAuditEventID = auditID
|
||||
alert.UpdatedAt = stamp
|
||||
if err := validator.ValidateAlertRecord(alert); err != nil {
|
||||
return domain.AlertRecord{}, err
|
||||
}
|
||||
if err := svc.store.Alerts().Update(alert); err != nil {
|
||||
return domain.AlertRecord{}, err
|
||||
}
|
||||
return domain.CopyAlertRecord(alert), nil
|
||||
}
|
||||
|
||||
func (svc *CoreService) ResolveAlertForSession(sessionID string, request domain.AlertResolveRequest) (domain.AlertRecord, error) {
|
||||
if err := validator.ValidateAlertResolveRequest(request); err != nil {
|
||||
return domain.AlertRecord{}, err
|
||||
}
|
||||
user, err := svc.GetCurrentUser(sessionID)
|
||||
if err != nil {
|
||||
return domain.AlertRecord{}, err
|
||||
}
|
||||
svc.productionMu.Lock()
|
||||
defer svc.productionMu.Unlock()
|
||||
alert, err := svc.store.Alerts().Get(request.AlertID)
|
||||
if err != nil {
|
||||
return domain.AlertRecord{}, err
|
||||
}
|
||||
if !svc.canAccessAlert(user, alert) {
|
||||
return domain.AlertRecord{}, ErrForbidden
|
||||
}
|
||||
if alert.State == domain.AlertStateResolved {
|
||||
return domain.CopyAlertRecord(alert), nil
|
||||
}
|
||||
stamp := svc.now()
|
||||
note := defaultAlertNote(request.Note, "alert resolved after operator review")
|
||||
auditID, err := svc.recordAuditEventWithID(user.ID, "alert.resolve", "alert", alert.ID, domain.AuditResultSuccess, note)
|
||||
if err != nil {
|
||||
return domain.AlertRecord{}, err
|
||||
}
|
||||
alert.State = domain.AlertStateResolved
|
||||
alert.ResolvedBy = user.ID
|
||||
alert.ResolvedAt = stamp
|
||||
alert.ResolutionNote = note
|
||||
alert.LastAuditEventID = auditID
|
||||
alert.UpdatedAt = stamp
|
||||
if err := validator.ValidateAlertRecord(alert); err != nil {
|
||||
return domain.AlertRecord{}, err
|
||||
}
|
||||
if err := svc.store.Alerts().Update(alert); err != nil {
|
||||
return domain.AlertRecord{}, err
|
||||
}
|
||||
return domain.CopyAlertRecord(alert), nil
|
||||
}
|
||||
|
||||
func (svc *CoreService) RetryAlertForSession(sessionID string, request domain.AlertRetryRequest) (domain.AlertRetryResult, error) {
|
||||
if err := validator.ValidateAlertRetryRequest(request); err != nil {
|
||||
return domain.AlertRetryResult{}, err
|
||||
}
|
||||
user, err := svc.GetCurrentUser(sessionID)
|
||||
if err != nil {
|
||||
return domain.AlertRetryResult{}, err
|
||||
}
|
||||
alert, err := svc.store.Alerts().Get(request.AlertID)
|
||||
if err != nil {
|
||||
return domain.AlertRetryResult{}, err
|
||||
}
|
||||
if !svc.canAccessAlert(user, alert) {
|
||||
return domain.AlertRetryResult{}, ErrForbidden
|
||||
}
|
||||
if !alert.Retryable {
|
||||
return domain.AlertRetryResult{}, validationError("alert source is not retryable")
|
||||
}
|
||||
switch alert.SourceKind {
|
||||
case "run-endpoint":
|
||||
endpoint, err := svc.store.RunEndpoints().Get(alert.SourceID)
|
||||
if err != nil {
|
||||
return domain.AlertRetryResult{}, err
|
||||
}
|
||||
capability := firstCapacityCapability(endpoint.Capabilities)
|
||||
decision, err := svc.CheckCapacityAdmissionForSession(sessionID, domain.CapacityAdmissionRequest{RunEndpointID: endpoint.ID, Capability: capability, IdempotencyKey: request.IdempotencyKey})
|
||||
if err != nil {
|
||||
return domain.AlertRetryResult{}, err
|
||||
}
|
||||
updated, err := svc.store.Alerts().Get(alert.ID)
|
||||
if err != nil {
|
||||
return domain.AlertRetryResult{}, err
|
||||
}
|
||||
return domain.CopyAlertRetryResult(domain.AlertRetryResult{Alert: updated, Decision: decision, Status: string(decision.State)}), nil
|
||||
case "plugin-lifecycle":
|
||||
installation, err := svc.store.PluginLifecycles().Get(alert.SourceID)
|
||||
if err != nil {
|
||||
return domain.AlertRetryResult{}, err
|
||||
}
|
||||
result, err := svc.RunPluginLifecycleForSession(sessionID, domain.PluginLifecycleRequest{PluginID: installation.PluginID, ServerInstanceID: installation.ServerInstanceID, Operation: installation.LastOperation, TargetVersion: installation.TargetVersion, IdempotencyKey: request.IdempotencyKey, Confirmed: true})
|
||||
if err != nil {
|
||||
return domain.AlertRetryResult{}, err
|
||||
}
|
||||
return domain.CopyAlertRetryResult(domain.AlertRetryResult{Alert: alert, Decision: result.Decision, Status: result.Status}), nil
|
||||
default:
|
||||
return domain.AlertRetryResult{}, validationError("alert source does not support scoped retry")
|
||||
}
|
||||
}
|
||||
|
||||
func (svc *CoreService) ListPluginLifecyclesForSession(sessionID string, filter domain.PluginLifecycleFilter) ([]domain.PluginLifecycleInstallation, error) {
|
||||
user, err := svc.GetCurrentUser(sessionID)
|
||||
if err != nil {
|
||||
@@ -383,13 +95,6 @@ func (svc *CoreService) RunPluginLifecycleForSession(sessionID string, request d
|
||||
if !errors.Is(jobErr, repo.ErrNotFound) {
|
||||
return domain.PluginLifecycleResult{}, jobErr
|
||||
}
|
||||
decision, err := svc.checkCapacityAdmission(user.ID, domain.CapacityAdmissionRequest{ServerInstanceID: instance.ID, RunEndpointID: endpoint.ID, Capability: capability, TargetKey: targetKey, IdempotencyKey: request.IdempotencyKey})
|
||||
if err != nil {
|
||||
return domain.PluginLifecycleResult{}, err
|
||||
}
|
||||
if !decision.Accepted {
|
||||
return domain.CopyPluginLifecycleResult(domain.PluginLifecycleResult{Installation: installation, Decision: decision, Status: string(decision.State)}), nil
|
||||
}
|
||||
job, err := svc.CreateJob(domain.Job{
|
||||
ID: jobIDFromParts("job-plugin-lifecycle", installation.ID, request.IdempotencyKey), ServerInstanceID: instance.ID,
|
||||
RunEndpointID: endpoint.ID, Capability: capability, TargetKey: targetKey, IdempotencyKey: request.IdempotencyKey,
|
||||
@@ -409,11 +114,6 @@ func (svc *CoreService) RunPluginLifecycleForSession(sessionID string, request d
|
||||
installation.IdempotencyKey = request.IdempotencyKey
|
||||
installation.FailureReason = ""
|
||||
installation.UpdatedAt = stamp
|
||||
auditID, err := svc.recordAuditEventWithID(user.ID, "plugin.lifecycle."+string(request.Operation), "plugin-lifecycle", installation.ID, domain.AuditResultQueued, "plugin lifecycle operation admitted and queued")
|
||||
if err != nil {
|
||||
return domain.PluginLifecycleResult{}, err
|
||||
}
|
||||
installation.AuditEventID = auditID
|
||||
if err := validator.ValidatePluginLifecycleInstallation(installation); err != nil {
|
||||
return domain.PluginLifecycleResult{}, err
|
||||
}
|
||||
@@ -425,7 +125,7 @@ func (svc *CoreService) RunPluginLifecycleForSession(sessionID string, request d
|
||||
if err != nil {
|
||||
return domain.PluginLifecycleResult{}, err
|
||||
}
|
||||
return domain.CopyPluginLifecycleResult(domain.PluginLifecycleResult{Installation: installation, Job: job, Decision: decision, Status: "queued"}), nil
|
||||
return domain.CopyPluginLifecycleResult(domain.PluginLifecycleResult{Installation: installation, Job: job, Status: "queued"}), nil
|
||||
}
|
||||
|
||||
func (svc *CoreService) ListAIConfigDiffsForSession(sessionID string, filter domain.AIConfigDiffFilter) ([]domain.AIConfigDiffPreview, error) {
|
||||
@@ -503,9 +203,6 @@ func (svc *CoreService) ApproveAIConfigDiffForSession(sessionID string, request
|
||||
if err := svc.store.AIConfigDiffs().Update(preview); err != nil {
|
||||
return domain.AIConfigDiffApprovalResult{}, err
|
||||
}
|
||||
if _, err := svc.recordAuditEventWithID(user.ID, "ai.config-diff.approve", "ai-config-diff", preview.ID, domain.AuditResultQueued, "approved reviewed AI config diff and queued one config write job"); err != nil {
|
||||
return domain.AIConfigDiffApprovalResult{}, err
|
||||
}
|
||||
return domain.CopyAIConfigDiffApprovalResult(domain.AIConfigDiffApprovalResult{Preview: preview, Dispatch: dispatch}), nil
|
||||
}
|
||||
|
||||
@@ -525,17 +222,9 @@ func (svc *CoreService) projectProductionOpsJobResult(job domain.Job, stamp time
|
||||
if job.State == domain.JobStateSucceeded {
|
||||
applyPluginLifecycleSuccess(&installation)
|
||||
installation.FailureReason = ""
|
||||
if err := svc.resolveAlertForSource("plugin-lifecycle", installation.ID, "plugin.lifecycle.failed", "run:"+job.RunEndpointID, "plugin lifecycle job completed", installation.AuditEventID); err != nil {
|
||||
return err
|
||||
}
|
||||
} else {
|
||||
installation.CurrentState = domain.PluginLifecycleStateFailed
|
||||
installation.FailureReason = "plugin lifecycle job did not complete successfully"
|
||||
alert, err := svc.upsertAlert(domain.AlertRecord{SourceKind: "plugin-lifecycle", SourceID: installation.ID, RuleKey: "plugin.lifecycle.failed", Severity: domain.AlertSeverityWarning, Title: "Plugin lifecycle operation failed", Message: installation.FailureReason, Retryable: true, LastJobID: job.ID, LastAuditEventID: installation.AuditEventID})
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
installation.AlertID = alert.ID
|
||||
}
|
||||
installation.UpdatedAt = stamp
|
||||
return svc.store.PluginLifecycles().Update(installation)
|
||||
@@ -598,193 +287,6 @@ func (svc *CoreService) getServerConfigForUser(userID, serverInstanceID string)
|
||||
return config, validator.ValidateServerConfig(config)
|
||||
}
|
||||
|
||||
func (svc *CoreService) authorizeCapacityRequest(user domain.User, request domain.CapacityAdmissionRequest) (domain.CapacityAdmissionRequest, error) {
|
||||
if request.ServerInstanceID != "" {
|
||||
instance, err := svc.store.ServerInstances().Get(request.ServerInstanceID)
|
||||
if err != nil {
|
||||
return request, err
|
||||
}
|
||||
if !canAccessServer(user, instance) {
|
||||
return request, ErrForbidden
|
||||
}
|
||||
if request.RunEndpointID != "" && request.RunEndpointID != instance.RunEndpointID {
|
||||
return request, validationError("runEndpointId must match server instance")
|
||||
}
|
||||
request.RunEndpointID = instance.RunEndpointID
|
||||
return request, nil
|
||||
}
|
||||
if request.RunEndpointID == "" {
|
||||
return request, validationError("serverInstanceId or runEndpointId is required")
|
||||
}
|
||||
if isPlatformAdmin(user) {
|
||||
return request, nil
|
||||
}
|
||||
visible, err := svc.visibleEndpointIDs(user)
|
||||
if err != nil {
|
||||
return request, err
|
||||
}
|
||||
if _, ok := visible[request.RunEndpointID]; !ok {
|
||||
return request, ErrForbidden
|
||||
}
|
||||
return request, nil
|
||||
}
|
||||
|
||||
func (svc *CoreService) visibleEndpointIDs(user domain.User) (map[string]struct{}, error) {
|
||||
instances, err := svc.store.ServerInstances().List(domain.ServerInstanceFilter{})
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
ids := map[string]struct{}{}
|
||||
for _, instance := range instances {
|
||||
if isPlatformAdmin(user) || canAccessServer(user, instance) {
|
||||
ids[instance.RunEndpointID] = struct{}{}
|
||||
}
|
||||
}
|
||||
return ids, nil
|
||||
}
|
||||
|
||||
func (svc *CoreService) capacityJobCounts(endpointID string) (int, int, error) {
|
||||
jobs, err := svc.store.Jobs().List(domain.JobFilter{RunEndpointID: endpointID})
|
||||
if err != nil {
|
||||
return 0, 0, err
|
||||
}
|
||||
running, queued := 0, 0
|
||||
for _, job := range jobs {
|
||||
switch job.State {
|
||||
case domain.JobStateAccepted, domain.JobStateRunning:
|
||||
running++
|
||||
case domain.JobStateQueued, domain.JobStateRetrying:
|
||||
queued++
|
||||
}
|
||||
}
|
||||
return running, queued, nil
|
||||
}
|
||||
|
||||
func (svc *CoreService) capacityProjection(endpoint domain.RunEndpoint, durableRunning, durableQueued int) domain.EndpointCapacityProjection {
|
||||
running := maxInt(endpoint.Capacity.RunningJobs, durableRunning)
|
||||
queued := maxInt(endpoint.Capacity.QueuedJobs, durableQueued)
|
||||
projection := domain.EndpointCapacityProjection{RunEndpointID: endpoint.ID, DisplayName: endpoint.DisplayName, Status: endpoint.Status, Capabilities: endpoint.Capabilities, MaxJobs: endpoint.Capacity.MaxJobs, RunningJobs: running, QueuedJobs: queued, LogBacklogBatches: endpoint.Capacity.LogBacklogBatches, ArtifactBacklogChunks: endpoint.Capacity.ArtifactBacklogChunks, Summary: safeBridgeReason(endpoint.Capacity.Summary), LastHeartbeatAt: endpoint.LastHeartbeatAt}
|
||||
if endpoint.Status != domain.RunEndpointStatusOnline && endpoint.Status != domain.RunEndpointStatusDegraded {
|
||||
projection.PressureCodes = appendCapacityPressure(projection.PressureCodes, domain.CapacityPressureEndpointOffline)
|
||||
}
|
||||
if endpoint.LastHeartbeatAt.IsZero() || svc.now().Sub(endpoint.LastHeartbeatAt) > capacityHeartbeatStaleAfter {
|
||||
projection.PressureCodes = appendCapacityPressure(projection.PressureCodes, domain.CapacityPressureEndpointStale)
|
||||
}
|
||||
if projection.MaxJobs <= 0 || running >= projection.MaxJobs {
|
||||
projection.PressureCodes = appendCapacityPressure(projection.PressureCodes, domain.CapacityPressureJobLimit)
|
||||
}
|
||||
queueLimit := maxInt(4, projection.MaxJobs*2)
|
||||
if queued >= queueLimit {
|
||||
projection.PressureCodes = appendCapacityPressure(projection.PressureCodes, domain.CapacityPressureQueueLimit)
|
||||
}
|
||||
if projection.LogBacklogBatches >= capacityLogBacklogLimit || projection.ArtifactBacklogChunks >= capacityArtifactBacklogLimit || len(endpoint.Capacity.PressureCodes) > 0 {
|
||||
projection.PressureCodes = appendCapacityPressure(projection.PressureCodes, domain.CapacityPressureBacklog)
|
||||
}
|
||||
return projection
|
||||
}
|
||||
|
||||
func (svc *CoreService) upsertAlert(candidate domain.AlertRecord) (domain.AlertRecord, error) {
|
||||
stamp := svc.now()
|
||||
candidate.ID = alertIDForSource(candidate.SourceKind, candidate.SourceID, candidate.RuleKey)
|
||||
existing, err := svc.store.Alerts().Get(candidate.ID)
|
||||
if err == nil {
|
||||
existing.Severity = candidate.Severity
|
||||
existing.State = domain.AlertStateActive
|
||||
existing.Title = candidate.Title
|
||||
existing.Message = safeBridgeReason(candidate.Message)
|
||||
existing.OccurrenceCount++
|
||||
existing.Retryable = candidate.Retryable
|
||||
existing.RetryAfterSeconds = candidate.RetryAfterSeconds
|
||||
existing.LastJobID = candidate.LastJobID
|
||||
existing.LastAuditEventID = candidate.LastAuditEventID
|
||||
existing.LastSeenAt = stamp
|
||||
existing.ResolvedBy = ""
|
||||
existing.ResolvedAt = time.Time{}
|
||||
existing.ResolutionNote = ""
|
||||
existing.UpdatedAt = stamp
|
||||
if err := validator.ValidateAlertRecord(existing); err != nil {
|
||||
return domain.AlertRecord{}, err
|
||||
}
|
||||
if err := svc.store.Alerts().Update(existing); err != nil {
|
||||
return domain.AlertRecord{}, err
|
||||
}
|
||||
return existing, nil
|
||||
}
|
||||
if !errors.Is(err, repo.ErrNotFound) {
|
||||
return domain.AlertRecord{}, err
|
||||
}
|
||||
candidate.State = domain.AlertStateActive
|
||||
candidate.Message = safeBridgeReason(candidate.Message)
|
||||
candidate.OccurrenceCount = 1
|
||||
candidate.LastSeenAt = stamp
|
||||
candidate.CreatedAt = stamp
|
||||
candidate.UpdatedAt = stamp
|
||||
if err := validator.ValidateAlertRecord(candidate); err != nil {
|
||||
return domain.AlertRecord{}, err
|
||||
}
|
||||
if err := svc.store.Alerts().Create(candidate); err != nil {
|
||||
return domain.AlertRecord{}, err
|
||||
}
|
||||
return candidate, nil
|
||||
}
|
||||
|
||||
func (svc *CoreService) resolveAlertForSource(sourceKind, sourceID, ruleKey, actorID, note, auditID string) error {
|
||||
alert, err := svc.store.Alerts().Get(alertIDForSource(sourceKind, sourceID, ruleKey))
|
||||
if errors.Is(err, repo.ErrNotFound) {
|
||||
return nil
|
||||
}
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
if alert.State == domain.AlertStateResolved {
|
||||
return nil
|
||||
}
|
||||
stamp := svc.now()
|
||||
alert.State = domain.AlertStateResolved
|
||||
alert.ResolvedBy = actorID
|
||||
alert.ResolvedAt = stamp
|
||||
alert.ResolutionNote = note
|
||||
alert.LastAuditEventID = auditID
|
||||
alert.UpdatedAt = stamp
|
||||
return svc.store.Alerts().Update(alert)
|
||||
}
|
||||
|
||||
func (svc *CoreService) canAccessAlert(user domain.User, alert domain.AlertRecord) bool {
|
||||
if isPlatformAdmin(user) {
|
||||
return true
|
||||
}
|
||||
switch alert.SourceKind {
|
||||
case "server-instance":
|
||||
instance, err := svc.store.ServerInstances().Get(alert.SourceID)
|
||||
return err == nil && canAccessServer(user, instance)
|
||||
case "run-endpoint":
|
||||
instances, err := svc.store.ServerInstances().List(domain.ServerInstanceFilter{RunEndpointID: alert.SourceID})
|
||||
if err != nil {
|
||||
return false
|
||||
}
|
||||
for _, instance := range instances {
|
||||
if canAccessServer(user, instance) {
|
||||
return true
|
||||
}
|
||||
}
|
||||
case "plugin-lifecycle":
|
||||
installation, err := svc.store.PluginLifecycles().Get(alert.SourceID)
|
||||
if err != nil {
|
||||
return false
|
||||
}
|
||||
instance, err := svc.store.ServerInstances().Get(installation.ServerInstanceID)
|
||||
return err == nil && canAccessServer(user, instance)
|
||||
case "ai-config-diff":
|
||||
preview, err := svc.store.AIConfigDiffs().Get(alert.SourceID)
|
||||
if err != nil {
|
||||
return false
|
||||
}
|
||||
instance, err := svc.store.ServerInstances().Get(preview.ServerInstanceID)
|
||||
return err == nil && canAccessServer(user, instance)
|
||||
}
|
||||
return false
|
||||
}
|
||||
|
||||
func (svc *CoreService) pluginLifecycleDenied(actorID string, instance domain.ServerInstance, plugin domain.GamePlugin, request domain.PluginLifecycleRequest, reason string) (domain.PluginLifecycleResult, error) {
|
||||
svc.productionMu.Lock()
|
||||
defer svc.productionMu.Unlock()
|
||||
@@ -798,21 +300,11 @@ func (svc *CoreService) pluginLifecycleDenied(actorID string, instance domain.Se
|
||||
|
||||
func (svc *CoreService) pluginLifecycleDeniedLocked(actorID string, installation domain.PluginLifecycleInstallation, request domain.PluginLifecycleRequest, reason string) (domain.PluginLifecycleResult, error) {
|
||||
stamp := svc.now()
|
||||
auditID, err := svc.recordAuditEventWithID(actorID, "plugin.lifecycle.denied", "plugin-lifecycle", installation.ID, domain.AuditResultDenied, reason)
|
||||
if err != nil {
|
||||
return domain.PluginLifecycleResult{}, err
|
||||
}
|
||||
installation.CurrentState = domain.PluginLifecycleStateFailed
|
||||
installation.LastOperation = request.Operation
|
||||
installation.TargetVersion = request.TargetVersion
|
||||
installation.FailureReason = safeBridgeReason(reason)
|
||||
installation.AuditEventID = auditID
|
||||
installation.UpdatedAt = stamp
|
||||
alert, err := svc.upsertAlert(domain.AlertRecord{SourceKind: "plugin-lifecycle", SourceID: installation.ID, RuleKey: "plugin.lifecycle.compatibility", Severity: domain.AlertSeverityWarning, Title: "Plugin lifecycle compatibility check failed", Message: installation.FailureReason, Retryable: true, LastAuditEventID: auditID})
|
||||
if err != nil {
|
||||
return domain.PluginLifecycleResult{}, err
|
||||
}
|
||||
installation.AlertID = alert.ID
|
||||
if err := validator.ValidatePluginLifecycleInstallation(installation); err != nil {
|
||||
return domain.PluginLifecycleResult{}, err
|
||||
}
|
||||
@@ -824,7 +316,7 @@ func (svc *CoreService) pluginLifecycleDeniedLocked(actorID string, installation
|
||||
if err != nil {
|
||||
return domain.PluginLifecycleResult{}, err
|
||||
}
|
||||
return domain.CopyPluginLifecycleResult(domain.PluginLifecycleResult{Installation: installation, Alert: &alert, Status: "denied"}), nil
|
||||
return domain.CopyPluginLifecycleResult(domain.PluginLifecycleResult{Installation: installation, Status: "denied"}), nil
|
||||
}
|
||||
|
||||
func pluginLifecycleDispatchMetadata(plugin domain.GamePlugin, operation domain.PluginLifecycleOperation) (string, string, error) {
|
||||
@@ -937,11 +429,6 @@ func applyPluginLifecycleSuccess(installation *domain.PluginLifecycleInstallatio
|
||||
}
|
||||
}
|
||||
|
||||
func alertIDForSource(sourceKind, sourceID, ruleKey string) string {
|
||||
sum := sha256.Sum256([]byte(sourceKind + "\x00" + sourceID + "\x00" + ruleKey))
|
||||
return "alert-" + hex.EncodeToString(sum[:12])
|
||||
}
|
||||
|
||||
func pluginLifecycleInstallationID(pluginID, serverInstanceID string) string {
|
||||
sum := sha256.Sum256([]byte(pluginID + "\x00" + serverInstanceID))
|
||||
return "plugin-lifecycle-" + hex.EncodeToString(sum[:12])
|
||||
@@ -951,42 +438,3 @@ func aiConfigDiffID(requestID, serverInstanceID string) string {
|
||||
sum := sha256.Sum256([]byte(requestID + "\x00" + serverInstanceID))
|
||||
return "ai-config-diff-" + hex.EncodeToString(sum[:12])
|
||||
}
|
||||
|
||||
func appendCapacityPressure(codes []domain.CapacityPressureCode, code domain.CapacityPressureCode) []domain.CapacityPressureCode {
|
||||
if !containsCapacityPressure(codes, code) {
|
||||
return append(codes, code)
|
||||
}
|
||||
return codes
|
||||
}
|
||||
|
||||
func containsCapacityPressure(codes []domain.CapacityPressureCode, target domain.CapacityPressureCode) bool {
|
||||
for _, code := range codes {
|
||||
if code == target {
|
||||
return true
|
||||
}
|
||||
}
|
||||
return false
|
||||
}
|
||||
|
||||
func firstCapacityCapability(capabilities []string) string {
|
||||
for _, capability := range capabilities {
|
||||
if strings.TrimSpace(capability) != "" {
|
||||
return capability
|
||||
}
|
||||
}
|
||||
return "control.heartbeat"
|
||||
}
|
||||
|
||||
func defaultAlertNote(note, fallback string) string {
|
||||
if strings.TrimSpace(note) == "" {
|
||||
return fallback
|
||||
}
|
||||
return safeBridgeReason(note)
|
||||
}
|
||||
|
||||
func maxInt(a, b int) int {
|
||||
if a > b {
|
||||
return a
|
||||
}
|
||||
return b
|
||||
}
|
||||
|
||||
Reference in New Issue
Block a user