diff --git a/launchpad/pkg/file.go b/launchpad/pkg/file.go index cf5dbc9..450f0e0 100644 --- a/launchpad/pkg/file.go +++ b/launchpad/pkg/file.go @@ -21,6 +21,14 @@ type FlagConfig struct { } `json:"flags"` } +const ( + // ChangingFlagFile is the only definition /change writes; RestoreChangingFlag relies on that. + ChangingFlagFile = "rawflags/changing-flag.json" + ChangingFlagKey = "changing-flag" + BaselineChangingVariant = "foo" + toggledChangingVariant = "bar" +) + var ( fileLock sync.Mutex // lock for file operations changeLock sync.Mutex // lock for change requests (so that multiple requests don't overlap) @@ -32,40 +40,84 @@ func ToggleChangingFlag() (string, error) { changeLock.Lock() defer changeLock.Unlock() - // Path to the configuration file - configFile := "rawflags/changing-flag.json" + current, err := readChangingVariant() + if err != nil { + return "", err + } - // Read the existing file - data, err := os.ReadFile(configFile) + next := BaselineChangingVariant + if current == BaselineChangingVariant { + next = toggledChangingVariant + } + + return next, writeChangingVariant(next) +} + +// RestoreChangingFlag flips changing-flag back to its shipped variant and reports +// whether it had to. The file is read, not remembered, so a restarted launchpad +// in a dirty container still restores; this is sound because writes wait for +// flagd to serve them. +func RestoreChangingFlag() (bool, error) { + changeLock.Lock() + defer changeLock.Unlock() + + current, err := readChangingVariant() + if err != nil { + return false, err + } + if current == BaselineChangingVariant { + return false, nil + } + + return true, writeChangingVariant(BaselineChangingVariant) +} + +func readChangingVariant() (string, error) { + data, err := os.ReadFile(ChangingFlagFile) if err != nil { return "", err } - // Parse the JSON into the FlagConfig struct var config FlagConfig if err := json.Unmarshal(data, &config); err != nil { return "", err } - // Find the "changing-flag" and toggle the default variant - flag, exists := config.Flags["changing-flag"] + flag, exists := config.Flags[ChangingFlagKey] if !exists { return "", errors.New("changing-flag not found in configuration") } + return flag.DefaultVariant, nil +} - // Toggle the defaultVariant between "foo" and "bar" - if flag.DefaultVariant == "foo" { - flag.DefaultVariant = "bar" - } else { - flag.DefaultVariant = "foo" +// writeChangingVariant returns once flagd serves the new variant, not just once +// our watcher has regenerated the merged file. +func writeChangingVariant(variant string) error { + // Read the existing file + data, err := os.ReadFile(ChangingFlagFile) + if err != nil { + return err } + // Parse the JSON into the FlagConfig struct + var config FlagConfig + if err := json.Unmarshal(data, &config); err != nil { + return err + } + + // Find the "changing-flag" and set the default variant + flag, exists := config.Flags[ChangingFlagKey] + if !exists { + return errors.New("changing-flag not found in configuration") + } + flag.DefaultVariant = variant + // Save the updated flag back to the configuration - config.Flags["changing-flag"] = flag + config.Flags[ChangingFlagKey] = flag // Serialize the updated configuration back to JSON updatedData, err := json.MarshalIndent(config, "", " ") if err != nil { - return "", err + return err } // the file watcher should be triggered instantly. If not, we add a timeout to prevent a hanging test @@ -86,20 +138,18 @@ func ToggleChangingFlag() (string, error) { // Write the updated JSON back to the file fileLock.Lock() - if err := atomicWriteFile(configFile, updatedData); err != nil { + if err := atomicWriteFile(ChangingFlagFile, updatedData); err != nil { fileLock.Unlock() - return "", err + return err } fileLock.Unlock() - select { - case <-ctx.Done(): - if errors.Is(ctx.Err(), context.DeadlineExceeded) { - return "", fmt.Errorf("Flags were not updated in time: %v", ctx.Err()) - } else { - return flag.DefaultVariant, nil - } + <-ctx.Done() + if errors.Is(ctx.Err(), context.DeadlineExceeded) { + return fmt.Errorf("flags were not updated in time: %v", ctx.Err()) } + + return awaitVariantInFlagd(variant) } func RestartFileWatcher() error { @@ -133,7 +183,7 @@ func RestartFileWatcher() error { fmt.Printf("%v config changed, regenerating JSON...\n", event.Name) if err := CombineJSONFiles(InputDir); err != nil { fmt.Printf("Error combining JSON files: %v\n", err) - return + continue } if strings.HasSuffix(event.Name, "changing-flag.json") { for _, v := range changeFlagUpdateListeners { diff --git a/launchpad/pkg/flagd.go b/launchpad/pkg/flagd.go index 99c13e5..3ca2f10 100644 --- a/launchpad/pkg/flagd.go +++ b/launchpad/pkg/flagd.go @@ -2,15 +2,29 @@ package flagd import ( "context" + "encoding/json" "errors" "fmt" "net/http" + "net/url" "os" "os/exec" + "path/filepath" + "strings" "sync" "time" ) +const ( + variantBudget = 10 * time.Second + servedPollInterval = 10 * time.Millisecond + probeTimeout = 500 * time.Millisecond + + readyzURL = "http://localhost:8014/readyz" + // OFREP is plain HTTP in every config, including the TLS one. + ofrepEvaluateURL = "http://localhost:8016/ofrep/v1/evaluate/flags/" +) + var ( flagdCmd *exec.Cmd flagdLock sync.Mutex @@ -70,13 +84,19 @@ func RestartFlagd(seconds int) { } func StartFlagd(config string) error { + flagdLock.Lock() + if config == "" { config = Config - } else { - Config = config } - flagdLock.Lock() + // Reuse the running flagd if it has the requested config and still answers; + // Process stays non-nil after a crash, hence the readiness probe. + reuse := flagdCmd != nil && flagdCmd.Process != nil && + restartCancelFunc == nil && config == Config && isReady() + + Config = config + // Cancel any pending restart attempts if restartCancelFunc != nil { restartCancelFunc() @@ -84,14 +104,22 @@ func StartFlagd(config string) error { restartCancelFunc = nil } + configPath := flagdConfigPath(config) + + if reuse { + // The restore runs unlocked; a concurrent /stop or /restart is not + // guarded against, harnesses call these sequentially. + flagdLock.Unlock() + return resumeRunningFlagd() + } + if err := stopFlagDWithoutLock(); err != nil { + flagdLock.Unlock() return err } ensureStartConditions() - configPath := fmt.Sprintf("./configs/%s.json", config) - flagdCmd = exec.Command("./flagd", "start", "--config", configPath) flagdCmd.Stdout = os.Stdout flagdCmd.Stderr = os.Stderr @@ -114,7 +142,7 @@ func StartFlagd(config string) error { _ = StopFlagd() return fmt.Errorf("flagd health check timed out") case <-ticker.C: - resp, err := client.Get("http://localhost:8014/readyz") + resp, err := client.Get(readyzURL) if err == nil { resp.Body.Close() if resp.StatusCode == http.StatusOK { @@ -126,6 +154,141 @@ func StartFlagd(config string) error { } } +func flagdConfigPath(config string) string { + return fmt.Sprintf("./configs/%s.json", config) +} + +// resumeRunningFlagd restores the baseline flag state instead of restarting flagd. +func resumeRunningFlagd() error { + restored, err := RestoreChangingFlag() + if err != nil { + return fmt.Errorf("failed to restore the baseline flag state: %w", err) + } + + if !restored { + fmt.Println("flagd reused; baseline flag state already in force.") + return nil + } + + fmt.Println("flagd reused; baseline flag state restored.") + return nil +} + +// awaitVariantInFlagd waits until flagd serves changing-flag as variant. Configs +// that do not read the merged flag file can never serve it, so they are skipped. +func awaitVariantInFlagd(variant string) error { + if !flagdIsRunning() { + return nil + } + if !servesCombinedFlags(flagdConfigPath(Config)) { + return nil + } + + ctx, cancel := context.WithTimeout(context.Background(), variantBudget) + defer cancel() + + return awaitVariantServed(ctx, &http.Client{Timeout: probeTimeout}, ChangingFlagKey, variant) +} + +func flagdIsRunning() bool { + flagdLock.Lock() + defer flagdLock.Unlock() + + return flagdCmd != nil && flagdCmd.Process != nil +} + +func isReady() bool { + resp, err := (&http.Client{Timeout: probeTimeout}).Get(readyzURL) + if err != nil { + return false + } + resp.Body.Close() + return resp.StatusCode == http.StatusOK +} + +func awaitVariantServed(ctx context.Context, client *http.Client, key, variant string) error { + ticker := time.NewTicker(servedPollInterval) + defer ticker.Stop() + + for { + if servedVariant(ctx, client, key) == variant { + return nil + } + select { + case <-ctx.Done(): + return fmt.Errorf("flagd did not serve %q as variant %q before the budget expired", key, variant) + case <-ticker.C: + } + } +} + +func servedVariant(ctx context.Context, client *http.Client, key string) string { + req, err := http.NewRequestWithContext(ctx, http.MethodPost, ofrepEvaluateURL+url.PathEscape(key), strings.NewReader("{}")) + if err != nil { + return "" + } + req.Header.Set("Content-Type", "application/json") + + resp, err := client.Do(req) + if err != nil { + return "" + } + defer resp.Body.Close() + + if resp.StatusCode != http.StatusOK { + return "" + } + + var evaluation struct { + Variant string `json:"variant"` + } + if err := json.NewDecoder(resp.Body).Decode(&evaluation); err != nil { + return "" + } + return evaluation.Variant +} + +func servesCombinedFlags(configPath string) bool { + sources, err := fileSources(configPath) + if err != nil { + fmt.Printf("Cannot tell whether %s serves the merged flag file: %v\n", configPath, err) + return false + } + + for _, uri := range sources { + if filepath.Clean(uri) == filepath.Clean(OutputFile) { + return true + } + } + return false +} + +func fileSources(configPath string) ([]string, error) { + content, err := os.ReadFile(configPath) + if err != nil { + return nil, fmt.Errorf("failed to read flagd config: %w", err) + } + + var cfg struct { + Sources []struct { + URI string `json:"uri"` + Provider string `json:"provider"` + } `json:"sources"` + } + if err := json.Unmarshal(content, &cfg); err != nil { + return nil, fmt.Errorf("failed to parse flagd config: %w", err) + } + + var uris []string + for _, source := range cfg.Sources { + if source.Provider != "file" { + continue + } + uris = append(uris, source.URI) + } + return uris, nil +} + func StopFlagd() error { flagdLock.Lock() defer flagdLock.Unlock() diff --git a/launchpad/pkg/flagd_test.go b/launchpad/pkg/flagd_test.go new file mode 100644 index 0000000..064035e --- /dev/null +++ b/launchpad/pkg/flagd_test.go @@ -0,0 +1,104 @@ +package flagd + +import ( + "os" + "path/filepath" + "testing" +) + +func writeConfig(t *testing.T, body string) string { + t.Helper() + + path := filepath.Join(t.TempDir(), "config.json") + if err := os.WriteFile(path, []byte(body), 0o644); err != nil { + t.Fatalf("failed to write config: %v", err) + } + return path +} + +func TestFileSources(t *testing.T) { + tests := []struct { + name string + body string + want []string + }{ + { + name: "file sources are returned in order", + body: `{"sources":[{"uri":"flags/allFlags.json","provider":"file"},{"uri":"rawflags/selector-flags.json","provider":"file"}]}`, + want: []string{"flags/allFlags.json", "rawflags/selector-flags.json"}, + }, + { + name: "non-file providers are skipped", + body: `{"sources":[{"uri":"http://example.com","provider":"http"},{"uri":"flags/allFlags.json","provider":"file"}]}`, + want: []string{"flags/allFlags.json"}, + }, + { + name: "a configuration without sources yields none", + body: `{}`, + want: nil, + }, + } + + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + got, err := fileSources(writeConfig(t, tt.body)) + if err != nil { + t.Fatalf("fileSources returned an error: %v", err) + } + if len(got) != len(tt.want) { + t.Fatalf("got %v, want %v", got, tt.want) + } + for i := range got { + if got[i] != tt.want[i] { + t.Fatalf("got %v, want %v", got, tt.want) + } + } + }) + } +} + +func TestFileSourcesRejectsUnreadableAndInvalidConfigs(t *testing.T) { + if _, err := fileSources(filepath.Join(t.TempDir(), "missing.json")); err == nil { + t.Fatal("expected an error for a missing configuration") + } + if _, err := fileSources(writeConfig(t, "not json")); err == nil { + t.Fatal("expected an error for an unparseable configuration") + } +} + +func TestServesCombinedFlags(t *testing.T) { + tests := []struct { + name string + body string + want bool + }{ + { + name: "the default shape reads the merged file", + body: `{"sources":[{"uri":"flags/allFlags.json","provider":"file"},{"uri":"rawflags/selector-flags.json","provider":"file"}]}`, + want: true, + }, + { + name: "the merged file is recognised through an unclean path", + body: `{"sources":[{"uri":"./flags/allFlags.json","provider":"file"}]}`, + want: true, + }, + { + name: "the metadata shape does not", + body: `{"sources":[{"uri":"rawflags/selector-flag-combined-metadata.json","provider":"file"}]}`, + want: false, + }, + { + name: "an unreadable configuration does not", + body: `not json`, + want: false, + }, + } + + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + if got := servesCombinedFlags(writeConfig(t, tt.body)); got != tt.want { + t.Fatalf("got %v, want %v", got, tt.want) + } + }) + } +}