diff --git a/.pipelines/cni/load-test-templates/migration-fault-injection-template.yaml b/.pipelines/cni/load-test-templates/migration-fault-injection-template.yaml new file mode 100644 index 00000000000..63feb99fb31 --- /dev/null +++ b/.pipelines/cni/load-test-templates/migration-fault-injection-template.yaml @@ -0,0 +1,56 @@ +parameters: + clusterName: "" + os: "linux" + # linux: cilium, cniv1, cniv2, dualstack; windows: cniv1, cniv2, stateless + cni: "cilium" + # all, add-before-endpoint-commit, delete-after-intent-commit, endpoint-patch, restart-during-scale + scenario: "all" + scaleReplicas: 20 + # Keep the test and task timeouts larger so context cancellation can restore the CNS DaemonSet. + timeoutMinutes: 60 + testTimeoutMinutes: 80 + taskTimeoutMinutes: 95 + runID: "$(Build.BuildId)-$(System.JobId)-$(System.JobAttempt)" + artifactName: "migration-fault-injection-$(System.JobId)" + workloadImage: "mcr.microsoft.com/oss/kubernetes/pause:3.6" + +steps: + - task: AzureCLI@2 + displayName: "Run CNS migration fault injection" + timeoutInMinutes: ${{ parameters.taskTimeoutMinutes }} + inputs: + azureSubscription: $(BUILD_VALIDATIONS_SERVICE_CONNECTION) + scriptLocation: "inlineScript" + scriptType: "bash" + addSpnToEnvironment: true + inlineScript: | + set -euo pipefail + + artifactDir="$(Build.ArtifactStagingDirectory)/migration-fault-injection" + mkdir -p "$artifactDir" + export KUBECONFIG="$(Build.ArtifactStagingDirectory)/migration-fault-injection-kubeconfig" + trap 'rm -f "$KUBECONFIG"' EXIT + make -C ./hack/aks set-kubeconf AZCLI=az CLUSTER=${{ parameters.clusterName }} + + MIGRATION_FAULT_SCENARIO="${{ parameters.scenario }}" \ + MIGRATION_FAULT_OS="${{ parameters.os }}" \ + MIGRATION_FAULT_CNI="${{ parameters.cni }}" \ + MIGRATION_FAULT_RUN_ID="${{ parameters.runID }}" \ + MIGRATION_FAULT_ARTIFACT_DIR="$artifactDir" \ + MIGRATION_FAULT_SCALE_REPLICAS="${{ parameters.scaleReplicas }}" \ + MIGRATION_FAULT_TIMEOUT_MINUTES="${{ parameters.timeoutMinutes }}" \ + MIGRATION_FAULT_WORKLOAD_IMAGE="${{ parameters.workloadImage }}" \ + VALIDATE_STATE_BACKEND=bolt \ + VALIDATE_CONVERGENCE_ATTEMPTS=12 \ + VALIDATE_CONVERGENCE_INTERVAL_SECONDS=10 \ + go test -mod=readonly -count=1 -timeout ${{ parameters.testTimeoutMinutes }}m -tags load \ + ./test/integration/state -run '^TestMigrationFaultInjection$' -v \ + -args -test-kubeconfig="$KUBECONFIG" 2>&1 | + tee "$artifactDir/go-test.log" + + - task: PublishPipelineArtifact@1 + displayName: "Publish migration fault artifacts" + condition: always() + inputs: + targetPath: "$(Build.ArtifactStagingDirectory)/migration-fault-injection" + artifact: "${{ parameters.artifactName }}" diff --git a/cns/restserver/fault_injection.go b/cns/restserver/fault_injection.go new file mode 100644 index 00000000000..e205c7eea18 --- /dev/null +++ b/cns/restserver/fault_injection.go @@ -0,0 +1,213 @@ +package restserver + +import ( + "crypto/subtle" + "encoding/json" + "errors" + "net/http" + "os" + "sync" + "time" +) + +const ( + faultInjectionPath = "/debug/faultinjection" + faultInjectionTokenHeader = "X-CNS-Test-Fault-Token" //nolint:gosec // This is a header name, not a credential. + faultInjectionTokenEnv = "CNS_TEST_FAULT_INJECTION_TOKEN" //nolint:gosec // This is an environment variable name, not a credential. + defaultFaultInjectionTimeout = 10 * time.Minute +) + +type faultPoint string + +const ( + faultPointAddBeforeEndpointCommit faultPoint = "add-before-endpoint-commit" + faultPointDeleteAfterIntentCommit faultPoint = "delete-after-intent-commit" + faultPointPatchBeforeEndpointCommit faultPoint = "patch-before-endpoint-commit" +) + +type faultState string + +const ( + faultStateIdle faultState = "idle" + faultStateArmed faultState = "armed" + faultStateReached faultState = "reached" +) + +var errFaultAlreadyArmed = errors.New("fault injection point is already armed") + +type faultInjector struct { + mu sync.Mutex + token string + timeout time.Duration + point faultPoint + target faultInjectionTarget + state faultState + release chan struct{} +} + +type faultInjectionRequest struct { + Point faultPoint `json:"point"` + Target faultInjectionTarget `json:"target"` +} + +type faultInjectionTarget struct { + PodName string `json:"podName,omitempty"` + PodNamespace string `json:"podNamespace,omitempty"` +} + +type faultInjectionStatus struct { + Point faultPoint `json:"point,omitempty"` + Target faultInjectionTarget `json:"target"` + State faultState `json:"state"` +} + +func newFaultInjectorFromEnv() *faultInjector { + token := os.Getenv(faultInjectionTokenEnv) + if token == "" { + return nil + } + return newFaultInjector(token, defaultFaultInjectionTimeout) +} + +func newFaultInjector(token string, timeout time.Duration) *faultInjector { + return &faultInjector{ + token: token, + timeout: timeout, + state: faultStateIdle, + } +} + +func validFaultPoint(point faultPoint) bool { + switch point { + case faultPointAddBeforeEndpointCommit, + faultPointDeleteAfterIntentCommit, + faultPointPatchBeforeEndpointCommit: + return true + default: + return false + } +} + +func (injector *faultInjector) arm(point faultPoint, target faultInjectionTarget) error { + injector.mu.Lock() + defer injector.mu.Unlock() + + if injector.state != faultStateIdle { + return errFaultAlreadyArmed + } + injector.point = point + injector.target = target + injector.state = faultStateArmed + injector.release = make(chan struct{}) + return nil +} + +func (injector *faultInjector) disarm() { + injector.mu.Lock() + defer injector.mu.Unlock() + + if injector.release != nil { + close(injector.release) + } + injector.point = "" + injector.target = faultInjectionTarget{} + injector.state = faultStateIdle + injector.release = nil +} + +func (injector *faultInjector) checkpoint(point faultPoint, target faultInjectionTarget) { + injector.mu.Lock() + if injector.point != point || injector.state != faultStateArmed || !injector.target.matches(target) { + injector.mu.Unlock() + return + } + release := injector.release + injector.state = faultStateReached + injector.mu.Unlock() + + timer := time.NewTimer(injector.timeout) + defer timer.Stop() + select { + case <-release: + case <-timer.C: + } + + injector.mu.Lock() + if injector.release == release { + injector.point = "" + injector.target = faultInjectionTarget{} + injector.state = faultStateIdle + injector.release = nil + } + injector.mu.Unlock() +} + +func (injector *faultInjector) status() faultInjectionStatus { + injector.mu.Lock() + defer injector.mu.Unlock() + return faultInjectionStatus{ + Point: injector.point, + Target: injector.target, + State: injector.state, + } +} + +func (target faultInjectionTarget) matches(candidate faultInjectionTarget) bool { + return target.PodNamespace == candidate.PodNamespace && + (target.PodName == "" || target.PodName == candidate.PodName) +} + +func (injector *faultInjector) handle(w http.ResponseWriter, r *http.Request) { + if subtle.ConstantTimeCompare([]byte(r.Header.Get(faultInjectionTokenHeader)), []byte(injector.token)) != 1 { + http.Error(w, "forbidden", http.StatusForbidden) + return + } + + switch r.Method { + case http.MethodGet: + writeFaultInjectionStatus(w, injector.status()) + case http.MethodPut: + var request faultInjectionRequest + decoder := json.NewDecoder(http.MaxBytesReader(w, r.Body, 1024)) + decoder.DisallowUnknownFields() + if err := decoder.Decode(&request); err != nil { + http.Error(w, "invalid request", http.StatusBadRequest) + return + } + if !validFaultPoint(request.Point) { + http.Error(w, "invalid fault injection point", http.StatusBadRequest) + return + } + if request.Target.PodNamespace == "" { + http.Error(w, "fault injection target namespace is required", http.StatusBadRequest) + return + } + if err := injector.arm(request.Point, request.Target); err != nil { + http.Error(w, err.Error(), http.StatusConflict) + return + } + writeFaultInjectionStatus(w, injector.status()) + case http.MethodDelete: + injector.disarm() + w.WriteHeader(http.StatusNoContent) + default: + w.Header().Set("Allow", "DELETE, GET, PUT") + http.Error(w, "method not allowed", http.StatusMethodNotAllowed) + } +} + +func writeFaultInjectionStatus(w http.ResponseWriter, status faultInjectionStatus) { + w.Header().Set("Content-Type", "application/json") + if err := json.NewEncoder(w).Encode(status); err != nil { + http.Error(w, "encoding response", http.StatusInternalServerError) + } +} + +func (service *HTTPRestService) reachFaultPoint(point faultPoint, podName, podNamespace string) { + if service.faultInjector != nil { + service.faultInjector.checkpoint(point, faultInjectionTarget{ + PodName: podName, + PodNamespace: podNamespace, + }) + } +} diff --git a/cns/restserver/fault_injection_test.go b/cns/restserver/fault_injection_test.go new file mode 100644 index 00000000000..3e49d2e54a5 --- /dev/null +++ b/cns/restserver/fault_injection_test.go @@ -0,0 +1,249 @@ +package restserver + +import ( + "bytes" + "context" + "encoding/json" + "errors" + "net/http" + "net/http/httptest" + "testing" + "time" + + "github.com/Azure/azure-container-networking/cns" + "github.com/Azure/azure-container-networking/cns/state" + "github.com/Azure/azure-container-networking/cns/types" + "github.com/stretchr/testify/require" +) + +func TestFaultInjectorDisabledByDefault(t *testing.T) { + t.Setenv(faultInjectionTokenEnv, "") + require.Nil(t, newFaultInjectorFromEnv()) + + t.Setenv(faultInjectionTokenEnv, "test-token") + require.NotNil(t, newFaultInjectorFromEnv()) +} + +func TestFaultInjectorHTTPContract(t *testing.T) { + injector := newFaultInjector("test-token", time.Minute) + t.Cleanup(injector.disarm) + target := faultInjectionTarget{PodName: "pod", PodNamespace: "namespace"} + + recorder := invokeFaultInjector(t, injector, http.MethodGet, "", nil) + require.Equal(t, http.StatusForbidden, recorder.Code) + + recorder = invokeFaultInjector(t, injector, http.MethodPost, "test-token", nil) + require.Equal(t, http.StatusMethodNotAllowed, recorder.Code) + require.Equal(t, "DELETE, GET, PUT", recorder.Header().Get("Allow")) + + recorder = invokeFaultInjector(t, injector, http.MethodPut, "test-token", faultInjectionRequest{Point: "unknown"}) + require.Equal(t, http.StatusBadRequest, recorder.Code) + + recorder = invokeFaultInjector(t, injector, http.MethodPut, "test-token", faultInjectionRequest{Point: faultPointAddBeforeEndpointCommit}) + require.Equal(t, http.StatusBadRequest, recorder.Code) + + recorder = invokeFaultInjector(t, injector, http.MethodPut, "test-token", faultInjectionRequest{ + Point: faultPointAddBeforeEndpointCommit, + Target: target, + }) + require.Equal(t, http.StatusOK, recorder.Code) + require.Equal(t, faultInjectionStatus{ + Point: faultPointAddBeforeEndpointCommit, + Target: target, + State: faultStateArmed, + }, decodeFaultStatus(t, recorder)) + + recorder = invokeFaultInjector(t, injector, http.MethodPut, "test-token", faultInjectionRequest{ + Point: faultPointDeleteAfterIntentCommit, + Target: target, + }) + require.Equal(t, http.StatusConflict, recorder.Code) + + injector.checkpoint(faultPointAddBeforeEndpointCommit, faultInjectionTarget{ + PodName: "other-pod", + PodNamespace: target.PodNamespace, + }) + require.Equal(t, faultStateArmed, injector.status().State) + + checkpointDone := make(chan struct{}) + go func() { + injector.checkpoint(faultPointAddBeforeEndpointCommit, target) + close(checkpointDone) + }() + require.Eventually(t, func() bool { + return injector.status().State == faultStateReached + }, time.Second, time.Millisecond) + + recorder = invokeFaultInjector(t, injector, http.MethodGet, "test-token", nil) + require.Equal(t, http.StatusOK, recorder.Code) + require.Equal(t, faultStateReached, decodeFaultStatus(t, recorder).State) + + recorder = invokeFaultInjector(t, injector, http.MethodDelete, "test-token", nil) + require.Equal(t, http.StatusNoContent, recorder.Code) + require.Eventually(t, func() bool { + select { + case <-checkpointDone: + return true + default: + return false + } + }, time.Second, time.Millisecond) + require.Equal(t, faultStateIdle, injector.status().State) +} + +func TestFaultInjectorCheckpointTimeout(t *testing.T) { + injector := newFaultInjector("test-token", time.Millisecond) + target := faultInjectionTarget{PodNamespace: "namespace"} + require.NoError(t, injector.arm(faultPointPatchBeforeEndpointCommit, target)) + + injector.checkpoint(faultPointPatchBeforeEndpointCommit, faultInjectionTarget{ + PodName: "pod", + PodNamespace: target.PodNamespace, + }) + + require.Equal(t, faultStateIdle, injector.status().State) +} + +func TestPersistentFaultPointAddBeforeEndpointCommit(t *testing.T) { + service, _, db, _ := newUnifiedAddFixture(t, map[string][]state.IPRecord{ + unifiedTestNCIPv4: {{ID: testIPID1, IPAddress: testIP1, NCID: unifiedTestNCIPv4, NCVersion: 1}}, + }, nil) + service.faultInjector = newFaultInjector("test-token", time.Minute) + t.Cleanup(service.faultInjector.disarm) + require.NoError(t, service.faultInjector.arm(faultPointAddBeforeEndpointCommit, faultTargetForPod(testPod1Info))) + + req := newTestIPConfigsRequest(t, testPod1Info) + errCh := make(chan error, 1) + go func() { + response, err := service.requestIPConfigHandlerHelper(context.Background(), req) + if err == nil && response.Response.ReturnCode != types.Success { + err = errUnexpectedResponseCode + } + errCh <- err + }() + waitForFaultPoint(t, service.faultInjector, faultPointAddBeforeEndpointCommit) + + snapshot, err := db.Snapshot(context.Background()) + require.NoError(t, err) + require.Empty(t, snapshot.Assignments) + require.Empty(t, snapshot.Endpoints) + ipState := service.PodIPConfigState[testIPID1] + require.Equal(t, types.Available, ipState.GetState()) + + service.faultInjector.disarm() + require.NoError(t, <-errCh) + snapshot, err = db.Snapshot(context.Background()) + require.NoError(t, err) + require.Contains(t, snapshot.Assignments, testPod1Info.Key()) + require.Contains(t, snapshot.Endpoints, testPod1Info.InfraContainerID()) +} + +func TestPersistentFaultPointDeleteAfterIntentCommit(t *testing.T) { + service, _, db, _ := newUnifiedAddFixture(t, map[string][]state.IPRecord{ + unifiedTestNCIPv4: {{ID: testIPID1, IPAddress: testIP1, NCID: unifiedTestNCIPv4, NCVersion: 1}}, + }, nil) + + req := newTestIPConfigsRequest(t, testPod1Info) + response, err := service.requestIPConfigHandlerHelper(context.Background(), req) + require.NoError(t, err) + require.Equal(t, types.Success, response.Response.ReturnCode) + + service.faultInjector = newFaultInjector("test-token", time.Minute) + t.Cleanup(service.faultInjector.disarm) + require.NoError(t, service.faultInjector.arm(faultPointDeleteAfterIntentCommit, faultTargetForPod(testPod1Info))) + errCh := make(chan error, 1) + go func() { + _, releaseErr := service.ReleaseIPConfigHandlerHelper(context.Background(), req) + errCh <- releaseErr + }() + waitForFaultPoint(t, service.faultInjector, faultPointDeleteAfterIntentCommit) + + snapshot, err := db.Snapshot(context.Background()) + require.NoError(t, err) + require.Contains(t, snapshot.DeleteIntents, testPod1Info.InfraContainerID()) + require.Empty(t, snapshot.Assignments) + require.Contains(t, snapshot.Endpoints, testPod1Info.InfraContainerID()) + require.Contains(t, service.EndpointState, testPod1Info.InfraContainerID()) + ipState := service.PodIPConfigState[testIPID1] + require.Equal(t, types.Assigned, ipState.GetState()) + + service.faultInjector.disarm() + require.NoError(t, <-errCh) + require.Contains(t, service.EndpointState, testPod1Info.InfraContainerID()) + ipState = service.PodIPConfigState[testIPID1] + require.Equal(t, types.Available, ipState.GetState()) +} + +func TestPersistentFaultPointPatchBeforeEndpointCommit(t *testing.T) { + service, _, db, _ := newUnifiedAddFixture(t, map[string][]state.IPRecord{ + unifiedTestNCIPv4: {{ID: testIPID1, IPAddress: testIP1, NCID: unifiedTestNCIPv4, NCVersion: 1}}, + }, nil) + + req := newTestIPConfigsRequest(t, testPod1Info) + response, err := service.requestIPConfigHandlerHelper(context.Background(), req) + require.NoError(t, err) + require.Equal(t, types.Success, response.Response.ReturnCode) + + service.faultInjector = newFaultInjector("test-token", time.Minute) + t.Cleanup(service.faultInjector.disarm) + require.NoError(t, service.faultInjector.arm(faultPointPatchBeforeEndpointCommit, faultTargetForPod(testPod1Info))) + errCh := make(chan error, 1) + go func() { + errCh <- service.updateEndpoint( + context.Background(), + testPod1Info.InfraContainerID(), + map[string]*IPInfo{ + InfraInterfaceName: {HnsEndpointID: "hns-endpoint"}, + }, + service.selectedUnifiedStateAdapter(), + ) + }() + waitForFaultPoint(t, service.faultInjector, faultPointPatchBeforeEndpointCommit) + + snapshot, err := db.Snapshot(context.Background()) + require.NoError(t, err) + require.Empty(t, snapshot.Endpoints[testPod1Info.InfraContainerID()].IfnameToIPMap[InfraInterfaceName].HNSEndpointID) + + service.faultInjector.disarm() + require.NoError(t, <-errCh) + snapshot, err = db.Snapshot(context.Background()) + require.NoError(t, err) + require.Equal(t, "hns-endpoint", snapshot.Endpoints[testPod1Info.InfraContainerID()].IfnameToIPMap[InfraInterfaceName].HNSEndpointID) +} + +var errUnexpectedResponseCode = errors.New("unexpected response code") + +func invokeFaultInjector(t *testing.T, injector *faultInjector, method, token string, body any) *httptest.ResponseRecorder { + t.Helper() + var requestBody bytes.Buffer + if body != nil { + require.NoError(t, json.NewEncoder(&requestBody).Encode(body)) + } + request := httptest.NewRequestWithContext(t.Context(), method, faultInjectionPath, &requestBody) + request.Header.Set(faultInjectionTokenHeader, token) + recorder := httptest.NewRecorder() + injector.handle(recorder, request) + return recorder +} + +func decodeFaultStatus(t *testing.T, recorder *httptest.ResponseRecorder) faultInjectionStatus { + t.Helper() + var status faultInjectionStatus + require.NoError(t, json.Unmarshal(recorder.Body.Bytes(), &status)) + return status +} + +func waitForFaultPoint(t *testing.T, injector *faultInjector, point faultPoint) { + t.Helper() + require.Eventually(t, func() bool { + status := injector.status() + return status.Point == point && status.State == faultStateReached + }, time.Second, time.Millisecond) +} + +func faultTargetForPod(podInfo cns.PodInfo) faultInjectionTarget { + return faultInjectionTarget{ + PodName: podInfo.Name(), + PodNamespace: podInfo.Namespace(), + } +} diff --git a/cns/restserver/restserver.go b/cns/restserver/restserver.go index bb2219927d2..8ad0804ea5e 100644 --- a/cns/restserver/restserver.go +++ b/cns/restserver/restserver.go @@ -131,6 +131,7 @@ type HTTPRestService struct { nodeName string stateRestoreLogger stateRestoreLogger unifiedStateAdapter *durableStateAdapter + faultInjector *faultInjector } type CNIConflistGenerator interface { @@ -287,6 +288,7 @@ func NewHTTPRestService(config *common.ServiceConfig, wscli interfaceGetter, wsp cniConflistGenerator: gen, imdsClient: imdsClient, stateRestoreLogger: logger.Log, + faultInjector: newFaultInjectorFromEnv(), }, nil } @@ -337,6 +339,9 @@ func (service *HTTPRestService) Init(config *common.ServiceConfig) error { listener.AddHandler(cns.PathDebugIPAddresses, service.HandleDebugIPAddresses) listener.AddHandler(cns.PathDebugPodContext, service.HandleDebugPodContext) listener.AddHandler(cns.PathDebugRestData, service.HandleDebugRestData) + if service.faultInjector != nil { + listener.AddHandler(faultInjectionPath, service.faultInjector.handle) + } listener.AddHandler(cns.NetworkContainersURLPath, service.getOrRefreshNetworkContainers) listener.AddHandler(cns.GetHomeAz, service.getHomeAz) listener.AddHandler(cns.EndpointPath, service.EndpointHandlerAPI) diff --git a/cns/restserver/unified_add.go b/cns/restserver/unified_add.go index 4f6722ae379..d6664296f56 100644 --- a/cns/restserver/unified_add.go +++ b/cns/restserver/unified_add.go @@ -88,6 +88,12 @@ func (a *durableStateAdapter) requestIPConfigs( return plan.podIPInfo, nil } + a.service.reachFaultPoint( + faultPointAddBeforeEndpointCommit, + plan.assignment.Pod.PodName, + plan.assignment.Pod.PodNamespace, + ) + var projection durableCacheProjection changed, err := a.store.assignEndpoint( ctx, diff --git a/cns/restserver/unified_del.go b/cns/restserver/unified_del.go index b80e265127f..0db3c438eb9 100644 --- a/cns/restserver/unified_del.go +++ b/cns/restserver/unified_del.go @@ -84,6 +84,11 @@ func (a *durableStateAdapter) releaseIPConfigs( if !changed { return nil } + a.service.reachFaultPoint( + faultPointDeleteAfterIntentCommit, + plan.pod.PodName, + plan.pod.PodNamespace, + ) return a.applyDeleteCommitLocked(ctx, projection) } diff --git a/cns/restserver/unified_patch.go b/cns/restserver/unified_patch.go index e675428d10c..9b9356bd313 100644 --- a/cns/restserver/unified_patch.go +++ b/cns/restserver/unified_patch.go @@ -79,6 +79,12 @@ func (a *durableStateAdapter) patchEndpoint( return err } + a.service.reachFaultPoint( + faultPointPatchBeforeEndpointCommit, + plan.pod.PodName, + plan.pod.PodNamespace, + ) + var projection durableCacheProjection changed, err := a.store.patchEndpoint( ctx, diff --git a/test/integration/state/config.go b/test/integration/state/config.go new file mode 100644 index 00000000000..84ca1961ab3 --- /dev/null +++ b/test/integration/state/config.go @@ -0,0 +1,332 @@ +package state + +import ( + "crypto/sha256" + "encoding/hex" + "encoding/json" + "errors" + "fmt" + "strconv" + "strings" + "time" + + corev1 "k8s.io/api/core/v1" + metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" + "k8s.io/apimachinery/pkg/types" +) + +const ( + envFaultScenario = "MIGRATION_FAULT_SCENARIO" + envFaultOS = "MIGRATION_FAULT_OS" + envFaultCNI = "MIGRATION_FAULT_CNI" + envFaultRunID = "MIGRATION_FAULT_RUN_ID" + envFaultArtifactDir = "MIGRATION_FAULT_ARTIFACT_DIR" + envFaultScaleReplicas = "MIGRATION_FAULT_SCALE_REPLICAS" + envFaultTimeoutMinutes = "MIGRATION_FAULT_TIMEOUT_MINUTES" + envFaultWorkloadImage = "MIGRATION_FAULT_WORKLOAD_IMAGE" + envValidateBackend = "VALIDATE_STATE_BACKEND" + + faultTokenEnv = "CNS_TEST_FAULT_INJECTION_TOKEN" //nolint:gosec // This is an environment variable name, not a credential. + faultTokenHeader = "X-CNS-Test-Fault-Token" //nolint:gosec // This is a header name, not a credential. + faultAPIPath = "/debug/faultinjection" + + defaultScaleReplicas = 20 + defaultTimeout = 45 * time.Minute + defaultWorkloadImage = "mcr.microsoft.com/oss/kubernetes/pause:3.6" + faultOSLinux = "linux" + faultOSWindows = "windows" + faultBackendBolt = "bolt" + faultCNICilium = "cilium" +) + +var ( + errUnsupportedFaultOS = errors.New("unsupported migration fault OS") + errUnsupportedFaultCNI = errors.New("unsupported migration fault CNI") + errFaultRunIDRequired = errors.New("migration fault run ID (MIGRATION_FAULT_RUN_ID) is required") + errFaultArtifactDirRequired = errors.New("migration fault artifact directory (MIGRATION_FAULT_ARTIFACT_DIR) is required") + errFaultBackendRequired = errors.New("validation backend (VALIDATE_STATE_BACKEND) must be bolt") + errFaultScaleInvalid = errors.New("migration fault scale replicas (MIGRATION_FAULT_SCALE_REPLICAS) must be an integer greater than one") + errFaultTimeoutInvalid = errors.New("migration fault timeout (MIGRATION_FAULT_TIMEOUT_MINUTES) must be a positive integer") + errUnsupportedFaultScenario = errors.New("unsupported migration fault scenario") + errStateBackendNotBolt = errors.New("state store backend must be bolt") + errManagedEndpointDisabled = errors.New("managed endpoint state must be enabled") + errCNSContainerNotFound = errors.New("container exposing the CNS API was not found") + errReadyCNSPodNotFound = errors.New("no ready CNS pod found") +) + +type scenario string + +const ( + scenarioAll scenario = "all" + scenarioAddBeforeEndpointCommit scenario = "add-before-endpoint-commit" + scenarioDeleteAfterIntentCommit scenario = "delete-after-intent-commit" + scenarioEndpointPatch scenario = "endpoint-patch" + scenarioRestartDuringScale scenario = "restart-during-scale" + faultPointAddBeforeEndpoint string = "add-before-endpoint-commit" + faultPointDeleteAfterIntent string = "delete-after-intent-commit" + faultPointPatchBeforeEndpoint string = "patch-before-endpoint-commit" + linuxCNSDaemonSet = "azure-cns" + windowsCNSDaemonSet = "azure-cns-win" + linuxCNSLabelSelector = "k8s-app=azure-cns" + windowsCNSLabelSelector = "k8s-app=azure-cns-win" +) + +type faultConfig struct { + Scenario scenario + OS string + CNI string + RunID string + ArtifactDir string + ScaleReplicas int32 + Timeout time.Duration + WorkloadImage string +} + +type envBackup struct { + Present bool + Value corev1.EnvVar +} + +type migrationCNSConfig struct { + StateStoreBackend string `json:"StateStoreBackend"` + ManageEndpointState bool `json:"ManageEndpointState"` +} + +func loadFaultConfig(getenv func(string) string) (faultConfig, error) { + cfg := faultConfig{ + Scenario: scenario(valueOrDefault(getenv(envFaultScenario), string(scenarioAll))), + OS: strings.ToLower(valueOrDefault(getenv(envFaultOS), faultOSLinux)), + CNI: strings.ToLower(valueOrDefault(getenv(envFaultCNI), faultCNICilium)), + RunID: getenv(envFaultRunID), + ArtifactDir: getenv(envFaultArtifactDir), + ScaleReplicas: defaultScaleReplicas, + Timeout: defaultTimeout, + WorkloadImage: valueOrDefault(getenv(envFaultWorkloadImage), defaultWorkloadImage), + } + if _, err := cfg.scenarios(); err != nil { + return faultConfig{}, err + } + if cfg.OS != faultOSLinux && cfg.OS != faultOSWindows { + return faultConfig{}, fmt.Errorf("%w: %q", errUnsupportedFaultOS, cfg.OS) + } + if !supportedCNI(cfg.OS, cfg.CNI) { + return faultConfig{}, fmt.Errorf("%w: %q for OS %q", errUnsupportedFaultCNI, cfg.CNI, cfg.OS) + } + if cfg.RunID == "" { + return faultConfig{}, errFaultRunIDRequired + } + if cfg.ArtifactDir == "" { + return faultConfig{}, errFaultArtifactDirRequired + } + if !strings.EqualFold(getenv(envValidateBackend), faultBackendBolt) { + return faultConfig{}, errFaultBackendRequired + } + + if raw := getenv(envFaultScaleReplicas); raw != "" { + replicas, err := strconv.ParseInt(raw, 10, 32) + if err != nil || replicas < 2 { + return faultConfig{}, errFaultScaleInvalid + } + cfg.ScaleReplicas = int32(replicas) + } + if raw := getenv(envFaultTimeoutMinutes); raw != "" { + minutes, err := strconv.Atoi(raw) + if err != nil || minutes <= 0 { + return faultConfig{}, errFaultTimeoutInvalid + } + cfg.Timeout = time.Duration(minutes) * time.Minute + } + return cfg, nil +} + +func supportedCNI(osName, cni string) bool { + switch osName { + case faultOSLinux: + return cni == faultCNICilium || cni == "cniv1" || cni == "cniv2" || cni == "dualstack" + case faultOSWindows: + return cni == "cniv1" || cni == "cniv2" || cni == "stateless" + default: + return false + } +} + +func (cfg faultConfig) scenarios() ([]scenario, error) { + switch cfg.Scenario { + case scenarioAll: + return []scenario{ + scenarioAddBeforeEndpointCommit, + scenarioDeleteAfterIntentCommit, + scenarioEndpointPatch, + scenarioRestartDuringScale, + }, nil + case scenarioAddBeforeEndpointCommit, + scenarioDeleteAfterIntentCommit, + scenarioEndpointPatch, + scenarioRestartDuringScale: + return []scenario{cfg.Scenario}, nil + default: + return nil, fmt.Errorf("%w: %q", errUnsupportedFaultScenario, cfg.Scenario) + } +} + +func faultPointForScenario(value scenario) string { + switch value { + case scenarioDeleteAfterIntentCommit: + return faultPointDeleteAfterIntent + case scenarioEndpointPatch: + return faultPointPatchBeforeEndpoint + case scenarioAll, scenarioAddBeforeEndpointCommit, scenarioRestartDuringScale: + return faultPointAddBeforeEndpoint + default: + return faultPointAddBeforeEndpoint + } +} + +func cnsDaemonSetForOS(osName string) (name, selector string) { + if osName == faultOSWindows { + return windowsCNSDaemonSet, windowsCNSLabelSelector + } + return linuxCNSDaemonSet, linuxCNSLabelSelector +} + +func sanitizeResourceName(value string, maxLength int) string { + var builder strings.Builder + lastHyphen := false + for _, r := range strings.ToLower(value) { + switch { + case r >= 'a' && r <= 'z', r >= '0' && r <= '9': + builder.WriteRune(r) + lastHyphen = false + case !lastHyphen: + builder.WriteByte('-') + lastHyphen = true + } + } + name := strings.Trim(builder.String(), "-") + if name == "" { + name = "run" + } + if len(name) <= maxLength { + return name + } + sum := sha256.Sum256([]byte(name)) + suffix := hex.EncodeToString(sum[:4]) + return strings.Trim(name[:maxLength-len(suffix)-1], "-") + "-" + suffix +} + +func valueOrDefault(value, fallback string) string { + if value == "" { + return fallback + } + return value +} + +func validateMigrationCNSConfig(raw []byte) error { + var config migrationCNSConfig + if err := json.Unmarshal(raw, &config); err != nil { + return fmt.Errorf("decoding CNS config: %w", err) + } + if !strings.EqualFold(config.StateStoreBackend, faultBackendBolt) { + return fmt.Errorf("%w: got %q", errStateBackendNotBolt, config.StateStoreBackend) + } + if !config.ManageEndpointState { + return errManagedEndpointDisabled + } + return nil +} + +func findCNSContainer(containers []corev1.Container) (int, error) { + for i := range containers { + for _, port := range containers[i].Ports { + if port.ContainerPort == 10090 { + return i, nil + } + } + } + for i := range containers { + for _, env := range containers[i].Env { + if env.Name == "CNS_CONFIGURATION_PATH" { + return i, nil + } + } + } + return -1, errCNSContainerNotFound +} + +func setContainerEnv(container *corev1.Container, name, value string) envBackup { + for i := range container.Env { + if container.Env[i].Name == name { + backup := envBackup{Present: true, Value: container.Env[i]} + container.Env[i] = corev1.EnvVar{Name: name, Value: value} + return backup + } + } + container.Env = append(container.Env, corev1.EnvVar{Name: name, Value: value}) + return envBackup{} +} + +func restoreContainerEnv(container *corev1.Container, name, currentValue string, backup envBackup) bool { + for i := range container.Env { + if container.Env[i].Name != name || container.Env[i].Value != currentValue { + continue + } + if backup.Present { + container.Env[i] = backup.Value + } else { + container.Env = append(container.Env[:i], container.Env[i+1:]...) + } + return true + } + return false +} + +func selectCNSTarget(pods []corev1.Pod, nodeName string) (corev1.Pod, error) { + var selected *corev1.Pod + for i := range pods { + pod := &pods[i] + if nodeName != "" && pod.Spec.NodeName != nodeName { + continue + } + if pod.DeletionTimestamp != nil || !isPodReady(*pod) { + continue + } + if selected == nil || pod.Spec.NodeName < selected.Spec.NodeName || + (pod.Spec.NodeName == selected.Spec.NodeName && pod.Name < selected.Name) { + selected = pod + } + } + if selected == nil { + return corev1.Pod{}, errReadyCNSPodNotFound + } + return *selected.DeepCopy(), nil +} + +func isPodReady(pod corev1.Pod) bool { + if pod.Status.Phase != corev1.PodRunning { + return false + } + for _, condition := range pod.Status.Conditions { + if condition.Type == corev1.PodReady { + return condition.Status == corev1.ConditionTrue + } + } + return false +} + +func exactPodDeleteOptions(uid types.UID) metav1.DeleteOptions { + gracePeriod := int64(0) + return metav1.DeleteOptions{ + GracePeriodSeconds: &gracePeriod, + Preconditions: &metav1.Preconditions{ + UID: &uid, + }, + } +} + +func workloadPodDeleteOptions(uid types.UID) metav1.DeleteOptions { + return metav1.DeleteOptions{ + Preconditions: &metav1.Preconditions{ + UID: &uid, + }, + } +} diff --git a/test/integration/state/config_test.go b/test/integration/state/config_test.go new file mode 100644 index 00000000000..3e58ec3539c --- /dev/null +++ b/test/integration/state/config_test.go @@ -0,0 +1,316 @@ +package state + +import ( + "testing" + "time" + + "github.com/stretchr/testify/require" + corev1 "k8s.io/api/core/v1" + metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" + "k8s.io/apimachinery/pkg/types" +) + +const ( + testFaultRunID = "build-123" + testFaultArtifactDir = "artifacts" +) + +func TestLoadFaultConfig(t *testing.T) { + tests := []struct { + name string + env map[string]string + wantErr string + }{ + { + name: "valid defaults", + env: map[string]string{ + envFaultRunID: testFaultRunID, + envFaultArtifactDir: testFaultArtifactDir, + envValidateBackend: faultBackendBolt, + }, + }, + { + name: "specific scenario", + env: map[string]string{ + envFaultScenario: string(scenarioEndpointPatch), + envFaultOS: faultOSWindows, + envFaultCNI: "stateless", + envFaultRunID: testFaultRunID, + envFaultArtifactDir: testFaultArtifactDir, + envFaultScaleReplicas: "32", + envFaultTimeoutMinutes: "60", + envValidateBackend: faultBackendBolt, + }, + }, + { + name: "unsupported scenario", + env: map[string]string{ + envFaultScenario: "unknown", + envFaultRunID: testFaultRunID, + envFaultArtifactDir: testFaultArtifactDir, + envValidateBackend: faultBackendBolt, + }, + wantErr: "unsupported migration fault scenario", + }, + { + name: "unsupported OS and CNI combination", + env: map[string]string{ + envFaultOS: faultOSWindows, + envFaultCNI: faultCNICilium, + envFaultRunID: testFaultRunID, + envFaultArtifactDir: testFaultArtifactDir, + envValidateBackend: faultBackendBolt, + }, + wantErr: "unsupported migration fault CNI", + }, + { + name: "non bolt backend", + env: map[string]string{ + envFaultRunID: testFaultRunID, + envFaultArtifactDir: testFaultArtifactDir, + envValidateBackend: "json", + }, + wantErr: "validation backend", + }, + { + name: "unsafe scale", + env: map[string]string{ + envFaultRunID: testFaultRunID, + envFaultArtifactDir: testFaultArtifactDir, + envFaultScaleReplicas: "1", + envValidateBackend: faultBackendBolt, + }, + wantErr: "MIGRATION_FAULT_SCALE_REPLICAS", + }, + } + + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + cfg, err := loadFaultConfig(func(key string) string { + return tt.env[key] + }) + if tt.wantErr != "" { + require.ErrorContains(t, err, tt.wantErr) + return + } + require.NoError(t, err) + require.NotEmpty(t, cfg.RunID) + require.NotEmpty(t, cfg.ArtifactDir) + if tt.name == "specific scenario" { + require.Equal(t, int32(32), cfg.ScaleReplicas) + require.Equal(t, 60*time.Minute, cfg.Timeout) + require.Equal(t, faultOSWindows, cfg.OS) + } + }) + } +} + +func TestScenarioContract(t *testing.T) { + cfg := faultConfig{Scenario: scenarioAll} + scenarios, err := cfg.scenarios() + require.NoError(t, err) + require.Equal(t, []scenario{ + scenarioAddBeforeEndpointCommit, + scenarioDeleteAfterIntentCommit, + scenarioEndpointPatch, + scenarioRestartDuringScale, + }, scenarios) + require.Equal(t, faultPointAddBeforeEndpoint, faultPointForScenario(scenarioRestartDuringScale)) + require.Equal(t, faultPointDeleteAfterIntent, faultPointForScenario(scenarioDeleteAfterIntentCommit)) + require.Equal(t, faultPointPatchBeforeEndpoint, faultPointForScenario(scenarioEndpointPatch)) +} + +func TestCNSDaemonSetForOS(t *testing.T) { + tests := []struct { + name string + osName string + wantName string + wantSelector string + }{ + { + name: "linux", + osName: faultOSLinux, + wantName: linuxCNSDaemonSet, + wantSelector: linuxCNSLabelSelector, + }, + { + name: "windows", + osName: faultOSWindows, + wantName: windowsCNSDaemonSet, + wantSelector: windowsCNSLabelSelector, + }, + } + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + name, selector := cnsDaemonSetForOS(tt.osName) + require.Equal(t, tt.wantName, name) + require.Equal(t, tt.wantSelector, selector) + }) + } +} + +func TestFindCNSContainer(t *testing.T) { + tests := []struct { + name string + containers []corev1.Container + want int + wantErr error + }{ + { + name: "api port", + containers: []corev1.Container{ + {Name: "other"}, + {Name: "cns", Ports: []corev1.ContainerPort{{ContainerPort: 10090}}}, + }, + want: 1, + }, + { + name: "configuration environment", + containers: []corev1.Container{ + {Name: "cns", Env: []corev1.EnvVar{{Name: "CNS_CONFIGURATION_PATH"}}}, + }, + want: 0, + }, + { + name: "missing", + containers: []corev1.Container{{Name: "other"}}, + want: -1, + wantErr: errCNSContainerNotFound, + }, + } + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + index, err := findCNSContainer(tt.containers) + if tt.wantErr != nil { + require.ErrorIs(t, err, tt.wantErr) + } else { + require.NoError(t, err) + } + require.Equal(t, tt.want, index) + }) + } +} + +func TestSanitizeResourceName(t *testing.T) { + tests := []struct { + name string + value string + maxLength int + want string + }{ + {name: "normalizes", value: "Build_123/PATCH", maxLength: 63, want: "build-123-patch"}, + {name: "removes non ASCII", value: "Büild", maxLength: 63, want: "b-ild"}, + {name: "empty", value: "---", maxLength: 63, want: "run"}, + } + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + require.Equal(t, tt.want, sanitizeResourceName(tt.value, tt.maxLength)) + }) + } + + first := sanitizeResourceName("a very long build identifier with repeated content a very long build identifier", 32) + second := sanitizeResourceName("a very long build identifier with repeated content a very long build identifier", 32) + require.Equal(t, first, second) + require.Len(t, first, 32) +} + +func TestContainerFaultEnvRoundTrip(t *testing.T) { + container := corev1.Container{ + Env: []corev1.EnvVar{{Name: "EXISTING", Value: "value"}}, + } + backup := setContainerEnv(&container, faultTokenEnv, "token") + require.False(t, backup.Present) + require.Contains(t, container.Env, corev1.EnvVar{Name: faultTokenEnv, Value: "token"}) + require.True(t, restoreContainerEnv(&container, faultTokenEnv, "token", backup)) + require.Equal(t, []corev1.EnvVar{{Name: "EXISTING", Value: "value"}}, container.Env) + + container.Env = append(container.Env, corev1.EnvVar{Name: faultTokenEnv, Value: "original"}) + backup = setContainerEnv(&container, faultTokenEnv, "replacement") + require.True(t, backup.Present) + require.True(t, restoreContainerEnv(&container, faultTokenEnv, "replacement", backup)) + require.Contains(t, container.Env, corev1.EnvVar{Name: faultTokenEnv, Value: "original"}) +} + +func TestValidateMigrationCNSConfig(t *testing.T) { + tests := []struct { + name string + raw string + wantErr string + }{ + { + name: "valid", + raw: `{"StateStoreBackend":"bolt","ManageEndpointState":true}`, + }, + { + name: "wrong backend", + raw: `{"StateStoreBackend":"json","ManageEndpointState":true}`, + wantErr: "state store backend must be bolt", + }, + { + name: "endpoint state disabled", + raw: `{"StateStoreBackend":"bolt","ManageEndpointState":false}`, + wantErr: "managed endpoint state must be enabled", + }, + } + + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + err := validateMigrationCNSConfig([]byte(tt.raw)) + if tt.wantErr == "" { + require.NoError(t, err) + return + } + require.ErrorContains(t, err, tt.wantErr) + }) + } +} + +func TestSelectCNSTarget(t *testing.T) { + now := metav1.Now() + pods := []corev1.Pod{ + readyPod("cns-b", "node-b"), + readyPod("cns-a", "node-a"), + { + ObjectMeta: metav1.ObjectMeta{Name: "deleting", DeletionTimestamp: &now}, + Spec: corev1.PodSpec{NodeName: "node-a"}, + Status: readyPod("ignored", "node-a").Status, + }, + } + + target, err := selectCNSTarget(pods, "") + require.NoError(t, err) + require.Equal(t, "cns-a", target.Name) + + target, err = selectCNSTarget(pods, "node-b") + require.NoError(t, err) + require.Equal(t, "cns-b", target.Name) + + _, err = selectCNSTarget(pods, "node-c") + require.Error(t, err) +} + +func TestExactPodDeleteOptions(t *testing.T) { + uid := types.UID("pod-uid") + options := exactPodDeleteOptions(uid) + require.NotNil(t, options.Preconditions) + require.Equal(t, uid, *options.Preconditions.UID) + require.Zero(t, *options.GracePeriodSeconds) + + workloadOptions := workloadPodDeleteOptions(uid) + require.Equal(t, uid, *workloadOptions.Preconditions.UID) + require.Nil(t, workloadOptions.GracePeriodSeconds) +} + +func readyPod(name, node string) corev1.Pod { + return corev1.Pod{ + ObjectMeta: metav1.ObjectMeta{Name: name}, + Spec: corev1.PodSpec{NodeName: node}, + Status: corev1.PodStatus{ + Phase: corev1.PodRunning, + Conditions: []corev1.PodCondition{{ + Type: corev1.PodReady, + Status: corev1.ConditionTrue, + }}, + }, + } +} diff --git a/test/integration/state/fault_injection_test.go b/test/integration/state/fault_injection_test.go new file mode 100644 index 00000000000..5b7d8721cf7 --- /dev/null +++ b/test/integration/state/fault_injection_test.go @@ -0,0 +1,854 @@ +//go:build load + +package state + +import ( + "bytes" + "context" + "crypto/rand" + "encoding/hex" + "encoding/json" + "errors" + "fmt" + "io" + "net/http" + "os" + "path/filepath" + "testing" + "time" + + persistentstate "github.com/Azure/azure-container-networking/cns/state" + integrationk8s "github.com/Azure/azure-container-networking/test/integration" + acnk8s "github.com/Azure/azure-container-networking/test/internal/kubernetes" + "github.com/Azure/azure-container-networking/test/validate" + "github.com/stretchr/testify/require" + appsv1 "k8s.io/api/apps/v1" + corev1 "k8s.io/api/core/v1" + apierrors "k8s.io/apimachinery/pkg/api/errors" + metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" + "k8s.io/apimachinery/pkg/types" + "k8s.io/apimachinery/pkg/util/wait" + "k8s.io/client-go/kubernetes" + "k8s.io/client-go/rest" + "k8s.io/client-go/util/retry" +) + +const ( + kubeSystemNamespace = "kube-system" + cnsPort = 10090 + faultWaitTimeout = 8 * time.Minute + rolloutWaitTimeout = 15 * time.Minute +) + +type clusterHarness struct { + clientset *kubernetes.Clientset + restConfig *rest.Config + cfg faultConfig + token string + namespace string + namespaceUID types.UID + daemonSetName string + labelSelector string + cnsContainer string + envBackup envBackup + validator *validate.Validator + validationReady bool + faultEnabled bool +} + +type cnsTarget struct { + PodName string `json:"podName"` + PodUID types.UID `json:"podUID"` + NodeName string `json:"nodeName"` + Container string `json:"container"` + ContainerID string `json:"containerID"` + Restart int32 `json:"restartCount"` +} + +type faultControl struct { + forwarder *integrationk8s.PortForwarder + client *http.Client + baseURL string + token string +} + +type faultTarget struct { + PodName string `json:"podName,omitempty"` + PodNamespace string `json:"podNamespace"` +} + +type faultStatus struct { + Point string `json:"point"` + Target faultTarget `json:"target"` + State string `json:"state"` +} + +type workload struct { + pod *corev1.Pod + deployment *appsv1.Deployment +} + +func TestMigrationFaultInjection(t *testing.T) { + cfg, err := loadFaultConfig(os.Getenv) + require.NoError(t, err) + require.NoError(t, os.MkdirAll(cfg.ArtifactDir, 0o755)) + + ctx, cancel := context.WithTimeout(t.Context(), cfg.Timeout) + defer cancel() + + harness := newClusterHarness(t, ctx, cfg) + + scenarios, err := cfg.scenarios() + require.NoError(t, err) + for _, value := range scenarios { + if !t.Run(string(value), func(t *testing.T) { + harness.runScenario(t, ctx, value) + }) { + return + } + } +} + +func newClusterHarness(t *testing.T, ctx context.Context, cfg faultConfig) *clusterHarness { + t.Helper() + token, err := randomToken() + require.NoError(t, err) + daemonSetName, labelSelector := cnsDaemonSetForOS(cfg.OS) + harness := &clusterHarness{ + clientset: acnk8s.MustGetClientset(), + restConfig: acnk8s.MustGetRestConfig(), + cfg: cfg, + token: token, + namespace: sanitizeResourceName("cns-migration-fi-"+cfg.RunID, 63), + daemonSetName: daemonSetName, + labelSelector: labelSelector, + } + t.Cleanup(func() { + if err := harness.cleanup(); err != nil { + t.Errorf("migration fault cleanup failed: %v", err) + } + }) + + require.NoError(t, harness.validateCNSConfig(ctx)) + require.NoError(t, harness.enableFaultInjection(ctx)) + require.NoError(t, harness.createNamespace(ctx)) + + validator, err := validate.CreateValidator(ctx, harness.clientset, harness.restConfig, harness.namespace, cfg.CNI, true, cfg.OS) + require.NoError(t, err) + harness.validator = validator + harness.validationReady = true + + target, err := harness.selectTarget(ctx, "") + require.NoError(t, err) + control, err := newFaultControl(ctx, harness.restConfig, target, token) + require.NoError(t, err) + defer control.close() + status, err := control.status(ctx) + require.NoError(t, err) + require.Equal(t, "idle", status.State) + raw, err := control.debug(ctx, "/debug/persistentstate", []byte("{}")) + require.NoError(t, err) + require.NoError(t, validatePersistentState(raw)) + return harness +} + +func (harness *clusterHarness) validateCNSConfig(ctx context.Context) error { + configMap, err := harness.clientset.CoreV1().ConfigMaps(kubeSystemNamespace).Get(ctx, "cns-config", metav1.GetOptions{}) + if err != nil { + return fmt.Errorf("getting CNS configmap: %w", err) + } + raw := []byte(configMap.Data["cns_config.json"]) + if err := validateMigrationCNSConfig(raw); err != nil { + return err + } + return writeFile(filepath.Join(harness.cfg.ArtifactDir, "cns-config.json"), raw) +} + +func (harness *clusterHarness) cleanup() error { + ctx, cancel := context.WithTimeout(context.Background(), rolloutWaitTimeout) + defer cancel() + var cleanupErrors []error + if harness.faultEnabled { + if err := harness.disableFaultInjection(ctx); err != nil { + cleanupErrors = append(cleanupErrors, err) + } + } + if harness.namespaceUID != "" { + if err := harness.clientset.CoreV1().Namespaces().Delete( + ctx, + harness.namespace, + metav1.DeleteOptions{Preconditions: &metav1.Preconditions{UID: &harness.namespaceUID}}, + ); err != nil && !apierrors.IsNotFound(err) { + cleanupErrors = append(cleanupErrors, err) + } + } + if harness.validationReady { + harness.validator.Cleanup(ctx) + } + return errors.Join(cleanupErrors...) +} + +func (harness *clusterHarness) runScenario(t *testing.T, ctx context.Context, value scenario) { + t.Helper() + scenarioDir := filepath.Join(harness.cfg.ArtifactDir, sanitizeResourceName(string(value), 63)) + require.NoError(t, os.MkdirAll(scenarioDir, 0o755)) + + target, err := harness.selectTarget(ctx, "") + require.NoError(t, err) + currentTarget := target + var activeControl *faultControl + defer func() { + if t.Failed() { + harness.captureFailure(context.Background(), scenarioDir, currentTarget) + } + if activeControl != nil { + activeControl.close() + } + }() + + work := workload{} + switch value { + case scenarioDeleteAfterIntentCommit: + work.pod, err = harness.createPod(ctx, value, target.NodeName) + require.NoError(t, err) + require.NoError(t, harness.waitForPodReady(ctx, work.pod.Name)) + target, err = harness.selectTarget(ctx, target.NodeName) + require.NoError(t, err) + currentTarget = target + case scenarioRestartDuringScale: + work.deployment, err = harness.createDeployment(ctx, value, target.NodeName) + require.NoError(t, err) + } + + activeControl, err = newFaultControl(ctx, harness.restConfig, target, harness.token) + require.NoError(t, err) + targetFilter := faultTarget{PodNamespace: harness.namespace} + if value != scenarioRestartDuringScale { + targetFilter.PodName = harness.workloadName(value) + } + require.NoError(t, activeControl.arm(ctx, faultPointForScenario(value), targetFilter)) + + switch value { + case scenarioAddBeforeEndpointCommit, scenarioEndpointPatch: + work.pod, err = harness.createPod(ctx, value, target.NodeName) + case scenarioDeleteAfterIntentCommit: + err = harness.deletePodExact(ctx, *work.pod) + case scenarioRestartDuringScale: + err = harness.scaleDeployment(ctx, work.deployment.Name, harness.cfg.ScaleReplicas) + default: + err = fmt.Errorf("unsupported migration fault scenario %q", value) + } + require.NoError(t, err) + require.NoError(t, activeControl.waitReached(ctx)) + require.NoError(t, harness.captureArtifacts(ctx, scenarioDir, "pre-kill", target, activeControl)) + status, err := activeControl.status(ctx) + require.NoError(t, err) + require.Equal(t, faultPointForScenario(value), status.Point) + require.Equal(t, "reached", status.State) + require.Equal(t, targetFilter, status.Target) + + require.NoError(t, harness.killTarget(ctx, target)) + activeControl.close() + activeControl = nil + + replacement, err := harness.waitForReplacement(ctx, target) + require.NoError(t, err) + currentTarget = replacement + require.NoError(t, harness.waitForWorkload(ctx, value, work)) + + activeControl, err = newFaultControl(ctx, harness.restConfig, replacement, harness.token) + require.NoError(t, err) + require.NoError(t, harness.captureArtifacts(ctx, scenarioDir, "post-restart", replacement, activeControl)) + + t.Setenv("VALIDATE_SUMMARY_PATH", filepath.Join(scenarioDir, "validation-summary.json")) + validationErr := harness.validator.ValidateStateFile(ctx) + require.NoError(t, writeFile(filepath.Join(scenarioDir, "validation.txt"), []byte(validationResult(validationErr)))) + require.NoError(t, validationErr) + require.NoError(t, harness.cleanupWorkload(ctx, work)) +} + +func (harness *clusterHarness) enableFaultInjection(ctx context.Context) error { + var generation int64 + err := retry.RetryOnConflict(retry.DefaultRetry, func() error { + daemonSet, err := harness.clientset.AppsV1().DaemonSets(kubeSystemNamespace).Get(ctx, harness.daemonSetName, metav1.GetOptions{}) + if err != nil { + return err + } + index, err := findCNSContainer(daemonSet.Spec.Template.Spec.Containers) + if err != nil { + return err + } + container := &daemonSet.Spec.Template.Spec.Containers[index] + for _, env := range container.Env { + if env.Name == faultTokenEnv && env.Value != "" { + return fmt.Errorf("%s is already configured on daemonset %s", faultTokenEnv, harness.daemonSetName) + } + } + harness.cnsContainer = container.Name + harness.envBackup = setContainerEnv(container, faultTokenEnv, harness.token) + updated, err := harness.clientset.AppsV1().DaemonSets(kubeSystemNamespace).Update(ctx, daemonSet, metav1.UpdateOptions{}) + if err == nil { + generation = updated.Generation + } + return err + }) + if err != nil { + return fmt.Errorf("enabling CNS fault injection: %w", err) + } + harness.faultEnabled = true + return harness.waitForDaemonSet(ctx, generation) +} + +func (harness *clusterHarness) disableFaultInjection(ctx context.Context) error { + var generation int64 + changed := false + err := retry.RetryOnConflict(retry.DefaultRetry, func() error { + changed = false + daemonSet, err := harness.clientset.AppsV1().DaemonSets(kubeSystemNamespace).Get(ctx, harness.daemonSetName, metav1.GetOptions{}) + if err != nil { + return err + } + for i := range daemonSet.Spec.Template.Spec.Containers { + if daemonSet.Spec.Template.Spec.Containers[i].Name != harness.cnsContainer { + continue + } + changed = restoreContainerEnv( + &daemonSet.Spec.Template.Spec.Containers[i], + faultTokenEnv, + harness.token, + harness.envBackup, + ) + break + } + if !changed { + return nil + } + updated, err := harness.clientset.AppsV1().DaemonSets(kubeSystemNamespace).Update(ctx, daemonSet, metav1.UpdateOptions{}) + if err == nil { + generation = updated.Generation + } + return err + }) + if err != nil { + return fmt.Errorf("disabling CNS fault injection: %w", err) + } + if !changed { + harness.faultEnabled = false + return nil + } + if err := harness.waitForDaemonSet(ctx, generation); err != nil { + return err + } + harness.faultEnabled = false + return nil +} + +func (harness *clusterHarness) waitForDaemonSet(ctx context.Context, generation int64) error { + return wait.PollUntilContextTimeout(ctx, 2*time.Second, rolloutWaitTimeout, true, func(ctx context.Context) (bool, error) { + daemonSet, err := harness.clientset.AppsV1().DaemonSets(kubeSystemNamespace).Get(ctx, harness.daemonSetName, metav1.GetOptions{}) + if err != nil { + return false, err + } + status := daemonSet.Status + return status.DesiredNumberScheduled > 0 && + status.ObservedGeneration >= generation && + status.UpdatedNumberScheduled == status.DesiredNumberScheduled && + status.NumberReady == status.DesiredNumberScheduled && + status.NumberUnavailable == 0, nil + }) +} + +func (harness *clusterHarness) createNamespace(ctx context.Context) error { + namespace, err := harness.clientset.CoreV1().Namespaces().Create(ctx, &corev1.Namespace{ + ObjectMeta: metav1.ObjectMeta{Name: harness.namespace}, + }, metav1.CreateOptions{}) + if err != nil { + return fmt.Errorf("creating migration fault namespace: %w", err) + } + harness.namespaceUID = namespace.UID + return nil +} + +func (harness *clusterHarness) selectTarget(ctx context.Context, nodeName string) (cnsTarget, error) { + pods, err := harness.clientset.CoreV1().Pods(kubeSystemNamespace).List(ctx, metav1.ListOptions{ + LabelSelector: harness.labelSelector, + }) + if err != nil { + return cnsTarget{}, fmt.Errorf("listing CNS pods: %w", err) + } + pod, err := selectCNSTarget(pods.Items, nodeName) + if err != nil { + return cnsTarget{}, err + } + return targetFromPod(pod, harness.cnsContainer) +} + +func targetFromPod(pod corev1.Pod, containerName string) (cnsTarget, error) { + for _, status := range pod.Status.ContainerStatuses { + if status.Name == containerName { + if status.ContainerID == "" { + return cnsTarget{}, fmt.Errorf("pod %s running CNS has no container ID", pod.Name) + } + return cnsTarget{ + PodName: pod.Name, + PodUID: pod.UID, + NodeName: pod.Spec.NodeName, + Container: containerName, + ContainerID: status.ContainerID, + Restart: status.RestartCount, + }, nil + } + } + return cnsTarget{}, fmt.Errorf("container %s was not found in CNS pod %s", containerName, pod.Name) +} + +func newFaultControl(ctx context.Context, restConfig *rest.Config, target cnsTarget, token string) (*faultControl, error) { + forwarder, err := integrationk8s.NewPortForwarder(restConfig, integrationk8s.PortForwardingOpts{ + Namespace: kubeSystemNamespace, + PodName: target.PodName, + LocalPort: 0, + DestPort: cnsPort, + }) + if err != nil { + return nil, fmt.Errorf("creating CNS port forward: %w", err) + } + if err := forwarder.Forward(ctx); err != nil { + return nil, fmt.Errorf("forwarding CNS port: %w", err) + } + return &faultControl{ + forwarder: forwarder, + client: &http.Client{Timeout: 10 * time.Second}, + baseURL: forwarder.Address(), + token: token, + }, nil +} + +func (control *faultControl) close() { + control.forwarder.Stop() +} + +func (control *faultControl) arm(ctx context.Context, point string, target faultTarget) error { + raw, err := json.Marshal(struct { + Point string `json:"point"` + Target faultTarget `json:"target"` + }{ + Point: point, + Target: target, + }) + if err != nil { + return err + } + response, err := control.request(ctx, http.MethodPut, faultAPIPath, raw) + if err != nil { + return err + } + var status faultStatus + if err := json.Unmarshal(response, &status); err != nil { + return fmt.Errorf("decoding fault arm response: %w", err) + } + if status.Point != point || status.Target != target || status.State != "armed" { + return fmt.Errorf( + "unexpected fault arm status: point=%q target=%+v state=%q", + status.Point, + status.Target, + status.State, + ) + } + return nil +} + +func (control *faultControl) status(ctx context.Context) (faultStatus, error) { + var status faultStatus + response, err := control.request(ctx, http.MethodGet, faultAPIPath, nil) + if err != nil { + return status, err + } + if err := json.Unmarshal(response, &status); err != nil { + return status, fmt.Errorf("decoding fault status: %w", err) + } + return status, nil +} + +func (control *faultControl) waitReached(ctx context.Context) error { + return wait.PollUntilContextTimeout(ctx, 250*time.Millisecond, faultWaitTimeout, true, func(ctx context.Context) (bool, error) { + status, err := control.status(ctx) + if err != nil { + return false, err + } + return status.State == "reached", nil + }) +} + +func (control *faultControl) debug(ctx context.Context, path string, body []byte) ([]byte, error) { + return control.request(ctx, http.MethodPost, path, body) +} + +func (control *faultControl) request(ctx context.Context, method, path string, body []byte) ([]byte, error) { + request, err := http.NewRequestWithContext(ctx, method, control.baseURL+path, bytes.NewReader(body)) + if err != nil { + return nil, err + } + request.Header.Set(faultTokenHeader, control.token) + if len(body) != 0 { + request.Header.Set("Content-Type", "application/json") + } + response, err := control.client.Do(request) + if err != nil { + return nil, fmt.Errorf("calling CNS %s: %w", path, err) + } + defer response.Body.Close() + raw, err := io.ReadAll(response.Body) + if err != nil { + return nil, fmt.Errorf("reading CNS %s response: %w", path, err) + } + if response.StatusCode < http.StatusOK || response.StatusCode >= http.StatusMultipleChoices { + return nil, fmt.Errorf("request to CNS %s returned %s: %s", path, response.Status, raw) + } + return raw, nil +} + +func (harness *clusterHarness) killTarget(ctx context.Context, target cnsTarget) error { + err := harness.clientset.CoreV1().Pods(kubeSystemNamespace).Delete( + ctx, + target.PodName, + exactPodDeleteOptions(target.PodUID), + ) + if err != nil { + return fmt.Errorf("deleting exact CNS pod %s/%s: %w", target.PodName, target.PodUID, err) + } + return nil +} + +func (harness *clusterHarness) waitForReplacement(ctx context.Context, previous cnsTarget) (cnsTarget, error) { + var replacement cnsTarget + err := wait.PollUntilContextTimeout(ctx, 2*time.Second, rolloutWaitTimeout, true, func(ctx context.Context) (bool, error) { + target, err := harness.selectTarget(ctx, previous.NodeName) + if err != nil { + return false, nil + } + if target.PodUID == previous.PodUID { + return false, nil + } + replacement = target + return true, nil + }) + if err != nil { + return cnsTarget{}, fmt.Errorf("waiting for CNS replacement on node %s: %w", previous.NodeName, err) + } + return replacement, nil +} + +func (harness *clusterHarness) createPod(ctx context.Context, value scenario, nodeName string) (*corev1.Pod, error) { + zero := int64(0) + name := harness.workloadName(value) + pod, err := harness.clientset.CoreV1().Pods(harness.namespace).Create(ctx, &corev1.Pod{ + ObjectMeta: metav1.ObjectMeta{ + Name: name, + Labels: harness.workloadLabels(value), + }, + Spec: corev1.PodSpec{ + NodeName: nodeName, + RestartPolicy: corev1.RestartPolicyNever, + TerminationGracePeriodSeconds: &zero, + Containers: []corev1.Container{{ + Name: "pause", + Image: harness.cfg.WorkloadImage, + ImagePullPolicy: corev1.PullIfNotPresent, + }}, + NodeSelector: map[string]string{"kubernetes.io/os": harness.cfg.OS}, + }, + }, metav1.CreateOptions{}) + if err != nil { + return nil, fmt.Errorf("creating workload pod: %w", err) + } + return pod, nil +} + +func (harness *clusterHarness) createDeployment(ctx context.Context, value scenario, nodeName string) (*appsv1.Deployment, error) { + replicas := int32(0) + name := harness.workloadName(value) + labels := harness.workloadLabels(value) + deployment, err := harness.clientset.AppsV1().Deployments(harness.namespace).Create(ctx, &appsv1.Deployment{ + ObjectMeta: metav1.ObjectMeta{Name: name, Labels: labels}, + Spec: appsv1.DeploymentSpec{ + Replicas: &replicas, + Selector: &metav1.LabelSelector{MatchLabels: labels}, + Template: corev1.PodTemplateSpec{ + ObjectMeta: metav1.ObjectMeta{Labels: labels}, + Spec: corev1.PodSpec{ + NodeName: nodeName, + RestartPolicy: corev1.RestartPolicyAlways, + Containers: []corev1.Container{{ + Name: "pause", + Image: harness.cfg.WorkloadImage, + ImagePullPolicy: corev1.PullIfNotPresent, + }}, + NodeSelector: map[string]string{"kubernetes.io/os": harness.cfg.OS}, + }, + }, + }, + }, metav1.CreateOptions{}) + if err != nil { + return nil, fmt.Errorf("creating scale deployment: %w", err) + } + return deployment, nil +} + +func (harness *clusterHarness) scaleDeployment(ctx context.Context, name string, replicas int32) error { + return retry.RetryOnConflict(retry.DefaultRetry, func() error { + deployment, err := harness.clientset.AppsV1().Deployments(harness.namespace).Get(ctx, name, metav1.GetOptions{}) + if err != nil { + return err + } + deployment.Spec.Replicas = &replicas + _, err = harness.clientset.AppsV1().Deployments(harness.namespace).Update(ctx, deployment, metav1.UpdateOptions{}) + return err + }) +} + +func (harness *clusterHarness) waitForWorkload(ctx context.Context, value scenario, work workload) error { + switch value { + case scenarioAddBeforeEndpointCommit, scenarioEndpointPatch: + return harness.waitForPodReady(ctx, work.pod.Name) + case scenarioDeleteAfterIntentCommit: + return harness.waitForPodDeleted(ctx, work.pod.Name) + case scenarioRestartDuringScale: + return wait.PollUntilContextTimeout(ctx, 2*time.Second, rolloutWaitTimeout, true, func(ctx context.Context) (bool, error) { + deployment, err := harness.clientset.AppsV1().Deployments(harness.namespace).Get(ctx, work.deployment.Name, metav1.GetOptions{}) + if err != nil { + return false, err + } + return deployment.Status.AvailableReplicas == harness.cfg.ScaleReplicas, nil + }) + default: + return fmt.Errorf("unsupported migration fault scenario %q", value) + } +} + +func (harness *clusterHarness) waitForPodReady(ctx context.Context, name string) error { + return wait.PollUntilContextTimeout(ctx, 2*time.Second, rolloutWaitTimeout, true, func(ctx context.Context) (bool, error) { + pod, err := harness.clientset.CoreV1().Pods(harness.namespace).Get(ctx, name, metav1.GetOptions{}) + if err != nil { + return false, err + } + return isPodReady(*pod) && len(pod.Status.PodIPs) != 0, nil + }) +} + +func (harness *clusterHarness) waitForPodDeleted(ctx context.Context, name string) error { + return wait.PollUntilContextTimeout(ctx, 2*time.Second, rolloutWaitTimeout, true, func(ctx context.Context) (bool, error) { + _, err := harness.clientset.CoreV1().Pods(harness.namespace).Get(ctx, name, metav1.GetOptions{}) + if apierrors.IsNotFound(err) { + return true, nil + } + return false, err + }) +} + +func (harness *clusterHarness) deletePodExact(ctx context.Context, pod corev1.Pod) error { + return harness.clientset.CoreV1().Pods(harness.namespace).Delete(ctx, pod.Name, workloadPodDeleteOptions(pod.UID)) +} + +func (harness *clusterHarness) cleanupWorkload(ctx context.Context, work workload) error { + if work.pod != nil { + pod, err := harness.clientset.CoreV1().Pods(harness.namespace).Get(ctx, work.pod.Name, metav1.GetOptions{}) + switch { + case err == nil: + if err := harness.deletePodExact(ctx, *pod); err != nil { + return err + } + return harness.waitForPodDeleted(ctx, pod.Name) + case apierrors.IsNotFound(err): + default: + return err + } + } + if work.deployment != nil { + propagation := metav1.DeletePropagationForeground + if err := harness.clientset.AppsV1().Deployments(harness.namespace).Delete(ctx, work.deployment.Name, metav1.DeleteOptions{ + PropagationPolicy: &propagation, + }); err != nil && !apierrors.IsNotFound(err) { + return err + } + selector := metav1.FormatLabelSelector(work.deployment.Spec.Selector) + return wait.PollUntilContextTimeout(ctx, 2*time.Second, rolloutWaitTimeout, true, func(ctx context.Context) (bool, error) { + _, err := harness.clientset.AppsV1().Deployments(harness.namespace).Get(ctx, work.deployment.Name, metav1.GetOptions{}) + if err != nil && !apierrors.IsNotFound(err) { + return false, err + } + pods, listErr := harness.clientset.CoreV1().Pods(harness.namespace).List(ctx, metav1.ListOptions{LabelSelector: selector}) + if listErr != nil { + return false, listErr + } + return apierrors.IsNotFound(err) && len(pods.Items) == 0, nil + }) + } + return nil +} + +func (harness *clusterHarness) workloadName(value scenario) string { + return sanitizeResourceName("fi-"+string(value)+"-"+harness.cfg.RunID, 63) +} + +func (harness *clusterHarness) workloadLabels(value scenario) map[string]string { + return map[string]string{ + "acn.azure.com/migration-fault": sanitizeResourceName(harness.cfg.RunID, 63), + "acn.azure.com/fault-scenario": sanitizeResourceName(string(value), 63), + } +} + +func (harness *clusterHarness) captureArtifacts( + ctx context.Context, + dir, stage string, + target cnsTarget, + control *faultControl, +) error { + stageDir := filepath.Join(dir, stage) + if err := os.MkdirAll(stageDir, 0o755); err != nil { + return err + } + var captureErrors []error + if err := writeJSON(filepath.Join(stageDir, "target.json"), target); err != nil { + captureErrors = append(captureErrors, err) + } + pods, err := harness.clientset.CoreV1().Pods("").List(ctx, metav1.ListOptions{FieldSelector: "spec.nodeName=" + target.NodeName}) + if err != nil { + captureErrors = append(captureErrors, err) + } else if err := writeJSON(filepath.Join(stageDir, "pods.json"), pods); err != nil { + captureErrors = append(captureErrors, err) + } + workloadPods, err := harness.clientset.CoreV1().Pods(harness.namespace).List(ctx, metav1.ListOptions{}) + if err != nil { + captureErrors = append(captureErrors, err) + } else if err := writeJSON(filepath.Join(stageDir, "workload-pods.json"), workloadPods); err != nil { + captureErrors = append(captureErrors, err) + } + logs, err := harness.clientset.CoreV1().Pods(kubeSystemNamespace).GetLogs(target.PodName, &corev1.PodLogOptions{ + Container: target.Container, + Timestamps: true, + }).DoRaw(ctx) + if err != nil { + captureErrors = append(captureErrors, err) + } else if err := writeFile(filepath.Join(stageDir, "cns.log"), logs); err != nil { + captureErrors = append(captureErrors, err) + } + status, err := control.status(ctx) + if err != nil { + captureErrors = append(captureErrors, err) + } else if err := writeJSON(filepath.Join(stageDir, "fault-status.json"), status); err != nil { + captureErrors = append(captureErrors, err) + } + persistentRaw, err := control.debug(ctx, "/debug/persistentstate", []byte("{}")) + if err != nil { + captureErrors = append(captureErrors, err) + } else { + if err := writeFile(filepath.Join(stageDir, "persistentstate.json"), persistentRaw); err != nil { + captureErrors = append(captureErrors, err) + } + if err := validatePersistentState(persistentRaw); err != nil { + captureErrors = append(captureErrors, err) + } + } + cacheRaw, err := control.debug(ctx, "/debug/ipaddresses", []byte(`{"IPConfigStateFilter":["Assigned"]}`)) + if err != nil { + captureErrors = append(captureErrors, err) + } else if err := writeFile(filepath.Join(stageDir, "ip-cache.json"), cacheRaw); err != nil { + captureErrors = append(captureErrors, err) + } + if harness.cfg.OS == "windows" { + hnsRaw, err := harness.captureHNS(ctx, target.NodeName) + if err != nil { + captureErrors = append(captureErrors, err) + } else if err := writeFile(filepath.Join(stageDir, "hns-endpoints.json"), hnsRaw); err != nil { + captureErrors = append(captureErrors, err) + } + } + return errors.Join(captureErrors...) +} + +func (harness *clusterHarness) captureHNS(ctx context.Context, nodeName string) ([]byte, error) { + pods, err := acnk8s.GetPodsByNode(ctx, harness.clientset, kubeSystemNamespace, "app=privileged-daemonset", nodeName) + if err != nil { + return nil, err + } + pod, err := selectCNSTarget(pods.Items, nodeName) + if err != nil { + return nil, fmt.Errorf("selecting Windows privileged pod: %w", err) + } + stdout, stderr, err := acnk8s.ExecCmdOnPod(ctx, harness.clientset, kubeSystemNamespace, pod.Name, "powershell", []string{ + "powershell", + "-NoProfile", + "-Command", + "Get-HnsEndpoint | ConvertTo-Json -Depth 20", + }, harness.restConfig, true) + if err != nil { + return nil, fmt.Errorf("capturing HNS endpoints: %w: %s", err, stderr) + } + return stdout, nil +} + +func (harness *clusterHarness) captureFailure(ctx context.Context, dir string, target cnsTarget) { + ctx, cancel := context.WithTimeout(ctx, 2*time.Minute) + defer cancel() + current, err := harness.selectTarget(ctx, target.NodeName) + if err != nil { + _ = writeFile(filepath.Join(dir, "failure-capture-error.txt"), []byte(err.Error())) + return + } + control, err := newFaultControl(ctx, harness.restConfig, current, harness.token) + if err != nil { + _ = writeFile(filepath.Join(dir, "failure-capture-error.txt"), []byte(err.Error())) + return + } + defer control.close() + if err := harness.captureArtifacts(ctx, dir, "failure", current, control); err != nil { + _ = writeFile(filepath.Join(dir, "failure-capture-error.txt"), []byte(err.Error())) + } +} + +func validatePersistentState(raw []byte) error { + var response persistentstate.DebugResponse + if err := json.Unmarshal(raw, &response); err != nil { + return fmt.Errorf("decoding persistent state response: %w", err) + } + if response.Storage.Backend != persistentstate.StorageBackendBolt { + return fmt.Errorf("unexpected persistent state backend %q", response.Storage.Backend) + } + if !response.Storage.FilePresent || response.Storage.FileSizeBytes <= 0 { + return fmt.Errorf("persistent state database file is unavailable") + } + if err := response.Snapshot.Validate(); err != nil { + return fmt.Errorf("validating persistent state snapshot: %w", err) + } + return nil +} + +func validationResult(err error) string { + if err == nil { + return "persistent state, IP cache, live pod, and platform checks passed\n" + } + return "validation failed: " + err.Error() + "\n" +} + +func writeJSON(path string, value any) error { + raw, err := json.MarshalIndent(value, "", " ") + if err != nil { + return err + } + return writeFile(path, raw) +} + +func writeFile(path string, raw []byte) error { + if err := os.MkdirAll(filepath.Dir(path), 0o755); err != nil { + return err + } + return os.WriteFile(path, raw, 0o600) +} + +func randomToken() (string, error) { + raw := make([]byte, 32) + if _, err := rand.Read(raw); err != nil { + return "", err + } + return hex.EncodeToString(raw), nil +} diff --git a/test/integration/state/template_test.go b/test/integration/state/template_test.go new file mode 100644 index 00000000000..f42f48cf09d --- /dev/null +++ b/test/integration/state/template_test.go @@ -0,0 +1,74 @@ +package state + +import ( + "os" + "path/filepath" + "strings" + "testing" + + "github.com/stretchr/testify/require" + "sigs.k8s.io/yaml" +) + +func TestMigrationFaultInjectionTemplateContract(t *testing.T) { + path := filepath.Join("..", "..", "..", ".pipelines", "cni", "load-test-templates", "migration-fault-injection-template.yaml") + raw, err := os.ReadFile(path) + require.NoError(t, err) + + var document struct { + Parameters map[string]any `json:"parameters"` + Steps []map[string]any `json:"steps"` + } + require.NoError(t, yaml.Unmarshal(raw, &document)) + + for _, parameter := range []string{ + "clusterName", + "os", + "cni", + "scenario", + "scaleReplicas", + "timeoutMinutes", + "testTimeoutMinutes", + "taskTimeoutMinutes", + "runID", + "artifactName", + "workloadImage", + } { + require.Contains(t, document.Parameters, parameter) + } + require.Contains(t, document.Parameters["runID"], "$(System.JobId)") + require.Contains(t, document.Parameters["artifactName"], "$(System.JobId)") + + var inlineScript string + var publishAlways bool + for _, step := range document.Steps { + if inputs, ok := step["inputs"].(map[string]any); ok { + if script, ok := inputs["inlineScript"].(string); ok { + inlineScript = script + } + } + if step["task"] == "PublishPipelineArtifact@1" && step["condition"] == "always()" { + publishAlways = true + } + } + require.NotEmpty(t, inlineScript) + require.True(t, publishAlways) + + for _, value := range []string{ + "MIGRATION_FAULT_SCENARIO", + "MIGRATION_FAULT_OS", + "MIGRATION_FAULT_CNI", + "MIGRATION_FAULT_RUN_ID", + "MIGRATION_FAULT_ARTIFACT_DIR", + "VALIDATE_STATE_BACKEND=bolt", + "export KUBECONFIG=", + "-test-kubeconfig=\"$KUBECONFIG\"", + "./test/integration/state", + "tee \"$artifactDir/go-test.log\"", + } { + require.Contains(t, inlineScript, value) + } + for _, forbidden := range []string{"killall", "pkill", "rollout restart"} { + require.NotContains(t, strings.ToLower(inlineScript), forbidden) + } +}