diff --git a/drivers/resfssgcp_nfs_cg/main.go b/drivers/resfssgcp_nfs_cg/main.go index 62593f241..353809532 100644 --- a/drivers/resfssgcp_nfs_cg/main.go +++ b/drivers/resfssgcp_nfs_cg/main.go @@ -2,25 +2,697 @@ package resfssgcp_nfs_cg import ( "context" + "encoding/json" + "errors" + "fmt" + "net/http" + "os" + "sort" + "strings" + "time" - "github.com/opensvc/om3/v3/core/datarecv" + "github.com/opensvc/om3/v3/core/actioncontext" + "github.com/opensvc/om3/v3/core/rawconfig" "github.com/opensvc/om3/v3/core/resource" "github.com/opensvc/om3/v3/core/status" + "github.com/opensvc/om3/v3/drivers/sgcphelper" + "github.com/opensvc/om3/v3/util/httpclientcache" + "github.com/opensvc/om3/v3/util/sgcp" ) +const ( + RetryWaitDelay = 2 * time.Second + waitMsgInterval = 10 * time.Second +) + +var ( + ErrAlreadyResumed = errors.New("already resumed") + ErrResumeInProgress = errors.New("resume in progress") +) + +type PreConditionError struct{ Err error } + +func (e *PreConditionError) Error() string { return fmt.Sprintf("precondition error: %s", e.Err) } +func (e *PreConditionError) Unwrap() error { return e.Err } + type ( - T struct { - resource.T - resource.Restart - datarecv.DataRecv + AZStatus struct { + AvailabilityZone string `json:"availabilityZone"` + Status string `json:"status"` + } + GeoRedundancyInfo struct { + Region string `json:"region"` + TargetAvailabilityZones []AZStatus `json:"targetAvailabilityZones"` + } + ReplicationInfo struct { + ReplicationMode string `json:"replicationMode"` + TargetAvailabilityZones []AZStatus `json:"targetAvailabilityZones"` + } + GeoTargetDetail struct { + Region string + AZ string + Status string + } + RepTargetDetail struct { + Mode string + AZ string + Status string + } + CgInfo struct { + UUID string `json:"uuid"` + Name string `json:"name"` + AvailabilityZone string `json:"availabilityZone"` + Status string `json:"status"` + GeoRedundancy GeoRedundancyInfo `json:"georedundancy"` + Replication ReplicationInfo `json:"replication"` } ) -// New creates a new SGCP NFS filesystem resource driver +func (cg *CgInfo) String() string { + if cg == nil { + return "NfsCg " + } + return fmt.Sprintf("NfsCg uuid:%s, name:%s, status:%s az:%s geo_redundancy:%+v replication:%+v", + cg.UUID, cg.Name, cg.Status, cg.AvailabilityZone, cg.GeoRedundancy, cg.Replication) +} + +func (cg *CgInfo) GeoRedundancies() []GeoTargetDetail { + region := cg.GeoRedundancy.Region + if region == "" { + region = "undef" + } + details := make([]GeoTargetDetail, 0, len(cg.GeoRedundancy.TargetAvailabilityZones)) + for _, target := range cg.GeoRedundancy.TargetAvailabilityZones { + details = append(details, GeoTargetDetail{ + Region: region, + AZ: target.AvailabilityZone, + Status: target.Status, + }) + } + return details +} + +func (cg *CgInfo) Replications() []RepTargetDetail { + mode := cg.Replication.ReplicationMode + if mode == "" { + mode = "undef" + } + details := make([]RepTargetDetail, 0, len(cg.Replication.TargetAvailabilityZones)) + for _, target := range cg.Replication.TargetAvailabilityZones { + details = append(details, RepTargetDetail{ + Mode: mode, + AZ: target.AvailabilityZone, + Status: target.Status, + }) + } + return details +} + +func (cg *CgInfo) hasReplication() bool { + return cg.Replication.ReplicationMode != "" || len(cg.Replication.TargetAvailabilityZones) > 0 +} + +func (cg *CgInfo) hasGeoRedundancy() bool { + return cg.GeoRedundancy.Region != "" || len(cg.GeoRedundancy.TargetAvailabilityZones) > 0 +} + +type ( + GetAuthInfoer interface { + GetAuthInfo(string) (*sgcp.AuthInfo, error) + } + logger interface { + Debugf(format string, args ...any) + Infof(format string, args ...any) + Warnf(format string, args ...any) + Errorf(format string, args ...any) + } + cgAPI interface { + GetConsistencyGroup(ctx context.Context, uuid string) (method, url string, code int, data []byte, err error) + PatchConsistencyGroup(ctx context.Context, uuid string, payload any) (method, url string, code int, data []byte, err error) + } + cgMgr struct { + uuid string + log logger + api cgAPI + } +) + +func (m *cgMgr) GetCg(ctx context.Context) (*CgInfo, error) { + m.log.Debugf("get consistency group %s info", m.uuid) + ts := time.Now() + method, url, code, data, err := m.api.GetConsistencyGroup(ctx, m.uuid) + if err != nil { + return nil, err + } + if code != http.StatusOK { + return nil, fmt.Errorf("get consistency group %s: unexpected status %d (method=%s url=%s)", m.uuid, code, method, url) + } + var cg CgInfo + if err := json.Unmarshal(data, &cg); err != nil { + return nil, fmt.Errorf("unmarshal consistency group %s: %w", m.uuid, err) + } + m.log.Debugf("consistency group details: %+v (duration %.2f)", &cg, time.Since(ts).Seconds()) + return &cg, nil +} + +func (m *cgMgr) Switchover(ctx context.Context, targetAZ string) error { + payload := map[string]any{ + "operation": "switchover", + "operationParameters": map[string]any{ + "availabilityZone": targetAZ, + }, + } + m.log.Infof("switchover consistency group %s to az %s ...", m.uuid, targetAZ) + ts := time.Now() + method, url, code, data, err := m.api.PatchConsistencyGroup(ctx, m.uuid, payload) + if err != nil { + return err + } + if code == http.StatusPreconditionFailed { + return &PreConditionError{Err: fmt.Errorf("switchover consistency group %s: status %d (method=%s url=%s)", m.uuid, code, method, url)} + } + if code != http.StatusAccepted { + return fmt.Errorf("switchover consistency group %s: unexpected status %d (method=%s url=%s body=%s)", m.uuid, code, method, url, string(data)) + } + m.log.Infof("| switched %s (duration %.2f)", m.uuid, time.Since(ts).Seconds()) + return nil +} + +func (m *cgMgr) Failover(ctx context.Context, az string) error { + payload := map[string]any{ + "operation": "failover", + "operationParameters": map[string]any{ + "availabilityZone": az, + "force": true, + }, + } + m.log.Infof("failover consistency group %s to az %s ...", m.uuid, az) + ts := time.Now() + method, url, code, data, err := m.api.PatchConsistencyGroup(ctx, m.uuid, payload) + if err != nil { + return err + } + if code != http.StatusAccepted { + return fmt.Errorf("failover consistency group %s: unexpected status %d (method=%s url=%s body=%s)", m.uuid, code, method, url, string(data)) + } + m.log.Infof("| failover %s (duration %.2f)", m.uuid, time.Since(ts).Seconds()) + return nil +} + +func (m *cgMgr) ResumeReplication(ctx context.Context, az string) error { + _ = az + payload := map[string]any{"operation": "resume-replication"} + m.log.Infof("resume-replication consistency group %s ...", m.uuid) + ts := time.Now() + method, url, code, data, err := m.api.PatchConsistencyGroup(ctx, m.uuid, payload) + if err != nil { + return err + } + if code != http.StatusAccepted { + return fmt.Errorf("resume-replication consistency group %s: unexpected status %d (method=%s url=%s body=%s)", m.uuid, code, method, url, string(data)) + } + m.log.Infof("| resume-replication %s (duration %.2f)", m.uuid, time.Since(ts).Seconds()) + return nil +} + +type T struct { + resource.T + + UUID string `json:"uuid"` + AZ string `json:"az,omitempty"` + Secret string `json:"secret,omitempty"` + Endpoint string `json:"endpoint,omitempty"` + Timeout time.Duration `json:"timeout"` + Failover bool `json:"failover"` + + lastWaitMsg time.Time + cgInfoCache *CgInfo + mgr *cgMgr + authInfoer GetAuthInfoer +} + func New() resource.Driver { return &T{} } +func (t *T) Configure() error { + cfg := sgcp.GetConfig() + if cfg == nil { + return fmt.Errorf("mandatory config file is required: %s", sgcp.DefaultConfigPath) + } + if t.Secret == "" { + t.Secret = cfg.Auth.DefaultSecret + } + if t.Secret == "" { + return fmt.Errorf("secret is required (neither defined into secret keyword nor config file %s", sgcp.DefaultConfigPath) + } + cfg = cfg.WithAuthSecret(t.Secret) + if t.Endpoint == "" { + t.Endpoint = cfg.Files.BaseURL + } + if t.Endpoint == "" { + return fmt.Errorf("file endpoint is required (neither defined into endpoint keyword nor config file %s", sgcp.DefaultConfigPath) + } + cfg = cfg.WithFileURL(t.Endpoint) + return t.configureMgr(cfg) +} + +func (t *T) configureMgr(cfg *sgcp.Config) error { + httpClient, err := httpclientcache.Client(httpclientcache.Options{ + Timeout: 30 * time.Second, + }) + if err != nil { + return fmt.Errorf("failed to create HTTP client: %w", err) + } + if t.authInfoer == nil { + t.authInfoer = &sgcphelper.GetAuthInfoFromDatastorePather{} + } + authInfo, err := t.authInfoer.GetAuthInfo(t.Secret) + if err != nil { + return fmt.Errorf("get auth info: %w", err) + } + tk := sgcp.NewTokenFactory(t.Log(), httpClient, &cfg.Auth, authInfo) + t.mgr = &cgMgr{ + uuid: t.UUID, + log: t.Log(), + api: sgcp.NewFilesAPI(cfg, httpClient, t.Log(), tk), + } + return nil +} + +func (t *T) Label(_ context.Context) string { + return t.UUID +} + +func (t *T) getCgCached(ctx context.Context) (*CgInfo, error) { + if t.cgInfoCache != nil { + return t.cgInfoCache, nil + } + cg, err := t.mgr.GetCg(ctx) + if err == nil { + t.cgInfoCache = cg + } + return cg, err +} + +func (t *T) clearGetCgCache() { + t.cgInfoCache = nil +} + +func (t *T) logGeoRedundancies(cg *CgInfo) { + geos := cg.GeoRedundancies() + if len(geos) == 0 { + return + } + geoStates := map[string]struct{}{} + for _, g := range geos { + geoStates[g.Status] = struct{}{} + } + switch cg.Status { + case "passive": + t.StatusLog().Info("geo mode remote -> local") + case "ready": + if cg.AvailabilityZone == t.AZ { + switch { + case isOnlyStatus(geoStates, "replicated"): + t.StatusLog().Info("geo mode local -> remote") + case isOnlyStatus(geoStates, "broken"): + t.StatusLog().Info("geo mode failover on remote") + case isOnlyStatus(geoStates, "unknown"): + t.StatusLog().Warn("geo mode failover on local ?") + default: + t.StatusLog().Info(fmt.Sprintf("geo mode transitioning states %s", joinStates(geoStates))) + } + } else { + t.StatusLog().Info("geo mode remote -> remote") + } + default: + t.StatusLog().Warn(fmt.Sprintf("geo mode states '%s'", joinStates(geoStates))) + } + t.StatusLog().Info(fmt.Sprintf("geo local az %s", t.AZ)) + regions := map[string]struct{}{} + for _, target := range geos { + regions[target.Region] = struct{}{} + } + for _, region := range sortedKeys(regions) { + t.StatusLog().Info(fmt.Sprintf("geo remote region %s", region)) + } + for _, target := range geos { + msg := fmt.Sprintf("geo %s %s", target.AZ, target.Status) + if target.Status == "replicated" { + t.StatusLog().Info(msg) + } else { + t.StatusLog().Warn(msg) + } + } +} + +func (t *T) logReplications(cg *CgInfo) { + reps := cg.Replications() + if len(reps) == 0 { + return + } + mode := cg.Replication.ReplicationMode + if mode == "" { + mode = "undef" + } + switch { + case cg.AvailabilityZone == t.AZ: + t.StatusLog().Info(fmt.Sprintf("rep mode %s local -> remote", mode)) + case repTargetsContainAZ(reps, t.AZ): + t.StatusLog().Info(fmt.Sprintf("rep mode %s remote -> local", mode)) + default: + t.StatusLog().Info(fmt.Sprintf("rep mode %s", mode)) + } + for _, target := range reps { + var msg string + if target.AZ == t.AZ { + msg = fmt.Sprintf("rep local %s %s", target.AZ, target.Status) + } else { + msg = fmt.Sprintf("rep remote %s %s", target.AZ, target.Status) + } + if target.Status == "replicated" { + t.StatusLog().Info(msg) + } else { + t.StatusLog().Warn(msg) + } + } +} + func (t *T) Status(ctx context.Context) status.T { + if sgcp.IsDisabled(rawconfig.NodeVarDir()) { + t.StatusLog().Info("xaas status disabled") + return status.NotApplicable + } + cg, err := t.getCgCached(ctx) + if err != nil { + t.StatusLog().Warn(fmt.Sprintf("get consistency group: %s", err)) + return status.NotApplicable + } + msg := fmt.Sprintf("status %s %s", cg.Status, cg.AvailabilityZone) + if cg.Status == "ready" || cg.Status == "passive" { + t.StatusLog().Info(msg) + } else { + t.StatusLog().Warn(msg) + } + t.logGeoRedundancies(cg) + t.logReplications(cg) return status.NotApplicable } + +func (t *T) waitStatus(ctx context.Context, expectedStates []string) error { + t.lastWaitMsg = time.Time{} + fn := func() (bool, error) { + cg, err := t.mgr.GetCg(ctx) + if err != nil { + return false, err + } + now := time.Now() + if contains(expectedStates, cg.Status) { + t.Log().Infof("| consistency group %s status is now %s", t.UUID, cg.Status) + return true, nil + } + if strings.Contains(cg.Status, "failed") || strings.Contains(cg.Status, "rollback") { + msg := fmt.Sprintf("abort waiting for consistency group %s status in '%v' because found status %s", + t.UUID, expectedStates, cg.Status) + t.Log().Warnf(msg) + return false, errors.New(msg) + } + if now.Sub(t.lastWaitMsg) >= waitMsgInterval { + t.Log().Infof("| waiting for consistency group %s status in %v. current status is %s", + t.UUID, expectedStates, cg.Status) + t.lastWaitMsg = now + } + return false, nil + } + errMsg := fmt.Sprintf("timeout waiting for consistency group %s status in %v", t.UUID, expectedStates) + return t.waitForFn(ctx, fn, t.Timeout, RetryWaitDelay, errMsg) +} + +func (t *T) waitForFn(ctx context.Context, fn func() (bool, error), timeout, retryDelay time.Duration, errMsg string) error { + deadline := time.Now().Add(timeout) + for { + ok, err := fn() + if err != nil { + return err + } + if ok { + return nil + } + if time.Now().After(deadline) { + return errors.New(errMsg) + } + select { + case <-ctx.Done(): + return ctx.Err() + case <-time.After(retryDelay): + } + } +} + +func (t *T) waitReady(ctx context.Context) error { + return t.waitStatus(ctx, []string{"ready", "passive"}) +} + +func (t *T) Start(ctx context.Context) error { + t.clearGetCgCache() + defer t.clearGetCgCache() + return t.start(ctx) +} + +func (t *T) start(ctx context.Context) error { + if sgcp.IsDisabled(rawconfig.NodeVarDir()) { + return nil + } + cg, err := t.mgr.GetCg(ctx) + if err != nil { + return err + } + if !contains([]string{"ready", "passive"}, cg.Status) { + t.Log().Infof("consistency group %s has an operation in progress. waiting ...", t.UUID) + if err := t.waitReady(ctx); err != nil { + return err + } + cg, err = t.mgr.GetCg(ctx) + if err != nil { + return err + } + } + if cg.Status == "ready" && cg.AvailabilityZone == t.AZ { + t.Log().Infof("consistency group %s is already up", t.UUID) + return nil + } + if actioncontext.IsForce(ctx) { + if err := t.mgr.Failover(ctx, t.AZ); err != nil { + return err + } + } else if err := t.mgr.Switchover(ctx, t.AZ); err != nil { + var preErr *PreConditionError + if !errors.As(err, &preErr) { + return err + } + msg := fmt.Sprintf("consistency group %s switchover 412 error", t.UUID) + if !t.Failover { + t.Log().Errorf("%s, skip failover fallback (resource failover is False)", msg) + return err + } + if os.Getenv("OSVC_ACTION_ORIGIN") != "daemon" { + t.Log().Errorf("%s, skip failover fallback, use --force if you want to try failover", msg) + return err + } + t.Log().Infof("%s, try failover", msg) + if err := t.mgr.Failover(ctx, t.AZ); err != nil { + return err + } + } + return t.waitStatus(ctx, []string{"ready"}) +} + +func (t *T) Stop(ctx context.Context) error { + _ = ctx + if sgcp.IsDisabled(rawconfig.NodeVarDir()) { + return nil + } + t.clearGetCgCache() + defer t.clearGetCgCache() + return nil +} + +func (t *T) SyncResume(ctx context.Context) error { + t.Log().Infof("sync resume ...") + t.clearGetCgCache() + defer t.clearGetCgCache() + if err := t.syncResume(ctx); err != nil { + t.Log().Errorf("sync resume failed") + return err + } + t.Log().Infof("sync resume succeed") + return nil +} + +func (t *T) syncResume(ctx context.Context) error { + msgPrefix := fmt.Sprintf("consistency group %s", t.UUID) + pendingResume := false + cg, err := t.mgr.GetCg(ctx) + if err != nil { + return err + } + if err := t.checkResumable(cg); err != nil { + switch { + case errors.Is(err, ErrAlreadyResumed): + t.Log().Infof("%s doesn't require sync resume", t.UUID) + return nil + case errors.Is(err, ErrResumeInProgress): + pendingResume = true + default: + return err + } + } + if !pendingResume { + if err := t.mgr.ResumeReplication(ctx, t.AZ); err != nil { + return err + } + } + if err := t.waitStatus(ctx, []string{"ready", "passive"}); err != nil { + return err + } + cg, err = t.mgr.GetCg(ctx) + if err != nil { + return err + } + if err := t.checkResumable(cg); err != nil { + if errors.Is(err, ErrAlreadyResumed) { + t.Log().Infof("%s now resumed", msgPrefix) + return nil + } + t.Log().Errorf("ERREUR : %s", err) + } + return fmt.Errorf("%s still not resumed", msgPrefix) +} + +func (t *T) checkResumable(cg *CgInfo) error { + hasRep := cg.hasReplication() + hasGeo := cg.hasGeoRedundancy() + + switch { + case hasRep && hasGeo: + return t.checkResumableReplicationAndGeo(cg) + case hasRep: + return t.checkResumableReplicationOnly(cg) + case hasGeo: + return t.checkResumableGeoOnly(cg) + default: + return fmt.Errorf("sync resume not allowed on cg %s without replication or georedundancy", t.UUID) + } +} + +func (t *T) checkResumableReplicationAndGeo(cg *CgInfo) error { + localRepStatus := t.localRepStatus(cg) + geoStatus := cg.GeoRedundancies()[0].Status + + if !contains([]string{"passive", "resuming"}, cg.Status) && + !contains([]string{"unknown", "replicated", "replicating"}, localRepStatus) { + return fmt.Errorf("sync resume not allowed on cg %s where status is %s and local replication status is %s", + t.UUID, cg.Status, localRepStatus) + } + if contains([]string{"passive", "replicated"}, localRepStatus) && geoStatus == "replicated" { + return ErrAlreadyResumed + } + if cg.Status == "resuming" { + return ErrResumeInProgress + } + return nil +} + +func (t *T) checkResumableReplicationOnly(cg *CgInfo) error { + localRepStatus := t.localRepStatus(cg) + + if !contains([]string{"ready", "resuming"}, cg.Status) { + return fmt.Errorf("sync resume not allowed when cg %s status is %s", t.UUID, cg.Status) + } + if cg.AvailabilityZone == t.AZ { + return fmt.Errorf("sync resume not allowed on cg %s where cg az is local az", t.UUID) + } + if localRepStatus == "replicated" { + return ErrAlreadyResumed + } + if cg.Status == "resuming" { + return ErrResumeInProgress + } + if localRepStatus != "unknown" { + return fmt.Errorf("sync resume not allowed on cg %s where local replication status is %s", t.UUID, localRepStatus) + } + return nil +} + +func (t *T) checkResumableGeoOnly(cg *CgInfo) error { + if cg.Status == "passive" { + return ErrAlreadyResumed + } + if cg.Status == "resuming" { + return ErrResumeInProgress + } + if cg.Status != "ready" { + return fmt.Errorf("sync resume not allowed when cg %s status is %s", t.UUID, cg.Status) + } + geoStatus := cg.GeoRedundancies()[0].Status + if !contains([]string{"broken", "unknown"}, geoStatus) { + return fmt.Errorf("sync resume not allowed on '%s' cg %s where georedundancy status is '%s'", cg.Status, t.UUID, geoStatus) + } + return nil +} + +func (t *T) localRepStatus(cg *CgInfo) string { + var localRep []RepTargetDetail + for _, rep := range cg.Replications() { + if rep.AZ == t.AZ { + localRep = append(localRep, rep) + } + } + if len(localRep) == 1 { + return localRep[0].Status + } + if cg.AvailabilityZone == t.AZ { + return cg.Status + } + return "" +} + +func contains(list []string, v string) bool { + for _, item := range list { + if item == v { + return true + } + } + return false +} + +func isOnlyStatus(set map[string]struct{}, val string) bool { + if len(set) != 1 { + return false + } + _, ok := set[val] + return ok +} + +func joinStates(set map[string]struct{}) string { + return strings.Join(sortedKeys(set), ",") +} + +func sortedKeys(set map[string]struct{}) []string { + keys := make([]string, 0, len(set)) + for k := range set { + keys = append(keys, k) + } + sort.Strings(keys) + return keys +} + +func repTargetsContainAZ(targets []RepTargetDetail, az string) bool { + for _, target := range targets { + if target.AZ == az { + return true + } + } + return false +} diff --git a/drivers/resfssgcp_nfs_cg/main_test.go b/drivers/resfssgcp_nfs_cg/main_test.go new file mode 100644 index 000000000..7a186a870 --- /dev/null +++ b/drivers/resfssgcp_nfs_cg/main_test.go @@ -0,0 +1,774 @@ +package resfssgcp_nfs_cg + +import ( + "context" + "encoding/json" + "errors" + "fmt" + "strings" + "sync" + "testing" + "time" + + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" + + "github.com/opensvc/om3/v3/core/actioncontext" + "github.com/opensvc/om3/v3/core/status" + "github.com/opensvc/om3/v3/util/sgcp" + "github.com/opensvc/om3/v3/util/sgcpcgtesthelper" + "github.com/opensvc/om3/v3/util/testsgcphelper" +) + +func setup(t *testing.T) func() { + t.Helper() + cfgFile := testsgcphelper.InstallConfig(t) + sgcp.SetConfigForTest(cfgFile) + require.NotNil(t, sgcp.GetConfig()) + return func() { + sgcp.SetConfigForTest("") + } +} + +const ( + cgUUIDRegion1 = "cg-uuid-region1" + region1AZ1 = "region1-az1" + region1AZ2 = "region1-az2" + region2AZ1 = "region2-az1" +) + +const repCgAZ1Active = `{ + "availabilityZone": "region1-az1", + "replication": { + "replicationMode": "sync", + "targetAvailabilityZones": [ + {"availabilityZone": "region1-az2", "status": "replicated"} + ] + }, + "status": "ready", + "uuid": "cg-uuid-region1" +}` + +const repCgAZ2Active = `{ + "availabilityZone": "region1-az2", + "replication": { + "replicationMode": "sync", + "targetAvailabilityZones": [ + {"availabilityZone": "region1-az1", "status": "replicated"} + ] + }, + "status": "ready", + "uuid": "cg-uuid-region1" +}` + +const repCgAZ1FailoverInProgressToAZ2 = `{ + "availabilityZone": "region1-az1", + "replication": { + "replicationMode": "sync", + "targetAvailabilityZones": [ + {"availabilityZone": "region1-az2", "status": "replicated"} + ] + }, + "status": "failover", + "uuid": "cg-uuid-region1" +}` + +const repCgAZ1ResumingRemoteAZ2 = `{ + "availabilityZone": "region1-az1", + "replication": { + "replicationMode": "sync", + "targetAvailabilityZones": [ + {"availabilityZone": "region1-az2", "status": "unknown"} + ] + }, + "status": "resuming", + "uuid": "cg-uuid-region1" +}` + +const geoCgRegion1AZ1Active = `{ + "availabilityZone": "region1-az1", + "georedundancy": { + "region": "region2", + "targetAvailabilityZones": [ + {"availabilityZone": "region2-az1", "status": "replicated"} + ], + "uuid": "cg-uuid-region2" + }, + "status": "ready", + "uuid": "cg-uuid-region1" +}` + +const geoCgRegion1AZ1Passive = `{ + "availabilityZone": "region1-az1", + "georedundancy": { + "region": "region2", + "targetAvailabilityZones": [ + {"availabilityZone": "region2-az1", "status": "replicated"} + ], + "uuid": "cg-uuid-region2" + }, + "status": "passive", + "uuid": "cg-uuid-region1" +}` + +const mixedRepGeoRegion1AZ1Active = `{ + "availabilityZone": "region1-az1", + "georedundancy": { + "region": "region2", + "targetAvailabilityZones": [ + {"availabilityZone": "region2-az1", "status": "replicated"} + ], + "uuid": "cg-uuid-region2" + }, + "replication": { + "replicationMode": "sync", + "targetAvailabilityZones": [ + {"availabilityZone": "region1-az2", "status": "replicated"} + ] + }, + "status": "ready", + "uuid": "cg-uuid-region1" +}` + +func mustParseCg(t *testing.T, raw string) *CgInfo { + t.Helper() + var cg CgInfo + if err := json.Unmarshal([]byte(raw), &cg); err != nil { + t.Fatalf("unmarshal fixture: %s", err) + } + return &cg +} + +func TestCgInfo_Replications(t *testing.T) { + cg := mustParseCg(t, repCgAZ1Active) + reps := cg.Replications() + if len(reps) != 1 { + t.Fatalf("expected 1 replication target, got %d", len(reps)) + } + if reps[0].AZ != region1AZ2 || reps[0].Status != "replicated" || reps[0].Mode != "sync" { + t.Fatalf("unexpected replication target: %+v", reps[0]) + } + if !cg.hasReplication() { + t.Fatal("expected hasReplication() to be true") + } + if cg.hasGeoRedundancy() { + t.Fatal("expected hasGeoRedundancy() to be false") + } +} + +func TestCgInfo_GeoRedundancies(t *testing.T) { + cg := mustParseCg(t, geoCgRegion1AZ1Active) + geos := cg.GeoRedundancies() + if len(geos) != 1 { + t.Fatalf("expected 1 geo-redundancy target, got %d", len(geos)) + } + if geos[0].AZ != region2AZ1 || geos[0].Status != "replicated" || geos[0].Region != "region2" { + t.Fatalf("unexpected geo-redundancy target: %+v", geos[0]) + } + if !cg.hasGeoRedundancy() { + t.Fatal("expected hasGeoRedundancy() to be true") + } + if cg.hasReplication() { + t.Fatal("expected hasReplication() to be false") + } +} + +func TestCgInfo_Mixed(t *testing.T) { + cg := mustParseCg(t, mixedRepGeoRegion1AZ1Active) + if !cg.hasReplication() || !cg.hasGeoRedundancy() { + t.Fatalf("expected both replication and geo-redundancy, got hasRep=%v hasGeo=%v", + cg.hasReplication(), cg.hasGeoRedundancy()) + } +} + +func TestCheckResumable_NotSyncable(t *testing.T) { + tests := []struct { + name string + az string + raw string + wantMsg string + }{ + { + name: "replication active az", + az: region1AZ1, + raw: repCgAZ1Active, + wantMsg: "sync resume not allowed on cg cg-uuid-region1 where cg az is local az", + }, + { + name: "replication active az while failover in progress", + az: region1AZ1, + raw: repCgAZ1FailoverInProgressToAZ2, + wantMsg: "sync resume not allowed when cg cg-uuid-region1 status is failover", + }, + { + name: "replication active az while resuming", + az: region1AZ1, + raw: repCgAZ1ResumingRemoteAZ2, + wantMsg: "sync resume not allowed on cg cg-uuid-region1 where cg az is local az", + }, + { + name: "georedundancy active az region az and status is not broken", + az: region1AZ1, + raw: geoCgRegion1AZ1Active, + wantMsg: "sync resume not allowed on 'ready' cg cg-uuid-region1 where georedundancy status is 'replicated'", + }, + { + name: "cg is ready and mix replication and georedundancy", + az: region1AZ1, + raw: mixedRepGeoRegion1AZ1Active, + wantMsg: "sync resume not allowed on cg cg-uuid-region1 where status is ready and local replication" + + " status is ready", + }, + } + + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + cg := mustParseCg(t, tt.raw) + rt := &T{UUID: cgUUIDRegion1, AZ: tt.az} + + err := rt.checkResumable(cg) + if err == nil { + t.Fatal("expected an error, got nil") + } + if errors.Is(err, ErrAlreadyResumed) || errors.Is(err, ErrResumeInProgress) { + t.Fatalf("expected a plain error, got sentinel: %v", err) + } + if !strings.Contains(err.Error(), tt.wantMsg) { + t.Fatalf("error = %q, want to contain %q", err.Error(), tt.wantMsg) + } + }) + } +} + +func TestCheckResumable_AlreadyResumed(t *testing.T) { + tests := []struct { + name string + az string + raw string + }{ + {name: "replication called from non active az", az: region1AZ1, raw: repCgAZ2Active}, + {name: "geo called from passive region", az: region1AZ1, raw: geoCgRegion1AZ1Passive}, + } + + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + cg := mustParseCg(t, tt.raw) + rt := &T{UUID: cgUUIDRegion1, AZ: tt.az} + + err := rt.checkResumable(cg) + if !errors.Is(err, ErrAlreadyResumed) { + t.Fatalf("expected ErrAlreadyResumed, got %v", err) + } + }) + } +} + +func TestLocalRepStatus(t *testing.T) { + cg := mustParseCg(t, repCgAZ2Active) + rt := &T{UUID: cgUUIDRegion1, AZ: region1AZ1} + if got := rt.localRepStatus(cg); got != "replicated" { + t.Fatalf("localRepStatus = %q, want %q", got, "replicated") + } + + rt2 := &T{UUID: cgUUIDRegion1, AZ: "region3-az1"} + if got := rt2.localRepStatus(cg); got != "" { + t.Fatalf("localRepStatus = %q, want empty", got) + } +} + +func TestContains(t *testing.T) { + if !contains([]string{"ready", "passive"}, "ready") { + t.Fatal("expected contains to find 'ready'") + } + if contains([]string{"ready", "passive"}, "resuming") { + t.Fatal("expected contains to not find 'resuming'") + } +} + +func TestIsOnlyStatus(t *testing.T) { + set := map[string]struct{}{"replicated": {}} + if !isOnlyStatus(set, "replicated") { + t.Fatal("expected isOnlyStatus to be true for a single matching entry") + } + set["broken"] = struct{}{} + if isOnlyStatus(set, "replicated") { + t.Fatal("expected isOnlyStatus to be false once a second state is present") + } +} + +func TestJoinStates(t *testing.T) { + set := map[string]struct{}{"broken": {}, "unknown": {}} + if got := joinStates(set); got != "broken,unknown" { + t.Fatalf("joinStates = %q, want %q (sorted)", got, "broken,unknown") + } +} + +func TestRepTargetsContainAZ(t *testing.T) { + targets := []RepTargetDetail{{AZ: region1AZ2, Status: "replicated"}} + if !repTargetsContainAZ(targets, region1AZ2) { + t.Fatal("expected repTargetsContainAZ to find region1-az2") + } + if repTargetsContainAZ(targets, region1AZ1) { + t.Fatal("expected repTargetsContainAZ to not find region1-az1") + } +} + +func TestWaitForFn_SucceedsBeforeTimeout(t *testing.T) { + rt := &T{} + calls := 0 + fn := func() (bool, error) { + calls++ + return calls >= 3, nil + } + ctx := context.Background() + err := rt.waitForFn(ctx, fn, time.Second, time.Millisecond, "timed out") + if err != nil { + t.Fatalf("unexpected error: %s", err) + } + if calls != 3 { + t.Fatalf("expected 3 calls, got %d", calls) + } +} + +func TestWaitForFn_TimesOut(t *testing.T) { + rt := &T{} + fn := func() (bool, error) { return false, nil } + ctx := context.Background() + err := rt.waitForFn(ctx, fn, 5*time.Millisecond, time.Millisecond, "timed out waiting") + if err == nil || !strings.Contains(err.Error(), "timed out waiting") { + t.Fatalf("expected timeout error, got %v", err) + } +} + +func TestWaitForFn_PropagatesError(t *testing.T) { + rt := &T{} + boom := errors.New("boom") + fn := func() (bool, error) { return false, boom } + ctx := context.Background() + err := rt.waitForFn(ctx, fn, time.Second, time.Millisecond, "timed out") + if !errors.Is(err, boom) { + t.Fatalf("expected boom error, got %v", err) + } +} + +func setupMockCG(t *testing.T, entries []sgcpcgtesthelper.CgEntry) (*sgcpcgtesthelper.DB, *sgcpcgtesthelper.API) { + t.Helper() + db := sgcpcgtesthelper.NewDB() + db.Setup(entries) + api := sgcpcgtesthelper.NewAPI(db) + return db, api +} + +func newTestDriver(t *testing.T, uuid, az string, timeout time.Duration, failover bool, api *sgcpcgtesthelper.API) *T { + t.Helper() + drv := &T{ + UUID: uuid, + AZ: az, + Timeout: timeout, + Failover: failover, + } + drv.mgr = &cgMgr{ + uuid: uuid, + log: drv.Log(), + api: api, + } + return drv +} + +func TestStart_SwitchoverSuccess(t *testing.T) { + cleanup := setup(t) + defer cleanup() + + uuid := "test-uuid" + db, api := setupMockCG(t, []sgcpcgtesthelper.CgEntry{ + { + UUID: uuid, + AvailabilityZone: region1AZ2, + Status: "ready", + Replication: sgcpcgtesthelper.ReplicationInfo{ + ReplicationMode: "sync", + TargetAvailabilityZones: []sgcpcgtesthelper.AZStatus{ + {AvailabilityZone: region1AZ1, Status: "replicated"}, + }, + }, + }, + }) + drv := newTestDriver(t, uuid, region1AZ1, 5*time.Second, false, api) + + ctx := context.Background() + err := drv.Start(ctx) + assert.NoError(t, err) + + entry, ok := db.Search(uuid) + require.True(t, ok) + assert.Equal(t, region1AZ1, entry.AvailabilityZone) + assert.Equal(t, "ready", entry.Status) + + calls := db.CallCounts() + assert.Equal(t, 2, calls.Get) + assert.Equal(t, 1, calls.Patch) + assert.Equal(t, 1, calls.Switch) + assert.Equal(t, 0, calls.Fail) +} + +func TestStart_Switchover412_FailoverAllowed(t *testing.T) { + cleanup := setup(t) + defer cleanup() + + uuid := "test-uuid" + db, api := setupMockCG(t, []sgcpcgtesthelper.CgEntry{ + { + UUID: uuid, + AvailabilityZone: region1AZ2, + Status: "ready", + }, + }) + db.PatchSwitchoverFunc = func(ctx context.Context, u, targetAZ string) error { + return &PreConditionError{Err: fmt.Errorf("simulated precondition failed")} + } + drv := newTestDriver(t, uuid, region1AZ1, 5*time.Second, true, api) + t.Setenv("OSVC_ACTION_ORIGIN", "daemon") + + ctx := context.Background() + err := drv.Start(ctx) + assert.NoError(t, err) + + entry, ok := db.Search(uuid) + require.True(t, ok) + assert.Equal(t, region1AZ1, entry.AvailabilityZone) + assert.Equal(t, "ready", entry.Status) + + calls := db.CallCounts() + assert.Equal(t, 2, calls.Get) + assert.Equal(t, 2, calls.Patch) + assert.Equal(t, 1, calls.Switch) + assert.Equal(t, 1, calls.Fail) +} + +func TestStart_Switchover412_FailoverNotAllowed_NoDaemon(t *testing.T) { + cleanup := setup(t) + defer cleanup() + + uuid := "test-uuid" + db, api := setupMockCG(t, []sgcpcgtesthelper.CgEntry{ + { + UUID: uuid, + AvailabilityZone: region1AZ2, + Status: "ready", + }, + }) + db.PatchSwitchoverFunc = func(ctx context.Context, u, targetAZ string) error { + return &PreConditionError{Err: fmt.Errorf("simulated precondition failed")} + } + drv := newTestDriver(t, uuid, region1AZ1, 5*time.Second, false, api) + t.Setenv("OSVC_ACTION_ORIGIN", "cli") + + ctx := context.Background() + err := drv.Start(ctx) + require.Error(t, err) + var preErr *PreConditionError + assert.True(t, errors.As(err, &preErr)) + + calls := db.CallCounts() + assert.Equal(t, 1, calls.Get) + assert.Equal(t, 1, calls.Patch) + assert.Equal(t, 1, calls.Switch) + assert.Equal(t, 0, calls.Fail) +} + +func TestStart_ForceFailover(t *testing.T) { + cleanup := setup(t) + defer cleanup() + + uuid := "test-uuid" + db, api := setupMockCG(t, []sgcpcgtesthelper.CgEntry{ + { + UUID: uuid, + AvailabilityZone: region1AZ2, + Status: "ready", + }, + }) + drv := newTestDriver(t, uuid, region1AZ1, 5*time.Second, false, api) + + ctx := actioncontext.WithForce(context.Background(), true) + err := drv.Start(ctx) + assert.NoError(t, err) + + entry, ok := db.Search(uuid) + require.True(t, ok) + assert.Equal(t, region1AZ1, entry.AvailabilityZone) + assert.Equal(t, "ready", entry.Status) + + calls := db.CallCounts() + assert.Equal(t, 2, calls.Get) + assert.Equal(t, 1, calls.Patch) + assert.Equal(t, 0, calls.Switch) + assert.Equal(t, 1, calls.Fail) +} + +func TestStart_AlreadyUp(t *testing.T) { + cleanup := setup(t) + defer cleanup() + + uuid := "test-uuid" + db, api := setupMockCG(t, []sgcpcgtesthelper.CgEntry{ + { + UUID: uuid, + AvailabilityZone: region1AZ1, + Status: "ready", + }, + }) + drv := newTestDriver(t, uuid, region1AZ1, 5*time.Second, false, api) + + ctx := context.Background() + err := drv.Start(ctx) + assert.NoError(t, err) + + calls := db.CallCounts() + assert.Equal(t, 1, calls.Get) + assert.Equal(t, 0, calls.Patch) +} + +func TestStart_OperationInProgress_WaitReady(t *testing.T) { + cleanup := setup(t) + defer cleanup() + + uuid := "test-uuid" + db, api := setupMockCG(t, []sgcpcgtesthelper.CgEntry{ + { + UUID: uuid, + AvailabilityZone: region1AZ1, + Status: "failover", + }, + }) + go func() { + time.Sleep(100 * time.Millisecond) + entry, _ := db.Search(uuid) + entry.Status = "ready" + _ = db.Update(entry) + }() + + drv := newTestDriver(t, uuid, region1AZ1, 2*time.Second, false, api) + ctx := context.Background() + err := drv.Start(ctx) + assert.NoError(t, err) + + entry, ok := db.Search(uuid) + require.True(t, ok) + assert.Equal(t, "ready", entry.Status) + assert.Equal(t, region1AZ1, entry.AvailabilityZone) + + calls := db.CallCounts() + assert.GreaterOrEqual(t, calls.Get, 2) + assert.Equal(t, 0, calls.Patch) +} + +func TestSyncResume_ReplicationOnly_Success(t *testing.T) { + cleanup := setup(t) + defer cleanup() + + uuid := "test-uuid" + db, api := setupMockCG(t, []sgcpcgtesthelper.CgEntry{ + { + UUID: uuid, + AvailabilityZone: region1AZ2, + Status: "ready", + Replication: sgcpcgtesthelper.ReplicationInfo{ + ReplicationMode: "sync", + TargetAvailabilityZones: []sgcpcgtesthelper.AZStatus{ + {AvailabilityZone: region1AZ1, Status: "unknown"}, + }, + }, + }, + }) + + db.PatchResumeFunc = func(ctx context.Context, u string) error { + entry, _ := db.Search(u) + entry.Replication.TargetAvailabilityZones[0].Status = "replicated" + entry.Status = "ready" + _ = db.Update(entry) + return nil + } + + drv := newTestDriver(t, uuid, region1AZ1, 5*time.Second, false, api) + + ctx := context.Background() + err := drv.SyncResume(ctx) + assert.NoError(t, err) + + entry, ok := db.Search(uuid) + require.True(t, ok) + var localStatus string + for _, rep := range entry.Replication.TargetAvailabilityZones { + if rep.AvailabilityZone == region1AZ1 { + localStatus = rep.Status + break + } + } + assert.Equal(t, "replicated", localStatus) + assert.Equal(t, "ready", entry.Status) + + calls := db.CallCounts() + assert.Equal(t, 3, calls.Get) + assert.Equal(t, 1, calls.Patch) + assert.Equal(t, 1, calls.Resume) +} + +func TestSyncResume_AlreadyResumed(t *testing.T) { + cleanup := setup(t) + defer cleanup() + + uuid := "test-uuid" + db, api := setupMockCG(t, []sgcpcgtesthelper.CgEntry{ + { + UUID: uuid, + AvailabilityZone: region1AZ2, + Status: "ready", + Replication: sgcpcgtesthelper.ReplicationInfo{ + ReplicationMode: "sync", + TargetAvailabilityZones: []sgcpcgtesthelper.AZStatus{ + {AvailabilityZone: region1AZ1, Status: "replicated"}, + }, + }, + }, + }) + drv := newTestDriver(t, uuid, region1AZ1, 5*time.Second, false, api) + + ctx := context.Background() + err := drv.SyncResume(ctx) + assert.NoError(t, err) + + calls := db.CallCounts() + assert.Equal(t, 1, calls.Get) + assert.Equal(t, 0, calls.Patch) +} + +func TestSyncResume_ResumeInProgress(t *testing.T) { + cleanup := setup(t) + defer cleanup() + + uuid := "test-uuid" + db, api := setupMockCG(t, []sgcpcgtesthelper.CgEntry{ + { + UUID: uuid, + AvailabilityZone: region1AZ2, + Status: "resuming", + Replication: sgcpcgtesthelper.ReplicationInfo{ + ReplicationMode: "sync", + TargetAvailabilityZones: []sgcpcgtesthelper.AZStatus{ + {AvailabilityZone: region1AZ1, Status: "unknown"}, + }, + }, + }, + }) + + var mu sync.Mutex + getCount := 0 + + api.GetConsistencyGroupFunc = func(ctx context.Context, u string) (method, url string, code int, data []byte, err error) { + mu.Lock() + getCount++ + if getCount >= 2 { + entry, _ := db.Search(u) + entry.Status = "ready" + entry.Replication.TargetAvailabilityZones[0].Status = "replicated" + _ = db.Update(entry) + } + mu.Unlock() + + savedFunc := api.GetConsistencyGroupFunc + api.GetConsistencyGroupFunc = nil + defer func() { api.GetConsistencyGroupFunc = savedFunc }() + + return api.GetConsistencyGroup(ctx, u) + } + + drv := newTestDriver(t, uuid, region1AZ1, 2*time.Second, false, api) + ctx := context.Background() + err := drv.SyncResume(ctx) + assert.NoError(t, err) + + entry, ok := db.Search(uuid) + require.True(t, ok) + assert.Equal(t, "ready", entry.Status) + localStatus := entry.Replication.TargetAvailabilityZones[0].Status + assert.Equal(t, "replicated", localStatus) + + calls := db.CallCounts() + assert.GreaterOrEqual(t, calls.Get, 2) + assert.Equal(t, 0, calls.Patch) + assert.Equal(t, 0, calls.Resume) +} + +func TestSyncResume_GeoOnly_Success(t *testing.T) { + cleanup := setup(t) + defer cleanup() + + uuid := "test-uuid" + db, api := setupMockCG(t, []sgcpcgtesthelper.CgEntry{ + { + UUID: uuid, + AvailabilityZone: region1AZ2, + Status: "ready", + GeoRedundancy: sgcpcgtesthelper.GeoRedundancyInfo{ + Region: "region2", + TargetAvailabilityZones: []sgcpcgtesthelper.AZStatus{ + {AvailabilityZone: region1AZ1, Status: "broken"}, + }, + }, + }, + }) + db.PatchResumeFunc = func(ctx context.Context, u string) error { + entry, _ := db.Search(u) + entry.GeoRedundancy.TargetAvailabilityZones[0].Status = "replicated" + entry.Status = "passive" + _ = db.Update(entry) + return nil + } + + drv := newTestDriver(t, uuid, region1AZ1, 5*time.Second, false, api) + + ctx := context.Background() + err := drv.SyncResume(ctx) + assert.NoError(t, err) + + entry, ok := db.Search(uuid) + require.True(t, ok) + geoStatus := entry.GeoRedundancy.TargetAvailabilityZones[0].Status + assert.Equal(t, "replicated", geoStatus) + assert.Equal(t, "passive", entry.Status) + + calls := db.CallCounts() + assert.Equal(t, 3, calls.Get) + assert.Equal(t, 1, calls.Patch) + assert.Equal(t, 1, calls.Resume) +} + +func TestStatus(t *testing.T) { + cleanup := setup(t) + defer cleanup() + + uuid := "test-uuid" + db, api := setupMockCG(t, []sgcpcgtesthelper.CgEntry{ + { + UUID: uuid, + AvailabilityZone: region1AZ1, + Status: "ready", + Replication: sgcpcgtesthelper.ReplicationInfo{ + ReplicationMode: "sync", + TargetAvailabilityZones: []sgcpcgtesthelper.AZStatus{ + {AvailabilityZone: region1AZ2, Status: "replicated"}, + }, + }, + }, + }) + drv := newTestDriver(t, uuid, region1AZ1, 5*time.Second, false, api) + + ctx := context.Background() + statusVal := drv.Status(ctx) + assert.Equal(t, status.NotApplicable, statusVal) + + calls := db.CallCounts() + assert.Equal(t, 1, calls.Get) + assert.Equal(t, 0, calls.Patch) +} diff --git a/util/sgcp/file.go b/util/sgcp/file.go index 4166c068b..43082e348 100644 --- a/util/sgcp/file.go +++ b/util/sgcp/file.go @@ -71,6 +71,29 @@ func (a *FilesAPI) DeleteNFSClients(ctx context.Context, fsUUID, clientUUID stri return } +// GetConsistencyGroup fetches a consistency group by uuid. +func (a *FilesAPI) GetConsistencyGroup(ctx context.Context, uuid string) (method, url string, code int, data []byte, err error) { + method = http.MethodGet + url = a.GetConsistencyGroupURL(uuid) + code, data, err = a.do(ctx, method, url, nil, a.GetScopes("files_read")...) + return +} + +func (a *FilesAPI) PatchConsistencyGroup(ctx context.Context, uuid string, payload any) (method, url string, code int, data []byte, err error) { + var b []byte + method = http.MethodPatch + url = a.GetConsistencyGroupURL(uuid) + + b, err = json.Marshal(payload) + if err != nil { + err = fmt.Errorf("failed to marshal consistency group patch: %w", err) + return + } + a.log.Infof("%s %s data=%s", method, url, string(b)) + code, data, err = a.do(ctx, method, url, bytes.NewReader(b), a.GetScopes("files_write")...) + return +} + func (a *FilesAPI) GetScopes(scopeType string) []string { return a.config.GetScopes(scopeType) } diff --git a/util/sgcpcgtesthelper/main.go b/util/sgcpcgtesthelper/main.go new file mode 100644 index 000000000..6bdb1b06b --- /dev/null +++ b/util/sgcpcgtesthelper/main.go @@ -0,0 +1,249 @@ +package sgcpcgtesthelper + +import ( + "context" + "encoding/json" + "fmt" + "net/http" + "sync" +) + +type AZStatus struct { + AvailabilityZone string `json:"availabilityZone"` + Status string `json:"status"` +} + +type GeoRedundancyInfo struct { + Region string `json:"region"` + TargetAvailabilityZones []AZStatus `json:"targetAvailabilityZones"` +} + +type ReplicationInfo struct { + ReplicationMode string `json:"replicationMode"` + TargetAvailabilityZones []AZStatus `json:"targetAvailabilityZones"` +} + +type CgEntry struct { + UUID string `json:"uuid"` + Name string `json:"name"` + AvailabilityZone string `json:"availabilityZone"` + Status string `json:"status"` + GeoRedundancy GeoRedundancyInfo `json:"georedundancy"` + Replication ReplicationInfo `json:"replication"` +} + +type DB struct { + mu sync.RWMutex + byUUID map[string]*CgEntry + callCount Counters + PatchSwitchoverFunc func(ctx context.Context, uuid, targetAZ string) error + PatchFailoverFunc func(ctx context.Context, uuid, targetAZ string) error + PatchResumeFunc func(ctx context.Context, uuid string) error + GetConsistencyGroupFunc func(ctx context.Context, uuid string) (method, url string, code int, data []byte, err error) +} + +type Counters struct { + Get int + Patch int + Switch int + Fail int + Resume int +} + +type API struct { + *DB +} + +func NewDB() *DB { + return &DB{ + byUUID: make(map[string]*CgEntry), + } +} + +func NewAPI(db *DB) *API { + return &API{DB: db} +} + +func (db *DB) Setup(entries []CgEntry) { + db.mu.Lock() + defer db.mu.Unlock() + db.byUUID = make(map[string]*CgEntry) + for _, e := range entries { + db.byUUID[e.UUID] = deepCopy(&e) + } +} + +func (db *DB) ResetCalls() { + db.mu.Lock() + defer db.mu.Unlock() + db.callCount = Counters{} +} + +func (db *DB) CallCounts() Counters { + db.mu.RLock() + defer db.mu.RUnlock() + return db.callCount +} + +func (a *API) GetConsistencyGroup(ctx context.Context, uuid string) (method, url string, code int, data []byte, err error) { + _ = ctx + if a.GetConsistencyGroupFunc != nil { + return a.GetConsistencyGroupFunc(ctx, uuid) + } + a.mu.Lock() + a.callCount.Get++ + a.mu.Unlock() + + a.mu.RLock() + defer a.mu.RUnlock() + entry, ok := a.byUUID[uuid] + if !ok { + return http.MethodGet, "/consistency-groups/" + uuid, http.StatusNotFound, nil, nil + } + b, err := json.Marshal(entry) + if err != nil { + return http.MethodGet, "/consistency-groups/" + uuid, http.StatusInternalServerError, nil, err + } + return http.MethodGet, "/consistency-groups/" + uuid, http.StatusOK, b, nil +} + +func (a *API) PatchConsistencyGroup(ctx context.Context, uuid string, payload any) (method, url string, code int, data []byte, err error) { + a.mu.Lock() + a.callCount.Patch++ + a.mu.Unlock() + + a.mu.RLock() + entry, ok := a.byUUID[uuid] + a.mu.RUnlock() + if !ok { + return http.MethodPatch, "/consistency-groups/" + uuid, http.StatusNotFound, nil, fmt.Errorf("cg not found") + } + + payloadMap, ok := payload.(map[string]any) + if !ok { + return http.MethodPatch, "/consistency-groups/" + uuid, http.StatusBadRequest, nil, fmt.Errorf("invalid payload type") + } + op, _ := payloadMap["operation"].(string) + + switch op { + case "switchover": + a.mu.Lock() + a.callCount.Switch++ + a.mu.Unlock() + if a.PatchSwitchoverFunc != nil { + params, _ := payloadMap["operationParameters"].(map[string]any) + targetAZ, _ := params["availabilityZone"].(string) + if err := a.PatchSwitchoverFunc(ctx, uuid, targetAZ); err != nil { + return http.MethodPatch, "/consistency-groups/" + uuid, http.StatusPreconditionFailed, nil, err + } + } + a.mu.Lock() + params, _ := payloadMap["operationParameters"].(map[string]any) + if targetAZ, ok := params["availabilityZone"].(string); ok { + entry.AvailabilityZone = targetAZ + entry.Status = "ready" + } + a.mu.Unlock() + return http.MethodPatch, "/consistency-groups/" + uuid, http.StatusAccepted, nil, nil + + case "failover": + a.mu.Lock() + a.callCount.Fail++ + a.mu.Unlock() + if a.PatchFailoverFunc != nil { + params, _ := payloadMap["operationParameters"].(map[string]any) + targetAZ, _ := params["availabilityZone"].(string) + if err := a.PatchFailoverFunc(ctx, uuid, targetAZ); err != nil { + return http.MethodPatch, "/consistency-groups/" + uuid, http.StatusInternalServerError, nil, err + } + } + a.mu.Lock() + params, _ := payloadMap["operationParameters"].(map[string]any) + if targetAZ, ok := params["availabilityZone"].(string); ok { + entry.AvailabilityZone = targetAZ + entry.Status = "ready" + } + a.mu.Unlock() + return http.MethodPatch, "/consistency-groups/" + uuid, http.StatusAccepted, nil, nil + + case "resume-replication": + a.mu.Lock() + a.callCount.Resume++ + a.mu.Unlock() + if a.PatchResumeFunc != nil { + if err := a.PatchResumeFunc(ctx, uuid); err != nil { + return http.MethodPatch, "/consistency-groups/" + uuid, http.StatusInternalServerError, nil, err + } + } + a.mu.Lock() + for i := range entry.Replication.TargetAvailabilityZones { + if entry.Replication.TargetAvailabilityZones[i].AvailabilityZone == entry.AvailabilityZone { + entry.Replication.TargetAvailabilityZones[i].Status = "replicated" + } + } + for i := range entry.GeoRedundancy.TargetAvailabilityZones { + if entry.GeoRedundancy.TargetAvailabilityZones[i].AvailabilityZone == entry.AvailabilityZone { + entry.GeoRedundancy.TargetAvailabilityZones[i].Status = "replicated" + } + } + entry.Status = "ready" + a.mu.Unlock() + return http.MethodPatch, "/consistency-groups/" + uuid, http.StatusAccepted, nil, nil + + default: + return http.MethodPatch, "/consistency-groups/" + uuid, http.StatusBadRequest, nil, fmt.Errorf("unknown operation %s", op) + } +} + +func (db *DB) Search(uuid string) (*CgEntry, bool) { + db.mu.RLock() + defer db.mu.RUnlock() + entry, ok := db.byUUID[uuid] + if !ok { + return nil, false + } + return deepCopy(entry), true +} + +func (db *DB) Update(entry *CgEntry) error { + db.mu.Lock() + defer db.mu.Unlock() + if _, ok := db.byUUID[entry.UUID]; !ok { + return fmt.Errorf("entry not found") + } + db.byUUID[entry.UUID] = deepCopy(entry) + return nil +} + +func deepCopy(src *CgEntry) *CgEntry { + if src == nil { + return nil + } + cp := &CgEntry{ + UUID: src.UUID, + Name: src.Name, + AvailabilityZone: src.AvailabilityZone, + Status: src.Status, + } + if len(src.GeoRedundancy.TargetAvailabilityZones) > 0 || src.GeoRedundancy.Region != "" { + cp.GeoRedundancy.Region = src.GeoRedundancy.Region + cp.GeoRedundancy.TargetAvailabilityZones = make([]AZStatus, len(src.GeoRedundancy.TargetAvailabilityZones)) + for i, az := range src.GeoRedundancy.TargetAvailabilityZones { + cp.GeoRedundancy.TargetAvailabilityZones[i] = AZStatus{ + AvailabilityZone: az.AvailabilityZone, + Status: az.Status, + } + } + } + if len(src.Replication.TargetAvailabilityZones) > 0 || src.Replication.ReplicationMode != "" { + cp.Replication.ReplicationMode = src.Replication.ReplicationMode + cp.Replication.TargetAvailabilityZones = make([]AZStatus, len(src.Replication.TargetAvailabilityZones)) + for i, az := range src.Replication.TargetAvailabilityZones { + cp.Replication.TargetAvailabilityZones[i] = AZStatus{ + AvailabilityZone: az.AvailabilityZone, + Status: az.Status, + } + } + } + return cp +}