diff --git a/.github/workflows/release.yaml b/.github/workflows/release.yaml index 74c9a71..487c07e 100644 --- a/.github/workflows/release.yaml +++ b/.github/workflows/release.yaml @@ -24,6 +24,11 @@ jobs: go-version-file: go.mod cache: false + - name: Validate release configuration + env: + PB_CLOUD_ORCHESTRATOR_URL: ${{ vars.PB_CLOUD_ORCHESTRATOR_URL }} + run: test -n "$PB_CLOUD_ORCHESTRATOR_URL" + - name: Run GoReleaser uses: goreleaser/goreleaser-action@v6 with: @@ -32,6 +37,7 @@ jobs: args: release --parallelism 1 --clean env: GITHUB_TOKEN: ${{ secrets.GITHUB_TOKEN }} + PB_CLOUD_ORCHESTRATOR_URL: ${{ vars.PB_CLOUD_ORCHESTRATOR_URL }} - name: Update Homebrew tap env: diff --git a/.goreleaser.yml b/.goreleaser.yml index a2a726c..96ed2e8 100644 --- a/.goreleaser.yml +++ b/.goreleaser.yml @@ -18,7 +18,7 @@ builds: - -trimpath - -tags=kqueue ldflags: - - -s -w -X main.Version=v{{ .Version }} -X main.Commit={{ .ShortCommit }} + - -s -w -X main.Version=v{{ .Version }} -X main.Commit={{ .ShortCommit }} -X github.com/parseablehq/pb/pkg/config.CloudOrchestratorURL={{ .Env.PB_CLOUD_ORCHESTRATOR_URL }} archives: - name_template: "pb_{{ .Version }}_{{ .Os }}_{{ .Arch }}" diff --git a/buildscripts/gen-ldflags.go b/buildscripts/gen-ldflags.go index 5d8bccf..11e9be9 100644 --- a/buildscripts/gen-ldflags.go +++ b/buildscripts/gen-ldflags.go @@ -29,6 +29,9 @@ func genLDFlags(version string) string { var ldflagsStr string ldflagsStr = "-s -w -X main.Version=" + version + " " ldflagsStr = ldflagsStr + "-X main.Commit=" + commitID()[:6] + if orchestratorURL := os.Getenv("PB_CLOUD_ORCHESTRATOR_URL"); orchestratorURL != "" { + ldflagsStr += " -X github.com/parseablehq/pb/pkg/config.CloudOrchestratorURL=" + orchestratorURL + } return ldflagsStr } diff --git a/cmd/cloud.go b/cmd/cloud.go new file mode 100644 index 0000000..cfe7be8 --- /dev/null +++ b/cmd/cloud.go @@ -0,0 +1,551 @@ +// Copyright (c) 2024 Parseable, Inc +// +// This program is free software: you can redistribute it and/or modify +// it under the terms of the GNU Affero General Public License as published by +// the Free Software Foundation, either version 3 of the License, or +// (at your option) any later version. + +package cmd + +import ( + "bytes" + "context" + "encoding/json" + "errors" + "fmt" + "io" + "net/http" + "net/url" + "os" + "os/exec" + "runtime" + "strings" + "time" + + "github.com/charmbracelet/bubbles/spinner" + tea "github.com/charmbracelet/bubbletea" + "github.com/charmbracelet/lipgloss" + "github.com/parseablehq/pb/pkg/config" + "github.com/parseablehq/pb/pkg/ui" + "github.com/spf13/cobra" +) + +const ( + cloudDeviceClientID = "pb-cli" + cloudDefaultPollInterval = 5 * time.Second + cloudDefaultLoginTimeout = 5 * time.Minute +) + +var ( + cloudAPIKey string + cloudProfileName string + cloudOrchestratorURL string + cloudForceOverwrite bool +) + +type cloudAPIKeyValidationResponse struct { + OrgID string `json:"org_id"` + OrgName string `json:"org_name"` + WorkspaceID string `json:"workspace_id"` + WorkspaceName string `json:"workspace_name"` + TenantID string `json:"tenant_id"` + URL string `json:"url"` + IngestURL string `json:"ingest_url"` + State string `json:"state"` + MultiTenant bool `json:"multi_tenant"` +} + +type cloudDeviceCodeResponse struct { + DeviceCode string `json:"device_code"` + UserCode string `json:"user_code"` + VerificationURI string `json:"verification_uri"` + VerificationURIComplete string `json:"verification_uri_complete"` + ExpiresIn int `json:"expires_in"` + Interval int `json:"interval"` +} + +type cloudDeviceTokenRequest struct { + GrantType string `json:"grant_type"` + DeviceCode string `json:"device_code,omitempty"` + RefreshToken string `json:"refresh_token,omitempty"` +} + +type cloudDeviceTokenResponse struct { + AccessToken string `json:"access_token"` + TokenType string `json:"token_type"` + RefreshToken string `json:"refresh_token"` + ExpiresIn int `json:"expires_in"` + IDToken string `json:"id_token"` + Workspace cloudDeviceWorkspaceRef `json:"workspace"` +} + +type cloudDeviceWorkspaceRef struct { + ID string `json:"id"` + URL string `json:"url"` + TenantID string `json:"tenant_id"` +} + +type cloudOAuthError struct { + Error string `json:"error"` + ErrorDescription string `json:"error_description"` +} + +type cloudDeviceLoginResultMsg struct { + tokens *cloudDeviceTokenResponse + err error +} + +type cloudDeviceLoginModel struct { + spinner spinner.Model + verificationURL string + userCode string + poll func() (*cloudDeviceTokenResponse, error) + openBrowser tea.Cmd + cancel context.CancelFunc + tokens *cloudDeviceTokenResponse + err error + done bool +} + +var cloudLoginTitleStyle = lipgloss.NewStyle().Bold(true).Foreground( + ui.Adaptive(func(p ui.Palette) lipgloss.Color { return p.Accent }), +) + +func newCloudDeviceLoginModel( + verificationURL string, + userCode string, + poll func() (*cloudDeviceTokenResponse, error), + openBrowser tea.Cmd, + cancel context.CancelFunc, +) cloudDeviceLoginModel { + sp := spinner.New() + sp.Spinner = spinner.Dot + return cloudDeviceLoginModel{ + spinner: sp, + verificationURL: verificationURL, + userCode: userCode, + poll: poll, + openBrowser: openBrowser, + cancel: cancel, + } +} + +func (m cloudDeviceLoginModel) Init() tea.Cmd { + return tea.Batch(m.spinner.Tick, m.openBrowser, func() tea.Msg { + tokens, err := m.poll() + return cloudDeviceLoginResultMsg{tokens: tokens, err: err} + }) +} + +func (m cloudDeviceLoginModel) Update(msg tea.Msg) (tea.Model, tea.Cmd) { + switch msg := msg.(type) { + case tea.KeyMsg: + if msg.String() == "ctrl+c" { + m.cancel() + m.err = errors.New("cloud device login aborted") + m.done = true + return m, tea.Quit + } + case spinner.TickMsg: + var cmd tea.Cmd + m.spinner, cmd = m.spinner.Update(msg) + return m, cmd + case cloudDeviceLoginResultMsg: + m.tokens = msg.tokens + m.err = msg.err + m.done = true + return m, tea.Quit + } + return m, nil +} + +func (m cloudDeviceLoginModel) View() string { + var status string + switch { + case !m.done: + status = m.spinner.View() + " Waiting for authorization... [Ctrl+C to abort]" + case m.err == nil: + status = "✓ Authorization successful!" + default: + status = "✗ Authorization failed." + } + return fmt.Sprintf( + "\n %s\n %s\n\n To authenticate, please visit:\n → %s\n\n Enter this Confirmation Code:\n ❯ %s\n\n %s\n", + cloudLoginTitleStyle.Render("PARSEABLE LOGIN"), + strings.Repeat("─", 60), + m.verificationURL, + m.userCode, + status, + ) +} + +var CloudCmd = &cobra.Command{ + Use: "cloud", + Short: "Manage Parseable Cloud profiles", + Long: "\ncloud command is used to configure Parseable Cloud access.", +} + +var CloudProfileCmd = &cobra.Command{ + Use: "profile", + Short: "Manage Parseable Cloud profiles", + Long: "\ncloud profile command is used to add Parseable Cloud profiles.", +} + +var CloudProfileAddCmd = &cobra.Command{ + Use: "add", + Example: " pb cloud profile add --api-key psk_xxx\n pb cloud profile add --api-key psk_xxx --name prod", + Short: "Add a Parseable Cloud profile", + Long: "\nAdd a Parseable Cloud profile using an API key created in Parseable Cloud.", + RunE: func(_ *cobra.Command, _ []string) error { + apiKey := strings.TrimSpace(cloudAPIKey) + if apiKey == "" { + return errors.New("api key is required. pass --api-key") + } + + orchestratorURL := cloudOrchestratorEndpoint() + result, err := validateCloudAPIKey(orchestratorURL, apiKey) + if err != nil { + return err + } + + profileName := strings.TrimSpace(cloudProfileName) + if profileName == "" { + profileName = cloudProfileNameFromWorkspace(result) + } + profile := config.Profile{ + URL: result.URL, + Cloud: true, + APIKey: apiKey, + TenantID: result.TenantID, + IngestURL: result.IngestURL, + WorkspaceID: result.WorkspaceID, + WorkspaceName: result.WorkspaceName, + OrchestratorURL: orchestratorURL, + } + if err := saveCloudProfile(profileName, profile, cloudForceOverwrite, true); err != nil { + return err + } + + fmt.Printf("Cloud profile %s added successfully\n", profileName) + if result.WorkspaceName != "" { + fmt.Printf("Workspace: %s (%s)\n", result.WorkspaceName, result.WorkspaceID) + } else { + fmt.Printf("Workspace: %s\n", result.WorkspaceID) + } + fmt.Printf("URL: %s\n", result.URL) + return nil + }, +} + +func init() { + CloudProfileAddCmd.Flags().StringVar(&cloudAPIKey, "api-key", "", "Parseable Cloud API key") + CloudProfileAddCmd.Flags().StringVar(&cloudProfileName, "name", "", "profile name") + CloudProfileAddCmd.Flags().StringVar(&cloudOrchestratorURL, "orchestrator-url", config.CloudOrchestratorURL, "Parseable Cloud orchestrator URL") + CloudProfileAddCmd.Flags().BoolVar(&cloudForceOverwrite, "force", false, "overwrite existing profile") + + CloudProfileCmd.AddCommand(CloudProfileAddCmd) + CloudCmd.AddCommand(CloudProfileCmd) +} + +func cloudOrchestratorEndpoint() string { + if endpoint := strings.TrimSpace(cloudOrchestratorURL); endpoint != "" { + return endpoint + } + return config.CloudOrchestratorURL +} + +func cloudProfileFromDeviceLogin(parent context.Context, orchestratorURL string) (*config.Profile, error) { + client := &http.Client{Timeout: 30 * time.Second} + device, err := requestCloudDeviceCode(parent, client, orchestratorURL) + if err != nil { + return nil, err + } + + verificationURL := strings.TrimSpace(device.VerificationURI) + if verificationURL == "" { + return nil, errors.New("cloud device authorization response missing verification URI") + } + + timeout := cloudDefaultLoginTimeout + if device.ExpiresIn > 0 { + timeout = time.Duration(device.ExpiresIn) * time.Second + } + ctx, cancel := context.WithTimeout(parent, timeout) + defer cancel() + + model := newCloudDeviceLoginModel( + verificationURL, + device.UserCode, + func() (*cloudDeviceTokenResponse, error) { + return pollCloudDeviceToken(ctx, client, orchestratorURL, device) + }, + func() tea.Msg { + if err := waitForCloudPoll(ctx, 5*time.Second); err == nil { + _ = openExternalBrowser(verificationURL) + } + return nil + }, + cancel, + ) + result, err := tea.NewProgram(model).Run() + if err != nil { + return nil, fmt.Errorf("failed to render cloud device login: %w", err) + } + loginResult, ok := result.(cloudDeviceLoginModel) + if !ok { + return nil, errors.New("cloud device login returned an unexpected result") + } + if loginResult.err != nil { + return nil, loginResult.err + } + return cloudProfileFromDeviceTokens(orchestratorURL, loginResult.tokens) +} + +func openExternalBrowser(rawURL string) error { + var command *exec.Cmd + switch runtime.GOOS { + case "darwin": + command = exec.Command("open", rawURL) + case "windows": + command = exec.Command("rundll32", "url.dll,FileProtocolHandler", rawURL) + default: + command = exec.Command("xdg-open", rawURL) + } + return command.Start() +} + +func requestCloudDeviceCode(ctx context.Context, client *http.Client, orchestratorURL string) (*cloudDeviceCodeResponse, error) { + body, err := json.Marshal(map[string]string{"client_id": cloudDeviceClientID}) + if err != nil { + return nil, err + } + endpoint, err := url.JoinPath(orchestratorURL, "api/v1/cli/device/code") + if err != nil { + return nil, fmt.Errorf("invalid orchestrator URL: %w", err) + } + + var result cloudDeviceCodeResponse + if err := doCloudJSON(ctx, client, http.MethodPost, endpoint, body, &result); err != nil { + return nil, fmt.Errorf("failed to create cloud device code: %w", err) + } + if result.DeviceCode == "" || result.UserCode == "" { + return nil, errors.New("cloud device authorization response missing device_code or user_code") + } + return &result, nil +} + +func pollCloudDeviceToken(ctx context.Context, client *http.Client, orchestratorURL string, device *cloudDeviceCodeResponse) (*cloudDeviceTokenResponse, error) { + endpoint, err := url.JoinPath(orchestratorURL, "api/v1/cli/oauth/token") + if err != nil { + return nil, fmt.Errorf("invalid orchestrator URL: %w", err) + } + interval := cloudDefaultPollInterval + if device.Interval > 0 { + interval = time.Duration(device.Interval) * time.Second + } + + for { + if err := waitForCloudPoll(ctx, interval); err != nil { + return nil, errors.New("cloud device login expired or timed out") + } + + payload, err := json.Marshal(cloudDeviceTokenRequest{ + GrantType: "urn:ietf:params:oauth:grant-type:device_code", + DeviceCode: device.DeviceCode, + }) + if err != nil { + return nil, err + } + var tokens cloudDeviceTokenResponse + oauthErr, err := doCloudOAuthTokenRequest(ctx, client, endpoint, payload, &tokens) + if err != nil { + return nil, err + } + if oauthErr == nil { + return &tokens, nil + } + + switch oauthErr.Error { + case "authorization_pending": + continue + case "expired_token": + return nil, errors.New("cloud device code expired; run login again") + default: + return nil, fmt.Errorf("cloud device login failed: %s", cloudOAuthErrorMessage(oauthErr)) + } + } +} + +func waitForCloudPoll(ctx context.Context, interval time.Duration) error { + timer := time.NewTimer(interval) + defer timer.Stop() + select { + case <-ctx.Done(): + return ctx.Err() + case <-timer.C: + return nil + } +} + +func doCloudOAuthTokenRequest(ctx context.Context, client *http.Client, endpoint string, payload []byte, out *cloudDeviceTokenResponse) (*cloudOAuthError, error) { + req, err := http.NewRequestWithContext(ctx, http.MethodPost, endpoint, bytes.NewReader(payload)) + if err != nil { + return nil, err + } + req.Header.Set("Accept", "application/json") + req.Header.Set("Content-Type", "application/json") + + resp, err := client.Do(req) + if err != nil { + return nil, fmt.Errorf("cloud token request failed: %w", err) + } + defer resp.Body.Close() + body, err := io.ReadAll(resp.Body) + if err != nil { + return nil, err + } + if resp.StatusCode >= http.StatusOK && resp.StatusCode < http.StatusMultipleChoices { + if err := json.Unmarshal(body, out); err != nil { + return nil, fmt.Errorf("failed to decode cloud token response: %w", err) + } + return nil, nil + } + + var oauthErr cloudOAuthError + if err := json.Unmarshal(body, &oauthErr); err != nil || oauthErr.Error == "" { + return nil, fmt.Errorf("cloud token request failed: %s: %s", resp.Status, strings.TrimSpace(string(body))) + } + return &oauthErr, nil +} + +func cloudOAuthErrorMessage(oauthErr *cloudOAuthError) string { + if oauthErr.ErrorDescription != "" { + return oauthErr.Error + ": " + oauthErr.ErrorDescription + } + return oauthErr.Error +} + +func cloudProfileFromDeviceTokens(orchestratorURL string, tokens *cloudDeviceTokenResponse) (*config.Profile, error) { + if tokens == nil || strings.TrimSpace(tokens.AccessToken) == "" { + return nil, errors.New("cloud token response missing access_token") + } + if strings.TrimSpace(tokens.RefreshToken) == "" { + return nil, errors.New("cloud token response missing refresh_token") + } + if strings.TrimSpace(tokens.Workspace.ID) == "" || strings.TrimSpace(tokens.Workspace.URL) == "" || strings.TrimSpace(tokens.Workspace.TenantID) == "" { + return nil, errors.New("cloud token response missing workspace id, url, or tenant_id") + } + return &config.Profile{ + URL: tokens.Workspace.URL, + Cloud: true, + SessionToken: tokens.AccessToken, + RefreshToken: tokens.RefreshToken, + TenantID: tokens.Workspace.TenantID, + WorkspaceID: tokens.Workspace.ID, + OrchestratorURL: orchestratorURL, + }, nil +} + +func doCloudJSON(ctx context.Context, client *http.Client, method, endpoint string, payload []byte, out interface{}) error { + req, err := http.NewRequestWithContext(ctx, method, endpoint, bytes.NewReader(payload)) + if err != nil { + return err + } + req.Header.Set("Accept", "application/json") + req.Header.Set("Content-Type", "application/json") + resp, err := client.Do(req) + if err != nil { + return err + } + defer resp.Body.Close() + body, err := io.ReadAll(resp.Body) + if err != nil { + return err + } + if resp.StatusCode < http.StatusOK || resp.StatusCode >= http.StatusMultipleChoices { + return fmt.Errorf("%s: %s", resp.Status, strings.TrimSpace(string(body))) + } + if err := json.Unmarshal(body, out); err != nil { + return fmt.Errorf("failed to decode response: %w", err) + } + return nil +} + +func validateCloudAPIKey(orchestratorURL, apiKey string) (*cloudAPIKeyValidationResponse, error) { + endpoint, err := url.JoinPath(orchestratorURL, "api/v1/apikey/validate") + if err != nil { + return nil, fmt.Errorf("invalid orchestrator URL: %w", err) + } + req, err := http.NewRequest(http.MethodGet, endpoint, nil) + if err != nil { + return nil, err + } + req.Header.Set("Authorization", "Bearer "+apiKey) + client := http.Client{Timeout: 30 * time.Second} + resp, err := client.Do(req) + if err != nil { + return nil, fmt.Errorf("failed to validate cloud api key: %w", err) + } + defer resp.Body.Close() + body, err := io.ReadAll(resp.Body) + if err != nil { + return nil, err + } + if resp.StatusCode < http.StatusOK || resp.StatusCode >= http.StatusMultipleChoices { + return nil, fmt.Errorf("cloud api key validation failed: %s: %s", resp.Status, strings.TrimSpace(string(body))) + } + var result cloudAPIKeyValidationResponse + if err := json.Unmarshal(body, &result); err != nil { + return nil, fmt.Errorf("failed to decode cloud api key validation response: %w", err) + } + if result.URL == "" || result.TenantID == "" || result.WorkspaceID == "" { + return nil, errors.New("cloud api key validation response missing url, tenant_id, or workspace_id") + } + return &result, nil +} + +func saveCloudProfile(name string, profile config.Profile, force, makeDefault bool) error { + if name == "" { + return errors.New("profile name is required") + } + fileConfig, err := config.ReadConfigFromFile() + if os.IsNotExist(err) { + fileConfig = &config.Config{Profiles: make(map[string]config.Profile)} + } else if err != nil { + return fmt.Errorf("failed to read config: %w", err) + } + if fileConfig.Profiles == nil { + fileConfig.Profiles = make(map[string]config.Profile) + } + if _, exists := fileConfig.Profiles[name]; exists && !force { + return fmt.Errorf("profile %q already exists. use --force to overwrite", name) + } + fileConfig.Profiles[name] = profile + if fileConfig.DefaultProfile == "" || makeDefault { + fileConfig.DefaultProfile = name + } + return config.WriteConfigToFile(fileConfig) +} + +func cloudProfileNameFromWorkspace(result *cloudAPIKeyValidationResponse) string { + name := strings.TrimSpace(result.WorkspaceName) + if name == "" { + name = strings.TrimSpace(result.WorkspaceID) + } + return normalizeCloudProfileName(name) +} + +func cloudProfileNameFromSession(profile *config.Profile) string { + name := strings.TrimSpace(profile.WorkspaceName) + if name == "" { + name = strings.TrimSpace(profile.WorkspaceID) + } + if name == "" { + name = "cloud" + } + return normalizeCloudProfileName(name) +} + +func normalizeCloudProfileName(name string) string { + return strings.ReplaceAll(strings.ToLower(name), " ", "-") +} diff --git a/cmd/cloud_profile_test.go b/cmd/cloud_profile_test.go new file mode 100644 index 0000000..83b72bb --- /dev/null +++ b/cmd/cloud_profile_test.go @@ -0,0 +1,51 @@ +package cmd + +import ( + "testing" + + "github.com/parseablehq/pb/pkg/config" +) + +func TestCloudCommandsDoNotExposeDefaultFlag(t *testing.T) { + if flag := CloudProfileAddCmd.Flags().Lookup("default"); flag != nil { + t.Fatal("cloud profile add still exposes --default") + } +} + +func TestSaveCloudProfilePreservesExistingDefault(t *testing.T) { + t.Setenv("XDG_CONFIG_HOME", t.TempDir()) + + if err := saveCloudProfile("first", config.Profile{URL: "https://first.example.com"}, false, false); err != nil { + t.Fatal(err) + } + if err := saveCloudProfile("second", config.Profile{URL: "https://second.example.com"}, false, false); err != nil { + t.Fatal(err) + } + + fileConfig, err := config.ReadConfigFromFile() + if err != nil { + t.Fatal(err) + } + if fileConfig.DefaultProfile != "first" { + t.Fatalf("default profile changed to %q", fileConfig.DefaultProfile) + } +} + +func TestSaveCloudProfileCanMakeAddedProfileDefault(t *testing.T) { + t.Setenv("XDG_CONFIG_HOME", t.TempDir()) + + if err := saveCloudProfile("first", config.Profile{URL: "https://first.example.com"}, false, false); err != nil { + t.Fatal(err) + } + if err := saveCloudProfile("second", config.Profile{URL: "https://second.example.com"}, false, true); err != nil { + t.Fatal(err) + } + + fileConfig, err := config.ReadConfigFromFile() + if err != nil { + t.Fatal(err) + } + if fileConfig.DefaultProfile != "second" { + t.Fatalf("added profile was not made default: %q", fileConfig.DefaultProfile) + } +} diff --git a/cmd/cluster.go b/cmd/cluster.go index 5f405ae..4a12483 100644 --- a/cmd/cluster.go +++ b/cmd/cluster.go @@ -23,6 +23,7 @@ import ( "github.com/olekukonko/tablewriter" "github.com/parseablehq/pb/pkg/common" + "github.com/parseablehq/pb/pkg/config" "github.com/parseablehq/pb/pkg/helm" "github.com/parseablehq/pb/pkg/installer" "github.com/spf13/cobra" @@ -196,7 +197,7 @@ func uninstallCluster(entry common.InstallerEntry) error { ReleaseName: entry.Name, Namespace: entry.Namespace, RepoName: "parseable", - RepoURL: "https://charts.parseable.com", + RepoURL: config.HelmChartRepositoryURL, ChartName: "parseable", Version: entry.Version, } diff --git a/cmd/dataset_test.go b/cmd/dataset_test.go index d8fd6bc..a0f0d7e 100644 --- a/cmd/dataset_test.go +++ b/cmd/dataset_test.go @@ -19,7 +19,7 @@ func TestFetchInfoUsesTelemetryType(t *testing.T) { })) defer server.Close() - client := internalHTTP.DefaultClient(&config.Profile{URL: server.URL}) + client := internalHTTP.DefaultClient(&config.Profile{URL: server.URL, APIKey: "test-api-key"}) got, err := fetchInfo(&client, "fly_logs") if err != nil { t.Fatalf("fetchInfo returned error: %v", err) @@ -63,7 +63,7 @@ func TestFetchInfoReturnsNotFoundError(t *testing.T) { server := httptest.NewServer(http.NotFoundHandler()) defer server.Close() - client := internalHTTP.DefaultClient(&config.Profile{URL: server.URL}) + client := internalHTTP.DefaultClient(&config.Profile{URL: server.URL, APIKey: "test-api-key"}) _, err := fetchInfo(&client, "missing") if err == nil { t.Fatal("expected not found error") diff --git a/cmd/login.go b/cmd/login.go index ff94c23..d4c0927 100644 --- a/cmd/login.go +++ b/cmd/login.go @@ -29,9 +29,9 @@ var LoginCmd = &cobra.Command{ Short: "Login to Parseable", Long: `Interactive login wizard for Parseable. -Select self-hosted and enter your server URL, credentials, and a -profile name. All settings are saved to ~/.config/pb/config.toml.`, - RunE: func(_ *cobra.Command, _ []string) error { +Select self-hosted or Parseable Cloud and enter the required +credentials. All settings are saved to ~/.config/pb/config.toml.`, + RunE: func(cmd *cobra.Command, _ []string) error { _m, err := tea.NewProgram(login.New()).Run() if err != nil { return err @@ -42,13 +42,58 @@ profile name. All settings are saved to ~/.config/pb/config.toml.`, return nil } + if m.CloudDeviceLogin { + profile, err := cloudProfileFromDeviceLogin(cmd.Context(), config.CloudOrchestratorURL) + if err != nil { + return err + } + m.Profile = *profile + if m.Name == "" { + m.Name = cloudProfileNameFromSession(profile) + } + } else if m.Profile.Cloud { + profile, err := cloudProfileFromAPIKey(m.Profile.APIKey) + if err != nil { + return err + } + m.Profile = *profile + } + if err := writeProfile(m.Profile, m.Name); err != nil { return fmt.Errorf("failed to save profile: %w", err) } + if m.CloudDeviceLogin { + printDeviceLoginSuccess(m.Name) + } return nil }, } +func printDeviceLoginSuccess(profileName string) { + fmt.Printf(" ✓ Profile '%s' saved.\n\n", profileName) + fmt.Println(" 💡 Tip: You can add more profiles anytime using:") + fmt.Println(" pb profile add [user] [pass]") +} + +func cloudProfileFromAPIKey(apiKey string) (*config.Profile, error) { + orchestratorURL := config.CloudOrchestratorURL + result, err := validateCloudAPIKey(orchestratorURL, apiKey) + if err != nil { + return nil, err + } + + return &config.Profile{ + URL: result.URL, + Cloud: true, + APIKey: apiKey, + TenantID: result.TenantID, + IngestURL: result.IngestURL, + WorkspaceID: result.WorkspaceID, + WorkspaceName: result.WorkspaceName, + OrchestratorURL: orchestratorURL, + }, nil +} + func writeProfile(profile config.Profile, profileName string) error { fileConfig, err := config.ReadConfigFromFile() if err != nil { @@ -63,8 +108,6 @@ func writeProfile(profile config.Profile, profileName string) error { fileConfig.Profiles = make(map[string]config.Profile) } fileConfig.Profiles[profileName] = profile - if fileConfig.DefaultProfile == "" { - fileConfig.DefaultProfile = profileName - } + fileConfig.DefaultProfile = profileName return config.WriteConfigToFile(fileConfig) } diff --git a/cmd/logout.go b/cmd/logout.go index b1dc315..37db252 100644 --- a/cmd/logout.go +++ b/cmd/logout.go @@ -17,8 +17,10 @@ package cmd import ( "bufio" + "encoding/json" "fmt" "os" + "sort" "strings" tea "github.com/charmbracelet/bubbletea" @@ -31,8 +33,20 @@ var LogoutCmd = &cobra.Command{ Use: "logout", Short: "Logout from the current Parseable profile", Long: "Removes the active profile (URL and credentials) from config.", - Example: " pb logout", - RunE: func(_ *cobra.Command, _ []string) error { + Example: " pb logout\n pb logout --yes -o json", + RunE: func(cmd *cobra.Command, _ []string) error { + outputFormat, err := cmd.Flags().GetString("output") + if err != nil { + return err + } + yes, err := cmd.Flags().GetBool("yes") + if err != nil { + return err + } + if outputFormat == "json" && !yes { + return fmt.Errorf("use --yes with -o json to avoid interactive prompts") + } + fileConfig, err := config.ReadConfigFromFile() if err != nil { return fmt.Errorf("no config found — nothing to logout from") @@ -44,6 +58,9 @@ var LogoutCmd = &cobra.Command{ if len(fileConfig.Profiles) == 0 { return fmt.Errorf("no active profile found") } + if len(fileConfig.Profiles) > 1 && (yes || outputFormat == "json") { + return fmt.Errorf("no active profile found; set one with: pb profile default ") + } selectedProfile, err := selectLogoutProfile(fileConfig.Profiles) if err != nil { return err @@ -56,7 +73,7 @@ var LogoutCmd = &cobra.Command{ activeProfile = fileConfig.Profiles[profileName] } - if !confirmLogout(profileName, activeProfile.URL) { + if !yes && !confirmLogout(profileName, activeProfile.URL) { fmt.Println("Logout canceled") return nil } @@ -72,17 +89,22 @@ var LogoutCmd = &cobra.Command{ newDefaultProfile = name } default: - fmt.Println("Select a new default profile:") - _m, err := tea.NewProgram(defaultprofile.New(fileConfig.Profiles)).Run() - if err != nil { - return fmt.Errorf("error selecting new default profile: %w", err) - } - m := _m.(defaultprofile.Model) - if m.Success { - fileConfig.DefaultProfile = m.Choice - newDefaultProfile = m.Choice + if yes { + fileConfig.DefaultProfile = firstSortedProfileName(fileConfig.Profiles) + newDefaultProfile = fileConfig.DefaultProfile } else { - fileConfig.DefaultProfile = "" + fmt.Println("Select a new default profile:") + _m, err := tea.NewProgram(defaultprofile.New(fileConfig.Profiles)).Run() + if err != nil { + return fmt.Errorf("error selecting new default profile: %w", err) + } + m := _m.(defaultprofile.Model) + if m.Success { + fileConfig.DefaultProfile = m.Choice + newDefaultProfile = m.Choice + } else { + fileConfig.DefaultProfile = "" + } } } @@ -90,6 +112,16 @@ var LogoutCmd = &cobra.Command{ return fmt.Errorf("failed to update config: %w", err) } + if outputFormat == "json" { + return printLogoutJSON(logoutOutput{ + Status: "ok", + Profile: profileName, + URL: activeProfile.URL, + Removed: true, + DefaultProfile: fileConfig.DefaultProfile, + }) + } + fmt.Printf("Logged out and removed profile '%s'\n", profileName) if newDefaultProfile != "" { fmt.Printf("'%s' is now set as the default profile\n", newDefaultProfile) @@ -98,6 +130,35 @@ var LogoutCmd = &cobra.Command{ }, } +type logoutOutput struct { + Status string `json:"status"` + Profile string `json:"profile"` + URL string `json:"url"` + Removed bool `json:"removed"` + DefaultProfile string `json:"default_profile"` +} + +func printLogoutJSON(result logoutOutput) error { + jsonData, err := json.MarshalIndent(result, "", " ") + if err != nil { + return fmt.Errorf("failed to marshal logout JSON: %w", err) + } + fmt.Println(string(jsonData)) + return nil +} + +func firstSortedProfileName(profiles map[string]config.Profile) string { + names := make([]string, 0, len(profiles)) + for name := range profiles { + names = append(names, name) + } + sort.Strings(names) + if len(names) == 0 { + return "" + } + return names[0] +} + func selectLogoutProfile(profiles map[string]config.Profile) (string, error) { if len(profiles) == 1 { for name := range profiles { @@ -138,3 +199,8 @@ func confirmLogout(profileName, profileURL string) bool { return false } } + +func init() { + LogoutCmd.Flags().Bool("yes", false, "confirm logout without prompting") + LogoutCmd.Flags().StringP("output", "o", "text", "Output format (text|json)") +} diff --git a/cmd/logout_test.go b/cmd/logout_test.go new file mode 100644 index 0000000..d9934e4 --- /dev/null +++ b/cmd/logout_test.go @@ -0,0 +1,45 @@ +package cmd + +import ( + "strings" + "testing" + + "github.com/parseablehq/pb/pkg/config" +) + +func TestLogoutYesDoesNotPromptWhenDefaultIsAmbiguous(t *testing.T) { + t.Setenv("XDG_CONFIG_HOME", t.TempDir()) + fileConfig := &config.Config{ + Profiles: map[string]config.Profile{ + "first": {URL: "https://first.example.com"}, + "second": {URL: "https://second.example.com"}, + }, + } + if err := config.WriteConfigToFile(fileConfig); err != nil { + t.Fatal(err) + } + + if err := LogoutCmd.Flags().Set("yes", "true"); err != nil { + t.Fatal(err) + } + if err := LogoutCmd.Flags().Set("output", "text"); err != nil { + t.Fatal(err) + } + t.Cleanup(func() { + _ = LogoutCmd.Flags().Set("yes", "false") + _ = LogoutCmd.Flags().Set("output", "text") + }) + + err := LogoutCmd.RunE(LogoutCmd, nil) + if err == nil || !strings.Contains(err.Error(), "pb profile default ") { + t.Fatalf("expected non-interactive ambiguity error, got %v", err) + } + + saved, err := config.ReadConfigFromFile() + if err != nil { + t.Fatal(err) + } + if len(saved.Profiles) != 2 || saved.DefaultProfile != "" { + t.Fatalf("config changed after ambiguous logout: %#v", saved) + } +} diff --git a/cmd/profile.go b/cmd/profile.go index 008f92b..4b3464e 100644 --- a/cmd/profile.go +++ b/cmd/profile.go @@ -20,6 +20,7 @@ import ( "errors" "fmt" "net/url" + "strings" "time" tea "github.com/charmbracelet/bubbletea" @@ -34,31 +35,69 @@ type ProfileListItem struct { title, url, user string } +type profileOutput struct { + URL string `json:"url"` + Username string `json:"username,omitempty"` + Cloud bool `json:"cloud"` + TenantID string `json:"tenant_id,omitempty"` + IngestURL string `json:"ingest_url,omitempty"` + WorkspaceID string `json:"workspace_id,omitempty"` + WorkspaceName string `json:"workspace_name,omitempty"` + OrchestratorURL string `json:"orchestrator_url,omitempty"` +} + +func safeProfileOutput(profile config.Profile) profileOutput { + return profileOutput{ + URL: profile.URL, + Username: profile.Username, + Cloud: profile.Cloud, + TenantID: profile.TenantID, + IngestURL: profile.IngestURL, + WorkspaceID: profile.WorkspaceID, + WorkspaceName: profile.WorkspaceName, + OrchestratorURL: profile.OrchestratorURL, + } +} + +func safeProfilesOutput(profiles map[string]config.Profile) map[string]profileOutput { + result := make(map[string]profileOutput, len(profiles)) + for name, profile := range profiles { + result[name] = safeProfileOutput(profile) + } + return result +} + func (item *ProfileListItem) Render(highlight bool) string { if highlight { - render := fmt.Sprintf( - "%s\n%s\n%s", + lines := []string{ SelectedStyle.Render(item.title), SelectedStyleAlt.Render(fmt.Sprintf("url: %s", item.url)), - SelectedStyleAlt.Render(fmt.Sprintf("user: %s", item.user)), - ) - return SelectedItemOuter.Render(render) + } + if item.user != "" { + lines = append(lines, SelectedStyleAlt.Render(fmt.Sprintf("user: %s", item.user))) + } + return SelectedItemOuter.Render(strings.Join(lines, "\n")) } - render := fmt.Sprintf( - "%s\n%s\n%s", + lines := []string{ StandardStyle.Render(item.title), StandardStyleAlt.Render(fmt.Sprintf("url: %s", item.url)), - StandardStyleAlt.Render(fmt.Sprintf("user: %s", item.user)), - ) - return ItemOuter.Render(render) + } + if item.user != "" { + lines = append(lines, StandardStyleAlt.Render(fmt.Sprintf("user: %s", item.user))) + } + return ItemOuter.Render(strings.Join(lines, "\n")) } // Add an output flag to specify the output format. -var outputFormat string +var ( + outputFormat string + profileAPIKey string +) // Initialize flags func init() { AddProfileCmd.Flags().StringVarP(&outputFormat, "output", "o", "", "Output format (text|json)") + AddProfileCmd.Flags().StringVar(&profileAPIKey, "api-key", "", "API key for self-hosted authentication") RemoveProfileCmd.Flags().StringVarP(&outputFormat, "output", "o", "", "Output format (text|json)") DefaultProfileCmd.Flags().StringVarP(&outputFormat, "output", "o", "", "Output format (text|json)") ListProfileCmd.Flags().StringVarP(&outputFormat, "output", "o", "", "Output format (text|json)") @@ -78,14 +117,17 @@ func outputResult(v interface{}) error { } var AddProfileCmd = &cobra.Command{ - Use: "add profile-name url ", - Example: " pb profile add local_parseable http://0.0.0.0:8000 admin admin", + Use: "add profile-name url [username] [password]", + Example: " pb profile add local_parseable http://0.0.0.0:8000 admin admin\n pb profile add local_parseable http://0.0.0.0:8000 --api-key psk_xxx", Short: "Add a new profile", Long: "Add a new profile to the config file", Args: func(cmd *cobra.Command, args []string) error { if err := cobra.MinimumNArgs(2)(cmd, args); err != nil { return err } + if strings.TrimSpace(profileAPIKey) != "" && len(args) != 2 { + return errors.New("--api-key cannot be combined with username/password") + } return cobra.MaximumNArgs(4)(cmd, args) }, RunE: func(cmd *cobra.Command, args []string) error { @@ -104,22 +146,29 @@ var AddProfileCmd = &cobra.Command{ return commandError } - var username, password string - if len(args) < 4 { - _m, err := tea.NewProgram(credential.New()).Run() - if err != nil { - commandError = fmt.Errorf("error reading credentials: %s", err) - cmd.Annotations["error"] = commandError.Error() - return commandError - } - m := _m.(credential.Model) - username, password = m.Values() + apiKey := strings.TrimSpace(profileAPIKey) + profile := config.Profile{Cloud: false, URL: url.String()} + if apiKey != "" { + profile.APIKey = apiKey } else { - username = args[2] - password = args[3] + var username, password string + if len(args) < 4 { + _m, err := tea.NewProgram(credential.New()).Run() + if err != nil { + commandError = fmt.Errorf("error reading credentials: %s", err) + cmd.Annotations["error"] = commandError.Error() + return commandError + } + m := _m.(credential.Model) + username, password = m.Values() + } else { + username = args[2] + password = args[3] + } + profile.Username = username + profile.Password = password } - profile := config.Profile{URL: url.String(), Username: username, Password: password} fileConfig, err := config.ReadConfigFromFile() if err != nil { newConfig := config.Config{ @@ -133,9 +182,7 @@ var AddProfileCmd = &cobra.Command{ fileConfig.Profiles = make(map[string]config.Profile) } fileConfig.Profiles[name] = profile - if fileConfig.DefaultProfile == "" { - fileConfig.DefaultProfile = name - } + fileConfig.DefaultProfile = name commandError = config.WriteConfigToFile(fileConfig) } @@ -146,7 +193,7 @@ var AddProfileCmd = &cobra.Command{ } if outputFormat == "json" { - return outputResult(profile) + return outputResult(safeProfileOutput(profile)) } fmt.Printf("Profile %s added successfully\n", name) return nil @@ -320,7 +367,7 @@ var UpdateProfileCmd = &cobra.Command{ } if outputFormat == "json" { - return outputResult(profile) + return outputResult(safeProfileOutput(profile)) } fmt.Printf("Profile '%s' URL updated to %s\n", name, rawURL) return nil @@ -345,7 +392,7 @@ var ListProfileCmd = &cobra.Command{ } if outputFormat == "json" { - commandError := outputResult(fileConfig.Profiles) + commandError := outputResult(safeProfilesOutput(fileConfig.Profiles)) cmd.Annotations["executionTime"] = time.Since(startTime).String() if commandError != nil { cmd.Annotations["error"] = commandError.Error() diff --git a/cmd/promql.go b/cmd/promql.go index b99faa8..eb670ce 100644 --- a/cmd/promql.go +++ b/cmd/promql.go @@ -23,6 +23,7 @@ import ( "io" "net/http" "net/url" + "sort" "strings" "time" @@ -44,6 +45,9 @@ var PromqlCmd = &cobra.Command{ Example: " pb promql run -i\n pb promql run \"http_requests_total\" --dataset otel_metrics --from=1h -i", } +// PromqlCardinalityCmd exposes the cardinality parent for shared CLI output setup. +func PromqlCardinalityCmd() *cobra.Command { return promqlCardinalityCmd } + func init() { PromqlCmd.SetHelpFunc(renderPromqlHelp) @@ -115,6 +119,7 @@ func init() { promqlCardinalityActiveSeriesCmd.Flags().StringP("output", "o", "text", "Output format: text or json") // flags: tsdb + promqlActiveQueriesCmd.Flags().StringP("output", "o", "text", "Output format: text or json") promqlTSDBCmd.Flags().StringP("dataset", "d", defaultMetricsStream, "Metrics dataset") promqlTSDBCmd.Flags().Int("top", 10, "Max entries per category") promqlTSDBCmd.Flags().String("date", "", "Date to analyze (YYYY-MM-DD, defaults to today)") @@ -179,11 +184,10 @@ func renderPromqlHelp(cmd *cobra.Command, _ []string) { // --------------------------------------------------------------------------- func promqlGet(path string, params url.Values) ([]byte, error) { - client := internalHTTP.DefaultClient(&DefaultProfile) - client.Client.Timeout = 120 * time.Second - client.Client.Transport = &http.Transport{ + client := internalHTTP.DefaultClientWithTransport(&DefaultProfile, &http.Transport{ TLSNextProto: make(map[string]func(string, *tls.Conn) http.RoundTripper), - } + }) + client.Client.Timeout = 120 * time.Second reqURL, err := url.JoinPath(DefaultProfile.URL, path) if err != nil { return nil, err @@ -195,10 +199,8 @@ func promqlGet(path string, params url.Values) ([]byte, error) { if err != nil { return nil, err } - if DefaultProfile.Token != "" { - req.Header.Set("Authorization", "Bearer "+DefaultProfile.Token) - } else { - req.SetBasicAuth(DefaultProfile.Username, DefaultProfile.Password) + if err := internalHTTP.AddAuthHeaders(req, &DefaultProfile); err != nil { + return nil, err } stopSpinner := startSpinner() resp, err := client.Client.Do(req) @@ -378,13 +380,16 @@ func promqlInteractiveFromDefault(from string, interactive, fromChanged bool) st // --------------------------------------------------------------------------- var promqlLabelsCmd = &cobra.Command{ - Use: "labels", + Use: "labels [stream]", Short: "List all label names in a metrics stream", - Example: " pb promql labels --dataset otel_metrics", - Args: cobra.NoArgs, + Example: " pb promql labels otel_metrics\n pb promql labels --dataset otel_metrics", + Args: cobra.MaximumNArgs(1), PreRunE: PreRunDefaultProfile, - RunE: func(cmd *cobra.Command, _ []string) error { + RunE: func(cmd *cobra.Command, args []string) error { stream, _ := cmd.Flags().GetString("dataset") + if len(args) > 0 { + stream = args[0] + } outputFmt, _ := cmd.Flags().GetString("output") params := url.Values{} @@ -413,8 +418,12 @@ var promqlLabelsCmd = &cobra.Command{ if resp.Status == "error" { return fmt.Errorf("%s", resp.Error) } + if len(resp.Data) == 0 { + fmt.Println("No labels found.") + return nil + } for _, label := range resp.Data { - fmt.Println((&DatasetListItem{Name: label}).Render()) + fmt.Println(ItemOuter.Render(fmt.Sprintf("%s %s", SelectedStyle.Render("•"), StandardStyle.Render(label)))) } return nil }, @@ -425,14 +434,17 @@ var promqlLabelsCmd = &cobra.Command{ // --------------------------------------------------------------------------- var promqlLabelValuesCmd = &cobra.Command{ - Use: "label-values [label_name]", + Use: "label-values [label_name] [stream]", Short: "List distinct values for a label", - Example: " pb promql label-values job --dataset otel_metrics\n pb promql label-values __name__ --dataset otel_metrics", - Args: cobra.ExactArgs(1), + Example: " pb promql label-values job otel_metrics\n pb promql label-values job --dataset otel_metrics\n pb promql label-values __name__ --dataset otel_metrics", + Args: cobra.RangeArgs(1, 2), PreRunE: PreRunDefaultProfile, RunE: func(cmd *cobra.Command, args []string) error { label := args[0] stream, _ := cmd.Flags().GetString("dataset") + if len(args) > 1 { + stream = args[1] + } outputFmt, _ := cmd.Flags().GetString("output") params := url.Values{} @@ -461,8 +473,12 @@ var promqlLabelValuesCmd = &cobra.Command{ if resp.Status == "error" { return fmt.Errorf("%s", resp.Error) } + if len(resp.Data) == 0 { + fmt.Println("No label values found.") + return nil + } for _, v := range resp.Data { - fmt.Println(v) + fmt.Println(ItemOuter.Render(fmt.Sprintf("%s %s", SelectedStyle.Render("•"), StandardStyle.Render(v)))) } fmt.Printf("\nlabel=%s total=%d\n", label, len(resp.Data)) return nil @@ -474,13 +490,16 @@ var promqlLabelValuesCmd = &cobra.Command{ // --------------------------------------------------------------------------- var promqlSeriesCmd = &cobra.Command{ - Use: "series", + Use: "series [stream]", Short: "Find time series matching a label selector", - Example: " pb promql series --match 'http_requests_total' --dataset otel_metrics\n pb promql series --match '{job=\"api\"}' --dataset otel_metrics", - Args: cobra.NoArgs, + Example: " pb promql series otel_metrics --match 'http_requests_total'\n pb promql series --match '{job=\"api\"}' --dataset otel_metrics", + Args: cobra.MaximumNArgs(1), PreRunE: PreRunDefaultProfile, - RunE: func(cmd *cobra.Command, _ []string) error { + RunE: func(cmd *cobra.Command, args []string) error { stream, _ := cmd.Flags().GetString("dataset") + if len(args) > 0 { + stream = args[0] + } matchers, _ := cmd.Flags().GetStringArray("match") outputFmt, _ := cmd.Flags().GetString("output") @@ -517,8 +536,12 @@ var promqlSeriesCmd = &cobra.Command{ if resp.Status == "error" { return fmt.Errorf("%s", resp.Error) } + if len(resp.Data) == 0 { + fmt.Println("No series found.") + return nil + } for _, series := range resp.Data { - fmt.Println(formatPromqlLabels(series)) + printPromqlSeries(series) } fmt.Printf("\ntotal=%d\n", len(resp.Data)) return nil @@ -542,13 +565,16 @@ type cardinalityEntry struct { // cardinality label-names var promqlCardinalityLabelNamesCmd = &cobra.Command{ - Use: "label-names", + Use: "label-names [stream]", Short: "Labels with the highest number of distinct values", - Example: " pb promql cardinality label-names --dataset otel_metrics --limit 20", - Args: cobra.NoArgs, + Example: " pb promql cardinality label-names otel_metrics\n pb promql cardinality label-names --dataset otel_metrics --limit 20", + Args: cobra.MaximumNArgs(1), PreRunE: PreRunDefaultProfile, - RunE: func(cmd *cobra.Command, _ []string) error { + RunE: func(cmd *cobra.Command, args []string) error { stream, _ := cmd.Flags().GetString("dataset") + if len(args) > 0 { + stream = args[0] + } lookback, _ := cmd.Flags().GetInt("lookback") limit, _ := cmd.Flags().GetInt("limit") selector, _ := cmd.Flags().GetString("selector") @@ -590,14 +616,20 @@ var promqlCardinalityLabelNamesCmd = &cobra.Command{ // cardinality label-values var promqlCardinalityLabelValuesCmd = &cobra.Command{ - Use: "label-values", + Use: "label-values [stream] [label]", Short: "Series count per value for a specific label", - Example: " pb promql cardinality label-values --label job --dataset otel_metrics", - Args: cobra.NoArgs, + Example: " pb promql cardinality label-values otel_metrics service.name\n pb promql cardinality label-values --label job --dataset otel_metrics", + Args: cobra.MaximumNArgs(2), PreRunE: PreRunDefaultProfile, - RunE: func(cmd *cobra.Command, _ []string) error { + RunE: func(cmd *cobra.Command, args []string) error { stream, _ := cmd.Flags().GetString("dataset") + if len(args) > 0 { + stream = args[0] + } labelName, _ := cmd.Flags().GetString("label") + if len(args) > 1 { + labelName = args[1] + } lookback, _ := cmd.Flags().GetInt("lookback") limit, _ := cmd.Flags().GetInt("limit") outputFmt, _ := cmd.Flags().GetString("output") @@ -681,16 +713,22 @@ func printCardinalityEntryTable(nameHeader, valueHeader string, entries []cardin // cardinality active-series var promqlCardinalityActiveSeriesCmd = &cobra.Command{ - Use: "active-series", + Use: "active-series [stream] [selector]", Short: "List currently active series", - Example: " pb promql cardinality active-series --dataset otel_metrics --selector '{job=\"api\"}'", - Args: cobra.NoArgs, + Example: " pb promql cardinality active-series otel_metrics '{job=\"api\"}'\n pb promql cardinality active-series --dataset otel_metrics --selector '{job=\"api\"}'", + Args: cobra.MaximumNArgs(2), PreRunE: PreRunDefaultProfile, - RunE: func(cmd *cobra.Command, _ []string) error { + RunE: func(cmd *cobra.Command, args []string) error { stream, _ := cmd.Flags().GetString("dataset") + if len(args) > 0 { + stream = args[0] + } lookback, _ := cmd.Flags().GetInt("lookback") limit, _ := cmd.Flags().GetInt("limit") selector, _ := cmd.Flags().GetString("selector") + if len(args) > 1 { + selector = args[1] + } outputFmt, _ := cmd.Flags().GetString("output") params := url.Values{} @@ -725,9 +763,15 @@ var promqlCardinalityActiveSeriesCmd = &cobra.Command{ if resp.Status == "error" { return fmt.Errorf("%s", resp.Error) } - fmt.Printf("total_active_series=%d\n\n", resp.Data.TotalActiveSeries) + fmt.Println(SelectedStyle.Bold(true).Render("ACTIVE SERIES")) + fmt.Printf("total=%d\n\n", resp.Data.TotalActiveSeries) + if len(resp.Data.Series) == 0 { + fmt.Println("No active series found.") + return nil + } for _, s := range resp.Data.Series { - fmt.Println(formatPromqlLabels(s)) + printPromqlSeries(s) + fmt.Println() } return nil }, @@ -741,14 +785,20 @@ var promqlActiveQueriesCmd = &cobra.Command{ Use: "active-queries", Aliases: []string{"ps"}, Short: "Show currently executing PromQL queries", - Example: " pb promql active-queries", + Example: " pb promql active-queries\n pb promql active-queries -o json", Args: cobra.NoArgs, PreRunE: PreRunDefaultProfile, - RunE: func(_ *cobra.Command, _ []string) error { + RunE: func(cmd *cobra.Command, _ []string) error { + outputFmt, _ := cmd.Flags().GetString("output") + body, err := promqlGet("prometheus/api/v1/status/active_queries", nil) if err != nil { return err } + if outputFmt == "json" { + printRawJSON(body) + return nil + } var resp struct { Status string `json:"status"` @@ -789,13 +839,16 @@ var promqlActiveQueriesCmd = &cobra.Command{ // --------------------------------------------------------------------------- var promqlTSDBCmd = &cobra.Command{ - Use: "tsdb", + Use: "tsdb [stream]", Short: "Show TSDB statistics for a metrics stream", - Example: " pb promql tsdb --dataset otel_metrics\n pb promql tsdb --dataset otel_metrics --top 20 --focus-label job", - Args: cobra.NoArgs, + Example: " pb promql tsdb otel_metrics\n pb promql tsdb --dataset otel_metrics\n pb promql tsdb --dataset otel_metrics --top 20 --focus-label job", + Args: cobra.MaximumNArgs(1), PreRunE: PreRunDefaultProfile, - RunE: func(cmd *cobra.Command, _ []string) error { + RunE: func(cmd *cobra.Command, args []string) error { stream, _ := cmd.Flags().GetString("dataset") + if len(args) > 0 { + stream = args[0] + } topN, _ := cmd.Flags().GetInt("top") date, _ := cmd.Flags().GetString("date") focusLabel, _ := cmd.Flags().GetString("focus-label") @@ -820,27 +873,37 @@ var promqlTSDBCmd = &cobra.Command{ return nil } - var resp struct { - Status string `json:"status"` - Data struct { - TotalSeries int `json:"totalSeries"` - TotalLabelValuePairs int `json:"totalLabelValuePairs"` - SeriesByMetric []cardinalityEntry `json:"seriesCountByMetricName"` - SeriesByLabel []cardinalityEntry `json:"seriesCountByLabelName"` - SeriesByFocusLabel []cardinalityEntry `json:"seriesCountByFocusLabelValue"` - LabelValueCount []cardinalityEntry `json:"labelValueCountByLabelName"` - } `json:"data"` - Error string `json:"error,omitempty"` + var envelope struct { + Status string `json:"status"` + Data json.RawMessage `json:"data"` + Error string `json:"error,omitempty"` } - if err := json.Unmarshal(body, &resp); err != nil { + if err := json.Unmarshal(body, &envelope); err != nil { fmt.Println(string(body)) return nil } - if resp.Status == "error" { - return fmt.Errorf("%s", resp.Error) + if envelope.Status == "error" { + return fmt.Errorf("%s", envelope.Error) + } + if isEmptyJSONList(envelope.Data) { + fmt.Printf("No TSDB stats found. Pass a metrics dataset name, for example: pb promql tsdb \n") + return nil + } + + var data struct { + TotalSeries int `json:"totalSeries"` + TotalLabelValuePairs int `json:"totalLabelValuePairs"` + SeriesByMetric []cardinalityEntry `json:"seriesCountByMetricName"` + SeriesByLabel []cardinalityEntry `json:"seriesCountByLabelName"` + SeriesByFocusLabel []cardinalityEntry `json:"seriesCountByFocusLabelValue"` + LabelValueCount []cardinalityEntry `json:"labelValueCountByLabelName"` + } + if err := json.Unmarshal(envelope.Data, &data); err != nil { + fmt.Println(string(body)) + return nil } - d := resp.Data + d := data printTSDBSummary(d.TotalSeries, d.TotalLabelValuePairs) if len(d.SeriesByMetric) > 0 { @@ -859,6 +922,11 @@ var promqlTSDBCmd = &cobra.Command{ }, } +func isEmptyJSONList(data json.RawMessage) bool { + var list []json.RawMessage + return json.Unmarshal(data, &list) == nil && len(list) == 0 +} + func printTSDBSummary(totalSeries, totalLabelValuePairs int) { headerStyle := SelectedStyle.Bold(true) bodyStyle := StandardStyle @@ -905,6 +973,7 @@ func formatPromqlLabels(m map[string]string) string { labels = append(labels, k+"=\""+v+"\"") } } + sort.Strings(labels) if len(labels) == 0 { return name } @@ -914,6 +983,25 @@ func formatPromqlLabels(m map[string]string) string { return fmt.Sprintf("%s{%s}", name, strings.Join(labels, ", ")) } +func printPromqlSeries(m map[string]string) { + name := m["__name__"] + if name == "" { + name = "(unnamed series)" + } + fmt.Println(ItemOuter.Render(fmt.Sprintf("%s %s", SelectedStyle.Render("•"), StandardStyle.Render(name)))) + + keys := make([]string, 0, len(m)) + for k := range m { + if k != "__name__" { + keys = append(keys, k) + } + } + sort.Strings(keys) + for _, k := range keys { + fmt.Printf(" %s=%q\n", k, m[k]) + } +} + func promqlTS(v any) string { if f, ok := v.(float64); ok { return time.Unix(int64(f), 0).UTC().Format("2006-01-02T15:04:05Z") diff --git a/cmd/promql_test.go b/cmd/promql_test.go index 45fdb6a..ec05f64 100644 --- a/cmd/promql_test.go +++ b/cmd/promql_test.go @@ -1,6 +1,15 @@ package cmd -import "testing" +import ( + "fmt" + "net/http" + "net/http/httptest" + "net/url" + "testing" + + "github.com/parseablehq/pb/pkg/config" + "github.com/spf13/cobra" +) func TestPromqlInteractiveFromDefaultUsesTenMinutes(t *testing.T) { got := promqlInteractiveFromDefault("5m", true, false) @@ -22,3 +31,114 @@ func TestPromqlInteractiveFromDefaultKeepsNonInteractiveDefault(t *testing.T) { t.Fatalf("got %q, want 5m", got) } } + +func TestPromqlPositionalArgumentsOverrideFlags(t *testing.T) { + requests := make(chan *url.URL, 1) + server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + requestURL := *r.URL + requests <- &requestURL + w.Header().Set("Content-Type", "application/json") + fmt.Fprint(w, `{"status":"success","data":[]}`) + })) + defer server.Close() + + originalProfile := DefaultProfile + DefaultProfile = config.Profile{URL: server.URL, APIKey: "test-key"} + t.Cleanup(func() { DefaultProfile = originalProfile }) + + tests := []struct { + name string + cmd *cobra.Command + args []string + flagSetup map[string]string + wantPath string + wantQuery map[string]string + }{ + { + name: "labels stream", + cmd: promqlLabelsCmd, + args: []string{"positional-labels"}, + wantPath: "/prometheus/api/v1/labels", + wantQuery: map[string]string{"stream": "positional-labels"}, + }, + { + name: "label values label and stream", + cmd: promqlLabelValuesCmd, + args: []string{"job", "positional-label-values"}, + wantPath: "/prometheus/api/v1/label/job/values", + wantQuery: map[string]string{"stream": "positional-label-values"}, + }, + { + name: "series stream", + cmd: promqlSeriesCmd, + args: []string{"positional-series"}, + flagSetup: map[string]string{ + "match": "metric_name", + }, + wantPath: "/prometheus/api/v1/series", + wantQuery: map[string]string{ + "stream": "positional-series", + "match[]": "metric_name", + }, + }, + { + name: "cardinality label values stream and label", + cmd: promqlCardinalityLabelValuesCmd, + args: []string{"positional-cardinality", "service.name"}, + flagSetup: map[string]string{ + "label": "flag-label", + }, + wantPath: "/prometheus/api/v1/cardinality/label_values", + wantQuery: map[string]string{"stream": "positional-cardinality", "label_name": "service.name"}, + }, + { + name: "active series stream and selector", + cmd: promqlCardinalityActiveSeriesCmd, + args: []string{"positional-active", `{job="api"}`}, + flagSetup: map[string]string{ + "selector": `{job="flag-value"}`, + }, + wantPath: "/prometheus/api/v1/cardinality/active_series", + wantQuery: map[string]string{"stream": "positional-active", "selector": `{job="api"}`}, + }, + { + name: "tsdb stream", + cmd: promqlTSDBCmd, + args: []string{"positional-tsdb"}, + wantPath: "/prometheus/api/v1/status/tsdb", + wantQuery: map[string]string{"stream": "positional-tsdb"}, + }, + } + + for _, test := range tests { + t.Run(test.name, func(t *testing.T) { + if err := test.cmd.Flags().Set("dataset", "flag-dataset"); err != nil { + t.Fatal(err) + } + if err := test.cmd.Flags().Set("output", "json"); err != nil { + t.Fatal(err) + } + for name, value := range test.flagSetup { + if err := test.cmd.Flags().Set(name, value); err != nil { + t.Fatal(err) + } + } + if err := test.cmd.Args(test.cmd, test.args); err != nil { + t.Fatal(err) + } + if err := test.cmd.RunE(test.cmd, test.args); err != nil { + t.Fatal(err) + } + + requestURL := <-requests + if requestURL.Path != test.wantPath { + t.Fatalf("path=%q want=%q", requestURL.Path, test.wantPath) + } + for key, want := range test.wantQuery { + if got := requestURL.Query().Get(key); got != want { + t.Fatalf("query %s=%q want=%q", key, got, want) + } + } + }) + } +} diff --git a/cmd/queryList.go b/cmd/queryList.go index 3938462..49ba40a 100644 --- a/cmd/queryList.go +++ b/cmd/queryList.go @@ -19,9 +19,7 @@ import ( "encoding/json" "fmt" "io" - "net/http" "strings" - "time" "github.com/parseablehq/pb/pkg/config" internalHTTP "github.com/parseablehq/pb/pkg/http" @@ -52,10 +50,8 @@ var SavedQueryList = &cobra.Command{ userProfile = profile } - client := &http.Client{ - Timeout: time.Second * 60, - } - userSavedQueries := fetchFilters(client, &userProfile) + client := internalHTTP.DefaultClient(&userProfile) + userSavedQueries := fetchFilters(&client) // Collect all filter titles in a slice and join with commas var filterDetails []string @@ -182,17 +178,14 @@ type Item struct { To string `json:"to,omitempty"` } -func fetchFilters(client *http.Client, profile *config.Profile) []Item { - endpoint := fmt.Sprintf("%s/%s", profile.URL, "api/v1/filters") - req, err := http.NewRequest("GET", endpoint, nil) +func fetchFilters(client *internalHTTP.HTTPClient) []Item { + req, err := client.NewRequest("GET", "filters", nil) if err != nil { fmt.Println("Error creating request:", err) return nil } - req.SetBasicAuth(profile.Username, profile.Password) - req.Header.Add("Content-Type", "application/json") - resp, err := client.Do(req) + resp, err := client.Client.Do(req) if err != nil { fmt.Println("Error making request:", err) return nil diff --git a/cmd/status.go b/cmd/status.go index 599deac..3670991 100644 --- a/cmd/status.go +++ b/cmd/status.go @@ -16,6 +16,8 @@ package cmd import ( + "encoding/json" + "errors" "fmt" "strings" @@ -30,32 +32,61 @@ import ( var StatusCmd = &cobra.Command{ Use: "status", Short: "Check connection status for the active profile", - Example: " pb status", - RunE: func(_ *cobra.Command, _ []string) error { + Example: " pb status\n pb status -o json", + RunE: func(cmd *cobra.Command, _ []string) error { + outputFormat, err := cmd.Flags().GetString("output") + if err != nil { + return err + } + fileConfig, err := config.ReadConfigFromFile() if err != nil { - return fmt.Errorf("no profile configured. run: pb login") + return statusPreflightError(outputFormat, "no profile configured. run: pb login") } profileName := fileConfig.DefaultProfile profile, exists := fileConfig.Profiles[profileName] if !exists || profileName == "" { - return fmt.Errorf("no active profile. run: pb login") + return statusPreflightError(outputFormat, "no active profile. run: pb login") } - fmt.Printf("Profile : %s\n", profileName) - fmt.Printf("URL : %s\n", profile.URL) + if outputFormat != "json" { + fmt.Printf("Profile : %s\n", profileName) + fmt.Printf("URL : %s\n", profile.URL) + } client := internalHTTP.DefaultClient(&profile) about, err := analytics.FetchAbout(&client) if err != nil { statusMessage := statusErrorMessage(err) - errStyle := lipgloss.NewStyle().Foreground(ui.Active.Err).Bold(true) - fmt.Printf("Status : %s\n", errStyle.Render("✗ Not connected")) - fmt.Printf("Error : %s\n", statusMessage) + if outputFormat == "json" { + if jsonErr := printStatusJSON(statusOutput{ + Status: "error", + Healthy: false, + Profile: profileName, + URL: profile.URL, + Error: statusMessage, + }); jsonErr != nil { + return jsonErr + } + } else { + errStyle := lipgloss.NewStyle().Foreground(ui.Active.Err).Bold(true) + fmt.Printf("Status : %s\n", errStyle.Render("✗ Not connected")) + fmt.Printf("Error : %s\n", statusMessage) + } return fmt.Errorf("status check failed: %s", statusMessage) } + if outputFormat == "json" { + return printStatusJSON(statusOutput{ + Status: "ok", + Healthy: true, + Profile: profileName, + URL: profile.URL, + Version: about.Version, + }) + } + okStyle := lipgloss.NewStyle().Foreground(ui.Active.Ok).Bold(true) fmt.Printf("Status : %s\n", okStyle.Render("✓ Connected")) fmt.Printf("Version : %s\n", about.Version) @@ -63,6 +94,37 @@ var StatusCmd = &cobra.Command{ }, } +type statusOutput struct { + Status string `json:"status"` + Healthy bool `json:"healthy"` + Profile string `json:"profile,omitempty"` + URL string `json:"url,omitempty"` + Version string `json:"version,omitempty"` + Error string `json:"error,omitempty"` +} + +func statusPreflightError(outputFormat, message string) error { + if outputFormat == "json" { + if err := printStatusJSON(statusOutput{ + Status: "error", + Healthy: false, + Error: message, + }); err != nil { + return err + } + } + return errors.New(message) +} + +func printStatusJSON(result statusOutput) error { + jsonData, err := json.MarshalIndent(result, "", " ") + if err != nil { + return fmt.Errorf("failed to marshal status JSON: %w", err) + } + fmt.Println(string(jsonData)) + return nil +} + func statusErrorMessage(err error) string { message := err.Error() if strings.Contains(message, "Status Code: 401") || strings.Contains(message, "Status Code: 403") { @@ -70,3 +132,7 @@ func statusErrorMessage(err error) string { } return message } + +func init() { + StatusCmd.Flags().StringP("output", "o", "text", "Output format (text|json)") +} diff --git a/cmd/status_test.go b/cmd/status_test.go index 9aaf10c..529fad5 100644 --- a/cmd/status_test.go +++ b/cmd/status_test.go @@ -1,8 +1,13 @@ package cmd import ( + "encoding/json" "errors" + "io" + "os" "testing" + + "github.com/parseablehq/pb/pkg/config" ) func TestStatusErrorMessageTreatsAuthFailuresAsCredentialErrors(t *testing.T) { @@ -21,3 +26,62 @@ func TestStatusErrorMessageKeepsOtherErrors(t *testing.T) { t.Fatalf("got %q, want %q", got, input.Error()) } } + +func TestStatusJSONWhenConfigIsMissing(t *testing.T) { + t.Setenv("XDG_CONFIG_HOME", t.TempDir()) + result, err := runStatusJSON(t) + if err == nil { + t.Fatal("expected nonzero status error") + } + if result.Status != "error" || result.Healthy || result.Error != "no profile configured. run: pb login" { + t.Fatalf("unexpected result: %#v", result) + } +} + +func TestStatusJSONWhenDefaultProfileIsMissing(t *testing.T) { + t.Setenv("XDG_CONFIG_HOME", t.TempDir()) + if err := config.WriteConfigToFile(&config.Config{ + Profiles: map[string]config.Profile{ + "available": {URL: "https://example.com"}, + }, + }); err != nil { + t.Fatal(err) + } + + result, err := runStatusJSON(t) + if err == nil { + t.Fatal("expected nonzero status error") + } + if result.Status != "error" || result.Healthy || result.Error != "no active profile. run: pb login" { + t.Fatalf("unexpected result: %#v", result) + } +} + +func runStatusJSON(t *testing.T) (statusOutput, error) { + t.Helper() + if err := StatusCmd.Flags().Set("output", "json"); err != nil { + t.Fatal(err) + } + t.Cleanup(func() { _ = StatusCmd.Flags().Set("output", "text") }) + + reader, writer, err := os.Pipe() + if err != nil { + t.Fatal(err) + } + originalStdout := os.Stdout + os.Stdout = writer + commandErr := StatusCmd.RunE(StatusCmd, nil) + _ = writer.Close() + os.Stdout = originalStdout + + data, readErr := io.ReadAll(reader) + _ = reader.Close() + if readErr != nil { + t.Fatal(readErr) + } + var result statusOutput + if err := json.Unmarshal(data, &result); err != nil { + t.Fatalf("invalid JSON %q: %v", data, err) + } + return result, commandErr +} diff --git a/cmd/tail.go b/cmd/tail.go index e0b17c2..1d9904e 100644 --- a/cmd/tail.go +++ b/cmd/tail.go @@ -19,11 +19,14 @@ package cmd import ( "bytes" "context" + "crypto/tls" "encoding/base64" "encoding/json" "fmt" "io" + "net/url" "os" + "strings" "time" "github.com/apache/arrow/go/v13/arrow/array" @@ -34,6 +37,7 @@ import ( "github.com/spf13/cobra" "google.golang.org/grpc" "google.golang.org/grpc/codes" + "google.golang.org/grpc/credentials" "google.golang.org/grpc/credentials/insecure" "google.golang.org/grpc/metadata" "google.golang.org/grpc/status" @@ -45,56 +49,81 @@ var TailCmd = &cobra.Command{ Short: "Stream live events from a dataset", Args: cobra.ExactArgs(1), PreRunE: PreRunDefaultProfile, - RunE: func(_ *cobra.Command, args []string) error { + RunE: func(cmd *cobra.Command, args []string) error { + output, err := cmd.Flags().GetString("output") + if err != nil { + return err + } + if output != "text" && output != "json" { + return fmt.Errorf("unsupported output format %q (expected text or json)", output) + } name := args[0] profile := DefaultProfile - return tail(profile, name) + return tail(profile, name, output == "json") }, } -func tail(profile config.Profile, stream string) error { +func init() { + TailCmd.Flags().StringP("output", "o", "text", "Output format (text|json)") +} + +func tail(profile config.Profile, stream string, jsonOutput bool) error { payload, _ := json.Marshal(struct { Stream string `json:"stream"` }{ Stream: stream, }) - stopConnect := tailSpinner("connecting...") - httpClient := internalHTTP.DefaultClient(&DefaultProfile) + stopConnect := func() {} + if !jsonOutput { + stopConnect = tailSpinner("connecting...") + } + httpClient := internalHTTP.DefaultClient(&profile) about, err := analytics.FetchAbout(&httpClient) stopConnect() if err != nil { return err } url := profile.GrpcAddr(fmt.Sprint(about.GRPCPort)) + transportCredentials, err := tailTransportCredentials(profile) + if err != nil { + return err + } - flightClient, err := flight.NewClientWithMiddleware(url, nil, nil, grpc.WithTransportCredentials(insecure.NewCredentials())) + flightClient, err := flight.NewClientWithMiddleware(url, nil, nil, grpc.WithTransportCredentials(transportCredentials)) if err != nil { return err } - authHeader := basicAuth(profile.Username, profile.Password) + authMetadata, err := tailAuthMetadata(profile) + if err != nil { + return err + } watching := func() { - fmt.Fprintf(os.Stderr, "\r\033[K● watching %s... (ctrl+c to stop)", stream) + if !jsonOutput { + fmt.Fprintf(os.Stderr, "\r\033[K● watching %s... (ctrl+c to stop)", stream) + } } watching() for { resp, err := flightClient.DoGet( - metadata.NewOutgoingContext(context.Background(), metadata.New(map[string]string{ - "Authorization": "Basic " + authHeader, - })), + metadata.NewOutgoingContext(context.Background(), authMetadata), &flight.Ticket{Ticket: payload}, ) if err != nil { - fmt.Fprint(os.Stderr, "\r\033[K") + if !jsonOutput { + fmt.Fprint(os.Stderr, "\r\033[K") + } return err } records, err := flight.NewRecordReader(resp) if err != nil { - fmt.Fprint(os.Stderr, "\r\033[K") + if !jsonOutput { + fmt.Fprint(os.Stderr, "\r\033[K") + } return err } @@ -105,10 +134,14 @@ func tail(profile config.Profile, stream string) error { if isStreamEnd(err) { break } - fmt.Fprint(os.Stderr, "\r\033[K") + if !jsonOutput { + fmt.Fprint(os.Stderr, "\r\033[K") + } return err } - fmt.Fprint(os.Stderr, "\r\033[K") // clear watching line before printing record + if !jsonOutput { + fmt.Fprint(os.Stderr, "\r\033[K") + } // clear watching line before printing record var buf bytes.Buffer array.RecordToJSON(record, &buf) fmt.Println(buf.String()) @@ -119,6 +152,53 @@ func tail(profile config.Profile, stream string) error { } } +func tailTransportCredentials(profile config.Profile) (credentials.TransportCredentials, error) { + profileURL, err := url.Parse(profile.URL) + if err != nil || profileURL.Hostname() == "" { + return nil, fmt.Errorf("invalid profile URL %q", profile.URL) + } + + scheme := strings.ToLower(profileURL.Scheme) + if profile.Cloud || scheme == "https" { + return credentials.NewTLS(&tls.Config{ + MinVersion: tls.VersionTLS12, + ServerName: profileURL.Hostname(), + }), nil + } + if scheme == "http" { + return insecure.NewCredentials(), nil + } + return nil, fmt.Errorf("unsupported profile URL scheme %q", profileURL.Scheme) +} + +func tailAuthMetadata(profile config.Profile) (metadata.MD, error) { + mode, err := profile.AuthMode() + if err != nil { + return nil, err + } + + switch mode { + case config.AuthCloudAPIKey: + values := map[string]string{"x-api-key": profile.APIKey} + values["x-p-tenant"] = profile.TenantID + return metadata.New(values), nil + case config.AuthCloudOAuth: + values := map[string]string{"cookie": "session=" + profile.SessionToken} + values["x-p-tenant"] = profile.TenantID + return metadata.New(values), nil + case config.AuthSelfHostedAPIKey: + return metadata.New(map[string]string{ + "x-api-key": profile.APIKey, + }), nil + case config.AuthSelfHostedBasic: + return metadata.New(map[string]string{ + "Authorization": "Basic " + basicAuth(profile.Username, profile.Password), + }), nil + default: + return nil, fmt.Errorf("unsupported auth mode %q", mode) + } +} + // isStreamEnd returns true for normal stream termination codes that warrant a reconnect. func isStreamEnd(err error) bool { if err == io.EOF { diff --git a/cmd/tail_test.go b/cmd/tail_test.go new file mode 100644 index 0000000..ee59a19 --- /dev/null +++ b/cmd/tail_test.go @@ -0,0 +1,56 @@ +package cmd + +import ( + "testing" + + "github.com/parseablehq/pb/pkg/config" +) + +func TestTailTransportCredentials(t *testing.T) { + tests := []struct { + name string + profile config.Profile + wantProtocol string + wantServer string + }{ + { + name: "cloud always uses TLS", + profile: config.Profile{URL: "https://workspace.example.com", Cloud: true}, + wantProtocol: "tls", + wantServer: "workspace.example.com", + }, + { + name: "self-hosted HTTPS uses TLS", + profile: config.Profile{URL: "https://logs.example.com:8000"}, + wantProtocol: "tls", + wantServer: "logs.example.com", + }, + { + name: "self-hosted HTTP preserves insecure transport", + profile: config.Profile{URL: "http://localhost:8000"}, + wantProtocol: "insecure", + }, + } + + for _, test := range tests { + t.Run(test.name, func(t *testing.T) { + transportCredentials, err := tailTransportCredentials(test.profile) + if err != nil { + t.Fatal(err) + } + info := transportCredentials.Info() + if info.SecurityProtocol != test.wantProtocol { + t.Fatalf("protocol=%q want=%q", info.SecurityProtocol, test.wantProtocol) + } + if info.ServerName != test.wantServer { + t.Fatalf("server name=%q want=%q", info.ServerName, test.wantServer) + } + }) + } +} + +func TestTailTransportCredentialsRejectsInvalidURL(t *testing.T) { + if _, err := tailTransportCredentials(config.Profile{URL: "not-a-url"}); err == nil { + t.Fatal("expected invalid URL error") + } +} diff --git a/main.go b/main.go index 3764269..cea8257 100644 --- a/main.go +++ b/main.go @@ -18,6 +18,7 @@ package main import ( "bytes" + "encoding/json" "fmt" "os" "strings" @@ -28,6 +29,7 @@ import ( "github.com/parseablehq/pb/pkg/ui" "github.com/spf13/cobra" + "github.com/spf13/pflag" ) // populated at build time @@ -39,23 +41,32 @@ var ( var ( versionFlag = "version" versionFlagShort = "v" + rootOutputFormat string ) +const analyticsEnabled = false + // Root command var cli = &cobra.Command{ Use: "pb", Short: "\nParseable command line interface", Long: "\npb is the command line interface for Parseable", - PersistentPreRunE: analytics.CheckAndCreateULID, + PersistentPreRunE: analyticsPreRun, RunE: func(command *cobra.Command, _ []string) error { if p, _ := command.Flags().GetBool(versionFlag); p { pb.PrintVersion(Version, Commit) return nil } + if rootOutputFormat == "json" { + return printCommandJSON(command) + } + if rootOutputFormat != "text" { + return fmt.Errorf("unsupported output format %q (expected text or json)", rootOutputFormat) + } return command.Help() }, PersistentPostRun: func(cmd *cobra.Command, args []string) { - if os.Getenv("PB_ANALYTICS") == "disable" { + if !analyticsEnabled || os.Getenv("PB_ANALYTICS") == "disable" { return } go func() { @@ -66,13 +77,20 @@ var cli = &cobra.Command{ func postRunAnalytics(name string) func(*cobra.Command, []string) { return func(cmd *cobra.Command, args []string) { - if os.Getenv("PB_ANALYTICS") == "disable" { + if !analyticsEnabled || os.Getenv("PB_ANALYTICS") == "disable" { return } go analytics.PostRunAnalytics(cmd, name, args) } } +func analyticsPreRun(cmd *cobra.Command, args []string) error { + if !analyticsEnabled { + return nil + } + return analytics.CheckAndCreateULID(cmd, args) +} + type rootHelpCommand struct { name string desc string @@ -102,6 +120,7 @@ var rootHelpGroups = []rootHelpGroup{ title: "Identity & Access:", commands: []rootHelpCommand{ {name: "login", desc: "Login to Parseable"}, + {name: "cloud", desc: "Manage Parseable Cloud profiles"}, {name: "logout", desc: "Logout from the current Parseable profile"}, {name: "profile", desc: "Manage different Parseable targets"}, {name: "user", desc: "Manage users"}, @@ -128,7 +147,7 @@ var profile = &cobra.Command{ Use: "profile", Short: "Manage different Parseable targets", Long: "\nuse profile command to configure different Parseable instances. Each profile takes a URL and credentials.", - PersistentPreRunE: analytics.CheckAndCreateULID, + PersistentPreRunE: analyticsPreRun, PersistentPostRun: postRunAnalytics("profile"), } @@ -234,6 +253,16 @@ func main() { pb.PromqlCmd.PersistentPreRunE = combinedPreRun pb.PromqlCmd.PersistentPostRun = postRunAnalytics("promql") + configureParentOutput(profile) + configureParentOutput(sql) + configureParentOutput(pb.PromqlCmd) + configureParentOutput(pb.PromqlCardinalityCmd()) + configureParentOutput(dataset) + configureParentOutput(pb.CloudCmd) + configureParentOutput(pb.CloudProfileCmd) + configureParentOutput(user) + configureParentOutput(role) + schema.AddCommand(pb.GenerateSchemaCmd) schema.AddCommand(pb.CreateSchemaCmd) @@ -258,6 +287,7 @@ func main() { // cli.AddCommand(cluster) cli.AddCommand(pb.LoginCmd) + cli.AddCommand(pb.CloudCmd) cli.AddCommand(pb.LogoutCmd) cli.AddCommand(pb.StatusCmd) @@ -269,6 +299,8 @@ func main() { cli.AddCommand(pb.VersionCmd) // set as flag cli.Flags().BoolP(versionFlag, versionFlagShort, false, "Print version") + cli.Flags().StringVarP(&rootOutputFormat, "output", "o", "text", "Output format (text|json)") + cli.SetHelpCommand(newHelpCommand()) cli.SetHelpFunc(renderRootHelp) cli.CompletionOptions.HiddenDefaultCmd = true @@ -279,6 +311,98 @@ func main() { } } +type jsonCommand struct { + Command string `json:"command"` + Usage string `json:"usage"` + Description string `json:"description,omitempty"` + Aliases []string `json:"aliases,omitempty"` + Flags []jsonFlag `json:"flags,omitempty"` + Subcommands []jsonCommand `json:"subcommands,omitempty"` +} + +type jsonFlag struct { + Name string `json:"name"` + Shorthand string `json:"shorthand,omitempty"` + Usage string `json:"usage"` +} + +func commandJSON(cmd *cobra.Command) jsonCommand { + description := strings.TrimSpace(cmd.Long) + if description == "" { + description = strings.TrimSpace(cmd.Short) + } + result := jsonCommand{ + Command: cmd.CommandPath(), + Usage: cmd.UseLine(), + Description: description, + Aliases: cmd.Aliases, + } + seenFlags := make(map[string]struct{}) + appendFlags := func(flags *pflag.FlagSet) { + flags.VisitAll(func(flag *pflag.Flag) { + if _, exists := seenFlags[flag.Name]; exists || flag.Hidden { + return + } + seenFlags[flag.Name] = struct{}{} + result.Flags = append(result.Flags, jsonFlag{Name: flag.Name, Shorthand: flag.Shorthand, Usage: flag.Usage}) + }) + } + appendFlags(cmd.NonInheritedFlags()) + appendFlags(cmd.InheritedFlags()) + for _, child := range cmd.Commands() { + if child.IsAvailableCommand() && child.Name() != "help" { + result.Subcommands = append(result.Subcommands, commandJSON(child)) + } + } + return result +} + +func printCommandJSON(cmd *cobra.Command) error { + encoder := json.NewEncoder(cmd.OutOrStdout()) + encoder.SetIndent("", " ") + return encoder.Encode(commandJSON(cmd)) +} + +func configureParentOutput(cmd *cobra.Command) { + var output string + cmd.Flags().StringVarP(&output, "output", "o", "text", "Output format (text|json)") + cmd.PreRunE = func(_ *cobra.Command, _ []string) error { return nil } + cmd.RunE = func(command *cobra.Command, _ []string) error { + switch output { + case "json": + return printCommandJSON(command) + case "text": + return command.Help() + default: + return fmt.Errorf("unsupported output format %q (expected text or json)", output) + } + } +} + +func newHelpCommand() *cobra.Command { + var output string + help := &cobra.Command{ + Use: "help [command]", + Short: "Help about any command", + Args: cobra.ArbitraryArgs, + RunE: func(command *cobra.Command, args []string) error { + target, _, err := command.Root().Find(args) + if err != nil { + return err + } + if output == "json" { + return printCommandJSON(target) + } + if output != "text" { + return fmt.Errorf("unsupported output format %q (expected text or json)", output) + } + return target.Help() + }, + } + help.Flags().StringVarP(&output, "output", "o", "text", "Output format (text|json)") + return help +} + func renderRootHelp(cmd *cobra.Command, _ []string) { if cmd != cli { renderCommandHelp(cmd) @@ -431,8 +555,10 @@ func combinedPreRun(cmd *cobra.Command, args []string) error { return fmt.Errorf("error initializing default profile: %w", err) } - if err := analytics.CheckAndCreateULID(cmd, args); err != nil { - return fmt.Errorf("error while creating ulid: %v", err) + if analyticsEnabled { + if err := analytics.CheckAndCreateULID(cmd, args); err != nil { + return fmt.Errorf("error while creating ulid: %v", err) + } } return nil diff --git a/pkg/analytics/analytics.go b/pkg/analytics/analytics.go index e3fafd5..af18203 100644 --- a/pkg/analytics/analytics.go +++ b/pkg/analytics/analytics.go @@ -35,7 +35,6 @@ import ( "github.com/oklog/ulid/v2" "github.com/spf13/cobra" - "github.com/spf13/pflag" "gopkg.in/yaml.v2" ) @@ -148,63 +147,41 @@ func CheckAndCreateULID(_ *cobra.Command, _ []string) error { return nil } -func PostRunAnalytics(cmd *cobra.Command, name string, args []string) { +func PostRunAnalytics(cmd *cobra.Command, name string, _ []string) { executionTime := cmd.Annotations["executionTime"] - commandError := cmd.Annotations["error"] - flags := make(map[string]string) - cmd.Flags().VisitAll(func(flag *pflag.Flag) { - flags[flag.Name] = flag.Value.String() - }) - - // Call SendEvent in PostRunE - err := sendEvent( - name, - append(args, cmd.Name()), - &commandError, // Pass the error here if there was one - executionTime, - flags, - ) + + // Never collect arguments, flag values, errors, or profile data. They may + // contain queries, passwords, API keys, session tokens, or server details. + err := sendEvent(name, executionTime) if err != nil { - fmt.Println("Error sending analytics event:", err) + fmt.Fprintln(os.Stderr, "Error sending analytics event:", err) } } // sendEvent is a placeholder function to simulate sending an event after command execution. -func sendEvent(commandName string, arguments []string, errors *string, executionTimestamp string, flags map[string]string) error { +func sendEvent(commandName string, executionTimestamp string) error { ulid, err := ReadUULD() if err != nil { return fmt.Errorf("could not load ULID: %v", err) } - profile, err := GetProfile() - if err != nil { - return fmt.Errorf("failed to get profile: %v", err) - } - - httpClient := internalHTTP.DefaultClient(&profile) - - about, _ := FetchAbout(&httpClient) - // if err != nil { - // return fmt.Errorf("failed to get about metadata for profile: %v", err) - // } - // Create the Command struct cmd := Command{ Name: commandName, - Arguments: arguments, - Flags: flags, + Arguments: []string{}, + Flags: map[string]string{}, } // Populate the Event struct with OS details and timestamp event := Event{ - CLIVersion: about.Commit, + CLIVersion: "", ULID: ulid, - CommitHash: about.Commit, + CommitHash: "", OSName: GetOSName(), OSVersion: GetOSVersion(), ReportCreatedAt: GetCurrentTimestamp(), Command: cmd, - Errors: errors, + Errors: nil, ExecutionTimestamp: executionTimestamp, } @@ -215,7 +192,7 @@ func sendEvent(commandName string, arguments []string, errors *string, execution } // Define the target URL for the HTTP request - url := "https://analytics.parseable.io:80/pb" + url := config.AnalyticsURL // Create the HTTP POST request req, err := http.NewRequest("POST", url, bytes.NewBuffer(eventJSON)) @@ -226,7 +203,8 @@ func sendEvent(commandName string, arguments []string, errors *string, execution req.Header.Set("X-P-Stream", "pb-usage") // Execute the HTTP request - resp, err := httpClient.Client.Do(req) + httpClient := &http.Client{Timeout: 10 * time.Second} + resp, err := httpClient.Do(req) if err != nil { return fmt.Errorf("failed to send event: %v", err) } diff --git a/pkg/analytics/analytics_test.go b/pkg/analytics/analytics_test.go new file mode 100644 index 0000000..986efad --- /dev/null +++ b/pkg/analytics/analytics_test.go @@ -0,0 +1,59 @@ +package analytics + +import ( + "encoding/json" + "io" + "net/http" + "strings" + "testing" + + "github.com/spf13/cobra" +) + +type roundTripFunc func(*http.Request) (*http.Response, error) + +func (fn roundTripFunc) RoundTrip(req *http.Request) (*http.Response, error) { + return fn(req) +} + +func TestPostRunAnalyticsDoesNotCollectPrivateCommandData(t *testing.T) { + t.Setenv("HOME", t.TempDir()) + if err := CheckAndCreateULID(nil, nil); err != nil { + t.Fatal(err) + } + + var got Event + originalTransport := http.DefaultTransport + http.DefaultTransport = roundTripFunc(func(req *http.Request) (*http.Response, error) { + if err := json.NewDecoder(req.Body).Decode(&got); err != nil { + t.Fatal(err) + } + return &http.Response{ + StatusCode: http.StatusOK, + Status: "200 OK", + Header: make(http.Header), + Body: io.NopCloser(strings.NewReader("")), + Request: req, + }, nil + }) + t.Cleanup(func() { http.DefaultTransport = originalTransport }) + + cmd := &cobra.Command{Use: "profile"} + cmd.Annotations = map[string]string{ + "error": "secret-error", + "executionTime": "1s", + } + cmd.Flags().String("api-key", "", "") + if err := cmd.Flags().Set("api-key", "secret-api-key"); err != nil { + t.Fatal(err) + } + + PostRunAnalytics(cmd, "profile", []string{"prod", "https://example.com", "user", "secret-password"}) + + if got.Command.Name != "profile" { + t.Fatalf("command name mismatch: %q", got.Command.Name) + } + if len(got.Command.Arguments) != 0 || len(got.Command.Flags) != 0 || got.Errors != nil { + t.Fatalf("private command data collected: %#v", got) + } +} diff --git a/pkg/config/config.go b/pkg/config/config.go index 7c6bdc2..7a1b043 100644 --- a/pkg/config/config.go +++ b/pkg/config/config.go @@ -24,6 +24,7 @@ import ( "os" path "path/filepath" "runtime" + "sync" toml "github.com/pelletier/go-toml/v2" ) @@ -31,6 +32,7 @@ import ( var ( configFilename = "config.toml" configAppName = "pb" + configFileMu sync.Mutex ) // Path returns the config file path. @@ -58,6 +60,27 @@ func Path() (string, error) { return path.Join(dir, configAppName, configFilename), nil } +// UpdateCloudTokens persists a rotated cloud session and refresh token. +func UpdateCloudTokens(oldRefreshToken, sessionToken, refreshToken string) error { + configFileMu.Lock() + defer configFileMu.Unlock() + + fileConfig, err := ReadConfigFromFile() + if err != nil { + return err + } + for name, profile := range fileConfig.Profiles { + if !profile.Cloud || profile.RefreshToken != oldRefreshToken { + continue + } + profile.SessionToken = sessionToken + profile.RefreshToken = refreshToken + fileConfig.Profiles[name] = profile + return WriteConfigToFile(fileConfig) + } + return errors.New("cloud profile for refresh token not found") +} + // Config is the struct that holds the configuration type Config struct { Profiles map[string]Profile @@ -69,7 +92,71 @@ type Profile struct { URL string `toml:"url" json:"url"` Username string `toml:"username,omitempty" json:"username,omitempty"` Password string `toml:"password,omitempty" json:"password,omitempty"` - Token string `toml:"token,omitempty" json:"token,omitempty"` + + Cloud bool `toml:"cloud" json:"cloud"` + APIKey string `toml:"apiKey,omitempty" json:"apiKey,omitempty"` + SessionToken string `toml:"session_token,omitempty" json:"session_token,omitempty"` + RefreshToken string `toml:"refresh_token,omitempty" json:"refresh_token,omitempty"` + TenantID string `toml:"tenant_id,omitempty" json:"tenant_id,omitempty"` + IngestURL string `toml:"ingest_url,omitempty" json:"ingest_url,omitempty"` + WorkspaceID string `toml:"workspace_id,omitempty" json:"workspace_id,omitempty"` + WorkspaceName string `toml:"workspace_name,omitempty" json:"workspace_name,omitempty"` + OrchestratorURL string `toml:"orchestrator_url,omitempty" json:"orchestrator_url,omitempty"` +} + +type AuthMode string + +const ( + // AuthSelfHostedBasic identifies username/password authentication. + AuthSelfHostedBasic AuthMode = "self_hosted_basic" + // AuthSelfHostedAPIKey identifies API-key authentication for a self-hosted server. + AuthSelfHostedAPIKey AuthMode = "self_hosted_api_key" + // AuthCloudAPIKey identifies API-key authentication for Parseable Cloud. + AuthCloudAPIKey AuthMode = "cloud_api_key" + // AuthCloudOAuth identifies browser-based OAuth authentication for Parseable Cloud. + AuthCloudOAuth AuthMode = "cloud_oauth" +) + +func (p Profile) AuthMode() (AuthMode, error) { + hasBasic := p.Username != "" || p.Password != "" + hasAPIKey := p.APIKey != "" + hasCloudOAuth := p.SessionToken != "" + + if p.Cloud { + if p.TenantID == "" { + return "", errors.New("cloud profile missing tenant_id") + } + if hasBasic { + return "", errors.New("cloud profile cannot use self-hosted credentials") + } + switch { + case hasAPIKey && !hasCloudOAuth: + return AuthCloudAPIKey, nil + case hasCloudOAuth && !hasAPIKey: + return AuthCloudOAuth, nil + case hasAPIKey && hasCloudOAuth: + return "", errors.New("cloud profile has both apiKey and session_token") + default: + return "", errors.New("cloud profile missing apiKey or session_token") + } + } + + if hasCloudOAuth || p.TenantID != "" { + return "", errors.New("self-hosted profile cannot use cloud credentials") + } + switch { + case hasBasic && !hasAPIKey: + if p.Username == "" || p.Password == "" { + return "", errors.New("self-hosted basic profile missing username or password") + } + return AuthSelfHostedBasic, nil + case hasAPIKey && !hasBasic: + return AuthSelfHostedAPIKey, nil + case hasBasic && hasAPIKey: + return "", errors.New("self-hosted profile has both basic credentials and api key") + default: + return "", errors.New("self-hosted profile missing credentials") + } } func (p *Profile) GrpcAddr(port string) string { @@ -79,7 +166,10 @@ func (p *Profile) GrpcAddr(port string) string { // WriteConfigToFile writes the configuration to the config file func WriteConfigToFile(config *Config) error { - tomlData, _ := toml.Marshal(config) + tomlData, err := toml.Marshal(config) + if err != nil { + return fmt.Errorf("failed to encode config: %w", err) + } filePath, err := Path() if err != nil { return err @@ -90,12 +180,16 @@ func WriteConfigToFile(config *Config) error { return err } - file, err := os.Create(filePath) + file, err := os.OpenFile(filePath, os.O_WRONLY|os.O_CREATE|os.O_TRUNC, 0o600) if err != nil { fmt.Println("Error creating the file:", err) return err } defer file.Close() + if err := file.Chmod(0o600); err != nil { + fmt.Println("Error setting file permissions:", err) + return err + } // Write the data into the file _, err = file.Write(tomlData) if err != nil { @@ -106,7 +200,7 @@ func WriteConfigToFile(config *Config) error { } // ReadConfigFromFile reads the configuration from the config file -func ReadConfigFromFile() (config *Config, err error) { +func ReadConfigFromFile() (*Config, error) { filePath, err := Path() if err != nil { return &Config{}, err @@ -117,12 +211,43 @@ func ReadConfigFromFile() (config *Config, err error) { return &Config{}, err } - err = toml.Unmarshal(data, &config) - if err != nil { + var config Config + if err := toml.Unmarshal(data, &config); err != nil { return &Config{}, err } + if config.Profiles == nil { + config.Profiles = make(map[string]Profile) + } + applyLegacyAPIKeyConfig(data, &config) - return config, nil + return &config, nil +} + +func applyLegacyAPIKeyConfig(data []byte, config *Config) { + type legacyProfile struct { + Token string `toml:"token,omitempty"` + APIKey string `toml:"api_key,omitempty"` + } + type legacyConfig struct { + Profiles map[string]legacyProfile + } + + var legacy legacyConfig + if err := toml.Unmarshal(data, &legacy); err != nil { + return + } + for name, legacyProfile := range legacy.Profiles { + profile := config.Profiles[name] + if profile.APIKey == "" { + profile.APIKey = legacyProfile.APIKey + } + if profile.APIKey == "" { + profile.APIKey = legacyProfile.Token + } + if profile.APIKey != "" { + config.Profiles[name] = profile + } + } } func GetProfile() (Profile, error) { diff --git a/pkg/config/defaults.go b/pkg/config/defaults.go new file mode 100644 index 0000000..a265bdd --- /dev/null +++ b/pkg/config/defaults.go @@ -0,0 +1,42 @@ +// Copyright (c) 2024 Parseable, Inc +// +// This program is free software: you can redistribute it and/or modify +// it under the terms of the GNU Affero General Public License as published by +// the Free Software Foundation, either version 3 of the License, or +// (at your option) any later version. + +package config + +import "fmt" + +// CloudOrchestratorURL uses staging for local/development builds. Release +// builds replace it with the production URL through Go's -X linker flag. +var CloudOrchestratorURL = "ADD ORCHESTRATOR URL HERE" + +// Service and user-facing URL defaults used by pb. +const ( + // Self-hosted defaults used by self-hosted login (placeholder values). + LocalParseableURL = "http://localhost:8000" + + // Analytics is retained in the codebase but disabled for this release by + // analyticsEnabled in main.go. + AnalyticsURL = "https://analytics.parseable.io:80/pb" + + // Installer defaults are retained for future development. The cluster + // install/uninstall commands are not registered in this release, so these + // values are not reachable through the released CLI. + HelmChartRepositoryURL = "https://charts.parseable.com" + GoogleCloudStorageURL = "https://storage.googleapis.com" + DocumentationURL = "https://www.parseable.com/docs/server/introduction" + StreamManagementDocsURL = "https://www.parseable.com/docs/server/api" +) + +// AmazonS3URL is used only by the currently unregistered cluster installer. +func AmazonS3URL(region string) string { + return fmt.Sprintf("https://s3.%s.amazonaws.com", region) +} + +// AzureBlobStorageURL is used only by the currently unregistered cluster installer. +func AzureBlobStorageURL(storageAccount string) string { + return fmt.Sprintf("https://%s.blob.core.windows.net", storageAccount) +} diff --git a/pkg/datasets/datasets.go b/pkg/datasets/datasets.go index 654d971..f2df91a 100644 --- a/pkg/datasets/datasets.go +++ b/pkg/datasets/datasets.go @@ -11,6 +11,7 @@ import ( "time" "github.com/parseablehq/pb/pkg/config" + internalHTTP "github.com/parseablehq/pb/pkg/http" ) const ( @@ -24,6 +25,7 @@ const ( type Dataset struct { Title string `json:"title"` + Name string `json:"name,omitempty"` DatasetType string `json:"datasetType"` DatasetFormat string `json:"datasetFormat"` Ingestion bool `json:"ingestion"` @@ -33,20 +35,37 @@ type homeResponse struct { Datasets []Dataset `json:"datasets"` } +type httpStatusError struct { + statusCode int + status string + body string +} + +func (err httpStatusError) Error() string { + return fmt.Sprintf("HTTP %d: %s", err.statusCode, strings.TrimSpace(err.body)) +} + func FetchHomeDatasets(profile config.Profile) ([]Dataset, error) { + return fetchPrismHomeDatasets(profile) +} + +func fetchPrismHomeDatasets(profile config.Profile) ([]Dataset, error) { reqURL, err := url.JoinPath(profile.URL, "api/prism/v1/home") if err != nil { return nil, err } - client := &http.Client{Timeout: 15 * time.Second} + client := internalHTTP.DefaultClient(&profile) + client.Client.Timeout = 15 * time.Second req, err := http.NewRequest(http.MethodGet, reqURL, nil) if err != nil { return nil, err } - authenticate(req, profile) + if err := authenticate(req, profile); err != nil { + return nil, err + } - resp, err := client.Do(req) + resp, err := client.Client.Do(req) if err != nil { return nil, err } @@ -57,19 +76,27 @@ func FetchHomeDatasets(profile config.Profile) ([]Dataset, error) { return nil, err } if resp.StatusCode >= 400 { - return nil, fmt.Errorf("HTTP %d: %s", resp.StatusCode, strings.TrimSpace(string(body))) + return nil, httpStatusError{statusCode: resp.StatusCode, status: resp.Status, body: string(body)} } var result homeResponse if err := json.Unmarshal(body, &result); err != nil { return nil, err } - sort.Slice(result.Datasets, func(i, j int) bool { - return result.Datasets[i].Title < result.Datasets[j].Title - }) + sortDatasets(result.Datasets) return result.Datasets, nil } +func FetchAPIDatasets(profile config.Profile) ([]Dataset, error) { + return fetchPrismHomeDatasets(profile) +} + +func sortDatasets(items []Dataset) { + sort.Slice(items, func(i, j int) bool { + return items[i].Title < items[j].Title + }) +} + func NamesByType(items []Dataset, datasetType string) []string { var names []string for _, item := range items { @@ -81,11 +108,10 @@ func NamesByType(items []Dataset, datasetType string) []string { return names } -func authenticate(req *http.Request, profile config.Profile) { - if profile.Token != "" { - req.Header.Set("Authorization", "Bearer "+profile.Token) - } else { - req.SetBasicAuth(profile.Username, profile.Password) +func authenticate(req *http.Request, profile config.Profile) error { + if err := internalHTTP.AddAuthHeaders(req, &profile); err != nil { + return err } req.Header.Set("Content-Type", "application/json") + return nil } diff --git a/pkg/datasets/datasets_test.go b/pkg/datasets/datasets_test.go index dd089cc..986f489 100644 --- a/pkg/datasets/datasets_test.go +++ b/pkg/datasets/datasets_test.go @@ -1,8 +1,13 @@ package datasets import ( + "fmt" + "net/http" + "net/http/httptest" "reflect" "testing" + + "github.com/parseablehq/pb/pkg/config" ) func TestNamesByTypeUsesDatasetTypeMetadata(t *testing.T) { @@ -25,3 +30,26 @@ func TestNamesByTypeUsesDatasetTypeMetadata(t *testing.T) { t.Fatalf("metrics mismatch\nwant: %#v\n got: %#v", wantMetrics, gotMetrics) } } + +func TestFetchAPIDatasetsUsesSinglePrismHomeRequest(t *testing.T) { + requestCount := 0 + server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + requestCount++ + if r.URL.Path != "/api/prism/v1/home" { + t.Fatalf("unexpected path: %s", r.URL.Path) + } + fmt.Fprint(w, `{"datasets":[{"title":"traces","datasetType":"traces","datasetFormat":"otel-traces","ingestion":false},{"title":"logs","datasetType":"logs","datasetFormat":"otel-logs","ingestion":false}]}`) + })) + defer server.Close() + + items, err := FetchAPIDatasets(config.Profile{URL: server.URL, APIKey: "test-key"}) + if err != nil { + t.Fatal(err) + } + if requestCount != 1 { + t.Fatalf("request count=%d want=1", requestCount) + } + if len(items) != 2 || items[0].Title != "logs" || items[0].DatasetType != TypeLogs || items[1].DatasetType != TypeTraces { + t.Fatalf("unexpected datasets: %#v", items) + } +} diff --git a/pkg/http/http.go b/pkg/http/http.go index aeb6618..546cced 100644 --- a/pkg/http/http.go +++ b/pkg/http/http.go @@ -17,23 +17,43 @@ package cmd import ( + "errors" + "fmt" "io" "net/http" "net/url" + "os" + "strings" "time" "github.com/parseablehq/pb/pkg/config" ) +const ( + apiKeyHeader = "x-api-key" + tenantHeader = "X-P-Tenant" +) + type HTTPClient struct { Client http.Client Profile *config.Profile } func DefaultClient(profile *config.Profile) HTTPClient { + return DefaultClientWithTransport(profile, http.DefaultTransport) +} + +func DefaultClientWithTransport(profile *config.Profile, transport http.RoundTripper) HTTPClient { + if transport == nil { + transport = http.DefaultTransport + } return HTTPClient{ Client: http.Client{ Timeout: 60 * time.Second, + Transport: &cloudRefreshTransport{ + base: transport, + profile: profile, + }, }, Profile: profile, } @@ -45,15 +65,58 @@ func (client *HTTPClient) baseAPIURL(path string) (x string) { } func (client *HTTPClient) NewRequest(method string, path string, body io.Reader) (req *http.Request, err error) { + if client.Profile == nil { + return nil, errors.New("profile is nil") + } req, err = http.NewRequest(method, client.baseAPIURL(path), body) if err != nil { return } - if client.Profile.Token != "" { - req.Header.Set("Authorization", "Bearer "+client.Profile.Token) - } else { - req.SetBasicAuth(client.Profile.Username, client.Profile.Password) + if err = AddAuthHeaders(req, client.Profile); err != nil { + return } req.Header.Add("Content-Type", "application/json") + debugRequestHeaders(req) return } + +func AddAuthHeaders(req *http.Request, profile *config.Profile) error { + if profile == nil { + return errors.New("profile is nil") + } + + mode, err := profile.AuthMode() + if err != nil { + return err + } + + switch mode { + case config.AuthCloudAPIKey: + req.Header.Set(apiKeyHeader, profile.APIKey) + req.Header.Set(tenantHeader, profile.TenantID) + case config.AuthCloudOAuth: + req.AddCookie(&http.Cookie{Name: "session", Value: profile.SessionToken}) + req.Header.Set(tenantHeader, profile.TenantID) + case config.AuthSelfHostedAPIKey: + req.Header.Set(apiKeyHeader, profile.APIKey) + case config.AuthSelfHostedBasic: + req.SetBasicAuth(profile.Username, profile.Password) + } + return nil +} + +func debugRequestHeaders(req *http.Request) { + if os.Getenv("PB_DEBUG_HTTP_HEADERS") == "" { + return + } + + fmt.Fprintf(os.Stderr, "pb debug request: %s %s\n", req.Method, req.URL.String()) + for key, values := range req.Header { + value := strings.Join(values, ",") + switch strings.ToLower(key) { + case "authorization", apiKeyHeader, "cookie": + value = "" + } + fmt.Fprintf(os.Stderr, "pb debug header: %s: %s\n", key, value) + } +} diff --git a/pkg/http/http_test.go b/pkg/http/http_test.go new file mode 100644 index 0000000..9fc33fa --- /dev/null +++ b/pkg/http/http_test.go @@ -0,0 +1,188 @@ +package cmd + +import ( + "encoding/json" + "io" + "net/http" + "strings" + "testing" + + "github.com/parseablehq/pb/pkg/config" +) + +type testRoundTripFunc func(*http.Request) (*http.Response, error) + +func (fn testRoundTripFunc) RoundTrip(req *http.Request) (*http.Response, error) { + return fn(req) +} + +func testHTTPResponse(req *http.Request, statusCode int, body string) *http.Response { + return &http.Response{ + StatusCode: statusCode, + Status: http.StatusText(statusCode), + Header: make(http.Header), + Body: io.NopCloser(strings.NewReader(body)), + Request: req, + } +} + +func TestNewRequestWithNilProfileReturnsError(t *testing.T) { + client := DefaultClient(nil) + req, err := client.NewRequest(http.MethodGet, "about", nil) + if err == nil || err.Error() != "profile is nil" { + t.Fatalf("request=%v error=%v", req, err) + } + if req != nil { + t.Fatalf("expected nil request, got %v", req) + } +} + +func TestCloudSessionRefreshesAndRetriesWorkspaceRequest(t *testing.T) { + t.Setenv("XDG_CONFIG_HOME", t.TempDir()) + profile := config.Profile{ + URL: "https://workspace.example.com", + Cloud: true, + SessionToken: "old-session", + RefreshToken: "old-refresh", + TenantID: "tenant-id", + WorkspaceID: "workspace-id", + OrchestratorURL: "https://orchestrator.example.com", + } + if err := config.WriteConfigToFile(&config.Config{ + Profiles: map[string]config.Profile{"cloud": profile}, + DefaultProfile: "cloud", + }); err != nil { + t.Fatal(err) + } + + workspaceCalls := 0 + refreshCalls := 0 + base := testRoundTripFunc(func(req *http.Request) (*http.Response, error) { + switch req.URL.Host { + case "orchestrator.example.com": + refreshCalls++ + var payload cloudRefreshRequest + if err := json.NewDecoder(req.Body).Decode(&payload); err != nil { + t.Fatal(err) + } + if payload.RefreshToken != "old-refresh" { + t.Fatalf("refresh token=%q", payload.RefreshToken) + } + return testHTTPResponse(req, http.StatusOK, `{"access_token":"new-session","refresh_token":"new-refresh"}`), nil + case "workspace.example.com": + workspaceCalls++ + cookie, _ := req.Cookie("session") + if cookie != nil && cookie.Value == "new-session" { + return testHTTPResponse(req, http.StatusOK, `{}`), nil + } + return testHTTPResponse(req, http.StatusUnauthorized, ``), nil + default: + t.Fatalf("unexpected host: %s", req.URL.Host) + return nil, nil + } + }) + + client := DefaultClientWithTransport(&profile, base) + req, err := client.NewRequest(http.MethodGet, "about", nil) + if err != nil { + t.Fatal(err) + } + resp, err := client.Client.Do(req) + if err != nil { + t.Fatal(err) + } + resp.Body.Close() + + if resp.StatusCode != http.StatusOK || workspaceCalls != 2 || refreshCalls != 1 { + t.Fatalf("status=%d workspace_calls=%d refresh_calls=%d", resp.StatusCode, workspaceCalls, refreshCalls) + } + if profile.SessionToken != "new-session" || profile.RefreshToken != "new-refresh" { + t.Fatalf("tokens not updated: %#v", profile) + } +} + +func TestCloudSessionDoesNotRefreshCrossOriginRequest(t *testing.T) { + profile := config.Profile{ + URL: "https://workspace.example.com", + Cloud: true, + SessionToken: "session-token", + RefreshToken: "refresh-token", + TenantID: "tenant-id", + WorkspaceID: "workspace-id", + OrchestratorURL: "https://orchestrator.example.com", + } + externalCalls := 0 + refreshCalls := 0 + base := testRoundTripFunc(func(req *http.Request) (*http.Response, error) { + if req.URL.Host == "orchestrator.example.com" { + refreshCalls++ + } else { + externalCalls++ + } + if _, err := req.Cookie("session"); err == nil || req.Header.Get("X-P-Tenant") != "" { + t.Fatal("cloud credentials sent cross-origin") + } + return testHTTPResponse(req, http.StatusUnauthorized, ``), nil + }) + + client := DefaultClientWithTransport(&profile, base) + req, err := http.NewRequest(http.MethodGet, "https://analytics.example.com/event", nil) + if err != nil { + t.Fatal(err) + } + resp, err := client.Client.Do(req) + if err != nil { + t.Fatal(err) + } + resp.Body.Close() + + if externalCalls != 1 || refreshCalls != 0 { + t.Fatalf("external_calls=%d refresh_calls=%d", externalCalls, refreshCalls) + } +} + +func TestCloudSessionReturnsRefreshPersistenceError(t *testing.T) { + t.Setenv("XDG_CONFIG_HOME", t.TempDir()) + profile := config.Profile{ + URL: "https://workspace.example.com", + Cloud: true, + SessionToken: "old-session", + RefreshToken: "old-refresh", + TenantID: "tenant-id", + WorkspaceID: "workspace-id", + OrchestratorURL: "https://orchestrator.example.com", + } + if err := config.WriteConfigToFile(&config.Config{ + Profiles: map[string]config.Profile{ + "different": { + Cloud: true, + RefreshToken: "different-refresh", + }, + }, + }); err != nil { + t.Fatal(err) + } + + base := testRoundTripFunc(func(req *http.Request) (*http.Response, error) { + if req.URL.Host == "orchestrator.example.com" { + return testHTTPResponse(req, http.StatusOK, `{"access_token":"new-session","refresh_token":"new-refresh"}`), nil + } + return testHTTPResponse(req, http.StatusUnauthorized, ``), nil + }) + + client := DefaultClientWithTransport(&profile, base) + req, err := client.NewRequest(http.MethodGet, "about", nil) + if err != nil { + t.Fatal(err) + } + resp, err := client.Client.Do(req) + if err == nil || !strings.Contains(err.Error(), "failed to persist refreshed cloud tokens") { + t.Fatalf("response=%v error=%v", resp, err) + } + if resp != nil { + t.Fatalf("expected nil response, got %v", resp) + } + if profile.SessionToken != "new-session" || profile.RefreshToken != "new-refresh" { + t.Fatalf("rotated tokens were not retained in memory: %#v", profile) + } +} diff --git a/pkg/http/refresh.go b/pkg/http/refresh.go new file mode 100644 index 0000000..f31ccf3 --- /dev/null +++ b/pkg/http/refresh.go @@ -0,0 +1,153 @@ +package cmd + +import ( + "bytes" + "encoding/json" + "fmt" + "io" + "net/http" + "net/url" + "strings" + "sync" + + "github.com/parseablehq/pb/pkg/config" +) + +type cloudRefreshTransport struct { + base http.RoundTripper + profile *config.Profile + mu sync.Mutex +} + +type cloudRefreshRequest struct { + GrantType string `json:"grant_type"` + RefreshToken string `json:"refresh_token"` +} + +type cloudRefreshResponse struct { + AccessToken string `json:"access_token"` + RefreshToken string `json:"refresh_token"` +} + +func (transport *cloudRefreshTransport) RoundTrip(req *http.Request) (*http.Response, error) { + resp, err := transport.base.RoundTrip(req) + if err != nil || resp.StatusCode != http.StatusUnauthorized || !transport.isProfileRequest(req) || !transport.canRefresh() { + return resp, err + } + + transport.mu.Lock() + defer transport.mu.Unlock() + + requestToken := requestSessionCookie(req) + if requestToken == transport.profile.SessionToken { + if err := transport.refresh(req); err != nil { + resp.Body.Close() + return nil, fmt.Errorf("cloud session refresh failed: %w", err) + } + } + + retry, err := cloneRequestForCloudRetry(req) + if err != nil { + return resp, nil + } + if _, err := io.Copy(io.Discard, resp.Body); err != nil { + resp.Body.Close() + return nil, err + } + resp.Body.Close() + retry.Header.Del("Cookie") + if err := AddAuthHeaders(retry, transport.profile); err != nil { + return nil, err + } + return transport.base.RoundTrip(retry) +} + +func (transport *cloudRefreshTransport) isProfileRequest(req *http.Request) bool { + if transport.profile == nil || req == nil || req.URL == nil { + return false + } + profileURL, err := url.Parse(transport.profile.URL) + if err != nil { + return false + } + return strings.EqualFold(req.URL.Scheme, profileURL.Scheme) && + strings.EqualFold(req.URL.Host, profileURL.Host) +} + +func (transport *cloudRefreshTransport) canRefresh() bool { + if transport.profile == nil || transport.profile.RefreshToken == "" || transport.profile.OrchestratorURL == "" { + return false + } + mode, err := transport.profile.AuthMode() + return err == nil && mode == config.AuthCloudOAuth +} + +func (transport *cloudRefreshTransport) refresh(original *http.Request) error { + oldRefreshToken := transport.profile.RefreshToken + payload, err := json.Marshal(cloudRefreshRequest{ + GrantType: "refresh_token", + RefreshToken: oldRefreshToken, + }) + if err != nil { + return err + } + endpoint, err := url.JoinPath(transport.profile.OrchestratorURL, "api/v1/cli/oauth/token") + if err != nil { + return err + } + req, err := http.NewRequestWithContext(original.Context(), http.MethodPost, endpoint, bytes.NewReader(payload)) + if err != nil { + return err + } + req.Header.Set("Accept", "application/json") + req.Header.Set("Content-Type", "application/json") + resp, err := transport.base.RoundTrip(req) + if err != nil { + return err + } + defer resp.Body.Close() + body, err := io.ReadAll(resp.Body) + if err != nil { + return err + } + if resp.StatusCode < http.StatusOK || resp.StatusCode >= http.StatusMultipleChoices { + return fmt.Errorf("cloud session refresh failed: %s: %s", resp.Status, strings.TrimSpace(string(body))) + } + var result cloudRefreshResponse + if err := json.Unmarshal(body, &result); err != nil { + return err + } + if result.AccessToken == "" || result.RefreshToken == "" { + return fmt.Errorf("cloud session refresh response missing access_token or refresh_token") + } + transport.profile.SessionToken = result.AccessToken + transport.profile.RefreshToken = result.RefreshToken + if err := config.UpdateCloudTokens(oldRefreshToken, result.AccessToken, result.RefreshToken); err != nil { + return fmt.Errorf("failed to persist refreshed cloud tokens: %w", err) + } + return nil +} + +func requestSessionCookie(req *http.Request) string { + cookie, err := req.Cookie("session") + if err != nil { + return "" + } + return cookie.Value +} + +func cloneRequestForCloudRetry(req *http.Request) (*http.Request, error) { + retry := req.Clone(req.Context()) + if req.Body == nil { + return retry, nil + } + if req.GetBody == nil { + return nil, fmt.Errorf("request body cannot be replayed after cloud session refresh") + } + body, err := req.GetBody() + if err != nil { + return nil, err + } + retry.Body = body + return retry, nil +} diff --git a/pkg/installer/installer.go b/pkg/installer/installer.go index 0b6f981..19b745c 100644 --- a/pkg/installer/installer.go +++ b/pkg/installer/installer.go @@ -33,6 +33,7 @@ import ( "time" "github.com/parseablehq/pb/pkg/common" + "github.com/parseablehq/pb/pkg/config" "github.com/parseablehq/pb/pkg/helm" "github.com/manifoldco/promptui" @@ -95,7 +96,7 @@ func waterFall(verbose bool) { ReleaseName: pbInfo.Name, Namespace: pbInfo.Namespace, RepoName: "parseable", - RepoURL: "https://charts.parseable.com", + RepoURL: config.HelmChartRepositoryURL, ChartName: "parseable", Version: "2.6.6", Values: agentValues, @@ -157,7 +158,7 @@ func waterFall(verbose bool) { ReleaseName: pbInfo.Name, Namespace: pbInfo.Namespace, RepoName: "parseable", - RepoURL: "https://charts.parseable.com", + RepoURL: config.HelmChartRepositoryURL, ChartName: "parseable", Version: "2.6.6", Values: storeConfigs, @@ -536,8 +537,8 @@ func promptStoreConfigs(store ObjectStore, chartValues []string, plan Plan) (Obj // Dynamically construct the URL after Region is set storeValues.S3Store.URL = promptForInputWithDefault( - common.Yellow+" Enter S3 URL (default: https://s3."+storeValues.S3Store.Region+".amazonaws.com): "+common.Reset, - "https://s3."+storeValues.S3Store.Region+".amazonaws.com", + common.Yellow+" Enter S3 URL (default: "+config.AmazonS3URL(storeValues.S3Store.Region)+"): "+common.Reset, + config.AmazonS3URL(storeValues.S3Store.Region), ) sc, err := promptStorageClass() @@ -574,9 +575,9 @@ func promptStoreConfigs(store ObjectStore, chartValues []string, plan Plan) (Obj // Dynamically construct the URL after Region is set storeValues.BlobStore.URL = promptForInputWithDefault( common.Yellow+ - " Enter Blob URL (default: https://"+storeValues.BlobStore.StorageAccountName+".blob.core.windows.net): "+ + " Enter Blob URL (default: "+config.AzureBlobStorageURL(storeValues.BlobStore.StorageAccountName)+"): "+ common.Reset, - "https://"+storeValues.BlobStore.StorageAccountName+".blob.core.windows.net") + config.AzureBlobStorageURL(storeValues.BlobStore.StorageAccountName)) storeValues.StorageClass = sc storeValues.ObjectStore = BlobStore @@ -598,7 +599,7 @@ func promptStoreConfigs(store ObjectStore, chartValues []string, plan Plan) (Obj storeValues.GCSStore = GCS{ Bucket: promptForInputWithDefault(common.Yellow+" Enter GCS Bucket: "+common.Reset, ""), Region: promptForInputWithDefault(common.Yellow+" Enter GCS Region (default: us-east1): "+common.Reset, "us-east1"), - URL: promptForInputWithDefault(common.Yellow+" Enter GCS URL (default: https://storage.googleapis.com):", "https://storage.googleapis.com"), + URL: promptForInputWithDefault(common.Yellow+" Enter GCS URL (default: "+config.GoogleCloudStorageURL+"):", config.GoogleCloudStorageURL), AccessKey: promptForInputWithDefault(common.Yellow+" Enter GCS Access Key: "+common.Reset, ""), SecretKey: promptForInputWithDefault(common.Yellow+" Enter GCS Secret Key: "+common.Reset, ""), } @@ -841,8 +842,8 @@ func printSuccessBanner(pbInfo ParseableInfo, version, ingestorURL, queryURL str fmt.Printf(" • Ingestion URL: %s\n", ingestorURL) fmt.Println("\n" + common.Blue + "🔗 Resources:" + common.Reset) - fmt.Println(common.Blue + " • Documentation: https://www.parseable.com/docs/server/introduction") - fmt.Println(common.Blue + " • Stream Management: https://www.parseable.com/docs/server/api") + fmt.Println(common.Blue + " • Documentation: " + config.DocumentationURL) + fmt.Println(common.Blue + " • Stream Management: " + config.StreamManagementDocsURL) fmt.Println("\n" + common.Blue + "Happy Logging!" + common.Reset) diff --git a/pkg/installer/uninstaller.go b/pkg/installer/uninstaller.go index 072f6ff..054e1f4 100644 --- a/pkg/installer/uninstaller.go +++ b/pkg/installer/uninstaller.go @@ -26,6 +26,7 @@ import ( "github.com/manifoldco/promptui" "github.com/parseablehq/pb/pkg/common" + "github.com/parseablehq/pb/pkg/config" "github.com/parseablehq/pb/pkg/helm" apierrors "k8s.io/apimachinery/pkg/api/errors" v1 "k8s.io/apimachinery/pkg/apis/meta/v1" @@ -93,7 +94,7 @@ func Uninstaller(verbose bool) error { ReleaseName: selectedCluster.Name, Namespace: selectedCluster.Namespace, RepoName: "parseable", - RepoURL: "https://charts.parseable.com", + RepoURL: config.HelmChartRepositoryURL, ChartName: "parseable", Version: selectedCluster.Version, } diff --git a/pkg/model/login/login.go b/pkg/model/login/login.go index 2645a8a..a8d9bf7 100644 --- a/pkg/model/login/login.go +++ b/pkg/model/login/login.go @@ -30,12 +30,12 @@ type step int const ( stepChooseType step = iota - stepCloudSoon stepEnterURL + stepChooseCloudAuth stepChooseAuth stepEnterUsername stepEnterPassword - stepEnterToken + stepEnterAPIKey stepEnterProfileName stepConfirmReplace stepDone @@ -65,24 +65,26 @@ var ( // Model is the BubbleTea model for the interactive login wizard. type Model struct { - step step - typeIndex int // 0 = self-hosted, 1 = cloud - authIndex int // 0 = username+password, 1 = token - replaceIndex int // 0 = Replace, 1 = Change name + step step + typeIndex int // 0 = self-hosted, 1 = cloud + cloudAuthIndex int // 0 = OAuth browser, 1 = API key + authIndex int // 0 = username+password, 1 = api key + replaceIndex int // 0 = Replace, 1 = Change name urlInput textinput.Model usernameInput textinput.Model passwordInput textinput.Model - tokenInput textinput.Model + apiKeyInput textinput.Model profileNameInput textinput.Model serverURL string errMsg string // Result fields — set when Done is true. - Done bool - Profile config.Profile - Name string + Done bool + Profile config.Profile + Name string + CloudDeviceLogin bool } func newInput(placeholder string, charLimit int) textinput.Model { @@ -97,7 +99,7 @@ func newInput(placeholder string, charLimit int) textinput.Model { // New returns a fresh login wizard model. func New() Model { - urlInput := newInput("http://localhost:8000", 256) + urlInput := newInput(config.LocalParseableURL, 256) usernameInput := newInput("admin", 64) @@ -105,7 +107,9 @@ func New() Model { passwordInput.EchoMode = textinput.EchoPassword passwordInput.EchoCharacter = '•' - tokenInput := newInput("paste API key here", 512) + apiKeyInput := newInput("paste API key here", 512) + apiKeyInput.EchoMode = textinput.EchoPassword + apiKeyInput.EchoCharacter = '•' profileInput := newInput("e.g. local, staging, prod", 64) profileInput.SetValue("default") @@ -115,7 +119,7 @@ func New() Model { urlInput: urlInput, usernameInput: usernameInput, passwordInput: passwordInput, - tokenInput: tokenInput, + apiKeyInput: apiKeyInput, profileNameInput: profileInput, } } @@ -152,16 +156,43 @@ func (m Model) Update(msg tea.Msg) (tea.Model, tea.Cmd) { m.urlInput.Focus() return m, textinput.Blink } - m.step = stepCloudSoon + m.errMsg = "" + m.step = stepChooseCloudAuth + return m, nil } return m, nil - // ── Coming-soon screen ─────────────────────────────────────────────── - case stepCloudSoon: - m.step = stepChooseType + // ── Step 2a: cloud auth method ─────────────────────────────────────── + case stepChooseCloudAuth: + switch key.Type { + case tea.KeyEsc: + m.errMsg = "" + m.step = stepChooseType + return m, nil + case tea.KeyUp: + if m.cloudAuthIndex > 0 { + m.cloudAuthIndex-- + } + return m, nil + case tea.KeyDown: + if m.cloudAuthIndex < 1 { + m.cloudAuthIndex++ + } + return m, nil + case tea.KeyEnter: + m.errMsg = "" + if m.cloudAuthIndex == 0 { + m.step = stepEnterProfileName + m.profileNameInput.Focus() + } else { + m.step = stepEnterAPIKey + m.apiKeyInput.Focus() + } + return m, textinput.Blink + } return m, nil - // ── Step 2: server URL ─────────────────────────────────────────────── + // ── Step 2b: server URL ────────────────────────────────────────────── case stepEnterURL: switch key.Type { case tea.KeyEsc: @@ -206,8 +237,8 @@ func (m Model) Update(msg tea.Msg) (tea.Model, tea.Cmd) { m.step = stepEnterUsername m.usernameInput.Focus() } else { - m.step = stepEnterToken - m.tokenInput.Focus() + m.step = stepEnterAPIKey + m.apiKeyInput.Focus() } return m, textinput.Blink } @@ -254,22 +285,26 @@ func (m Model) Update(msg tea.Msg) (tea.Model, tea.Cmd) { return m, textinput.Blink } - // ── Step 4c: token ─────────────────────────────────────────────────── - case stepEnterToken: + // ── Step 4c: API key ───────────────────────────────────────────────── + case stepEnterAPIKey: switch key.Type { case tea.KeyEsc: m.errMsg = "" - m.step = stepChooseAuth - m.tokenInput.Blur() + if m.typeIndex == 1 { + m.step = stepChooseCloudAuth + } else { + m.step = stepChooseAuth + } + m.apiKeyInput.Blur() return m, nil case tea.KeyEnter: - if strings.TrimSpace(m.tokenInput.Value()) == "" { + if strings.TrimSpace(m.apiKeyInput.Value()) == "" { m.errMsg = "API key is required" return m, nil } m.errMsg = "" m.step = stepEnterProfileName - m.tokenInput.Blur() + m.apiKeyInput.Blur() m.profileNameInput.Focus() return m, textinput.Blink } @@ -280,12 +315,19 @@ func (m Model) Update(msg tea.Msg) (tea.Model, tea.Cmd) { case tea.KeyEsc: m.errMsg = "" m.profileNameInput.Blur() - if m.authIndex == 0 { + if m.typeIndex == 1 { + if m.cloudAuthIndex == 0 { + m.step = stepChooseCloudAuth + } else { + m.step = stepEnterAPIKey + m.apiKeyInput.Focus() + } + } else if m.authIndex == 0 { m.step = stepEnterPassword m.passwordInput.Focus() } else { - m.step = stepEnterToken - m.tokenInput.Focus() + m.step = stepEnterAPIKey + m.apiKeyInput.Focus() } return m, textinput.Blink case tea.KeyEnter: @@ -349,8 +391,8 @@ func (m Model) Update(msg tea.Msg) (tea.Model, tea.Cmd) { m.usernameInput, cmd = m.usernameInput.Update(msg) case stepEnterPassword: m.passwordInput, cmd = m.passwordInput.Update(msg) - case stepEnterToken: - m.tokenInput, cmd = m.tokenInput.Update(msg) + case stepEnterAPIKey: + m.apiKeyInput, cmd = m.apiKeyInput.Update(msg) case stepEnterProfileName: m.profileNameInput, cmd = m.profileNameInput.Update(msg) } @@ -359,16 +401,28 @@ func (m Model) Update(msg tea.Msg) (tea.Model, tea.Cmd) { func (m Model) finalize(name string) (tea.Model, tea.Cmd) { m.Name = name - if m.authIndex == 0 { + if m.typeIndex == 1 && m.cloudAuthIndex == 0 { + m.CloudDeviceLogin = true + m.Profile = config.Profile{ + Cloud: true, + } + } else if m.typeIndex == 1 { m.Profile = config.Profile{ + Cloud: true, + APIKey: strings.TrimSpace(m.apiKeyInput.Value()), + } + } else if m.authIndex == 0 { + m.Profile = config.Profile{ + Cloud: false, URL: m.serverURL, Username: strings.TrimSpace(m.usernameInput.Value()), Password: m.passwordInput.Value(), } } else { m.Profile = config.Profile{ - URL: m.serverURL, - Token: strings.TrimSpace(m.tokenInput.Value()), + Cloud: false, + URL: m.serverURL, + APIKey: strings.TrimSpace(m.apiKeyInput.Value()), } } m.Done = true @@ -401,6 +455,11 @@ func hint(pairs ...[2]string) string { // card with a fixed UPPERCASE title strip. Each step writes its own // label row + body + hint row, joined into the card. func (m Model) View() string { + // Device authorization and persistence happen after the wizard exits. + if m.step == stepDone && m.CloudDeviceLogin { + return "" + } + var b strings.Builder b.WriteString(titleStyle.Render("PARSEABLE LOGIN")) @@ -413,7 +472,7 @@ func (m Model) View() string { b.WriteString("\n\n") entries := []struct{ label, badge string }{ {"Self-hosted", ""}, - {"Parseable Cloud", " (coming soon)"}, + {"Parseable Cloud", ""}, } for i, e := range entries { if i == m.typeIndex { @@ -427,15 +486,6 @@ func (m Model) View() string { b.WriteString("\n") b.WriteString(hint([2]string{"↑↓", "navigate"}, [2]string{"enter", "select"}, [2]string{"ctrl-c", "quit"})) - case stepCloudSoon: - b.WriteString(labelStyle.Render("PARSEABLE CLOUD")) - b.WriteString("\n\n") - b.WriteString(normalStyle.Render(" We're working on it.")) - b.WriteString("\n") - b.WriteString(dimStyle.Render(" Cloud login is coming soon.")) - b.WriteString("\n\n") - b.WriteString(hint([2]string{"any key", "back"})) - case stepEnterURL: b.WriteString(labelStyle.Render("SERVER URL")) b.WriteString("\n\n ") @@ -444,6 +494,21 @@ func (m Model) View() string { b.WriteString(renderErr(m.errMsg)) b.WriteString(hint([2]string{"esc", "back"}, [2]string{"enter", "continue"})) + case stepChooseCloudAuth: + b.WriteString(labelStyle.Render("CLOUD AUTHENTICATION")) + b.WriteString("\n\n") + authEntries := []string{"Device login (browser)", "API key"} + for i, entry := range authEntries { + if i == m.cloudAuthIndex { + b.WriteString(rowSelected(entry)) + } else { + b.WriteString(rowIdle(entry)) + } + b.WriteString("\n") + } + b.WriteString("\n") + b.WriteString(hint([2]string{"esc", "back"}, [2]string{"↑↓", "navigate"}, [2]string{"enter", "select"})) + case stepChooseAuth: b.WriteString(labelStyle.Render("AUTHENTICATION")) b.WriteString("\n\n") @@ -475,10 +540,14 @@ func (m Model) View() string { b.WriteString(renderErr(m.errMsg)) b.WriteString(hint([2]string{"esc", "back"}, [2]string{"enter", "continue"})) - case stepEnterToken: - b.WriteString(labelStyle.Render("API KEY")) + case stepEnterAPIKey: + if m.typeIndex == 1 { + b.WriteString(labelStyle.Render("CLOUD API KEY")) + } else { + b.WriteString(labelStyle.Render("API KEY")) + } b.WriteString("\n\n ") - b.WriteString(m.tokenInput.View()) + b.WriteString(m.apiKeyInput.View()) b.WriteString("\n\n") b.WriteString(renderErr(m.errMsg)) b.WriteString(hint([2]string{"esc", "back"}, [2]string{"enter", "continue"})) @@ -509,23 +578,32 @@ func (m Model) View() string { case stepDone: b.WriteString(successStyle.Render("✓ profile '" + m.Name + "' saved")) b.WriteString("\n\n") - b.WriteString(" " + labelStyle.Render("URL ")) - b.WriteString(normalStyle.Render(m.Profile.URL)) - b.WriteString("\n") + if m.Profile.URL != "" { + b.WriteString(" " + labelStyle.Render("URL ")) + b.WriteString(normalStyle.Render(m.Profile.URL)) + b.WriteString("\n") + } + if m.Profile.Cloud { + b.WriteString(" " + labelStyle.Render("AUTH ")) + b.WriteString(normalStyle.Render("Cloud API key (stored after validation)")) + b.WriteString("\n") + } if m.Profile.Username != "" { b.WriteString(" " + labelStyle.Render("USER ")) b.WriteString(normalStyle.Render(m.Profile.Username)) b.WriteString("\n") } - if m.Profile.Token != "" { + if m.Profile.APIKey != "" { b.WriteString(" " + labelStyle.Render("AUTH ")) b.WriteString(normalStyle.Render("API key (stored)")) b.WriteString("\n") } b.WriteString("\n") - b.WriteString(dimStyle.Render(" add more profiles:")) + b.WriteString(dimStyle.Render(" Next steps:")) + b.WriteString("\n ") + b.WriteString(hintStyle.Render("pb status Verify active connection")) b.WriteString("\n ") - b.WriteString(hintStyle.Render("pb profile add [user] [pass]")) + b.WriteString(hintStyle.Render("pb dataset list Explore available datasets")) } return lipgloss.NewStyle(). diff --git a/pkg/model/login/login_test.go b/pkg/model/login/login_test.go new file mode 100644 index 0000000..84baf81 --- /dev/null +++ b/pkg/model/login/login_test.go @@ -0,0 +1,17 @@ +package login + +import ( + "testing" + + "github.com/charmbracelet/bubbles/textinput" +) + +func TestAPIKeyInputIsMasked(t *testing.T) { + model := New() + if model.apiKeyInput.EchoMode != textinput.EchoPassword { + t.Fatalf("API key echo mode=%v", model.apiKeyInput.EchoMode) + } + if model.apiKeyInput.EchoCharacter != '•' { + t.Fatalf("API key echo character=%q", model.apiKeyInput.EchoCharacter) + } +} diff --git a/pkg/model/promql.go b/pkg/model/promql.go index 05c2b1a..47c2b7c 100644 --- a/pkg/model/promql.go +++ b/pkg/model/promql.go @@ -44,6 +44,7 @@ import ( table "github.com/evertras/bubble-table/table" "github.com/parseablehq/pb/pkg/config" "github.com/parseablehq/pb/pkg/datasets" + internalHTTP "github.com/parseablehq/pb/pkg/http" "github.com/parseablehq/pb/pkg/ui" "golang.org/x/term" ) @@ -2214,24 +2215,20 @@ func promqlModelFetch(profile config.Profile, path string, params url.Values) ([ reqURL += "?" + params.Encode() } - client := &http.Client{ - Timeout: 120 * time.Second, - Transport: &http.Transport{ - TLSNextProto: make(map[string]func(string, *tls.Conn) http.RoundTripper), - }, - } + client := internalHTTP.DefaultClientWithTransport(&profile, &http.Transport{ + TLSNextProto: make(map[string]func(string, *tls.Conn) http.RoundTripper), + }) + client.Client.Timeout = 120 * time.Second req, err := http.NewRequest("GET", reqURL, nil) if err != nil { return nil, err } - if profile.Token != "" { - req.Header.Set("Authorization", "Bearer "+profile.Token) - } else { - req.SetBasicAuth(profile.Username, profile.Password) + if err := internalHTTP.AddAuthHeaders(req, &profile); err != nil { + return nil, err } - resp, err := client.Do(req) + resp, err := client.Client.Do(req) if err != nil { if strings.Contains(err.Error(), "connection reset") { return nil, fmt.Errorf("server reset the connection — query timed out") @@ -2602,17 +2599,16 @@ type promqlLabelListResp struct { } func builderHTTPGetCtx(ctx context.Context, profile config.Profile, rawURL string) ([]byte, error) { - client := &http.Client{Timeout: 30 * time.Second} + client := internalHTTP.DefaultClient(&profile) + client.Client.Timeout = 30 * time.Second req, err := http.NewRequestWithContext(ctx, "GET", rawURL, nil) if err != nil { return nil, err } - if profile.Token != "" { - req.Header.Set("Authorization", "Bearer "+profile.Token) - } else { - req.SetBasicAuth(profile.Username, profile.Password) + if err := internalHTTP.AddAuthHeaders(req, &profile); err != nil { + return nil, err } - resp, err := client.Do(req) + resp, err := client.Client.Do(req) if err != nil { return nil, err } diff --git a/pkg/model/query.go b/pkg/model/query.go index 273aa58..fec2c00 100644 --- a/pkg/model/query.go +++ b/pkg/model/query.go @@ -40,6 +40,7 @@ import ( "github.com/muesli/reflow/truncate" "github.com/parseablehq/pb/pkg/config" "github.com/parseablehq/pb/pkg/datasets" + internalHTTP "github.com/parseablehq/pb/pkg/http" "github.com/parseablehq/pb/pkg/iterator" "github.com/parseablehq/pb/pkg/ui" "golang.org/x/exp/slices" @@ -2582,11 +2583,10 @@ func NewFetchTask(profile config.Profile, query string, startTime string, endTim } }() - client := &http.Client{ - Timeout: time.Second * 50, - } + client := internalHTTP.DefaultClient(&profile) + client.Client.Timeout = time.Second * 50 - data, status, errMsg := fetchData(client, &profile, query, startTime, endTime) + data, status, errMsg := fetchData(&client, &profile, query, startTime, endTime) if status == fetchOk { res.data = data.Records @@ -2616,11 +2616,10 @@ func NewSQLWindowFetchTask(profile config.Profile, runID int, baseQuery string, } }() - client := &http.Client{ - Timeout: time.Second * 50, - } + client := internalHTTP.DefaultClient(&profile) + client.Client.Timeout = time.Second * 50 query := injectSQLWindow(baseQuery, fetchLimit, baseOffset+windowIndex*windowSize) - data, status, errMsg := fetchDataRaw(client, &profile, query, startTime, endTime) + data, status, errMsg := fetchDataRaw(&client, &profile, query, startTime, endTime) if status == fetchOk { res.data = data.Records res.schema = data.Fields @@ -2673,14 +2672,14 @@ func IteratorPrev(iter *iterator.QueryIterator[QueryData, FetchResult]) tea.Cmd } } -func fetchData(client *http.Client, profile *config.Profile, query string, startTime string, endTime string) (data QueryData, res FetchResult, errMsg string) { +func fetchData(client *internalHTTP.HTTPClient, profile *config.Profile, query string, startTime string, endTime string) (data QueryData, res FetchResult, errMsg string) { query = quoteUnsafeSQLTableNames(query) query = quoteUnsafeSQLFieldNames(query) query = ensureDefaultSQLLimit(query) return fetchDataRaw(client, profile, query, startTime, endTime) } -func fetchDataRaw(client *http.Client, profile *config.Profile, query string, startTime string, endTime string) (data QueryData, res FetchResult, errMsg string) { +func fetchDataRaw(client *internalHTTP.HTTPClient, _ *config.Profile, query string, startTime string, endTime string) (data QueryData, res FetchResult, errMsg string) { data = QueryData{} res = fetchErr body, err := json.Marshal(map[string]string{ @@ -2693,20 +2692,16 @@ func fetchDataRaw(client *http.Client, profile *config.Profile, query string, st return } - endpoint, _ := url.JoinPath(profile.URL, "api/v1/query") - endpoint += "?fields=true" - req, err := http.NewRequest("POST", endpoint, bytes.NewBuffer(body)) + req, err := client.NewRequest("POST", "query", bytes.NewBuffer(body)) if err != nil { errMsg = err.Error() return } - if profile.Token != "" { - req.Header.Set("Authorization", "Bearer "+profile.Token) - } else { - req.SetBasicAuth(profile.Username, profile.Password) - } - req.Header.Add("Content-Type", "application/json") - resp, err := client.Do(req) + queryParams := req.URL.Query() + queryParams.Set("fields", "true") + req.URL.RawQuery = queryParams.Encode() + + resp, err := client.Client.Do(req) if err != nil { errMsg = err.Error() return @@ -3048,18 +3043,17 @@ func fetchStreamSchema(profile config.Profile, stream string) tea.Cmd { if stream == "" { return schemaMsg{} } - client := &http.Client{Timeout: 15 * time.Second} + client := internalHTTP.DefaultClient(&profile) + client.Client.Timeout = 15 * time.Second // Primary: GET /api/v1/logstream/{stream}/schema schemaURL, _ := url.JoinPath(profile.URL, "api/v1/logstream", stream, "schema") req, err := http.NewRequest("GET", schemaURL, nil) if err == nil { - if profile.Token != "" { - req.Header.Set("Authorization", "Bearer "+profile.Token) - } else { - req.SetBasicAuth(profile.Username, profile.Password) + if err := internalHTTP.AddAuthHeaders(req, &profile); err != nil { + return schemaMsg{errMsg: err.Error()} } - if resp, err := client.Do(req); err == nil { + if resp, err := client.Client.Do(req); err == nil { body, _ := io.ReadAll(resp.Body) resp.Body.Close() if resp.StatusCode == http.StatusOK { @@ -3088,7 +3082,7 @@ func fetchStreamSchema(profile config.Profile, stream string) tea.Cmd { // Stream names with hyphens (e.g. my-stream) are invalid bare SQL // identifiers, so single-quote them — same as resolveDatasetPlaceholder. endTime := time.Now().UTC().Format(time.RFC3339) - data, res, errMsg := fetchData(client, &profile, + data, res, errMsg := fetchData(&client, &profile, fmt.Sprintf("SELECT * FROM '%s' LIMIT 1", stream), "2000-01-01T00:00:00+00:00", endTime) if res != fetchOk { diff --git a/pkg/model/savedQueries.go b/pkg/model/savedQueries.go index d387924..5a08795 100644 --- a/pkg/model/savedQueries.go +++ b/pkg/model/savedQueries.go @@ -22,9 +22,9 @@ import ( "io" "net/http" "strings" - "time" "github.com/parseablehq/pb/pkg/config" + internalHTTP "github.com/parseablehq/pb/pkg/http" "github.com/parseablehq/pb/pkg/ui" "github.com/charmbracelet/bubbles/key" @@ -282,10 +282,8 @@ func SavedQueriesMenu() *tea.Program { userProfile = profile } - client := &http.Client{ - Timeout: time.Second * 60, - } - userSavedQueries := fetchFilters(client, &userProfile) + client := internalHTTP.DefaultClient(&userProfile) + userSavedQueries := fetchFilters(&client) m := modelSavedQueries{list: list.New(userSavedQueries, itemDelegate{}, 0, 0)} m.list.SetShowTitle(false) @@ -297,17 +295,14 @@ func SavedQueriesMenu() *tea.Program { } // fetchFilters fetches saved SQL queries for the active user from the server -func fetchFilters(client *http.Client, profile *config.Profile) []list.Item { - endpoint := fmt.Sprintf("%s/%s", profile.URL, "api/v1/filters") - req, err := http.NewRequest("GET", endpoint, nil) +func fetchFilters(client *internalHTTP.HTTPClient) []list.Item { + req, err := client.NewRequest("GET", "filters", nil) if err != nil { fmt.Println("Error creating request:", err) return nil } - req.SetBasicAuth(profile.Username, profile.Password) - req.Header.Add("Content-Type", "application/json") - resp, err := client.Do(req) + resp, err := client.Client.Do(req) if err != nil { fmt.Println("Error making request:", err) return nil @@ -357,7 +352,7 @@ func QueryToDelete() Item { return selectedQueryDelete } -func RunQuery(client *http.Client, profile *config.Profile, query string, startTime string, endTime string) (string, error) { +func RunQuery(client *internalHTTP.HTTPClient, _ *config.Profile, query string, startTime string, endTime string) (string, error) { queryTemplate := `{ "query": "%s", "startTime": "%s", @@ -366,15 +361,12 @@ func RunQuery(client *http.Client, profile *config.Profile, query string, startT finalQuery := fmt.Sprintf(queryTemplate, query, startTime, endTime) - endpoint := fmt.Sprintf("%s/%s", profile.URL, "api/v1/query") - req, err := http.NewRequest("POST", endpoint, bytes.NewBuffer([]byte(finalQuery))) + req, err := client.NewRequest("POST", "query", bytes.NewBuffer([]byte(finalQuery))) if err != nil { return "", err } - req.SetBasicAuth(profile.Username, profile.Password) - req.Header.Add("Content-Type", "application/json") - resp, err := client.Do(req) + resp, err := client.Client.Do(req) if err != nil { return "", err }