diff --git a/e2e/stress/README.md b/e2e/stress/README.md new file mode 100644 index 00000000..ddc5cfaa --- /dev/null +++ b/e2e/stress/README.md @@ -0,0 +1,150 @@ +# Eno stress runner + +`eno-stress` is a standalone binary for running YAML-defined workloads against a live Kubernetes cluster. It is stored under `e2e/`, but it is not invoked by the Eno e2e test suite. + +## Build + +Build the standalone client binary from the repository root: + +```bash +mkdir -p bin +go build -o bin/eno-stress ./e2e/stress/cmd/eno-stress +``` + +Build and push the synthesizer image to a registry reachable by the target cluster: + +```bash +docker build \ + -f e2e/stress/synthesizer/Dockerfile \ + -t REGISTRY/eno-stress-synthesizer:TAG . +docker push REGISTRY/eno-stress-synthesizer:TAG +``` + +The ACR image is referenced by `spec.run.variables.synthesizerImage` in +`e2e/stress/plans/missing-input-fanout/plan.yaml`: + +```yaml +spec: + run: + variables: + synthesizerImage: MYACR.azurecr.io/eno-stress-synthesizer:TAG +``` + +The Synthesizer template consumes that value as `spec.image`: + +```yaml +spec: + image: "${synthesizerImage}" +``` + +The target cluster must be able to pull the image from ACR, either through its +managed identity or its configured image-pull credentials. + +## Completion stages + +Set `spec.run.completionStage` to control when `run` stops and writes its +report: + +```yaml +spec: + run: + completionStage: PreSynthesis +``` + +- `PreSynthesis`: every Composition has observed the acknowledged input + revisions and exited `MissingInputs`. +- `PostSynthesis`: `PreSynthesis` plus every Composition has a non-empty + `status.*Synthesis.synthesized` timestamp. +- `Reconcile`: `PostSynthesis` plus every Composition has a non-empty + `status.currentSynthesis.reconciled` timestamp. This does not wait for + resource readiness. + +Later stages include all measurements from earlier stages. `Reconcile` is the +default when `completionStage` is omitted. + +## Reusable setup resources + +Setup resources can opt into reuse: + +```yaml +setup: + resources: + - name: synthesizer + reuse: true +``` + +When a reusable resource already exists, the runner performs a server-side +dry-run update and compares the resulting `spec` with the existing `spec`. An +unchanged resource is reused without relabeling or modifying it. A changed +image, command, ref, or other spec field fails preparation explicitly. + +## Namespace naming + +The included plan sets `namespacePrefix: 6a`. Each namespace is named `6a` +followed immediately by 16 lowercase hexadecimal characters, for example +`6a9e3a86214e85de76`. The suffix is deterministically derived from the run ID +and namespace index so all 150 namespaces are unique and resuming `prepare` +uses the same names. + +## Symphony topology + +For every generated namespace, `prepare` creates one Symphony from +`plans/missing-input-fanout/manifests/symphony.yaml`. The Symphony has 20 +variations referencing 20 Synthesizer objects that all use the same image. +Every variation inherits the same three input bindings and creates one unique +output ConfigMap. This produces 3,000 Compositions and outputs across 150 +namespaces. The plan declares the generated Compositions with +`operation: observe`, so the runner records controller-generated objects +instead of creating duplicate Compositions. + +The 20-variation plan uses state schema `v1alpha2`. Do not reuse a state file +created by an older stress binary or by the one-Composition plan. + +## Run against a live cluster + +```bash +bin/eno-stress validate \ + --plan e2e/stress/plans/missing-input-fanout/plan.yaml \ + --kubeconfig "$KUBECONFIG" + +bin/eno-stress prepare \ + --plan e2e/stress/plans/missing-input-fanout/plan.yaml \ + --state /tmp/eno-stress-state.json \ + --kubeconfig "$KUBECONFIG" + +bin/eno-stress run \ + --state /tmp/eno-stress-state.json \ + --kubeconfig "$KUBECONFIG" +``` + +`prepare` creates namespaces, the Synthesizer, and Compositions, then waits for every Composition to report `MissingInputs`. It never creates test inputs. `run` reloads the state, verifies that Compositions remain in `MissingInputs`, establishes watches, and only then creates inputs. + +Use `status` to inspect partial results and `cleanup` to delete only the test namespaces recorded in state. Namespace deletion cascades namespaced Symphonies, Compositions, inputs, and outputs. Cluster-scoped Synthesizers are retained. Cleanup requires both the recorded run label and UID before deleting a namespace. + +```bash +bin/eno-stress status --state /tmp/eno-stress-state.json +bin/eno-stress cleanup --state /tmp/eno-stress-state.json --kubeconfig "$KUBECONFIG" +``` + +The generated JSON report is written relative to the plan directory unless `metrics.reportFile` is absolute. The state file is written atomically with mode `0600` and should not be committed. + +## Repeated cycles + +`run-cycles.sh` validates once, waits for all generated `6a<16 hex>` +namespaces to disappear, and runs `prepare`, `run`, namespace-only `cleanup`, +and deletion waiting for each cycle. Full command output is appended to +`results.txt`, and each state file is moved to `state-history/`. + +```bash +CYCLES=5 ./run-cycles.sh +``` + +For a session that should survive disconnects: + +```bash +nohup env CYCLES=5 ./run-cycles.sh >> cycle-console.log 2>&1 & +echo $! > cycle.pid +``` + +Settings can be overridden with `CYCLES`, `WAIT_SECONDS`, `KUBECONFIG`, +`ENO_STRESS`, `PLAN`, `STATE`, `RESULTS_FILE`, and `STATE_HISTORY`. \ No newline at end of file diff --git a/e2e/stress/cmd/eno-stress/main.go b/e2e/stress/cmd/eno-stress/main.go new file mode 100644 index 00000000..1233b2f7 --- /dev/null +++ b/e2e/stress/cmd/eno-stress/main.go @@ -0,0 +1,79 @@ +// Command eno-stress executes YAML-driven Eno stress plans against a live cluster. +package main + +import ( + "context" + "flag" + "fmt" + "os" + "os/signal" + "syscall" + + "github.com/Azure/eno/e2e/stress/internal/runner" +) + +func main() { + ctx, cancel := signal.NotifyContext(context.Background(), os.Interrupt, syscall.SIGTERM) + defer cancel() + if err := execute(ctx, os.Args[1:]); err != nil { + fmt.Fprintf(os.Stderr, "eno-stress: %v\n", err) + os.Exit(1) + } +} + +func execute(ctx context.Context, arguments []string) error { + if len(arguments) == 0 { + return usageError() + } + command := arguments[0] + set := flag.NewFlagSet(command, flag.ContinueOnError) + set.SetOutput(os.Stderr) + options := runner.Options{Output: func(format string, values ...any) { fmt.Printf(format, values...) }} + set.StringVar(&options.Kubeconfig, "kubeconfig", "", "path to the kubeconfig for the live cluster (defaults to standard kubeconfig loading)") + + switch command { + case "validate": + set.StringVar(&options.PlanPath, "plan", "", "path to the stress test plan") + set.BoolVar(&options.ServerDryRun, "server-dry-run", false, "send rendered resources to Kubernetes server-side dry-run") + case "prepare": + set.StringVar(&options.PlanPath, "plan", "", "path to the stress test plan") + set.StringVar(&options.StatePath, "state", "state.json", "path to durable run state") + set.StringVar(&options.RunID, "run-id", "", "optional stable run ID") + case "run", "status", "cleanup": + set.StringVar(&options.StatePath, "state", "state.json", "path to durable run state") + default: + return usageError() + } + set.Usage = func() { + fmt.Fprintf(set.Output(), "usage: eno-stress %s [flags]\n", command) + set.PrintDefaults() + } + if err := set.Parse(arguments[1:]); err != nil { + return err + } + if set.NArg() != 0 { + return fmt.Errorf("unexpected arguments: %v", set.Args()) + } + if (command == "validate" || command == "prepare") && options.PlanPath == "" { + return fmt.Errorf("--plan is required") + } + + switch command { + case "validate": + return runner.Validate(ctx, options) + case "prepare": + return runner.Prepare(ctx, options) + case "run": + return runner.Run(ctx, options) + case "status": + return runner.Status(options) + case "cleanup": + return runner.Cleanup(ctx, options) + default: + return usageError() + } +} + +func usageError() error { + return fmt.Errorf("usage: eno-stress [flags]") +} diff --git a/e2e/stress/internal/kube/client.go b/e2e/stress/internal/kube/client.go new file mode 100644 index 00000000..36b31570 --- /dev/null +++ b/e2e/stress/internal/kube/client.go @@ -0,0 +1,57 @@ +package kube + +import ( + "fmt" + + "k8s.io/apimachinery/pkg/api/meta" + "k8s.io/client-go/discovery" + "k8s.io/client-go/dynamic" + "k8s.io/client-go/rest" + "k8s.io/client-go/restmapper" + "k8s.io/client-go/tools/clientcmd" +) + +type Client struct { + Config *rest.Config + Dynamic dynamic.Interface + Discovery discovery.DiscoveryInterface + Mapper meta.RESTMapper +} + +func New(kubeconfig string, qps float32, burst int) (*Client, error) { + loadingRules := clientcmd.NewDefaultClientConfigLoadingRules() + if kubeconfig != "" { + loadingRules.ExplicitPath = kubeconfig + } + + config, err := clientcmd.NewNonInteractiveDeferredLoadingClientConfig( + loadingRules, + &clientcmd.ConfigOverrides{}, + ).ClientConfig() + if err != nil { + return nil, fmt.Errorf("loading kubeconfig: %w", err) + } + config.QPS = qps + config.Burst = burst + config.UserAgent = "eno-stress" + + dynamicClient, err := dynamic.NewForConfig(config) + if err != nil { + return nil, fmt.Errorf("creating dynamic client: %w", err) + } + discoveryClient, err := discovery.NewDiscoveryClientForConfig(config) + if err != nil { + return nil, fmt.Errorf("creating discovery client: %w", err) + } + resources, err := restmapper.GetAPIGroupResources(discoveryClient) + if err != nil { + return nil, fmt.Errorf("discovering API resources: %w", err) + } + + return &Client{ + Config: config, + Dynamic: dynamicClient, + Discovery: discoveryClient, + Mapper: restmapper.NewDiscoveryRESTMapper(resources), + }, nil +} diff --git a/e2e/stress/internal/kube/resources.go b/e2e/stress/internal/kube/resources.go new file mode 100644 index 00000000..10c54359 --- /dev/null +++ b/e2e/stress/internal/kube/resources.go @@ -0,0 +1,128 @@ +// Dynamic resource operations used by the stress runner. +package kube + +import ( + "context" + "encoding/json" + "fmt" + "reflect" + + apierrors "k8s.io/apimachinery/pkg/api/errors" + metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" + "k8s.io/apimachinery/pkg/apis/meta/v1/unstructured" + "k8s.io/apimachinery/pkg/runtime/schema" + "k8s.io/apimachinery/pkg/types" + "k8s.io/client-go/dynamic" + + "github.com/Azure/eno/e2e/stress/internal/render" +) + +type Resource struct { + Interface dynamic.ResourceInterface + GVR schema.GroupVersionResource + Scope string +} + +func (c *Client) ResourceFor(object *unstructured.Unstructured) (*Resource, error) { + gvk := object.GroupVersionKind() + mapping, err := c.Mapper.RESTMapping(gvk.GroupKind(), gvk.Version) + if err != nil { + return nil, fmt.Errorf("mapping %s: %w", gvk.String(), err) + } + resource := c.Dynamic.Resource(mapping.Resource) + if mapping.Scope.Name() == "namespace" { + return &Resource{Interface: resource.Namespace(object.GetNamespace()), GVR: mapping.Resource, Scope: "namespace"}, nil + } + return &Resource{Interface: resource, GVR: mapping.Resource, Scope: "cluster"}, nil +} + +func (c *Client) CreateOwned(ctx context.Context, object *unstructured.Unstructured, runID string, dryRun bool) (*unstructured.Unstructured, *Resource, error) { + return c.CreateSetup(ctx, object, runID, dryRun, false) +} + +func (c *Client) CreateSetup(ctx context.Context, object *unstructured.Unstructured, runID string, dryRun, reuse bool) (*unstructured.Unstructured, *Resource, error) { + resource, err := c.ResourceFor(object) + if err != nil { + return nil, nil, err + } + options := metav1.CreateOptions{FieldManager: "eno-stress"} + if dryRun { + options.DryRun = []string{metav1.DryRunAll} + } + created, err := resource.Interface.Create(ctx, object, options) + if err == nil { + return created, resource, nil + } + if !apierrors.IsAlreadyExists(err) || dryRun { + return nil, nil, err + } + existing, getErr := resource.Interface.Get(ctx, object.GetName(), metav1.GetOptions{}) + if getErr != nil { + return nil, nil, fmt.Errorf("getting existing resource after create conflict: %w", getErr) + } + if existing.GetLabels()[render.RunIDLabel] == runID { + return existing, resource, nil + } + if !reuse { + return nil, nil, fmt.Errorf("refusing to adopt unowned %s %s/%s", object.GetKind(), object.GetNamespace(), object.GetName()) + } + candidate := object.DeepCopy() + candidate.SetResourceVersion(existing.GetResourceVersion()) + validated, updateErr := resource.Interface.Update(ctx, candidate, metav1.UpdateOptions{ + DryRun: []string{metav1.DryRunAll}, + FieldManager: "eno-stress-reuse-check", + }) + if updateErr != nil { + return nil, nil, fmt.Errorf("validating reusable %s %s/%s: %w", object.GetKind(), object.GetNamespace(), object.GetName(), updateErr) + } + existingSpec, _, _ := unstructured.NestedFieldNoCopy(existing.Object, "spec") + validatedSpec, _, _ := unstructured.NestedFieldNoCopy(validated.Object, "spec") + if !reflect.DeepEqual(existingSpec, validatedSpec) { + return nil, nil, fmt.Errorf("refusing to reuse changed %s %s/%s", object.GetKind(), object.GetNamespace(), object.GetName()) + } + return existing, resource, nil +} + +func (c *Client) ApplyOwned(ctx context.Context, object *unstructured.Unstructured, runID string) (*unstructured.Unstructured, *Resource, error) { + resource, err := c.ResourceFor(object) + if err != nil { + return nil, nil, err + } + existing, err := resource.Interface.Get(ctx, object.GetName(), metav1.GetOptions{}) + if err == nil && existing.GetLabels()[render.RunIDLabel] != runID { + return nil, nil, fmt.Errorf("refusing to update unowned %s %s/%s", object.GetKind(), object.GetNamespace(), object.GetName()) + } + if err != nil && !apierrors.IsNotFound(err) { + return nil, nil, err + } + raw, err := json.Marshal(object.Object) + if err != nil { + return nil, nil, err + } + force := true + applied, err := resource.Interface.Patch(ctx, object.GetName(), types.ApplyPatchType, raw, metav1.PatchOptions{ + FieldManager: "eno-stress", + Force: &force, + }) + return applied, resource, err +} + +func (c *Client) DeleteOwned(ctx context.Context, gvr schema.GroupVersionResource, namespace, name, uid, runID string) error { + resource := c.Dynamic.Resource(gvr) + var resourceInterface dynamic.ResourceInterface = resource + if namespace != "" { + resourceInterface = resource.Namespace(namespace) + } + existing, err := resourceInterface.Get(ctx, name, metav1.GetOptions{}) + if apierrors.IsNotFound(err) { + return nil + } + if err != nil { + return err + } + if existing.GetLabels()[render.RunIDLabel] != runID || string(existing.GetUID()) != uid { + return fmt.Errorf("refusing to delete %s/%s because ownership or UID changed", namespace, name) + } + uidValue := types.UID(uid) + return resourceInterface.Delete(ctx, name, metav1.DeleteOptions{Preconditions: &metav1.Preconditions{UID: &uidValue}}) +} diff --git a/e2e/stress/internal/kube/watch.go b/e2e/stress/internal/kube/watch.go new file mode 100644 index 00000000..58e0cca8 --- /dev/null +++ b/e2e/stress/internal/kube/watch.go @@ -0,0 +1,93 @@ +// Restartable watch support for live-cluster measurements. +package kube + +import ( + "context" + "fmt" + "time" + + metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" + "k8s.io/apimachinery/pkg/apis/meta/v1/unstructured" + "k8s.io/apimachinery/pkg/runtime/schema" + "k8s.io/apimachinery/pkg/watch" + "k8s.io/client-go/dynamic" +) + +type WatchHandler func(*unstructured.Unstructured, time.Time) + +func (c *Client) Watch(ctx context.Context, gvr schema.GroupVersionResource, namespace, labelSelector string, ready chan<- struct{}, handler WatchHandler) error { + var signalReady = ready + for ctx.Err() == nil { + resource := c.Dynamic.Resource(gvr) + var resourceInterface dynamic.ResourceInterface = resource + if namespace != "" { + resourceInterface = resource.Namespace(namespace) + } + list, err := resourceInterface.List(ctx, metav1.ListOptions{LabelSelector: labelSelector}) + if err != nil { + if !waitForRetry(ctx) { + break + } + continue + } + observedAt := time.Now() + for index := range list.Items { + handler(&list.Items[index], observedAt) + } + + stream, err := resourceInterface.Watch(ctx, metav1.ListOptions{ + LabelSelector: labelSelector, + ResourceVersion: list.GetResourceVersion(), + }) + if err != nil { + if !waitForRetry(ctx) { + break + } + continue + } + if signalReady != nil { + close(signalReady) + signalReady = nil + } + watchEnded := consumeWatch(ctx, stream, handler) + stream.Stop() + if !watchEnded { + break + } + } + if ctx.Err() != nil { + return ctx.Err() + } + return fmt.Errorf("watch for %s stopped", gvr.String()) +} + +func consumeWatch(ctx context.Context, stream watch.Interface, handler WatchHandler) bool { + for { + select { + case <-ctx.Done(): + return false + case event, open := <-stream.ResultChan(): + if !open { + return true + } + if event.Type == watch.Error { + return true + } + object, ok := event.Object.(*unstructured.Unstructured) + if ok { + handler(object, time.Now()) + } + } + } +} + +func waitForRetry(ctx context.Context) bool { + timer := time.NewTimer(250 * time.Millisecond) + defer timer.Stop() + select { + case <-ctx.Done(): + return false + case <-timer.C: + return true + } +} diff --git a/e2e/stress/internal/plan/load.go b/e2e/stress/internal/plan/load.go new file mode 100644 index 00000000..5d1ee110 --- /dev/null +++ b/e2e/stress/internal/plan/load.go @@ -0,0 +1,30 @@ +package plan + +import ( + "bytes" + "fmt" + "os" + "path/filepath" + + "gopkg.in/yaml.v3" +) + +func Load(path string) (*Plan, error) { + raw, err := os.ReadFile(path) + if err != nil { + return nil, fmt.Errorf("reading plan: %w", err) + } + + decoder := yaml.NewDecoder(bytes.NewReader(raw)) + decoder.KnownFields(true) + + result := &Plan{} + if err := decoder.Decode(result); err != nil { + return nil, fmt.Errorf("decoding plan: %w", err) + } + result.ApplyDefaults() + if err := result.Validate(filepath.Dir(path)); err != nil { + return nil, err + } + return result, nil +} diff --git a/e2e/stress/internal/plan/load_test.go b/e2e/stress/internal/plan/load_test.go new file mode 100644 index 00000000..94a4d372 --- /dev/null +++ b/e2e/stress/internal/plan/load_test.go @@ -0,0 +1,74 @@ +package plan + +import ( + "os" + "path/filepath" + "testing" + + "github.com/stretchr/testify/require" +) + +func TestLoadRejectsUnknownFields(t *testing.T) { + path := writePlan(t, validPlan+"\n unknown: true\n") + _, err := Load(path) + require.ErrorContains(t, err, "field unknown not found") +} + +func TestLoadAppliesDefaults(t *testing.T) { + path := writePlan(t, validPlan) + loaded, err := Load(path) + require.NoError(t, err) + require.Equal(t, 4, loaded.Spec.Run.Concurrency) + require.Equal(t, 3, loaded.Spec.Run.BatchSize) + require.Equal(t, "create", loaded.Spec.Test.Phases[0].Resources[0].Operation) +} + +func writePlan(t *testing.T, contents string) string { + t.Helper() + dir := t.TempDir() + path := filepath.Join(dir, "plan.yaml") + require.NoError(t, os.WriteFile(path, []byte(contents), 0o600)) + return path +} + +const validPlan = `apiVersion: stress.eno.azure.io/v1alpha1 +kind: StressTestPlan +metadata: + name: test +spec: + run: + namespacePrefix: eno-stress + namespaceCount: 3 + concurrency: 4 + timeout: 1m + setup: + resources: + - name: composition + scope: namespace + forEachNamespace: true + template: + apiVersion: eno.azure.io/v1 + kind: Composition + metadata: + name: test + readiness: + - resource: composition + condition: + status: MissingInputs + test: + phases: + - name: inputs + resources: + - name: input + scope: namespace + forEachNamespace: true + template: + apiVersion: v1 + kind: ConfigMap + metadata: + name: input + output: + kind: ConfigMap + name: output + expectedData: + value: ok` diff --git a/e2e/stress/internal/plan/types.go b/e2e/stress/internal/plan/types.go new file mode 100644 index 00000000..7d8b9903 --- /dev/null +++ b/e2e/stress/internal/plan/types.go @@ -0,0 +1,165 @@ +package plan + +import ( + "fmt" + "time" +) + +const ( + APIVersion = "stress.eno.azure.io/v1alpha1" + Kind = "StressTestPlan" + + CompletionStagePreSynthesis CompletionStage = "PreSynthesis" + CompletionStagePostSynthesis CompletionStage = "PostSynthesis" + CompletionStageReconcile CompletionStage = "Reconcile" +) + +type CompletionStage string + +type Plan struct { + APIVersion string `yaml:"apiVersion" json:"apiVersion"` + Kind string `yaml:"kind" json:"kind"` + Metadata Metadata `yaml:"metadata" json:"metadata"` + Spec Spec `yaml:"spec" json:"spec"` +} + +type Metadata struct { + Name string `yaml:"name" json:"name"` +} + +type Spec struct { + Run RunSpec `yaml:"run" json:"run"` + Setup SetupSpec `yaml:"setup" json:"setup"` + Test TestSpec `yaml:"test" json:"test"` + Output OutputSpec `yaml:"output" json:"output"` + Metrics MetricsSpec `yaml:"metrics,omitempty" json:"metrics,omitempty"` +} + +type RunSpec struct { + NamespacePrefix string `yaml:"namespacePrefix" json:"namespacePrefix"` + NamespaceCount int `yaml:"namespaceCount" json:"namespaceCount"` + CompletionStage CompletionStage `yaml:"completionStage,omitempty" json:"completionStage,omitempty"` + Concurrency int `yaml:"concurrency" json:"concurrency"` + BatchSize int `yaml:"batchSize,omitempty" json:"batchSize,omitempty"` + BatchDelay string `yaml:"batchDelay,omitempty" json:"batchDelay,omitempty"` + Repetitions int `yaml:"repetitions,omitempty" json:"repetitions,omitempty"` + Timeout string `yaml:"timeout" json:"timeout"` + ClientQPS float32 `yaml:"clientQPS,omitempty" json:"clientQPS,omitempty"` + ClientBurst int `yaml:"clientBurst,omitempty" json:"clientBurst,omitempty"` + Labels map[string]string `yaml:"labels,omitempty" json:"labels,omitempty"` + Variables map[string]string `yaml:"variables,omitempty" json:"variables,omitempty"` +} + +type SetupSpec struct { + Resources []ResourceSpec `yaml:"resources" json:"resources"` + Readiness []ReadinessSpec `yaml:"readiness" json:"readiness"` +} + +type ResourceSpec struct { + Name string `yaml:"name" json:"name"` + Count int `yaml:"count,omitempty" json:"count,omitempty"` + Reuse bool `yaml:"reuse,omitempty" json:"reuse,omitempty"` + APIVersion string `yaml:"apiVersion,omitempty" json:"apiVersion,omitempty"` + Kind string `yaml:"kind,omitempty" json:"kind,omitempty"` + Scope string `yaml:"scope,omitempty" json:"scope,omitempty"` + ForEachNamespace bool `yaml:"forEachNamespace,omitempty" json:"forEachNamespace,omitempty"` + TemplateFile string `yaml:"templateFile,omitempty" json:"templateFile,omitempty"` + Template map[string]any `yaml:"template,omitempty" json:"template,omitempty"` + DependsOn []string `yaml:"dependsOn,omitempty" json:"dependsOn,omitempty"` + Operation string `yaml:"operation,omitempty" json:"operation,omitempty"` +} + +func (r ResourceSpec) ExpandedCount() int { + if r.Count > 0 { + return r.Count + } + return 1 +} + +type ReadinessSpec struct { + Resource string `yaml:"resource" json:"resource"` + Condition ReadinessCondition `yaml:"condition" json:"condition"` +} + +type ReadinessCondition struct { + Status string `yaml:"status" json:"status"` + ExpectedMissingInputs []string `yaml:"expectedMissingInputs,omitempty" json:"expectedMissingInputs,omitempty"` +} + +type TestSpec struct { + Phases []PhaseSpec `yaml:"phases" json:"phases"` + SuccessCriteria []SuccessCriterion `yaml:"successCriteria" json:"successCriteria"` +} + +type PhaseSpec struct { + Name string `yaml:"name" json:"name"` + Concurrency int `yaml:"concurrency,omitempty" json:"concurrency,omitempty"` + BatchSize int `yaml:"batchSize,omitempty" json:"batchSize,omitempty"` + Delay string `yaml:"delay,omitempty" json:"delay,omitempty"` + Repetitions int `yaml:"repetitions,omitempty" json:"repetitions,omitempty"` + DependsOn []string `yaml:"dependsOn,omitempty" json:"dependsOn,omitempty"` + Resources []ResourceSpec `yaml:"resources" json:"resources"` +} + +type SuccessCriterion map[string]string + +type OutputSpec struct { + APIVersion string `yaml:"apiVersion,omitempty" json:"apiVersion,omitempty"` + Kind string `yaml:"kind" json:"kind"` + Name string `yaml:"name" json:"name"` + ExpectedData map[string]string `yaml:"expectedData" json:"expectedData"` +} + +type MetricsSpec struct { + Collect []string `yaml:"collect,omitempty" json:"collect,omitempty"` + ReportFile string `yaml:"reportFile,omitempty" json:"reportFile,omitempty"` +} + +func (p *Plan) ApplyDefaults() { + if p.Spec.Run.CompletionStage == "" { + p.Spec.Run.CompletionStage = CompletionStageReconcile + } + if p.Spec.Run.Concurrency == 0 { + p.Spec.Run.Concurrency = 1 + } + if p.Spec.Run.BatchSize == 0 { + p.Spec.Run.BatchSize = p.Spec.Run.NamespaceCount + } + if p.Spec.Run.Repetitions == 0 { + p.Spec.Run.Repetitions = 1 + } + if p.Spec.Run.ClientQPS == 0 { + p.Spec.Run.ClientQPS = float32(max(100, p.Spec.Run.Concurrency*2)) + } + if p.Spec.Run.ClientBurst == 0 { + p.Spec.Run.ClientBurst = max(200, p.Spec.Run.Concurrency*4) + } + if p.Spec.Output.APIVersion == "" { + p.Spec.Output.APIVersion = "v1" + } + for phaseIndex := range p.Spec.Test.Phases { + phase := &p.Spec.Test.Phases[phaseIndex] + if phase.Concurrency == 0 { + phase.Concurrency = p.Spec.Run.Concurrency + } + if phase.BatchSize == 0 { + phase.BatchSize = p.Spec.Run.BatchSize + } + if phase.Repetitions == 0 { + phase.Repetitions = p.Spec.Run.Repetitions + } + for resourceIndex := range phase.Resources { + if phase.Resources[resourceIndex].Operation == "" { + phase.Resources[resourceIndex].Operation = "create" + } + } + } +} + +func (p *Plan) Timeout() (time.Duration, error) { + timeout, err := time.ParseDuration(p.Spec.Run.Timeout) + if err != nil || timeout <= 0 { + return 0, fmt.Errorf("spec.run.timeout must be a positive duration: %q", p.Spec.Run.Timeout) + } + return timeout, nil +} diff --git a/e2e/stress/internal/plan/validate.go b/e2e/stress/internal/plan/validate.go new file mode 100644 index 00000000..c7bd588f --- /dev/null +++ b/e2e/stress/internal/plan/validate.go @@ -0,0 +1,159 @@ +package plan + +import ( + "fmt" + "os" + "path/filepath" + "slices" + "strings" + "time" + + "k8s.io/apimachinery/pkg/util/validation" +) + +func (p *Plan) Validate(baseDir string) error { + if p.APIVersion != APIVersion { + return fmt.Errorf("apiVersion must be %q", APIVersion) + } + if p.Kind != Kind { + return fmt.Errorf("kind must be %q", Kind) + } + if errs := validation.IsDNS1123Subdomain(p.Metadata.Name); len(errs) > 0 { + return fmt.Errorf("metadata.name is invalid: %s", strings.Join(errs, ", ")) + } + if errs := validation.IsDNS1123Label(p.Spec.Run.NamespacePrefix); len(errs) > 0 { + return fmt.Errorf("spec.run.namespacePrefix is invalid: %s", strings.Join(errs, ", ")) + } + if p.Spec.Run.NamespaceCount <= 0 { + return fmt.Errorf("spec.run.namespaceCount must be positive") + } + if !slices.Contains([]CompletionStage{CompletionStagePreSynthesis, CompletionStagePostSynthesis, CompletionStageReconcile}, p.Spec.Run.CompletionStage) { + return fmt.Errorf("spec.run.completionStage must be PreSynthesis, PostSynthesis, or Reconcile") + } + if p.Spec.Run.Concurrency <= 0 || p.Spec.Run.BatchSize <= 0 { + return fmt.Errorf("spec.run concurrency and batchSize must be positive") + } + if _, err := p.Timeout(); err != nil { + return err + } + if p.Spec.Run.BatchDelay != "" { + if delay, err := time.ParseDuration(p.Spec.Run.BatchDelay); err != nil || delay < 0 { + return fmt.Errorf("spec.run.batchDelay must be a non-negative duration") + } + } + + setupNames := map[string]struct{}{} + for index := range p.Spec.Setup.Resources { + resource := &p.Spec.Setup.Resources[index] + if resource.Operation != "" && resource.Operation != "create" && resource.Operation != "observe" { + return fmt.Errorf("setup resource %q has unsupported operation %q", resource.Name, resource.Operation) + } + if err := validateResource(baseDir, resource); err != nil { + return fmt.Errorf("setup resource %q: %w", resource.Name, err) + } + if _, found := setupNames[resource.Name]; found { + return fmt.Errorf("duplicate setup resource name %q", resource.Name) + } + setupNames[resource.Name] = struct{}{} + } + for _, resource := range p.Spec.Setup.Resources { + for _, dependency := range resource.DependsOn { + if _, found := setupNames[dependency]; !found { + return fmt.Errorf("setup resource %q depends on unknown resource %q", resource.Name, dependency) + } + } + } + for _, readiness := range p.Spec.Setup.Readiness { + if _, found := setupNames[readiness.Resource]; !found { + return fmt.Errorf("readiness references unknown resource %q", readiness.Resource) + } + } + + phaseNames := map[string]struct{}{} + for phaseIndex := range p.Spec.Test.Phases { + phase := &p.Spec.Test.Phases[phaseIndex] + if phase.Name == "" { + return fmt.Errorf("test phase %d has no name", phaseIndex) + } + if _, found := phaseNames[phase.Name]; found { + return fmt.Errorf("duplicate test phase name %q", phase.Name) + } + for _, dependency := range phase.DependsOn { + if _, found := phaseNames[dependency]; !found { + return fmt.Errorf("test phase %q depends on unknown or later phase %q", phase.Name, dependency) + } + } + phaseNames[phase.Name] = struct{}{} + resourceNames := map[string]struct{}{} + for resourceIndex := range phase.Resources { + resource := &phase.Resources[resourceIndex] + if resource.ExpandedCount() != 1 { + return fmt.Errorf("test phase %q resource %q cannot set count", phase.Name, resource.Name) + } + if err := validateResource(baseDir, resource); err != nil { + return fmt.Errorf("test phase %q resource %q: %w", phase.Name, resource.Name, err) + } + if _, found := resourceNames[resource.Name]; found { + return fmt.Errorf("duplicate resource name %q in phase %q", resource.Name, phase.Name) + } + resourceNames[resource.Name] = struct{}{} + if !slices.Contains([]string{"create", "apply", "update"}, resource.Operation) { + return fmt.Errorf("test phase %q resource %q has unsupported operation %q", phase.Name, resource.Name, resource.Operation) + } + } + } + if len(p.Spec.Test.Phases) == 0 { + return fmt.Errorf("spec.test.phases must not be empty") + } + if p.Spec.Output.Kind == "" || p.Spec.Output.Name == "" { + return fmt.Errorf("spec.output kind and name are required") + } + return nil +} + +func validateResource(baseDir string, resource *ResourceSpec) error { + if resource.Name == "" { + return fmt.Errorf("name is required") + } + if resource.Count < 0 { + return fmt.Errorf("count must not be negative") + } + if resource.Operation == "observe" { + if resource.TemplateFile != "" || len(resource.Template) != 0 { + return fmt.Errorf("observed resources cannot define templateFile or template") + } + if resource.APIVersion == "" || resource.Kind == "" { + return fmt.Errorf("observed resources require apiVersion and kind") + } + if resource.Scope != "cluster" && resource.Scope != "namespace" { + return fmt.Errorf("observed resources require scope cluster or namespace") + } + return nil + } + if resource.TemplateFile == "" && len(resource.Template) == 0 { + return fmt.Errorf("exactly one of templateFile or template is required") + } + if resource.TemplateFile != "" && len(resource.Template) != 0 { + return fmt.Errorf("templateFile and template are mutually exclusive") + } + if resource.Scope != "" && resource.Scope != "cluster" && resource.Scope != "namespace" { + return fmt.Errorf("scope must be cluster or namespace") + } + if resource.Scope == "cluster" && resource.ForEachNamespace { + return fmt.Errorf("cluster-scoped resources cannot set forEachNamespace") + } + if resource.TemplateFile != "" { + path := resource.TemplateFile + if !filepath.IsAbs(path) { + path = filepath.Join(baseDir, path) + } + info, err := os.Stat(path) + if err != nil { + return fmt.Errorf("template file: %w", err) + } + if info.IsDir() { + return fmt.Errorf("template file %q is a directory", resource.TemplateFile) + } + } + return nil +} diff --git a/e2e/stress/internal/render/render.go b/e2e/stress/internal/render/render.go new file mode 100644 index 00000000..28a11022 --- /dev/null +++ b/e2e/stress/internal/render/render.go @@ -0,0 +1,190 @@ +package render + +import ( + "bytes" + "encoding/json" + "fmt" + "os" + "path/filepath" + "regexp" + "strconv" + "strings" + + "k8s.io/apimachinery/pkg/apis/meta/v1/unstructured" + "k8s.io/apimachinery/pkg/runtime" + utilyaml "k8s.io/apimachinery/pkg/util/yaml" + + "github.com/Azure/eno/e2e/stress/internal/plan" +) + +const ( + RunIDLabel = "stress.eno.azure.io/run-id" + PlanLabel = "stress.eno.azure.io/plan" + ResourceIDLabel = "stress.eno.azure.io/resource-id" + VariationLabel = "stress.eno.azure.io/variation" +) + +var variablePattern = regexp.MustCompile(`\$\{([A-Za-z][A-Za-z0-9]*)\}`) + +type Context struct { + RunID string + PlanName string + Namespace string + NamespaceIndex int + ResourceIndex int + Variation string + Phase string + Iteration int + Variables map[string]string + Labels map[string]string +} + +func Resource(baseDir string, spec plan.ResourceSpec, context Context) (*unstructured.Unstructured, error) { + object, err := loadObject(baseDir, spec) + if err != nil { + return nil, err + } + + variables := map[string]string{ + "runID": context.RunID, + "namespace": context.Namespace, + "namespaceIndex": strconv.Itoa(context.NamespaceIndex), + "resourceIndex": strconv.Itoa(context.ResourceIndex), + "variation": context.Variation, + "phase": context.Phase, + "iteration": strconv.Itoa(context.Iteration), + } + for key, value := range context.Variables { + variables[key] = value + } + expanded, err := expand(object.Object, variables) + if err != nil { + return nil, fmt.Errorf("expanding resource %q: %w", spec.Name, err) + } + object.Object = expanded.(map[string]any) + + if object.GetAPIVersion() == "" { + object.SetAPIVersion(spec.APIVersion) + } + if object.GetKind() == "" { + object.SetKind(spec.Kind) + } + if object.GetName() == "" { + object.SetName(spec.Name) + } + if spec.Scope != "cluster" && context.Namespace != "" { + object.SetNamespace(context.Namespace) + } + if object.GetAPIVersion() == "" || object.GetKind() == "" || object.GetName() == "" { + return nil, fmt.Errorf("resource %q must render apiVersion, kind, and metadata.name", spec.Name) + } + + labels := object.GetLabels() + if labels == nil { + labels = map[string]string{} + } + for key, value := range context.Labels { + labels[key] = value + } + labels[RunIDLabel] = context.RunID + labels[PlanLabel] = context.PlanName + labels[ResourceIDLabel] = spec.Name + object.SetLabels(labels) + return object, nil +} + +func String(value string, context Context) (string, error) { + variables := map[string]string{ + "runID": context.RunID, + "namespace": context.Namespace, + "namespaceIndex": strconv.Itoa(context.NamespaceIndex), + "resourceIndex": strconv.Itoa(context.ResourceIndex), + "variation": context.Variation, + "phase": context.Phase, + "iteration": strconv.Itoa(context.Iteration), + } + for key, variable := range context.Variables { + variables[key] = variable + } + expanded, err := expand(value, variables) + if err != nil { + return "", err + } + return expanded.(string), nil +} + +func loadObject(baseDir string, spec plan.ResourceSpec) (*unstructured.Unstructured, error) { + if spec.TemplateFile == "" { + raw, err := json.Marshal(spec.Template) + if err != nil { + return nil, fmt.Errorf("encoding inline template: %w", err) + } + object := &unstructured.Unstructured{} + if err := json.Unmarshal(raw, &object.Object); err != nil { + return nil, fmt.Errorf("decoding inline template: %w", err) + } + return object, nil + } + + path := spec.TemplateFile + if !filepath.IsAbs(path) { + path = filepath.Join(baseDir, path) + } + raw, err := os.ReadFile(path) + if err != nil { + return nil, fmt.Errorf("reading template %q: %w", spec.TemplateFile, err) + } + + decoder := utilyaml.NewYAMLOrJSONDecoder(bytes.NewReader(raw), 4096) + object := &unstructured.Unstructured{} + if err := decoder.Decode(object); err != nil { + return nil, fmt.Errorf("decoding template: %w", err) + } + var extra runtime.RawExtension + if err := decoder.Decode(&extra); err == nil && len(bytes.TrimSpace(extra.Raw)) > 0 { + return nil, fmt.Errorf("template must contain exactly one Kubernetes object") + } + return object, nil +} + +func expand(value any, variables map[string]string) (any, error) { + switch typed := value.(type) { + case map[string]any: + result := make(map[string]any, len(typed)) + for key, item := range typed { + expanded, err := expand(item, variables) + if err != nil { + return nil, err + } + result[key] = expanded + } + return result, nil + case []any: + result := make([]any, len(typed)) + for index, item := range typed { + expanded, err := expand(item, variables) + if err != nil { + return nil, err + } + result[index] = expanded + } + return result, nil + case string: + var missing string + result := variablePattern.ReplaceAllStringFunc(typed, func(match string) string { + key := strings.TrimSuffix(strings.TrimPrefix(match, "${"), "}") + value, found := variables[key] + if !found { + missing = key + return match + } + return value + }) + if missing != "" { + return nil, fmt.Errorf("unknown variable %q", missing) + } + return result, nil + default: + return value, nil + } +} diff --git a/e2e/stress/internal/report/report.go b/e2e/stress/internal/report/report.go new file mode 100644 index 00000000..aa0ad25d --- /dev/null +++ b/e2e/stress/internal/report/report.go @@ -0,0 +1,198 @@ +// Machine-readable and human-readable stress latency reports. +package report + +import ( + "encoding/json" + "fmt" + "math" + "os" + "path/filepath" + "sort" + "time" + + stressstate "github.com/Azure/eno/e2e/stress/internal/state" +) + +type Report struct { + SchemaVersion string `json:"schemaVersion"` + RunID string `json:"runID"` + PlanName string `json:"planName"` + PlanDigest string `json:"planDigest"` + CompletionStage string `json:"completionStage"` + StartedAt *time.Time `json:"startedAt,omitempty"` + CompletedAt *time.Time `json:"completedAt,omitempty"` + Compositions map[string]Composition `json:"compositions"` + Aggregate map[string]Statistics `json:"aggregate"` + Failures []string `json:"failures,omitempty"` +} + +type Composition struct { + Namespace string `json:"namespace"` + CompositionName string `json:"compositionName"` + Variation string `json:"variation"` + SynthesisUUID string `json:"synthesisUUID,omitempty"` + LastStatus string `json:"lastStatus,omitempty"` + OutputValid bool `json:"outputValid"` + Failure string `json:"failure,omitempty"` + LatencyMS map[string]float64 `json:"latencyMs,omitempty"` +} + +type Statistics struct { + Count int `json:"count"` + P50MS float64 `json:"p50Ms"` + P95MS float64 `json:"p95Ms"` + P99MS float64 `json:"p99Ms"` + MaxMS float64 `json:"maxMs"` +} + +func Build(data *stressstate.State) *Report { + result := &Report{ + SchemaVersion: "v1alpha2", + RunID: data.RunID, + PlanName: data.PlanName, + PlanDigest: data.PlanDigest, + CompletionStage: data.CompletionStage, + StartedAt: data.WorkloadStart, + CompletedAt: data.CompletedAt, + Compositions: map[string]Composition{}, + Aggregate: map[string]Statistics{}, + } + values := map[string][]float64{} + for key, state := range data.CompositionState { + entry := Composition{LatencyMS: map[string]float64{}} + if state == nil { + entry.Failure = "no composition result" + result.Failures = append(result.Failures, key+": "+entry.Failure) + result.Compositions[key] = entry + continue + } + entry.Namespace = state.Namespace + entry.CompositionName = state.CompositionName + entry.Variation = state.Variation + entry.SynthesisUUID = state.SynthesisUUID + entry.LastStatus = state.LastStatus + entry.OutputValid = state.OutputValid + entry.Failure = state.Failure + inputStart, inputAck := inputBounds(data.NamespaceInputs[state.Namespace]) + addLatency(entry.LatencyMS, values, "inputCreateToAPIAck", inputStart, inputAck) + addLatency(entry.LatencyMS, values, "apiAckToInputRevision", inputAck, state.InputRevisionObservedAt) + addLatency(entry.LatencyMS, values, "apiAckToExitMissingInputs", inputAck, state.ExitedMissingInputsAt) + if data.CompletionStage == "PostSynthesis" || data.CompletionStage == "Reconcile" { + addLatency(entry.LatencyMS, values, "exitMissingInputsToSynthesisStart", state.ExitedMissingInputsAt, state.SynthesisInitializedObservedAt) + addLatency(entry.LatencyMS, values, "synthesisStartToSynthesisComplete", state.SynthesisInitializedObservedAt, state.SynthesisCompletedObservedAt) + addLatency(entry.LatencyMS, values, "inputCreateToSynthesisComplete", inputStart, state.SynthesisCompletedObservedAt) + } + if data.CompletionStage == "Reconcile" { + addLatency(entry.LatencyMS, values, "synthesisCompleteToReconciled", state.SynthesisCompletedObservedAt, state.ReconciledObservedAt) + addLatency(entry.LatencyMS, values, "inputCreateToReconciled", inputStart, state.ReconciledObservedAt) + } + if entry.Failure != "" { + result.Failures = append(result.Failures, key+": "+entry.Failure) + } + result.Compositions[key] = entry + } + for metric, samples := range values { + result.Aggregate[metric] = statistics(samples) + } + return result +} + +func Write(path string, result *Report) error { + raw, err := json.MarshalIndent(result, "", " ") + if err != nil { + return err + } + if err := os.MkdirAll(filepath.Dir(path), 0o755); err != nil { + return err + } + return os.WriteFile(path, raw, 0o600) +} + +func Print(result *Report, output func(string, ...any)) { + output("run %s: stage=%s, %d compositions, %d failures\n", result.RunID, result.CompletionStage, len(result.Compositions), len(result.Failures)) + metrics := make([]string, 0, len(result.Aggregate)) + for metric := range result.Aggregate { + metrics = append(metrics, metric) + } + sort.Strings(metrics) + output("%-38s %8s %8s %8s %8s\n", "latency", "p50", "p95", "p99", "max") + for _, metric := range metrics { + stats := result.Aggregate[metric] + output("%-38s %7.1fms %7.1fms %7.1fms %7.1fms\n", metric, stats.P50MS, stats.P95MS, stats.P99MS, stats.MaxMS) + } + for _, failure := range result.Failures { + output("failure: %s\n", failure) + } +} + +func inputBounds(inputs map[string]*stressstate.InputTiming) (*time.Time, *time.Time) { + var earliest, latest time.Time + for _, input := range inputs { + if earliest.IsZero() || input.RequestStartedAt.Before(earliest) { + earliest = input.RequestStartedAt + } + if latest.IsZero() || input.APIAcknowledgedAt.After(latest) { + latest = input.APIAcknowledgedAt + } + } + if earliest.IsZero() || latest.IsZero() { + return nil, nil + } + return &earliest, &latest +} + +func addLatency(target map[string]float64, aggregate map[string][]float64, name string, start, end *time.Time) { + if start == nil || end == nil { + return + } + value := float64(end.Sub(*start).Microseconds()) / 1000 + target[name] = value + aggregate[name] = append(aggregate[name], value) +} + +func statistics(values []float64) Statistics { + sort.Float64s(values) + return Statistics{ + Count: len(values), + P50MS: percentile(values, 0.50), + P95MS: percentile(values, 0.95), + P99MS: percentile(values, 0.99), + MaxMS: values[len(values)-1], + } +} + +func percentile(values []float64, quantile float64) float64 { + index := int(math.Ceil(quantile*float64(len(values)))) - 1 + if index < 0 { + return 0 + } + return values[index] +} + +func ResolvePath(baseDir, configured, runID string) string { + if configured == "" { + configured = filepath.Join("results", runID+".json") + } + configured = filepath.Clean(configured) + configured = stringReplace(configured, "${runID}", runID) + if !filepath.IsAbs(configured) { + configured = filepath.Join(baseDir, configured) + } + return configured +} + +func stringReplace(value, old, replacement string) string { + for { + index := -1 + for position := 0; position+len(old) <= len(value); position++ { + if value[position:position+len(old)] == old { + index = position + break + } + } + if index < 0 { + return value + } + value = fmt.Sprintf("%s%s%s", value[:index], replacement, value[index+len(old):]) + } +} diff --git a/e2e/stress/internal/runner/common.go b/e2e/stress/internal/runner/common.go new file mode 100644 index 00000000..e78be860 --- /dev/null +++ b/e2e/stress/internal/runner/common.go @@ -0,0 +1,380 @@ +// Shared orchestration helpers for live stress commands. +package runner + +import ( + "context" + "crypto/rand" + "encoding/hex" + "fmt" + "path/filepath" + "sort" + "strings" + "time" + + "github.com/google/uuid" + "golang.org/x/sync/errgroup" + metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" + "k8s.io/apimachinery/pkg/apis/meta/v1/unstructured" + "k8s.io/apimachinery/pkg/runtime/schema" + + "github.com/Azure/eno/e2e/stress/internal/kube" + "github.com/Azure/eno/e2e/stress/internal/plan" + "github.com/Azure/eno/e2e/stress/internal/render" + stressstate "github.com/Azure/eno/e2e/stress/internal/state" +) + +type Options struct { + PlanPath string + StatePath string + Kubeconfig string + RunID string + ServerDryRun bool + Output func(string, ...any) +} + +type runtime struct { + plan *plan.Plan + planPath string + baseDir string + client *kube.Client + store *stressstate.Store + output func(string, ...any) +} + +func loadRuntime(options Options, withState bool) (*runtime, error) { + planPath, err := filepath.Abs(options.PlanPath) + if err != nil { + return nil, err + } + loadedPlan, err := plan.Load(planPath) + if err != nil { + return nil, err + } + client, err := kube.New(options.Kubeconfig, loadedPlan.Spec.Run.ClientQPS, loadedPlan.Spec.Run.ClientBurst) + if err != nil { + return nil, err + } + r := &runtime{ + plan: loadedPlan, + planPath: planPath, + baseDir: filepath.Dir(planPath), + client: client, + output: options.Output, + } + if r.output == nil { + r.output = func(string, ...any) {} + } + if withState { + if options.StatePath == "" { + return nil, fmt.Errorf("--state is required") + } + r.store, err = prepareStore(options, loadedPlan, planPath) + if err != nil { + return nil, err + } + } + return r, nil +} + +func loadRunRuntime(options Options) (*runtime, error) { + if options.StatePath == "" { + return nil, fmt.Errorf("--state is required") + } + store, err := stressstate.Load(options.StatePath) + if err != nil { + return nil, err + } + var planPath, expectedDigest string + store.Read(func(data *stressstate.State) error { + planPath = data.PlanPath + expectedDigest = data.PlanDigest + return nil + }) + loadedPlan, err := plan.Load(planPath) + if err != nil { + return nil, err + } + digest, err := stressstate.FileDigest(planPath) + if err != nil { + return nil, err + } + if digest != expectedDigest { + return nil, fmt.Errorf("plan changed after prepare: expected %s, got %s", expectedDigest, digest) + } + client, err := kube.New(options.Kubeconfig, loadedPlan.Spec.Run.ClientQPS, loadedPlan.Spec.Run.ClientBurst) + if err != nil { + return nil, err + } + output := options.Output + if output == nil { + output = func(string, ...any) {} + } + return &runtime{ + plan: loadedPlan, + planPath: planPath, + baseDir: filepath.Dir(planPath), + client: client, + store: store, + output: output, + }, nil +} + +func prepareStore(options Options, loadedPlan *plan.Plan, planPath string) (*stressstate.Store, error) { + digest, err := stressstate.FileDigest(planPath) + if err != nil { + return nil, err + } + if existing, err := stressstate.Load(options.StatePath); err == nil { + var mismatch error + existing.Read(func(data *stressstate.State) error { + if data.PlanDigest != digest { + mismatch = fmt.Errorf("state plan digest does not match %s", planPath) + } + return nil + }) + return existing, mismatch + } + runID := options.RunID + if runID == "" { + runID = newRunID(loadedPlan.Metadata.Name) + } + now := time.Now() + data := &stressstate.State{ + SchemaVersion: stressstate.SchemaVersion, + RunID: runID, + PlanName: loadedPlan.Metadata.Name, + PlanPath: planPath, + PlanDigest: digest, + CompletionStage: string(loadedPlan.Spec.Run.CompletionStage), + Phase: "preparing", + CreatedAt: now, + NamespaceInputs: map[string]map[string]*stressstate.InputTiming{}, + CompositionState: map[string]*stressstate.CompositionResult{}, + } + store := stressstate.NewStore(options.StatePath, data) + return store, store.Save() +} + +func newRunID(planName string) string { + random := make([]byte, 4) + if _, err := rand.Read(random); err != nil { + panic(err) + } + suffix := hex.EncodeToString(random) + maxPrefix := 63 - len(suffix) - 1 + if len(planName) > maxPrefix { + planName = strings.TrimRight(planName[:maxPrefix], "-") + } + return planName + "-" + suffix +} + +func namespaceNames(p *plan.Plan, runID string) []string { + names := make([]string, p.Spec.Run.NamespaceCount) + for index := range names { + prefix := p.Spec.Run.NamespacePrefix + guid := uuid.NewSHA1(uuid.NameSpaceOID, []byte(fmt.Sprintf("%s/%d", runID, index+1))) + guidSuffix := strings.ReplaceAll(guid.String(), "-", "")[:16] + if len(prefix)+len(guidSuffix) > 63 { + prefix = strings.TrimRight(prefix[:63-len(guidSuffix)], "-") + } + names[index] = prefix + guidSuffix + } + return names +} + +func renderContexts(p *plan.Plan, runID string, namespaces []string, phase string, iteration int, spec plan.ResourceSpec) []render.Context { + base := render.Context{ + RunID: runID, + PlanName: p.Metadata.Name, + Phase: phase, + Iteration: iteration, + Variables: p.Spec.Run.Variables, + Labels: p.Spec.Run.Labels, + } + count := spec.ExpandedCount() + if spec.ForEachNamespace { + contexts := make([]render.Context, 0, len(namespaces)*count) + for namespaceIndex, namespace := range namespaces { + for resourceIndex := 1; resourceIndex <= count; resourceIndex++ { + context := base + context.Namespace = namespace + context.NamespaceIndex = namespaceIndex + 1 + context.ResourceIndex = resourceIndex + contexts = append(contexts, context) + } + } + return contexts + } + contexts := make([]render.Context, count) + for resourceIndex := 1; resourceIndex <= count; resourceIndex++ { + context := base + context.ResourceIndex = resourceIndex + contexts[resourceIndex-1] = context + } + return contexts +} + +func orderedSetup(resources []plan.ResourceSpec) ([]plan.ResourceSpec, error) { + pending := make(map[string]plan.ResourceSpec, len(resources)) + for _, resource := range resources { + pending[resource.Name] = resource + } + done := map[string]bool{} + ordered := make([]plan.ResourceSpec, 0, len(resources)) + for len(pending) > 0 { + progress := false + names := make([]string, 0, len(pending)) + for name := range pending { + names = append(names, name) + } + sort.Strings(names) + for _, name := range names { + resource := pending[name] + ready := true + for _, dependency := range resource.DependsOn { + ready = ready && done[dependency] + } + if !ready { + continue + } + ordered = append(ordered, resource) + done[name] = true + delete(pending, name) + progress = true + } + if !progress { + return nil, fmt.Errorf("setup resource dependency cycle") + } + } + return ordered, nil +} + +func parallel(ctx context.Context, concurrency int, count int, fn func(context.Context, int) error) error { + group, groupContext := errgroup.WithContext(ctx) + group.SetLimit(concurrency) + for index := 0; index < count; index++ { + index := index + group.Go(func() error { return fn(groupContext, index) }) + } + return group.Wait() +} + +func recordResource(store *stressstate.Store, logicalName, phase string, object *unstructured.Unstructured, resource *kube.Resource, startedAt, acknowledgedAt time.Time) error { + return store.Update(func(data *stressstate.State) error { + upsertResource(data, resourceRecord(logicalName, phase, object, resource, startedAt, acknowledgedAt)) + return nil + }) +} + +func recordResourceMemory(store *stressstate.Store, logicalName, phase string, object *unstructured.Unstructured, resource *kube.Resource, startedAt, acknowledgedAt time.Time) error { + return store.Mutate(func(data *stressstate.State) error { + upsertResource(data, resourceRecord(logicalName, phase, object, resource, startedAt, acknowledgedAt)) + return nil + }) +} + +func resourceRecord(logicalName, phase string, object *unstructured.Unstructured, resource *kube.Resource, startedAt, acknowledgedAt time.Time) stressstate.Resource { + record := stressstate.Resource{ + LogicalName: logicalName, + Phase: phase, + APIVersion: object.GetAPIVersion(), + Kind: object.GetKind(), + Group: resource.GVR.Group, + Version: resource.GVR.Version, + Resource: resource.GVR.Resource, + Scope: resource.Scope, + Namespace: object.GetNamespace(), + Name: object.GetName(), + UID: string(object.GetUID()), + ResourceVersion: object.GetResourceVersion(), + CreationTimestamp: object.GetCreationTimestamp().Time, + RequestStartedAt: startedAt, + APIAcknowledgedAt: acknowledgedAt, + } + return record +} + +func upsertResource(data *stressstate.State, record stressstate.Resource) { + for index := range data.Resources { + existing := &data.Resources[index] + if existing.Phase == record.Phase && existing.LogicalName == record.LogicalName && existing.Namespace == record.Namespace && existing.Name == record.Name { + *existing = record + return + } + } + data.Resources = append(data.Resources, record) +} + +func gvrFor(record stressstate.Resource) schema.GroupVersionResource { + return schema.GroupVersionResource{Group: record.Group, Version: record.Version, Resource: record.Resource} +} + +func compositionKey(namespace, name string) string { + return namespace + "/" + name +} + +func expandedResourceCount(p *plan.Plan, logicalName string) int { + for _, resource := range p.Spec.Setup.Resources { + if resource.Name != logicalName { + continue + } + count := resource.ExpandedCount() + if resource.ForEachNamespace { + count *= p.Spec.Run.NamespaceCount + } + return count + } + return 0 +} + +func simplifiedStatus(object *unstructured.Unstructured) string { + value, _, _ := unstructured.NestedString(object.Object, "status", "simplified", "status") + return value +} + +func synthesisExists(object *unstructured.Unstructured) bool { + inFlight, foundInFlight, _ := unstructured.NestedMap(object.Object, "status", "inFlightSynthesis") + current, foundCurrent, _ := unstructured.NestedMap(object.Object, "status", "currentSynthesis") + return (foundInFlight && len(inFlight) > 0) || (foundCurrent && len(current) > 0) +} + +func inputRevisionKeys(object *unstructured.Unstructured) map[string]string { + items, _, _ := unstructured.NestedSlice(object.Object, "status", "inputRevisions") + result := map[string]string{} + for _, item := range items { + entry, ok := item.(map[string]any) + if !ok { + continue + } + key, _, _ := unstructured.NestedString(entry, "key") + resourceVersion, _, _ := unstructured.NestedString(entry, "resourceVersion") + result[key] = resourceVersion + } + return result +} + +func namespaceObject(name, runID, planName string, labels map[string]string) *unstructured.Unstructured { + object := &unstructured.Unstructured{Object: map[string]any{ + "apiVersion": "v1", + "kind": "Namespace", + "metadata": map[string]any{ + "name": name, + }, + }} + allLabels := map[string]string{} + for key, value := range labels { + allLabels[key] = value + } + allLabels[render.RunIDLabel] = runID + allLabels[render.PlanLabel] = planName + allLabels[render.ResourceIDLabel] = "namespace" + object.SetLabels(allLabels) + return object +} + +func creationTime(object *unstructured.Unstructured) time.Time { + if object.GetCreationTimestamp() == (metav1.Time{}) { + return time.Time{} + } + return object.GetCreationTimestamp().Time +} diff --git a/e2e/stress/internal/runner/monitor.go b/e2e/stress/internal/runner/monitor.go new file mode 100644 index 00000000..3764d77f --- /dev/null +++ b/e2e/stress/internal/runner/monitor.go @@ -0,0 +1,295 @@ +// Composition and output event correlation for workload runs. +package runner + +import ( + "context" + "fmt" + "sync" + "time" + + metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" + "k8s.io/apimachinery/pkg/apis/meta/v1/unstructured" + "k8s.io/apimachinery/pkg/runtime/schema" + + "github.com/Azure/eno/e2e/stress/internal/plan" + "github.com/Azure/eno/e2e/stress/internal/render" + stressstate "github.com/Azure/eno/e2e/stress/internal/state" +) + +type runMonitor struct { + runtime *runtime + runID string + stage plan.CompletionStage + compositionGVR schema.GroupVersionResource + expectedInputs []string + + mutex sync.Mutex + latest map[string]*unstructured.Unstructured + completed map[string]bool + total int + armed bool + done chan struct{} + errors chan error + complete sync.Once + watchCancel context.CancelFunc +} + +func newRunMonitor(r *runtime, compositionGVR schema.GroupVersionResource, expectedInputs []string) (*runMonitor, error) { + monitor := &runMonitor{ + runtime: r, + stage: r.plan.Spec.Run.CompletionStage, + compositionGVR: compositionGVR, + expectedInputs: expectedInputs, + latest: map[string]*unstructured.Unstructured{}, + completed: map[string]bool{}, + done: make(chan struct{}), + errors: make(chan error, 1), + } + r.store.Read(func(data *stressstate.State) error { + monitor.runID = data.RunID + monitor.total = len(data.CompositionState) + for key, result := range data.CompositionState { + if monitor.resultComplete(result) { + monitor.completed[key] = true + } + } + return nil + }) + return monitor, nil +} + +func (m *runMonitor) Start(ctx context.Context) error { + watchContext, cancel := context.WithCancel(ctx) + m.watchCancel = cancel + compositionReady := make(chan struct{}) + selector := render.RunIDLabel + "=" + m.runID + go func() { + err := m.runtime.client.Watch(watchContext, m.compositionGVR, metav1.NamespaceAll, selector, compositionReady, m.handleComposition) + if watchContext.Err() == nil { + m.errors <- err + } + }() + go m.checkpoint(watchContext) + select { + case <-ctx.Done(): + return ctx.Err() + case err := <-m.errors: + return err + case <-compositionReady: + } + return nil +} + +func (m *runMonitor) checkpoint(ctx context.Context) { + ticker := time.NewTicker(10 * time.Second) + defer ticker.Stop() + for { + select { + case <-ctx.Done(): + return + case <-ticker.C: + if err := m.runtime.store.Save(); err != nil { + select { + case <-ctx.Done(): + return + case m.errors <- fmt.Errorf("checkpointing run state: %w", err): + return + } + } + } + } +} + +func (m *runMonitor) Stop() { + if m.watchCancel != nil { + m.watchCancel() + } +} + +func (m *runMonitor) Arm() { + m.mutex.Lock() + defer m.mutex.Unlock() + m.armed = true + if len(m.completed) == m.total { + m.complete.Do(func() { close(m.done) }) + } +} + +func (m *runMonitor) Wait(ctx context.Context) error { + select { + case <-ctx.Done(): + m.markIncomplete(ctx.Err().Error()) + return ctx.Err() + case err := <-m.errors: + m.markIncomplete(err.Error()) + return err + case <-m.done: + return nil + } +} + +func (m *runMonitor) InputAcknowledged(namespace, logicalName string, object *unstructured.Unstructured, startedAt, acknowledgedAt time.Time, overwrite bool) error { + if err := m.runtime.store.Mutate(func(data *stressstate.State) error { + inputs := data.NamespaceInputs[namespace] + if inputs == nil { + return fmt.Errorf("no input state for namespace %s", namespace) + } + if _, found := inputs[logicalName]; found && !overwrite { + return nil + } + inputs[logicalName] = &stressstate.InputTiming{ + Name: object.GetName(), + Kind: object.GetKind(), + RequestStartedAt: startedAt, + APIAcknowledgedAt: acknowledgedAt, + ResourceVersion: object.GetResourceVersion(), + UID: string(object.GetUID()), + } + return nil + }); err != nil { + return err + } + m.mutex.Lock() + var latest []*unstructured.Unstructured + for _, object := range m.latest { + if object.GetNamespace() == namespace { + latest = append(latest, object.DeepCopy()) + } + } + m.mutex.Unlock() + for _, object := range latest { + m.handleComposition(object, time.Now()) + } + return nil +} + +func (m *runMonitor) handleComposition(object *unstructured.Unstructured, observedAt time.Time) { + namespace := object.GetNamespace() + key := compositionKey(namespace, object.GetName()) + m.mutex.Lock() + m.latest[key] = object.DeepCopy() + m.mutex.Unlock() + nowComplete := false + _ = m.runtime.store.Mutate(func(data *stressstate.State) error { + result := data.CompositionState[key] + if result == nil { + return nil + } + inputs := data.NamespaceInputs[namespace] + result.LastStatus = simplifiedStatus(object) + revisions := inputRevisionKeys(object) + allRevisions := len(m.expectedInputs) > 0 + for _, key := range m.expectedInputs { + input := inputs[key] + if input == nil || revisions[key] != input.ResourceVersion { + allRevisions = false + break + } + } + if allRevisions && result.InputRevisionObservedAt == nil { + result.InputRevisionObservedAt = timePointer(observedAt) + } + if result.InputRevisionObservedAt != nil && result.ExitedMissingInputsAt == nil && result.LastStatus != "MissingInputs" { + result.ExitedMissingInputsAt = timePointer(observedAt) + } + if result.InputRevisionObservedAt != nil { + if synthesis, found := synthesisMap(object); found { + if uuid, _, _ := unstructured.NestedString(synthesis, "uuid"); uuid != "" { + result.SynthesisUUID = uuid + } + if initialized, _, _ := unstructured.NestedString(synthesis, "initialized"); initialized != "" && result.SynthesisInitializedObservedAt == nil { + result.SynthesisInitializedAt = parseTime(initialized) + result.SynthesisInitializedObservedAt = timePointer(observedAt) + } + if synthesized, _, _ := unstructured.NestedString(synthesis, "synthesized"); synthesized != "" && result.SynthesisCompletedObservedAt == nil { + result.SynthesisCompletedAt = parseTime(synthesized) + result.SynthesisCompletedObservedAt = timePointer(observedAt) + } + } + } + current, found, _ := unstructured.NestedMap(object.Object, "status", "currentSynthesis") + if found { + if synthesized, _, _ := unstructured.NestedString(current, "synthesized"); synthesized != "" && result.SynthesisCompletedObservedAt == nil { + result.SynthesisCompletedAt = parseTime(synthesized) + result.SynthesisCompletedObservedAt = timePointer(observedAt) + } + if reconciled, _, _ := unstructured.NestedString(current, "reconciled"); reconciled != "" && result.ReconciledObservedAt == nil { + result.ReconciledAt = parseTime(reconciled) + result.ReconciledObservedAt = timePointer(observedAt) + } + if ready, _, _ := unstructured.NestedString(current, "ready"); ready != "" && result.ReadyObservedAt == nil { + result.ReadyAt = parseTime(ready) + result.ReadyObservedAt = timePointer(observedAt) + } + } + nowComplete = m.resultComplete(result) + return nil + }) + if nowComplete { + m.recordCompletion(key) + } +} + +func (m *runMonitor) recordCompletion(key string) { + m.mutex.Lock() + defer m.mutex.Unlock() + m.completed[key] = true + if m.armed && len(m.completed) == m.total { + m.complete.Do(func() { close(m.done) }) + } +} + +func (m *runMonitor) resultComplete(result *stressstate.CompositionResult) bool { + if result == nil || result.InputRevisionObservedAt == nil || result.ExitedMissingInputsAt == nil { + return false + } + switch m.stage { + case plan.CompletionStagePreSynthesis: + return true + case plan.CompletionStagePostSynthesis: + return result.SynthesisCompletedObservedAt != nil + case plan.CompletionStageReconcile: + return result.ReconciledObservedAt != nil + default: + return false + } +} + +func (m *runMonitor) markIncomplete(reason string) { + _ = m.runtime.store.Mutate(func(data *stressstate.State) error { + for _, result := range data.CompositionState { + if result == nil { + continue + } + if m.resultComplete(result) { + continue + } + if result.Failure == "" { + result.Failure = "incomplete: " + reason + } + } + return nil + }) + _ = m.runtime.store.Save() +} + +func synthesisMap(object *unstructured.Unstructured) (map[string]any, bool) { + if inFlight, found, _ := unstructured.NestedMap(object.Object, "status", "inFlightSynthesis"); found && len(inFlight) > 0 { + return inFlight, true + } + current, found, _ := unstructured.NestedMap(object.Object, "status", "currentSynthesis") + return current, found && len(current) > 0 +} + +func parseTime(value string) *time.Time { + parsed, err := time.Parse(time.RFC3339Nano, value) + if err != nil { + return nil + } + return &parsed +} + +func timePointer(value time.Time) *time.Time { + copy := value + return © +} diff --git a/e2e/stress/internal/runner/prepare.go b/e2e/stress/internal/runner/prepare.go new file mode 100644 index 00000000..162b0cf6 --- /dev/null +++ b/e2e/stress/internal/runner/prepare.go @@ -0,0 +1,298 @@ +// Setup-resource creation and MissingInputs gating. +package runner + +import ( + "context" + "fmt" + "sync" + "time" + + metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" + "k8s.io/apimachinery/pkg/apis/meta/v1/unstructured" + "k8s.io/apimachinery/pkg/runtime/schema" + + "github.com/Azure/eno/e2e/stress/internal/plan" + "github.com/Azure/eno/e2e/stress/internal/render" + stressstate "github.com/Azure/eno/e2e/stress/internal/state" +) + +func Prepare(ctx context.Context, options Options) error { + r, err := loadRuntime(options, true) + if err != nil { + return err + } + timeout, _ := r.plan.Timeout() + ctx, cancel := context.WithTimeout(ctx, timeout) + defer cancel() + + var runID string + r.store.Read(func(data *stressstate.State) error { + runID = data.RunID + return nil + }) + namespaces := namespaceNames(r.plan, runID) + if err := r.store.Update(func(data *stressstate.State) error { + data.Phase = "preparing" + data.Namespaces = namespaces + return nil + }); err != nil { + return err + } + r.output("run ID: %s\n", runID) + r.output("creating %d namespaces\n", len(namespaces)) + if err := parallel(ctx, r.plan.Spec.Run.Concurrency, len(namespaces), func(ctx context.Context, index int) error { + object := namespaceObject(namespaces[index], runID, r.plan.Metadata.Name, r.plan.Spec.Run.Labels) + startedAt := time.Now() + created, resource, err := r.client.CreateOwned(ctx, object, runID, false) + acknowledgedAt := time.Now() + if err != nil { + return fmt.Errorf("creating namespace %s: %w", namespaces[index], err) + } + return recordResource(r.store, "namespace", "setup", created, resource, startedAt, acknowledgedAt) + }); err != nil { + return err + } + + ordered, err := orderedSetup(r.plan.Spec.Setup.Resources) + if err != nil { + return err + } + for _, resourceSpec := range ordered { + if resourceSpec.Operation == "observe" { + r.output("observing generated setup resource %s\n", resourceSpec.Name) + if err := r.observeSetupResource(ctx, resourceSpec, namespaces, runID); err != nil { + return err + } + continue + } + contexts := renderContexts(r.plan, runID, namespaces, "setup", 1, resourceSpec) + r.output("creating setup resource %s (%d objects)\n", resourceSpec.Name, len(contexts)) + if err := parallel(ctx, r.plan.Spec.Run.Concurrency, len(contexts), func(ctx context.Context, index int) error { + object, err := render.Resource(r.baseDir, resourceSpec, contexts[index]) + if err != nil { + return err + } + startedAt := time.Now() + created, resource, err := r.client.CreateSetup(ctx, object, runID, false, resourceSpec.Reuse) + acknowledgedAt := time.Now() + if err != nil { + return fmt.Errorf("creating %s %s/%s: %w", object.GetKind(), object.GetNamespace(), object.GetName(), err) + } + return recordResource(r.store, resourceSpec.Name, "setup", created, resource, startedAt, acknowledgedAt) + }); err != nil { + return err + } + } + + for _, readiness := range r.plan.Spec.Setup.Readiness { + r.output("waiting for %s resources to enter %s\n", readiness.Resource, readiness.Condition.Status) + if err := r.waitForReadiness(ctx, readiness.Resource, readiness.Condition.Status, readiness.Condition.ExpectedMissingInputs); err != nil { + return err + } + } + now := time.Now() + if err := r.store.Update(func(data *stressstate.State) error { + data.Phase = "prepared" + data.PreparedAt = &now + return nil + }); err != nil { + return err + } + r.output("prepared %d compositions in MissingInputs; state: %s\n", expandedResourceCount(r.plan, "composition"), options.StatePath) + return nil +} + +func (r *runtime) observeSetupResource(ctx context.Context, resourceSpec plan.ResourceSpec, namespaces []string, runID string) error { + probe := &unstructured.Unstructured{} + probe.SetAPIVersion(resourceSpec.APIVersion) + probe.SetKind(resourceSpec.Kind) + if resourceSpec.Scope == "namespace" && len(namespaces) > 0 { + probe.SetNamespace(namespaces[0]) + } + resource, err := r.client.ResourceFor(probe) + if err != nil { + return err + } + + expectedNamespaces := map[string]bool{} + expectedPerNamespace := resourceSpec.ExpandedCount() + expectedCount := expectedPerNamespace + if resourceSpec.ForEachNamespace { + expectedCount *= len(namespaces) + for _, namespace := range namespaces { + expectedNamespaces[namespace] = true + } + } + + watchContext, cancel := context.WithCancel(ctx) + defer cancel() + ready := make(chan struct{}) + done := make(chan struct{}) + errChannel := make(chan error, 1) + var once sync.Once + var mutex sync.Mutex + seen := map[string]string{} + seenByNamespace := map[string]int{} + handler := func(object *unstructured.Unstructured, observedAt time.Time) { + if resourceSpec.ForEachNamespace && !expectedNamespaces[object.GetNamespace()] { + return + } + key := compositionKey(object.GetNamespace(), object.GetName()) + mutex.Lock() + defer mutex.Unlock() + if previousUID, found := seen[key]; found { + if previousUID != string(object.GetUID()) { + once.Do(func() { + errChannel <- fmt.Errorf("generated %s %q changed UID", resourceSpec.Kind, key) + close(done) + }) + } + return + } + startedAt := object.GetCreationTimestamp().Time + if startedAt.IsZero() { + startedAt = observedAt + } + if err := recordResourceMemory(r.store, resourceSpec.Name, "setup", object, resource, startedAt, observedAt); err != nil { + once.Do(func() { + errChannel <- err + close(done) + }) + return + } + seen[key] = string(object.GetUID()) + seenByNamespace[object.GetNamespace()]++ + if resourceSpec.ForEachNamespace && seenByNamespace[object.GetNamespace()] > expectedPerNamespace { + once.Do(func() { + errChannel <- fmt.Errorf("observed more than %d generated %s resources in namespace %s", expectedPerNamespace, resourceSpec.Kind, object.GetNamespace()) + close(done) + }) + return + } + if len(seen) == expectedCount { + once.Do(func() { close(done) }) + } + } + selector := render.RunIDLabel + "=" + runID + "," + render.ResourceIDLabel + "=" + resourceSpec.Name + go func() { + err := r.client.Watch(watchContext, resource.GVR, metav1.NamespaceAll, selector, ready, handler) + if watchContext.Err() == nil { + errChannel <- err + } + }() + select { + case <-ctx.Done(): + return ctx.Err() + case err := <-errChannel: + return err + case <-ready: + } + select { + case <-ctx.Done(): + return ctx.Err() + case err := <-errChannel: + return err + case <-done: + select { + case err := <-errChannel: + return err + default: + return r.store.Save() + } + } +} + +func (r *runtime) waitForReadiness(ctx context.Context, logicalName, expectedStatus string, expectedMissing []string) error { + var records []stressstate.Resource + var runID string + r.store.Read(func(data *stressstate.State) error { + runID = data.RunID + for _, resource := range data.Resources { + if resource.Phase == "setup" && resource.LogicalName == logicalName { + records = append(records, resource) + } + } + return nil + }) + if len(records) == 0 { + return fmt.Errorf("readiness resource %q has no recorded objects", logicalName) + } + gvr := gvrFor(records[0]) + for _, record := range records[1:] { + if gvrFor(record) != gvr { + return fmt.Errorf("readiness resource %q expands to multiple GVRs", logicalName) + } + } + + watchContext, cancel := context.WithCancel(ctx) + defer cancel() + ready := make(chan struct{}) + done := make(chan struct{}) + errChannel := make(chan error, 1) + var once sync.Once + var mutex sync.Mutex + seen := map[string]bool{} + handler := func(object *unstructured.Unstructured, _ time.Time) { + if object.GetLabels()[render.ResourceIDLabel] != logicalName { + return + } + key := object.GetNamespace() + "/" + object.GetName() + mutex.Lock() + defer mutex.Unlock() + if synthesisExists(object) { + once.Do(func() { + errChannel <- fmt.Errorf("%s became synthesizable before run", key) + close(done) + }) + return + } + if simplifiedStatus(object) != expectedStatus { + delete(seen, key) + return + } + revisions := inputRevisionKeys(object) + for _, missing := range expectedMissing { + if _, found := revisions[missing]; found { + once.Do(func() { + errChannel <- fmt.Errorf("%s unexpectedly has required input %q", key, missing) + close(done) + }) + return + } + } + seen[key] = true + if len(seen) == len(records) { + once.Do(func() { close(done) }) + } + } + go func() { + err := r.client.Watch(watchContext, gvr, metav1.NamespaceAll, render.RunIDLabel+"="+runID, ready, handler) + if watchContext.Err() == nil { + errChannel <- err + } + }() + select { + case <-ctx.Done(): + return ctx.Err() + case err := <-errChannel: + return err + case <-ready: + } + select { + case <-ctx.Done(): + return ctx.Err() + case err := <-errChannel: + return err + case <-done: + select { + case err := <-errChannel: + return err + default: + return nil + } + } +} + +func listResource(ctx context.Context, r *runtime, gvr schema.GroupVersionResource, namespace string) (*unstructured.UnstructuredList, error) { + return r.client.Dynamic.Resource(gvr).Namespace(namespace).List(ctx, metav1.ListOptions{}) +} diff --git a/e2e/stress/internal/runner/run.go b/e2e/stress/internal/runner/run.go new file mode 100644 index 00000000..aeb8e833 --- /dev/null +++ b/e2e/stress/internal/runner/run.go @@ -0,0 +1,281 @@ +// Workload-only execution for prepared stress runs. +package runner + +import ( + "context" + "fmt" + "path/filepath" + "time" + + metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" + + "github.com/Azure/eno/e2e/stress/internal/render" + "github.com/Azure/eno/e2e/stress/internal/report" + stressstate "github.com/Azure/eno/e2e/stress/internal/state" +) + +func Run(ctx context.Context, options Options) error { + r, err := loadRunRuntime(options) + if err != nil { + return err + } + timeout, _ := r.plan.Timeout() + ctx, cancel := context.WithTimeout(ctx, timeout) + defer cancel() + + r.output("verifying prepared Compositions\n") + compositionRecords, expectedInputs, err := r.verifyPrepared(ctx) + if err != nil { + return err + } + r.output("verified %d prepared Compositions; establishing watches\n", len(compositionRecords)) + monitor, err := newRunMonitor(r, gvrFor(compositionRecords[0]), expectedInputs) + if err != nil { + return err + } + defer monitor.Stop() + if err := monitor.Start(ctx); err != nil { + return fmt.Errorf("establishing workload watches: %w", err) + } + start := time.Now() + if err := r.store.Update(func(data *stressstate.State) error { + data.Phase = "running" + data.WorkloadStart = &start + return nil + }); err != nil { + return err + } + monitor.Arm() + r.output("workload started at %s; watches established\n", start.Format(time.RFC3339Nano)) + if err := r.executePhases(ctx, monitor); err != nil { + monitor.markIncomplete(err.Error()) + return err + } + if err := monitor.Wait(ctx); err != nil { + return fmt.Errorf("waiting for %s completion: %w", r.plan.Spec.Run.CompletionStage, err) + } + completed := time.Now() + if err := r.store.Update(func(data *stressstate.State) error { + data.Phase = "completed" + data.CompletedAt = &completed + return nil + }); err != nil { + return err + } + return r.writeReport() +} + +func (r *runtime) verifyPrepared(ctx context.Context) ([]stressstate.Resource, []string, error) { + var records []stressstate.Resource + var phase, runID string + r.store.Read(func(data *stressstate.State) error { + phase = data.Phase + runID = data.RunID + for _, resource := range data.Resources { + if resource.Phase == "setup" && resource.Kind == "Composition" { + records = append(records, resource) + } + } + return nil + }) + if phase != "prepared" && phase != "running" { + return nil, nil, fmt.Errorf("state phase must be prepared or running, got %q", phase) + } + if len(records) == 0 { + return nil, nil, fmt.Errorf("no prepared Compositions found in state") + } + expectedCount := expandedResourceCount(r.plan, records[0].LogicalName) + if len(records) != expectedCount { + return nil, nil, fmt.Errorf("expected %d prepared Compositions, found %d", expectedCount, len(records)) + } + gvr := gvrFor(records[0]) + recordsByKey := make(map[string]stressstate.Resource, len(records)) + for _, record := range records { + if gvrFor(record) != gvr { + return nil, nil, fmt.Errorf("prepared Compositions span multiple resource types") + } + recordsByKey[compositionKey(record.Namespace, record.Name)] = record + } + + results := map[string]*stressstate.CompositionResult{} + selector := render.RunIDLabel + "=" + runID + "," + render.ResourceIDLabel + "=composition" + resource := r.client.Dynamic.Resource(gvr).Namespace(metav1.NamespaceAll) + continueToken := "" + for { + list, err := resource.List(ctx, metav1.ListOptions{LabelSelector: selector, Limit: 500, Continue: continueToken}) + if err != nil { + return nil, nil, fmt.Errorf("listing prepared Compositions: %w", err) + } + for index := range list.Items { + object := &list.Items[index] + key := compositionKey(object.GetNamespace(), object.GetName()) + record, found := recordsByKey[key] + if !found { + return nil, nil, fmt.Errorf("found unrecorded Composition %s during preflight", key) + } + if string(object.GetUID()) != record.UID { + return nil, nil, fmt.Errorf("Composition %s UID changed after prepare", key) + } + if simplifiedStatus(object) != "MissingInputs" || synthesisExists(object) { + return nil, nil, fmt.Errorf("Composition %s is no longer pristine MissingInputs (status=%s)", key, simplifiedStatus(object)) + } + results[key] = &stressstate.CompositionResult{ + Namespace: object.GetNamespace(), + CompositionName: object.GetName(), + Variation: object.GetLabels()[render.VariationLabel], + } + } + continueToken = list.GetContinue() + if continueToken == "" { + break + } + } + if len(results) != len(records) { + return nil, nil, fmt.Errorf("expected %d live Compositions, found %d", len(records), len(results)) + } + expected := []string{} + for _, readiness := range r.plan.Spec.Setup.Readiness { + if readiness.Condition.Status == "MissingInputs" { + expected = append(expected, readiness.Condition.ExpectedMissingInputs...) + break + } + } + if len(expected) == 0 { + for _, resource := range r.plan.Spec.Test.Phases[0].Resources { + expected = append(expected, resource.Name) + } + } + if err := r.store.Update(func(data *stressstate.State) error { + for key, result := range results { + if existing := data.CompositionState[key]; existing != nil { + result.InputRevisionObservedAt = existing.InputRevisionObservedAt + result.ExitedMissingInputsAt = existing.ExitedMissingInputsAt + result.SynthesisInitializedAt = existing.SynthesisInitializedAt + result.SynthesisInitializedObservedAt = existing.SynthesisInitializedObservedAt + result.SynthesisCompletedAt = existing.SynthesisCompletedAt + result.SynthesisCompletedObservedAt = existing.SynthesisCompletedObservedAt + result.ReconciledAt = existing.ReconciledAt + result.ReconciledObservedAt = existing.ReconciledObservedAt + result.ReadyAt = existing.ReadyAt + result.ReadyObservedAt = existing.ReadyObservedAt + result.OutputObservedAt = existing.OutputObservedAt + result.OutputValid = existing.OutputValid + result.SynthesisUUID = existing.SynthesisUUID + result.LastStatus = existing.LastStatus + result.Failure = existing.Failure + } + data.CompositionState[key] = result + if data.NamespaceInputs[result.Namespace] == nil { + data.NamespaceInputs[result.Namespace] = map[string]*stressstate.InputTiming{} + } + } + return nil + }); err != nil { + return nil, nil, err + } + return records, expected, nil +} + +func (r *runtime) executePhases(ctx context.Context, monitor *runMonitor) error { + var namespaces []string + var runID string + r.store.Read(func(data *stressstate.State) error { + namespaces = append(namespaces, data.Namespaces...) + runID = data.RunID + return nil + }) + for _, phase := range r.plan.Spec.Test.Phases { + for iteration := 1; iteration <= phase.Repetitions; iteration++ { + for start := 0; start < len(namespaces); start += phase.BatchSize { + end := min(start+phase.BatchSize, len(namespaces)) + batch := namespaces[start:end] + taskCount := len(batch) * len(phase.Resources) + r.output("phase %s iteration %d: namespaces %d-%d (%d objects)\n", phase.Name, iteration, start+1, end, taskCount) + if err := parallel(ctx, phase.Concurrency, taskCount, func(ctx context.Context, task int) error { + namespaceOffset := task / len(phase.Resources) + resourceOffset := task % len(phase.Resources) + namespace := batch[namespaceOffset] + namespaceIndex := start + namespaceOffset + 1 + resourceSpec := phase.Resources[resourceOffset] + context := render.Context{ + RunID: runID, + PlanName: r.plan.Metadata.Name, + Namespace: namespace, + NamespaceIndex: namespaceIndex, + Phase: phase.Name, + Iteration: iteration, + Variables: r.plan.Spec.Run.Variables, + Labels: r.plan.Spec.Run.Labels, + } + object, err := render.Resource(r.baseDir, resourceSpec, context) + if err != nil { + return err + } + startedAt := time.Now() + if resourceSpec.Operation == "create" { + created, resource, err := r.client.CreateOwned(ctx, object, runID, false) + acknowledgedAt := time.Now() + if err != nil { + return err + } + if err := recordResourceMemory(r.store, resourceSpec.Name, phase.Name, created, resource, startedAt, acknowledgedAt); err != nil { + return err + } + return monitor.InputAcknowledged(namespace, resourceSpec.Name, created, startedAt, acknowledgedAt, iteration > 1) + } + created, resource, err := r.client.ApplyOwned(ctx, object, runID) + acknowledgedAt := time.Now() + if err != nil { + return err + } + if err := recordResourceMemory(r.store, resourceSpec.Name, phase.Name, created, resource, startedAt, acknowledgedAt); err != nil { + return err + } + return monitor.InputAcknowledged(namespace, resourceSpec.Name, created, startedAt, acknowledgedAt, true) + }); err != nil { + return fmt.Errorf("phase %s iteration %d: %w", phase.Name, iteration, err) + } + if err := r.store.Save(); err != nil { + return fmt.Errorf("persisting phase %s iteration %d: %w", phase.Name, iteration, err) + } + if end < len(namespaces) { + delay := phase.Delay + if delay == "" { + delay = r.plan.Spec.Run.BatchDelay + } + if delay != "" { + duration, _ := time.ParseDuration(delay) + timer := time.NewTimer(duration) + select { + case <-ctx.Done(): + timer.Stop() + return ctx.Err() + case <-timer.C: + } + } + } + } + } + } + return nil +} + +func (r *runtime) writeReport() error { + var result *report.Report + var runID string + r.store.Read(func(data *stressstate.State) error { + runID = data.RunID + result = report.Build(data) + return nil + }) + path := report.ResolvePath(r.baseDir, r.plan.Spec.Metrics.ReportFile, runID) + if err := report.Write(path, result); err != nil { + return err + } + report.Print(result, r.output) + r.output("JSON report: %s\n", filepath.Clean(path)) + if len(result.Failures) > 0 { + return fmt.Errorf("run completed with %d failures", len(result.Failures)) + } + return nil +} diff --git a/e2e/stress/internal/runner/status_cleanup.go b/e2e/stress/internal/runner/status_cleanup.go new file mode 100644 index 00000000..f0819b1a --- /dev/null +++ b/e2e/stress/internal/runner/status_cleanup.go @@ -0,0 +1,66 @@ +// Status reporting and deterministic owned-resource cleanup. +package runner + +import ( + "context" + "fmt" + "time" + + "github.com/Azure/eno/e2e/stress/internal/kube" + "github.com/Azure/eno/e2e/stress/internal/report" + stressstate "github.com/Azure/eno/e2e/stress/internal/state" +) + +func Status(options Options) error { + store, err := stressstate.Load(options.StatePath) + if err != nil { + return err + } + output := options.Output + if output == nil { + output = func(string, ...any) {} + } + return store.Read(func(data *stressstate.State) error { + output("run ID: %s\nphase: %s\ncompletion stage: %s\nnamespaces: %d\ncompositions: %d\nresources: %d\n", data.RunID, data.Phase, data.CompletionStage, len(data.Namespaces), len(data.CompositionState), len(data.Resources)) + report.Print(report.Build(data), output) + return nil + }) +} + +func Cleanup(ctx context.Context, options Options) error { + store, err := stressstate.Load(options.StatePath) + if err != nil { + return err + } + client, err := kube.New(options.Kubeconfig, 200, 400) + if err != nil { + return err + } + output := options.Output + if output == nil { + output = func(string, ...any) {} + } + var resources []stressstate.Resource + var runID string + store.Read(func(data *stressstate.State) error { + runID = data.RunID + resources = append(resources, data.Resources...) + return nil + }) + for index := len(resources) - 1; index >= 0; index-- { + resource := resources[index] + if resource.Kind != "Namespace" || resource.LogicalName != "namespace" { + continue + } + if err := client.DeleteOwned(ctx, gvrFor(resource), resource.Namespace, resource.Name, resource.UID, runID); err != nil { + return fmt.Errorf("deleting %s %s/%s: %w", resource.Kind, resource.Namespace, resource.Name, err) + } + output("deleted %s %s/%s\n", resource.Kind, resource.Namespace, resource.Name) + } + now := time.Now() + return store.Update(func(data *stressstate.State) error { + data.Phase = "cleaned" + data.CompletedAt = &now + return nil + }) +} diff --git a/e2e/stress/internal/runner/validate.go b/e2e/stress/internal/runner/validate.go new file mode 100644 index 00000000..3bd1786b --- /dev/null +++ b/e2e/stress/internal/runner/validate.go @@ -0,0 +1,110 @@ +// Plan validation and expansion for the stress CLI. +package runner + +import ( + "context" + "fmt" + "sort" + + "k8s.io/apimachinery/pkg/apis/meta/v1/unstructured" + + "github.com/Azure/eno/e2e/stress/internal/plan" + "github.com/Azure/eno/e2e/stress/internal/render" +) + +func Validate(ctx context.Context, options Options) error { + r, err := loadRuntime(options, false) + if err != nil { + return err + } + namespaces := namespaceNames(r.plan, "validation-00000000") + type identity struct { + phase string + name string + } + identities := map[string]identity{} + counts := map[string]int{"Namespace": len(namespaces)} + renderOne := func(phase string, resource plan.ResourceSpec, context render.Context) error { + object, err := render.Resource(r.baseDir, resource, context) + if err != nil { + return err + } + mapped, err := r.client.ResourceFor(object) + if err != nil { + return err + } + key := mapped.GVR.String() + "/" + object.GetNamespace() + "/" + object.GetName() + if previous, found := identities[key]; found && resource.Operation != "update" && resource.Operation != "apply" { + return fmt.Errorf("duplicate rendered identity %s from %s/%s and %s/%s", key, previous.phase, previous.name, phase, resource.Name) + } + identities[key] = identity{phase: phase, name: resource.Name} + counts[object.GetKind()]++ + if options.ServerDryRun { + if _, _, err := r.client.CreateSetup(ctx, object, context.RunID, true, resource.Reuse); err != nil { + return fmt.Errorf("server dry-run %s/%s: %w", phase, resource.Name, err) + } + } + return nil + } + + ordered, err := orderedSetup(r.plan.Spec.Setup.Resources) + if err != nil { + return err + } + for _, resource := range ordered { + if resource.Operation == "observe" { + probe := &unstructured.Unstructured{} + probe.SetAPIVersion(resource.APIVersion) + probe.SetKind(resource.Kind) + if resource.ForEachNamespace && len(namespaces) > 0 { + probe.SetNamespace(namespaces[0]) + } + if _, err := r.client.ResourceFor(probe); err != nil { + return err + } + count := resource.ExpandedCount() + if resource.ForEachNamespace { + count *= len(namespaces) + } + counts[resource.Kind] += count + continue + } + for _, context := range renderContexts(r.plan, "validation-00000000", namespaces, "setup", 1, resource) { + if err := renderOne("setup", resource, context); err != nil { + return err + } + } + } + for _, phase := range r.plan.Spec.Test.Phases { + for _, resource := range phase.Resources { + for _, context := range renderContexts(r.plan, "validation-00000000", namespaces, phase.Name, 1, resource) { + if err := renderOne(phase.Name, resource, context); err != nil { + return err + } + } + } + } + + kinds := make([]string, 0, len(counts)) + for kind := range counts { + kinds = append(kinds, kind) + } + sort.Strings(kinds) + r.output("plan %q is valid\n", r.plan.Metadata.Name) + r.output("expected resources:\n") + for _, kind := range kinds { + r.output(" %-24s %d\n", kind, counts[kind]) + } + r.output("execution plan:\n prepare: namespaces -> ") + for index, resource := range ordered { + if index > 0 { + r.output(" -> ") + } + r.output("%s", resource.Name) + } + r.output(" -> readiness\n") + for _, phase := range r.plan.Spec.Test.Phases { + r.output(" run: %s (concurrency=%d batchSize=%d repetitions=%d)\n", phase.Name, phase.Concurrency, phase.BatchSize, phase.Repetitions) + } + return nil +} diff --git a/e2e/stress/internal/state/state.go b/e2e/stress/internal/state/state.go new file mode 100644 index 00000000..b6a93978 --- /dev/null +++ b/e2e/stress/internal/state/state.go @@ -0,0 +1,182 @@ +package state + +import ( + "crypto/sha256" + "encoding/hex" + "encoding/json" + "fmt" + "os" + "path/filepath" + "sync" + "time" +) + +const SchemaVersion = "v1alpha2" + +type State struct { + SchemaVersion string `json:"schemaVersion"` + RunID string `json:"runID"` + PlanName string `json:"planName"` + PlanPath string `json:"planPath"` + PlanDigest string `json:"planDigest"` + CompletionStage string `json:"completionStage"` + Phase string `json:"phase"` + CreatedAt time.Time `json:"createdAt"` + PreparedAt *time.Time `json:"preparedAt,omitempty"` + WorkloadStart *time.Time `json:"workloadStart,omitempty"` + CompletedAt *time.Time `json:"completedAt,omitempty"` + Namespaces []string `json:"namespaces"` + Resources []Resource `json:"resources"` + NamespaceInputs map[string]map[string]*InputTiming `json:"namespaceInputs"` + CompositionState map[string]*CompositionResult `json:"compositionState"` +} + +type Resource struct { + LogicalName string `json:"logicalName"` + Phase string `json:"phase"` + APIVersion string `json:"apiVersion"` + Kind string `json:"kind"` + Group string `json:"group,omitempty"` + Version string `json:"version"` + Resource string `json:"resource"` + Scope string `json:"scope"` + Namespace string `json:"namespace,omitempty"` + Name string `json:"name"` + UID string `json:"uid"` + ResourceVersion string `json:"resourceVersion"` + CreationTimestamp time.Time `json:"creationTimestamp"` + RequestStartedAt time.Time `json:"requestStartedAt"` + APIAcknowledgedAt time.Time `json:"apiAcknowledgedAt"` +} + +type CompositionResult struct { + Namespace string `json:"namespace"` + CompositionName string `json:"compositionName"` + Variation string `json:"variation"` + InputRevisionObservedAt *time.Time `json:"inputRevisionObservedAt,omitempty"` + ExitedMissingInputsAt *time.Time `json:"exitedMissingInputsAt,omitempty"` + SynthesisInitializedAt *time.Time `json:"synthesisInitializedAt,omitempty"` + SynthesisInitializedObservedAt *time.Time `json:"synthesisInitializedObservedAt,omitempty"` + SynthesisCompletedAt *time.Time `json:"synthesisCompletedAt,omitempty"` + SynthesisCompletedObservedAt *time.Time `json:"synthesisCompletedObservedAt,omitempty"` + ReconciledAt *time.Time `json:"reconciledAt,omitempty"` + ReconciledObservedAt *time.Time `json:"reconciledObservedAt,omitempty"` + ReadyAt *time.Time `json:"readyAt,omitempty"` + ReadyObservedAt *time.Time `json:"readyObservedAt,omitempty"` + OutputObservedAt *time.Time `json:"outputObservedAt,omitempty"` + OutputValid bool `json:"outputValid"` + SynthesisUUID string `json:"synthesisUUID,omitempty"` + LastStatus string `json:"lastStatus,omitempty"` + Failure string `json:"failure,omitempty"` +} + +type InputTiming struct { + Name string `json:"name"` + Kind string `json:"kind"` + RequestStartedAt time.Time `json:"requestStartedAt"` + APIAcknowledgedAt time.Time `json:"apiAcknowledgedAt"` + ResourceVersion string `json:"resourceVersion"` + UID string `json:"uid"` +} + +type Store struct { + path string + mu sync.RWMutex + data *State +} + +func NewStore(path string, data *State) *Store { + return &Store{path: path, data: data} +} + +func Load(path string) (*Store, error) { + raw, err := os.ReadFile(path) + if err != nil { + return nil, fmt.Errorf("reading state: %w", err) + } + data := &State{} + if err := json.Unmarshal(raw, data); err != nil { + return nil, fmt.Errorf("decoding state: %w", err) + } + if data.SchemaVersion != SchemaVersion && data.SchemaVersion != "v1alpha1" { + return nil, fmt.Errorf("unsupported state schema %q", data.SchemaVersion) + } + if data.NamespaceInputs == nil { + data.NamespaceInputs = map[string]map[string]*InputTiming{} + } + if data.CompositionState == nil { + data.CompositionState = map[string]*CompositionResult{} + } + return NewStore(path, data), nil +} + +func (s *Store) Read(fn func(*State) error) error { + s.mu.RLock() + defer s.mu.RUnlock() + return fn(s.data) +} + +func (s *Store) Update(fn func(*State) error) error { + s.mu.Lock() + defer s.mu.Unlock() + if err := fn(s.data); err != nil { + return err + } + return save(s.path, s.data) +} + +func (s *Store) Mutate(fn func(*State) error) error { + s.mu.Lock() + defer s.mu.Unlock() + return fn(s.data) +} + +func (s *Store) Save() error { + s.mu.RLock() + defer s.mu.RUnlock() + return save(s.path, s.data) +} + +func save(path string, data *State) error { + raw, err := json.MarshalIndent(data, "", " ") + if err != nil { + return fmt.Errorf("encoding state: %w", err) + } + if err := os.MkdirAll(filepath.Dir(path), 0o755); err != nil { + return fmt.Errorf("creating state directory: %w", err) + } + temporary, err := os.CreateTemp(filepath.Dir(path), ".eno-stress-state-*") + if err != nil { + return fmt.Errorf("creating temporary state: %w", err) + } + temporaryPath := temporary.Name() + defer os.Remove(temporaryPath) + if err := temporary.Chmod(0o600); err != nil { + temporary.Close() + return err + } + if _, err := temporary.Write(raw); err != nil { + temporary.Close() + return fmt.Errorf("writing state: %w", err) + } + if err := temporary.Sync(); err != nil { + temporary.Close() + return fmt.Errorf("syncing state: %w", err) + } + if err := temporary.Close(); err != nil { + return fmt.Errorf("closing state: %w", err) + } + if err := os.Rename(temporaryPath, path); err != nil { + return fmt.Errorf("replacing state: %w", err) + } + return nil +} + +func FileDigest(path string) (string, error) { + raw, err := os.ReadFile(path) + if err != nil { + return "", err + } + sum := sha256.Sum256(raw) + return "sha256:" + hex.EncodeToString(sum[:]), nil +} diff --git a/e2e/stress/plans/missing-input-fanout/manifests/symphony.yaml b/e2e/stress/plans/missing-input-fanout/manifests/symphony.yaml new file mode 100644 index 00000000..e908978e --- /dev/null +++ b/e2e/stress/plans/missing-input-fanout/manifests/symphony.yaml @@ -0,0 +1,282 @@ +apiVersion: eno.azure.io/v1 +kind: Symphony +metadata: + name: stress-symphony +spec: + bindings: + - key: input1 + resource: + name: input1 + namespace: "${namespace}" + - key: input2 + resource: + name: input2 + namespace: "${namespace}" + - key: input3 + resource: + name: input3 + namespace: "${namespace}" + synthesisEnv: + - name: ENO_STRESS_RUN_ID + value: "${runID}" + variations: + - labels: + stress.eno.azure.io/run-id: "${runID}" + stress.eno.azure.io/plan: missing-input-fanout + stress.eno.azure.io/resource-id: composition + stress.eno.azure.io/variation: "1" + eno.azure.io/stress-test: "true" + synthesizer: + name: eno-stress-missing-inputs-1 + synthesisEnv: + - name: ENO_STRESS_VARIATION + value: "1" + - name: ENO_STRESS_OUTPUT_NAME + value: combined-output-1 + - labels: + stress.eno.azure.io/run-id: "${runID}" + stress.eno.azure.io/plan: missing-input-fanout + stress.eno.azure.io/resource-id: composition + stress.eno.azure.io/variation: "2" + eno.azure.io/stress-test: "true" + synthesizer: + name: eno-stress-missing-inputs-2 + synthesisEnv: + - name: ENO_STRESS_VARIATION + value: "2" + - name: ENO_STRESS_OUTPUT_NAME + value: combined-output-2 + - labels: + stress.eno.azure.io/run-id: "${runID}" + stress.eno.azure.io/plan: missing-input-fanout + stress.eno.azure.io/resource-id: composition + stress.eno.azure.io/variation: "3" + eno.azure.io/stress-test: "true" + synthesizer: + name: eno-stress-missing-inputs-3 + synthesisEnv: + - name: ENO_STRESS_VARIATION + value: "3" + - name: ENO_STRESS_OUTPUT_NAME + value: combined-output-3 + - labels: + stress.eno.azure.io/run-id: "${runID}" + stress.eno.azure.io/plan: missing-input-fanout + stress.eno.azure.io/resource-id: composition + stress.eno.azure.io/variation: "4" + eno.azure.io/stress-test: "true" + synthesizer: + name: eno-stress-missing-inputs-4 + synthesisEnv: + - name: ENO_STRESS_VARIATION + value: "4" + - name: ENO_STRESS_OUTPUT_NAME + value: combined-output-4 + - labels: + stress.eno.azure.io/run-id: "${runID}" + stress.eno.azure.io/plan: missing-input-fanout + stress.eno.azure.io/resource-id: composition + stress.eno.azure.io/variation: "5" + eno.azure.io/stress-test: "true" + synthesizer: + name: eno-stress-missing-inputs-5 + synthesisEnv: + - name: ENO_STRESS_VARIATION + value: "5" + - name: ENO_STRESS_OUTPUT_NAME + value: combined-output-5 + - labels: + stress.eno.azure.io/run-id: "${runID}" + stress.eno.azure.io/plan: missing-input-fanout + stress.eno.azure.io/resource-id: composition + stress.eno.azure.io/variation: "6" + eno.azure.io/stress-test: "true" + synthesizer: + name: eno-stress-missing-inputs-6 + synthesisEnv: + - name: ENO_STRESS_VARIATION + value: "6" + - name: ENO_STRESS_OUTPUT_NAME + value: combined-output-6 + - labels: + stress.eno.azure.io/run-id: "${runID}" + stress.eno.azure.io/plan: missing-input-fanout + stress.eno.azure.io/resource-id: composition + stress.eno.azure.io/variation: "7" + eno.azure.io/stress-test: "true" + synthesizer: + name: eno-stress-missing-inputs-7 + synthesisEnv: + - name: ENO_STRESS_VARIATION + value: "7" + - name: ENO_STRESS_OUTPUT_NAME + value: combined-output-7 + - labels: + stress.eno.azure.io/run-id: "${runID}" + stress.eno.azure.io/plan: missing-input-fanout + stress.eno.azure.io/resource-id: composition + stress.eno.azure.io/variation: "8" + eno.azure.io/stress-test: "true" + synthesizer: + name: eno-stress-missing-inputs-8 + synthesisEnv: + - name: ENO_STRESS_VARIATION + value: "8" + - name: ENO_STRESS_OUTPUT_NAME + value: combined-output-8 + - labels: + stress.eno.azure.io/run-id: "${runID}" + stress.eno.azure.io/plan: missing-input-fanout + stress.eno.azure.io/resource-id: composition + stress.eno.azure.io/variation: "9" + eno.azure.io/stress-test: "true" + synthesizer: + name: eno-stress-missing-inputs-9 + synthesisEnv: + - name: ENO_STRESS_VARIATION + value: "9" + - name: ENO_STRESS_OUTPUT_NAME + value: combined-output-9 + - labels: + stress.eno.azure.io/run-id: "${runID}" + stress.eno.azure.io/plan: missing-input-fanout + stress.eno.azure.io/resource-id: composition + stress.eno.azure.io/variation: "10" + eno.azure.io/stress-test: "true" + synthesizer: + name: eno-stress-missing-inputs-10 + synthesisEnv: + - name: ENO_STRESS_VARIATION + value: "10" + - name: ENO_STRESS_OUTPUT_NAME + value: combined-output-10 + - labels: + stress.eno.azure.io/run-id: "${runID}" + stress.eno.azure.io/plan: missing-input-fanout + stress.eno.azure.io/resource-id: composition + stress.eno.azure.io/variation: "11" + eno.azure.io/stress-test: "true" + synthesizer: + name: eno-stress-missing-inputs-11 + synthesisEnv: + - name: ENO_STRESS_VARIATION + value: "11" + - name: ENO_STRESS_OUTPUT_NAME + value: combined-output-11 + - labels: + stress.eno.azure.io/run-id: "${runID}" + stress.eno.azure.io/plan: missing-input-fanout + stress.eno.azure.io/resource-id: composition + stress.eno.azure.io/variation: "12" + eno.azure.io/stress-test: "true" + synthesizer: + name: eno-stress-missing-inputs-12 + synthesisEnv: + - name: ENO_STRESS_VARIATION + value: "12" + - name: ENO_STRESS_OUTPUT_NAME + value: combined-output-12 + - labels: + stress.eno.azure.io/run-id: "${runID}" + stress.eno.azure.io/plan: missing-input-fanout + stress.eno.azure.io/resource-id: composition + stress.eno.azure.io/variation: "13" + eno.azure.io/stress-test: "true" + synthesizer: + name: eno-stress-missing-inputs-13 + synthesisEnv: + - name: ENO_STRESS_VARIATION + value: "13" + - name: ENO_STRESS_OUTPUT_NAME + value: combined-output-13 + - labels: + stress.eno.azure.io/run-id: "${runID}" + stress.eno.azure.io/plan: missing-input-fanout + stress.eno.azure.io/resource-id: composition + stress.eno.azure.io/variation: "14" + eno.azure.io/stress-test: "true" + synthesizer: + name: eno-stress-missing-inputs-14 + synthesisEnv: + - name: ENO_STRESS_VARIATION + value: "14" + - name: ENO_STRESS_OUTPUT_NAME + value: combined-output-14 + - labels: + stress.eno.azure.io/run-id: "${runID}" + stress.eno.azure.io/plan: missing-input-fanout + stress.eno.azure.io/resource-id: composition + stress.eno.azure.io/variation: "15" + eno.azure.io/stress-test: "true" + synthesizer: + name: eno-stress-missing-inputs-15 + synthesisEnv: + - name: ENO_STRESS_VARIATION + value: "15" + - name: ENO_STRESS_OUTPUT_NAME + value: combined-output-15 + - labels: + stress.eno.azure.io/run-id: "${runID}" + stress.eno.azure.io/plan: missing-input-fanout + stress.eno.azure.io/resource-id: composition + stress.eno.azure.io/variation: "16" + eno.azure.io/stress-test: "true" + synthesizer: + name: eno-stress-missing-inputs-16 + synthesisEnv: + - name: ENO_STRESS_VARIATION + value: "16" + - name: ENO_STRESS_OUTPUT_NAME + value: combined-output-16 + - labels: + stress.eno.azure.io/run-id: "${runID}" + stress.eno.azure.io/plan: missing-input-fanout + stress.eno.azure.io/resource-id: composition + stress.eno.azure.io/variation: "17" + eno.azure.io/stress-test: "true" + synthesizer: + name: eno-stress-missing-inputs-17 + synthesisEnv: + - name: ENO_STRESS_VARIATION + value: "17" + - name: ENO_STRESS_OUTPUT_NAME + value: combined-output-17 + - labels: + stress.eno.azure.io/run-id: "${runID}" + stress.eno.azure.io/plan: missing-input-fanout + stress.eno.azure.io/resource-id: composition + stress.eno.azure.io/variation: "18" + eno.azure.io/stress-test: "true" + synthesizer: + name: eno-stress-missing-inputs-18 + synthesisEnv: + - name: ENO_STRESS_VARIATION + value: "18" + - name: ENO_STRESS_OUTPUT_NAME + value: combined-output-18 + - labels: + stress.eno.azure.io/run-id: "${runID}" + stress.eno.azure.io/plan: missing-input-fanout + stress.eno.azure.io/resource-id: composition + stress.eno.azure.io/variation: "19" + eno.azure.io/stress-test: "true" + synthesizer: + name: eno-stress-missing-inputs-19 + synthesisEnv: + - name: ENO_STRESS_VARIATION + value: "19" + - name: ENO_STRESS_OUTPUT_NAME + value: combined-output-19 + - labels: + stress.eno.azure.io/run-id: "${runID}" + stress.eno.azure.io/plan: missing-input-fanout + stress.eno.azure.io/resource-id: composition + stress.eno.azure.io/variation: "20" + eno.azure.io/stress-test: "true" + synthesizer: + name: eno-stress-missing-inputs-20 + synthesisEnv: + - name: ENO_STRESS_VARIATION + value: "20" + - name: ENO_STRESS_OUTPUT_NAME + value: combined-output-20 diff --git a/e2e/stress/plans/missing-input-fanout/manifests/synthesizer.yaml b/e2e/stress/plans/missing-input-fanout/manifests/synthesizer.yaml new file mode 100644 index 00000000..e669ab38 --- /dev/null +++ b/e2e/stress/plans/missing-input-fanout/manifests/synthesizer.yaml @@ -0,0 +1,24 @@ +apiVersion: eno.azure.io/v1 +kind: Synthesizer +metadata: + name: eno-stress-missing-inputs-${resourceIndex} +spec: + image: "${synthesizerImage}" + command: + - /bin/synthesize + refs: + - key: input1 + resource: + group: "" + version: v1 + kind: ConfigMap + - key: input2 + resource: + group: "" + version: v1 + kind: ConfigMap + - key: input3 + resource: + group: "" + version: v1 + kind: Secret \ No newline at end of file diff --git a/e2e/stress/plans/missing-input-fanout/plan.yaml b/e2e/stress/plans/missing-input-fanout/plan.yaml new file mode 100644 index 00000000..1590cf06 --- /dev/null +++ b/e2e/stress/plans/missing-input-fanout/plan.yaml @@ -0,0 +1,99 @@ +apiVersion: stress.eno.azure.io/v1alpha1 +kind: StressTestPlan +metadata: + name: missing-input-fanout +spec: + run: + namespacePrefix: 6a + namespaceCount: 150 + completionStage: PreSynthesis + concurrency: 50 + batchSize: 150 + timeout: 90m + clientQPS: 200 + clientBurst: 400 + labels: + eno.azure.io/stress-test: "true" + variables: + synthesizerImage: ruinanliuacr.azurecr.io/eno-stress-synthesizer/0.0.2:latest + setup: + resources: + - name: synthesizer + count: 20 + reuse: true + scope: cluster + templateFile: manifests/synthesizer.yaml + - name: symphony + scope: namespace + forEachNamespace: true + dependsOn: + - synthesizer + templateFile: manifests/symphony.yaml + - name: composition + count: 20 + apiVersion: eno.azure.io/v1 + kind: Composition + scope: namespace + forEachNamespace: true + operation: observe + dependsOn: + - symphony + readiness: + - resource: composition + condition: + status: MissingInputs + expectedMissingInputs: + - input1 + - input2 + - input3 + test: + phases: + - name: create-inputs + concurrency: 50 + batchSize: 150 + resources: + - name: input1 + apiVersion: v1 + kind: ConfigMap + scope: namespace + forEachNamespace: true + template: + data: + value: "${namespace}-input1" + - name: input2 + apiVersion: v1 + kind: ConfigMap + scope: namespace + forEachNamespace: true + template: + data: + value: "${namespace}-input2" + - name: input3 + apiVersion: v1 + kind: Secret + scope: namespace + forEachNamespace: true + template: + stringData: + value: "${namespace}-input3" + successCriteria: + - compositionsExitMissingInputs: 100% + - p95InputToExitMissingInputs: <30s + - maxInputToExitMissingInputs: <120s + output: + apiVersion: v1 + kind: ConfigMap + name: combined-output-${variation} + expectedData: + input1: "${namespace}-input1" + input2: "${namespace}-input2" + input3: "${namespace}-input3" + combined: "${namespace}-input1-${namespace}-input2-${namespace}-input3" + metrics: + collect: + - enoControllerResources + - inputRevisionBuffer + - workqueues + - apiServerRequests + - compositionTransitions + reportFile: results/${runID}.json \ No newline at end of file diff --git a/e2e/stress/run-cycles.sh b/e2e/stress/run-cycles.sh new file mode 100755 index 00000000..67751619 --- /dev/null +++ b/e2e/stress/run-cycles.sh @@ -0,0 +1,136 @@ +#!/usr/bin/env bash +set -Eeuo pipefail + +SCRIPT_DIR=$(cd -- "$(dirname -- "${BASH_SOURCE[0]}")" && pwd) +cd "$SCRIPT_DIR" + +CYCLES=${CYCLES:-5} +WAIT_SECONDS=${WAIT_SECONDS:-10} +KUBECTL=${KUBECTL:-kubectl} +ENO_STRESS=${ENO_STRESS:-./eno-stress} +PLAN=${PLAN:-./plan.yaml} +STATE=${STATE:-./state.json} +RESULTS_FILE=${RESULTS_FILE:-./results.txt} +STATE_HISTORY=${STATE_HISTORY:-./state-history} +KUBECONFIG=${KUBECONFIG:-./e2e-underlay-kubeconfig/kubeconfig-cx-1/kubeconfig-cx-1} + +mkdir -p "$STATE_HISTORY" +touch "$RESULTS_FILE" + +log() { + printf '[%s] %s\n' "$(date --iso-8601=seconds)" "$*" | tee -a "$RESULTS_FILE" +} + +run_logged() { + set +e + "$@" 2>&1 | tee -a "$RESULTS_FILE" + local status=${PIPESTATUS[0]} + set -e + return "$status" +} + +stress_namespaces() { + "$KUBECTL" --kubeconfig "$KUBECONFIG" get namespaces \ + -o jsonpath='{range .items[*]}{.metadata.name}{"\t"}{.metadata.deletionTimestamp}{"\n"}{end}' | + awk '$1 ~ /^6a[0-9a-f]{16}$/ {print}' +} + +wait_for_no_stress_namespaces() { + while true; do + local namespaces + namespaces=$(stress_namespaces) + if [[ -z "$namespaces" ]]; then + log 'No stress namespaces remain.' + return + fi + + local total terminating active + total=$(awk 'NF {count++} END {print count+0}' <<<"$namespaces") + terminating=$(awk 'NF >= 2 && $2 != "/dev/null || printf 'unknown-run') + timestamp=$(date -u +%Y%m%dT%H%M%SZ) + mv "$STATE" "$STATE_HISTORY/${run_id}-${timestamp}.json" +} + +cleanup_state() { + [[ -f "$STATE" ]] || return 0 + log "Cleaning namespaces recorded in $STATE" + run_logged "$ENO_STRESS" cleanup --state "$STATE" --kubeconfig "$KUBECONFIG" +} + +on_exit() { + local status=$? + trap - EXIT INT TERM + if [[ -f "$STATE" ]]; then + cleanup_state || true + fi + exit "$status" +} +trap on_exit EXIT INT TERM + +if ! [[ "$CYCLES" =~ ^[1-9][0-9]*$ ]]; then + log "CYCLES must be a positive integer, got: $CYCLES" + exit 2 +fi +if [[ ! -x "$ENO_STRESS" ]]; then + log "Stress binary is not executable: $ENO_STRESS" + exit 2 +fi +if [[ ! -f "$PLAN" ]]; then + log "Plan does not exist: $PLAN" + exit 2 +fi +if [[ ! -f "$KUBECONFIG" ]]; then + log "Kubeconfig does not exist: $KUBECONFIG" + exit 2 +fi + +log "Validating plan before $CYCLES cycles." +run_logged "$ENO_STRESS" validate --plan "$PLAN" --kubeconfig "$KUBECONFIG" + +if [[ -f "$STATE" ]]; then + log 'Found existing state; cleaning and archiving it before the first cycle.' + cleanup_state + archive_state +fi +wait_for_no_stress_namespaces + +successful=0 +failed=0 +for ((cycle = 1; cycle <= CYCLES; cycle++)); do + log "========== cycle $cycle/$CYCLES: prepare ==========" + if ! run_logged "$ENO_STRESS" prepare --plan "$PLAN" --state "$STATE" --kubeconfig "$KUBECONFIG"; then + log "Cycle $cycle prepare failed." + failed=$((failed + 1)) + cleanup_state || true + archive_state + wait_for_no_stress_namespaces + continue + fi + + log "========== cycle $cycle/$CYCLES: run ==========" + if run_logged "$ENO_STRESS" run --state "$STATE" --kubeconfig "$KUBECONFIG"; then + log "Cycle $cycle run completed successfully." + successful=$((successful + 1)) + else + log "Cycle $cycle run failed." + failed=$((failed + 1)) + fi + + log "========== cycle $cycle/$CYCLES: cleanup ==========" + cleanup_state || log "Cycle $cycle cleanup returned an error." + archive_state + wait_for_no_stress_namespaces +done + +log "All cycles finished: successful=$successful failed=$failed" +trap - EXIT INT TERM \ No newline at end of file diff --git a/e2e/stress/synthesizer/Dockerfile b/e2e/stress/synthesizer/Dockerfile new file mode 100644 index 00000000..3d0975a7 --- /dev/null +++ b/e2e/stress/synthesizer/Dockerfile @@ -0,0 +1,15 @@ +FROM mcr.microsoft.com/devcontainers/go:1.25 AS builder + +ENV GOTOOLCHAIN=auto +ENV GOWORK=off +WORKDIR /app + +COPY go.mod go.sum ./ +RUN --mount=type=cache,target=/root/.cache/go-build go mod download + +COPY . . +RUN --mount=type=cache,target=/root/.cache/go-build CGO_ENABLED=0 go build -o /eno-stress-synthesizer ./e2e/stress/synthesizer + +FROM scratch +USER 65532:65532 +COPY --from=builder /eno-stress-synthesizer /bin/synthesize \ No newline at end of file diff --git a/e2e/stress/synthesizer/main.go b/e2e/stress/synthesizer/main.go new file mode 100644 index 00000000..eb0ba17e --- /dev/null +++ b/e2e/stress/synthesizer/main.go @@ -0,0 +1,67 @@ +// Command eno-stress-synthesizer produces one deterministic ConfigMap from three inputs. +package main + +import ( + "fmt" + "os" + + "github.com/Azure/eno/pkg/function" + corev1 "k8s.io/api/core/v1" + metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" + "k8s.io/apimachinery/pkg/runtime" + "sigs.k8s.io/controller-runtime/pkg/client" +) + +type inputs struct { + Input1 *corev1.ConfigMap `eno_key:"input1"` + Input2 *corev1.ConfigMap `eno_key:"input2"` + Input3 *corev1.Secret `eno_key:"input3"` +} + +func synthesize(input inputs) ([]client.Object, error) { + input1, found := input.Input1.Data["value"] + if !found { + return nil, fmt.Errorf("input1 ConfigMap has no data.value") + } + input2, found := input.Input2.Data["value"] + if !found { + return nil, fmt.Errorf("input2 ConfigMap has no data.value") + } + input3Raw, found := input.Input3.Data["value"] + if !found { + return nil, fmt.Errorf("input3 Secret has no data.value") + } + input3 := string(input3Raw) + variation := os.Getenv("ENO_STRESS_VARIATION") + outputName := os.Getenv("ENO_STRESS_OUTPUT_NAME") + if outputName == "" { + outputName = "combined-output" + } + + output := &corev1.ConfigMap{ + TypeMeta: metav1.TypeMeta{APIVersion: "v1", Kind: "ConfigMap"}, + ObjectMeta: metav1.ObjectMeta{ + Name: outputName, + Namespace: os.Getenv("COMPOSITION_NAMESPACE"), + Labels: map[string]string{ + "stress.eno.azure.io/run-id": os.Getenv("ENO_STRESS_RUN_ID"), + "stress.eno.azure.io/variation": variation, + }, + }, + Data: map[string]string{ + "input1": input1, + "input2": input2, + "input3": input3, + "combined": input1 + "-" + input2 + "-" + input3, + }, + } + return []client.Object{output}, nil +} + +func main() { + scheme := runtime.NewScheme() + if err := corev1.AddToScheme(scheme); err != nil { + panic(err) + } + function.Main(synthesize, function.WithScheme(scheme)) +}