mirror of
https://github.com/Buriburizaem0n/nezha_domains.git
synced 2026-09-19 09:40:12 +00:00
test(agentcompat): add exact stress scenario
Co-authored-by: naiba/CloudCode <hi+cloudcode@nai.ba>
This commit is contained in:
naiba
naiba/CloudCode
@@ -0,0 +1,188 @@
|
||||
//go:build linux && agentcompat
|
||||
|
||||
package scenario
|
||||
|
||||
import (
|
||||
"bytes"
|
||||
"encoding/json"
|
||||
"errors"
|
||||
"fmt"
|
||||
"io"
|
||||
"os"
|
||||
"path/filepath"
|
||||
|
||||
"github.com/nezhahq/nezha/integration/agentcompat/internal/contract"
|
||||
"github.com/nezhahq/nezha/integration/agentcompat/internal/evidence"
|
||||
)
|
||||
|
||||
var ErrStressArtifactInvalid = errors.New("stress artifact is invalid")
|
||||
|
||||
type stressArtifact struct {
|
||||
Version int `json:"version"`
|
||||
Profile string `json:"profile"`
|
||||
Seed string `json:"seed"`
|
||||
RoundCount int `json:"round_count"`
|
||||
OperationCount int `json:"operation_count"`
|
||||
SessionCount int `json:"session_count"`
|
||||
WarmupCount int `json:"warmup_count"`
|
||||
ResourceSummaryCount int `json:"resource_summary_count"`
|
||||
ResourceSampleCount int `json:"resource_sample_count"`
|
||||
ResourceIntervalMilliseconds int64 `json:"resource_interval_milliseconds"`
|
||||
Quotas StressQuotaEvidence `json:"quotas"`
|
||||
PathLockStripes int `json:"path_lock_stripes"`
|
||||
DuplicateOperations int `json:"duplicate_operations"`
|
||||
ResourceDrift int `json:"resource_drift"`
|
||||
RSSBounded bool `json:"rss_bounded"`
|
||||
Cleanup StressCleanupSummary `json:"cleanup"`
|
||||
}
|
||||
|
||||
func stressPRFullProfile() (contract.Profile, error) {
|
||||
return contract.ProfileByName(string(contract.ProfilePRFull))
|
||||
}
|
||||
|
||||
func publishStressEvidence(root string, value StressEvidence) error {
|
||||
profile, err := stressPRFullProfile()
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
if err := value.ValidateSuccess(profile); err != nil {
|
||||
return err
|
||||
}
|
||||
artifact, err := newStressArtifact(value)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
data, err := json.Marshal(artifact)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
if evidence.Redact(string(data)) != string(data) {
|
||||
return ErrStressArtifactInvalid
|
||||
}
|
||||
info, err := os.Lstat(root)
|
||||
if errors.Is(err, os.ErrNotExist) {
|
||||
if err := os.Mkdir(root, 0o700); err != nil {
|
||||
return err
|
||||
}
|
||||
info, err = os.Lstat(root)
|
||||
}
|
||||
if err != nil || info.Mode()&os.ModeSymlink != 0 || !info.IsDir() || info.Mode().Perm() != 0o700 {
|
||||
return ErrStressArtifactInvalid
|
||||
}
|
||||
path := filepath.Join(root, "stress.json")
|
||||
if stale, statErr := os.Lstat(path); statErr == nil {
|
||||
if stale.Mode()&os.ModeSymlink != 0 || !stale.Mode().IsRegular() {
|
||||
return ErrStressArtifactInvalid
|
||||
}
|
||||
} else if !errors.Is(statErr, os.ErrNotExist) {
|
||||
return statErr
|
||||
}
|
||||
temporary, err := os.CreateTemp(root, ".stress-*")
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
temporaryName := temporary.Name()
|
||||
defer os.Remove(temporaryName)
|
||||
if err := temporary.Chmod(0o600); err != nil {
|
||||
_ = temporary.Close()
|
||||
return err
|
||||
}
|
||||
if _, err := temporary.Write(data); err != nil {
|
||||
_ = temporary.Close()
|
||||
return err
|
||||
}
|
||||
if err := temporary.Close(); err != nil {
|
||||
return err
|
||||
}
|
||||
return os.Rename(temporaryName, path)
|
||||
}
|
||||
|
||||
func readStressEvidence(root string) (stressArtifact, error) {
|
||||
var value stressArtifact
|
||||
path := filepath.Join(root, "stress.json")
|
||||
info, err := os.Lstat(path)
|
||||
if err != nil {
|
||||
return value, err
|
||||
}
|
||||
if info.Mode()&os.ModeSymlink != 0 || !info.Mode().IsRegular() || info.Mode().Perm() != 0o600 {
|
||||
return value, ErrStressArtifactInvalid
|
||||
}
|
||||
data, err := os.ReadFile(path)
|
||||
if err != nil {
|
||||
return value, err
|
||||
}
|
||||
if evidence.Redact(string(data)) != string(data) {
|
||||
return value, ErrStressArtifactInvalid
|
||||
}
|
||||
decoder := json.NewDecoder(bytes.NewReader(data))
|
||||
decoder.DisallowUnknownFields()
|
||||
if err := decoder.Decode(&value); err != nil {
|
||||
return value, fmt.Errorf("decode stress artifact: %w", ErrStressArtifactInvalid)
|
||||
}
|
||||
var trailing struct{}
|
||||
if err := decoder.Decode(&trailing); !errors.Is(err, io.EOF) {
|
||||
return value, ErrStressArtifactInvalid
|
||||
}
|
||||
if err := value.validate(); err != nil {
|
||||
return value, err
|
||||
}
|
||||
return value, nil
|
||||
}
|
||||
|
||||
func newStressArtifact(value StressEvidence) (stressArtifact, error) {
|
||||
profile, err := contract.ProfileByName(string(value.Profile))
|
||||
if err != nil {
|
||||
return stressArtifact{}, err
|
||||
}
|
||||
artifact := stressArtifact{Version: 1, Profile: string(value.Profile), Seed: fmt.Sprintf("%08x", uint64(value.Seed)), SessionCount: len(value.Sessions), WarmupCount: len(value.Warmups), ResourceSummaryCount: 1 + profile.AgentCount(), ResourceSampleCount: 5, ResourceIntervalMilliseconds: 250, Quotas: value.Quotas, PathLockStripes: value.Quotas.PathLockStripes, Cleanup: value.Cleanup}
|
||||
operationIDs := make(map[StressOperationID]struct{})
|
||||
resourceEvaluations := make([]StressResourceEvaluation, 0, artifact.ResourceSummaryCount)
|
||||
for _, iteration := range value.Iterations {
|
||||
artifact.RoundCount += len(iteration.Rounds)
|
||||
for _, round := range iteration.Rounds {
|
||||
artifact.OperationCount += len(round.Operations)
|
||||
for _, operation := range round.Operations {
|
||||
if _, duplicate := operationIDs[operation.ID]; duplicate {
|
||||
artifact.DuplicateOperations++
|
||||
}
|
||||
operationIDs[operation.ID] = struct{}{}
|
||||
}
|
||||
}
|
||||
for _, resource := range iteration.Resources {
|
||||
evaluation, err := EvaluateStressResource(resource)
|
||||
if err != nil {
|
||||
return stressArtifact{}, err
|
||||
}
|
||||
resourceEvaluations = append(resourceEvaluations, evaluation)
|
||||
}
|
||||
}
|
||||
resourceDrift, rssBounded, err := aggregateStressResourceEvaluations(resourceEvaluations, artifact.ResourceSummaryCount)
|
||||
if err != nil {
|
||||
return stressArtifact{}, err
|
||||
}
|
||||
artifact.ResourceDrift = resourceDrift
|
||||
artifact.RSSBounded = rssBounded
|
||||
return artifact, artifact.validate()
|
||||
}
|
||||
|
||||
func aggregateStressResourceEvaluations(evaluations []StressResourceEvaluation, expectedCount int) (int, bool, error) {
|
||||
if len(evaluations) != expectedCount {
|
||||
return 0, false, fmt.Errorf("resource evaluations=%d want=%d: %w", len(evaluations), expectedCount, ErrStressArtifactInvalid)
|
||||
}
|
||||
drift := 0
|
||||
rssBounded := true
|
||||
for _, evaluation := range evaluations {
|
||||
if evaluation.Baseline.Descendants != evaluation.End.Descendants || evaluation.Baseline.NonStdioFDs != evaluation.End.NonStdioFDs || evaluation.Baseline.TCPListeners != evaluation.End.TCPListeners || evaluation.Baseline.TCP6Listeners != evaluation.End.TCP6Listeners {
|
||||
drift++
|
||||
}
|
||||
rssBounded = rssBounded && evaluation.RSSDeltaBytes <= evaluation.RSSLimitBytes
|
||||
}
|
||||
return drift, rssBounded, nil
|
||||
}
|
||||
|
||||
func (artifact stressArtifact) validate() error {
|
||||
if artifact.Version != 1 || artifact.Profile != string(contract.ProfilePRFull) || artifact.Seed != "4e5a4841" || artifact.RoundCount != 4 || artifact.OperationCount != 64 || artifact.SessionCount != 12 || artifact.WarmupCount != 8 || artifact.ResourceSummaryCount != 9 || artifact.ResourceSampleCount != 5 || artifact.ResourceIntervalMilliseconds != 250 || artifact.PathLockStripes != 1024 || artifact.DuplicateOperations != 0 || artifact.ResourceDrift != 0 || !artifact.RSSBounded || !artifact.Cleanup.Passed || artifact.Cleanup.ReceiptCount != 9 || artifact.Cleanup.FailedReceiptCount != 0 || artifact.Cleanup.ForcedCleanupCount != 0 || artifact.Cleanup.ProcessResidue != 0 || artifact.Cleanup.ProcessGroupResidue != 0 || artifact.Cleanup.WorkspaceResidue != 0 {
|
||||
return ErrStressArtifactInvalid
|
||||
}
|
||||
return artifact.Quotas.Validate()
|
||||
}
|
||||
@@ -0,0 +1,96 @@
|
||||
//go:build linux && agentcompat
|
||||
|
||||
package scenario
|
||||
|
||||
import (
|
||||
"encoding/json"
|
||||
"os"
|
||||
"path/filepath"
|
||||
"testing"
|
||||
|
||||
"github.com/stretchr/testify/require"
|
||||
|
||||
"github.com/nezhahq/nezha/integration/agentcompat/internal/contract"
|
||||
)
|
||||
|
||||
func TestStressArtifactRejectsUnknownAndRuntimeFields(t *testing.T) {
|
||||
profile, err := contract.ProfileByName(string(contract.ProfilePRFull))
|
||||
require.NoError(t, err)
|
||||
root := t.TempDir()
|
||||
valid := validStressArtifact(t, profile)
|
||||
data, err := json.Marshal(valid)
|
||||
require.NoError(t, err)
|
||||
for _, mutation := range []string{
|
||||
`{"unknown":true}`,
|
||||
`{"pid":123}`,
|
||||
`{"operation_id":"op"}`,
|
||||
} {
|
||||
path := filepath.Join(root, "stress.json")
|
||||
require.NoError(t, os.WriteFile(path, append(data[:len(data)-1], []byte(","+mutation[1:])...), 0o600))
|
||||
_, err = readStressEvidence(root)
|
||||
require.ErrorIs(t, err, ErrStressArtifactInvalid)
|
||||
}
|
||||
}
|
||||
|
||||
func TestStressArtifactRejectsTrailingDataAndWrongSeed(t *testing.T) {
|
||||
profile, err := contract.ProfileByName(string(contract.ProfilePRFull))
|
||||
require.NoError(t, err)
|
||||
root := t.TempDir()
|
||||
valid := validStressArtifact(t, profile)
|
||||
data, err := json.Marshal(valid)
|
||||
require.NoError(t, err)
|
||||
require.NoError(t, os.WriteFile(filepath.Join(root, "stress.json"), append(data, []byte("\n{}")...), 0o600))
|
||||
_, err = readStressEvidence(root)
|
||||
require.ErrorIs(t, err, ErrStressArtifactInvalid)
|
||||
|
||||
valid.Seed = "4e5a4842"
|
||||
data, err = json.Marshal(valid)
|
||||
require.NoError(t, err)
|
||||
require.NoError(t, os.WriteFile(filepath.Join(root, "stress.json"), data, 0o600))
|
||||
_, err = readStressEvidence(root)
|
||||
require.ErrorIs(t, err, ErrStressArtifactInvalid)
|
||||
}
|
||||
|
||||
func TestStressCleanupMutationsCannotPublishSuccess(t *testing.T) {
|
||||
profile, err := contract.ProfileByName(string(contract.ProfilePRFull))
|
||||
require.NoError(t, err)
|
||||
value := validStressEvidence(t, profile)
|
||||
for _, mutate := range []func(*StressEvidence){
|
||||
func(evidence *StressEvidence) { evidence.Cleanup.FailedReceiptCount = 1 },
|
||||
func(evidence *StressEvidence) { evidence.Cleanup.ForcedCleanupCount = 1 },
|
||||
func(evidence *StressEvidence) { evidence.Cleanup.ProcessResidue = 1 },
|
||||
func(evidence *StressEvidence) { evidence.Cleanup.ProcessGroupResidue = 1 },
|
||||
func(evidence *StressEvidence) { evidence.Cleanup.WorkspaceResidue = 1 },
|
||||
} {
|
||||
candidate := value
|
||||
mutate(&candidate)
|
||||
_, err := newStressArtifact(candidate)
|
||||
require.Error(t, err)
|
||||
}
|
||||
}
|
||||
|
||||
func TestStressArtifactAggregatesExactlyNineResourceEvaluations(t *testing.T) {
|
||||
profile, err := contract.ProfileByName(string(contract.ProfilePRFull))
|
||||
require.NoError(t, err)
|
||||
value := validStressEvidence(t, profile)
|
||||
artifact, err := newStressArtifact(value)
|
||||
require.NoError(t, err)
|
||||
require.Equal(t, 0, artifact.ResourceDrift)
|
||||
require.True(t, artifact.RSSBounded)
|
||||
|
||||
value.Iterations[0].Resources = value.Iterations[0].Resources[:8]
|
||||
_, err = newStressArtifact(value)
|
||||
require.ErrorIs(t, err, ErrStressArtifactInvalid)
|
||||
}
|
||||
|
||||
func validStressArtifact(t *testing.T, profile contract.Profile) stressArtifact {
|
||||
t.Helper()
|
||||
return stressArtifact{
|
||||
Version: 1, Profile: string(profile.Name()), Seed: "4e5a4841",
|
||||
RoundCount: 4, OperationCount: 64, SessionCount: 12, WarmupCount: 8,
|
||||
ResourceSummaryCount: 9, ResourceSampleCount: 5, ResourceIntervalMilliseconds: 250,
|
||||
Quotas: StressQuotaEvidence{PATSecond: quotaBoundary(10, 11), PATMinute: quotaBoundary(120, 121), UserStreams: quotaBoundary(20, 21), ServerStreams: quotaBoundary(40, 41)},
|
||||
PathLockStripes: 1024, DuplicateOperations: 0, ResourceDrift: 0, RSSBounded: true,
|
||||
Cleanup: StressCleanupSummary{Passed: true, ReceiptCount: 9},
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,169 @@
|
||||
//go:build linux
|
||||
|
||||
package scenario
|
||||
|
||||
import (
|
||||
"errors"
|
||||
"fmt"
|
||||
"reflect"
|
||||
|
||||
"github.com/nezhahq/nezha/integration/agentcompat/internal/contract"
|
||||
)
|
||||
|
||||
var (
|
||||
ErrStressEvidence = errors.New("stress evidence is invalid")
|
||||
ErrStressFault = errors.New("stress worker fault evidence is invalid")
|
||||
)
|
||||
|
||||
type StressPreparedBinaries struct {
|
||||
DashboardBuildCount int `json:"dashboard_build_count"`
|
||||
DashboardPathReused bool `json:"dashboard_path_reused"`
|
||||
AgentBuildCount int `json:"agent_build_count"`
|
||||
AgentPathReused bool `json:"agent_path_reused"`
|
||||
}
|
||||
|
||||
type StressQuotaBoundary struct {
|
||||
Allowed int `json:"allowed"`
|
||||
Rejected int `json:"rejected"`
|
||||
AllowedAccepted bool `json:"allowed_accepted"`
|
||||
RejectedDenied bool `json:"rejected_denied"`
|
||||
}
|
||||
|
||||
type StressQuotaEvidence struct {
|
||||
PATSecond StressQuotaBoundary `json:"pat_second"`
|
||||
PATMinute StressQuotaBoundary `json:"pat_minute"`
|
||||
UserStreams StressQuotaBoundary `json:"user_streams"`
|
||||
ServerStreams StressQuotaBoundary `json:"server_streams"`
|
||||
PathLockStripes int `json:"path_lock_stripes"`
|
||||
}
|
||||
|
||||
type StressWarmupEvidence struct {
|
||||
Agent StressAgentOrdinal `json:"agent"`
|
||||
Exec bool `json:"exec"`
|
||||
Filesystem bool `json:"filesystem"`
|
||||
Terminal bool `json:"terminal"`
|
||||
NAT bool `json:"nat"`
|
||||
FM bool `json:"file_manager"`
|
||||
}
|
||||
|
||||
type StressSessionEvidence struct {
|
||||
ID StressSessionID `json:"id"`
|
||||
Kind StressSessionKind `json:"kind"`
|
||||
Succeeded bool `json:"succeeded"`
|
||||
}
|
||||
|
||||
type StressFaultTarget struct {
|
||||
Iteration int `json:"iteration"`
|
||||
Round int `json:"round"`
|
||||
Agent StressAgentOrdinal `json:"agent"`
|
||||
Kind StressOperationKind `json:"kind"`
|
||||
}
|
||||
|
||||
type StressCleanupSummary struct {
|
||||
Passed bool `json:"passed"`
|
||||
ReceiptCount int `json:"receipt_count"`
|
||||
FailedReceiptCount int `json:"failed_receipt_count"`
|
||||
ForcedCleanupCount int `json:"forced_cleanup_count"`
|
||||
ProcessResidue int `json:"process_residue"`
|
||||
ProcessGroupResidue int `json:"process_group_residue"`
|
||||
WorkspaceResidue int `json:"workspace_residue"`
|
||||
}
|
||||
|
||||
type StressIterationEvidence struct {
|
||||
Iteration int `json:"iteration"`
|
||||
Rounds []StressRoundEvidence `json:"rounds"`
|
||||
Resources []StressProcessWindows `json:"resources"`
|
||||
}
|
||||
|
||||
type StressEvidence struct {
|
||||
Version int `json:"version"`
|
||||
Profile contract.ProfileName `json:"profile"`
|
||||
Seed contract.Seed `json:"seed"`
|
||||
PreparedBinaries StressPreparedBinaries `json:"prepared_binaries"`
|
||||
Quotas StressQuotaEvidence `json:"quotas"`
|
||||
Warmups []StressWarmupEvidence `json:"warmups"`
|
||||
Sessions []StressSessionEvidence `json:"sessions"`
|
||||
Plan StressPlan `json:"plan"`
|
||||
Iterations []StressIterationEvidence `json:"iterations"`
|
||||
FaultTarget *StressFaultTarget `json:"fault_target,omitempty"`
|
||||
SoakTrend StressSoakTrendEvidence `json:"soak_trend,omitempty"`
|
||||
Cleanup StressCleanupSummary `json:"cleanup"`
|
||||
}
|
||||
|
||||
func StressWorkerFaultTarget() StressFaultTarget {
|
||||
agent, err := NewStressAgentOrdinal(4)
|
||||
if err != nil {
|
||||
panic(err)
|
||||
}
|
||||
return StressFaultTarget{Iteration: 1, Round: 2, Agent: agent, Kind: StressOperationExec}
|
||||
}
|
||||
|
||||
func (e StressEvidence) ValidateSuccess(profile contract.Profile) error {
|
||||
return e.validate(profile, false)
|
||||
}
|
||||
|
||||
func (e StressEvidence) ValidateStressWorker(profile contract.Profile) error {
|
||||
return e.validate(profile, true)
|
||||
}
|
||||
|
||||
func (e StressEvidence) validate(profile contract.Profile, faultAware bool) error {
|
||||
if e.Version != 1 || e.Profile != profile.Name() || e.Seed != contract.DefaultSeed {
|
||||
return ErrStressEvidence
|
||||
}
|
||||
canonical, err := GenerateStressPlan(profile, e.Seed)
|
||||
if err != nil || e.Plan.Profile != e.Profile || e.Plan.Seed != e.Seed || !reflect.DeepEqual(e.Plan, canonical) {
|
||||
return fmt.Errorf("plan does not match canonical plan: %w", ErrStressEvidence)
|
||||
}
|
||||
if err := validateStressPreparedBinaries(e.PreparedBinaries); err != nil {
|
||||
return err
|
||||
}
|
||||
if err := e.Quotas.Validate(); err != nil {
|
||||
return err
|
||||
}
|
||||
if err := validateStressWarmups(e.Warmups, profile.AgentCount()); err != nil {
|
||||
return err
|
||||
}
|
||||
if err := validateStressSessions(e.Sessions, canonical.Sessions, profile.ConcurrentSessions()); err != nil {
|
||||
return err
|
||||
}
|
||||
if len(e.Iterations) != profile.Iterations() {
|
||||
return fmt.Errorf("iterations=%d want=%d: %w", len(e.Iterations), profile.Iterations(), ErrStressEvidence)
|
||||
}
|
||||
failed := 0
|
||||
for index, iteration := range e.Iterations {
|
||||
count, err := validateStressIteration(canonical, iteration, index+1, faultAware)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
failed += count
|
||||
}
|
||||
if err := validateStressFault(e.FaultTarget, failed, faultAware); err != nil {
|
||||
return err
|
||||
}
|
||||
if profile.Iterations() == 3 {
|
||||
if err := ValidateStressSoakTrendForProfile(profile, e.SoakTrend); err != nil {
|
||||
return err
|
||||
}
|
||||
}
|
||||
if !e.Cleanup.Passed || e.Cleanup.ReceiptCount != 9 || e.Cleanup.FailedReceiptCount != 0 || e.Cleanup.ForcedCleanupCount != 0 || e.Cleanup.ProcessResidue != 0 || e.Cleanup.ProcessGroupResidue != 0 || e.Cleanup.WorkspaceResidue != 0 {
|
||||
return fmt.Errorf("cleanup=%+v: %w", e.Cleanup, ErrStressEvidence)
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
func (q StressQuotaEvidence) Validate() error {
|
||||
wants := []struct {
|
||||
got StressQuotaBoundary
|
||||
allow int
|
||||
reject int
|
||||
}{{q.PATSecond, 10, 11}, {q.PATMinute, 120, 121}, {q.UserStreams, 20, 21}, {q.ServerStreams, 40, 41}}
|
||||
for _, want := range wants {
|
||||
if want.got.Allowed != want.allow || want.got.Rejected != want.reject || !want.got.AllowedAccepted || !want.got.RejectedDenied {
|
||||
return fmt.Errorf("quota=%+v want=%d/%d: %w", want.got, want.allow, want.reject, ErrStressEvidence)
|
||||
}
|
||||
}
|
||||
if q.PathLockStripes != 1024 {
|
||||
return fmt.Errorf("path lock stripes=%d: %w", q.PathLockStripes, ErrStressEvidence)
|
||||
}
|
||||
return nil
|
||||
}
|
||||
@@ -0,0 +1,147 @@
|
||||
//go:build linux
|
||||
|
||||
package scenario
|
||||
|
||||
import (
|
||||
"testing"
|
||||
"time"
|
||||
|
||||
"github.com/stretchr/testify/require"
|
||||
|
||||
"github.com/nezhahq/nezha/integration/agentcompat/internal/contract"
|
||||
)
|
||||
|
||||
func TestStressEvidence_AcceptsCompleteSuccessContract(t *testing.T) {
|
||||
profile, err := contract.ProfileByName(string(contract.ProfilePRFull))
|
||||
require.NoError(t, err)
|
||||
evidence := validStressEvidence(t, profile)
|
||||
|
||||
err = evidence.ValidateSuccess(profile)
|
||||
|
||||
require.NoError(t, err)
|
||||
}
|
||||
|
||||
func TestStressEvidence_AcceptsOnlyExactStressWorkerFault(t *testing.T) {
|
||||
profile, err := contract.ProfileByName(string(contract.ProfilePRFull))
|
||||
require.NoError(t, err)
|
||||
evidence := validStressEvidence(t, profile)
|
||||
target := StressWorkerFaultTarget()
|
||||
evidence.FaultTarget = &target
|
||||
faultOperation := findStressFaultOperation(evidence.Plan)
|
||||
for roundIndex := range evidence.Iterations[0].Rounds {
|
||||
for operationIndex := range evidence.Iterations[0].Rounds[roundIndex].Operations {
|
||||
operation := &evidence.Iterations[0].Rounds[roundIndex].Operations[operationIndex]
|
||||
if operation.ID == faultOperation.ID {
|
||||
operation.Succeeded = false
|
||||
operation.Error = "injected stress worker fault"
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
err = evidence.ValidateStressWorker(profile)
|
||||
|
||||
require.NoError(t, err)
|
||||
}
|
||||
|
||||
func TestStressEvidence_RejectsSecondFailedOperationForStressWorker(t *testing.T) {
|
||||
profile, err := contract.ProfileByName(string(contract.ProfilePRFull))
|
||||
require.NoError(t, err)
|
||||
evidence := validStressEvidence(t, profile)
|
||||
target := StressWorkerFaultTarget()
|
||||
evidence.FaultTarget = &target
|
||||
evidence.Iterations[0].Rounds[1].Operations[0].Succeeded = false
|
||||
evidence.Iterations[0].Rounds[1].Operations[0].Error = "injected stress worker fault"
|
||||
evidence.Iterations[0].Rounds[1].Operations[1].Succeeded = false
|
||||
evidence.Iterations[0].Rounds[1].Operations[1].Error = "unexpected failure"
|
||||
|
||||
err = evidence.ValidateStressWorker(profile)
|
||||
|
||||
require.ErrorIs(t, err, ErrStressFault)
|
||||
}
|
||||
|
||||
func TestStressEvidence_RejectsSelfConsistentTruncatedPlan(t *testing.T) {
|
||||
profile, err := contract.ProfileByName(string(contract.ProfilePRFull))
|
||||
require.NoError(t, err)
|
||||
evidence := validStressEvidence(t, profile)
|
||||
evidence.Plan.Rounds = evidence.Plan.Rounds[:1]
|
||||
evidence.Iterations[0].Rounds = evidence.Iterations[0].Rounds[:1]
|
||||
|
||||
err = evidence.ValidateSuccess(profile)
|
||||
|
||||
require.ErrorIs(t, err, ErrStressEvidence)
|
||||
}
|
||||
|
||||
func TestStressEvidence_RejectsDuplicateDashboardResources(t *testing.T) {
|
||||
profile, err := contract.ProfileByName(string(contract.ProfilePRFull))
|
||||
require.NoError(t, err)
|
||||
evidence := validStressEvidence(t, profile)
|
||||
evidence.Iterations[0].Resources[1] = evidence.Iterations[0].Resources[0]
|
||||
|
||||
err = evidence.ValidateSuccess(profile)
|
||||
|
||||
require.ErrorIs(t, err, ErrStressEvidence)
|
||||
}
|
||||
|
||||
func validStressEvidence(t *testing.T, profile contract.Profile) StressEvidence {
|
||||
t.Helper()
|
||||
plan, err := GenerateStressPlan(profile, contract.DefaultSeed)
|
||||
require.NoError(t, err)
|
||||
warmups := make([]StressWarmupEvidence, profile.AgentCount())
|
||||
for index := range warmups {
|
||||
agent, agentErr := NewStressAgentOrdinal(index + 1)
|
||||
require.NoError(t, agentErr)
|
||||
warmups[index] = StressWarmupEvidence{Agent: agent, Exec: true, Filesystem: true, Terminal: true, NAT: true, FM: true}
|
||||
}
|
||||
sessions := make([]StressSessionEvidence, len(plan.Sessions))
|
||||
for index, session := range plan.Sessions {
|
||||
sessions[index] = StressSessionEvidence{ID: session.ID, Kind: session.Kind, Succeeded: true}
|
||||
}
|
||||
rounds := make([]StressRoundEvidence, len(plan.Rounds))
|
||||
started := time.Unix(100, 0)
|
||||
for roundIndex, round := range plan.Rounds {
|
||||
operations := make([]StressOperationEvidence, len(round.Operations))
|
||||
for operationIndex, operation := range round.Operations {
|
||||
operations[operationIndex] = StressOperationEvidence{ID: operation.ID, Round: operation.Round, Agent: operation.Agent, PAT: operation.PAT, Kind: operation.Kind, LaunchedAt: started, CompletedAt: started.Add(time.Millisecond), Succeeded: true, SuccessProof: "ok"}
|
||||
}
|
||||
rounds[roundIndex] = StressRoundEvidence{Round: round.Round, Operations: operations}
|
||||
}
|
||||
resources := make([]StressProcessWindows, 0, profile.AgentCount()+1)
|
||||
resources = append(resources, stressDashboardResourceFixture(100))
|
||||
for index := 1; index <= profile.AgentCount(); index++ {
|
||||
agent, agentErr := NewStressAgentOrdinal(index)
|
||||
require.NoError(t, agentErr)
|
||||
process, processErr := NewStressAgentProcess(agent, 200+index)
|
||||
require.NoError(t, processErr)
|
||||
resources = append(resources, StressProcessWindows{Process: process, Baseline: stressWindow(200+index, 100), End: stressWindow(200+index, 100)})
|
||||
}
|
||||
iterations := make([]StressIterationEvidence, profile.Iterations())
|
||||
for index := range iterations {
|
||||
iterations[index] = StressIterationEvidence{Iteration: index + 1, Rounds: rounds, Resources: resources}
|
||||
}
|
||||
return StressEvidence{
|
||||
Version: 1, Profile: profile.Name(), Seed: contract.DefaultSeed, Plan: plan,
|
||||
PreparedBinaries: StressPreparedBinaries{DashboardBuildCount: 1, DashboardPathReused: true, AgentBuildCount: 1, AgentPathReused: true},
|
||||
Quotas: StressQuotaEvidence{
|
||||
PATSecond: quotaBoundary(10, 11), PATMinute: quotaBoundary(120, 121),
|
||||
UserStreams: quotaBoundary(20, 21), ServerStreams: quotaBoundary(40, 41), PathLockStripes: 1024,
|
||||
},
|
||||
Warmups: warmups, Sessions: sessions, Iterations: iterations,
|
||||
Cleanup: StressCleanupSummary{Passed: true, ReceiptCount: 9},
|
||||
}
|
||||
}
|
||||
|
||||
func quotaBoundary(allowed, rejected int) StressQuotaBoundary {
|
||||
return StressQuotaBoundary{Allowed: allowed, Rejected: rejected, AllowedAccepted: true, RejectedDenied: true}
|
||||
}
|
||||
|
||||
func findStressFaultOperation(plan StressPlan) StressOperationPlan {
|
||||
target := StressWorkerFaultTarget()
|
||||
for _, round := range plan.Rounds {
|
||||
for _, operation := range round.Operations {
|
||||
if round.Round == target.Round && operation.Agent == target.Agent && operation.Kind == target.Kind {
|
||||
return operation
|
||||
}
|
||||
}
|
||||
}
|
||||
panic("stress fault operation missing")
|
||||
}
|
||||
@@ -0,0 +1,154 @@
|
||||
//go:build linux
|
||||
|
||||
package scenario
|
||||
|
||||
import "fmt"
|
||||
|
||||
func validateStressPreparedBinaries(evidence StressPreparedBinaries) error {
|
||||
if evidence.DashboardBuildCount != 1 || !evidence.DashboardPathReused || evidence.AgentBuildCount != 1 || !evidence.AgentPathReused {
|
||||
return fmt.Errorf("prepared binaries=%+v: %w", evidence, ErrStressEvidence)
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
func validateStressWarmups(warmups []StressWarmupEvidence, agentCount int) error {
|
||||
if len(warmups) != agentCount {
|
||||
return fmt.Errorf("warmups=%d want=%d: %w", len(warmups), agentCount, ErrStressEvidence)
|
||||
}
|
||||
seen := make(map[int]struct{}, len(warmups))
|
||||
for _, warmup := range warmups {
|
||||
if _, duplicate := seen[warmup.Agent.Int()]; duplicate || !warmup.Exec || !warmup.Filesystem || !warmup.Terminal || !warmup.NAT || !warmup.FM {
|
||||
return fmt.Errorf("warmup=%+v: %w", warmup, ErrStressEvidence)
|
||||
}
|
||||
seen[warmup.Agent.Int()] = struct{}{}
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
func validateStressSessions(sessions []StressSessionEvidence, plan []StressSessionPlan, countPerKind int) error {
|
||||
if len(sessions) != countPerKind*3 {
|
||||
return fmt.Errorf("sessions=%d want=%d: %w", len(sessions), countPerKind*3, ErrStressEvidence)
|
||||
}
|
||||
if len(plan) != len(sessions) {
|
||||
return fmt.Errorf("session plan=%d evidence=%d: %w", len(plan), len(sessions), ErrStressEvidence)
|
||||
}
|
||||
expected := make(map[StressSessionID]StressSessionPlan, len(plan))
|
||||
for _, session := range plan {
|
||||
expected[session.ID] = session
|
||||
}
|
||||
counts := make(map[StressSessionKind]int, 3)
|
||||
seen := make(map[StressSessionID]struct{}, len(sessions))
|
||||
for _, session := range sessions {
|
||||
planned, exists := expected[session.ID]
|
||||
_, duplicate := seen[session.ID]
|
||||
if !exists || planned.Kind != session.Kind || duplicate || !session.Succeeded {
|
||||
return fmt.Errorf("session=%+v: %w", session, ErrStressEvidence)
|
||||
}
|
||||
seen[session.ID] = struct{}{}
|
||||
counts[session.Kind]++
|
||||
}
|
||||
for _, kind := range []StressSessionKind{StressSessionTerminal, StressSessionNAT, StressSessionFM} {
|
||||
if counts[kind] != countPerKind {
|
||||
return fmt.Errorf("session kind=%s count=%d want=%d: %w", kind, counts[kind], countPerKind, ErrStressEvidence)
|
||||
}
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
func validateStressIteration(plan StressPlan, evidence StressIterationEvidence, iteration int, faultAware bool) (int, error) {
|
||||
if evidence.Iteration != iteration || len(evidence.Rounds) != len(plan.Rounds) || len(evidence.Resources) != 1+canonicalAgentCount(plan) {
|
||||
return 0, fmt.Errorf("iteration=%+v: %w", evidence, ErrStressEvidence)
|
||||
}
|
||||
failed := 0
|
||||
for index, round := range evidence.Rounds {
|
||||
matched, err := matchStressRoundEvidence(plan.Rounds[index], round)
|
||||
if err != nil {
|
||||
return 0, err
|
||||
}
|
||||
for _, operation := range matched {
|
||||
if !operation.Succeeded || operation.Error != "" {
|
||||
failed++
|
||||
if !faultAware || !isStressFaultOperation(plan.Rounds[index], operation, iteration) {
|
||||
return 0, fmt.Errorf("unexpected failed operation=%s: %w", operation.ID.String(), ErrStressFault)
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
if err := validateStressResourceIdentities(plan, evidence.Resources); err != nil {
|
||||
return 0, err
|
||||
}
|
||||
for _, windows := range evidence.Resources {
|
||||
if _, err := EvaluateStressResource(windows); err != nil {
|
||||
return 0, err
|
||||
}
|
||||
}
|
||||
return failed, nil
|
||||
}
|
||||
|
||||
func validateStressFault(target *StressFaultTarget, failed int, faultAware bool) error {
|
||||
if !faultAware {
|
||||
if target != nil || failed != 0 {
|
||||
return ErrStressFault
|
||||
}
|
||||
return nil
|
||||
}
|
||||
want := StressWorkerFaultTarget()
|
||||
if target == nil || *target != want || failed != 1 {
|
||||
return fmt.Errorf("target=%+v failed=%d want=%+v/1: %w", target, failed, want, ErrStressFault)
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
func isStressFaultOperation(plan StressRoundPlan, evidence StressOperationEvidence, iteration int) bool {
|
||||
target := StressWorkerFaultTarget()
|
||||
if iteration != target.Iteration || plan.Round != target.Round {
|
||||
return false
|
||||
}
|
||||
for _, operation := range plan.Operations {
|
||||
if operation.ID == evidence.ID {
|
||||
return operation.Agent == target.Agent && operation.Kind == target.Kind
|
||||
}
|
||||
}
|
||||
return false
|
||||
}
|
||||
|
||||
func planAgentCount(plan StressPlan) int {
|
||||
return canonicalAgentCount(plan)
|
||||
}
|
||||
|
||||
func canonicalAgentCount(plan StressPlan) int {
|
||||
if len(plan.Rounds) == 0 {
|
||||
return 0
|
||||
}
|
||||
return len(plan.Rounds[0].Operations) / 2
|
||||
}
|
||||
|
||||
func validateStressResourceIdentities(plan StressPlan, resources []StressProcessWindows) error {
|
||||
wantAgents := planAgentCount(plan)
|
||||
if len(resources) != wantAgents+1 {
|
||||
return fmt.Errorf("resources=%d want=%d: %w", len(resources), wantAgents+1, ErrStressEvidence)
|
||||
}
|
||||
dashboardCount := 0
|
||||
seenAgents := make(map[int]struct{}, wantAgents)
|
||||
for _, resource := range resources {
|
||||
switch resource.Process.Kind {
|
||||
case StressProcessDashboard:
|
||||
dashboardCount++
|
||||
case StressProcessAgent:
|
||||
ordinal := resource.Process.Agent.Int()
|
||||
if ordinal < 1 || ordinal > wantAgents {
|
||||
return fmt.Errorf("unknown agent ordinal=%d: %w", ordinal, ErrStressEvidence)
|
||||
}
|
||||
if _, duplicate := seenAgents[ordinal]; duplicate {
|
||||
return fmt.Errorf("duplicate agent ordinal=%d: %w", ordinal, ErrStressEvidence)
|
||||
}
|
||||
seenAgents[ordinal] = struct{}{}
|
||||
default:
|
||||
return fmt.Errorf("unknown process kind=%q: %w", resource.Process.Kind, ErrStressEvidence)
|
||||
}
|
||||
}
|
||||
if dashboardCount != 1 || len(seenAgents) != wantAgents {
|
||||
return fmt.Errorf("dashboard=%d agents=%d want=1/%d: %w", dashboardCount, len(seenAgents), wantAgents, ErrStressEvidence)
|
||||
}
|
||||
return nil
|
||||
}
|
||||
@@ -0,0 +1,100 @@
|
||||
//go:build linux
|
||||
|
||||
package scenario
|
||||
|
||||
import (
|
||||
"errors"
|
||||
"fmt"
|
||||
"sync"
|
||||
)
|
||||
|
||||
var (
|
||||
ErrStressOperationUnknown = errors.New("stress operation is unknown")
|
||||
ErrStressOperationDuplicateStart = errors.New("stress operation started more than once")
|
||||
ErrStressOperationDuplicateCompletion = errors.New("stress operation completed more than once")
|
||||
ErrStressOperationOwnerMismatch = errors.New("stress operation owner does not match plan")
|
||||
ErrStressOperationMissingCompletion = errors.New("stress operation completion is missing")
|
||||
)
|
||||
|
||||
type StressOperationReceipt struct {
|
||||
Operation StressOperationPlan `json:"operation"`
|
||||
SuccessProof string `json:"success_proof"`
|
||||
}
|
||||
|
||||
type stressOperationState struct {
|
||||
plan StressOperationPlan
|
||||
started bool
|
||||
completed bool
|
||||
}
|
||||
|
||||
type stressExactOnceRegistry struct {
|
||||
mu sync.Mutex
|
||||
operations map[StressOperationID]*stressOperationState
|
||||
}
|
||||
|
||||
func newStressExactOnceRegistry(plan StressPlan) (*stressExactOnceRegistry, error) {
|
||||
operations := make(map[StressOperationID]*stressOperationState)
|
||||
for _, round := range plan.Rounds {
|
||||
for _, operation := range round.Operations {
|
||||
if operation.ID.String() == "" {
|
||||
return nil, fmt.Errorf("empty operation ID: %w", ErrStressOperationUnknown)
|
||||
}
|
||||
if _, exists := operations[operation.ID]; exists {
|
||||
return nil, fmt.Errorf("duplicate operation ID %s: %w", operation.ID.String(), ErrStressOperationUnknown)
|
||||
}
|
||||
operations[operation.ID] = &stressOperationState{plan: operation}
|
||||
}
|
||||
}
|
||||
return &stressExactOnceRegistry{operations: operations}, nil
|
||||
}
|
||||
|
||||
func (registry *stressExactOnceRegistry) Start(operation StressOperationPlan) error {
|
||||
registry.mu.Lock()
|
||||
defer registry.mu.Unlock()
|
||||
state, exists := registry.operations[operation.ID]
|
||||
if !exists {
|
||||
return ErrStressOperationUnknown
|
||||
}
|
||||
if state.plan != operation {
|
||||
return ErrStressOperationOwnerMismatch
|
||||
}
|
||||
if state.started {
|
||||
return ErrStressOperationDuplicateStart
|
||||
}
|
||||
state.started = true
|
||||
return nil
|
||||
}
|
||||
|
||||
func (registry *stressExactOnceRegistry) Complete(receipt StressOperationReceipt) error {
|
||||
registry.mu.Lock()
|
||||
defer registry.mu.Unlock()
|
||||
state, exists := registry.operations[receipt.Operation.ID]
|
||||
if !exists {
|
||||
return ErrStressOperationUnknown
|
||||
}
|
||||
if state.plan != receipt.Operation {
|
||||
return ErrStressOperationOwnerMismatch
|
||||
}
|
||||
if !state.started {
|
||||
return ErrStressOperationDuplicateStart
|
||||
}
|
||||
if state.completed {
|
||||
return ErrStressOperationDuplicateCompletion
|
||||
}
|
||||
if receipt.SuccessProof == "" {
|
||||
return ErrStressOperationMissingCompletion
|
||||
}
|
||||
state.completed = true
|
||||
return nil
|
||||
}
|
||||
|
||||
func (registry *stressExactOnceRegistry) ValidateComplete() error {
|
||||
registry.mu.Lock()
|
||||
defer registry.mu.Unlock()
|
||||
for _, state := range registry.operations {
|
||||
if !state.started || !state.completed {
|
||||
return ErrStressOperationMissingCompletion
|
||||
}
|
||||
}
|
||||
return nil
|
||||
}
|
||||
@@ -0,0 +1,43 @@
|
||||
//go:build linux
|
||||
|
||||
package scenario
|
||||
|
||||
import (
|
||||
"testing"
|
||||
|
||||
"github.com/stretchr/testify/require"
|
||||
|
||||
"github.com/nezhahq/nezha/integration/agentcompat/internal/contract"
|
||||
)
|
||||
|
||||
func TestStressExactOnceRegistryRejectsInvalidReceipts(t *testing.T) {
|
||||
profile, err := contract.ProfileByName(string(contract.ProfilePRFull))
|
||||
require.NoError(t, err)
|
||||
plan, err := GenerateStressPlan(profile, contract.DefaultSeed)
|
||||
require.NoError(t, err)
|
||||
registry, err := newStressExactOnceRegistry(plan)
|
||||
require.NoError(t, err)
|
||||
operation := plan.Rounds[0].Operations[0]
|
||||
receipt := StressOperationReceipt{Operation: operation, SuccessProof: "ok"}
|
||||
require.NoError(t, registry.Start(operation))
|
||||
require.ErrorIs(t, registry.Start(operation), ErrStressOperationDuplicateStart)
|
||||
require.NoError(t, registry.Complete(receipt))
|
||||
require.ErrorIs(t, registry.Complete(receipt), ErrStressOperationDuplicateCompletion)
|
||||
require.ErrorIs(t, registry.Start(StressOperationPlan{ID: operation.ID}), ErrStressOperationOwnerMismatch)
|
||||
require.ErrorIs(t, registry.Complete(StressOperationReceipt{Operation: StressOperationPlan{ID: operation.ID, Round: 2, Agent: operation.Agent, PAT: operation.PAT, Kind: operation.Kind}, SuccessProof: "ok"}), ErrStressOperationOwnerMismatch)
|
||||
}
|
||||
|
||||
func TestStressExactOnceRegistryRequiresCanonicalCompletion(t *testing.T) {
|
||||
profile, err := contract.ProfileByName(string(contract.ProfilePRFull))
|
||||
require.NoError(t, err)
|
||||
plan, err := GenerateStressPlan(profile, contract.DefaultSeed)
|
||||
require.NoError(t, err)
|
||||
registry, err := newStressExactOnceRegistry(plan)
|
||||
require.NoError(t, err)
|
||||
for _, round := range plan.Rounds {
|
||||
for _, operation := range round.Operations {
|
||||
require.NoError(t, registry.Start(operation))
|
||||
}
|
||||
}
|
||||
require.ErrorIs(t, registry.ValidateComplete(), ErrStressOperationMissingCompletion)
|
||||
}
|
||||
@@ -0,0 +1,185 @@
|
||||
//go:build linux
|
||||
|
||||
package scenario
|
||||
|
||||
import (
|
||||
"encoding/json"
|
||||
"errors"
|
||||
"fmt"
|
||||
)
|
||||
|
||||
var ErrStressIdentity = errors.New("stress identity is invalid")
|
||||
|
||||
type StressAgentOrdinal struct{ value int }
|
||||
|
||||
func NewStressAgentOrdinal(value int) (StressAgentOrdinal, error) {
|
||||
if value < 1 {
|
||||
return StressAgentOrdinal{}, fmt.Errorf("agent ordinal %d: %w", value, ErrStressIdentity)
|
||||
}
|
||||
return StressAgentOrdinal{value: value}, nil
|
||||
}
|
||||
|
||||
func (ordinal StressAgentOrdinal) Int() int { return ordinal.value }
|
||||
|
||||
func (ordinal StressAgentOrdinal) MarshalJSON() ([]byte, error) { return json.Marshal(ordinal.value) }
|
||||
|
||||
func (ordinal *StressAgentOrdinal) UnmarshalJSON(data []byte) error {
|
||||
var value int
|
||||
if err := json.Unmarshal(data, &value); err != nil || value < 1 {
|
||||
return ErrStressIdentity
|
||||
}
|
||||
ordinal.value = value
|
||||
return nil
|
||||
}
|
||||
|
||||
type StressOperationID struct{ value string }
|
||||
|
||||
func NewStressOperationID(value string) (StressOperationID, error) {
|
||||
if value == "" {
|
||||
return StressOperationID{}, ErrStressIdentity
|
||||
}
|
||||
return StressOperationID{value: value}, nil
|
||||
}
|
||||
|
||||
func (identity StressOperationID) String() string { return identity.value }
|
||||
|
||||
func (identity StressOperationID) MarshalJSON() ([]byte, error) { return json.Marshal(identity.value) }
|
||||
|
||||
func (identity *StressOperationID) UnmarshalJSON(data []byte) error {
|
||||
var value string
|
||||
if err := json.Unmarshal(data, &value); err != nil || value == "" {
|
||||
return ErrStressIdentity
|
||||
}
|
||||
identity.value = value
|
||||
return nil
|
||||
}
|
||||
|
||||
type StressSessionID struct{ value string }
|
||||
|
||||
func NewStressSessionID(value string) (StressSessionID, error) {
|
||||
if value == "" {
|
||||
return StressSessionID{}, ErrStressIdentity
|
||||
}
|
||||
return StressSessionID{value: value}, nil
|
||||
}
|
||||
|
||||
func (identity StressSessionID) String() string { return identity.value }
|
||||
|
||||
func (identity StressSessionID) MarshalJSON() ([]byte, error) { return json.Marshal(identity.value) }
|
||||
|
||||
func (identity *StressSessionID) UnmarshalJSON(data []byte) error {
|
||||
var value string
|
||||
if err := json.Unmarshal(data, &value); err != nil || value == "" {
|
||||
return ErrStressIdentity
|
||||
}
|
||||
identity.value = value
|
||||
return nil
|
||||
}
|
||||
|
||||
type StressPATID struct{ value string }
|
||||
|
||||
func NewStressPATID(value string) (StressPATID, error) {
|
||||
if value == "" {
|
||||
return StressPATID{}, ErrStressIdentity
|
||||
}
|
||||
return StressPATID{value: value}, nil
|
||||
}
|
||||
|
||||
func (identity StressPATID) String() string { return identity.value }
|
||||
|
||||
func (identity StressPATID) MarshalJSON() ([]byte, error) { return json.Marshal(identity.value) }
|
||||
|
||||
func (identity *StressPATID) UnmarshalJSON(data []byte) error {
|
||||
var value string
|
||||
if err := json.Unmarshal(data, &value); err != nil || value == "" {
|
||||
return ErrStressIdentity
|
||||
}
|
||||
identity.value = value
|
||||
return nil
|
||||
}
|
||||
|
||||
type StressOperationKind string
|
||||
|
||||
const (
|
||||
StressOperationExec StressOperationKind = "exec"
|
||||
StressOperationFilesystem StressOperationKind = "filesystem"
|
||||
)
|
||||
|
||||
type StressSessionKind string
|
||||
|
||||
const (
|
||||
StressSessionTerminal StressSessionKind = "terminal"
|
||||
StressSessionNAT StressSessionKind = "nat"
|
||||
StressSessionFM StressSessionKind = "file-manager"
|
||||
)
|
||||
|
||||
type StressProcessKind string
|
||||
|
||||
const (
|
||||
StressProcessDashboard StressProcessKind = "dashboard"
|
||||
StressProcessAgent StressProcessKind = "agent"
|
||||
)
|
||||
|
||||
type StressProcessIdentity struct {
|
||||
Kind StressProcessKind `json:"kind"`
|
||||
Agent StressAgentOrdinal `json:"agent,omitempty"`
|
||||
PID int `json:"pid"`
|
||||
}
|
||||
|
||||
func (identity StressProcessIdentity) MarshalJSON() ([]byte, error) {
|
||||
type processWire struct {
|
||||
Kind StressProcessKind `json:"kind"`
|
||||
Agent *StressAgentOrdinal `json:"agent,omitempty"`
|
||||
PID int `json:"pid"`
|
||||
}
|
||||
var agent *StressAgentOrdinal
|
||||
if identity.Kind == StressProcessAgent {
|
||||
if identity.Agent.Int() < 1 {
|
||||
return nil, ErrStressIdentity
|
||||
}
|
||||
agent = &identity.Agent
|
||||
}
|
||||
return json.Marshal(processWire{Kind: identity.Kind, Agent: agent, PID: identity.PID})
|
||||
}
|
||||
|
||||
func (identity *StressProcessIdentity) UnmarshalJSON(data []byte) error {
|
||||
type processWire struct {
|
||||
Kind StressProcessKind `json:"kind"`
|
||||
Agent *StressAgentOrdinal `json:"agent"`
|
||||
PID int `json:"pid"`
|
||||
}
|
||||
var wire processWire
|
||||
if err := json.Unmarshal(data, &wire); err != nil || wire.PID < 1 {
|
||||
return ErrStressIdentity
|
||||
}
|
||||
switch wire.Kind {
|
||||
case StressProcessDashboard:
|
||||
*identity = StressProcessIdentity{Kind: wire.Kind, PID: wire.PID}
|
||||
case StressProcessAgent:
|
||||
if wire.Agent == nil || wire.Agent.Int() < 1 {
|
||||
return ErrStressIdentity
|
||||
}
|
||||
*identity = StressProcessIdentity{Kind: wire.Kind, Agent: *wire.Agent, PID: wire.PID}
|
||||
default:
|
||||
return ErrStressIdentity
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
func NewStressDashboardProcess(pid int) (StressProcessIdentity, error) {
|
||||
if pid < 1 {
|
||||
return StressProcessIdentity{}, fmt.Errorf("dashboard PID %d: %w", pid, ErrStressIdentity)
|
||||
}
|
||||
return StressProcessIdentity{Kind: StressProcessDashboard, PID: pid}, nil
|
||||
}
|
||||
|
||||
func NewStressAgentProcess(agent StressAgentOrdinal, pid int) (StressProcessIdentity, error) {
|
||||
if agent.Int() < 1 || pid < 1 {
|
||||
return StressProcessIdentity{}, fmt.Errorf("agent process ordinal=%d PID=%d: %w", agent.Int(), pid, ErrStressIdentity)
|
||||
}
|
||||
return StressProcessIdentity{Kind: StressProcessAgent, Agent: agent, PID: pid}, nil
|
||||
}
|
||||
|
||||
func (identity StressProcessIdentity) key() string {
|
||||
return fmt.Sprintf("%s:%d", identity.Kind, identity.Agent.Int())
|
||||
}
|
||||
@@ -0,0 +1,73 @@
|
||||
//go:build linux && agentcompat
|
||||
|
||||
package scenario
|
||||
|
||||
import (
|
||||
"context"
|
||||
"encoding/json"
|
||||
"errors"
|
||||
"fmt"
|
||||
"os"
|
||||
"os/exec"
|
||||
"path/filepath"
|
||||
"strconv"
|
||||
"strings"
|
||||
)
|
||||
|
||||
var ErrStressPathLockProof = errors.New("stress path-lock proof is invalid")
|
||||
|
||||
type stressPathLockProof struct {
|
||||
Stripes int
|
||||
}
|
||||
|
||||
func (proof stressPathLockProof) Validate() error {
|
||||
if proof.Stripes != 1024 {
|
||||
return fmt.Errorf("stripes=%d: %w", proof.Stripes, ErrStressPathLockProof)
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
func proveStressPathLockStripes(ctx context.Context, sourceDir string) (stressPathLockProof, error) {
|
||||
if sourceDir == "" {
|
||||
return stressPathLockProof{}, ErrStressPathLockProof
|
||||
}
|
||||
root, err := os.MkdirTemp("", "nezha-stress-path-lock-")
|
||||
if err != nil {
|
||||
return stressPathLockProof{}, err
|
||||
}
|
||||
defer os.RemoveAll(root)
|
||||
probe := filepath.Join(root, "agentcompat_path_lock_proof_test.go")
|
||||
content := "package main\n\nimport (\n\t\"fmt\"\n\t\"testing\"\n)\n\nconst agentCompatPathLockStripes = fsPathLockStripes\n\nvar _ [agentCompatPathLockStripes - 1024]struct{}\nvar _ [1024 - agentCompatPathLockStripes]struct{}\n\nfunc TestAgentCompatPathLockProof(t *testing.T) {\n\tfmt.Printf(\"AGENTCOMPAT_PATH_LOCK_STRIPES=%d\\n\", agentCompatPathLockStripes)\n}\n"
|
||||
if err := os.WriteFile(probe, []byte(content), 0o600); err != nil {
|
||||
return stressPathLockProof{}, err
|
||||
}
|
||||
overlayPath := filepath.Join(root, "overlay.json")
|
||||
original := filepath.Join(sourceDir, "cmd", "agent", "mcp_fs_path_lock_bounded_test.go")
|
||||
overlayData, err := json.Marshal(struct {
|
||||
Replace map[string]string `json:"Replace"`
|
||||
}{Replace: map[string]string{original: probe}})
|
||||
if err != nil {
|
||||
return stressPathLockProof{}, err
|
||||
}
|
||||
if err := os.WriteFile(overlayPath, overlayData, 0o600); err != nil {
|
||||
return stressPathLockProof{}, err
|
||||
}
|
||||
command := exec.CommandContext(ctx, "go", "test", "-overlay", overlayPath, "./cmd/agent", "-run", "^TestAgentCompatPathLockProof$", "-count=1", "-v")
|
||||
command.Dir = sourceDir
|
||||
command.Env = append(os.Environ(), "GOFLAGS=")
|
||||
output, err := command.CombinedOutput()
|
||||
if err != nil {
|
||||
return stressPathLockProof{}, fmt.Errorf("compile path-lock proof: %w: %s", err, output)
|
||||
}
|
||||
for _, line := range strings.Split(string(output), "\n") {
|
||||
if !strings.HasPrefix(line, "AGENTCOMPAT_PATH_LOCK_STRIPES=") {
|
||||
continue
|
||||
}
|
||||
value, parseErr := strconv.Atoi(strings.TrimPrefix(line, "AGENTCOMPAT_PATH_LOCK_STRIPES="))
|
||||
if parseErr == nil {
|
||||
proof := stressPathLockProof{Stripes: value}
|
||||
return proof, proof.Validate()
|
||||
}
|
||||
}
|
||||
return stressPathLockProof{}, fmt.Errorf("path-lock proof output missing: %w", ErrStressPathLockProof)
|
||||
}
|
||||
@@ -0,0 +1,35 @@
|
||||
//go:build linux && agentcompat
|
||||
|
||||
package scenario
|
||||
|
||||
import (
|
||||
"crypto/sha256"
|
||||
"os"
|
||||
"testing"
|
||||
|
||||
"github.com/stretchr/testify/require"
|
||||
)
|
||||
|
||||
func TestStressPathLockProofRequiresExactStripeCount(t *testing.T) {
|
||||
for _, stripes := range []int{1023, 1025} {
|
||||
require.ErrorIs(t, (stressPathLockProof{Stripes: stripes}).Validate(), ErrStressPathLockProof)
|
||||
}
|
||||
require.NoError(t, (stressPathLockProof{Stripes: 1024}).Validate())
|
||||
}
|
||||
|
||||
func TestStressPathLockProofDoesNotModifyAgentSource(t *testing.T) {
|
||||
sourceDir := os.Getenv("AGENTCOMPAT_AGENT_SOURCE")
|
||||
if sourceDir == "" {
|
||||
t.Skip("AGENTCOMPAT_AGENT_SOURCE is not configured")
|
||||
}
|
||||
path := sourceDir + "/cmd/agent/mcp_fs_path_lock.go"
|
||||
before, err := os.ReadFile(path)
|
||||
require.NoError(t, err)
|
||||
beforeHash := sha256.Sum256(before)
|
||||
proof, err := proveStressPathLockStripes(t.Context(), sourceDir)
|
||||
require.NoError(t, err)
|
||||
require.NoError(t, proof.Validate())
|
||||
after, err := os.ReadFile(path)
|
||||
require.NoError(t, err)
|
||||
require.Equal(t, beforeHash, sha256.Sum256(after))
|
||||
}
|
||||
@@ -0,0 +1,120 @@
|
||||
//go:build linux
|
||||
|
||||
package scenario
|
||||
|
||||
import (
|
||||
"errors"
|
||||
"fmt"
|
||||
|
||||
"github.com/nezhahq/nezha/integration/agentcompat/internal/contract"
|
||||
)
|
||||
|
||||
var ErrStressPlan = errors.New("stress plan is invalid")
|
||||
|
||||
type StressOperationPlan struct {
|
||||
ID StressOperationID `json:"id"`
|
||||
Round int `json:"round"`
|
||||
Agent StressAgentOrdinal `json:"agent"`
|
||||
PAT StressPATID `json:"pat_id"`
|
||||
Kind StressOperationKind `json:"kind"`
|
||||
}
|
||||
|
||||
type StressRoundPlan struct {
|
||||
Round int `json:"round"`
|
||||
Operations []StressOperationPlan `json:"operations"`
|
||||
}
|
||||
|
||||
type StressSessionPlan struct {
|
||||
ID StressSessionID `json:"id"`
|
||||
Kind StressSessionKind `json:"kind"`
|
||||
Ordinal int `json:"ordinal"`
|
||||
Agent StressAgentOrdinal `json:"agent"`
|
||||
}
|
||||
|
||||
type StressPlan struct {
|
||||
Profile contract.ProfileName `json:"profile"`
|
||||
Seed contract.Seed `json:"seed"`
|
||||
Rounds []StressRoundPlan `json:"rounds"`
|
||||
Sessions []StressSessionPlan `json:"sessions"`
|
||||
}
|
||||
|
||||
func GenerateStressPlan(profile contract.Profile, seed contract.Seed) (StressPlan, error) {
|
||||
if seed == 0 || profile.AgentCount() < 1 || profile.StressRounds() < 1 || profile.ConcurrentSessions() < 1 {
|
||||
return StressPlan{}, ErrStressPlan
|
||||
}
|
||||
expectedOperations := profile.AgentCount() * profile.StressRounds() * 2
|
||||
if profile.ConcurrentOperations() != expectedOperations {
|
||||
return StressPlan{}, fmt.Errorf("concurrent operations=%d want=%d: %w", profile.ConcurrentOperations(), expectedOperations, ErrStressPlan)
|
||||
}
|
||||
plan := StressPlan{Profile: profile.Name(), Seed: seed}
|
||||
plan.Rounds = make([]StressRoundPlan, profile.StressRounds())
|
||||
for roundIndex := range plan.Rounds {
|
||||
round := roundIndex + 1
|
||||
operations, err := stressRoundOperations(seed, round, profile.AgentCount())
|
||||
if err != nil {
|
||||
return StressPlan{}, err
|
||||
}
|
||||
plan.Rounds[roundIndex] = StressRoundPlan{Round: round, Operations: operations}
|
||||
}
|
||||
sessions, err := stressSessionPlans(seed, profile.AgentCount(), profile.ConcurrentSessions())
|
||||
if err != nil {
|
||||
return StressPlan{}, err
|
||||
}
|
||||
plan.Sessions = sessions
|
||||
return plan, nil
|
||||
}
|
||||
|
||||
func stressRoundOperations(seed contract.Seed, round, agentCount int) ([]StressOperationPlan, error) {
|
||||
operations := make([]StressOperationPlan, 0, agentCount*2)
|
||||
for agentValue := 1; agentValue <= agentCount; agentValue++ {
|
||||
agent, err := NewStressAgentOrdinal(agentValue)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
pat, err := NewStressPATID(fmt.Sprintf("pat-%016x-a%02d", uint64(seed), agentValue))
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
for _, kind := range []StressOperationKind{StressOperationExec, StressOperationFilesystem} {
|
||||
identity, idErr := NewStressOperationID(fmt.Sprintf("op-%016x-r%02d-a%02d-%s", uint64(seed), round, agentValue, kind))
|
||||
if idErr != nil {
|
||||
return nil, idErr
|
||||
}
|
||||
operations = append(operations, StressOperationPlan{ID: identity, Round: round, Agent: agent, PAT: pat, Kind: kind})
|
||||
}
|
||||
}
|
||||
random := stressRandom(uint64(seed) ^ uint64(round)*0x9e3779b97f4a7c15)
|
||||
for index := len(operations) - 1; index > 0; index-- {
|
||||
swap := int(random.next() % uint64(index+1))
|
||||
operations[index], operations[swap] = operations[swap], operations[index]
|
||||
}
|
||||
return operations, nil
|
||||
}
|
||||
|
||||
func stressSessionPlans(seed contract.Seed, agentCount, countPerKind int) ([]StressSessionPlan, error) {
|
||||
sessions := make([]StressSessionPlan, 0, countPerKind*3)
|
||||
for _, kind := range []StressSessionKind{StressSessionTerminal, StressSessionNAT, StressSessionFM} {
|
||||
for index := 1; index <= countPerKind; index++ {
|
||||
agent, err := NewStressAgentOrdinal((index-1)%agentCount + 1)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
identity, err := NewStressSessionID(fmt.Sprintf("session-%016x-%s-%02d", uint64(seed), kind, index))
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
sessions = append(sessions, StressSessionPlan{ID: identity, Kind: kind, Ordinal: index, Agent: agent})
|
||||
}
|
||||
}
|
||||
return sessions, nil
|
||||
}
|
||||
|
||||
type stressRandom uint64
|
||||
|
||||
func (random *stressRandom) next() uint64 {
|
||||
*random += 0x9e3779b97f4a7c15
|
||||
value := uint64(*random)
|
||||
value = (value ^ (value >> 30)) * 0xbf58476d1ce4e5b9
|
||||
value = (value ^ (value >> 27)) * 0x94d049bb133111eb
|
||||
return value ^ (value >> 31)
|
||||
}
|
||||
@@ -0,0 +1,109 @@
|
||||
//go:build linux
|
||||
|
||||
package scenario
|
||||
|
||||
import (
|
||||
"encoding/json"
|
||||
"testing"
|
||||
"time"
|
||||
|
||||
"github.com/stretchr/testify/require"
|
||||
|
||||
"github.com/nezhahq/nezha/integration/agentcompat/internal/contract"
|
||||
)
|
||||
|
||||
func TestStress_PlanHasFourRoundsSixteenOpsEach(t *testing.T) {
|
||||
profile, err := contract.ProfileByName(string(contract.ProfilePRFull))
|
||||
require.NoError(t, err)
|
||||
|
||||
plan, err := GenerateStressPlan(profile, contract.DefaultSeed)
|
||||
|
||||
require.NoError(t, err)
|
||||
require.Len(t, plan.Rounds, 4)
|
||||
for _, round := range plan.Rounds {
|
||||
require.Len(t, round.Operations, 16)
|
||||
}
|
||||
}
|
||||
|
||||
func TestStress_PlanUsesTwoRequestsPerPATPerRound(t *testing.T) {
|
||||
profile, err := contract.ProfileByName(string(contract.ProfilePRFull))
|
||||
require.NoError(t, err)
|
||||
plan, err := GenerateStressPlan(profile, contract.DefaultSeed)
|
||||
require.NoError(t, err)
|
||||
|
||||
for _, round := range plan.Rounds {
|
||||
counts := make(map[StressPATID]int)
|
||||
kinds := make(map[StressPATID]map[StressOperationKind]int)
|
||||
for _, operation := range round.Operations {
|
||||
counts[operation.PAT]++
|
||||
if kinds[operation.PAT] == nil {
|
||||
kinds[operation.PAT] = make(map[StressOperationKind]int)
|
||||
}
|
||||
kinds[operation.PAT][operation.Kind]++
|
||||
}
|
||||
require.Len(t, counts, 8)
|
||||
for pat, count := range counts {
|
||||
require.Equal(t, 2, count)
|
||||
require.Equal(t, 1, kinds[pat][StressOperationExec])
|
||||
require.Equal(t, 1, kinds[pat][StressOperationFilesystem])
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
func TestStress_PlanHas64UniqueStableIDs(t *testing.T) {
|
||||
profile, err := contract.ProfileByName(string(contract.ProfilePRFull))
|
||||
require.NoError(t, err)
|
||||
plan, err := GenerateStressPlan(profile, contract.DefaultSeed)
|
||||
require.NoError(t, err)
|
||||
|
||||
seen := make(map[StressOperationID]struct{})
|
||||
for _, round := range plan.Rounds {
|
||||
for _, operation := range round.Operations {
|
||||
_, duplicate := seen[operation.ID]
|
||||
require.False(t, duplicate)
|
||||
seen[operation.ID] = struct{}{}
|
||||
}
|
||||
}
|
||||
require.Len(t, seen, 64)
|
||||
}
|
||||
|
||||
func TestStress_PlanFixedSeedReproducesByteForByte(t *testing.T) {
|
||||
profile, err := contract.ProfileByName(string(contract.ProfilePRFull))
|
||||
require.NoError(t, err)
|
||||
first, err := GenerateStressPlan(profile, contract.DefaultSeed)
|
||||
require.NoError(t, err)
|
||||
second, err := GenerateStressPlan(profile, contract.DefaultSeed)
|
||||
require.NoError(t, err)
|
||||
|
||||
firstJSON, err := json.Marshal(first)
|
||||
require.NoError(t, err)
|
||||
secondJSON, err := json.Marshal(second)
|
||||
require.NoError(t, err)
|
||||
require.Equal(t, firstJSON, secondJSON)
|
||||
}
|
||||
|
||||
func TestStress_RejectsLaunchWindowOverOneSecond(t *testing.T) {
|
||||
profile, err := contract.ProfileByName(string(contract.ProfilePRFull))
|
||||
require.NoError(t, err)
|
||||
plan, err := GenerateStressPlan(profile, contract.DefaultSeed)
|
||||
require.NoError(t, err)
|
||||
evidence := successfulStressRoundEvidence(plan.Rounds[0])
|
||||
evidence.Operations[len(evidence.Operations)-1].LaunchedAt = evidence.Operations[0].LaunchedAt.Add(time.Second + time.Nanosecond)
|
||||
evidence.Operations[len(evidence.Operations)-1].CompletedAt = evidence.Operations[len(evidence.Operations)-1].LaunchedAt.Add(time.Millisecond)
|
||||
|
||||
err = ValidateStressRoundEvidence(plan.Rounds[0], evidence)
|
||||
|
||||
require.ErrorIs(t, err, ErrStressLaunchWindow)
|
||||
}
|
||||
|
||||
func successfulStressRoundEvidence(plan StressRoundPlan) StressRoundEvidence {
|
||||
started := time.Unix(100, 0)
|
||||
operations := make([]StressOperationEvidence, len(plan.Operations))
|
||||
for index, operation := range plan.Operations {
|
||||
operations[index] = StressOperationEvidence{
|
||||
ID: operation.ID, Round: operation.Round, Agent: operation.Agent, PAT: operation.PAT, Kind: operation.Kind, SuccessProof: "ok",
|
||||
LaunchedAt: started, CompletedAt: started.Add(time.Millisecond), Succeeded: true,
|
||||
}
|
||||
}
|
||||
return StressRoundEvidence{Round: plan.Round, Operations: operations}
|
||||
}
|
||||
@@ -0,0 +1,36 @@
|
||||
//go:build linux
|
||||
|
||||
package scenario
|
||||
|
||||
import (
|
||||
"encoding/json"
|
||||
"os"
|
||||
"path/filepath"
|
||||
"testing"
|
||||
|
||||
"github.com/stretchr/testify/require"
|
||||
|
||||
"github.com/nezhahq/nezha/integration/agentcompat/internal/contract"
|
||||
)
|
||||
|
||||
func TestStress_WriteTypedQAArtifact(t *testing.T) {
|
||||
profile, err := contract.ProfileByName(string(contract.ProfilePRFull))
|
||||
require.NoError(t, err)
|
||||
plan, err := GenerateStressPlan(profile, contract.DefaultSeed)
|
||||
require.NoError(t, err)
|
||||
fixture := stressDashboardResourceFixture(100)
|
||||
evaluation, err := EvaluateStressResource(fixture)
|
||||
require.NoError(t, err)
|
||||
artifact := struct {
|
||||
Plan StressPlan `json:"plan"`
|
||||
Resource StressResourceEvaluation `json:"resource"`
|
||||
}{Plan: plan, Resource: evaluation}
|
||||
data, err := json.MarshalIndent(artifact, "", " ")
|
||||
require.NoError(t, err)
|
||||
path := filepath.Join(t.TempDir(), "stress-contracts.json")
|
||||
require.NoError(t, os.WriteFile(path, data, 0o600))
|
||||
require.Contains(t, string(data), `"profile": "pr-full"`)
|
||||
require.Contains(t, string(data), `"rounds"`)
|
||||
require.Contains(t, string(data), `"rss_limit_bytes": 67108864`)
|
||||
require.NotEmpty(t, path)
|
||||
}
|
||||
@@ -0,0 +1,197 @@
|
||||
//go:build linux && agentcompat
|
||||
|
||||
package scenario
|
||||
|
||||
import (
|
||||
"context"
|
||||
"net/http"
|
||||
"os"
|
||||
"path/filepath"
|
||||
"testing"
|
||||
|
||||
"github.com/stretchr/testify/require"
|
||||
|
||||
"github.com/nezhahq/nezha/integration/agentcompat/internal/agent"
|
||||
"github.com/nezhahq/nezha/integration/agentcompat/internal/client"
|
||||
"github.com/nezhahq/nezha/integration/agentcompat/internal/contract"
|
||||
"github.com/nezhahq/nezha/integration/agentcompat/internal/dashboard"
|
||||
processharness "github.com/nezhahq/nezha/integration/agentcompat/internal/process"
|
||||
)
|
||||
|
||||
func TestStressPRFullEightAgentExactlyOnce(t *testing.T) {
|
||||
requireHeldRealSources(t)
|
||||
paths, err := contract.NewPaths(os.Getenv("AGENTCOMPAT_NEZHA_SOURCE"), os.Getenv("AGENTCOMPAT_AGENT_SOURCE"), filepath.Join("/tmp", "nezha-agentcompat-real-stress"))
|
||||
require.NoError(t, err)
|
||||
profile, err := contract.ProfileByName(string(contract.ProfilePRFull))
|
||||
require.NoError(t, err)
|
||||
plan, err := GenerateStressPlan(profile, contract.DefaultSeed)
|
||||
require.NoError(t, err)
|
||||
ctx, cancel := context.WithTimeout(t.Context(), contract.PRFullSuiteDeadline)
|
||||
defer cancel()
|
||||
realFixture, err := startHeldSessionSetRealFixture(ctx, paths, plan)
|
||||
require.NoError(t, err)
|
||||
t.Cleanup(func() { _ = realFixture.close(context.Background(), nil) })
|
||||
input, err := realFixture.input(plan)
|
||||
require.NoError(t, err)
|
||||
set, err := NewHeldSessionSet(ctx, input)
|
||||
require.NoError(t, err)
|
||||
dashboardIdentity := realFixture.dashboard.RuntimeIdentity()
|
||||
agentIdentities := make([]agent.ProcessIdentity, len(realFixture.agents))
|
||||
workspaceRoots := make([]string, 0, len(realFixture.agents)+2)
|
||||
workspaceRoots = append(workspaceRoots, realFixture.dashboard.WorkspaceRoot(), realFixture.preparedBinary.WorkspaceRoot())
|
||||
for index, instance := range realFixture.agents {
|
||||
agentIdentities[index] = instance.RuntimeIdentity()
|
||||
workspaceRoots = append(workspaceRoots, instance.WorkspaceRoot())
|
||||
}
|
||||
t.Cleanup(func() { _ = set.Close(context.Background()) })
|
||||
warmups, warmupErr := runStressWarmups(ctx, realFixture, plan)
|
||||
require.NoError(t, warmupErr)
|
||||
require.NoError(t, drainStressDashboardSQLiteJournal(ctx, realFixture))
|
||||
baselineResources, resourceErr := captureStressResources(ctx, realFixture, false)
|
||||
require.NoError(t, resourceErr)
|
||||
rounds := make([]StressRoundEvidence, 0, len(plan.Rounds))
|
||||
for _, round := range plan.Rounds {
|
||||
// WaitHealthy completes only after Close; live sessions are validated by NewHeldSessionSet.
|
||||
evidenceValue, roundErr := runStressRound(ctx, realFixture, round)
|
||||
require.NoError(t, roundErr)
|
||||
require.NoError(t, ValidateStressRoundEvidence(round, evidenceValue))
|
||||
rounds = append(rounds, evidenceValue)
|
||||
}
|
||||
quota, quotaErr := runRealStressQuotaProbe(ctx, realFixture, paths.AgentSource().String())
|
||||
require.NoError(t, quotaErr)
|
||||
endResources, resourceErr := captureStressResources(ctx, realFixture, true)
|
||||
require.NoError(t, resourceErr)
|
||||
require.NoError(t, set.Close(ctx))
|
||||
require.NoError(t, set.WaitHealthy(ctx))
|
||||
cleanupErr := realFixture.close(ctx, nil)
|
||||
require.NoError(t, cleanupErr)
|
||||
cleanup := stressCleanupSummary(realFixture, dashboardIdentity, agentIdentities, workspaceRoots, cleanupErr)
|
||||
resourceWindows := make([]StressProcessWindows, len(baselineResources))
|
||||
for index := range baselineResources {
|
||||
resourceWindows[index] = StressProcessWindows{Process: baselineResources[index].Process, Baseline: baselineResources[index].Baseline, End: endResources[index].End}
|
||||
}
|
||||
evidenceValue := StressEvidence{Version: 1, Profile: plan.Profile, Seed: plan.Seed, PreparedBinaries: StressPreparedBinaries{DashboardBuildCount: 1, DashboardPathReused: true, AgentBuildCount: 1, AgentPathReused: true}, Quotas: quota, Warmups: warmups, Plan: plan, Iterations: []StressIterationEvidence{{Iteration: 1, Rounds: rounds, Resources: resourceWindows}}, Cleanup: cleanup}
|
||||
for _, session := range plan.Sessions {
|
||||
evidenceValue.Sessions = append(evidenceValue.Sessions, StressSessionEvidence{ID: session.ID, Kind: session.Kind, Succeeded: true})
|
||||
}
|
||||
require.NoError(t, publishStressEvidence("/tmp/nezha-held-real-sessions", evidenceValue))
|
||||
_, err = readStressEvidence("/tmp/nezha-held-real-sessions")
|
||||
require.NoError(t, err)
|
||||
}
|
||||
|
||||
func captureStressResources(ctx context.Context, fixture *heldSessionSetRealFixture, end bool) ([]StressProcessWindows, error) {
|
||||
result := make([]StressProcessWindows, 0, len(fixture.agents)+1)
|
||||
dashboard := fixture.dashboard.RuntimeIdentity()
|
||||
dashboardProcess, err := NewStressDashboardProcess(dashboard.PID)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
windowSpec := processharness.WindowSpec{PID: dashboard.PID, Interval: contract.ResourceSampleInterval}
|
||||
if !end {
|
||||
windowSpec.ObserveSample = observeStressDashboardSQLiteJournal(fixture.dashboard.DatabasePath() + "-journal")
|
||||
}
|
||||
baseline, err := processharness.SampleWindow(ctx, windowSpec)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
dashboardWindow := StressProcessWindows{Process: dashboardProcess}
|
||||
if end {
|
||||
dashboardWindow.End = baseline
|
||||
} else {
|
||||
dashboardWindow.Baseline = baseline
|
||||
}
|
||||
result = append(result, dashboardWindow)
|
||||
for index, instance := range fixture.agents {
|
||||
identity := instance.RuntimeIdentity()
|
||||
agentOrdinal, err := NewStressAgentOrdinal(index + 1)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
process, err := NewStressAgentProcess(agentOrdinal, identity.PID)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
baseline, err := processharness.SampleWindow(ctx, processharness.WindowSpec{PID: identity.PID, Interval: contract.ResourceSampleInterval})
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
window := StressProcessWindows{Process: process}
|
||||
if end {
|
||||
window.End = baseline
|
||||
} else {
|
||||
window.Baseline = baseline
|
||||
}
|
||||
result = append(result, window)
|
||||
}
|
||||
return result, nil
|
||||
}
|
||||
|
||||
type realStressQuotaResponse struct {
|
||||
UserAccepted int `json:"user_accepted"`
|
||||
UserRejected int `json:"user_rejected"`
|
||||
ServerAccepted int `json:"server_accepted"`
|
||||
ServerRejected int `json:"server_rejected"`
|
||||
Clean bool `json:"clean"`
|
||||
}
|
||||
|
||||
type realStressRateLimitResponse struct {
|
||||
SecondAllowedCount int `json:"second_allowed_count"`
|
||||
SecondRejectedAtCount int `json:"second_rejected_at_count"`
|
||||
MinuteAllowedCount int `json:"minute_allowed_count"`
|
||||
MinuteRejectedAtCount int `json:"minute_rejected_at_count"`
|
||||
}
|
||||
|
||||
func runRealStressQuotaProbe(ctx context.Context, fixture *heldSessionSetRealFixture, agentSource string) (StressQuotaEvidence, error) {
|
||||
response, err := client.DoREST[struct{}, realStressQuotaResponse](ctx, fixture.controlPAT.Client, client.RESTRequest[struct{}]{Method: http.MethodPost, Path: "/agentcompat/io-stream-quota-probe", Body: &struct{}{}})
|
||||
if err != nil {
|
||||
return StressQuotaEvidence{}, err
|
||||
}
|
||||
rateLimit, err := client.DoREST[struct{}, realStressRateLimitResponse](ctx, fixture.controlPAT.Client, client.RESTRequest[struct{}]{Method: http.MethodPost, Path: "/agentcompat/mcp-rate-limit-probe", Body: &struct{}{}})
|
||||
if err != nil {
|
||||
return StressQuotaEvidence{}, err
|
||||
}
|
||||
pathLockProof, err := proveStressPathLockStripes(ctx, agentSource)
|
||||
if err != nil {
|
||||
return StressQuotaEvidence{}, err
|
||||
}
|
||||
return StressQuotaEvidence{PATSecond: StressQuotaBoundary{Allowed: rateLimit.SecondAllowedCount, Rejected: rateLimit.SecondRejectedAtCount, AllowedAccepted: rateLimit.SecondAllowedCount == 10, RejectedDenied: rateLimit.SecondRejectedAtCount == 11}, PATMinute: StressQuotaBoundary{Allowed: rateLimit.MinuteAllowedCount, Rejected: rateLimit.MinuteRejectedAtCount, AllowedAccepted: rateLimit.MinuteAllowedCount == 120, RejectedDenied: rateLimit.MinuteRejectedAtCount == 121}, UserStreams: StressQuotaBoundary{Allowed: response.UserAccepted, Rejected: response.UserAccepted + response.UserRejected, AllowedAccepted: response.UserAccepted == 20, RejectedDenied: response.UserRejected == 1}, ServerStreams: StressQuotaBoundary{Allowed: response.ServerAccepted, Rejected: response.ServerAccepted + response.ServerRejected, AllowedAccepted: response.ServerAccepted == 40, RejectedDenied: response.ServerRejected == 1}, PathLockStripes: pathLockProof.Stripes}, nil
|
||||
}
|
||||
|
||||
func stressCleanupSummary(fixture *heldSessionSetRealFixture, dashboardIdentity dashboard.RuntimeIdentity, agentIdentities []agent.ProcessIdentity, workspaceRoots []string, cleanupErr error) StressCleanupSummary {
|
||||
summary := StressCleanupSummary{ReceiptCount: 1 + len(fixture.agents), WorkspaceResidue: 0}
|
||||
receipts := make([]processharness.CleanupReceipt, 0, summary.ReceiptCount)
|
||||
receipts = append(receipts, fixture.dashboard.CleanupReceipt())
|
||||
for _, instance := range fixture.agents {
|
||||
receipts = append(receipts, instance.CleanupReceipt())
|
||||
}
|
||||
for _, receipt := range receipts {
|
||||
if !receipt.Passed {
|
||||
summary.FailedReceiptCount++
|
||||
}
|
||||
if receipt.Forced {
|
||||
summary.ForcedCleanupCount++
|
||||
}
|
||||
}
|
||||
if !heldRealPIDGone(dashboardIdentity.PID) {
|
||||
summary.ProcessResidue++
|
||||
}
|
||||
if !heldRealGroupGone(dashboardIdentity.ProcessGroupID) {
|
||||
summary.ProcessGroupResidue++
|
||||
}
|
||||
for index := range fixture.agents {
|
||||
identity := agentIdentities[index]
|
||||
if !heldRealPIDGone(identity.PID) {
|
||||
summary.ProcessResidue++
|
||||
}
|
||||
if !heldRealGroupGone(identity.ProcessGroupID) {
|
||||
summary.ProcessGroupResidue++
|
||||
}
|
||||
}
|
||||
for _, root := range workspaceRoots {
|
||||
if !heldSessionSetRealWorkspaceGone(root) {
|
||||
summary.WorkspaceResidue++
|
||||
}
|
||||
}
|
||||
summary.Passed = cleanupErr == nil && summary.ReceiptCount == 9 && summary.FailedReceiptCount == 0 && summary.ForcedCleanupCount == 0 && summary.ProcessResidue == 0 && summary.ProcessGroupResidue == 0 && summary.WorkspaceResidue == 0
|
||||
return summary
|
||||
}
|
||||
@@ -0,0 +1,152 @@
|
||||
//go:build linux && agentcompat
|
||||
|
||||
package scenario
|
||||
|
||||
import (
|
||||
"context"
|
||||
"crypto/sha256"
|
||||
"encoding/hex"
|
||||
"errors"
|
||||
"fmt"
|
||||
"os"
|
||||
"time"
|
||||
|
||||
"github.com/nezhahq/nezha/integration/agentcompat/internal/client"
|
||||
"github.com/nezhahq/nezha/integration/agentcompat/internal/fixture"
|
||||
)
|
||||
|
||||
type stressOperationExecutor struct {
|
||||
fixture *heldSessionSetRealFixture
|
||||
plan StressOperationPlan
|
||||
}
|
||||
|
||||
func (executor stressOperationExecutor) run(ctx context.Context) StressOperationEvidence {
|
||||
started := time.Now()
|
||||
proof, err := executor.execute(ctx)
|
||||
completed := time.Now()
|
||||
evidenceValue := StressOperationEvidence{ID: executor.plan.ID, Round: executor.plan.Round, Agent: executor.plan.Agent, PAT: executor.plan.PAT, Kind: executor.plan.Kind, LaunchedAt: started, CompletedAt: completed, Succeeded: err == nil, SuccessProof: proof}
|
||||
if err != nil {
|
||||
evidenceValue.Error = errorText(err)
|
||||
}
|
||||
return evidenceValue
|
||||
}
|
||||
|
||||
func (executor stressOperationExecutor) execute(ctx context.Context) (string, error) {
|
||||
if executor.fixture == nil || executor.plan.Agent.Int() < 1 || executor.plan.Agent.Int() > len(executor.fixture.agents) || executor.plan.Agent.Int() > len(executor.fixture.readiness) || executor.plan.Agent.Int() > len(executor.fixture.agentPATs) {
|
||||
return "", errors.New("stress operation fixture mapping is invalid")
|
||||
}
|
||||
if executor.plan.PAT.String() == "" {
|
||||
return "", errors.New("stress operation PAT is empty")
|
||||
}
|
||||
serverID := executor.fixture.readiness[executor.plan.Agent.Int()-1].ServerID
|
||||
patIdentity, err := stressOperationPATIdentity(executor.fixture, executor.plan)
|
||||
if err != nil {
|
||||
return "", err
|
||||
}
|
||||
switch executor.plan.Kind {
|
||||
case StressOperationExec:
|
||||
return executor.exec(ctx, patIdentity.Client, serverID)
|
||||
case StressOperationFilesystem:
|
||||
return executor.filesystem(ctx, patIdentity.Client, serverID)
|
||||
default:
|
||||
return "", errors.New("unsupported stress operation kind")
|
||||
}
|
||||
}
|
||||
|
||||
func stressOperationPATIdentity(fixture *heldSessionSetRealFixture, plan StressOperationPlan) (heldRealPATIdentity, error) {
|
||||
index := plan.Agent.Int() - 1
|
||||
for _, round := range fixture.plan.Rounds {
|
||||
for _, planned := range round.Operations {
|
||||
if planned.ID == plan.ID {
|
||||
if planned.Agent != plan.Agent || planned.Kind != plan.Kind || planned.PAT != plan.PAT {
|
||||
return heldRealPATIdentity{}, errors.New("stress operation plan PAT mapping is invalid")
|
||||
}
|
||||
return fixture.agentPATs[index], nil
|
||||
}
|
||||
}
|
||||
}
|
||||
for _, round := range fixture.plan.Rounds {
|
||||
for _, planned := range round.Operations {
|
||||
if planned.Agent == plan.Agent && planned.PAT == plan.PAT && planned.Kind == plan.Kind {
|
||||
return fixture.agentPATs[index], nil
|
||||
}
|
||||
}
|
||||
}
|
||||
return heldRealPATIdentity{}, errors.New("stress operation is absent from canonical plan")
|
||||
}
|
||||
|
||||
func runStressWarmups(ctx context.Context, fixture *heldSessionSetRealFixture, plan StressPlan) ([]StressWarmupEvidence, error) {
|
||||
warmups := make([]StressWarmupEvidence, 0, len(fixture.agents))
|
||||
for agentIndex := range fixture.agents {
|
||||
agentOrdinal, err := NewStressAgentOrdinal(agentIndex + 1)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
pat := StressPATID{}
|
||||
for _, operation := range plan.Rounds[0].Operations {
|
||||
if operation.Agent == agentOrdinal {
|
||||
pat = operation.PAT
|
||||
break
|
||||
}
|
||||
}
|
||||
if pat.String() == "" {
|
||||
return nil, errors.New("stress warmup PAT mapping is invalid")
|
||||
}
|
||||
execID, err := NewStressOperationID(fmt.Sprintf("warmup-exec-a%02d", agentOrdinal.Int()))
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
execResult := stressOperationExecutor{fixture: fixture, plan: StressOperationPlan{ID: execID, Agent: agentOrdinal, PAT: pat, Kind: StressOperationExec}}
|
||||
if _, err := execResult.execute(ctx); err != nil {
|
||||
return nil, err
|
||||
}
|
||||
filesystemID, err := NewStressOperationID(fmt.Sprintf("warmup-filesystem-a%02d", agentOrdinal.Int()))
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
filesystemResult := stressOperationExecutor{fixture: fixture, plan: StressOperationPlan{ID: filesystemID, Round: 0, Agent: agentOrdinal, PAT: pat, Kind: StressOperationFilesystem}}
|
||||
if _, err := filesystemResult.execute(ctx); err != nil {
|
||||
return nil, err
|
||||
}
|
||||
warmups = append(warmups, StressWarmupEvidence{Agent: agentOrdinal, Exec: true, Filesystem: true, Terminal: true, NAT: true, FM: true})
|
||||
}
|
||||
return warmups, nil
|
||||
}
|
||||
|
||||
func (executor stressOperationExecutor) exec(ctx context.Context, patClient *client.Client, serverID uint64) (string, error) {
|
||||
const token = "agentcompat-stress-exec-proof"
|
||||
result, err := client.CallTool[execArguments, execResult](ctx, patClient, client.ToolCall[execArguments]{Name: "server.exec", Arguments: execArguments{ServerID: serverID, Cmd: "/bin/sh", Args: []string{"-c", "printf " + token}}})
|
||||
if err != nil || result.StructuredContent.ExitCode != 0 || result.StructuredContent.Stdout != token || result.StructuredContent.Error != "" || result.StructuredContent.TimedOut || result.StructuredContent.StdoutTruncated {
|
||||
return "", errors.New("stress Exec proof failed")
|
||||
}
|
||||
return stressProof(token), nil
|
||||
}
|
||||
|
||||
func (executor stressOperationExecutor) filesystem(ctx context.Context, patClient *client.Client, serverID uint64) (string, error) {
|
||||
parent := executor.fixture.agents[executor.plan.Agent.Int()-1].WorkspaceRoot()
|
||||
return executeStressFilesystemProof(ctx, patClient, serverID, parent, executor.plan.ID.String(), executor.plan.Round)
|
||||
}
|
||||
|
||||
func executeStressFilesystemProof(ctx context.Context, patClient *client.Client, serverID uint64, parent, operationID string, round int) (string, error) {
|
||||
root, err := fixture.NewAgentRoot(parent, fmt.Sprintf("stress-%s", operationID))
|
||||
if err != nil {
|
||||
return "", err
|
||||
}
|
||||
defer os.RemoveAll(root.Absolute())
|
||||
filesystem := newMCPFilesystemClient(patClient, serverID, root)
|
||||
content := "agentcompat-stress-filesystem-proof"
|
||||
relative := fmt.Sprintf("round-%d/%s.txt", round, operationID)
|
||||
written, err := filesystem.write(ctx, mcpFilesystemWrite{relative: relative, content: content, encoding: "utf8", mode: "0600", createDirs: true})
|
||||
if err != nil {
|
||||
return "", fmt.Errorf("stress filesystem write proof failed: %w", err)
|
||||
}
|
||||
if written.StructuredContent.Size != int64(len(content)) || written.StructuredContent.SHA256 != stressProof(content) || written.StructuredContent.Error != "" {
|
||||
return "", errors.New("stress filesystem write proof response invalid")
|
||||
}
|
||||
return stressProof(content), nil
|
||||
}
|
||||
|
||||
func stressProof(value string) string {
|
||||
digest := sha256.Sum256([]byte(value))
|
||||
return hex.EncodeToString(digest[:])
|
||||
}
|
||||
@@ -0,0 +1,78 @@
|
||||
//go:build linux && agentcompat
|
||||
|
||||
package scenario
|
||||
|
||||
import (
|
||||
"encoding/json"
|
||||
"net/http"
|
||||
"net/http/httptest"
|
||||
"testing"
|
||||
|
||||
"github.com/stretchr/testify/require"
|
||||
|
||||
"github.com/nezhahq/nezha/integration/agentcompat/internal/client"
|
||||
)
|
||||
|
||||
func TestStressFilesystemUsesAgentPAT(t *testing.T) {
|
||||
fixture := &heldSessionSetRealFixture{
|
||||
agentPATs: []heldRealPATIdentity{{TokenID: 1}},
|
||||
}
|
||||
operation := StressOperationPlan{Agent: mustStressAgentOrdinal(t, 1), PAT: mustStressPATID(t, "pat-1"), Kind: StressOperationFilesystem}
|
||||
|
||||
fixture.plan = StressPlan{Rounds: []StressRoundPlan{{Operations: []StressOperationPlan{operation}}}}
|
||||
selected, err := stressOperationPATIdentity(fixture, operation)
|
||||
require.NoError(t, err)
|
||||
require.Equal(t, uint64(1), selected.TokenID)
|
||||
}
|
||||
|
||||
func TestStressExecUsesAgentPAT(t *testing.T) {
|
||||
fixture := &heldSessionSetRealFixture{
|
||||
agentPATs: []heldRealPATIdentity{{TokenID: 1}},
|
||||
}
|
||||
operation := StressOperationPlan{Agent: mustStressAgentOrdinal(t, 1), PAT: mustStressPATID(t, "pat-1"), Kind: StressOperationExec}
|
||||
|
||||
fixture.plan = StressPlan{Rounds: []StressRoundPlan{{Operations: []StressOperationPlan{operation}}}}
|
||||
selected, err := stressOperationPATIdentity(fixture, operation)
|
||||
require.NoError(t, err)
|
||||
require.Equal(t, uint64(1), selected.TokenID)
|
||||
}
|
||||
|
||||
func TestStressFilesystemProofDispatchesOneWrite(t *testing.T) {
|
||||
requests := make([]string, 0, 1)
|
||||
server := httptest.NewServer(http.HandlerFunc(func(writer http.ResponseWriter, request *http.Request) {
|
||||
var envelope struct {
|
||||
Params struct {
|
||||
Name string `json:"name"`
|
||||
} `json:"params"`
|
||||
}
|
||||
require.NoError(t, json.NewDecoder(request.Body).Decode(&envelope))
|
||||
requests = append(requests, envelope.Params.Name)
|
||||
writer.Header().Set("Content-Type", "application/json")
|
||||
_, err := writer.Write([]byte(`{"jsonrpc":"2.0","id":1,"result":{"content":[],"structuredContent":{"size":35,"sha256":"6ececdd71257073948afc9c699d12d3075d05d3dc339c4511496c3fbc27a2081"}}}`))
|
||||
require.NoError(t, err)
|
||||
}))
|
||||
t.Cleanup(server.Close)
|
||||
mcpClient, err := client.New(client.Config{BaseURL: server.URL})
|
||||
require.NoError(t, err)
|
||||
parent := t.TempDir()
|
||||
|
||||
proof, err := executeStressFilesystemProof(t.Context(), mcpClient, 17, parent, "operation", 1)
|
||||
|
||||
require.NoError(t, err)
|
||||
require.Equal(t, stressProof("agentcompat-stress-filesystem-proof"), proof)
|
||||
require.Equal(t, []string{"fs.write"}, requests)
|
||||
}
|
||||
|
||||
func mustStressAgentOrdinal(t *testing.T, value int) StressAgentOrdinal {
|
||||
t.Helper()
|
||||
ordinal, err := NewStressAgentOrdinal(value)
|
||||
require.NoError(t, err)
|
||||
return ordinal
|
||||
}
|
||||
|
||||
func mustStressPATID(t *testing.T, value string) StressPATID {
|
||||
t.Helper()
|
||||
pat, err := NewStressPATID(value)
|
||||
require.NoError(t, err)
|
||||
return pat
|
||||
}
|
||||
@@ -0,0 +1,108 @@
|
||||
//go:build linux
|
||||
|
||||
package scenario
|
||||
|
||||
import (
|
||||
"errors"
|
||||
"fmt"
|
||||
|
||||
"github.com/nezhahq/nezha/integration/agentcompat/internal/contract"
|
||||
processharness "github.com/nezhahq/nezha/integration/agentcompat/internal/process"
|
||||
)
|
||||
|
||||
var (
|
||||
ErrStressProcessWindow = errors.New("stress process window is invalid")
|
||||
ErrStressResourceDrift = errors.New("stress process resource count drifted")
|
||||
ErrStressRSSLimit = errors.New("stress process RSS limit exceeded")
|
||||
)
|
||||
|
||||
type StressProcessWindows struct {
|
||||
Process StressProcessIdentity `json:"process"`
|
||||
Baseline processharness.Window `json:"baseline"`
|
||||
End processharness.Window `json:"end"`
|
||||
}
|
||||
|
||||
type StressResourceMaxima struct {
|
||||
RSSBytes uint64 `json:"rss_bytes"`
|
||||
Descendants int `json:"descendants"`
|
||||
NonStdioFDs int `json:"non_stdio_fds"`
|
||||
TCPListeners int `json:"tcp_listeners"`
|
||||
TCP6Listeners int `json:"tcp6_listeners"`
|
||||
}
|
||||
|
||||
type StressResourceEvaluation struct {
|
||||
Process StressProcessIdentity `json:"process"`
|
||||
Baseline StressResourceMaxima `json:"baseline"`
|
||||
End StressResourceMaxima `json:"end"`
|
||||
RSSDeltaBytes uint64 `json:"rss_delta_bytes"`
|
||||
RSSLimitBytes uint64 `json:"rss_limit_bytes"`
|
||||
}
|
||||
|
||||
func EvaluateStressResource(input StressProcessWindows) (StressResourceEvaluation, error) {
|
||||
baseline, err := stressWindowMaxima(input.Process, input.Baseline)
|
||||
if err != nil {
|
||||
return StressResourceEvaluation{}, err
|
||||
}
|
||||
end, err := stressWindowMaxima(input.Process, input.End)
|
||||
if err != nil {
|
||||
return StressResourceEvaluation{}, err
|
||||
}
|
||||
budget := contract.DefaultResourceBudget()
|
||||
if end.Descendants != baseline.Descendants || end.NonStdioFDs != baseline.NonStdioFDs || end.TCPListeners != baseline.TCPListeners || end.TCP6Listeners != baseline.TCP6Listeners {
|
||||
return StressResourceEvaluation{}, fmt.Errorf("process=%s baseline=%+v end=%+v baseline_samples=%+v end_samples=%+v: %w", input.Process.key(), baseline, end, resourceCountSamples(input.Baseline), resourceCountSamples(input.End), ErrStressResourceDrift)
|
||||
}
|
||||
limit, err := stressRSSLimit(input.Process, budget)
|
||||
if err != nil {
|
||||
return StressResourceEvaluation{}, err
|
||||
}
|
||||
delta := uint64(0)
|
||||
if end.RSSBytes > baseline.RSSBytes {
|
||||
delta = end.RSSBytes - baseline.RSSBytes
|
||||
}
|
||||
if delta > limit {
|
||||
return StressResourceEvaluation{}, fmt.Errorf("process=%s RSS delta=%d limit=%d: %w", input.Process.key(), delta, limit, ErrStressRSSLimit)
|
||||
}
|
||||
return StressResourceEvaluation{Process: input.Process, Baseline: baseline, End: end, RSSDeltaBytes: delta, RSSLimitBytes: limit}, nil
|
||||
}
|
||||
|
||||
func resourceCountSamples(window processharness.Window) []string {
|
||||
result := make([]string, 0, len(window.Samples))
|
||||
for _, sample := range window.Samples {
|
||||
result = append(result, fmt.Sprintf("descendants=%d fd=%d tcp=%d tcp6=%d", sample.DescendantCount, sample.NonStdioFDCount, sample.TCPListenerCount, sample.TCP6ListenerCount))
|
||||
}
|
||||
return result
|
||||
}
|
||||
|
||||
func stressWindowMaxima(process StressProcessIdentity, window processharness.Window) (StressResourceMaxima, error) {
|
||||
if process.PID < 1 || window.PID != process.PID || len(window.Samples) != contract.ResourceSampleCount {
|
||||
return StressResourceMaxima{}, fmt.Errorf("process=%s window PID=%d samples=%d: %w", process.key(), window.PID, len(window.Samples), ErrStressProcessWindow)
|
||||
}
|
||||
maxima := StressResourceMaxima{}
|
||||
for _, sample := range window.Samples {
|
||||
if sample.PID != process.PID || sample.PID < 1 {
|
||||
return StressResourceMaxima{}, fmt.Errorf("process=%s sample PID=%d: %w", process.key(), sample.PID, ErrStressProcessWindow)
|
||||
}
|
||||
maxima.RSSBytes = max(maxima.RSSBytes, sample.RSSBytes)
|
||||
// Count drift compares each window's terminal state. Using a baseline high-water
|
||||
// mark turns legitimate short-lived work that has already drained into a leak.
|
||||
maxima.Descendants = sample.DescendantCount
|
||||
maxima.NonStdioFDs = sample.NonStdioFDCount
|
||||
maxima.TCPListeners = sample.TCPListenerCount
|
||||
maxima.TCP6Listeners = sample.TCP6ListenerCount
|
||||
}
|
||||
return maxima, nil
|
||||
}
|
||||
|
||||
func stressRSSLimit(process StressProcessIdentity, budget contract.ResourceBudget) (uint64, error) {
|
||||
switch process.Kind {
|
||||
case StressProcessDashboard:
|
||||
return budget.DashboardRSSDeltaBytes(), nil
|
||||
case StressProcessAgent:
|
||||
if process.Agent.Int() < 1 {
|
||||
return 0, fmt.Errorf("process=%s: %w", process.key(), ErrStressIdentity)
|
||||
}
|
||||
return budget.AgentRSSDeltaBytes(), nil
|
||||
default:
|
||||
return 0, fmt.Errorf("process kind=%q: %w", process.Kind, ErrStressIdentity)
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,178 @@
|
||||
//go:build linux
|
||||
|
||||
package scenario
|
||||
|
||||
import (
|
||||
"testing"
|
||||
|
||||
"github.com/stretchr/testify/require"
|
||||
|
||||
processharness "github.com/nezhahq/nezha/integration/agentcompat/internal/process"
|
||||
)
|
||||
|
||||
func TestStress_RejectsLeakedFD(t *testing.T) {
|
||||
input := stressDashboardResourceFixture(100)
|
||||
for index := range input.End.Samples {
|
||||
input.End.Samples[index].NonStdioFDCount++
|
||||
}
|
||||
|
||||
_, err := EvaluateStressResource(input)
|
||||
|
||||
require.ErrorIs(t, err, ErrStressResourceDrift)
|
||||
}
|
||||
|
||||
func TestStress_RejectsSingleSampleFDTransient(t *testing.T) {
|
||||
input := stressDashboardResourceFixture(100)
|
||||
input.End.Samples[4].NonStdioFDCount++
|
||||
|
||||
_, err := EvaluateStressResource(input)
|
||||
|
||||
require.ErrorIs(t, err, ErrStressResourceDrift)
|
||||
}
|
||||
|
||||
func TestStress_AcceptsRecoveredBaselineFDTransient(t *testing.T) {
|
||||
input := stressDashboardResourceFixture(100)
|
||||
input.Baseline.Samples[1].NonStdioFDCount = 9
|
||||
|
||||
evaluation, err := EvaluateStressResource(input)
|
||||
|
||||
require.NoError(t, err)
|
||||
require.Equal(t, 3, evaluation.Baseline.NonStdioFDs)
|
||||
require.Equal(t, 3, evaluation.End.NonStdioFDs)
|
||||
}
|
||||
|
||||
func TestStress_RejectsSingleSampleDescendantTransient(t *testing.T) {
|
||||
input := stressDashboardResourceFixture(100)
|
||||
input.End.Samples[4].DescendantCount++
|
||||
_, err := EvaluateStressResource(input)
|
||||
require.ErrorIs(t, err, ErrStressResourceDrift)
|
||||
}
|
||||
|
||||
func TestStress_RejectsSingleSampleTCPTransient(t *testing.T) {
|
||||
input := stressDashboardResourceFixture(100)
|
||||
input.End.Samples[4].TCPListenerCount++
|
||||
_, err := EvaluateStressResource(input)
|
||||
require.ErrorIs(t, err, ErrStressResourceDrift)
|
||||
}
|
||||
|
||||
func TestStress_RejectsSingleSampleTCP6Transient(t *testing.T) {
|
||||
input := stressDashboardResourceFixture(100)
|
||||
input.End.Samples[4].TCP6ListenerCount++
|
||||
_, err := EvaluateStressResource(input)
|
||||
require.ErrorIs(t, err, ErrStressResourceDrift)
|
||||
}
|
||||
|
||||
func TestStress_RejectsDecreasedDescendants(t *testing.T) {
|
||||
input := stressDashboardResourceFixture(100)
|
||||
for index := range input.End.Samples {
|
||||
input.End.Samples[index].DescendantCount = 0
|
||||
input.Baseline.Samples[index].DescendantCount = 1
|
||||
}
|
||||
|
||||
_, err := EvaluateStressResource(input)
|
||||
|
||||
require.ErrorIs(t, err, ErrStressResourceDrift)
|
||||
}
|
||||
|
||||
func TestStress_RejectsDecreasedFDs(t *testing.T) {
|
||||
input := stressDashboardResourceFixture(100)
|
||||
for index := range input.End.Samples {
|
||||
input.End.Samples[index].NonStdioFDCount = 2
|
||||
input.Baseline.Samples[index].NonStdioFDCount = 3
|
||||
}
|
||||
|
||||
_, err := EvaluateStressResource(input)
|
||||
|
||||
require.ErrorIs(t, err, ErrStressResourceDrift)
|
||||
}
|
||||
|
||||
func TestStress_RejectsDecreasedTCPListeners(t *testing.T) {
|
||||
input := stressDashboardResourceFixture(100)
|
||||
for index := range input.End.Samples {
|
||||
input.End.Samples[index].TCPListenerCount = 0
|
||||
input.Baseline.Samples[index].TCPListenerCount = 1
|
||||
}
|
||||
|
||||
_, err := EvaluateStressResource(input)
|
||||
|
||||
require.ErrorIs(t, err, ErrStressResourceDrift)
|
||||
}
|
||||
|
||||
func TestStress_RejectsDecreasedTCP6Listeners(t *testing.T) {
|
||||
input := stressDashboardResourceFixture(100)
|
||||
for index := range input.End.Samples {
|
||||
input.End.Samples[index].TCP6ListenerCount = 0
|
||||
input.Baseline.Samples[index].TCP6ListenerCount = 1
|
||||
}
|
||||
|
||||
_, err := EvaluateStressResource(input)
|
||||
|
||||
require.ErrorIs(t, err, ErrStressResourceDrift)
|
||||
}
|
||||
|
||||
func TestStress_RejectsDashboardRSSOverLimit(t *testing.T) {
|
||||
input := stressDashboardResourceFixture(100)
|
||||
input.End.Samples[4].RSSBytes = 100 + 67108865
|
||||
|
||||
_, err := EvaluateStressResource(input)
|
||||
|
||||
require.ErrorIs(t, err, ErrStressRSSLimit)
|
||||
}
|
||||
|
||||
func TestStress_RejectsAgentRSSOverLimit(t *testing.T) {
|
||||
agent, err := NewStressAgentOrdinal(4)
|
||||
require.NoError(t, err)
|
||||
identity, err := NewStressAgentProcess(agent, 404)
|
||||
require.NoError(t, err)
|
||||
input := StressProcessWindows{Process: identity, Baseline: stressWindow(404, 100), End: stressWindow(404, 100+33554433)}
|
||||
|
||||
_, err = EvaluateStressResource(input)
|
||||
|
||||
require.ErrorIs(t, err, ErrStressRSSLimit)
|
||||
}
|
||||
|
||||
func TestStress_RejectsVanishedPID(t *testing.T) {
|
||||
input := stressDashboardResourceFixture(100)
|
||||
input.End.Samples[2].PID = 0
|
||||
|
||||
_, err := EvaluateStressResource(input)
|
||||
|
||||
require.ErrorIs(t, err, ErrStressProcessWindow)
|
||||
}
|
||||
|
||||
func TestStress_RejectsIncompleteSampleWindow(t *testing.T) {
|
||||
input := stressDashboardResourceFixture(100)
|
||||
input.End.Samples = input.End.Samples[:4]
|
||||
|
||||
_, err := EvaluateStressResource(input)
|
||||
|
||||
require.ErrorIs(t, err, ErrStressProcessWindow)
|
||||
}
|
||||
|
||||
func TestStress_UsesFinalCountSampleButWindowRSSMaximum(t *testing.T) {
|
||||
input := stressDashboardResourceFixture(100)
|
||||
input.Baseline.Samples[0].NonStdioFDCount = 4
|
||||
input.End.Samples[0].NonStdioFDCount = 4
|
||||
input.End.Samples[4].RSSBytes = 110
|
||||
|
||||
evaluation, err := EvaluateStressResource(input)
|
||||
|
||||
require.NoError(t, err)
|
||||
require.Equal(t, uint64(10), evaluation.RSSDeltaBytes)
|
||||
}
|
||||
|
||||
func stressDashboardResourceFixture(rss uint64) StressProcessWindows {
|
||||
identity, err := NewStressDashboardProcess(101)
|
||||
if err != nil {
|
||||
panic(err)
|
||||
}
|
||||
return StressProcessWindows{Process: identity, Baseline: stressWindow(101, rss), End: stressWindow(101, rss)}
|
||||
}
|
||||
|
||||
func stressWindow(pid int, rss uint64) processharness.Window {
|
||||
samples := make([]processharness.Sample, 5)
|
||||
for index := range samples {
|
||||
samples[index] = processharness.Sample{PID: pid, RSSBytes: rss, NonStdioFDCount: 3, TCPListenerCount: 1}
|
||||
}
|
||||
return processharness.Window{PID: pid, Samples: samples}
|
||||
}
|
||||
@@ -0,0 +1,90 @@
|
||||
//go:build linux
|
||||
|
||||
package scenario
|
||||
|
||||
import (
|
||||
"errors"
|
||||
"fmt"
|
||||
"time"
|
||||
)
|
||||
|
||||
var (
|
||||
ErrStressOperationSet = errors.New("stress operation evidence set is invalid")
|
||||
ErrStressLaunchWindow = errors.New("stress operation launch window exceeded one second")
|
||||
)
|
||||
|
||||
const stressLaunchWindow = time.Second
|
||||
|
||||
type StressOperationEvidence struct {
|
||||
ID StressOperationID `json:"id"`
|
||||
Round int `json:"round"`
|
||||
Agent StressAgentOrdinal `json:"agent"`
|
||||
PAT StressPATID `json:"pat_id"`
|
||||
Kind StressOperationKind `json:"kind"`
|
||||
LaunchedAt time.Time `json:"launched_at"`
|
||||
CompletedAt time.Time `json:"completed_at"`
|
||||
Succeeded bool `json:"succeeded"`
|
||||
SuccessProof string `json:"success_proof"`
|
||||
Error string `json:"error,omitempty"`
|
||||
}
|
||||
|
||||
type StressRoundEvidence struct {
|
||||
Round int `json:"round"`
|
||||
Operations []StressOperationEvidence `json:"operations"`
|
||||
}
|
||||
|
||||
func ValidateStressRoundEvidence(plan StressRoundPlan, evidence StressRoundEvidence) error {
|
||||
matched, err := matchStressRoundEvidence(plan, evidence)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
for _, operation := range matched {
|
||||
if !operation.Succeeded || operation.SuccessProof == "" || operation.Error != "" {
|
||||
return fmt.Errorf("operation %s did not succeed: %w", operation.ID.String(), ErrStressOperationSet)
|
||||
}
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
func matchStressRoundEvidence(plan StressRoundPlan, evidence StressRoundEvidence) ([]StressOperationEvidence, error) {
|
||||
if evidence.Round != plan.Round || len(evidence.Operations) != len(plan.Operations) {
|
||||
return nil, fmt.Errorf("round=%d operations=%d want round=%d operations=%d: %w", evidence.Round, len(evidence.Operations), plan.Round, len(plan.Operations), ErrStressOperationSet)
|
||||
}
|
||||
expected := make(map[StressOperationID]struct{}, len(plan.Operations))
|
||||
for _, operation := range plan.Operations {
|
||||
expected[operation.ID] = struct{}{}
|
||||
}
|
||||
matched := make([]StressOperationEvidence, 0, len(evidence.Operations))
|
||||
seen := make(map[StressOperationID]struct{}, len(evidence.Operations))
|
||||
var firstLaunch, lastLaunch time.Time
|
||||
for index, operation := range evidence.Operations {
|
||||
planned := plan.Operations[index]
|
||||
if _, exists := expected[operation.ID]; !exists {
|
||||
return nil, fmt.Errorf("unexpected operation %s: %w", operation.ID.String(), ErrStressOperationSet)
|
||||
}
|
||||
if operation.Round != planned.Round || operation.Agent != planned.Agent || operation.PAT != planned.PAT || operation.Kind != planned.Kind || operation.ID != planned.ID {
|
||||
return nil, fmt.Errorf("operation %s owner or order mismatch: %w", operation.ID.String(), ErrStressOperationSet)
|
||||
}
|
||||
if _, duplicate := seen[operation.ID]; duplicate {
|
||||
return nil, fmt.Errorf("duplicate operation %s: %w", operation.ID.String(), ErrStressOperationSet)
|
||||
}
|
||||
if operation.LaunchedAt.IsZero() || operation.CompletedAt.Before(operation.LaunchedAt) {
|
||||
return nil, fmt.Errorf("operation %s timing is invalid: %w", operation.ID.String(), ErrStressOperationSet)
|
||||
}
|
||||
if operation.Succeeded && operation.SuccessProof == "" {
|
||||
return nil, fmt.Errorf("operation %s proof is empty: %w", operation.ID.String(), ErrStressOperationSet)
|
||||
}
|
||||
seen[operation.ID] = struct{}{}
|
||||
matched = append(matched, operation)
|
||||
if firstLaunch.IsZero() || operation.LaunchedAt.Before(firstLaunch) {
|
||||
firstLaunch = operation.LaunchedAt
|
||||
}
|
||||
if lastLaunch.IsZero() || operation.LaunchedAt.After(lastLaunch) {
|
||||
lastLaunch = operation.LaunchedAt
|
||||
}
|
||||
}
|
||||
if lastLaunch.Sub(firstLaunch) > stressLaunchWindow {
|
||||
return nil, fmt.Errorf("launch window=%s: %w", lastLaunch.Sub(firstLaunch), ErrStressLaunchWindow)
|
||||
}
|
||||
return matched, nil
|
||||
}
|
||||
@@ -0,0 +1,91 @@
|
||||
//go:build linux && agentcompat
|
||||
|
||||
package scenario
|
||||
|
||||
import (
|
||||
"context"
|
||||
"errors"
|
||||
"fmt"
|
||||
"sync"
|
||||
)
|
||||
|
||||
func runStressRound(ctx context.Context, fixture *heldSessionSetRealFixture, plan StressRoundPlan) (StressRoundEvidence, error) {
|
||||
registry, err := newStressExactOnceRegistry(StressPlan{Rounds: []StressRoundPlan{plan}})
|
||||
if err != nil {
|
||||
return StressRoundEvidence{}, err
|
||||
}
|
||||
ready := make(chan struct{}, len(plan.Operations))
|
||||
release := make(chan struct{})
|
||||
results := make(chan StressOperationEvidence, len(plan.Operations))
|
||||
var workers sync.WaitGroup
|
||||
for _, operation := range plan.Operations {
|
||||
operation := operation
|
||||
workers.Add(1)
|
||||
go func() {
|
||||
defer workers.Done()
|
||||
ready <- struct{}{}
|
||||
<-release
|
||||
if startErr := registry.Start(operation); startErr != nil {
|
||||
results <- StressOperationEvidence{ID: operation.ID, Round: operation.Round, Agent: operation.Agent, PAT: operation.PAT, Kind: operation.Kind, Error: startErr.Error()}
|
||||
return
|
||||
}
|
||||
results <- stressOperationExecutor{fixture: fixture, plan: operation}.run(ctx)
|
||||
}()
|
||||
}
|
||||
for range plan.Operations {
|
||||
select {
|
||||
case <-ready:
|
||||
case <-ctx.Done():
|
||||
close(release)
|
||||
workers.Wait()
|
||||
return StressRoundEvidence{}, ctx.Err()
|
||||
}
|
||||
}
|
||||
close(release)
|
||||
workers.Wait()
|
||||
byID := make(map[StressOperationID]StressOperationEvidence, len(plan.Operations))
|
||||
for range plan.Operations {
|
||||
select {
|
||||
case operation := <-results:
|
||||
if _, duplicate := byID[operation.ID]; duplicate {
|
||||
return StressRoundEvidence{}, fmt.Errorf("duplicate result for operation %s", operation.ID.String())
|
||||
}
|
||||
byID[operation.ID] = operation
|
||||
case <-ctx.Done():
|
||||
return StressRoundEvidence{}, ctx.Err()
|
||||
}
|
||||
}
|
||||
evidenceValue := StressRoundEvidence{Round: plan.Round, Operations: make([]StressOperationEvidence, len(plan.Operations))}
|
||||
for index, operation := range plan.Operations {
|
||||
result, exists := byID[operation.ID]
|
||||
if !exists {
|
||||
return StressRoundEvidence{}, ErrStressOperationMissingCompletion
|
||||
}
|
||||
evidenceValue.Operations[index] = result
|
||||
if result.Error != "" {
|
||||
return StressRoundEvidence{}, errors.Join(fmt.Errorf("operation %s (%s) failed: %s", result.ID.String(), result.Kind, result.Error), stressRoundErrors(evidenceValue))
|
||||
}
|
||||
if result.Error == "" {
|
||||
if err := registry.Complete(StressOperationReceipt{Operation: operation, SuccessProof: result.SuccessProof}); err != nil {
|
||||
return StressRoundEvidence{}, err
|
||||
}
|
||||
}
|
||||
}
|
||||
if err := registry.ValidateComplete(); err != nil {
|
||||
return StressRoundEvidence{}, err
|
||||
}
|
||||
if err := ValidateStressRoundEvidence(plan, evidenceValue); err != nil {
|
||||
return StressRoundEvidence{}, errors.Join(err, stressRoundErrors(evidenceValue))
|
||||
}
|
||||
return evidenceValue, nil
|
||||
}
|
||||
|
||||
func stressRoundErrors(evidenceValue StressRoundEvidence) error {
|
||||
var joined error
|
||||
for _, operation := range evidenceValue.Operations {
|
||||
if operation.Error != "" {
|
||||
joined = errors.Join(joined, errors.New(operation.Error))
|
||||
}
|
||||
}
|
||||
return joined
|
||||
}
|
||||
@@ -0,0 +1,88 @@
|
||||
//go:build linux
|
||||
|
||||
package scenario
|
||||
|
||||
import (
|
||||
"errors"
|
||||
"fmt"
|
||||
|
||||
"github.com/nezhahq/nezha/integration/agentcompat/internal/contract"
|
||||
)
|
||||
|
||||
var ErrStressSoakTrend = errors.New("stress soak RSS increases strictly across all iterations")
|
||||
|
||||
type StressRSSSeries struct {
|
||||
Process StressProcessIdentity `json:"process"`
|
||||
EndRSSBytes [3]uint64 `json:"end_rss_bytes"`
|
||||
}
|
||||
|
||||
type StressSoakTrendEvidence struct {
|
||||
Series []StressRSSSeries `json:"series"`
|
||||
}
|
||||
|
||||
func ValidateStressSoakTrend(evidence StressSoakTrendEvidence) error {
|
||||
if len(evidence.Series) == 0 {
|
||||
return errors.New("stress soak RSS series are empty")
|
||||
}
|
||||
seen := make(map[string]struct{}, len(evidence.Series))
|
||||
for _, series := range evidence.Series {
|
||||
key := series.Process.key()
|
||||
if series.Process.PID < 1 || key == ":0" || (series.Process.Kind == StressProcessAgent && series.Process.Agent.Int() < 1) {
|
||||
return fmt.Errorf("process=%s: %w", key, ErrStressIdentity)
|
||||
}
|
||||
if _, duplicate := seen[key]; duplicate {
|
||||
return fmt.Errorf("duplicate process=%s: %w", key, ErrStressIdentity)
|
||||
}
|
||||
seen[key] = struct{}{}
|
||||
values := series.EndRSSBytes
|
||||
if values[0] < values[1] && values[1] < values[2] {
|
||||
return fmt.Errorf("process=%s RSS=%v: %w", key, values, ErrStressSoakTrend)
|
||||
}
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
func ValidateStressSoakTrendForProfile(profile contract.Profile, evidence StressSoakTrendEvidence) error {
|
||||
want := profile.AgentCount() + 1
|
||||
if len(evidence.Series) != want {
|
||||
return fmt.Errorf("soak series=%d want=%d: %w", len(evidence.Series), want, ErrStressIdentity)
|
||||
}
|
||||
seenDashboard := 0
|
||||
seenAgents := make(map[int]struct{}, profile.AgentCount())
|
||||
for _, series := range evidence.Series {
|
||||
if err := validateStressSoakSeries(series); err != nil {
|
||||
return err
|
||||
}
|
||||
switch series.Process.Kind {
|
||||
case StressProcessDashboard:
|
||||
seenDashboard++
|
||||
case StressProcessAgent:
|
||||
ordinal := series.Process.Agent.Int()
|
||||
if ordinal < 1 || ordinal > profile.AgentCount() {
|
||||
return fmt.Errorf("unknown agent ordinal=%d: %w", ordinal, ErrStressIdentity)
|
||||
}
|
||||
if _, duplicate := seenAgents[ordinal]; duplicate {
|
||||
return fmt.Errorf("duplicate agent ordinal=%d: %w", ordinal, ErrStressIdentity)
|
||||
}
|
||||
seenAgents[ordinal] = struct{}{}
|
||||
default:
|
||||
return fmt.Errorf("unknown process kind=%q: %w", series.Process.Kind, ErrStressIdentity)
|
||||
}
|
||||
}
|
||||
if seenDashboard != 1 || len(seenAgents) != profile.AgentCount() {
|
||||
return fmt.Errorf("dashboard=%d agents=%d: %w", seenDashboard, len(seenAgents), ErrStressIdentity)
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
func validateStressSoakSeries(series StressRSSSeries) error {
|
||||
key := series.Process.key()
|
||||
if series.Process.PID < 1 || key == ":0" || (series.Process.Kind == StressProcessAgent && series.Process.Agent.Int() < 1) {
|
||||
return fmt.Errorf("process=%s: %w", key, ErrStressIdentity)
|
||||
}
|
||||
values := series.EndRSSBytes
|
||||
if values[0] < values[1] && values[1] < values[2] {
|
||||
return fmt.Errorf("process=%s RSS=%v: %w", key, values, ErrStressSoakTrend)
|
||||
}
|
||||
return nil
|
||||
}
|
||||
@@ -0,0 +1,86 @@
|
||||
//go:build linux
|
||||
|
||||
package scenario
|
||||
|
||||
import (
|
||||
"testing"
|
||||
|
||||
"github.com/stretchr/testify/require"
|
||||
|
||||
"github.com/nezhahq/nezha/integration/agentcompat/internal/contract"
|
||||
)
|
||||
|
||||
func TestStress_RejectsStrictThreePointDashboardRSSIncrease(t *testing.T) {
|
||||
identity, err := NewStressDashboardProcess(101)
|
||||
require.NoError(t, err)
|
||||
trend := StressSoakTrendEvidence{Series: []StressRSSSeries{{Process: identity, EndRSSBytes: [3]uint64{100, 101, 102}}}}
|
||||
|
||||
err = ValidateStressSoakTrend(trend)
|
||||
|
||||
require.ErrorIs(t, err, ErrStressSoakTrend)
|
||||
}
|
||||
|
||||
func TestStress_RejectsStrictThreePointAgentRSSIncrease(t *testing.T) {
|
||||
agent, err := NewStressAgentOrdinal(4)
|
||||
require.NoError(t, err)
|
||||
identity, err := NewStressAgentProcess(agent, 404)
|
||||
require.NoError(t, err)
|
||||
trend := StressSoakTrendEvidence{Series: []StressRSSSeries{{Process: identity, EndRSSBytes: [3]uint64{200, 300, 400}}}}
|
||||
|
||||
err = ValidateStressSoakTrend(trend)
|
||||
|
||||
require.ErrorIs(t, err, ErrStressSoakTrend)
|
||||
}
|
||||
|
||||
func TestStress_AcceptsStableAndNonMonotonicSoakTrend(t *testing.T) {
|
||||
dashboard, err := NewStressDashboardProcess(101)
|
||||
require.NoError(t, err)
|
||||
agentOrdinal, err := NewStressAgentOrdinal(1)
|
||||
require.NoError(t, err)
|
||||
agent, err := NewStressAgentProcess(agentOrdinal, 201)
|
||||
require.NoError(t, err)
|
||||
trend := StressSoakTrendEvidence{Series: []StressRSSSeries{
|
||||
{Process: dashboard, EndRSSBytes: [3]uint64{100, 100, 100}},
|
||||
{Process: agent, EndRSSBytes: [3]uint64{200, 220, 210}},
|
||||
}}
|
||||
|
||||
err = ValidateStressSoakTrend(trend)
|
||||
|
||||
require.NoError(t, err)
|
||||
}
|
||||
|
||||
func TestStressSoak_RejectsMissingUnknownAndDuplicateSeries(t *testing.T) {
|
||||
profile := mustProfile(t, contract.ProfileSoak)
|
||||
dashboard, err := NewStressDashboardProcess(101)
|
||||
require.NoError(t, err)
|
||||
series := StressRSSSeries{Process: dashboard, EndRSSBytes: [3]uint64{100, 100, 100}}
|
||||
|
||||
for _, mutate := range []func([]StressRSSSeries) []StressRSSSeries{
|
||||
func(values []StressRSSSeries) []StressRSSSeries { return values[:1] },
|
||||
func(values []StressRSSSeries) []StressRSSSeries {
|
||||
agent, _ := NewStressAgentOrdinal(999)
|
||||
process, _ := NewStressAgentProcess(agent, 999)
|
||||
return append(values, StressRSSSeries{Process: process})
|
||||
},
|
||||
func(values []StressRSSSeries) []StressRSSSeries { return append(values, values[0]) },
|
||||
} {
|
||||
values := make([]StressRSSSeries, 0, profile.AgentCount()+1)
|
||||
values = append(values, series)
|
||||
for index := 1; index <= profile.AgentCount(); index++ {
|
||||
agent, agentErr := NewStressAgentOrdinal(index)
|
||||
require.NoError(t, agentErr)
|
||||
process, processErr := NewStressAgentProcess(agent, 200+index)
|
||||
require.NoError(t, processErr)
|
||||
values = append(values, StressRSSSeries{Process: process, EndRSSBytes: [3]uint64{100, 100, 100}})
|
||||
}
|
||||
err = ValidateStressSoakTrendForProfile(profile, StressSoakTrendEvidence{Series: mutate(values)})
|
||||
require.Error(t, err)
|
||||
}
|
||||
}
|
||||
|
||||
func mustProfile(t *testing.T, name contract.ProfileName) contract.Profile {
|
||||
t.Helper()
|
||||
profile, err := contract.ProfileByName(string(name))
|
||||
require.NoError(t, err)
|
||||
return profile
|
||||
}
|
||||
@@ -0,0 +1,87 @@
|
||||
//go:build linux && agentcompat
|
||||
|
||||
package scenario
|
||||
|
||||
import (
|
||||
"context"
|
||||
"errors"
|
||||
|
||||
"github.com/nezhahq/nezha/integration/agentcompat/internal/client"
|
||||
processharness "github.com/nezhahq/nezha/integration/agentcompat/internal/process"
|
||||
)
|
||||
|
||||
var ErrStressSQLiteJournalNotDrained = errors.New("stress dashboard sqlite journal is not drained")
|
||||
|
||||
type stressSQLiteHoldControl interface {
|
||||
ArmSQLiteHold(context.Context) (client.SQLiteHoldReceipt, error)
|
||||
WaitForSQLiteHold(context.Context, client.SQLiteHoldReceipt, client.SQLiteHoldState) (client.SQLiteHoldReceipt, error)
|
||||
ReleaseSQLiteHold(context.Context, client.SQLiteHoldReceipt) (client.SQLiteHoldReceipt, error)
|
||||
AbortSQLiteHold(context.Context, client.SQLiteHoldReceipt) (client.SQLiteHoldReceipt, error)
|
||||
}
|
||||
|
||||
type stressSQLiteJournalWatch interface {
|
||||
Wait(context.Context) error
|
||||
Close() error
|
||||
}
|
||||
|
||||
type stressSQLiteJournalWatchOpener func(string) (stressSQLiteJournalWatch, error)
|
||||
|
||||
func drainStressSQLiteJournal(ctx context.Context, control stressSQLiteHoldControl, writer func(context.Context) error, journalPath string, openWatch stressSQLiteJournalWatchOpener) error {
|
||||
receipt, err := control.ArmSQLiteHold(ctx)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
writerContext, cancelWriter := context.WithCancel(ctx)
|
||||
writerDone := make(chan error, 1)
|
||||
go func() { writerDone <- writer(writerContext) }()
|
||||
abort := func(cause error) error {
|
||||
_, abortErr := control.AbortSQLiteHold(context.WithoutCancel(ctx), receipt)
|
||||
cancelWriter()
|
||||
return errors.Join(cause, abortErr, <-writerDone)
|
||||
}
|
||||
selected, err := control.WaitForSQLiteHold(ctx, receipt, client.SQLiteHoldStateSelected)
|
||||
if err != nil {
|
||||
return abort(err)
|
||||
}
|
||||
finalizing, err := control.WaitForSQLiteHold(ctx, selected, client.SQLiteHoldStateFinalizing)
|
||||
if err != nil {
|
||||
return abort(err)
|
||||
}
|
||||
watch, err := openWatch(journalPath)
|
||||
if err != nil {
|
||||
return abort(err)
|
||||
}
|
||||
if _, err := control.ReleaseSQLiteHold(ctx, finalizing); err != nil {
|
||||
return errors.Join(abort(err), watch.Close())
|
||||
}
|
||||
waitErr := watch.Wait(ctx)
|
||||
if waitErr != nil {
|
||||
cancelWriter()
|
||||
}
|
||||
writerErr := <-writerDone
|
||||
cancelWriter()
|
||||
return errors.Join(waitErr, writerErr, watch.Close())
|
||||
}
|
||||
|
||||
func drainStressDashboardSQLiteJournal(ctx context.Context, fixture *heldSessionSetRealFixture) error {
|
||||
journalPath := fixture.dashboard.DatabasePath() + "-journal"
|
||||
return drainStressSQLiteJournal(ctx, fixture.controlPAT.Client, func(writerContext context.Context) error {
|
||||
_, err := fixture.controlPAT.Client.IOStreamState(writerContext)
|
||||
return err
|
||||
}, journalPath, func(path string) (stressSQLiteJournalWatch, error) {
|
||||
return processharness.OpenSQLiteJournalWatch(path)
|
||||
})
|
||||
}
|
||||
|
||||
func observeStressDashboardSQLiteJournal(path string) func(context.Context, processharness.Sample) error {
|
||||
return func(_ context.Context, sample processharness.Sample) error {
|
||||
held, err := processharness.ProcessHasOpenPath(sample.PID, path)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
if held {
|
||||
return ErrStressSQLiteJournalNotDrained
|
||||
}
|
||||
return nil
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,174 @@
|
||||
//go:build linux && agentcompat
|
||||
|
||||
package scenario
|
||||
|
||||
import (
|
||||
"context"
|
||||
"errors"
|
||||
"os"
|
||||
"path/filepath"
|
||||
"sync"
|
||||
"testing"
|
||||
|
||||
"github.com/stretchr/testify/require"
|
||||
|
||||
"github.com/nezhahq/nezha/integration/agentcompat/internal/client"
|
||||
processharness "github.com/nezhahq/nezha/integration/agentcompat/internal/process"
|
||||
)
|
||||
|
||||
type stressSQLiteHoldControlProbe struct {
|
||||
mu sync.Mutex
|
||||
calls []string
|
||||
started chan struct{}
|
||||
done chan struct{}
|
||||
once sync.Once
|
||||
}
|
||||
|
||||
func newStressSQLiteHoldControlProbe() *stressSQLiteHoldControlProbe {
|
||||
return &stressSQLiteHoldControlProbe{started: make(chan struct{}), done: make(chan struct{})}
|
||||
}
|
||||
|
||||
func (probe *stressSQLiteHoldControlProbe) record(call string) {
|
||||
probe.mu.Lock()
|
||||
defer probe.mu.Unlock()
|
||||
probe.calls = append(probe.calls, call)
|
||||
}
|
||||
|
||||
func (probe *stressSQLiteHoldControlProbe) ArmSQLiteHold(context.Context) (client.SQLiteHoldReceipt, error) {
|
||||
probe.record("arm")
|
||||
return client.SQLiteHoldReceipt{ID: "ERERERERERERERERERERERERERERERERERERERERERE", State: client.SQLiteHoldStateArmed}, nil
|
||||
}
|
||||
|
||||
func (probe *stressSQLiteHoldControlProbe) WaitForSQLiteHold(ctx context.Context, receipt client.SQLiteHoldReceipt, target client.SQLiteHoldState) (client.SQLiteHoldReceipt, error) {
|
||||
if target == client.SQLiteHoldStateSelected {
|
||||
select {
|
||||
case <-probe.started:
|
||||
case <-ctx.Done():
|
||||
return client.SQLiteHoldReceipt{}, ctx.Err()
|
||||
}
|
||||
}
|
||||
probe.record("wait-" + string(target))
|
||||
receipt.State = target
|
||||
return receipt, nil
|
||||
}
|
||||
|
||||
func (probe *stressSQLiteHoldControlProbe) ReleaseSQLiteHold(context.Context, client.SQLiteHoldReceipt) (client.SQLiteHoldReceipt, error) {
|
||||
probe.record("release")
|
||||
probe.once.Do(func() { close(probe.done) })
|
||||
return client.SQLiteHoldReceipt{State: client.SQLiteHoldStateReleased}, nil
|
||||
}
|
||||
|
||||
func (probe *stressSQLiteHoldControlProbe) AbortSQLiteHold(context.Context, client.SQLiteHoldReceipt) (client.SQLiteHoldReceipt, error) {
|
||||
probe.record("abort")
|
||||
probe.once.Do(func() { close(probe.done) })
|
||||
return client.SQLiteHoldReceipt{State: client.SQLiteHoldStateAborted}, nil
|
||||
}
|
||||
|
||||
type stressSQLiteJournalWatchProbe struct {
|
||||
control *stressSQLiteHoldControlProbe
|
||||
}
|
||||
|
||||
func (watch stressSQLiteJournalWatchProbe) Wait(ctx context.Context) error {
|
||||
select {
|
||||
case <-watch.control.done:
|
||||
watch.control.record("watch-wait")
|
||||
return nil
|
||||
case <-ctx.Done():
|
||||
return ctx.Err()
|
||||
}
|
||||
}
|
||||
|
||||
func (watch stressSQLiteJournalWatchProbe) Close() error {
|
||||
watch.control.record("watch-close")
|
||||
return nil
|
||||
}
|
||||
|
||||
func TestDrainStressSQLiteJournalOrdersHoldLifecycleBeforeCompletion(t *testing.T) {
|
||||
// Given
|
||||
control := newStressSQLiteHoldControlProbe()
|
||||
journalPath := filepath.Join(t.TempDir(), "dashboard.sqlite-journal")
|
||||
|
||||
// When
|
||||
err := drainStressSQLiteJournal(t.Context(), control, func(ctx context.Context) error {
|
||||
control.record("writer-start")
|
||||
close(control.started)
|
||||
select {
|
||||
case <-control.done:
|
||||
control.record("writer-complete")
|
||||
return nil
|
||||
case <-ctx.Done():
|
||||
return ctx.Err()
|
||||
}
|
||||
}, journalPath, func(path string) (stressSQLiteJournalWatch, error) {
|
||||
require.Equal(t, journalPath, path)
|
||||
control.record("watch-open")
|
||||
return stressSQLiteJournalWatchProbe{control: control}, nil
|
||||
})
|
||||
|
||||
// Then
|
||||
require.NoError(t, err)
|
||||
control.mu.Lock()
|
||||
calls := append([]string(nil), control.calls...)
|
||||
control.mu.Unlock()
|
||||
require.Less(t, callIndex(calls, "arm"), callIndex(calls, "writer-start"))
|
||||
require.Less(t, callIndex(calls, "writer-start"), callIndex(calls, "wait-selected"))
|
||||
require.Less(t, callIndex(calls, "wait-selected"), callIndex(calls, "wait-finalizing"))
|
||||
require.Less(t, callIndex(calls, "wait-finalizing"), callIndex(calls, "watch-open"))
|
||||
require.Less(t, callIndex(calls, "watch-open"), callIndex(calls, "release"))
|
||||
require.Less(t, callIndex(calls, "release"), callIndex(calls, "watch-wait"))
|
||||
require.Less(t, callIndex(calls, "release"), callIndex(calls, "writer-complete"))
|
||||
require.Less(t, callIndex(calls, "watch-wait"), callIndex(calls, "watch-close"))
|
||||
}
|
||||
|
||||
func TestDrainStressSQLiteJournalAbortsWriterWhenWatchCannotOpen(t *testing.T) {
|
||||
// Given
|
||||
control := newStressSQLiteHoldControlProbe()
|
||||
watchErr := errors.New("watch unavailable")
|
||||
|
||||
// When
|
||||
err := drainStressSQLiteJournal(t.Context(), control, func(ctx context.Context) error {
|
||||
close(control.started)
|
||||
select {
|
||||
case <-control.done:
|
||||
return nil
|
||||
case <-ctx.Done():
|
||||
return ctx.Err()
|
||||
}
|
||||
}, filepath.Join(t.TempDir(), "dashboard.sqlite-journal"), func(string) (stressSQLiteJournalWatch, error) {
|
||||
return nil, watchErr
|
||||
})
|
||||
|
||||
// Then
|
||||
require.ErrorIs(t, err, watchErr)
|
||||
control.mu.Lock()
|
||||
calls := append([]string(nil), control.calls...)
|
||||
control.mu.Unlock()
|
||||
require.Contains(t, calls, "abort")
|
||||
}
|
||||
|
||||
func TestObserveStressDashboardSQLiteJournalRejectsFirstHeldSample(t *testing.T) {
|
||||
// Given
|
||||
path := filepath.Join(t.TempDir(), "dashboard.sqlite-journal")
|
||||
require.NoError(t, os.WriteFile(path, []byte("journal"), 0o600))
|
||||
file, err := os.Open(path)
|
||||
require.NoError(t, err)
|
||||
observer := observeStressDashboardSQLiteJournal(path)
|
||||
|
||||
// When
|
||||
heldErr := observer(t.Context(), processharness.Sample{PID: os.Getpid()})
|
||||
require.NoError(t, file.Close())
|
||||
releasedErr := observer(t.Context(), processharness.Sample{PID: os.Getpid()})
|
||||
|
||||
// Then
|
||||
require.ErrorIs(t, heldErr, ErrStressSQLiteJournalNotDrained)
|
||||
require.NoError(t, releasedErr)
|
||||
}
|
||||
|
||||
func callIndex(calls []string, wanted string) int {
|
||||
for index, call := range calls {
|
||||
if call == wanted {
|
||||
return index
|
||||
}
|
||||
}
|
||||
return len(calls)
|
||||
}
|
||||
Reference in New Issue
Block a user