diff --git a/Makefile b/Makefile index e457170d3..a2e8cd837 100644 --- a/Makefile +++ b/Makefile @@ -46,7 +46,7 @@ clean: rm -f $(OM) $(OX) compobj: - $(GOBUILD) -o $(COMPOBJ) ./util/compobj/ + $(GOBUILD) -trimpath -o $(COMPOBJ) ./util/compobj/ compobj-race: $(GOBUILDRACE) -o $(COMPOBJ) ./util/compobj/ @@ -94,13 +94,13 @@ install: $(PREFIX)/$(COMPOBJ) -i $(PREFIX)/$(COMPOBJ_D) om: - $(GOBUILD) -o $(OM) ./cmd/om/ + $(GOBUILD) -trimpath -o $(OM) ./cmd/om/ om-race: $(GOBUILDRACE) -o $(OM) ./cmd/om/ ox: - $(GOBUILD) -o $(OX) ./cmd/ox/ + $(GOBUILD) -trimpath -o $(OX) ./cmd/ox/ ox-race: $(GOBUILDRACE) -o $(OX) ./cmd/ox/ diff --git a/core/driverdb/drivers.go b/core/driverdb/drivers.go index 25765f0a9..d16e68fed 100644 --- a/core/driverdb/drivers.go +++ b/core/driverdb/drivers.go @@ -30,6 +30,7 @@ import ( _ "github.com/opensvc/om3/v3/drivers/resdisklv" _ "github.com/opensvc/om3/v3/drivers/resdiskmd" _ "github.com/opensvc/om3/v3/drivers/resdiskraw" + _ "github.com/opensvc/om3/v3/drivers/resdisksgcp_nfs_cg" _ "github.com/opensvc/om3/v3/drivers/resdiskvg" _ "github.com/opensvc/om3/v3/drivers/resdiskxp8" _ "github.com/opensvc/om3/v3/drivers/resfsdir" diff --git a/core/env/env.go b/core/env/env.go index 213133f17..00f21be17 100644 --- a/core/env/env.go +++ b/core/env/env.go @@ -26,7 +26,8 @@ var ( ) // HasDaemonOrigin returns true if the environment variable OSVC_ACTION_ORIGIN -// is set to "daemon". The opensvc daemon sets this variable on every command +// is set to one of the daemon origins: "daemon/monitor", "daemon/api" or +// "daemon/scheduler". The opensvc daemon sets this variable on every command // it executes. func HasDaemonOrigin() bool { switch Origin() { @@ -38,7 +39,7 @@ func HasDaemonOrigin() bool { } // HasDaemonMonitorOrigin returns true if the environment variable OSVC_ACTION_ORIGIN -// is set to "daemon/imon". The opensvc daemon sets this variable on every command +// is set to "daemon/monitor". The opensvc daemon sets this variable on every command // it executes. func HasDaemonMonitorOrigin() bool { switch Origin() { @@ -49,6 +50,18 @@ func HasDaemonMonitorOrigin() bool { } } +// HasDaemonSchedulerOrigin returns true if the environment variable +// OSVC_ACTION_ORIGIN is set to "daemon/scheduler", which the daemon sets on +// the commands its scheduler runs. +func HasDaemonSchedulerOrigin() bool { + switch Origin() { + case ActionOriginDaemonScheduler: + return true + default: + return false + } +} + // Origin returns the action origin using a env var that the daemon sets when // executing a CRM action. func Origin() ActionOrigin { diff --git a/core/env/env_test.go b/core/env/env_test.go new file mode 100644 index 000000000..48db56c6a --- /dev/null +++ b/core/env/env_test.go @@ -0,0 +1,35 @@ +package env + +import ( + "testing" + + "github.com/stretchr/testify/assert" +) + +func TestOriginPredicates(t *testing.T) { + cases := []struct { + origin string + daemon bool + monitor bool + scheduler bool + }{ + {origin: "", daemon: false, monitor: false, scheduler: false}, + {origin: string(ActionOriginUser), daemon: false, monitor: false, scheduler: false}, + {origin: string(ActionOriginDaemonMonitor), daemon: true, monitor: true, scheduler: false}, + {origin: string(ActionOriginDaemonAPI), daemon: true, monitor: false, scheduler: false}, + {origin: string(ActionOriginDaemonScheduler), daemon: true, monitor: false, scheduler: true}, + + // The daemon never sets these, and none of them is a daemon origin. + {origin: "daemon", daemon: false, monitor: false, scheduler: false}, + {origin: "scheduler", daemon: false, monitor: false, scheduler: false}, + } + + for _, tc := range cases { + t.Run("origin "+tc.origin, func(t *testing.T) { + t.Setenv(ActionOriginVar, tc.origin) + assert.Equal(t, tc.daemon, HasDaemonOrigin(), "HasDaemonOrigin") + assert.Equal(t, tc.monitor, HasDaemonMonitorOrigin(), "HasDaemonMonitorOrigin") + assert.Equal(t, tc.scheduler, HasDaemonSchedulerOrigin(), "HasDaemonSchedulerOrigin") + }) + } +} diff --git a/drivers/resfssgcp_nfs_cg/caps.go b/drivers/resdisksgcp_nfs_cg/caps.go similarity index 100% rename from drivers/resfssgcp_nfs_cg/caps.go rename to drivers/resdisksgcp_nfs_cg/caps.go diff --git a/drivers/resdisksgcp_nfs_cg/main.go b/drivers/resdisksgcp_nfs_cg/main.go new file mode 100644 index 000000000..a648ebc63 --- /dev/null +++ b/drivers/resdisksgcp_nfs_cg/main.go @@ -0,0 +1,854 @@ +package resfssgcp_nfs_cg + +import ( + "context" + "crypto/sha256" + "encoding/json" + "errors" + "fmt" + "net/http" + "sort" + "strings" + "time" + + "github.com/opensvc/om3/v3/core/actioncontext" + "github.com/opensvc/om3/v3/core/env" + "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/ageingcache" + "github.com/opensvc/om3/v3/util/httpclientcache" + "github.com/opensvc/om3/v3/util/sgcp" +) + +const ( + waitMsgInterval = 10 * time.Second +) + +var ( + retryWaitDelay = 2 * time.Second + + ErrAlreadyResumed = errors.New("already resumed") + ErrResumeInProgress = errors.New("resume in progress") + ErrPrecondition = errors.New("precondition error") +) + +type ( + 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"` + } +) + +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 + cache sgcp.CacheConfig + endpoint string + secret string + } +) + +// GetCg reads the consistency group past the cache. A cache that can not be +// dropped is an error: the read that follows would serve the entry GetCg was +// asked to go without, and the callers polling for a status change would spin +// on it until they time out. +func (m *cgMgr) GetCg(ctx context.Context) (*CgInfo, error) { + if err := m.cacheClearGetCg(); err != nil { + return nil, fmt.Errorf("clear the consistency group %s cache: %w", m.uuid, err) + } + return m.GetCachedCg(ctx) +} + +func (m *cgMgr) GetCachedCg(ctx context.Context) (*CgInfo, error) { + m.log.Debugf("get consistency group %s info", m.uuid) + ts := time.Now() + + sig := m.cacheSigGetCInfo() + ttl := time.Duration(m.cache.TTLSeconds) * time.Second + o := ageingcache.NewOutputter(m.getCgOutputter(ctx)) + data, err := ageingcache.Output(o, sig, ttl) + if err != nil { + return nil, fmt.Errorf("get consistency group %s: %w", m.uuid, err) + } + 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) getCgOutputter(ctx context.Context) func() ([]byte, error) { + return func() ([]byte, error) { + 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) + } + return data, nil + } +} + +func (m *cgMgr) cacheClearGetCg() error { + m.log.Debugf("clear consistency group %s cache", m.uuid) + return ageingcache.Clear(m.cacheSigGetCInfo()) +} + +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 code == http.StatusPreconditionFailed { + return fmt.Errorf("switchover consistency group %s: status %d (method=%s url=%s): %w", m.uuid, code, method, url, ErrPrecondition) + } + if err != nil { + return err + } + 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 +} + +// ResumeReplication asks the provider to resume the replication of the +// consistency group. Unlike the switchover and the failover, the operation +// takes no availability zone: the group resumes toward the targets it is +// already configured with. +func (m *cgMgr) ResumeReplication(ctx context.Context) error { + 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 +} + +// cacheSigGetCInfo generates the cache signature specific to the consistency +// group info operation. +func (m *cgMgr) cacheSigGetCInfo() string { + sig := m.cacheSig("get-cg-info") + m.log.Debugf("cacheSigGetCInfo %s: %s", m.uuid, sig) + return sig +} + +// cacheSig generates a unique cache signature by hashing the endpoint, secret, and UUID values of the consistency group. +func (m *cgMgr) cacheSig(s string) string { + data := fmt.Sprintf("%s|%s|%s", + m.endpoint, + m.secret, + m.uuid, + ) + hash := sha256.Sum256([]byte(data)) + return fmt.Sprintf("sgcp-nfs-cg-%s-%x", s, hash) +} + +type T struct { + resource.T + resource.Restart + + 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 + 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) + if t.AZ == "" { + // The az keyword defaults to {node.labels.az}, which evaluates to + // an empty string on a node where that label is not set. Refuse + // now: without a local az, start would ask for a switchover to + // the "" availability zone. + return fmt.Errorf("az is required (neither defined into the az keyword nor by the node az label)") + } + + if t.Timeout == nil { + timeout := time.Duration(cfg.Files.CG.Timeout) * time.Second + t.Timeout = &timeout + } + 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), + cache: cfg.Cache, + endpoint: t.Endpoint, + secret: t.Secret, + } + return nil +} + +func (t *T) Label(_ context.Context) string { + return t.UUID +} + +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("geo mode transitioning states %s", joinStates(geoStates)) + } + } else { + t.StatusLog().Info("geo mode remote -> remote") + } + default: + t.StatusLog().Warn("geo mode states '%s'", joinStates(geoStates)) + } + t.StatusLog().Info("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("geo remote region %s", region) + } + for _, target := range geos { + if target.Status == "replicated" { + t.StatusLog().Info("geo %s %s", target.AZ, target.Status) + } else { + t.StatusLog().Warn("geo %s %s", target.AZ, target.Status) + } + } +} + +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("rep mode %s local -> remote", mode) + case repTargetsContainAZ(reps, t.AZ): + t.StatusLog().Info("rep mode %s remote -> local", mode) + default: + t.StatusLog().Info("rep mode %s", mode) + } + for _, target := range reps { + where := "remote" + if target.AZ == t.AZ { + where = "local" + } + if target.Status == "replicated" { + t.StatusLog().Info("rep %s %s %s", where, target.AZ, target.Status) + } else { + t.StatusLog().Warn("rep %s %s %s", where, target.AZ, target.Status) + } + } +} + +func (t *T) Status(ctx context.Context) status.T { + if disabled, err := sgcp.IsDisabled(rawconfig.NodeVarDir()); err != nil { + t.StatusLog().Warn("%s", err) + return status.NotApplicable + } else if disabled { + t.StatusLog().Info("xaas status disabled") + return status.NotApplicable + } + if sgcphelper.NeedsCacheClear() { + if err := t.mgr.cacheClearGetCg(); err != nil { + t.Log().Debugf("clear get cg cache failed: %s", err) + t.StatusLog().Warn("possible stale value: clear get cg cache failed") + } + } + cg, err := t.mgr.GetCachedCg(ctx) + if err != nil { + t.StatusLog().Warn("get consistency group: %s", err) + return status.NotApplicable + } + if cg.Status == "ready" || cg.Status == "passive" { + t.StatusLog().Info("status %s %s", cg.Status, cg.AvailabilityZone) + } else { + t.StatusLog().Warn("status %s %s", cg.Status, cg.AvailabilityZone) + } + 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("%s", 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) waitStatusAndAZ(ctx context.Context, expectedStates []string, expectedAZ 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) { + if expectedAZ != "" && cg.AvailabilityZone != expectedAZ { + msg := fmt.Sprintf("consistency group %s reached status %s but in availability zone %s (expected %s)", + t.UUID, cg.Status, cg.AvailabilityZone, expectedAZ) + t.Log().Warnf("%s", msg) + // Continue waiting; the AZ may still be transitioning. + return false, nil + } + 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("%s", msg) + return false, errors.New(msg) + } + if now.Sub(t.lastWaitMsg) >= waitMsgInterval { + t.Log().Infof("| waiting for consistency group %s status in %v and availability zone %s. current status is %s (az %s)", + t.UUID, expectedStates, expectedAZ, cg.Status, cg.AvailabilityZone) + t.lastWaitMsg = now + } + return false, nil + } + errMsg := fmt.Sprintf("timeout waiting for consistency group %s status in %v and availability zone %s", t.UUID, expectedStates, expectedAZ) + 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 { + return t.start(ctx) +} + +func (t *T) start(ctx context.Context) error { + // An unknown disabled state has to stop the action here. Reading it as + // disabled would return success without switching the consistency group + // over, and the instance would be declared up in the wrong az. + disabled, err := sgcp.IsDisabled(rawconfig.NodeVarDir()) + if err != nil { + return err + } + if disabled { + t.Log().Infof("skip start of consistency group %s: sgcp support disabled", t.UUID) + 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.GetCachedCg(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 { + if !errors.Is(err, ErrPrecondition) { + 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 + } + // A failover is not a switchover: it is only tried on its own for an + // orchestration the daemon drives. An operator asks for it with + // --force, which does not come here. + if !env.HasDaemonOrigin() { + 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.waitStatusAndAZ(ctx, []string{"ready"}, t.AZ) +} + +// Stop does not move the consistency group. It stays in the availability zone +// the last start switched it to, and the start of the instance elsewhere is +// what moves it. The action says so rather than passing silently, so that an +// operator watching a stop is not left wondering whether it did anything. +// +// Having nothing to undo, it also treats an undecidable disabled flag as a +// warning where start has to refuse to go on. +func (t *T) Stop(ctx context.Context) error { + _ = ctx + disabled, err := sgcp.IsDisabled(rawconfig.NodeVarDir()) + switch { + case err != nil: + t.Log().Warnf("%s", err) + case disabled: + t.Log().Infof("skip stop of consistency group %s: sgcp support disabled", t.UUID) + default: + t.Log().Infof("stop leaves consistency group %s where it is, a start elsewhere moves it", t.UUID) + } + return nil +} + +// Resync implements the resource.Resyncer interface and is the entry point +// for the sync-resync action. +func (t *T) Resync(ctx context.Context) error { + return t.SyncResume(ctx) +} + +// SyncResume performs the actual resume-replication logic and is kept for +// backward compatibility with existing callers/tests. +func (t *T) SyncResume(ctx context.Context) error { + t.Log().Infof("sync resume ...") + 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", msgPrefix) + return nil + case errors.Is(err, ErrResumeInProgress): + pendingResume = true + default: + return err + } + } + if !pendingResume { + if err := t.mgr.ResumeReplication(ctx); err != nil { + return err + } + } + if err := t.waitStatus(ctx, []string{"ready", "passive"}); err != nil { + return err + } + cg, err = t.mgr.GetCachedCg(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 + } + return fmt.Errorf("%s still not resumed: %w", msgPrefix, err) + } + // The group accepts a resume again, which is not the resumed state the + // operation was waiting for. + 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 { + // Reject explicitly forbidden in-progress or failed operations + switch cg.Status { + case "failover", "failed", "rollback": + return fmt.Errorf("sync resume not allowed on cg %s in status %s", t.UUID, cg.Status) + } + + localRepStatus := t.localRepStatus(cg) + if localRepStatus == "" { + return fmt.Errorf("sync resume not allowed on cg %s: no local replication target found", t.UUID) + } + if !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) + } + + geos := cg.GeoRedundancies() + if len(geos) == 0 { + return fmt.Errorf("sync resume not allowed on cg %s: georedundancy has no target availability zones", t.UUID) + } + allGeoReplicated := true + for _, g := range geos { + if g.Status != "replicated" { + allGeoReplicated = false + break + } + } + + // Already resumed if local replication is passive/replicated and all geo targets are replicated + if contains([]string{"passive", "replicated"}, localRepStatus) && allGeoReplicated { + 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) + } + geos := cg.GeoRedundancies() + if len(geos) == 0 { + return fmt.Errorf("sync resume not allowed on cg %s: georedundancy has no target availability zones", t.UUID) + } + for _, g := range geos { + if !contains([]string{"broken", "unknown"}, g.Status) { + return fmt.Errorf("sync resume not allowed on '%s' cg %s where georedundancy status is '%s'", cg.Status, t.UUID, g.Status) + } + } + 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 multiple local targets (unusual) return the first non-empty status + for _, rep := range localRep { + if rep.Status != "" { + return rep.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/resdisksgcp_nfs_cg/main_test.go b/drivers/resdisksgcp_nfs_cg/main_test.go new file mode 100644 index 000000000..fd82d9ef9 --- /dev/null +++ b/drivers/resdisksgcp_nfs_cg/main_test.go @@ -0,0 +1,932 @@ +package resfssgcp_nfs_cg + +import ( + "context" + "encoding/json" + "errors" + "fmt" + "strings" + "sync" + "testing" + "time" + + "github.com/google/uuid" + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" + + "github.com/opensvc/om3/v3/core/actioncontext" + "github.com/opensvc/om3/v3/core/env" + "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()) + defaultRetryWaitDelay := retryWaitDelay + retryWaitDelay = 150 * time.Millisecond + return func() { + sgcp.SetConfigForTest("") + retryWaitDelay = defaultRetryWaitDelay + } +} + +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 TestConfigure_RequiresAZ(t *testing.T) { + cleanup := setup(t) + defer cleanup() + + // The az keyword defaults to {node.labels.az}, which evaluates to an + // empty string on a node where the label is not set. + drv := &T{UUID: cgUUIDRegion1} + err := drv.Configure() + require.Error(t, err) + assert.Contains(t, err.Error(), "az is required") +} + +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, id, az string, timeout time.Duration, failover bool, api *sgcpcgtesthelper.API) *T { + t.Helper() + drv := &T{ + UUID: id, + AZ: az, + Timeout: &timeout, + Failover: failover, + } + drv.mgr = &cgMgr{ + uuid: id, + log: drv.Log(), + api: api, + // Long enough that no test can outlive it, and that a slow run + // can not turn a cache hit into a miss. The tests that want a + // fresh read drop the entry, which is what the driver does. + cache: sgcp.CacheConfig{TTLSeconds: 60}, + } + return drv +} + +func TestStart_SwitchoverSuccess(t *testing.T) { + cleanup := setup(t) + defer cleanup() + + id := uuid.New().String() + + db, api := setupMockCG(t, []sgcpcgtesthelper.CgEntry{ + { + UUID: id, + AvailabilityZone: region1AZ2, + Status: "ready", + Replication: sgcpcgtesthelper.ReplicationInfo{ + ReplicationMode: "sync", + TargetAvailabilityZones: []sgcpcgtesthelper.AZStatus{ + {AvailabilityZone: region1AZ1, Status: "replicated"}, + }, + }, + }, + }) + drv := newTestDriver(t, id, region1AZ1, 5*time.Second, false, api) + + ctx := context.Background() + err := drv.Start(ctx) + assert.NoError(t, err) + + entry, ok := db.Search(id) + 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() + + id := uuid.New().String() + + db, api := setupMockCG(t, []sgcpcgtesthelper.CgEntry{ + { + UUID: id, + AvailabilityZone: region1AZ2, + Status: "ready", + }, + }) + db.PatchSwitchoverFunc = func(ctx context.Context, u, targetAZ string) error { + return fmt.Errorf("simulated precondition failed: %w", ErrPrecondition) + } + drv := newTestDriver(t, id, region1AZ1, 5*time.Second, true, api) + t.Setenv(env.ActionOriginVar, string(env.ActionOriginDaemonMonitor)) + + ctx := context.Background() + err := drv.Start(ctx) + assert.NoError(t, err) + + entry, ok := db.Search(id) + 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() + + id := uuid.New().String() + db, api := setupMockCG(t, []sgcpcgtesthelper.CgEntry{ + { + UUID: id, + AvailabilityZone: region1AZ2, + Status: "ready", + }, + }) + db.PatchSwitchoverFunc = func(ctx context.Context, u, targetAZ string) error { + return fmt.Errorf("simulated precondition failed: %w", ErrPrecondition) + } + drv := newTestDriver(t, id, region1AZ1, 5*time.Second, false, api) + t.Setenv(env.ActionOriginVar, string(env.ActionOriginUser)) + + ctx := context.Background() + err := drv.Start(ctx) + require.Error(t, err) + assert.True(t, errors.Is(err, ErrPrecondition)) + + 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() + + id := uuid.New().String() + + db, api := setupMockCG(t, []sgcpcgtesthelper.CgEntry{ + { + UUID: id, + AvailabilityZone: region1AZ2, + Status: "ready", + }, + }) + drv := newTestDriver(t, id, region1AZ1, 5*time.Second, false, api) + + ctx := actioncontext.WithForce(context.Background(), true) + err := drv.Start(ctx) + assert.NoError(t, err) + + entry, ok := db.Search(id) + 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() + + id := uuid.New().String() + + db, api := setupMockCG(t, []sgcpcgtesthelper.CgEntry{ + { + UUID: id, + AvailabilityZone: region1AZ1, + Status: "ready", + }, + }) + drv := newTestDriver(t, id, 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() + + id := uuid.New().String() + + db, api := setupMockCG(t, []sgcpcgtesthelper.CgEntry{ + { + UUID: id, + AvailabilityZone: region1AZ1, + Status: "failover", + }, + }) + go func() { + time.Sleep(100 * time.Millisecond) + entry, _ := db.Search(id) + entry.Status = "ready" + _ = db.Update(entry) + }() + + drv := newTestDriver(t, id, region1AZ1, 2*time.Second, false, api) + ctx := context.Background() + err := drv.Start(ctx) + assert.NoError(t, err) + + entry, ok := db.Search(id) + 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() + + id := uuid.New().String() + + db, api := setupMockCG(t, []sgcpcgtesthelper.CgEntry{ + { + UUID: id, + 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, id, region1AZ1, 5*time.Second, false, api) + ctx := context.Background() + err := drv.SyncResume(ctx) + assert.NoError(t, err) + + entry, ok := db.Search(id) + 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, 2, calls.Get) + assert.Equal(t, 1, calls.Patch) + assert.Equal(t, 1, calls.Resume) +} + +// TestSyncResume_ReportsTheReason verifies a resume that does not take is +// reported with what checkResumable said about the group, and not with the +// bare "still not resumed" it used to return, the reason going to the log +// alone. +func TestSyncResume_ReportsTheReason(t *testing.T) { + cleanup := setup(t) + defer cleanup() + + id := uuid.New().String() + + db, api := setupMockCG(t, []sgcpcgtesthelper.CgEntry{ + { + UUID: id, + AvailabilityZone: region1AZ2, + Status: "ready", + Replication: sgcpcgtesthelper.ReplicationInfo{ + ReplicationMode: "sync", + TargetAvailabilityZones: []sgcpcgtesthelper.AZStatus{ + {AvailabilityZone: region1AZ1, Status: "unknown"}, + }, + }, + }, + }) + + // The group settles in a state a resume is not allowed from. + db.PatchResumeFunc = func(ctx context.Context, u string) error { + entry, _ := db.Search(u) + entry.Status = "passive" + _ = db.Update(entry) + return nil + } + + drv := newTestDriver(t, id, region1AZ1, 5*time.Second, false, api) + ctx := context.Background() + + err := drv.SyncResume(ctx) + require.Error(t, err) + assert.Contains(t, err.Error(), "still not resumed") + assert.Containsf(t, err.Error(), "status is passive", "the returned error drops the reason: %s", err) +} + +// TestSyncResume_FailedResumeLeavesTheGroupAlone verifies a refused resume +// leaves the group as it was. The mock used to apply its own outcome before +// asking the hook, so a test injecting a failure still got a resumed group. +func TestSyncResume_FailedResumeLeavesTheGroupAlone(t *testing.T) { + cleanup := setup(t) + defer cleanup() + + id := uuid.New().String() + + // A target in the group's own az: the outcome the mock applies on its + // own is visible there, and has to stay out of a failed resume. + db, api := setupMockCG(t, []sgcpcgtesthelper.CgEntry{ + { + UUID: id, + AvailabilityZone: region1AZ2, + Status: "ready", + Replication: sgcpcgtesthelper.ReplicationInfo{ + ReplicationMode: "sync", + TargetAvailabilityZones: []sgcpcgtesthelper.AZStatus{ + {AvailabilityZone: region1AZ2, Status: "unknown"}, + {AvailabilityZone: region1AZ1, Status: "unknown"}, + }, + }, + }, + }) + db.PatchResumeFunc = func(ctx context.Context, u string) error { + return fmt.Errorf("the provider refused the resume") + } + + drv := newTestDriver(t, id, region1AZ1, 5*time.Second, false, api) + ctx := context.Background() + + err := drv.SyncResume(ctx) + require.Error(t, err) + + entry, ok := db.Search(id) + require.True(t, ok) + assert.Equal(t, 1, db.CallCounts().Resume) + assert.Equal(t, "ready", entry.Status) + for _, target := range entry.Replication.TargetAvailabilityZones { + assert.Equalf(t, "unknown", target.Status, "az %s replicates after a refused resume", target.AvailabilityZone) + } +} + +func TestSyncResume_AlreadyResumed(t *testing.T) { + cleanup := setup(t) + defer cleanup() + + id := uuid.New().String() + + db, api := setupMockCG(t, []sgcpcgtesthelper.CgEntry{ + { + UUID: id, + AvailabilityZone: region1AZ2, + Status: "ready", + Replication: sgcpcgtesthelper.ReplicationInfo{ + ReplicationMode: "sync", + TargetAvailabilityZones: []sgcpcgtesthelper.AZStatus{ + {AvailabilityZone: region1AZ1, Status: "replicated"}, + }, + }, + }, + }) + drv := newTestDriver(t, id, 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() + + id := uuid.New().String() + + db, api := setupMockCG(t, []sgcpcgtesthelper.CgEntry{ + { + UUID: id, + 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, id, region1AZ1, 2*time.Second, false, api) + ctx := context.Background() + err := drv.SyncResume(ctx) + assert.NoError(t, err) + + entry, ok := db.Search(id) + 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() + + id := uuid.New().String() + + db, api := setupMockCG(t, []sgcpcgtesthelper.CgEntry{ + { + UUID: id, + 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, id, region1AZ1, 5*time.Second, false, api) + + ctx := context.Background() + err := drv.SyncResume(ctx) + assert.NoError(t, err) + + entry, ok := db.Search(id) + 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, 2, calls.Get) + assert.Equal(t, 1, calls.Patch) + assert.Equal(t, 1, calls.Resume) +} + +// TestStatus_ReadsTheProviderOutsideTheScheduler verifies every origin but the +// scheduler reads the provider: the point of asking for a status is to see +// where the group is now. +func TestStatus_ReadsTheProviderOutsideTheScheduler(t *testing.T) { + for _, origin := range []env.ActionOrigin{ + env.ActionOriginUser, + env.ActionOriginDaemonMonitor, + env.ActionOriginDaemonAPI, + } { + t.Run(string(origin), func(t *testing.T) { + cleanup := setup(t) + defer cleanup() + t.Setenv(env.ActionOriginVar, string(origin)) + + id := uuid.New().String() + db, api := setupMockCG(t, []sgcpcgtesthelper.CgEntry{ + {UUID: id, AvailabilityZone: region1AZ1, Status: "ready"}, + }) + drv := newTestDriver(t, id, region1AZ1, 5*time.Second, false, api) + + ctx := context.Background() + for i := 0; i < 3; i++ { + assert.Equal(t, status.NotApplicable, drv.Status(ctx)) + } + assert.Equal(t, 3, db.CallCounts().Get, "a status evaluation was served by the cache") + }) + } +} + +// TestStatus covers the status evaluations the daemon scheduler makes, which +// are the ones the cache is for. +func TestStatus(t *testing.T) { + cleanup := setup(t) + defer cleanup() + t.Setenv(env.ActionOriginVar, string(env.ActionOriginDaemonScheduler)) + + id := uuid.New().String() + db, api := setupMockCG(t, []sgcpcgtesthelper.CgEntry{ + { + UUID: id, + AvailabilityZone: region1AZ1, + Status: "ready", + Replication: sgcpcgtesthelper.ReplicationInfo{ + ReplicationMode: "sync", + TargetAvailabilityZones: []sgcpcgtesthelper.AZStatus{ + {AvailabilityZone: region1AZ2, Status: "replicated"}, + }, + }, + }, + }) + drv := newTestDriver(t, id, region1AZ1, 5*time.Second, false, api) + + ctx := context.Background() + assert.Equal(t, status.NotApplicable, drv.Status(ctx)) + + calls := db.CallCounts() + expectedGetCall := 1 + assert.Equal(t, expectedGetCall, calls.Get) + assert.Equal(t, 0, calls.Patch) + + t.Log("the next evaluations are served by the cache") + for i := 0; i < 5; i++ { + assert.Equalf(t, status.NotApplicable, drv.Status(ctx), "status evaluation %d", i+2) + } + calls = db.CallCounts() + assert.Equal(t, expectedGetCall, calls.Get, "cache has not been used as expected") + + t.Log("dropping the entry sends the next evaluation to the api") + require.NoError(t, drv.mgr.cacheClearGetCg()) + assert.Equal(t, status.NotApplicable, drv.Status(ctx)) + calls = db.CallCounts() + assert.Equal(t, expectedGetCall+1, calls.Get, "expected call get after clear cache") +} diff --git a/drivers/resfssgcp_nfs_cg/manifest.go b/drivers/resdisksgcp_nfs_cg/manifest.go similarity index 93% rename from drivers/resfssgcp_nfs_cg/manifest.go rename to drivers/resdisksgcp_nfs_cg/manifest.go index 9f6d77f54..632b2c5dc 100644 --- a/drivers/resfssgcp_nfs_cg/manifest.go +++ b/drivers/resdisksgcp_nfs_cg/manifest.go @@ -14,7 +14,7 @@ var ( //go:embed text fs embed.FS - drvID = driver.NewID(driver.GroupFS, "sgcp_nfs_cg") + drvID = driver.NewID(driver.GroupDisk, "sgcp_nfs_cg") kws = []*keywords.Keyword{ { @@ -47,7 +47,6 @@ var ( Attr: "Timeout", Option: "timeout", Converter: converters.Duration, - Default: "300s", // TODO: move to config Scopable: true, Text: keywords.NewText(fs, "text/kw/timeout"), }, diff --git a/drivers/resdisksgcp_nfs_cg/text/kw/az b/drivers/resdisksgcp_nfs_cg/text/kw/az new file mode 100644 index 000000000..7d8a14360 --- /dev/null +++ b/drivers/resdisksgcp_nfs_cg/text/kw/az @@ -0,0 +1,5 @@ +The availability zone of the local node. Use a reference to a node label or a scoped value. + +The default resolves the az node label, so a node where that label is unset +has to set this keyword. The resource refuses to configure without it: the +start action would then ask for a switchover to an empty availability zone. diff --git a/drivers/resfssgcp_nfs_cg/text/kw/endpoint b/drivers/resdisksgcp_nfs_cg/text/kw/endpoint similarity index 100% rename from drivers/resfssgcp_nfs_cg/text/kw/endpoint rename to drivers/resdisksgcp_nfs_cg/text/kw/endpoint diff --git a/drivers/resdisksgcp_nfs_cg/text/kw/failover b/drivers/resdisksgcp_nfs_cg/text/kw/failover new file mode 100644 index 000000000..51f1bb9d7 --- /dev/null +++ b/drivers/resdisksgcp_nfs_cg/text/kw/failover @@ -0,0 +1,3 @@ +The failover keyword is used during the start action when --force is not specified. +When set to true, it enables fallback to the failover operation if the initial switchover operation fails with status code 412 and the action originates from the daemon. +When start --force is used, the failover operation is used directly, without first trying switchover, regardless of the failover keyword value. diff --git a/drivers/resfssgcp_nfs_cg/text/kw/secret b/drivers/resdisksgcp_nfs_cg/text/kw/secret similarity index 100% rename from drivers/resfssgcp_nfs_cg/text/kw/secret rename to drivers/resdisksgcp_nfs_cg/text/kw/secret diff --git a/drivers/resfssgcp_nfs_cg/text/kw/timeout b/drivers/resdisksgcp_nfs_cg/text/kw/timeout similarity index 100% rename from drivers/resfssgcp_nfs_cg/text/kw/timeout rename to drivers/resdisksgcp_nfs_cg/text/kw/timeout diff --git a/drivers/resfssgcp_nfs_cg/text/kw/uuid b/drivers/resdisksgcp_nfs_cg/text/kw/uuid similarity index 100% rename from drivers/resfssgcp_nfs_cg/text/kw/uuid rename to drivers/resdisksgcp_nfs_cg/text/kw/uuid diff --git a/drivers/resfssgcp_nfs/api.go b/drivers/resfssgcp_nfs/api.go index 2af12f7de..f96801519 100644 --- a/drivers/resfssgcp_nfs/api.go +++ b/drivers/resfssgcp_nfs/api.go @@ -2,10 +2,10 @@ package resfssgcp_nfs import ( "context" + "crypto/sha256" "encoding/json" "fmt" "net/http" - "slices" "time" "github.com/opensvc/om3/v3/util/ageingcache" @@ -22,6 +22,9 @@ type ( protocol string nfsIgnored []string + endpoint string + secret string + api *sgcp.FilesAPI cacheConfig *sgcp.CacheConfig log *plog.Logger @@ -88,19 +91,27 @@ func (mgr *nfsClientMgr) startExclusive(ctx context.Context) error { return err } -func (mgr *nfsClientMgr) cacheSig(name string) string { - return fmt.Sprintf("%s:%s", name, mgr.uuid) +// cacheSigGetFileInfo generates a cache signature specific to fetching file information. +func (mgr *nfsClientMgr) cacheSigGetFileInfo() string { + sig := mgr.cacheSig("get-file-info") + mgr.log.Debugf("cacheSigGetFileInfo %s: %s", mgr.uuid, sig) + return sig } -func (mgr *nfsClientMgr) cacheClear(name string) error { - cacheSig := mgr.cacheSig(name) - return ageingcache.Clear(cacheSig) +func (mgr *nfsClientMgr) cacheSig(name string) string { + data := fmt.Sprintf("%s|%s|%s", + mgr.endpoint, + mgr.secret, + mgr.uuid, + ) + hash := sha256.Sum256([]byte(data)) + return fmt.Sprintf("sgcp-nfs-%s-%x", name, hash) } func (mgr *nfsClientMgr) getFileInfo(ctx context.Context) (*FilesystemInfo, error) { var fileInfo FilesystemInfo - cacheSig := mgr.cacheSig("getFileInfo") + cacheSig := mgr.cacheSigGetFileInfo() ttl := time.Duration(mgr.cacheConfig.TTLSeconds) * time.Second o := ageingcache.NewOutputter(mgr.getFileInfoFactory(ctx)) data, err := ageingcache.Output(o, cacheSig, ttl) @@ -109,6 +120,13 @@ func (mgr *nfsClientMgr) getFileInfo(ctx context.Context) (*FilesystemInfo, erro return nil, fmt.Errorf("getFileInfo failed: %w", err) } + // The factory records an absent filesystem as a null document, the only + // way for it to travel through the cache. Hand it back as no filesystem + // at all, which is what every caller tests for. + if len(data) == 0 || string(data) == "null" { + return nil, nil + } + if err := json.Unmarshal(data, &fileInfo); err != nil { return nil, err } @@ -123,16 +141,14 @@ func (mgr *nfsClientMgr) getFileInfoFactory(ctx context.Context) func() ([]byte, return nil, err } - if err := mgr.api.CheckStatusCode(method, url, statusCode, http.StatusNotFound, http.StatusOK); err != nil { + if err := mgr.api.CheckStatusCode(method, url, statusCode, http.StatusOK, http.StatusNotFound); err != nil { return nil, err } - switch statusCode { - case http.StatusOK: - case http.StatusNotFound: - return nil, nil - default: - // paranoid, should never happen - return nil, fmt.Errorf("%s %s got unexpected status code %d", method, url, statusCode) + // A filesystem that is not there is an answer, not a failure. Cache + // it as a null document, so the absence ages like a presence would. + if statusCode == http.StatusNotFound { + mgr.log.Debugf("%s %s: no such filesystem", method, url) + return []byte("null"), nil } return data, nil } @@ -223,33 +239,23 @@ func (mgr *nfsClientMgr) deleteNFSClient(ctx context.Context, client NfsClient) mgr.log.Infof("drop permission %s for host %s on filesystem %s%s ...", client.Permission, client.Host, mgr.uuid, cgMsg) method, url, statusCode, _, err := mgr.api.DeleteNFSClients(ctx, mgr.uuid, client.UUID) + + // The provider refusing the drop while the consistency group is busy is a + // state to report, not a transport failure. + if statusCode == http.StatusPreconditionFailed { + return fmt.Errorf("consistency group is not in ready (status_code %d)", statusCode) + } if err != nil { return err } - if err := mgr.api.CheckStatusCode(method, url, statusCode, http.StatusNoContent, http.StatusPreconditionFailed); err != nil { + if err := mgr.api.CheckStatusCode(method, url, statusCode, http.StatusNoContent); err != nil { return err } - switch statusCode { - case http.StatusNoContent: - case http.StatusPreconditionFailed: - return fmt.Errorf("consistency group is not in ready (status_code %d)", statusCode) - default: - // paranoid, should never happen - return fmt.Errorf("unexpected status code %d", statusCode) - } mgr.log.Infof("deleted %s on filesystem %s", client, mgr.uuid) return nil } -func (mgr *nfsClientMgr) checkStatusCode(method, url string, got int, wanted ...int) error { - mgr.log.Debugf("%s %s status code: %d", method, url, got) - if slices.Contains(wanted, got) { - return nil - } - return fmt.Errorf("unexpected status code for %s %s got %d wanted %v", method, url, got, wanted) -} - // isClientIgnored checks if a client host should be ignored func (mgr *nfsClientMgr) isClientIgnored(host string) bool { for _, ignored := range mgr.nfsIgnored { diff --git a/drivers/resfssgcp_nfs/main.go b/drivers/resfssgcp_nfs/main.go index 6971202b2..eab6971d6 100644 --- a/drivers/resfssgcp_nfs/main.go +++ b/drivers/resfssgcp_nfs/main.go @@ -3,7 +3,6 @@ package resfssgcp_nfs import ( "context" - "errors" "fmt" "time" @@ -13,6 +12,7 @@ import ( "github.com/opensvc/om3/v3/core/status" "github.com/opensvc/om3/v3/drivers/resfshost" "github.com/opensvc/om3/v3/drivers/sgcphelper" + "github.com/opensvc/om3/v3/util/ageingcache" "github.com/opensvc/om3/v3/util/httpclientcache" "github.com/opensvc/om3/v3/util/sgcp" ) @@ -62,10 +62,9 @@ type ( CheckRead bool `json:"check_read"` // Internal state - resFs fsDriver - fileInfoCache *FilesystemInfo - mgr *nfsClientMgr - authInfoer GetAuthInfoer + resFs fsDriver + mgr *nfsClientMgr + authInfoer GetAuthInfoer } GetAuthInfoer interface { @@ -83,8 +82,10 @@ type ( } ) -// NfsClientIgnored is a list of NFS client hosts to ignore -var NfsClientIgnored = []string{} +var ( + // NfsClientIgnored is a list of NFS client hosts to ignore + NfsClientIgnored = []string{} +) // New creates a new SGCP NFS filesystem resource driver func New() resource.Driver { @@ -162,6 +163,8 @@ func (t *T) configureMgr(cfg *sgcp.Config) error { protocol: t.Protocol, log: t.Log(), nfsIgnored: NfsClientIgnored, + endpoint: t.Endpoint, + secret: t.Secret, api: sgcp.NewFilesAPI(cfg, httpClient, t.Log(), tk), cacheConfig: &cfg.Cache, } @@ -231,6 +234,12 @@ func (t *T) Stop(ctx context.Context) error { // Status returns the combined status of the file and fs func (t *T) Status(ctx context.Context) status.T { + if sgcphelper.NeedsCacheClear() { + if err := t.clearFileStatusCache(); err != nil { + t.Log().Debugf("clear get file status cache failed: %s", err) + t.StatusLog().Warn("possible stale value: clear get file status cache failed") + } + } fileStatus := t.fileStatus(ctx) // Get underlying filesystem status @@ -266,7 +275,11 @@ func (t *T) fileStart(ctx context.Context) error { defer func() { _ = t.clearFileStatusCache() }() - if sgcp.IsDisabled(rawconfig.NodeVarDir()) { + disabled, err := sgcp.IsDisabled(rawconfig.NodeVarDir()) + if err != nil { + return err + } + if disabled { t.Log().Infof("skipping file start %s: SGCP API disabled", t.UUID) return nil } @@ -293,7 +306,11 @@ func (t *T) fileStop(ctx context.Context) error { defer func() { _ = t.clearFileStatusCache() }() - if sgcp.IsDisabled(rawconfig.NodeVarDir()) { + disabled, err := sgcp.IsDisabled(rawconfig.NodeVarDir()) + if err != nil { + return err + } + if disabled { t.Log().Infof("skipping file stop %s: SGCP API disabled", t.UUID) return nil } @@ -316,7 +333,10 @@ func (t *T) fileStop(ctx context.Context) error { // fileStatus returns the status of the filesystem from the SGCP API func (t *T) fileStatus(ctx context.Context) status.T { // Check if XaaS status is disabled - if sgcp.IsDisabled(rawconfig.NodeVarDir()) { + if disabled, err := sgcp.IsDisabled(rawconfig.NodeVarDir()); err != nil { + t.StatusLog().Warn("%s", err) + return status.NotApplicable + } else if disabled { t.Log().Debugf("skipping file status %s: SGCP API disabled", t.UUID) return status.NotApplicable } @@ -370,18 +390,7 @@ func (t *T) fileStatus(ctx context.Context) status.T { // getFileInfo retrieves filesystem information from the API func (t *T) getFileInfo(ctx context.Context) (*FilesystemInfo, error) { - // Use cached value if available - if t.fileInfoCache != nil { - return t.fileInfoCache, nil - } - - fileInfo, err := t.mgr.getFileInfo(ctx) - if err == nil { - // Cache the result - t.fileInfoCache = fileInfo - } - - return fileInfo, err + return t.mgr.getFileInfo(ctx) } // getNFSClients returns the NFS clients for the filesystem, filtered by ignored hosts @@ -412,12 +421,8 @@ func (t *T) isClientIgnored(host string) bool { // clearFileStatusCache clears the filesystem info cache func (t *T) clearFileStatusCache() error { - var errs error - t.fileInfoCache = nil - for _, s := range []string{"getFileInfo"} { - errs = errors.Join(errs, t.mgr.cacheClear(s)) - } - return errs + t.Log().Debugf("clear get file info cache") + return ageingcache.Clear(t.mgr.cacheSigGetFileInfo()) } // String returns a string representation of an NfsClient diff --git a/drivers/resfssgcp_nfs/main_test.go b/drivers/resfssgcp_nfs/main_test.go index 4fcdfcdf1..2f8508165 100644 --- a/drivers/resfssgcp_nfs/main_test.go +++ b/drivers/resfssgcp_nfs/main_test.go @@ -3,9 +3,13 @@ package resfssgcp_nfs import ( "context" "encoding/json" + "fmt" "net/http" "net/http/httptest" + "os" + "path/filepath" "testing" + "time" "github.com/opensvc/om3/v3/drivers/sgcpauthtesthelper" "github.com/opensvc/om3/v3/util/testsgcphelper" @@ -14,8 +18,11 @@ import ( "github.com/stretchr/testify/require" "github.com/opensvc/om3/v3/core/driver" + "github.com/opensvc/om3/v3/core/env" + "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/util/ageingcache" "github.com/opensvc/om3/v3/util/sgcp" ) @@ -174,6 +181,161 @@ func TestGetNFSClients(t *testing.T) { } // TestFileStatusWithNoClients tests status when no clients are available +// sgcpServer runs an sgcp api the driver can talk to, and points the +// configuration at it. The handler answers the token request; everything else +// is up to the test. +func sgcpServer(t *testing.T, handler http.HandlerFunc) *httptest.Server { + t.Helper() + srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + w.Header().Set("Content-Type", "application/json") + if r.URL.Path == "/auth/access_token" { + fmt.Fprint(w, `{"access_token": "a-token"}`) + return + } + handler(w, r) + })) + t.Cleanup(srv.Close) + + root := filepath.Dir(testsgcphelper.InstallConfig(t)) + cfgFile := filepath.Join(root, "server.yaml") + require.NoError(t, os.WriteFile(cfgFile, []byte(` +auth: + base_url: "`+srv.URL+`/auth" + default_secret: "the-secret" + scopes: + files_read: ["files:read"] + files_write: ["files:write"] + timeout: 5 + ttl_seconds: 60 +files: + base_url: "`+srv.URL+`/file" + path: + fs: "/fs" + client: "/client" + cg: "/cg" +cache: + ttl_seconds: 0 +`), 0644)) + sgcp.SetConfigForTest(cfgFile) + t.Cleanup(func() { sgcp.SetConfigForTest("") }) + return srv +} + +func sgcpDriver(t *testing.T) *T { + t.Helper() + drv := newDrvWithRid("test-rid") + drv.authInfoer = sgcpauthtesthelper.NewMockGetAuthInfoProvider("id1") + drv.UUID = "fs-uuid" + drv.Host = "test-host" + drv.Permission = "read-write" + require.NoError(t, drv.Configure()) + return drv +} + +// TestFilesystemNotFound verifies a filesystem the provider does not have is +// an absence the driver converges on, not an error. The api reports every +// status over 400 as an error, which used to leave the not-found branch and +// every "fileInfo == nil" guard behind it unreachable. +func TestFilesystemNotFound(t *testing.T) { + sgcpServer(t, func(w http.ResponseWriter, r *http.Request) { + w.WriteHeader(http.StatusNotFound) + fmt.Fprint(w, `{"message": "no such filesystem"}`) + }) + drv := sgcpDriver(t) + ctx := context.Background() + + fileInfo, err := drv.mgr.getFileInfo(ctx) + require.NoError(t, err, "an absent filesystem is reported as a failure") + assert.Nil(t, fileInfo, "an absent filesystem is reported as a filesystem") + + assert.Equal(t, status.Down, drv.fileStatus(ctx)) + assert.NoError(t, drv.fileStop(ctx), "the stop of an absent filesystem does not converge") +} + +// TestDeleteNFSClientPreconditionFailed verifies the provider refusing to drop +// a client while the consistency group is busy reaches the operator as the +// message written for it. +func TestDeleteNFSClientPreconditionFailed(t *testing.T) { + sgcpServer(t, func(w http.ResponseWriter, r *http.Request) { + switch { + case r.Method == http.MethodDelete: + w.WriteHeader(http.StatusPreconditionFailed) + fmt.Fprint(w, `{"message": "consistency group busy"}`) + default: + fmt.Fprint(w, `{"uuid": "fs-uuid", "status": "online", "nfsClients": [ + {"uuid": "client-uuid", "host": "test-host", "permission": "read-write", "protocol": "nfs4.1"}]}`) + } + }) + drv := sgcpDriver(t) + ctx := context.Background() + + require.Equal(t, status.Up, drv.fileStatus(ctx), "the test needs a granted client to drop") + + err := drv.fileStop(ctx) + require.Error(t, err) + assert.Contains(t, err.Error(), "consistency group is not in ready") +} + +// TestFileStatusCache verifies the cached filesystem info the scheduler fills +// is not served to anyone else asking for a status. +func TestFileStatusCache(t *testing.T) { + cachedFiles := func(t *testing.T) []string { + t.Helper() + entries, err := os.ReadDir(filepath.Join(rawconfig.Paths.Cache, "ageing")) + if err != nil { + return nil + } + l := make([]string, 0, len(entries)) + for _, e := range entries { + l = append(l, e.Name()) + } + return l + } + + // fill the cache the way a status evaluation does, then ask for a status + // and see what became of the entry. + fill := func(t *testing.T) *T { + t.Helper() + drv := newDrvWithRid("test-rid") + drv.authInfoer = sgcpauthtesthelper.NewMockGetAuthInfoProvider("id1") + drv.UUID = "test-uuid" + drv.Host = "test-host" + drv.Permission = "read-write" + require.NoError(t, drv.Configure()) + + sig := drv.mgr.cacheSigGetFileInfo() + o := ageingcache.NewOutputter(func() ([]byte, error) { return []byte("null"), nil }) + _, err := ageingcache.Output(o, sig, time.Hour) + require.NoError(t, err) + require.NotEmpty(t, cachedFiles(t), "the cache was not filled") + return drv + } + + for _, origin := range []env.ActionOrigin{ + env.ActionOriginUser, + env.ActionOriginDaemonMonitor, + env.ActionOriginDaemonAPI, + } { + t.Run(string(origin)+" drops it", func(t *testing.T) { + defer Setup(t)() + t.Setenv(env.ActionOriginVar, string(origin)) + drv := fill(t) + + drv.Status(context.Background()) + assert.Empty(t, cachedFiles(t), "the status evaluation was served the cache") + }) + } + + t.Run("the scheduler keeps it", func(t *testing.T) { + defer Setup(t)() + t.Setenv(env.ActionOriginVar, string(env.ActionOriginDaemonScheduler)) + drv := fill(t) + + drv.Status(context.Background()) + assert.NotEmpty(t, cachedFiles(t), "the scheduler evaluation dropped the cache") + }) +} + func TestFileStatusWithNoClients(t *testing.T) { // Create a test server server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { diff --git a/drivers/resfssgcp_nfs_cg/main.go b/drivers/resfssgcp_nfs_cg/main.go deleted file mode 100644 index 62593f241..000000000 --- a/drivers/resfssgcp_nfs_cg/main.go +++ /dev/null @@ -1,26 +0,0 @@ -package resfssgcp_nfs_cg - -import ( - "context" - - "github.com/opensvc/om3/v3/core/datarecv" - "github.com/opensvc/om3/v3/core/resource" - "github.com/opensvc/om3/v3/core/status" -) - -type ( - T struct { - resource.T - resource.Restart - datarecv.DataRecv - } -) - -// New creates a new SGCP NFS filesystem resource driver -func New() resource.Driver { - return &T{} -} - -func (t *T) Status(ctx context.Context) status.T { - return status.NotApplicable -} diff --git a/drivers/resfssgcp_nfs_cg/text/kw/az b/drivers/resfssgcp_nfs_cg/text/kw/az deleted file mode 100644 index 494d5e5af..000000000 --- a/drivers/resfssgcp_nfs_cg/text/kw/az +++ /dev/null @@ -1 +0,0 @@ -The availability zone of the local node. Use a reference to a node label or a scoped value. \ No newline at end of file diff --git a/drivers/resfssgcp_nfs_cg/text/kw/failover b/drivers/resfssgcp_nfs_cg/text/kw/failover deleted file mode 100644 index f14fedda9..000000000 --- a/drivers/resfssgcp_nfs_cg/text/kw/failover +++ /dev/null @@ -1,2 +0,0 @@ -Allows daemon resource start to call operation 'switchover' when the 'failover' operation fails with status code 412. -When '--force' option is used, the setting has no effect because operation during start is always 'failover'. \ No newline at end of file diff --git a/drivers/resipsgcp_dnsalias/main.go b/drivers/resipsgcp_dnsalias/main.go index cf0a6aacd..5399cacb2 100644 --- a/drivers/resipsgcp_dnsalias/main.go +++ b/drivers/resipsgcp_dnsalias/main.go @@ -7,6 +7,7 @@ import ( "strings" "time" + "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" @@ -32,13 +33,24 @@ type ( // for tests api apiProvider + authInfoer GetAuthInfoer noneTarget string } + // GetAuthInfoer resolves the api credentials named by the secret keyword. + // The resfssgcp_nfs driver has the same seam, for the same reason: a test + // needs the real api without a datastore behind it. + GetAuthInfoer interface { + GetAuthInfo(string) (*sgcp.AuthInfo, error) + } + mgr struct { - alias alias - api apiProvider - log *plog.Logger + alias alias + api apiProvider + log *plog.Logger + CacheTTL time.Duration + Endpoint string + Secret string } // alias decoupled from sgcp.Alias to allow for future changes @@ -96,6 +108,7 @@ func (t *T) Configure() error { if t.Endpoint == "" { return errors.New("endpoint is required") } + cfg = cfg.WithDNSBaseURL(t.Endpoint) // zoneid is mandatory if t.ZoneID == "" { @@ -110,8 +123,11 @@ func (t *T) Configure() error { func (t *T) configureMgr(cfg *sgcp.Config) error { mgr := &mgr{ - alias: alias{UUID: t.UUID, Name: t.Name, Target: t.Target, ZoneID: t.ZoneID}, - log: t.Log(), + alias: alias{UUID: t.UUID, Name: t.Name, Target: t.Target, ZoneID: t.ZoneID}, + log: t.Log(), + CacheTTL: time.Duration(cfg.Cache.TTLSeconds) * time.Second, + Endpoint: t.Endpoint, + Secret: t.Secret, } if t.api != nil { // allow custom api for tests @@ -125,38 +141,66 @@ func (t *T) configureMgr(cfg *sgcp.Config) error { return fmt.Errorf("failed to create http client: %w", err) } - authInfo, err := sgcphelper.AuthInfoFromPath(t.Secret) + 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) } tokenFactory := sgcp.NewTokenFactory(t.Log(), httpClient, &cfg.Auth, authInfo) - if t.api != nil { - mgr.api = t.api - } else { - mgr.api = sgcp.NewDNSAPI(cfg, httpClient, t.Log(), tokenFactory) - } + mgr.api = sgcp.NewDNSAPI(cfg, httpClient, t.Log(), tokenFactory) t.mgr = mgr return nil } func (t *T) Start(ctx context.Context) error { - // TODO: implement cache cleanup + if disabled, err := t.isDisabled(); err != nil { + return err + } else if disabled { + t.Log().Infof("skip start of alias %s: sgcp support disabled", t.Name) + return nil + } return t.mgr.createOrUpdate(ctx, t.Target) } func (t *T) Stop(ctx context.Context) error { - // TODO: implement cache cleanup + if disabled, err := t.isDisabled(); err != nil { + return err + } else if disabled { + t.Log().Infof("skip stop of alias %s: sgcp support disabled", t.Name) + return nil + } if t.UUID != "" { return t.mgr.createOrUpdate(ctx, t.noneTarget) } return t.mgr.delete(ctx) } +// isDisabled tells whether the operator has disabled the sgcp support. An +// undecidable flag is an error the actions report, rather than a reason to +// quietly skip the work they were asked to do. +func (t *T) isDisabled() (bool, error) { + return sgcp.IsDisabled(rawconfig.NodeVarDir()) +} + func (t *T) Status(ctx context.Context) status.T { - // TODO: implement cache cleanup if command is not called from the scheduler + if disabled, err := t.isDisabled(); err != nil { + t.StatusLog().Warn("%s", err) + return status.NotApplicable + } else if disabled { + t.StatusLog().Info("xaas status disabled") + return status.NotApplicable + } + if sgcphelper.NeedsCacheClear() { + if err := t.mgr.cacheClear(t.mgr.cacheSigGetAliases()); err != nil { + t.Log().Debugf("clear get alias cache failed: %s", err) + t.StatusLog().Warn("possible stale value: clear get alias cache failed") + } + } aliases, err := t.mgr.getAliases(ctx) if err != nil { t.StatusLog().Error("get alias failed: %s", err) diff --git a/drivers/resipsgcp_dnsalias/main_test.go b/drivers/resipsgcp_dnsalias/main_test.go index af59ce53c..dd919204f 100644 --- a/drivers/resipsgcp_dnsalias/main_test.go +++ b/drivers/resipsgcp_dnsalias/main_test.go @@ -3,14 +3,22 @@ package resipsgcp_dnsalias import ( "context" "fmt" + "net/http" + "net/http/httptest" + "os" + "path/filepath" + "strings" + "sync" "testing" "time" "github.com/stretchr/testify/assert" "github.com/stretchr/testify/require" + "github.com/opensvc/om3/v3/core/env" "github.com/opensvc/om3/v3/core/resource" "github.com/opensvc/om3/v3/core/status" + "github.com/opensvc/om3/v3/drivers/sgcpauthtesthelper" "github.com/opensvc/om3/v3/util/sgcpdnstesthelper" "github.com/opensvc/om3/v3/util/testsgcphelper" @@ -42,6 +50,267 @@ func newDBAndDrv(t *testing.T, s string, entries []sgcpdnstesthelper.DBEntry) (* return db, drv } +// TestConfigureEndpoint verifies the endpoint keyword is the api the driver +// talks to. It used to be validated, stored and never applied, every request +// going to the base url of the configuration instead. +func TestConfigureEndpoint(t *testing.T) { + ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second) + defer cancel() + + var mu sync.Mutex + var seen []string + + srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + mu.Lock() + seen = append(seen, r.URL.Path) + mu.Unlock() + w.Header().Set("Content-Type", "application/json") + switch { + case r.URL.Path == "/auth/access_token": + fmt.Fprint(w, `{"access_token": "a-token"}`) + case strings.HasPrefix(r.URL.Path, "/dns/zone/"): + fmt.Fprint(w, `{"cnameRecords": [{"id": "uuid1", "name": "name1", "target": "target1", "zoneId": "z1"}]}`) + default: + w.WriteHeader(http.StatusNotFound) + } + })) + defer srv.Close() + + // The agent root goes to a tempdir, then the configuration is replaced by + // one whose dns base url is a port nothing listens on: reaching the alias + // is only possible through the endpoint keyword. + root := filepath.Dir(testsgcphelper.InstallConfig(t)) + cfgFile := filepath.Join(root, "endpoint.yaml") + require.NoError(t, os.WriteFile(cfgFile, []byte(` +auth: + base_url: "`+srv.URL+`/auth" + default_secret: "the-secret" + scopes: + dns_read: ["dns:read"] + dns_write: ["dns:write"] + timeout: 5 + ttl_seconds: 60 +dns: + base_url: "http://127.0.0.1:1/dns" + none_target: "none.xxx" + path: + cname: "/cname" + zone: "/zone" +cache: + ttl_seconds: 0 +`), 0644)) + sgcp.SetConfigForTest(cfgFile) + defer sgcp.SetConfigForTest("") + + drv := New().(*T) + require.NoError(t, drv.SetRID("rid1")) + drv.authInfoer = sgcpauthtesthelper.NewMockGetAuthInfoProvider("id1") + drv.Name = "name1" + drv.Target = "target1" + drv.ZoneID = "z1" + drv.Endpoint = srv.URL + "/dns" + require.NoError(t, drv.Configure()) + + assert.Equal(t, status.Up, drv.Status(ctx), "the alias was not read through the endpoint") + + mu.Lock() + defer mu.Unlock() + assert.Containsf(t, seen, "/dns/zone/z1/cname", "the endpoint keyword was not used: %v", seen) +} + +// TestConfigureCacheTTL verifies the alias cache ages as the configuration +// says, and not on a duration of the driver's own. +func TestConfigureCacheTTL(t *testing.T) { + defer setup(t)() + + _, drv := newDBAndDrv(t, "rid1", nil) + drv.Name = "name1" + drv.Target = "target1" + drv.ZoneID = "z1" + require.NoError(t, drv.Configure()) + + expected := time.Duration(sgcp.GetConfig().Cache.TTLSeconds) * time.Second + require.NotZerof(t, expected, "the test configuration has to set a cache ttl") + assert.Equal(t, expected, drv.mgr.CacheTTL) +} + +// TestStatus_ReadsTheProviderOutsideTheScheduler verifies every origin but the +// scheduler reads the api, the scheduler alone being served the cache. +func TestStatus_ReadsTheProviderOutsideTheScheduler(t *testing.T) { + ctx, cancel := context.WithTimeout(context.Background(), time.Second) + defer cancel() + + entries := []sgcpdnstesthelper.DBEntry{ + { + Name: "name1", UUID: "uuid1", ZoneID: "z1", + AliasL: []sgcp.Alias{{UUID: "uuid1", Name: "name1", Target: "target1", ZoneID: "z1"}}, + }, + } + newDrv := func(t *testing.T) (*sgcpdnstesthelper.DB, *T) { + db, drv := newDBAndDrv(t, "rid1", entries) + drv.UUID = "uuid1" + drv.Name = "name1" + drv.Target = "target1" + drv.ZoneID = "z1" + require.NoError(t, drv.Configure()) + require.NotZerof(t, drv.mgr.CacheTTL, "the test needs the cache enabled") + return db, drv + } + + for _, origin := range []env.ActionOrigin{ + env.ActionOriginUser, + env.ActionOriginDaemonMonitor, + env.ActionOriginDaemonAPI, + } { + t.Run(string(origin)+" reads the api every time", func(t *testing.T) { + defer setup(t)() + t.Setenv(env.ActionOriginVar, string(origin)) + db, drv := newDrv(t) + + for i := 0; i < 3; i++ { + assert.Equal(t, status.Up, drv.Status(ctx)) + } + assert.Equal(t, 3, db.CallCounts().Search, "a status evaluation was served by the cache") + }) + } + + t.Run("the scheduler is served the cache", func(t *testing.T) { + defer setup(t)() + t.Setenv(env.ActionOriginVar, string(env.ActionOriginDaemonScheduler)) + db, drv := newDrv(t) + + for i := 0; i < 3; i++ { + assert.Equal(t, status.Up, drv.Status(ctx)) + } + assert.Equal(t, 1, db.CallCounts().Search, "the cache did not serve the repeated evaluations") + }) +} + +// TestCacheInvalidation verifies a started alias is not read back from the +// ageing cache the status evaluation filled before the change. The two drivers +// stand for the two processes an om run uses: the one that starts the resource +// and the one that evaluates its status afterwards. +func TestCacheInvalidation(t *testing.T) { + defer setup(t)() + // The scheduler origin: every other status reads the provider instead + // of the cache, and would not show the staleness this test is about. + t.Setenv(env.ActionOriginVar, string(env.ActionOriginDaemonScheduler)) + + ctx, cancel := context.WithTimeout(context.Background(), time.Second) + defer cancel() + + db := sgcpdnstesthelper.NewDB() + db.Setup([]sgcpdnstesthelper.DBEntry{ + { + Name: "name1", + UUID: "uuid1", + ZoneID: "z1", + AliasL: []sgcp.Alias{ + {UUID: "uuid1", Name: "name1", Target: "target1", FQDN: "name1.z1", ZoneID: "z1"}, + }, + }, + }) + + // No uuid keyword: the alias is looked up by name, and the api completes + // its identity. The cache signature thus differs before and after a start. + newDrv := func(rid string) *T { + drv := New().(*T) + require.NoError(t, drv.SetRID(rid)) + drv.api = sgcpdnstesthelper.NewApi(db) + drv.Name = "name1" + drv.Target = "target2" + drv.ZoneID = "z1" + require.NoError(t, drv.Configure()) + require.NotZerof(t, drv.mgr.CacheTTL, "the test needs the cache enabled") + return drv + } + + starter := newDrv("rid1") + t.Log("evaluate the status first, so the aliases are cached") + require.Equal(t, status.Down, starter.Status(ctx)) + + require.NoError(t, starter.Start(ctx)) + alias, ok := db.Search("z1", "name1", "uuid1") + require.True(t, ok) + require.Equal(t, "target2", alias.Target, "the alias was not retargeted") + + t.Log("another driver sees the change, not the entry cached before it") + assert.Equal(t, status.Up, newDrv("rid2").Status(ctx)) +} + +// TestDisabled verifies the actions honor the disabled flag of the sgcp +// configuration, and that they report a flag they can not decide about +// instead of taking it for a disable. +func TestDisabled(t *testing.T) { + ctx, cancel := context.WithTimeout(context.Background(), time.Second) + defer cancel() + + dbEntries := []sgcpdnstesthelper.DBEntry{ + { + Name: "name1", + UUID: "uuid1", + ZoneID: "z1", + AliasL: []sgcp.Alias{ + {UUID: "uuid1", Name: "name1", Target: "target1", ZoneID: "z1"}, + }, + }, + } + + newDrv := func(t *testing.T) (*sgcpdnstesthelper.DB, *T) { + db, drv := newDBAndDrv(t, "rid1", dbEntries) + drv.UUID = "uuid1" + drv.Name = "name1" + drv.Target = "target2" + drv.ZoneID = "z1" + require.NoError(t, drv.Configure()) + drv.mgr.CacheTTL = 0 + db.ResetCalls() + return db, drv + } + + t.Run("the actions are skipped when the flag exists", func(t *testing.T) { + defer setup(t)() + require.NoError(t, os.WriteFile(sgcp.GetConfig().DisabledFlag, nil, 0644)) + db, drv := newDrv(t) + + require.NoError(t, drv.Start(ctx)) + require.NoError(t, drv.Stop(ctx)) + assert.Equal(t, status.NotApplicable, drv.Status(ctx)) + + call := db.CallCounts() + assert.Zerof(t, call.Update, "the api was called on a disabled node: %+v", call) + assert.Zerof(t, call.Delete, "the api was called on a disabled node: %+v", call) + assert.Zerof(t, call.Search, "the api was called on a disabled node: %+v", call) + }) + + t.Run("the actions run when the flag is absent", func(t *testing.T) { + defer setup(t)() + db, drv := newDrv(t) + + require.NoError(t, drv.Start(ctx)) + assert.NotZerof(t, db.CallCounts().Update, "the alias was not updated") + }) + + t.Run("an undecidable flag is reported", func(t *testing.T) { + defer setup(t)() + db, drv := newDrv(t) + + // A regular file as the flag parent directory: stat then fails with + // something else than "does not exist", whatever the user running + // the test. + notADir := filepath.Join(t.TempDir(), "file") + require.NoError(t, os.WriteFile(notADir, nil, 0644)) + cfgFile := filepath.Join(t.TempDir(), "sgcp.yaml") + require.NoError(t, os.WriteFile(cfgFile, []byte("disabled_flag: "+filepath.Join(notADir, "flag")+"\n"), 0644)) + sgcp.SetConfigForTest(cfgFile) + + require.Error(t, drv.Start(ctx)) + require.Error(t, drv.Stop(ctx)) + assert.Equal(t, status.NotApplicable, drv.Status(ctx)) + assert.Zerof(t, db.CallCounts().Update, "the api was called with an undecidable flag") + }) +} + func TestStatus(t *testing.T) { cleanup := setup(t) defer cleanup() @@ -249,6 +518,7 @@ func TestStatus(t *testing.T) { drv.Target = tc.resTarget drv.ZoneID = tc.resZoneID require.NoError(t, drv.Configure()) + drv.mgr.CacheTTL = 0 dStatus := drv.Status(ctx) assert.Equalf(t, tc.expectedStatus, dStatus, "expected %s, got %s", tc.expectedStatus, dStatus) @@ -300,6 +570,7 @@ func TestStart(t *testing.T) { drv.Target = "foo-target" drv.ZoneID = "z1" require.NoError(t, drv.Configure()) + drv.mgr.CacheTTL = 0 t.Log("verify alias doesn't exits") alias, ok := db.Search("z1", "foo", "") @@ -334,6 +605,7 @@ func TestStart(t *testing.T) { drv.Target = "target1" drv.ZoneID = "z1" require.NoError(t, drv.Configure()) + drv.mgr.CacheTTL = 0 t.Log("verify alias initially exits") alias, ok := db.Search("z1", "name1", "uuid1") @@ -362,6 +634,7 @@ func TestStart(t *testing.T) { drv.Target = "newTarget2" drv.ZoneID = "z1" require.NoError(t, drv.Configure()) + drv.mgr.CacheTTL = 0 t.Log("verify alias exits initially, with alternate target") alias, ok := db.Search("z1", "name2", "uuid2") @@ -452,6 +725,7 @@ func TestStop(t *testing.T) { drv.Target = "target" drv.ZoneID = "z1" require.NoError(t, drv.Configure()) + drv.mgr.CacheTTL = 0 t.Log("verify alias doesn't exits") _, ok := db.Search(drv.ZoneID, drv.Name, drv.UUID) @@ -477,6 +751,7 @@ func TestStop(t *testing.T) { drv.Target = "target" drv.ZoneID = "z1" require.NoError(t, drv.Configure()) + drv.mgr.CacheTTL = 0 t.Log("verify alias doesn't exits") _, ok := db.Search(drv.ZoneID, drv.Name, drv.UUID) @@ -500,6 +775,7 @@ func TestStop(t *testing.T) { drv.Target = "target1" drv.ZoneID = "z1" require.NoError(t, drv.Configure()) + drv.mgr.CacheTTL = 0 t.Log("verify initial exits") alias, ok := db.Search(drv.ZoneID, drv.Name, drv.UUID) @@ -531,6 +807,7 @@ func TestStop(t *testing.T) { drv.Target = "none.xxx" drv.ZoneID = "z1" require.NoError(t, drv.Configure()) + drv.mgr.CacheTTL = 0 t.Log("verify initial exits with target none") alias, ok := db.Search(drv.ZoneID, drv.Name, drv.UUID) @@ -561,6 +838,7 @@ func TestStop(t *testing.T) { drv.Target = "target1" drv.ZoneID = "z1" require.NoError(t, drv.Configure()) + drv.mgr.CacheTTL = 0 t.Log("verify initial exits") initial, ok := db.Search(drv.ZoneID, drv.Name, drv.UUID) diff --git a/drivers/resipsgcp_dnsalias/mgr.go b/drivers/resipsgcp_dnsalias/mgr.go index e087c61c6..5f8feceaf 100644 --- a/drivers/resipsgcp_dnsalias/mgr.go +++ b/drivers/resipsgcp_dnsalias/mgr.go @@ -2,10 +2,12 @@ package resipsgcp_dnsalias import ( "context" + "crypto/sha256" "encoding/json" "fmt" "net/http" + "github.com/opensvc/om3/v3/util/ageingcache" "github.com/opensvc/om3/v3/util/sgcp" ) @@ -14,6 +16,12 @@ var ( ) func (m *mgr) createOrUpdate(ctx context.Context, target string) error { + // The signature the read below caches under. The alias identity changes + // on the way when the create assigns the uuid the keyword did not carry, + // so the entry to drop afterwards is this one, not the one the alias + // computes once updated: nothing was ever cached under that one. + sig := m.cacheSigGetAliases() + aliases, err := m.getAliases(ctx) if err != nil { return fmt.Errorf("get aliases: %w", err) @@ -27,6 +35,9 @@ func (m *mgr) createOrUpdate(ctx context.Context, target string) error { return fmt.Errorf("create alias: %w", err) } else { m.alias = *v + if err := m.cacheClear(sig); err != nil { + m.log.Debugf("cache clear error: %s", err) + } return nil } } @@ -51,11 +62,16 @@ func (m *mgr) createOrUpdate(ctx context.Context, target string) error { return fmt.Errorf("update alias unexpected nil") } else { m.alias = *v + if err := m.cacheClear(sig); err != nil { + m.log.Debugf("cache clear error: %s", err) + } return nil } } func (m *mgr) delete(ctx context.Context) error { + sig := m.cacheSigGetAliases() + aliases, err := m.getAliases(ctx) if err != nil { return fmt.Errorf("get aliases: %w", err) @@ -72,20 +88,37 @@ func (m *mgr) delete(ctx context.Context) error { if err := m.api.DeleteAlias(ctx, alias.ZoneID, alias.UUID); err != nil { return fmt.Errorf("delete alias: %w", err) } + if err := m.cacheClear(sig); err != nil { + m.log.Debugf("cache clear error: %s", err) + } return nil } // getAliases retrieves a list of aliases for the specified zone, name, and UUID or returns an error if unsuccessful. func (m *mgr) getAliases(ctx context.Context) ([]sgcp.Alias, error) { - // TODO: Use ageing cache - method, url, code, data, err := m.api.GetAliases(ctx, m.alias.ZoneID, m.alias.Name, m.alias.UUID) - if err != nil { - return nil, err + if m.CacheTTL <= 0 { + data, err := m.getAliasesFactory(ctx)() + if err != nil { + return nil, err + } + if data == nil || string(data) == "null" { + return nil, nil + } + var resp aliasListResponse + if err := json.Unmarshal(data, &resp); err != nil { + return nil, fmt.Errorf("decode aliases: %w", err) + } + return resp.CnameRecords, nil } - if err := m.api.CheckStatusCode(method, url, code, http.StatusOK, http.StatusNotFound); err != nil { + + o := ageingcache.NewOutputter(m.getAliasesFactory(ctx)) + sig := m.cacheSigGetAliases() + data, err := ageingcache.Output(o, sig, m.CacheTTL) + if err != nil { + m.log.Debugf("getAliases cache miss: %s", err) return nil, err } - if code == http.StatusNotFound { + if data == nil || string(data) == "null" { return nil, nil } var resp aliasListResponse @@ -95,7 +128,22 @@ func (m *mgr) getAliases(ctx context.Context) ([]sgcp.Alias, error) { return resp.CnameRecords, nil } -// create creates a new alias with the specified target and returns the created alias or an error if the operation fails. +func (m *mgr) getAliasesFactory(ctx context.Context) func() ([]byte, error) { + return func() ([]byte, error) { + method, url, code, data, err := m.api.GetAliases(ctx, m.alias.ZoneID, m.alias.Name, m.alias.UUID) + if err != nil { + return nil, err + } + if err := m.api.CheckStatusCode(method, url, code, http.StatusOK, http.StatusNotFound); err != nil { + return nil, err + } + if code == http.StatusNotFound { + return []byte("null"), nil + } + return data, nil + } +} + func (m *mgr) create(ctx context.Context, target string) (*alias, error) { v, err := m.api.CreateAlias(ctx, m.alias.ZoneID, m.alias.Name, target) if err != nil { @@ -104,7 +152,6 @@ func (m *mgr) create(ctx context.Context, target string) (*alias, error) { return toAlias(v), nil } -// update modifies an existing alias with the specified parameters and returns the updated alias or an error if any occurs. func (m *mgr) update(ctx context.Context, zoneID, aliasUUID, aliasName, target string) (*alias, error) { v, err := m.api.UpdateAlias(ctx, zoneID, aliasUUID, aliasName, target) if err != nil { @@ -113,7 +160,6 @@ func (m *mgr) update(ctx context.Context, zoneID, aliasUUID, aliasName, target s return toAlias(v), nil } -// toAlias converts a sgcp.Alias object to an alias object by mapping corresponding fields. func toAlias(v *sgcp.Alias) *alias { return &alias{ UUID: v.UUID, @@ -124,8 +170,6 @@ func toAlias(v *sgcp.Alias) *alias { } } -// Equal compares two alias objects and returns true if they are equal, or false otherwise. -// It doesn't compare the FQDN field. func (a *alias) Equal(b *alias) bool { if a == nil && b == nil { return true @@ -138,3 +182,36 @@ func (a *alias) Equal(b *alias) bool { a.Target == b.Target && a.ZoneID == b.ZoneID } + +// cacheSigGetAliases generates the cache signature specific to retrieving +// alias data based on predefined constants. +func (m *mgr) cacheSigGetAliases() string { + sig := m.cacheSig("get-aliases") + m.log.Debugf("cacheSigGetAliases %s: %s", m.alias.Name, sig) + return sig +} + +// cacheSig generates a unique cache signature for by hashing a formatted +// string combining endpoint, secret, and alias details. +func (m *mgr) cacheSig(s string) string { + data := fmt.Sprintf("%s|%s|%s|%s|%s", + m.Endpoint, + m.Secret, + m.alias.ZoneID, + m.alias.Name, + m.alias.UUID, + ) + hash := sha256.Sum256([]byte(data)) + return fmt.Sprintf("sgcp-dnsalias-%s-%x", s, hash) +} + +// cacheClear drops the cached aliases read under sig. The caller captures +// that signature before the alias it names is changed, so an alias whose +// identity the api completed is still invalidated where it was cached. +func (m *mgr) cacheClear(sig string) error { + if m.CacheTTL <= 0 { + return nil + } + m.log.Debugf("clear alias cache") + return ageingcache.Clear(sig) +} diff --git a/drivers/sgcphelper/main.go b/drivers/sgcphelper/main.go index bdec5b753..17050ed11 100644 --- a/drivers/sgcphelper/main.go +++ b/drivers/sgcphelper/main.go @@ -3,11 +3,24 @@ package sgcphelper import ( "fmt" + "github.com/opensvc/om3/v3/core/env" "github.com/opensvc/om3/v3/core/naming" "github.com/opensvc/om3/v3/core/object" "github.com/opensvc/om3/v3/util/sgcp" ) +// NeedsCacheClear tells whether a driver has to drop what it cached and read +// the provider again. Only the daemon scheduler is served the cache: its +// status evaluations run over and over on their own, and the cache is what +// keeps them off the provider api. Everyone else, an operator asking for a +// status first of all, is asking what the provider says now. +// +// This lives here rather than in util/sgcp because it reads the action +// origin, and a util package does not depend on core. +func NeedsCacheClear() bool { + return !env.HasDaemonSchedulerOrigin() +} + type ( GetAuthInfoFromDatastorePather struct{} ) diff --git a/util/sgcp/api.go b/util/sgcp/api.go index c9dbcc4e3..479b5b5d1 100644 --- a/util/sgcp/api.go +++ b/util/sgcp/api.go @@ -30,6 +30,13 @@ func (a *Api) CheckStatusCode(method, url string, got int, wanted ...int) error return fmt.Errorf("unexpected status code for %s %s got %d wanted %v", method, url, got, wanted) } +// do executes the request and reports the response status code and body. +// +// The status code is an answer, not a failure: a 404 saying the filesystem is +// gone, or a 412 saying the consistency group is busy, is information the +// callers act on. So err is reserved for what prevents an answer altogether +// (token, request build, transport, body read), and each caller declares the +// status codes it accepts, with CheckStatusCode. func (a *Api) do(ctx context.Context, method, url string, body io.Reader, scopes ...string) (statusCode int, b []byte, err error) { var req *http.Request var resp *http.Response @@ -57,11 +64,10 @@ func (a *Api) do(ctx context.Context, method, url string, body io.Reader, scopes defer func() { _ = resp.Body.Close() }() a.log.Debugf("request: %s %s status code: %d", method, url, resp.StatusCode) - if resp.StatusCode >= 400 { - body, _ := io.ReadAll(resp.Body) - return resp.StatusCode, nil, fmt.Errorf("HTTP %d: %s", resp.StatusCode, string(body)) - } b, err = io.ReadAll(resp.Body) + if err != nil { + return resp.StatusCode, nil, fmt.Errorf("read %s %s response body: %w", method, url, err) + } - return resp.StatusCode, b, err + return resp.StatusCode, b, nil } diff --git a/util/sgcp/api_test.go b/util/sgcp/api_test.go index 22bbd4cd2..2f422e2b1 100644 --- a/util/sgcp/api_test.go +++ b/util/sgcp/api_test.go @@ -3,6 +3,7 @@ package sgcp import ( "context" "encoding/json" + "fmt" "net/http" "net/http/httptest" "strings" @@ -84,3 +85,42 @@ func TestCheckStatusCode(t *testing.T) { err = a.CheckStatusCode(http.MethodGet, "https://localhost:1215/foo", 201, 200, 201) assert.Nil(t, err) } + +// TestDoErrorStatusIsNotAnError verifies do hands back the status code and the +// response body of a failed request without an error. The status is an answer: +// the callers branch on a 404 or a 412, and an error here would shadow them, +// as every caller tests the error first. +func TestDoErrorStatusIsNotAnError(t *testing.T) { + defer Setup(t)() + + for _, wanted := range []int{ + http.StatusNotFound, + http.StatusPreconditionFailed, + http.StatusInternalServerError, + } { + t.Run(fmt.Sprint(wanted), func(t *testing.T) { + handler := http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + w.Header().Set("Content-Type", "application/json") + w.WriteHeader(wanted) + _, _ = w.Write([]byte(`{"message":"nope"}`)) + }) + ts := httptest.NewTLSServer(handler) + defer ts.Close() + + api := Api{ + client: ts.Client(), + log: plog.NewDefaultLogger(), + tk: &TTkBuilder{}, + } + + code, data, err := api.do(context.Background(), "GET", ts.URL+"/foo/bar", nil, "scope1") + require.NoError(t, err) + assert.Equal(t, wanted, code) + assert.Equal(t, `{"message":"nope"}`, string(data), "the error body reached the caller") + + // The caller is the one turning an unwanted status into an error. + assert.Error(t, api.CheckStatusCode("GET", ts.URL+"/foo/bar", code, http.StatusOK)) + assert.NoError(t, api.CheckStatusCode("GET", ts.URL+"/foo/bar", code, http.StatusOK, wanted)) + }) + } +} diff --git a/util/sgcp/config.go b/util/sgcp/config.go index 00107160e..8499e4d94 100644 --- a/util/sgcp/config.go +++ b/util/sgcp/config.go @@ -28,6 +28,9 @@ type ( Client string `yaml:"client"` CG string `yaml:"cg"` } `yaml:"path"` + CG struct { + Timeout int `yaml:"timeout"` + } `yaml:"cg"` } // DNSConfig contains DNS-related API configuration @@ -49,10 +52,10 @@ type ( Timeout int `yaml:"timeout"` } - // CacheConfig contains caching configuration + // CacheConfig contains caching configuration. A zero ttl_seconds + // disables the caching. CacheConfig struct { - TTLSeconds int `yaml:"ttl_seconds"` - Enabled bool `yaml:"enabled"` + TTLSeconds int `yaml:"ttl_seconds"` } ) diff --git a/util/sgcp/config_test.go b/util/sgcp/config_test.go index 95128ded6..79a1808c2 100644 --- a/util/sgcp/config_test.go +++ b/util/sgcp/config_test.go @@ -80,5 +80,4 @@ func TestLoadConfig(t *testing.T) { // Test cache configuration assert.Equal(t, 14400, cfg.Cache.TTLSeconds) - assert.True(t, cfg.Cache.Enabled) } diff --git a/util/sgcp/dns.go b/util/sgcp/dns.go index 11b836489..a747b7f64 100644 --- a/util/sgcp/dns.go +++ b/util/sgcp/dns.go @@ -64,7 +64,7 @@ func (a *DNSAPI) CreateAlias(ctx context.Context, zoneID, name, target string) ( if code, data, err := a.do(ctx, method, path, bytes.NewReader(b), a.GetScopes("dns_write")...); err != nil { return nil, fmt.Errorf("%s %s: %w", method, path, err) } else if err := a.CheckStatusCode(method, path, code, http.StatusCreated); err != nil { - return nil, err + return nil, fmt.Errorf("%w: '%s'", err, string(data)) } else if err := json.Unmarshal(data, &result); err != nil { return nil, fmt.Errorf("%s %s unmarshal created alias: %w", method, path, err) } @@ -87,7 +87,7 @@ func (a *DNSAPI) UpdateAlias(ctx context.Context, zoneID, aliasUUID, name, targe if code, data, err := a.do(ctx, method, path, bytes.NewReader(b), a.GetScopes("dns_write")...); err != nil { return nil, err } else if err := a.CheckStatusCode(method, path, code, http.StatusOK); err != nil { - return nil, err + return nil, fmt.Errorf("%w: '%s'", err, string(data)) } else if err := json.Unmarshal(data, &alias); err != nil { return nil, fmt.Errorf("failed to unmarshal alias: %w", err) } @@ -100,10 +100,10 @@ func (a *DNSAPI) DeleteAlias(ctx context.Context, zoneID, aliasUUID string) erro path := a.getAliasURL(zoneID, aliasUUID) a.log.Infof("%s %s", method, path) - if code, _, err := a.do(ctx, method, path, nil, a.GetScopes("dns_write")...); err != nil { + if code, data, err := a.do(ctx, method, path, nil, a.GetScopes("dns_write")...); err != nil { return err } else if err := a.CheckStatusCode(method, path, code, http.StatusNoContent); err != nil { - return err + return fmt.Errorf("%w: '%s'", err, string(data)) } return nil } 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/sgcp/main.go b/util/sgcp/main.go index c2ef5c594..fd089abce 100644 --- a/util/sgcp/main.go +++ b/util/sgcp/main.go @@ -3,12 +3,35 @@ package sgcp import ( - "github.com/opensvc/om3/v3/util/file" + "errors" + "fmt" + "os" + "path/filepath" ) -func IsDisabled(s string) bool { - if config.DisabledFlag == "" { - return false +// IsDisabled returns true when the sgcp support is administratively disabled, +// which an operator declares by creating the disabled_flag file named by the +// configuration. A relative flag path is resolved against dir. +// +// A flag that can be neither confirmed present nor absent, because of a +// permission error on a parent directory or an io error, is reported as an +// error and not as a disable. The callers act on that error, where reading it +// as "disabled" would have them quietly skip the work they were asked to do. +func IsDisabled(dir string) (bool, error) { + cfg := GetConfig() + if cfg == nil || cfg.DisabledFlag == "" { + return false, nil + } + path := cfg.DisabledFlag + if !filepath.IsAbs(path) { + path = filepath.Join(dir, path) + } + switch _, err := os.Stat(path); { + case err == nil: + return true, nil + case errors.Is(err, os.ErrNotExist): + return false, nil + default: + return false, fmt.Errorf("stat the sgcp disabled flag: %w", err) } - return file.Exists(config.DisabledFlag) } diff --git a/util/sgcp/main_test.go b/util/sgcp/main_test.go new file mode 100644 index 000000000..83f70b5d8 --- /dev/null +++ b/util/sgcp/main_test.go @@ -0,0 +1,75 @@ +package sgcp + +import ( + "os" + "path/filepath" + "testing" + + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" +) + +// installConfigWithDisabledFlag writes a minimal configuration naming flag as +// its disabled_flag, and makes it the configuration for the test duration. +func installConfigWithDisabledFlag(t *testing.T, flag string) { + t.Helper() + cfgFile := filepath.Join(t.TempDir(), "sgcp.yaml") + require.NoError(t, os.WriteFile(cfgFile, []byte("disabled_flag: "+flag+"\n"), 0644)) + SetConfigForTest(cfgFile) + t.Cleanup(func() { SetConfigForTest("") }) +} + +func TestIsDisabled(t *testing.T) { + t.Run("no flag configured", func(t *testing.T) { + installConfigWithDisabledFlag(t, "") + + disabled, err := IsDisabled(t.TempDir()) + require.NoError(t, err) + assert.False(t, disabled) + }) + + t.Run("flag absent", func(t *testing.T) { + dir := t.TempDir() + installConfigWithDisabledFlag(t, filepath.Join(dir, "sgcp_disabled")) + + disabled, err := IsDisabled(dir) + require.NoError(t, err) + assert.False(t, disabled) + }) + + t.Run("flag present", func(t *testing.T) { + dir := t.TempDir() + flag := filepath.Join(dir, "sgcp_disabled") + require.NoError(t, os.WriteFile(flag, nil, 0644)) + installConfigWithDisabledFlag(t, flag) + + disabled, err := IsDisabled(dir) + require.NoError(t, err) + assert.True(t, disabled) + }) + + t.Run("relative flag is resolved against dir", func(t *testing.T) { + dir := t.TempDir() + require.NoError(t, os.WriteFile(filepath.Join(dir, "sgcp_disabled"), nil, 0644)) + installConfigWithDisabledFlag(t, "sgcp_disabled") + + disabled, err := IsDisabled(dir) + require.NoError(t, err) + assert.True(t, disabled) + + disabled, err = IsDisabled(t.TempDir()) + require.NoError(t, err) + assert.False(t, disabled) + }) + + t.Run("an undecidable flag is an error, not a disable", func(t *testing.T) { + dir := t.TempDir() + notADir := filepath.Join(dir, "file") + require.NoError(t, os.WriteFile(notADir, nil, 0644)) + installConfigWithDisabledFlag(t, filepath.Join(notADir, "sgcp_disabled")) + + disabled, err := IsDisabled(dir) + require.Error(t, err) + assert.False(t, disabled) + }) +} diff --git a/util/sgcpcgtesthelper/main.go b/util/sgcpcgtesthelper/main.go new file mode 100644 index 000000000..07dd6eef2 --- /dev/null +++ b/util/sgcpcgtesthelper/main.go @@ -0,0 +1,266 @@ +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 and PatchFailoverFunc inject a failure: returning + // an error makes the operation fail and leaves the group alone, and the + // group moves as it would otherwise. + PatchSwitchoverFunc func(ctx context.Context, uuid, targetAZ string) error + PatchFailoverFunc func(ctx context.Context, uuid, targetAZ string) error + + // PatchResumeFunc decides the outcome of a resume instead of injecting + // one: what it writes through Update() is what the group becomes, and + // the error it returns fails the operation without touching the group. + 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() + + // PatchResumeFunc says what the group becomes, the test writing it + // through Update(). A hook that fails leaves the group as it was, + // which is what a provider refusing the operation does. + if a.PatchResumeFunc != nil { + if err := a.PatchResumeFunc(ctx, uuid); err != nil { + return http.MethodPatch, "/consistency-groups/" + uuid, http.StatusInternalServerError, nil, err + } + return http.MethodPatch, "/consistency-groups/" + uuid, http.StatusAccepted, nil, nil + } + + // Without a hook, the targets in the group's own availability zone + // replicate again, and the group is ready. + 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 +} diff --git a/util/sgcpserverfortest/main.go b/util/sgcpserverfortest/main.go index b463a5972..99c57ee59 100644 --- a/util/sgcpserverfortest/main.go +++ b/util/sgcpserverfortest/main.go @@ -16,12 +16,10 @@ import ( "github.com/google/uuid" - "github.com/opensvc/om3/v3/util/sgcp" "github.com/opensvc/om3/v3/util/sgcpdnstesthelper" ) type ( - // NfsClient represents an NFS client configuration NfsClient struct { UUID string `json:"uuid"` Host string `json:"host"` @@ -30,7 +28,6 @@ type ( ConsistencyGroupID string `json:"consistencyGroupId,omitempty"` } - // FilesystemInfo represents the filesystem information from the API FilesystemInfo struct { UUID string `json:"uuid"` ConsistencyGroupID string `json:"consistencyGroupId"` @@ -55,7 +52,7 @@ var ( users *Users files = map[string]FilesystemInfo{ - "1ab7d139-dd35-4f9c-ad82-cd6a93675cfd": FilesystemInfo{ + "1ab7d139-dd35-4f9c-ad82-cd6a93675cfd": { UUID: "1ab7d139-dd35-4f9c-ad82-cd6a93675cfd", ConsistencyGroupID: "12", NFSClients: nil, @@ -64,6 +61,9 @@ var ( } createdTokenCount atomic.Int64 + + dnsDB *sgcpdnstesthelper.DB + dnsApi *sgcpdnstesthelper.Api ) func assertAuth(desc string, w http.ResponseWriter, r *http.Request, scope ...string) bool { @@ -87,6 +87,7 @@ func setHeader(w http.ResponseWriter, desc string, code int) { w.WriteHeader(code) } +// ========== File handlers ========== func getFileAPI(w http.ResponseWriter, r *http.Request) { id := r.PathValue("id") slog.Info(GetFile, "id", id) @@ -99,40 +100,34 @@ func getFileAPI(w http.ResponseWriter, r *http.Request) { return } setHeader(w, GetFile, http.StatusNotFound) - return } + func getFileClient(w http.ResponseWriter, r *http.Request) { id := r.PathValue("id") slog.Info(GetFileClient, "id", id) if !assertAuth(GetFileClient, w, r, "account1:sgcp:files:read") { return } - if v, ok := files[id]; ok { setHeader(w, GetFileClient, http.StatusOK) json.NewEncoder(w).Encode(v.NFSClients) return } setHeader(w, GetFileClient, http.StatusNotFound) - - return } + func postFileClient(w http.ResponseWriter, r *http.Request) { id := r.PathValue("id") slog.Info(PostFileClient, "id", id) if !assertAuth(PostFileClient, w, r, "account1:sgcp:files:read", "account1:sgcp:files:write") { return } - var body NfsClient - err := json.NewDecoder(r.Body).Decode(&body) - - if err != nil { + if err := json.NewDecoder(r.Body).Decode(&body); err != nil { logStatusCode(PostFileClient, http.StatusBadRequest) http.Error(w, fmt.Sprintf("invalid JSON: %s", err), http.StatusBadRequest) return } - if v, ok := files[id]; ok { body.UUID = uuid.New().String() v.NFSClients = append(v.NFSClients, body) @@ -142,11 +137,10 @@ func postFileClient(w http.ResponseWriter, r *http.Request) { slog.Info("Created client", "id", id, "clientID", body.UUID) return } - logStatusCode(PostFileClient, http.StatusNotFound) - http.Error(w, fmt.Sprintf("no such fs %s", err), http.StatusNotFound) - return + http.Error(w, fmt.Sprintf("no such fs %s", id), http.StatusNotFound) } + func deleteFileClient(w http.ResponseWriter, r *http.Request) { id := r.PathValue("id") clientID := r.PathValue("clientID") @@ -154,7 +148,6 @@ func deleteFileClient(w http.ResponseWriter, r *http.Request) { if !assertAuth(DeleteFileClient, w, r, "account1:sgcp:files:read", "account1:sgcp:files:write") { return } - if v, ok := files[id]; ok { l := make([]NfsClient, 0) for _, c := range v.NFSClients { @@ -167,35 +160,132 @@ func deleteFileClient(w http.ResponseWriter, r *http.Request) { setHeader(w, DeleteFileClient, http.StatusNoContent) return } - logStatusCode(DeleteFileClient, http.StatusNotFound) http.Error(w, fmt.Sprintf("no such fs %s", id), http.StatusNotFound) - return } -func dnsGetAliasHandler(a *sgcpdnstesthelper.Api) http.HandlerFunc { - return func(w http.ResponseWriter, r *http.Request) { - zoneID := r.PathValue("zoneID") - query := r.URL.Query() - name := query.Get("name") - id := query.Get("id") - - slog.Info(GetDnsAlias, "zoneID", zoneID, "cnameID", id, "name", name) - alias, ok := a.DB.Search(zoneID, name, id) - if !ok { - logStatusCode(GetDnsAlias, http.StatusNotFound) - http.Error(w, fmt.Sprintf("no such alias %s", id), http.StatusNotFound) - return - } - setHeader(w, GetDnsAlias, http.StatusOK) - // TODO: verify mapping - body := map[string]sgcp.Alias{ - "alias": *alias, - } - json.NewEncoder(w).Encode(body) +// ========== DNS handlers (new API) ========== +func dnsListAliases(w http.ResponseWriter, r *http.Request) { + zoneID := r.PathValue("zoneID") + name := r.URL.Query().Get("name") + id := r.URL.Query().Get("id") + + slog.Info(ListDnsAliases, "zoneID", zoneID, "name", name, "id", id) + if !assertAuth(ListDnsAliases, w, r, "account1:sgcp:dns:read") { + return + } + _, _, code, data, err := dnsApi.GetAliases(r.Context(), zoneID, name, id) + if err != nil { + http.Error(w, err.Error(), http.StatusInternalServerError) + return + } + w.WriteHeader(code) + if data != nil { + w.Write(data) + } +} + +func dnsCreateAlias(w http.ResponseWriter, r *http.Request) { + zoneID := r.PathValue("zoneID") + slog.Info(CreateDnsAlias, "zoneID", zoneID) + if !assertAuth(CreateDnsAlias, w, r, "account1:sgcp:dns:read", "account1:sgcp:dns:write") { + return + } + var payload struct { + Name string `json:"name"` + Target string `json:"target"` + TTL int `json:"ttl"` + } + if err := json.NewDecoder(r.Body).Decode(&payload); err != nil { + http.Error(w, fmt.Sprintf("invalid JSON: %s", err), http.StatusBadRequest) + return } + alias, err := dnsApi.CreateAlias(r.Context(), zoneID, payload.Name, payload.Target) + if err != nil { + http.Error(w, err.Error(), http.StatusInternalServerError) + return + } + w.WriteHeader(http.StatusCreated) + json.NewEncoder(w).Encode(alias) } +func dnsGetAlias(w http.ResponseWriter, r *http.Request) { + zoneID := r.PathValue("zoneID") + id := r.PathValue("id") + slog.Info(GetDnsAlias, "zoneID", zoneID, "id", id) + if !assertAuth(GetDnsAlias, w, r, "account1:sgcp:dns:read") { + return + } + // Use search by id + a, ok := dnsApi.DB.Search(zoneID, "", id) + if !ok { + w.WriteHeader(http.StatusNotFound) + return + } + w.WriteHeader(http.StatusOK) + json.NewEncoder(w).Encode(a) +} + +func dnsUpdateAlias(w http.ResponseWriter, r *http.Request) { + zoneID := r.PathValue("zoneID") + id := r.PathValue("id") + slog.Info(UpdateDnsAlias, "zoneID", zoneID, "id", id) + if !assertAuth(UpdateDnsAlias, w, r, "account1:sgcp:dns:read", "account1:sgcp:dns:write") { + return + } + var payload struct { + Target string `json:"target"` + TTL int `json:"ttl"` + } + if err := json.NewDecoder(r.Body).Decode(&payload); err != nil { + http.Error(w, fmt.Sprintf("invalid JSON: %s", err), http.StatusBadRequest) + return + } + // name is ignored; the DB entry is found by id + updated, err := dnsApi.UpdateAlias(r.Context(), zoneID, id, "", payload.Target) + if err != nil { + http.Error(w, err.Error(), http.StatusNotFound) + return + } + w.WriteHeader(http.StatusOK) + json.NewEncoder(w).Encode(updated) +} + +func dnsDeleteAlias(w http.ResponseWriter, r *http.Request) { + zoneID := r.PathValue("zoneID") + id := r.PathValue("id") + slog.Info(DeleteDnsAlias, "zoneID", zoneID, "id", id) + if !assertAuth(DeleteDnsAlias, w, r, "account1:sgcp:dns:read", "account1:sgcp:dns:write") { + return + } + err := dnsApi.DeleteAlias(r.Context(), zoneID, id) + if err != nil { + http.Error(w, err.Error(), http.StatusNotFound) + return + } + w.WriteHeader(http.StatusNoContent) +} + +// ========== CG handlers (stub) ========== +func getCG(w http.ResponseWriter, r *http.Request) { + id := r.PathValue("id") + slog.Info(GetCG, "id", id) + if !assertAuth(GetCG, w, r, "account1:sgcp:files:read") { + return + } + // A group the driver can read: "ready" and "passive" are the states it + // settles in, "online" was none of them. + w.Header().Set("Content-Type", "application/json") + setHeader(w, GetCG, http.StatusOK) + json.NewEncoder(w).Encode(map[string]any{ + "uuid": id, + "name": "test-cg", + "availabilityZone": "region1-az1", + "status": "ready", + }) +} + +// ========== Auth ========== func postAuthToken(w http.ResponseWriter, r *http.Request) { slog.Info(PostAuth) auth := r.Header.Get("Authorization") @@ -213,11 +303,10 @@ func postAuthToken(w http.ResponseWriter, r *http.Request) { clientSecret := l[1] userScopes, ok := users.ScopesForAuth(clientID, clientSecret) if !ok { - slog.Info(PostAuth+" bad creadentials", "client_id", clientID) + slog.Info(PostAuth+" bad credentials", "client_id", clientID) setHeader(w, PostAuth, http.StatusForbidden) return } - if err := r.ParseForm(); err != nil { http.Error(w, err.Error(), http.StatusBadRequest) return @@ -231,14 +320,10 @@ func postAuthToken(w http.ResponseWriter, r *http.Request) { return } } - count := createdTokenCount.Add(1) - body := Token{ - AccessToken: fmt.Sprintf("%v", requestedScopes), - } + body := Token{AccessToken: fmt.Sprintf("%v", requestedScopes)} setHeader(w, PostAuth, http.StatusOK) slog.Info(PostAuth, "createdCount", count, "client_id", clientID, "createdToken", requestedScopes) - w.WriteHeader(http.StatusOK) json.NewEncoder(w).Encode(body) } @@ -249,14 +334,20 @@ func (u Users) ScopesForAuth(clientID, clientSecret string) ([]string, bool) { return nil, false } +// ========== Routes ========== var ( GetFile = "GET /file/fs/{id}" PostFileClient = "POST /file/fs/{id}/client" - GetFileClient = "Get /file/fs/{id}/client" + GetFileClient = "GET /file/fs/{id}/client" DeleteFileClient = "DELETE /file/fs/{id}/client/{clientID}" - // TODO: verify path - GetDnsAlias = "GET /dns/zone/{zoneID}/cname-entry" + ListDnsAliases = "GET /dns/zones/{zoneID}/cname-records" + CreateDnsAlias = "POST /dns/zones/{zoneID}/cname-records" + GetDnsAlias = "GET /dns/zones/{zoneID}/cname-records/{id}" + UpdateDnsAlias = "PATCH /dns/zones/{zoneID}/cname-records/{id}" + DeleteDnsAlias = "DELETE /dns/zones/{zoneID}/cname-records/{id}" + + GetCG = "GET /file/cg/{id}" PostAuth = "POST /auth/access_token" ) @@ -268,46 +359,48 @@ func loadUsers() (*Users, error) { if err != nil { return nil, fmt.Errorf("failed to read config file %s: %w", configFile, err) } - if err := yaml.Unmarshal(data, &u); err != nil { return nil, fmt.Errorf("failed to parse config file %s: %w", configFile, err) } - + // The client secrets stay out of the output: this reads a file an + // operator may well have filled with the credentials of a real account. for userID, user := range u { - slog.Info("loaded user", "userID", userID, "clientSecret", user.ClientSecret, "scopes", user.Scopes) + slog.Info("loaded user", "userID", userID, "clientID", user.ClientID, "scopes", user.Scopes) } - - fmt.Printf("loaded users: %#v\n", u) - return &u, nil } func main() { var err error createdTokenCount.Store(0) - mux := http.NewServeMux() - dnsDB := sgcpdnstesthelper.NewDB() - dnsApi := sgcpdnstesthelper.NewApi(dnsDB) + dnsDB = sgcpdnstesthelper.NewDB() + dnsApi = sgcpdnstesthelper.NewApi(dnsDB) users, err = loadUsers() if err != nil { slog.Error("failed to load users", "err", err) return } - // url: /file/fs/{id} - mux.HandleFunc(GetFile, getFileAPI) - // url: /file/fs/{id}/client + mux := http.NewServeMux() + + // File + mux.HandleFunc(GetFile, getFileAPI) mux.HandleFunc(PostFileClient, postFileClient) mux.HandleFunc(GetFileClient, getFileClient) - - // url: /file/fs/{id}/client/{clientID} mux.HandleFunc(DeleteFileClient, deleteFileClient) - // url: /dns/alias/{id} - mux.HandleFunc(GetDnsAlias, dnsGetAliasHandler(dnsApi)) + // DNS + mux.HandleFunc(ListDnsAliases, dnsListAliases) + mux.HandleFunc(CreateDnsAlias, dnsCreateAlias) + mux.HandleFunc(GetDnsAlias, dnsGetAlias) + mux.HandleFunc(UpdateDnsAlias, dnsUpdateAlias) + mux.HandleFunc(DeleteDnsAlias, dnsDeleteAlias) + + // CG + mux.HandleFunc(GetCG, getCG) - // url: /auth/access_token + // Auth mux.HandleFunc(PostAuth, postAuthToken) log.Println("Listening on :8000") diff --git a/util/sgcpserverfortest/users.yaml b/util/sgcpserverfortest/users.yaml new file mode 100644 index 000000000..7e220f9a6 --- /dev/null +++ b/util/sgcpserverfortest/users.yaml @@ -0,0 +1,8 @@ +account1: + client_id: account1 + client_secret: secret1 + scopes: + - account1:sgcp:files:read + - account1:sgcp:files:write + - account1:sgcp:dns:read + - account1:sgcp:dns:write diff --git a/util/testsgcphelper/main.go b/util/testsgcphelper/main.go index 9c5c69ce8..0bf34b35c 100644 --- a/util/testsgcphelper/main.go +++ b/util/testsgcphelper/main.go @@ -4,21 +4,39 @@ import ( "embed" "os" "path/filepath" + "regexp" "testing" "github.com/stretchr/testify/require" + + "github.com/opensvc/om3/v3/core/rawconfig" ) var ( //go:embed text fs embed.FS + + disabledFlagLine = regexp.MustCompile(`(?m)^disabled_flag:.*$`) ) +// InstallConfig writes the sgcp test configuration in a directory of its own +// and returns its path. +// +// That directory is also the agent root for the duration of the test, so the +// ageing caches the sgcp code keeps and the locks it takes land there instead +// of in the node /var/lib/opensvc. The disabled flag is moved there too, as +// its mere path is stat'ed on every action. A test run thus needs no +// privileges, and leaves nothing behind on the machine it runs on. func InstallConfig(t *testing.T) string { t.Helper() - tmpCfgFile := filepath.Join(t.TempDir(), "sgcp.yaml") + root := t.TempDir() + t.Cleanup(rawconfig.ReloadForTest(root)) + b, err := fs.ReadFile("text/config.yaml") require.NoError(t, err) - require.NoError(t, os.WriteFile(tmpCfgFile, b, 0755)) - return tmpCfgFile + b = disabledFlagLine.ReplaceAll(b, []byte("disabled_flag: "+filepath.Join(root, "sgcp_disabled"))) + + cfgFile := filepath.Join(root, "sgcp.yaml") + require.NoError(t, os.WriteFile(cfgFile, b, 0644)) + return cfgFile } diff --git a/util/testsgcphelper/text/config.yaml b/util/testsgcphelper/text/config.yaml index 7ac1e2f12..f0ea5ea96 100644 --- a/util/testsgcphelper/text/config.yaml +++ b/util/testsgcphelper/text/config.yaml @@ -8,6 +8,8 @@ files: fs: "/fs" client: "/client" cg: "/cg" + cg: + timeout: 300 dns: base_url: "https://127.0.0.1:1215/dns" @@ -30,7 +32,6 @@ auth: ttl_seconds: 1140 # 19 minutes cache: - ttl_seconds: 14400 # 4 hours - enabled: true + ttl_seconds: 14400 # 4 hours, 0 disables the caching disabled_flag: /var/lib/opensvc/sgcp_disabled